sgi-xp: setup the activate GRU message queue

Setup the activate GRU message queue that is used for partition activation
and channel connection on UV systems.

Signed-off-by: Dean Nelson <dcn@sgi.com>
Cc: Jack Steiner <steiner@sgi.com>
Cc: "Luck, Tony" <tony.luck@intel.com>
Signed-off-by: Andrew Morton <akpm@linux-foundation.org>
Signed-off-by: Linus Torvalds <torvalds@linux-foundation.org>

1 files changed, 751 insertions, 30 deletions

diff --git a/drivers/misc/sgi-xp/xpc_uv.c b/drivers/misc/sgi-xp/xpc_uv.c
index c2d4ddd6e95..689cb5c68cc 100644
--- a/drivers/misc/sgi-xp/xpc_uv.c
+++ b/drivers/misc/sgi-xp/xpc_uv.c
@@ -14,41 +14,528 @@
  */
 
 #include <linux/kernel.h>
+#include <linux/mm.h>
+#include <linux/interrupt.h>
+#include <linux/delay.h>
+#include <linux/device.h>
 #include <asm/uv/uv_hub.h>
+#include "../sgi-gru/gru.h"
 #include "../sgi-gru/grukservices.h"
 #include "xpc.h"
 
+static atomic64_t xpc_heartbeat_uv;
 static DECLARE_BITMAP(xpc_heartbeating_to_mask_uv, XP_MAX_NPARTITIONS_UV);
 
-static void *xpc_activate_mq;
+#define XPC_ACTIVATE_MSG_SIZE_UV	(1 * GRU_CACHE_LINE_BYTES)
+#define XPC_NOTIFY_MSG_SIZE_UV		(2 * GRU_CACHE_LINE_BYTES)
+
+#define XPC_ACTIVATE_MQ_SIZE_UV	(4 * XP_MAX_NPARTITIONS_UV * \
+				 XPC_ACTIVATE_MSG_SIZE_UV)
+#define XPC_NOTIFY_MQ_SIZE_UV	(4 * XP_MAX_NPARTITIONS_UV * \
+				 XPC_NOTIFY_MSG_SIZE_UV)
+
+static void *xpc_activate_mq_uv;
+static void *xpc_notify_mq_uv;
+
+static int
+xpc_setup_partitions_sn_uv(void)
+{
+	short partid;
+	struct xpc_partition_uv *part_uv;
+
+	for (partid = 0; partid < XP_MAX_NPARTITIONS_UV; partid++) {
+		part_uv = &xpc_partitions[partid].sn.uv;
+
+		spin_lock_init(&part_uv->flags_lock);
+		part_uv->remote_act_state = XPC_P_AS_INACTIVE;
+	}
+	return 0;
+}
+
+static void *
+xpc_create_gru_mq_uv(unsigned int mq_size, int cpuid, unsigned int irq,
+		     irq_handler_t irq_handler)
+{
+	int ret;
+	int nid;
+	int mq_order;
+	struct page *page;
+	void *mq;
+
+	nid = cpu_to_node(cpuid);
+	mq_order = get_order(mq_size);
+	page = alloc_pages_node(nid, GFP_KERNEL | __GFP_ZERO | GFP_THISNODE,
+				mq_order);
+	if (page == NULL)
+		return NULL;
+
+	mq = page_address(page);
+	ret = gru_create_message_queue(mq, mq_size);
+	if (ret != 0) {
+		dev_err(xpc_part, "gru_create_message_queue() returned "
+			"error=%d\n", ret);
+		free_pages((unsigned long)mq, mq_order);
+		return NULL;
+	}
+
+	/* !!! Need to do some other things to set up IRQ */
+
+	ret = request_irq(irq, irq_handler, 0, "xpc", NULL);
+	if (ret != 0) {
+		dev_err(xpc_part, "request_irq(irq=%d) returned error=%d\n",
+			irq, ret);
+		free_pages((unsigned long)mq, mq_order);
+		return NULL;
+	}
+
+	/* !!! enable generation of irq when GRU mq op occurs to this mq */
+
+	/* ??? allow other partitions to access GRU mq? */
+
+	return mq;
+}
 
 static void
-xpc_send_local_activate_IRQ_uv(struct xpc_partition *part)
+xpc_destroy_gru_mq_uv(void *mq, unsigned int mq_size, unsigned int irq)
+{
+	/* ??? disallow other partitions to access GRU mq? */
+
+	/* !!! disable generation of irq when GRU mq op occurs to this mq */
+
+	free_irq(irq, NULL);
+
+	free_pages((unsigned long)mq, get_order(mq_size));
+}
+
+static enum xp_retval
+xpc_send_gru_msg(unsigned long mq_gpa, void *msg, size_t msg_size)
 {
+	enum xp_retval xp_ret;
+	int ret;
+
+	while (1) {
+		ret = gru_send_message_gpa(mq_gpa, msg, msg_size);
+		if (ret == MQE_OK) {
+			xp_ret = xpSuccess;
+			break;
+		}
+
+		if (ret == MQE_QUEUE_FULL) {
+			dev_dbg(xpc_chan, "gru_send_message_gpa() returned "
+				"error=MQE_QUEUE_FULL\n");
+			/* !!! handle QLimit reached; delay & try again */
+			/* ??? Do we add a limit to the number of retries? */
+			(void)msleep_interruptible(10);
+		} else if (ret == MQE_CONGESTION) {
+			dev_dbg(xpc_chan, "gru_send_message_gpa() returned "
+				"error=MQE_CONGESTION\n");
+			/* !!! handle LB Overflow; simply try again */
+			/* ??? Do we add a limit to the number of retries? */
+		} else {
+			/* !!! Currently this is MQE_UNEXPECTED_CB_ERR */
+			dev_err(xpc_chan, "gru_send_message_gpa() returned "
+				"error=%d\n", ret);
+			xp_ret = xpGruSendMqError;
+			break;
+		}
+	}
+	return xp_ret;
+}
+
+static void
+xpc_process_activate_IRQ_rcvd_uv(void)
+{
+	unsigned long irq_flags;
+	short partid;
+	struct xpc_partition *part;
+	u8 act_state_req;
+
+	DBUG_ON(xpc_activate_IRQ_rcvd == 0);
+
+	spin_lock_irqsave(&xpc_activate_IRQ_rcvd_lock, irq_flags);
+	for (partid = 0; partid < XP_MAX_NPARTITIONS_UV; partid++) {
+		part = &xpc_partitions[partid];
+
+		if (part->sn.uv.act_state_req == 0)
+			continue;
+
+		xpc_activate_IRQ_rcvd--;
+		BUG_ON(xpc_activate_IRQ_rcvd < 0);
+
+		act_state_req = part->sn.uv.act_state_req;
+		part->sn.uv.act_state_req = 0;
+		spin_unlock_irqrestore(&xpc_activate_IRQ_rcvd_lock, irq_flags);
+
+		if (act_state_req == XPC_P_ASR_ACTIVATE_UV) {
+			if (part->act_state == XPC_P_AS_INACTIVE)
+				xpc_activate_partition(part);
+			else if (part->act_state == XPC_P_AS_DEACTIVATING)
+				XPC_DEACTIVATE_PARTITION(part, xpReactivating);
+
+		} else if (act_state_req == XPC_P_ASR_REACTIVATE_UV) {
+			if (part->act_state == XPC_P_AS_INACTIVE)
+				xpc_activate_partition(part);
+			else
+				XPC_DEACTIVATE_PARTITION(part, xpReactivating);
+
+		} else if (act_state_req == XPC_P_ASR_DEACTIVATE_UV) {
+			XPC_DEACTIVATE_PARTITION(part, part->sn.uv.reason);
+
+		} else {
+			BUG();
+		}
+
+		spin_lock_irqsave(&xpc_activate_IRQ_rcvd_lock, irq_flags);
+		if (xpc_activate_IRQ_rcvd == 0)
+			break;
+	}
+	spin_unlock_irqrestore(&xpc_activate_IRQ_rcvd_lock, irq_flags);
+
+}
+
+static irqreturn_t
+xpc_handle_activate_IRQ_uv(int irq, void *dev_id)
+{
+	unsigned long irq_flags;
+	struct xpc_activate_mq_msghdr_uv *msg_hdr;
+	short partid;
+	struct xpc_partition *part;
+	struct xpc_partition_uv *part_uv;
+	struct xpc_openclose_args *args;
+	int wakeup_hb_checker = 0;
+
+	while ((msg_hdr = gru_get_next_message(xpc_activate_mq_uv)) != NULL) {
+
+		partid = msg_hdr->partid;
+		if (partid < 0 || partid >= XP_MAX_NPARTITIONS_UV) {
+			dev_err(xpc_part, "xpc_handle_activate_IRQ_uv() invalid"
+				"partid=0x%x passed in message\n", partid);
+			gru_free_message(xpc_activate_mq_uv, msg_hdr);
+			continue;
+		}
+		part = &xpc_partitions[partid];
+		part_uv = &part->sn.uv;
+
+		part_uv->remote_act_state = msg_hdr->act_state;
+
+		switch (msg_hdr->type) {
+		case XPC_ACTIVATE_MQ_MSG_SYNC_ACT_STATE_UV:
+			/* syncing of remote_act_state was just done above */
+			break;
+
+		case XPC_ACTIVATE_MQ_MSG_INC_HEARTBEAT_UV: {
+			struct xpc_activate_mq_msg_heartbeat_req_uv *msg;
+
+			msg = (struct xpc_activate_mq_msg_heartbeat_req_uv *)
+			    msg_hdr;
+			part_uv->heartbeat = msg->heartbeat;
+			break;
+		}
+		case XPC_ACTIVATE_MQ_MSG_OFFLINE_HEARTBEAT_UV: {
+			struct xpc_activate_mq_msg_heartbeat_req_uv *msg;
+
+			msg = (struct xpc_activate_mq_msg_heartbeat_req_uv *)
+			    msg_hdr;
+			part_uv->heartbeat = msg->heartbeat;
+			spin_lock_irqsave(&part_uv->flags_lock, irq_flags);
+			part_uv->flags |= XPC_P_HEARTBEAT_OFFLINE_UV;
+			spin_unlock_irqrestore(&part_uv->flags_lock, irq_flags);
+			break;
+		}
+		case XPC_ACTIVATE_MQ_MSG_ONLINE_HEARTBEAT_UV: {
+			struct xpc_activate_mq_msg_heartbeat_req_uv *msg;
+
+			msg = (struct xpc_activate_mq_msg_heartbeat_req_uv *)
+			    msg_hdr;
+			part_uv->heartbeat = msg->heartbeat;
+			spin_lock_irqsave(&part_uv->flags_lock, irq_flags);
+			part_uv->flags &= ~XPC_P_HEARTBEAT_OFFLINE_UV;
+			spin_unlock_irqrestore(&part_uv->flags_lock, irq_flags);
+			break;
+		}
+		case XPC_ACTIVATE_MQ_MSG_ACTIVATE_REQ_UV: {
+			struct xpc_activate_mq_msg_activate_req_uv *msg;
+
+			/*
+			 * ??? Do we deal here with ts_jiffies being different
+			 * ??? if act_state != XPC_P_AS_INACTIVE instead of
+			 * ??? below?
+			 */
+			msg = (struct xpc_activate_mq_msg_activate_req_uv *)
+			    msg_hdr;
+			spin_lock_irqsave(&xpc_activate_IRQ_rcvd_lock,
+					  irq_flags);
+			if (part_uv->act_state_req == 0)
+				xpc_activate_IRQ_rcvd++;
+			part_uv->act_state_req = XPC_P_ASR_ACTIVATE_UV;
+			part->remote_rp_pa = msg->rp_gpa; /* !!! _pa is _gpa */
+			part->remote_rp_ts_jiffies = msg_hdr->rp_ts_jiffies;
+			part_uv->remote_activate_mq_gpa = msg->activate_mq_gpa;
+			spin_unlock_irqrestore(&xpc_activate_IRQ_rcvd_lock,
+					       irq_flags);
+			wakeup_hb_checker++;
+			break;
+		}
+		case XPC_ACTIVATE_MQ_MSG_DEACTIVATE_REQ_UV: {
+			struct xpc_activate_mq_msg_deactivate_req_uv *msg;
+
+			msg = (struct xpc_activate_mq_msg_deactivate_req_uv *)
+			    msg_hdr;
+			spin_lock_irqsave(&xpc_activate_IRQ_rcvd_lock,
+					  irq_flags);
+			if (part_uv->act_state_req == 0)
+				xpc_activate_IRQ_rcvd++;
+			part_uv->act_state_req = XPC_P_ASR_DEACTIVATE_UV;
+			part_uv->reason = msg->reason;
+			spin_unlock_irqrestore(&xpc_activate_IRQ_rcvd_lock,
+					       irq_flags);
+			wakeup_hb_checker++;
+			break;
+		}
+		case XPC_ACTIVATE_MQ_MSG_CHCTL_CLOSEREQUEST_UV: {
+			struct xpc_activate_mq_msg_chctl_closerequest_uv *msg;
+
+			msg = (struct xpc_activate_mq_msg_chctl_closerequest_uv
+			    *)msg_hdr;
+			args = &part->remote_openclose_args[msg->ch_number];
+			args->reason = msg->reason;
+
+			spin_lock_irqsave(&part->chctl_lock, irq_flags);
+			part->chctl.flags[msg->ch_number] |=
+			    XPC_CHCTL_CLOSEREQUEST;
+			spin_unlock_irqrestore(&part->chctl_lock, irq_flags);
+
+			xpc_wakeup_channel_mgr(part);
+			break;
+		}
+		case XPC_ACTIVATE_MQ_MSG_CHCTL_CLOSEREPLY_UV: {
+			struct xpc_activate_mq_msg_chctl_closereply_uv *msg;
+
+			msg = (struct xpc_activate_mq_msg_chctl_closereply_uv *)
+			    msg_hdr;
+
+			spin_lock_irqsave(&part->chctl_lock, irq_flags);
+			part->chctl.flags[msg->ch_number] |=
+			    XPC_CHCTL_CLOSEREPLY;
+			spin_unlock_irqrestore(&part->chctl_lock, irq_flags);
+
+			xpc_wakeup_channel_mgr(part);
+			break;
+		}
+		case XPC_ACTIVATE_MQ_MSG_CHCTL_OPENREQUEST_UV: {
+			struct xpc_activate_mq_msg_chctl_openrequest_uv *msg;
+
+			msg = (struct xpc_activate_mq_msg_chctl_openrequest_uv
+			    *)msg_hdr;
+			args = &part->remote_openclose_args[msg->ch_number];
+			args->msg_size = msg->msg_size;
+			args->local_nentries = msg->local_nentries;
+
+			spin_lock_irqsave(&part->chctl_lock, irq_flags);
+			part->chctl.flags[msg->ch_number] |=
+			    XPC_CHCTL_OPENREQUEST;
+			spin_unlock_irqrestore(&part->chctl_lock, irq_flags);
+
+			xpc_wakeup_channel_mgr(part);
+			break;
+		}
+		case XPC_ACTIVATE_MQ_MSG_CHCTL_OPENREPLY_UV: {
+			struct xpc_activate_mq_msg_chctl_openreply_uv *msg;
+
+			msg = (struct xpc_activate_mq_msg_chctl_openreply_uv *)
+			    msg_hdr;
+			args = &part->remote_openclose_args[msg->ch_number];
+			args->remote_nentries = msg->remote_nentries;
+			args->local_nentries = msg->local_nentries;
+			args->local_msgqueue_pa = msg->local_notify_mq_gpa;
+
+			spin_lock_irqsave(&part->chctl_lock, irq_flags);
+			part->chctl.flags[msg->ch_number] |=
+			    XPC_CHCTL_OPENREPLY;
+			spin_unlock_irqrestore(&part->chctl_lock, irq_flags);
+
+			xpc_wakeup_channel_mgr(part);
+			break;
+		}
+		case XPC_ACTIVATE_MQ_MSG_MARK_ENGAGED_UV:
+			spin_lock_irqsave(&part_uv->flags_lock, irq_flags);
+			part_uv->flags |= XPC_P_ENGAGED_UV;
+			spin_unlock_irqrestore(&part_uv->flags_lock, irq_flags);
+			break;
+
+		case XPC_ACTIVATE_MQ_MSG_MARK_DISENGAGED_UV:
+			spin_lock_irqsave(&part_uv->flags_lock, irq_flags);
+			part_uv->flags &= ~XPC_P_ENGAGED_UV;
+			spin_unlock_irqrestore(&part_uv->flags_lock, irq_flags);
+			break;
+
+		default:
+			dev_err(xpc_part, "received unknown activate_mq msg "
+				"type=%d from partition=%d\n", msg_hdr->type,
+				partid);
+		}
+
+		if (msg_hdr->rp_ts_jiffies != part->remote_rp_ts_jiffies &&
+		    part->remote_rp_ts_jiffies != 0) {
+			/*
+			 * ??? Does what we do here need to be sensitive to
+			 * ??? act_state or remote_act_state?
+			 */
+			spin_lock_irqsave(&xpc_activate_IRQ_rcvd_lock,
+					  irq_flags);
+			if (part_uv->act_state_req == 0)
+				xpc_activate_IRQ_rcvd++;
+			part_uv->act_state_req = XPC_P_ASR_REACTIVATE_UV;
+			spin_unlock_irqrestore(&xpc_activate_IRQ_rcvd_lock,
+					       irq_flags);
+			wakeup_hb_checker++;
+		}
+
+		gru_free_message(xpc_activate_mq_uv, msg_hdr);
+	}
+
+	if (wakeup_hb_checker)
+		wake_up_interruptible(&xpc_activate_IRQ_wq);
+
+	return IRQ_HANDLED;
+}
+
+static enum xp_retval
+xpc_send_activate_IRQ_uv(struct xpc_partition *part, void *msg, size_t msg_size,
+			 int msg_type)
+{
+	struct xpc_activate_mq_msghdr_uv *msg_hdr = msg;
+
+	DBUG_ON(msg_size > XPC_ACTIVATE_MSG_SIZE_UV);
+
+	msg_hdr->type = msg_type;
+	msg_hdr->partid = XPC_PARTID(part);
+	msg_hdr->act_state = part->act_state;
+	msg_hdr->rp_ts_jiffies = xpc_rsvd_page->ts_jiffies;
+
+	/* ??? Is holding a spin_lock (ch->lock) during this call a bad idea? */
+	return xpc_send_gru_msg(part->sn.uv.remote_activate_mq_gpa, msg,
+				msg_size);
+}
+
+static void
+xpc_send_activate_IRQ_part_uv(struct xpc_partition *part, void *msg,
+			      size_t msg_size, int msg_type)
+{
+	enum xp_retval ret;
+
+	ret = xpc_send_activate_IRQ_uv(part, msg, msg_size, msg_type);
+	if (unlikely(ret != xpSuccess))
+		XPC_DEACTIVATE_PARTITION(part, ret);
+}
+
+static void
+xpc_send_activate_IRQ_ch_uv(struct xpc_channel *ch, unsigned long *irq_flags,
+			 void *msg, size_t msg_size, int msg_type)
+{
+	struct xpc_partition *part = &xpc_partitions[ch->number];
+	enum xp_retval ret;
+
+	ret = xpc_send_activate_IRQ_uv(part, msg, msg_size, msg_type);
+	if (unlikely(ret != xpSuccess)) {
+		if (irq_flags != NULL)
+			spin_unlock_irqrestore(&ch->lock, *irq_flags);
+
+		XPC_DEACTIVATE_PARTITION(part, ret);
+
+		if (irq_flags != NULL)
+			spin_lock_irqsave(&ch->lock, *irq_flags);
+	}
+}
+
+static void
+xpc_send_local_activate_IRQ_uv(struct xpc_partition *part, int act_state_req)
+{
+	unsigned long irq_flags;
+	struct xpc_partition_uv *part_uv = &part->sn.uv;
+
 	/*
 	 * !!! Make our side think that the remote parition sent an activate
-	 * !!! message our way. Also do what the activate IRQ handler would
+	 * !!! message our way by doing what the activate IRQ handler would
 	 * !!! do had one really been sent.
 	 */
+
+	spin_lock_irqsave(&xpc_activate_IRQ_rcvd_lock, irq_flags);
+	if (part_uv->act_state_req == 0)
+		xpc_activate_IRQ_rcvd++;
+	part_uv->act_state_req = act_state_req;
+	spin_unlock_irqrestore(&xpc_activate_IRQ_rcvd_lock, irq_flags);
+
+	wake_up_interruptible(&xpc_activate_IRQ_wq);
 }
 
 static enum xp_retval
-xpc_rsvd_page_init_uv(struct xpc_rsvd_page *rp)
+xpc_get_partition_rsvd_page_pa_uv(void *buf, u64 *cookie, unsigned long *rp_pa,
+				  size_t *len)
 {
-	/* !!! need to have established xpc_activate_mq earlier */
-	rp->sn.activate_mq_gpa = uv_gpa(xpc_activate_mq);
-	return xpSuccess;
+	/* !!! call the UV version of sn_partition_reserved_page_pa() */
+	return xpUnsupported;
+}
+
+static int
+xpc_setup_rsvd_page_sn_uv(struct xpc_rsvd_page *rp)
+{
+	rp->sn.activate_mq_gpa = uv_gpa(xpc_activate_mq_uv);
+	return 0;
+}
+
+static void
+xpc_send_heartbeat_uv(int msg_type)
+{
+	short partid;
+	struct xpc_partition *part;
+	struct xpc_activate_mq_msg_heartbeat_req_uv msg;
+
+	/*
+	 * !!! On uv we're broadcasting a heartbeat message every 5 seconds.
+	 * !!! Whereas on sn2 we're bte_copy'ng the heartbeat info every 20
+	 * !!! seconds. This is an increase in numalink traffic.
+	 * ??? Is this good?
+	 */
+
+	msg.heartbeat = atomic64_inc_return(&xpc_heartbeat_uv);
+
+	partid = find_first_bit(xpc_heartbeating_to_mask_uv,
+				XP_MAX_NPARTITIONS_UV);
+
+	while (partid < XP_MAX_NPARTITIONS_UV) {
+		part = &xpc_partitions[partid];
+
+		xpc_send_activate_IRQ_part_uv(part, &msg, sizeof(msg),
+					      msg_type);
+
+		partid = find_next_bit(xpc_heartbeating_to_mask_uv,
+				       XP_MAX_NPARTITIONS_UV, partid + 1);
+	}
 }
 
 static void
 xpc_increment_heartbeat_uv(void)
 {
-	/* !!! send heartbeat msg to xpc_heartbeating_to_mask partids */
+	xpc_send_heartbeat_uv(XPC_ACTIVATE_MQ_MSG_INC_HEARTBEAT_UV);
+}
+
+static void
+xpc_offline_heartbeat_uv(void)
+{
+	xpc_send_heartbeat_uv(XPC_ACTIVATE_MQ_MSG_OFFLINE_HEARTBEAT_UV);
+}
+
+static void
+xpc_online_heartbeat_uv(void)
+{
+	xpc_send_heartbeat_uv(XPC_ACTIVATE_MQ_MSG_ONLINE_HEARTBEAT_UV);
 }
 
 static void
 xpc_heartbeat_init_uv(void)
 {
+	atomic64_set(&xpc_heartbeat_uv, 0);
 	bitmap_zero(xpc_heartbeating_to_mask_uv, XP_MAX_NPARTITIONS_UV);
 	xpc_heartbeating_to_mask = &xpc_heartbeating_to_mask_uv[0];
 }
@@ -56,48 +543,94 @@ xpc_heartbeat_init_uv(void)
 static void
 xpc_heartbeat_exit_uv(void)
 {
-	/* !!! send heartbeat_offline msg to xpc_heartbeating_to_mask partids */
+	xpc_send_heartbeat_uv(XPC_ACTIVATE_MQ_MSG_OFFLINE_HEARTBEAT_UV);
+}
+
+static enum xp_retval
+xpc_get_remote_heartbeat_uv(struct xpc_partition *part)
+{
+	struct xpc_partition_uv *part_uv = &part->sn.uv;
+	enum xp_retval ret = xpNoHeartbeat;
+
+	if (part_uv->remote_act_state != XPC_P_AS_INACTIVE &&
+	    part_uv->remote_act_state != XPC_P_AS_DEACTIVATING) {
+
+		if (part_uv->heartbeat != part->last_heartbeat ||
+		    (part_uv->flags & XPC_P_HEARTBEAT_OFFLINE_UV)) {
+
+			part->last_heartbeat = part_uv->heartbeat;
+			ret = xpSuccess;
+		}
+	}
+	return ret;
 }
 
 static void
 xpc_request_partition_activation_uv(struct xpc_rsvd_page *remote_rp,
-				    unsigned long remote_rp_pa, int nasid)
+				    unsigned long remote_rp_gpa, int nasid)
 {
 	short partid = remote_rp->SAL_partid;
 	struct xpc_partition *part = &xpc_partitions[partid];
+	struct xpc_activate_mq_msg_activate_req_uv msg;
 
-/*
- * !!! Setup part structure with the bits of info we can glean from the rp:
- * !!!	part->remote_rp_pa = remote_rp_pa;
- * !!!	part->sn.uv.activate_mq_gpa = remote_rp->sn.activate_mq_gpa;
- */
+	part->remote_rp_pa = remote_rp_gpa; /* !!! _pa here is really _gpa */
+	part->remote_rp_ts_jiffies = remote_rp->ts_jiffies;
+	part->sn.uv.remote_activate_mq_gpa = remote_rp->sn.activate_mq_gpa;
+
+	/*
+	 * ??? Is it a good idea to make this conditional on what is
+	 * ??? potentially stale state information?
+	 */
+	if (part->sn.uv.remote_act_state == XPC_P_AS_INACTIVE) {
+		msg.rp_gpa = uv_gpa(xpc_rsvd_page);
+		msg.activate_mq_gpa = xpc_rsvd_page->sn.activate_mq_gpa;
+		xpc_send_activate_IRQ_part_uv(part, &msg, sizeof(msg),
+					   XPC_ACTIVATE_MQ_MSG_ACTIVATE_REQ_UV);
+	}
 
-	xpc_send_local_activate_IRQ_uv(part);
+	if (part->act_state == XPC_P_AS_INACTIVE)
+		xpc_send_local_activate_IRQ_uv(part, XPC_P_ASR_ACTIVATE_UV);
 }
 
 static void
 xpc_request_partition_reactivation_uv(struct xpc_partition *part)
 {
-	xpc_send_local_activate_IRQ_uv(part);
+	xpc_send_local_activate_IRQ_uv(part, XPC_P_ASR_ACTIVATE_UV);
+}
+
+static void
+xpc_request_partition_deactivation_uv(struct xpc_partition *part)
+{
+	struct xpc_activate_mq_msg_deactivate_req_uv msg;
+
+	/*
+	 * ??? Is it a good idea to make this conditional on what is
+	 * ??? potentially stale state information?
+	 */
+	if (part->sn.uv.remote_act_state != XPC_P_AS_DEACTIVATING &&
+	    part->sn.uv.remote_act_state != XPC_P_AS_INACTIVE) {
+
+		msg.reason = part->reason;
+		xpc_send_activate_IRQ_part_uv(part, &msg, sizeof(msg),
+					 XPC_ACTIVATE_MQ_MSG_DEACTIVATE_REQ_UV);
+	}
 }
 
 /*
- * Setup the infrastructure necessary to support XPartition Communication
- * between the specified remote partition and the local one.
+ * Setup the channel structures that are uv specific.
  */
 static enum xp_retval
-xpc_setup_infrastructure_uv(struct xpc_partition *part)
+xpc_setup_ch_structures_sn_uv(struct xpc_partition *part)
 {
 	/* !!! this function needs fleshing out */
 	return xpUnsupported;
 }
 
 /*
- * Teardown the infrastructure necessary to support XPartition Communication
- * between the specified remote partition and the local one.
+ * Teardown the channel structures that are uv specific.
  */
 static void
-xpc_teardown_infrastructure_uv(struct xpc_partition *part)
+xpc_teardown_ch_structures_sn_uv(struct xpc_partition *part)
 {
 	/* !!! this function needs fleshing out */
 	return;
@@ -106,15 +639,163 @@ xpc_teardown_infrastructure_uv(struct xpc_partition *part)
 static enum xp_retval
 xpc_make_first_contact_uv(struct xpc_partition *part)
 {
-	/* !!! this function needs fleshing out */
-	return xpUnsupported;
+	struct xpc_activate_mq_msg_uv msg;
+
+	/*
+	 * We send a sync msg to get the remote partition's remote_act_state
+	 * updated to our current act_state which at this point should
+	 * be XPC_P_AS_ACTIVATING.
+	 */
+	xpc_send_activate_IRQ_part_uv(part, &msg, sizeof(msg),
+				      XPC_ACTIVATE_MQ_MSG_SYNC_ACT_STATE_UV);
+
+	while (part->sn.uv.remote_act_state != XPC_P_AS_ACTIVATING) {
+
+		dev_dbg(xpc_part, "waiting to make first contact with "
+			"partition %d\n", XPC_PARTID(part));
+
+		/* wait a 1/4 of a second or so */
+		(void)msleep_interruptible(250);
+
+		if (part->act_state == XPC_P_AS_DEACTIVATING)
+			return part->reason;
+	}
+
+	return xpSuccess;
 }
 
 static u64
 xpc_get_chctl_all_flags_uv(struct xpc_partition *part)
 {
+	unsigned long irq_flags;
+	union xpc_channel_ctl_flags chctl;
+
+	spin_lock_irqsave(&part->chctl_lock, irq_flags);
+	chctl = part->chctl;
+	if (chctl.all_flags != 0)
+		part->chctl.all_flags = 0;
+
+	spin_unlock_irqrestore(&part->chctl_lock, irq_flags);
+	return chctl.all_flags;
+}
+
+static enum xp_retval
+xpc_setup_msg_structures_uv(struct xpc_channel *ch)
+{
+	/* !!! this function needs fleshing out */
+	return xpUnsupported;
+}
+
+static void
+xpc_teardown_msg_structures_uv(struct xpc_channel *ch)
+{
+	struct xpc_channel_uv *ch_uv = &ch->sn.uv;
+
+	ch_uv->remote_notify_mq_gpa = 0;
+
 	/* !!! this function needs fleshing out */
-	return 0UL;
+}
+
+static void
+xpc_send_chctl_closerequest_uv(struct xpc_channel *ch, unsigned long *irq_flags)
+{
+	struct xpc_activate_mq_msg_chctl_closerequest_uv msg;
+
+	msg.ch_number = ch->number;
+	msg.reason = ch->reason;
+	xpc_send_activate_IRQ_ch_uv(ch, irq_flags, &msg, sizeof(msg),
+				    XPC_ACTIVATE_MQ_MSG_CHCTL_CLOSEREQUEST_UV);
+}
+
+static void
+xpc_send_chctl_closereply_uv(struct xpc_channel *ch, unsigned long *irq_flags)
+{
+	struct xpc_activate_mq_msg_chctl_closereply_uv msg;
+
+	msg.ch_number = ch->number;
+	xpc_send_activate_IRQ_ch_uv(ch, irq_flags, &msg, sizeof(msg),
+				    XPC_ACTIVATE_MQ_MSG_CHCTL_CLOSEREPLY_UV);
+}
+
+static void
+xpc_send_chctl_openrequest_uv(struct xpc_channel *ch, unsigned long *irq_flags)
+{
+	struct xpc_activate_mq_msg_chctl_openrequest_uv msg;
+
+	msg.ch_number = ch->number;
+	msg.msg_size = ch->msg_size;
+	msg.local_nentries = ch->local_nentries;
+	xpc_send_activate_IRQ_ch_uv(ch, irq_flags, &msg, sizeof(msg),
+				    XPC_ACTIVATE_MQ_MSG_CHCTL_OPENREQUEST_UV);
+}
+
+static void
+xpc_send_chctl_openreply_uv(struct xpc_channel *ch, unsigned long *irq_flags)
+{
+	struct xpc_activate_mq_msg_chctl_openreply_uv msg;
+
+	msg.ch_number = ch->number;
+	msg.local_nentries = ch->local_nentries;
+	msg.remote_nentries = ch->remote_nentries;
+	msg.local_notify_mq_gpa = uv_gpa(xpc_notify_mq_uv);
+	xpc_send_activate_IRQ_ch_uv(ch, irq_flags, &msg, sizeof(msg),
+				    XPC_ACTIVATE_MQ_MSG_CHCTL_OPENREPLY_UV);
+}
+
+static void
+xpc_save_remote_msgqueue_pa_uv(struct xpc_channel *ch,
+			       unsigned long msgqueue_pa)
+{
+	ch->sn.uv.remote_notify_mq_gpa = msgqueue_pa;
+}
+
+static void
+xpc_indicate_partition_engaged_uv(struct xpc_partition *part)
+{
+	struct xpc_activate_mq_msg_uv msg;
+
+	xpc_send_activate_IRQ_part_uv(part, &msg, sizeof(msg),
+				      XPC_ACTIVATE_MQ_MSG_MARK_ENGAGED_UV);
+}
+
+static void
+xpc_indicate_partition_disengaged_uv(struct xpc_partition *part)
+{
+	struct xpc_activate_mq_msg_uv msg;
+
+	xpc_send_activate_IRQ_part_uv(part, &msg, sizeof(msg),
+				      XPC_ACTIVATE_MQ_MSG_MARK_DISENGAGED_UV);
+}
+
+static void
+xpc_assume_partition_disengaged_uv(short partid)
+{
+	struct xpc_partition_uv *part_uv = &xpc_partitions[partid].sn.uv;
+	unsigned long irq_flags;
+
+	spin_lock_irqsave(&part_uv->flags_lock, irq_flags);
+	part_uv->flags &= ~XPC_P_ENGAGED_UV;
+	spin_unlock_irqrestore(&part_uv->flags_lock, irq_flags);
+}
+
+static int
+xpc_partition_engaged_uv(short partid)
+{
+	return (xpc_partitions[partid].sn.uv.flags & XPC_P_ENGAGED_UV) != 0;
+}
+
+static int
+xpc_any_partition_engaged_uv(void)
+{
+	struct xpc_partition_uv *part_uv;
+	short partid;
+
+	for (partid = 0; partid < XP_MAX_NPARTITIONS_UV; partid++) {
+		part_uv = &xpc_partitions[partid].sn.uv;
+		if ((part_uv->flags & XPC_P_ENGAGED_UV) != 0)
+			return 1;
+	}
+	return 0;
 }
 
 static struct xpc_msg *
@@ -124,24 +805,64 @@ xpc_get_deliverable_msg_uv(struct xpc_channel *ch)
 	return NULL;
 }
 
-void
+int
 xpc_init_uv(void)
 {
-	xpc_rsvd_page_init = xpc_rsvd_page_init_uv;
+	xpc_setup_partitions_sn = xpc_setup_partitions_sn_uv;
+	xpc_process_activate_IRQ_rcvd = xpc_process_activate_IRQ_rcvd_uv;
+	xpc_get_partition_rsvd_page_pa = xpc_get_partition_rsvd_page_pa_uv;
+	xpc_setup_rsvd_page_sn = xpc_setup_rsvd_page_sn_uv;
 	xpc_increment_heartbeat = xpc_increment_heartbeat_uv;
+	xpc_offline_heartbeat = xpc_offline_heartbeat_uv;
+	xpc_online_heartbeat = xpc_online_heartbeat_uv;
 	xpc_heartbeat_init = xpc_heartbeat_init_uv;
 	xpc_heartbeat_exit = xpc_heartbeat_exit_uv;
+	xpc_get_remote_heartbeat = xpc_get_remote_heartbeat_uv;
+
 	xpc_request_partition_activation = xpc_request_partition_activation_uv;
 	xpc_request_partition_reactivation =
 	    xpc_request_partition_reactivation_uv;
-	xpc_setup_infrastructure = xpc_setup_infrastructure_uv;
-	xpc_teardown_infrastructure = xpc_teardown_infrastructure_uv;
+	xpc_request_partition_deactivation =
+	    xpc_request_partition_deactivation_uv;
+
+	xpc_setup_ch_structures_sn = xpc_setup_ch_structures_sn_uv;
+	xpc_teardown_ch_structures_sn = xpc_teardown_ch_structures_sn_uv;
+
 	xpc_make_first_contact = xpc_make_first_contact_uv;
+
 	xpc_get_chctl_all_flags = xpc_get_chctl_all_flags_uv;
+	xpc_send_chctl_closerequest = xpc_send_chctl_closerequest_uv;
+	xpc_send_chctl_closereply = xpc_send_chctl_closereply_uv;
+	xpc_send_chctl_openrequest = xpc_send_chctl_openrequest_uv;
+	xpc_send_chctl_openreply = xpc_send_chctl_openreply_uv;
+
+	xpc_save_remote_msgqueue_pa = xpc_save_remote_msgqueue_pa_uv;
+
+	xpc_setup_msg_structures = xpc_setup_msg_structures_uv;
+	xpc_teardown_msg_structures = xpc_teardown_msg_structures_uv;
+
+	xpc_indicate_partition_engaged = xpc_indicate_partition_engaged_uv;
+	xpc_indicate_partition_disengaged =
+	    xpc_indicate_partition_disengaged_uv;
+	xpc_assume_partition_disengaged = xpc_assume_partition_disengaged_uv;
+	xpc_partition_engaged = xpc_partition_engaged_uv;
+	xpc_any_partition_engaged = xpc_any_partition_engaged_uv;
+
 	xpc_get_deliverable_msg = xpc_get_deliverable_msg_uv;
+
+	/* ??? The cpuid argument's value is 0, is that what we want? */
+	/* !!! The irq argument's value isn't correct. */
+	xpc_activate_mq_uv = xpc_create_gru_mq_uv(XPC_ACTIVATE_MQ_SIZE_UV, 0, 0,
+						  xpc_handle_activate_IRQ_uv);
+	if (xpc_activate_mq_uv == NULL)
+		return -ENOMEM;
+
+	return 0;
 }
 
 void
 xpc_exit_uv(void)
 {
+	/* !!! The irq argument's value isn't correct. */
+	xpc_destroy_gru_mq_uv(xpc_activate_mq_uv, XPC_ACTIVATE_MQ_SIZE_UV, 0);
 }