[IA64] move XP and XPC to drivers/misc/sgi-xp

Move XPC and XPNET from arch/ia64/sn/kernel to drivers/misc/sgi-xp.

Signed-off-by: Dean Nelson <dcn@sgi.com>
Signed-off-by: Tony Luck <tony.luck@intel.com>

8 files changed, 1 insertions, 6110 deletions

diff --git a/arch/ia64/Kconfig b/arch/ia64/Kconfig
index ed21737..cd13e13 100644
--- a/arch/ia64/Kconfig
+++ b/arch/ia64/Kconfig
@@ -266,17 +266,6 @@ config IOSAPIC
 	depends on !IA64_HP_SIM
 	default y
 
-config IA64_SGI_SN_XP
-	tristate "Support communication between SGI SSIs"
-	depends on IA64_GENERIC || IA64_SGI_SN2
-	select IA64_UNCACHED_ALLOCATOR
-	help
-	  An SGI machine can be divided into multiple Single System
-	  Images which act independently of each other and have
-	  hardware based memory protection from the others.  Enabling
-	  this feature will allow for direct communication between SSIs
-	  based on a network adapter and DMA messaging.
-
 config FORCE_MAX_ZONEORDER
 	int "MAX_ORDER (11 - 17)"  if !HUGETLB_PAGE
 	range 11 17  if !HUGETLB_PAGE
diff --git a/arch/ia64/sn/kernel/Makefile b/arch/ia64/sn/kernel/Makefile
index 688a3c27..0591038 100644
--- a/arch/ia64/sn/kernel/Makefile
+++ b/arch/ia64/sn/kernel/Makefile
@@ -4,7 +4,7 @@
 # License.  See the file "COPYING" in the main directory of this archive
 # for more details.
 #
-# Copyright (C) 1999,2001-2006 Silicon Graphics, Inc.  All Rights Reserved.
+# Copyright (C) 1999,2001-2006,2008 Silicon Graphics, Inc.  All Rights Reserved.
 #
 
 EXTRA_CFLAGS += -Iarch/ia64/sn/include
@@ -15,9 +15,4 @@ obj-y				+= setup.o bte.o bte_error.o irq.o mca.o idle.o \
 				   sn2/
 obj-$(CONFIG_IA64_GENERIC)      += machvec.o
 obj-$(CONFIG_SGI_TIOCX)		+= tiocx.o
-obj-$(CONFIG_IA64_SGI_SN_XP)	+= xp.o
-xp-y				:= xp_main.o xp_nofault.o
-obj-$(CONFIG_IA64_SGI_SN_XP)	+= xpc.o
-xpc-y				:= xpc_main.o xpc_channel.o xpc_partition.o
-obj-$(CONFIG_IA64_SGI_SN_XP)	+= xpnet.o
 obj-$(CONFIG_PCI_MSI)		+= msi_sn.o
diff --git a/arch/ia64/sn/kernel/xp_main.c b/arch/ia64/sn/kernel/xp_main.c
deleted file mode 100644
index b7ea466..0000000
--- a/arch/ia64/sn/kernel/xp_main.c
+++ /dev/null
@@ -1,290 +0,0 @@
-/*
- * This file is subject to the terms and conditions of the GNU General Public
- * License.  See the file "COPYING" in the main directory of this archive
- * for more details.
- *
- * Copyright (c) 2004-2005 Silicon Graphics, Inc.  All Rights Reserved.
- */
-
-
-/*
- * Cross Partition (XP) base.
- *
- *	XP provides a base from which its users can interact
- *	with XPC, yet not be dependent on XPC.
- *
- */
-
-
-#include <linux/kernel.h>
-#include <linux/interrupt.h>
-#include <linux/module.h>
-#include <linux/mutex.h>
-#include <asm/sn/intr.h>
-#include <asm/sn/sn_sal.h>
-#include <asm/sn/xp.h>
-
-
-/*
- * Target of nofault PIO read.
- */
-u64 xp_nofault_PIOR_target;
-
-
-/*
- * xpc_registrations[] keeps track of xpc_connect()'s done by the kernel-level
- * users of XPC.
- */
-struct xpc_registration xpc_registrations[XPC_NCHANNELS];
-
-
-/*
- * Initialize the XPC interface to indicate that XPC isn't loaded.
- */
-static enum xpc_retval xpc_notloaded(void) { return xpcNotLoaded; }
-
-struct xpc_interface xpc_interface = {
-	(void (*)(int)) xpc_notloaded,
-	(void (*)(int)) xpc_notloaded,
-	(enum xpc_retval (*)(partid_t, int, u32, void **)) xpc_notloaded,
-	(enum xpc_retval (*)(partid_t, int, void *)) xpc_notloaded,
-	(enum xpc_retval (*)(partid_t, int, void *, xpc_notify_func, void *))
-							xpc_notloaded,
-	(void (*)(partid_t, int, void *)) xpc_notloaded,
-	(enum xpc_retval (*)(partid_t, void *)) xpc_notloaded
-};
-
-
-/*
- * XPC calls this when it (the XPC module) has been loaded.
- */
-void
-xpc_set_interface(void (*connect)(int),
-		void (*disconnect)(int),
-		enum xpc_retval (*allocate)(partid_t, int, u32, void **),
-		enum xpc_retval (*send)(partid_t, int, void *),
-		enum xpc_retval (*send_notify)(partid_t, int, void *,
-						xpc_notify_func, void *),
-		void (*received)(partid_t, int, void *),
-		enum xpc_retval (*partid_to_nasids)(partid_t, void *))
-{
-	xpc_interface.connect = connect;
-	xpc_interface.disconnect = disconnect;
-	xpc_interface.allocate = allocate;
-	xpc_interface.send = send;
-	xpc_interface.send_notify = send_notify;
-	xpc_interface.received = received;
-	xpc_interface.partid_to_nasids = partid_to_nasids;
-}
-
-
-/*
- * XPC calls this when it (the XPC module) is being unloaded.
- */
-void
-xpc_clear_interface(void)
-{
-	xpc_interface.connect = (void (*)(int)) xpc_notloaded;
-	xpc_interface.disconnect = (void (*)(int)) xpc_notloaded;
-	xpc_interface.allocate = (enum xpc_retval (*)(partid_t, int, u32,
-					void **)) xpc_notloaded;
-	xpc_interface.send = (enum xpc_retval (*)(partid_t, int, void *))
-					xpc_notloaded;
-	xpc_interface.send_notify = (enum xpc_retval (*)(partid_t, int, void *,
-				    xpc_notify_func, void *)) xpc_notloaded;
-	xpc_interface.received = (void (*)(partid_t, int, void *))
-					xpc_notloaded;
-	xpc_interface.partid_to_nasids = (enum xpc_retval (*)(partid_t, void *))
-					xpc_notloaded;
-}
-
-
-/*
- * Register for automatic establishment of a channel connection whenever
- * a partition comes up.
- *
- * Arguments:
- *
- *	ch_number - channel # to register for connection.
- *	func - function to call for asynchronous notification of channel
- *	       state changes (i.e., connection, disconnection, error) and
- *	       the arrival of incoming messages.
- *      key - pointer to optional user-defined value that gets passed back
- *	      to the user on any callouts made to func.
- *	payload_size - size in bytes of the XPC message's payload area which
- *		       contains a user-defined message. The user should make
- *		       this large enough to hold their largest message.
- *	nentries - max #of XPC message entries a message queue can contain.
- *		   The actual number, which is determined when a connection
- * 		   is established and may be less then requested, will be
- *		   passed to the user via the xpcConnected callout.
- *	assigned_limit - max number of kthreads allowed to be processing
- * 			 messages (per connection) at any given instant.
- *	idle_limit - max number of kthreads allowed to be idle at any given
- * 		     instant.
- */
-enum xpc_retval
-xpc_connect(int ch_number, xpc_channel_func func, void *key, u16 payload_size,
-		u16 nentries, u32 assigned_limit, u32 idle_limit)
-{
-	struct xpc_registration *registration;
-
-
-	DBUG_ON(ch_number < 0 || ch_number >= XPC_NCHANNELS);
-	DBUG_ON(payload_size == 0 || nentries == 0);
-	DBUG_ON(func == NULL);
-	DBUG_ON(assigned_limit == 0 || idle_limit > assigned_limit);
-
-	registration = &xpc_registrations[ch_number];
-
-	if (mutex_lock_interruptible(&registration->mutex) != 0) {
-		return xpcInterrupted;
-	}
-
-	/* if XPC_CHANNEL_REGISTERED(ch_number) */
-	if (registration->func != NULL) {
-		mutex_unlock(&registration->mutex);
-		return xpcAlreadyRegistered;
-	}
-
-	/* register the channel for connection */
-	registration->msg_size = XPC_MSG_SIZE(payload_size);
-	registration->nentries = nentries;
-	registration->assigned_limit = assigned_limit;
-	registration->idle_limit = idle_limit;
-	registration->key = key;
-	registration->func = func;
-
-	mutex_unlock(&registration->mutex);
-
-	xpc_interface.connect(ch_number);
-
-	return xpcSuccess;
-}
-
-
-/*
- * Remove the registration for automatic connection of the specified channel
- * when a partition comes up.
- *
- * Before returning this xpc_disconnect() will wait for all connections on the
- * specified channel have been closed/torndown. So the caller can be assured
- * that they will not be receiving any more callouts from XPC to their
- * function registered via xpc_connect().
- *
- * Arguments:
- *
- *	ch_number - channel # to unregister.
- */
-void
-xpc_disconnect(int ch_number)
-{
-	struct xpc_registration *registration;
-
-
-	DBUG_ON(ch_number < 0 || ch_number >= XPC_NCHANNELS);
-
-	registration = &xpc_registrations[ch_number];
-
-	/*
-	 * We've decided not to make this a down_interruptible(), since we
-	 * figured XPC's users will just turn around and call xpc_disconnect()
-	 * again anyways, so we might as well wait, if need be.
-	 */
-	mutex_lock(&registration->mutex);
-
-	/* if !XPC_CHANNEL_REGISTERED(ch_number) */
-	if (registration->func == NULL) {
-		mutex_unlock(&registration->mutex);
-		return;
-	}
-
-	/* remove the connection registration for the specified channel */
-	registration->func = NULL;
-	registration->key = NULL;
-	registration->nentries = 0;
-	registration->msg_size = 0;
-	registration->assigned_limit = 0;
-	registration->idle_limit = 0;
-
-	xpc_interface.disconnect(ch_number);
-
-	mutex_unlock(&registration->mutex);
-
-	return;
-}
-
-
-int __init
-xp_init(void)
-{
-	int ret, ch_number;
-	u64 func_addr = *(u64 *) xp_nofault_PIOR;
-	u64 err_func_addr = *(u64 *) xp_error_PIOR;
-
-
-	if (!ia64_platform_is("sn2")) {
-		return -ENODEV;
-	}
-
-	/*
-	 * Register a nofault code region which performs a cross-partition
-	 * PIO read. If the PIO read times out, the MCA handler will consume
-	 * the error and return to a kernel-provided instruction to indicate
-	 * an error. This PIO read exists because it is guaranteed to timeout
-	 * if the destination is down (AMO operations do not timeout on at
-	 * least some CPUs on Shubs <= v1.2, which unfortunately we have to
-	 * work around).
-	 */
-	if ((ret = sn_register_nofault_code(func_addr, err_func_addr,
-						err_func_addr, 1, 1)) != 0) {
-		printk(KERN_ERR "XP: can't register nofault code, error=%d\n",
-			ret);
-	}
-	/*
-	 * Setup the nofault PIO read target. (There is no special reason why
-	 * SH_IPI_ACCESS was selected.)
-	 */
-	if (is_shub2()) {
-		xp_nofault_PIOR_target = SH2_IPI_ACCESS0;
-	} else {
-		xp_nofault_PIOR_target = SH1_IPI_ACCESS;
-	}
-
-	/* initialize the connection registration mutex */
-	for (ch_number = 0; ch_number < XPC_NCHANNELS; ch_number++) {
-		mutex_init(&xpc_registrations[ch_number].mutex);
-	}
-
-	return 0;
-}
-module_init(xp_init);
-
-
-void __exit
-xp_exit(void)
-{
-	u64 func_addr = *(u64 *) xp_nofault_PIOR;
-	u64 err_func_addr = *(u64 *) xp_error_PIOR;
-
-
-	/* unregister the PIO read nofault code region */
-	(void) sn_register_nofault_code(func_addr, err_func_addr,
-					err_func_addr, 1, 0);
-}
-module_exit(xp_exit);
-
-
-MODULE_AUTHOR("Silicon Graphics, Inc.");
-MODULE_DESCRIPTION("Cross Partition (XP) base");
-MODULE_LICENSE("GPL");
-
-EXPORT_SYMBOL(xp_nofault_PIOR);
-EXPORT_SYMBOL(xp_nofault_PIOR_target);
-EXPORT_SYMBOL(xpc_registrations);
-EXPORT_SYMBOL(xpc_interface);
-EXPORT_SYMBOL(xpc_clear_interface);
-EXPORT_SYMBOL(xpc_set_interface);
-EXPORT_SYMBOL(xpc_connect);
-EXPORT_SYMBOL(xpc_disconnect);
-
diff --git a/arch/ia64/sn/kernel/xp_nofault.S b/arch/ia64/sn/kernel/xp_nofault.S
deleted file mode 100644
index 98e7c7d..0000000
--- a/arch/ia64/sn/kernel/xp_nofault.S
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * This file is subject to the terms and conditions of the GNU General Public
- * License.  See the file "COPYING" in the main directory of this archive
- * for more details.
- *
- * Copyright (c) 2004-2007 Silicon Graphics, Inc.  All Rights Reserved.
- */
-
-
-/*
- * The xp_nofault_PIOR function takes a pointer to a remote PIO register
- * and attempts to load and consume a value from it.  This function
- * will be registered as a nofault code block.  In the event that the
- * PIO read fails, the MCA handler will force the error to look
- * corrected and vector to the xp_error_PIOR which will return an error.
- *
- * The definition of "consumption" and the time it takes for an MCA
- * to surface is processor implementation specific.  This code
- * is sufficient on Itanium through the Montvale processor family.
- * It may need to be adjusted for future processor implementations.
- *
- *	extern int xp_nofault_PIOR(void *remote_register);
- */
-
-	.global xp_nofault_PIOR
-xp_nofault_PIOR:
-	mov	r8=r0			// Stage a success return value
-	ld8.acq	r9=[r32];;		// PIO Read the specified register
-	adds	r9=1,r9;;		// Add to force consumption
-	srlz.i;;			// Allow time for MCA to surface
-	br.ret.sptk.many b0;;		// Return success
-
-	.global xp_error_PIOR
-xp_error_PIOR:
-	mov	r8=1			// Return value of 1
-	br.ret.sptk.many b0;;		// Return failure
diff --git a/arch/ia64/sn/kernel/xpc_channel.c b/arch/ia64/sn/kernel/xpc_channel.c
deleted file mode 100644
index 44ccc0d..0000000
--- a/arch/ia64/sn/kernel/xpc_channel.c
+++ /dev/null
@@ -1,2379 +0,0 @@
-/*
- * This file is subject to the terms and conditions of the GNU General Public
- * License.  See the file "COPYING" in the main directory of this archive
- * for more details.
- *
- * Copyright (c) 2004-2006 Silicon Graphics, Inc.  All Rights Reserved.
- */
-
-
-/*
- * Cross Partition Communication (XPC) channel support.
- *
- *	This is the part of XPC that manages the channels and
- *	sends/receives messages across them to/from other partitions.
- *
- */
-
-
-#include <linux/kernel.h>
-#include <linux/init.h>
-#include <linux/sched.h>
-#include <linux/cache.h>
-#include <linux/interrupt.h>
-#include <linux/mutex.h>
-#include <linux/completion.h>
-#include <asm/sn/bte.h>
-#include <asm/sn/sn_sal.h>
-#include <asm/sn/xpc.h>
-
-
-/*
- * Guarantee that the kzalloc'd memory is cacheline aligned.
- */
-static void *
-xpc_kzalloc_cacheline_aligned(size_t size, gfp_t flags, void **base)
-{
-	/* see if kzalloc will give us cachline aligned memory by default */
-	*base = kzalloc(size, flags);
-	if (*base == NULL) {
-		return NULL;
-	}
-	if ((u64) *base == L1_CACHE_ALIGN((u64) *base)) {
-		return *base;
-	}
-	kfree(*base);
-
-	/* nope, we'll have to do it ourselves */
-	*base = kzalloc(size + L1_CACHE_BYTES, flags);
-	if (*base == NULL) {
-		return NULL;
-	}
-	return (void *) L1_CACHE_ALIGN((u64) *base);
-}
-
-
-/*
- * Set up the initial values for the XPartition Communication channels.
- */
-static void
-xpc_initialize_channels(struct xpc_partition *part, partid_t partid)
-{
-	int ch_number;
-	struct xpc_channel *ch;
-
-
-	for (ch_number = 0; ch_number < part->nchannels; ch_number++) {
-		ch = &part->channels[ch_number];
-
-		ch->partid = partid;
-		ch->number = ch_number;
-		ch->flags = XPC_C_DISCONNECTED;
-
-		ch->local_GP = &part->local_GPs[ch_number];
-		ch->local_openclose_args =
-					&part->local_openclose_args[ch_number];
-
-		atomic_set(&ch->kthreads_assigned, 0);
-		atomic_set(&ch->kthreads_idle, 0);
-		atomic_set(&ch->kthreads_active, 0);
-
-		atomic_set(&ch->references, 0);
-		atomic_set(&ch->n_to_notify, 0);
-
-		spin_lock_init(&ch->lock);
-		mutex_init(&ch->msg_to_pull_mutex);
-		init_completion(&ch->wdisconnect_wait);
-
-		atomic_set(&ch->n_on_msg_allocate_wq, 0);
-		init_waitqueue_head(&ch->msg_allocate_wq);
-		init_waitqueue_head(&ch->idle_wq);
-	}
-}
-
-
-/*
- * Setup the infrastructure necessary to support XPartition Communication
- * between the specified remote partition and the local one.
- */
-enum xpc_retval
-xpc_setup_infrastructure(struct xpc_partition *part)
-{
-	int ret, cpuid;
-	struct timer_list *timer;
-	partid_t partid = XPC_PARTID(part);
-
-
-	/*
-	 * Zero out MOST of the entry for this partition. Only the fields
-	 * starting with `nchannels' will be zeroed. The preceding fields must
-	 * remain `viable' across partition ups and downs, since they may be
-	 * referenced during this memset() operation.
-	 */
-	memset(&part->nchannels, 0, sizeof(struct xpc_partition) -
-				offsetof(struct xpc_partition, nchannels));
-
-	/*
-	 * Allocate all of the channel structures as a contiguous chunk of
-	 * memory.
-	 */
-	part->channels = kzalloc(sizeof(struct xpc_channel) * XPC_NCHANNELS,
-								GFP_KERNEL);
-	if (part->channels == NULL) {
-		dev_err(xpc_chan, "can't get memory for channels\n");
-		return xpcNoMemory;
-	}
-
-	part->nchannels = XPC_NCHANNELS;
-
-
-	/* allocate all the required GET/PUT values */
-
-	part->local_GPs = xpc_kzalloc_cacheline_aligned(XPC_GP_SIZE,
-					GFP_KERNEL, &part->local_GPs_base);
-	if (part->local_GPs == NULL) {
-		kfree(part->channels);
-		part->channels = NULL;
-		dev_err(xpc_chan, "can't get memory for local get/put "
-			"values\n");
-		return xpcNoMemory;
-	}
-
-	part->remote_GPs = xpc_kzalloc_cacheline_aligned(XPC_GP_SIZE,
-					GFP_KERNEL, &part->remote_GPs_base);
-	if (part->remote_GPs == NULL) {
-		dev_err(xpc_chan, "can't get memory for remote get/put "
-			"values\n");
-		kfree(part->local_GPs_base);
-		part->local_GPs = NULL;
-		kfree(part->channels);
-		part->channels = NULL;
-		return xpcNoMemory;
-	}
-
-
-	/* allocate all the required open and close args */
-
-	part->local_openclose_args = xpc_kzalloc_cacheline_aligned(
-					XPC_OPENCLOSE_ARGS_SIZE, GFP_KERNEL,
-					&part->local_openclose_args_base);
-	if (part->local_openclose_args == NULL) {
-		dev_err(xpc_chan, "can't get memory for local connect args\n");
-		kfree(part->remote_GPs_base);
-		part->remote_GPs = NULL;
-		kfree(part->local_GPs_base);
-		part->local_GPs = NULL;
-		kfree(part->channels);
-		part->channels = NULL;
-		return xpcNoMemory;
-	}
-
-	part->remote_openclose_args = xpc_kzalloc_cacheline_aligned(
-					XPC_OPENCLOSE_ARGS_SIZE, GFP_KERNEL,
-					&part->remote_openclose_args_base);
-	if (part->remote_openclose_args == NULL) {
-		dev_err(xpc_chan, "can't get memory for remote connect args\n");
-		kfree(part->local_openclose_args_base);
-		part->local_openclose_args = NULL;
-		kfree(part->remote_GPs_base);
-		part->remote_GPs = NULL;
-		kfree(part->local_GPs_base);
-		part->local_GPs = NULL;
-		kfree(part->channels);
-		part->channels = NULL;
-		return xpcNoMemory;
-	}
-
-
-	xpc_initialize_channels(part, partid);
-
-	atomic_set(&part->nchannels_active, 0);
-	atomic_set(&part->nchannels_engaged, 0);
-
-
-	/* local_IPI_amo were set to 0 by an earlier memset() */
-
-	/* Initialize this partitions AMO_t structure */
-	part->local_IPI_amo_va = xpc_IPI_init(partid);
-
-	spin_lock_init(&part->IPI_lock);
-
-	atomic_set(&part->channel_mgr_requests, 1);
-	init_waitqueue_head(&part->channel_mgr_wq);
-
-	sprintf(part->IPI_owner, "xpc%02d", partid);
-	ret = request_irq(SGI_XPC_NOTIFY, xpc_notify_IRQ_handler, IRQF_SHARED,
-				part->IPI_owner, (void *) (u64) partid);
-	if (ret != 0) {
-		dev_err(xpc_chan, "can't register NOTIFY IRQ handler, "
-			"errno=%d\n", -ret);
-		kfree(part->remote_openclose_args_base);
-		part->remote_openclose_args = NULL;
-		kfree(part->local_openclose_args_base);
-		part->local_openclose_args = NULL;
-		kfree(part->remote_GPs_base);
-		part->remote_GPs = NULL;
-		kfree(part->local_GPs_base);
-		part->local_GPs = NULL;
-		kfree(part->channels);
-		part->channels = NULL;
-		return xpcLackOfResources;
-	}
-
-	/* Setup a timer to check for dropped IPIs */
-	timer = &part->dropped_IPI_timer;
-	init_timer(timer);
-	timer->function = (void (*)(unsigned long)) xpc_dropped_IPI_check;
-	timer->data = (unsigned long) part;
-	timer->expires = jiffies + XPC_P_DROPPED_IPI_WAIT;
-	add_timer(timer);
-
-	/*
-	 * With the setting of the partition setup_state to XPC_P_SETUP, we're
-	 * declaring that this partition is ready to go.
-	 */
-	part->setup_state = XPC_P_SETUP;
-
-
-	/*
-	 * Setup the per partition specific variables required by the
-	 * remote partition to establish channel connections with us.
-	 *
-	 * The setting of the magic # indicates that these per partition
-	 * specific variables are ready to be used.
-	 */
-	xpc_vars_part[partid].GPs_pa = __pa(part->local_GPs);
-	xpc_vars_part[partid].openclose_args_pa =
-					__pa(part->local_openclose_args);
-	xpc_vars_part[partid].IPI_amo_pa = __pa(part->local_IPI_amo_va);
-	cpuid = raw_smp_processor_id();	/* any CPU in this partition will do */
-	xpc_vars_part[partid].IPI_nasid = cpuid_to_nasid(cpuid);
-	xpc_vars_part[partid].IPI_phys_cpuid = cpu_physical_id(cpuid);
-	xpc_vars_part[partid].nchannels = part->nchannels;
-	xpc_vars_part[partid].magic = XPC_VP_MAGIC1;
-
-	return xpcSuccess;
-}
-
-
-/*
- * Create a wrapper that hides the underlying mechanism for pulling a cacheline
- * (or multiple cachelines) from a remote partition.
- *
- * src must be a cacheline aligned physical address on the remote partition.
- * dst must be a cacheline aligned virtual address on this partition.
- * cnt must be an cacheline sized
- */
-static enum xpc_retval
-xpc_pull_remote_cachelines(struct xpc_partition *part, void *dst,
-				const void *src, size_t cnt)
-{
-	bte_result_t bte_ret;
-
-
-	DBUG_ON((u64) src != L1_CACHE_ALIGN((u64) src));
-	DBUG_ON((u64) dst != L1_CACHE_ALIGN((u64) dst));
-	DBUG_ON(cnt != L1_CACHE_ALIGN(cnt));
-
-	if (part->act_state == XPC_P_DEACTIVATING) {
-		return part->reason;
-	}
-
-	bte_ret = xp_bte_copy((u64) src, (u64) dst, (u64) cnt,
-					(BTE_NORMAL | BTE_WACQUIRE), NULL);
-	if (bte_ret == BTE_SUCCESS) {
-		return xpcSuccess;
-	}
-
-	dev_dbg(xpc_chan, "xp_bte_copy() from partition %d failed, ret=%d\n",
-		XPC_PARTID(part), bte_ret);
-
-	return xpc_map_bte_errors(bte_ret);
-}
-
-
-/*
- * Pull the remote per partition specific variables from the specified
- * partition.
- */
-enum xpc_retval
-xpc_pull_remote_vars_part(struct xpc_partition *part)
-{
-	u8 buffer[L1_CACHE_BYTES * 2];
-	struct xpc_vars_part *pulled_entry_cacheline =
-			(struct xpc_vars_part *) L1_CACHE_ALIGN((u64) buffer);
-	struct xpc_vars_part *pulled_entry;
-	u64 remote_entry_cacheline_pa, remote_entry_pa;
-	partid_t partid = XPC_PARTID(part);
-	enum xpc_retval ret;
-
-
-	/* pull the cacheline that contains the variables we're interested in */
-
-	DBUG_ON(part->remote_vars_part_pa !=
-				L1_CACHE_ALIGN(part->remote_vars_part_pa));
-	DBUG_ON(sizeof(struct xpc_vars_part) != L1_CACHE_BYTES / 2);
-
-	remote_entry_pa = part->remote_vars_part_pa +
-			sn_partition_id * sizeof(struct xpc_vars_part);
-
-	remote_entry_cacheline_pa = (remote_entry_pa & ~(L1_CACHE_BYTES - 1));
-
-	pulled_entry = (struct xpc_vars_part *) ((u64) pulled_entry_cacheline +
-				(remote_entry_pa & (L1_CACHE_BYTES - 1)));
-
-	ret = xpc_pull_remote_cachelines(part, pulled_entry_cacheline,
-					(void *) remote_entry_cacheline_pa,
-					L1_CACHE_BYTES);
-	if (ret != xpcSuccess) {
-		dev_dbg(xpc_chan, "failed to pull XPC vars_part from "
-			"partition %d, ret=%d\n", partid, ret);
-		return ret;
-	}
-
-
-	/* see if they've been set up yet */
-
-	if (pulled_entry->magic != XPC_VP_MAGIC1 &&
-				pulled_entry->magic != XPC_VP_MAGIC2) {
-
-		if (pulled_entry->magic != 0) {
-			dev_dbg(xpc_chan, "partition %d's XPC vars_part for "
-				"partition %d has bad magic value (=0x%lx)\n",
-				partid, sn_partition_id, pulled_entry->magic);
-			return xpcBadMagic;
-		}
-
-		/* they've not been initialized yet */
-		return xpcRetry;
-	}
-
-	if (xpc_vars_part[partid].magic == XPC_VP_MAGIC1) {
-
-		/* validate the variables */
-
-		if (pulled_entry->GPs_pa == 0 ||
-				pulled_entry->openclose_args_pa == 0 ||
-					pulled_entry->IPI_amo_pa == 0) {
-
-			dev_err(xpc_chan, "partition %d's XPC vars_part for "
-				"partition %d are not valid\n", partid,
-				sn_partition_id);
-			return xpcInvalidAddress;
-		}
-
-		/* the variables we imported look to be valid */
-
-		part->remote_GPs_pa = pulled_entry->GPs_pa;
-		part->remote_openclose_args_pa =
-					pulled_entry->openclose_args_pa;
-		part->remote_IPI_amo_va =
-				      (AMO_t *) __va(pulled_entry->IPI_amo_pa);
-		part->remote_IPI_nasid = pulled_entry->IPI_nasid;
-		part->remote_IPI_phys_cpuid = pulled_entry->IPI_phys_cpuid;
-
-		if (part->nchannels > pulled_entry->nchannels) {
-			part->nchannels = pulled_entry->nchannels;
-		}
-
-		/* let the other side know that we've pulled their variables */
-
-		xpc_vars_part[partid].magic = XPC_VP_MAGIC2;
-	}
-
-	if (pulled_entry->magic == XPC_VP_MAGIC1) {
-		return xpcRetry;
-	}
-
-	return xpcSuccess;
-}
-
-
-/*
- * Get the IPI flags and pull the openclose args and/or remote GPs as needed.
- */
-static u64
-xpc_get_IPI_flags(struct xpc_partition *part)
-{
-	unsigned long irq_flags;
-	u64 IPI_amo;
-	enum xpc_retval ret;
-
-
-	/*
-	 * See if there are any IPI flags to be handled.
-	 */
-
-	spin_lock_irqsave(&part->IPI_lock, irq_flags);
-	if ((IPI_amo = part->local_IPI_amo) != 0) {
-		part->local_IPI_amo = 0;
-	}
-	spin_unlock_irqrestore(&part->IPI_lock, irq_flags);
-
-
-	if (XPC_ANY_OPENCLOSE_IPI_FLAGS_SET(IPI_amo)) {
-		ret = xpc_pull_remote_cachelines(part,
-					part->remote_openclose_args,
-					(void *) part->remote_openclose_args_pa,
-					XPC_OPENCLOSE_ARGS_SIZE);
-		if (ret != xpcSuccess) {
-			XPC_DEACTIVATE_PARTITION(part, ret);
-
-			dev_dbg(xpc_chan, "failed to pull openclose args from "
-				"partition %d, ret=%d\n", XPC_PARTID(part),
-				ret);
-
-			/* don't bother processing IPIs anymore */
-			IPI_amo = 0;
-		}
-	}
-
-	if (XPC_ANY_MSG_IPI_FLAGS_SET(IPI_amo)) {
-		ret = xpc_pull_remote_cachelines(part, part->remote_GPs,
-						(void *) part->remote_GPs_pa,
-						XPC_GP_SIZE);
-		if (ret != xpcSuccess) {
-			XPC_DEACTIVATE_PARTITION(part, ret);
-
-			dev_dbg(xpc_chan, "failed to pull GPs from partition "
-				"%d, ret=%d\n", XPC_PARTID(part), ret);
-
-			/* don't bother processing IPIs anymore */
-			IPI_amo = 0;
-		}
-	}
-
-	return IPI_amo;
-}
-
-
-/*
- * Allocate the local message queue and the notify queue.
- */
-static enum xpc_retval
-xpc_allocate_local_msgqueue(struct xpc_channel *ch)
-{
-	unsigned long irq_flags;
-	int nentries;
-	size_t nbytes;
-
-
-	// >>> may want to check for ch->flags & XPC_C_DISCONNECTING between
-	// >>> iterations of the for-loop, bail if set?
-
-	// >>> should we impose a minimum #of entries? like 4 or 8?
-	for (nentries = ch->local_nentries; nentries > 0; nentries--) {
-
-		nbytes = nentries * ch->msg_size;
-		ch->local_msgqueue = xpc_kzalloc_cacheline_aligned(nbytes,
-						GFP_KERNEL,
-						&ch->local_msgqueue_base);
-		if (ch->local_msgqueue == NULL) {
-			continue;
-		}
-
-		nbytes = nentries * sizeof(struct xpc_notify);
-		ch->notify_queue = kzalloc(nbytes, GFP_KERNEL);
-		if (ch->notify_queue == NULL) {
-			kfree(ch->local_msgqueue_base);
-			ch->local_msgqueue = NULL;
-			continue;
-		}
-
-		spin_lock_irqsave(&ch->lock, irq_flags);
-		if (nentries < ch->local_nentries) {
-			dev_dbg(xpc_chan, "nentries=%d local_nentries=%d, "
-				"partid=%d, channel=%d\n", nentries,
-				ch->local_nentries, ch->partid, ch->number);
-
-			ch->local_nentries = nentries;
-		}
-		spin_unlock_irqrestore(&ch->lock, irq_flags);
-		return xpcSuccess;
-	}
-
-	dev_dbg(xpc_chan, "can't get memory for local message queue and notify "
-		"queue, partid=%d, channel=%d\n", ch->partid, ch->number);
-	return xpcNoMemory;
-}
-
-
-/*
- * Allocate the cached remote message queue.
- */
-static enum xpc_retval
-xpc_allocate_remote_msgqueue(struct xpc_channel *ch)
-{
-	unsigned long irq_flags;
-	int nentries;
-	size_t nbytes;
-
-
-	DBUG_ON(ch->remote_nentries <= 0);
-
-	// >>> may want to check for ch->flags & XPC_C_DISCONNECTING between
-	// >>> iterations of the for-loop, bail if set?
-
-	// >>> should we impose a minimum #of entries? like 4 or 8?
-	for (nentries = ch->remote_nentries; nentries > 0; nentries--) {
-
-		nbytes = nentries * ch->msg_size;
-		ch->remote_msgqueue = xpc_kzalloc_cacheline_aligned(nbytes,
-						GFP_KERNEL,
-						&ch->remote_msgqueue_base);
-		if (ch->remote_msgqueue == NULL) {
-			continue;
-		}
-
-		spin_lock_irqsave(&ch->lock, irq_flags);
-		if (nentries < ch->remote_nentries) {
-			dev_dbg(xpc_chan, "nentries=%d remote_nentries=%d, "
-				"partid=%d, channel=%d\n", nentries,
-				ch->remote_nentries, ch->partid, ch->number);
-
-			ch->remote_nentries = nentries;
-		}
-		spin_unlock_irqrestore(&ch->lock, irq_flags);
-		return xpcSuccess;
-	}
-
-	dev_dbg(xpc_chan, "can't get memory for cached remote message queue, "
-		"partid=%d, channel=%d\n", ch->partid, ch->number);
-	return xpcNoMemory;
-}
-
-
-/*
- * Allocate message queues and other stuff associated with a channel.
- *
- * Note: Assumes all of the channel sizes are filled in.
- */
-static enum xpc_retval
-xpc_allocate_msgqueues(struct xpc_channel *ch)
-{
-	unsigned long irq_flags;
-	enum xpc_retval ret;
-
-
-	DBUG_ON(ch->flags & XPC_C_SETUP);
-
-	if ((ret = xpc_allocate_local_msgqueue(ch)) != xpcSuccess) {
-		return ret;
-	}
-
-	if ((ret = xpc_allocate_remote_msgqueue(ch)) != xpcSuccess) {
-		kfree(ch->local_msgqueue_base);
-		ch->local_msgqueue = NULL;
-		kfree(ch->notify_queue);
-		ch->notify_queue = NULL;
-		return ret;
-	}
-
-	spin_lock_irqsave(&ch->lock, irq_flags);
-	ch->flags |= XPC_C_SETUP;
-	spin_unlock_irqrestore(&ch->lock, irq_flags);
-
-	return xpcSuccess;
-}
-
-
-/*
- * Process a connect message from a remote partition.
- *
- * Note: xpc_process_connect() is expecting to be called with the
- * spin_lock_irqsave held and will leave it locked upon return.
- */
-static void
-xpc_process_connect(struct xpc_channel *ch, unsigned long *irq_flags)
-{
-	enum xpc_retval ret;
-
-
-	DBUG_ON(!spin_is_locked(&ch->lock));
-
-	if (!(ch->flags & XPC_C_OPENREQUEST) ||
-				!(ch->flags & XPC_C_ROPENREQUEST)) {
-		/* nothing more to do for now */
-		return;
-	}
-	DBUG_ON(!(ch->flags & XPC_C_CONNECTING));
-
-	if (!(ch->flags & XPC_C_SETUP)) {
-		spin_unlock_irqrestore(&ch->lock, *irq_flags);
-		ret = xpc_allocate_msgqueues(ch);
-		spin_lock_irqsave(&ch->lock, *irq_flags);
-
-		if (ret != xpcSuccess) {
-			XPC_DISCONNECT_CHANNEL(ch, ret, irq_flags);
-		}
-		if (ch->flags & (XPC_C_CONNECTED | XPC_C_DISCONNECTING)) {
-			return;
-		}
-
-		DBUG_ON(!(ch->flags & XPC_C_SETUP));
-		DBUG_ON(ch->local_msgqueue == NULL);
-		DBUG_ON(ch->remote_msgqueue == NULL);
-	}
-
-	if (!(ch->flags & XPC_C_OPENREPLY)) {
-		ch->flags |= XPC_C_OPENREPLY;
-		xpc_IPI_send_openreply(ch, irq_flags);
-	}
-
-	if (!(ch->flags & XPC_C_ROPENREPLY)) {
-		return;
-	}
-
-	DBUG_ON(ch->remote_msgqueue_pa == 0);
-
-	ch->flags = (XPC_C_CONNECTED | XPC_C_SETUP);	/* clear all else */
-
-	dev_info(xpc_chan, "channel %d to partition %d connected\n",
-		ch->number, ch->partid);
-
-	spin_unlock_irqrestore(&ch->lock, *irq_flags);
-	xpc_create_kthreads(ch, 1, 0);
-	spin_lock_irqsave(&ch->lock, *irq_flags);
-}
-
-
-/*
- * Notify those who wanted to be notified upon delivery of their message.
- */
-static void
-xpc_notify_senders(struct xpc_channel *ch, enum xpc_retval reason, s64 put)
-{
-	struct xpc_notify *notify;
-	u8 notify_type;
-	s64 get = ch->w_remote_GP.get - 1;
-
-
-	while (++get < put && atomic_read(&ch->n_to_notify) > 0) {
-
-		notify = &ch->notify_queue[get % ch->local_nentries];
-
-		/*
-		 * See if the notify entry indicates it was associated with
-		 * a message who's sender wants to be notified. It is possible
-		 * that it is, but someone else is doing or has done the
-		 * notification.
-		 */
-		notify_type = notify->type;
-		if (notify_type == 0 ||
-				cmpxchg(&notify->type, notify_type, 0) !=
-								notify_type) {
-			continue;
-		}
-
-		DBUG_ON(notify_type != XPC_N_CALL);
-
-		atomic_dec(&ch->n_to_notify);
-
-		if (notify->func != NULL) {
-			dev_dbg(xpc_chan, "notify->func() called, notify=0x%p, "
-				"msg_number=%ld, partid=%d, channel=%d\n",
-				(void *) notify, get, ch->partid, ch->number);
-
-			notify->func(reason, ch->partid, ch->number,
-								notify->key);
-
-			dev_dbg(xpc_chan, "notify->func() returned, "
-				"notify=0x%p, msg_number=%ld, partid=%d, "
-				"channel=%d\n", (void *) notify, get,
-				ch->partid, ch->number);
-		}
-	}
-}
-
-
-/*
- * Free up message queues and other stuff that were allocated for the specified
- * channel.
- *
- * Note: ch->reason and ch->reason_line are left set for debugging purposes,
- * they're cleared when XPC_C_DISCONNECTED is cleared.
- */
-static void
-xpc_free_msgqueues(struct xpc_channel *ch)
-{
-	DBUG_ON(!spin_is_locked(&ch->lock));
-	DBUG_ON(atomic_read(&ch->n_to_notify) != 0);
-
-	ch->remote_msgqueue_pa = 0;
-	ch->func = NULL;
-	ch->key = NULL;
-	ch->msg_size = 0;
-	ch->local_nentries = 0;
-	ch->remote_nentries = 0;
-	ch->kthreads_assigned_limit = 0;
-	ch->kthreads_idle_limit = 0;
-
-	ch->local_GP->get = 0;
-	ch->local_GP->put = 0;
-	ch->remote_GP.get = 0;
-	ch->remote_GP.put = 0;
-	ch->w_local_GP.get = 0;
-	ch->w_local_GP.put = 0;
-	ch->w_remote_GP.get = 0;
-	ch->w_remote_GP.put = 0;
-	ch->next_msg_to_pull = 0;
-
-	if (ch->flags & XPC_C_SETUP) {
-		ch->flags &= ~XPC_C_SETUP;
-
-		dev_dbg(xpc_chan, "ch->flags=0x%x, partid=%d, channel=%d\n",
-			ch->flags, ch->partid, ch->number);
-
-		kfree(ch->local_msgqueue_base);
-		ch->local_msgqueue = NULL;
-		kfree(ch->remote_msgqueue_base);
-		ch->remote_msgqueue = NULL;
-		kfree(ch->notify_queue);
-		ch->notify_queue = NULL;
-	}
-}
-
-
-/*
- * spin_lock_irqsave() is expected to be held on entry.
- */
-static void
-xpc_process_disconnect(struct xpc_channel *ch, unsigned long *irq_flags)
-{
-	struct xpc_partition *part = &xpc_partitions[ch->partid];
-	u32 channel_was_connected = (ch->flags & XPC_C_WASCONNECTED);
-
-
-	DBUG_ON(!spin_is_locked(&ch->lock));
-
-	if (!(ch->flags & XPC_C_DISCONNECTING)) {
-		return;
-	}
-
-	DBUG_ON(!(ch->flags & XPC_C_CLOSEREQUEST));
-
-	/* make sure all activity has settled down first */
-
-	if (atomic_read(&ch->kthreads_assigned) > 0 ||
-				atomic_read(&ch->references) > 0) {
-		return;
-	}
-	DBUG_ON((ch->flags & XPC_C_CONNECTEDCALLOUT_MADE) &&
-			!(ch->flags & XPC_C_DISCONNECTINGCALLOUT_MADE));
-
-	if (part->act_state == XPC_P_DEACTIVATING) {
-		/* can't proceed until the other side disengages from us */
-		if (xpc_partition_engaged(1UL << ch->partid)) {
-			return;
-		}
-
-	} else {
-
-		/* as long as the other side is up do the full protocol */
-
-		if (!(ch->flags & XPC_C_RCLOSEREQUEST)) {
-			return;
-		}
-
-		if (!(ch->flags & XPC_C_CLOSEREPLY)) {
-			ch->flags |= XPC_C_CLOSEREPLY;
-			xpc_IPI_send_closereply(ch, irq_flags);
-		}
-
-		if (!(ch->flags & XPC_C_RCLOSEREPLY)) {
-			return;
-		}
-	}
-
-	/* wake those waiting for notify completion */
-	if (atomic_read(&ch->n_to_notify) > 0) {
-		/* >>> we do callout while holding ch->lock */
-		xpc_notify_senders(ch, ch->reason, ch->w_local_GP.put);
-	}
-
-	/* both sides are disconnected now */
-
-	if (ch->flags & XPC_C_DISCONNECTINGCALLOUT_MADE) {
-		spin_unlock_irqrestore(&ch->lock, *irq_flags);
-		xpc_disconnect_callout(ch, xpcDisconnected);
-		spin_lock_irqsave(&ch->lock, *irq_flags);
-	}
-
-	/* it's now safe to free the channel's message queues */
-	xpc_free_msgqueues(ch);
-
-	/* mark disconnected, clear all other flags except XPC_C_WDISCONNECT */
-	ch->flags = (XPC_C_DISCONNECTED | (ch->flags & XPC_C_WDISCONNECT));
-
-	atomic_dec(&part->nchannels_active);
-
-	if (channel_was_connected) {
-		dev_info(xpc_chan, "channel %d to partition %d disconnected, "
-			"reason=%d\n", ch->number, ch->partid, ch->reason);
-	}
-
-	if (ch->flags & XPC_C_WDISCONNECT) {
-		/* we won't lose the CPU since we're holding ch->lock */
-		complete(&ch->wdisconnect_wait);
-	} else if (ch->delayed_IPI_flags) {
-		if (part->act_state != XPC_P_DEACTIVATING) {
-			/* time to take action on any delayed IPI flags */
-			spin_lock(&part->IPI_lock);
-			XPC_SET_IPI_FLAGS(part->local_IPI_amo, ch->number,
-							ch->delayed_IPI_flags);
-			spin_unlock(&part->IPI_lock);
-		}
-		ch->delayed_IPI_flags = 0;
-	}
-}
-
-
-/*
- * Process a change in the channel's remote connection state.
- */
-static void
-xpc_process_openclose_IPI(struct xpc_partition *part, int ch_number,
-				u8 IPI_flags)
-{
-	unsigned long irq_flags;
-	struct xpc_openclose_args *args =
-				&part->remote_openclose_args[ch_number];
-	struct xpc_channel *ch = &part->channels[ch_number];
-	enum xpc_retval reason;
-
-
-
-	spin_lock_irqsave(&ch->lock, irq_flags);
-
-again:
-
-	if ((ch->flags & XPC_C_DISCONNECTED) &&
-					(ch->flags & XPC_C_WDISCONNECT)) {
-		/*
-		 * Delay processing IPI flags until thread waiting disconnect
-		 * has had a chance to see that the channel is disconnected.
-		 */
-		ch->delayed_IPI_flags |= IPI_flags;
-		spin_unlock_irqrestore(&ch->lock, irq_flags);
-		return;
-	}
-
-
-	if (IPI_flags & XPC_IPI_CLOSEREQUEST) {
-
-		dev_dbg(xpc_chan, "XPC_IPI_CLOSEREQUEST (reason=%d) received "
-			"from partid=%d, channel=%d\n", args->reason,
-			ch->partid, ch->number);
-
-		/*
-		 * If RCLOSEREQUEST is set, we're probably waiting for
-		 * RCLOSEREPLY. We should find it and a ROPENREQUEST packed
-		 * with this RCLOSEREQUEST in the IPI_flags.
-		 */
-
-		if (ch->flags & XPC_C_RCLOSEREQUEST) {
-			DBUG_ON(!(ch->flags & XPC_C_DISCONNECTING));
-			DBUG_ON(!(ch->flags & XPC_C_CLOSEREQUEST));
-			DBUG_ON(!(ch->flags & XPC_C_CLOSEREPLY));
-			DBUG_ON(ch->flags & XPC_C_RCLOSEREPLY);
-
-			DBUG_ON(!(IPI_flags & XPC_IPI_CLOSEREPLY));
-			IPI_flags &= ~XPC_IPI_CLOSEREPLY;
-			ch->flags |= XPC_C_RCLOSEREPLY;
-
-			/* both sides have finished disconnecting */
-			xpc_process_disconnect(ch, &irq_flags);
-			DBUG_ON(!(ch->flags & XPC_C_DISCONNECTED));
-			goto again;
-		}
-
-		if (ch->flags & XPC_C_DISCONNECTED) {
-			if (!(IPI_flags & XPC_IPI_OPENREQUEST)) {
-				if ((XPC_GET_IPI_FLAGS(part->local_IPI_amo,
-					 ch_number) & XPC_IPI_OPENREQUEST)) {
-
-					DBUG_ON(ch->delayed_IPI_flags != 0);
-					spin_lock(&part->IPI_lock);
-					XPC_SET_IPI_FLAGS(part->local_IPI_amo,
-							ch_number,
-							XPC_IPI_CLOSEREQUEST);
-					spin_unlock(&part->IPI_lock);
-				}
-				spin_unlock_irqrestore(&ch->lock, irq_flags);
-				return;
-			}
-
-			XPC_SET_REASON(ch, 0, 0);
-			ch->flags &= ~XPC_C_DISCONNECTED;
-
-			atomic_inc(&part->nchannels_active);
-			ch->flags |= (XPC_C_CONNECTING | XPC_C_ROPENREQUEST);
-		}
-
-		IPI_flags &= ~(XPC_IPI_OPENREQUEST | XPC_IPI_OPENREPLY);
-
-		/*
-		 * The meaningful CLOSEREQUEST connection state fields are:
-		 *      reason = reason connection is to be closed
-		 */
-
-		ch->flags |= XPC_C_RCLOSEREQUEST;
-
-		if (!(ch->flags & XPC_C_DISCONNECTING)) {
-			reason = args->reason;
-			if (reason <= xpcSuccess || reason > xpcUnknownReason) {
-				reason = xpcUnknownReason;
-			} else if (reason == xpcUnregistering) {
-				reason = xpcOtherUnregistering;
-			}
-
-			XPC_DISCONNECT_CHANNEL(ch, reason, &irq_flags);
-
-			DBUG_ON(IPI_flags & XPC_IPI_CLOSEREPLY);
-			spin_unlock_irqrestore(&ch->lock, irq_flags);
-			return;
-		}
-
-		xpc_process_disconnect(ch, &irq_flags);
-	}
-
-
-	if (IPI_flags & XPC_IPI_CLOSEREPLY) {
-
-		dev_dbg(xpc_chan, "XPC_IPI_CLOSEREPLY received from partid=%d,"
-			" channel=%d\n", ch->partid, ch->number);
-
-		if (ch->flags & XPC_C_DISCONNECTED) {
-			DBUG_ON(part->act_state != XPC_P_DEACTIVATING);
-			spin_unlock_irqrestore(&ch->lock, irq_flags);
-			return;
-		}
-
-		DBUG_ON(!(ch->flags & XPC_C_CLOSEREQUEST));
-
-		if (!(ch->flags & XPC_C_RCLOSEREQUEST)) {
-			if ((XPC_GET_IPI_FLAGS(part->local_IPI_amo, ch_number)
-						& XPC_IPI_CLOSEREQUEST)) {
-
-				DBUG_ON(ch->delayed_IPI_flags != 0);
-				spin_lock(&part->IPI_lock);
-				XPC_SET_IPI_FLAGS(part->local_IPI_amo,
-						ch_number, XPC_IPI_CLOSEREPLY);
-				spin_unlock(&part->IPI_lock);
-			}
-			spin_unlock_irqrestore(&ch->lock, irq_flags);
-			return;
-		}
-
-		ch->flags |= XPC_C_RCLOSEREPLY;
-
-		if (ch->flags & XPC_C_CLOSEREPLY) {
-			/* both sides have finished disconnecting */
-			xpc_process_disconnect(ch, &irq_flags);
-		}
-	}
-
-
-	if (IPI_flags & XPC_IPI_OPENREQUEST) {
-
-		dev_dbg(xpc_chan, "XPC_IPI_OPENREQUEST (msg_size=%d, "
-			"local_nentries=%d) received from partid=%d, "
-			"channel=%d\n", args->msg_size, args->local_nentries,
-			ch->partid, ch->number);
-
-		if (part->act_state == XPC_P_DEACTIVATING ||
-					(ch->flags & XPC_C_ROPENREQUEST)) {
-			spin_unlock_irqrestore(&ch->lock, irq_flags);
-			return;
-		}
-
-		if (ch->flags & (XPC_C_DISCONNECTING | XPC_C_WDISCONNECT)) {
-			ch->delayed_IPI_flags |= XPC_IPI_OPENREQUEST;
-			spin_unlock_irqrestore(&ch->lock, irq_flags);
-			return;
-		}
-		DBUG_ON(!(ch->flags & (XPC_C_DISCONNECTED |
-							XPC_C_OPENREQUEST)));
-		DBUG_ON(ch->flags & (XPC_C_ROPENREQUEST | XPC_C_ROPENREPLY |
-					XPC_C_OPENREPLY | XPC_C_CONNECTED));
-
-		/*
-		 * The meaningful OPENREQUEST connection state fields are:
-		 *      msg_size = size of channel's messages in bytes
-		 *      local_nentries = remote partition's local_nentries
-		 */
-		if (args->msg_size == 0 || args->local_nentries == 0) {
-			/* assume OPENREQUEST was delayed by mistake */
-			spin_unlock_irqrestore(&ch->lock, irq_flags);
-			return;
-		}
-
-		ch->flags |= (XPC_C_ROPENREQUEST | XPC_C_CONNECTING);
-		ch->remote_nentries = args->local_nentries;
-
-
-		if (ch->flags & XPC_C_OPENREQUEST) {
-			if (args->msg_size != ch->msg_size) {
-				XPC_DISCONNECT_CHANNEL(ch, xpcUnequalMsgSizes,
-								&irq_flags);
-				spin_unlock_irqrestore(&ch->lock, irq_flags);
-				return;
-			}
-		} else {
-			ch->msg_size = args->msg_size;
-
-			XPC_SET_REASON(ch, 0, 0);
-			ch->flags &= ~XPC_C_DISCONNECTED;
-
-			atomic_inc(&part->nchannels_active);
-		}
-
-		xpc_process_connect(ch, &irq_flags);
-	}
-
-
-	if (IPI_flags & XPC_IPI_OPENREPLY) {
-
-		dev_dbg(xpc_chan, "XPC_IPI_OPENREPLY (local_msgqueue_pa=0x%lx, "
-			"local_nentries=%d, remote_nentries=%d) received from "
-			"partid=%d, channel=%d\n", args->local_msgqueue_pa,
-			args->local_nentries, args->remote_nentries,
-			ch->partid, ch->number);
-
-		if (ch->flags & (XPC_C_DISCONNECTING | XPC_C_DISCONNECTED)) {
-			spin_unlock_irqrestore(&ch->lock, irq_flags);
-			return;
-		}
-		if (!(ch->flags & XPC_C_OPENREQUEST)) {
-			XPC_DISCONNECT_CHANNEL(ch, xpcOpenCloseError,
-								&irq_flags);
-			spin_unlock_irqrestore(&ch->lock, irq_flags);
-			return;
-		}
-
-		DBUG_ON(!(ch->flags & XPC_C_ROPENREQUEST));
-		DBUG_ON(ch->flags & XPC_C_CONNECTED);
-
-		/*
-		 * The meaningful OPENREPLY connection state fields are:
-		 *      local_msgqueue_pa = physical address of remote
-		 *			    partition's local_msgqueue
-		 *      local_nentries = remote partition's local_nentries
-		 *      remote_nentries = remote partition's remote_nentries
-		 */
-		DBUG_ON(args->local_msgqueue_pa == 0);
-		DBUG_ON(args->local_nentries == 0);
-		DBUG_ON(args->remote_nentries == 0);
-
-		ch->flags |= XPC_C_ROPENREPLY;
-		ch->remote_msgqueue_pa = args->local_msgqueue_pa;
-
-		if (args->local_nentries < ch->remote_nentries) {
-			dev_dbg(xpc_chan, "XPC_IPI_OPENREPLY: new "
-				"remote_nentries=%d, old remote_nentries=%d, "
-				"partid=%d, channel=%d\n",
-				args->local_nentries, ch->remote_nentries,
-				ch->partid, ch->number);
-
-			ch->remote_nentries = args->local_nentries;
-		}
-		if (args->remote_nentries < ch->local_nentries) {
-			dev_dbg(xpc_chan, "XPC_IPI_OPENREPLY: new "
-				"local_nentries=%d, old local_nentries=%d, "
-				"partid=%d, channel=%d\n",
-				args->remote_nentries, ch->local_nentries,
-				ch->partid, ch->number);
-
-			ch->local_nentries = args->remote_nentries;
-		}
-
-		xpc_process_connect(ch, &irq_flags);
-	}
-
-	spin_unlock_irqrestore(&ch->lock, irq_flags);
-}
-
-
-/*
- * Attempt to establish a channel connection to a remote partition.
- */
-static enum xpc_retval
-xpc_connect_channel(struct xpc_channel *ch)
-{
-	unsigned long irq_flags;
-	struct xpc_registration *registration = &xpc_registrations[ch->number];
-
-
-	if (mutex_trylock(&registration->mutex) == 0) {
-		return xpcRetry;
-	}
-
-	if (!XPC_CHANNEL_REGISTERED(ch->number)) {
-		mutex_unlock(&registration->mutex);
-		return xpcUnregistered;
-	}
-
-	spin_lock_irqsave(&ch->lock, irq_flags);
-
-	DBUG_ON(ch->flags & XPC_C_CONNECTED);
-	DBUG_ON(ch->flags & XPC_C_OPENREQUEST);
-
-	if (ch->flags & XPC_C_DISCONNECTING) {
-		spin_unlock_irqrestore(&ch->lock, irq_flags);
-		mutex_unlock(&registration->mutex);
-		return ch->reason;
-	}
-
-
-	/* add info from the channel connect registration to the channel */
-
-	ch->kthreads_assigned_limit = registration->assigned_limit;
-	ch->kthreads_idle_limit = registration->idle_limit;
-	DBUG_ON(atomic_read(&ch->kthreads_assigned) != 0);
-	DBUG_ON(atomic_read(&ch->kthreads_idle) != 0);
-	DBUG_ON(atomic_read(&ch->kthreads_active) != 0);
-
-	ch->func = registration->func;
-	DBUG_ON(registration->func == NULL);
-	ch->key = registration->key;
-
-	ch->local_nentries = registration->nentries;
-
-	if (ch->flags & XPC_C_ROPENREQUEST) {
-		if (registration->msg_size != ch->msg_size) {
-			/* the local and remote sides aren't the same */
-
-			/*
-			 * Because XPC_DISCONNECT_CHANNEL() can block we're
-			 * forced to up the registration sema before we unlock
-			 * the channel lock. But that's okay here because we're
-			 * done with the part that required the registration
-			 * sema. XPC_DISCONNECT_CHANNEL() requires that the
-			 * channel lock be locked and will unlock and relock
-			 * the channel lock as needed.
-			 */
-			mutex_unlock(&registration->mutex);
-			XPC_DISCONNECT_CHANNEL(ch, xpcUnequalMsgSizes,
-								&irq_flags);
-			spin_unlock_irqrestore(&ch->lock, irq_flags);
-			return xpcUnequalMsgSizes;
-		}
-	} else {
-		ch->msg_size = registration->msg_size;
-
-		XPC_SET_REASON(ch, 0, 0);
-		ch->flags &= ~XPC_C_DISCONNECTED;
-
-		atomic_inc(&xpc_partitions[ch->partid].nchannels_active);
-	}
-
-	mutex_unlock(&registration->mutex);
-
-
-	/* initiate the connection */
-
-	ch->flags |= (XPC_C_OPENREQUEST | XPC_C_CONNECTING);
-	xpc_IPI_send_openrequest(ch, &irq_flags);
-
-	xpc_process_connect(ch, &irq_flags);
-
-	spin_unlock_irqrestore(&ch->lock, irq_flags);
-
-	return xpcSuccess;
-}
-
-
-/*
- * Clear some of the msg flags in the local message queue.
- */
-static inline void
-xpc_clear_local_msgqueue_flags(struct xpc_channel *ch)
-{
-	struct xpc_msg *msg;
-	s64 get;
-
-
-	get = ch->w_remote_GP.get;
-	do {
-		msg = (struct xpc_msg *) ((u64) ch->local_msgqueue +
-				(get % ch->local_nentries) * ch->msg_size);
-		msg->flags = 0;
-	} while (++get < (volatile s64) ch->remote_GP.get);
-}
-
-
-/*
- * Clear some of the msg flags in the remote message queue.
- */
-static inline void
-xpc_clear_remote_msgqueue_flags(struct xpc_channel *ch)
-{
-	struct xpc_msg *msg;
-	s64 put;
-
-
-	put = ch->w_remote_GP.put;
-	do {
-		msg = (struct xpc_msg *) ((u64) ch->remote_msgqueue +
-				(put % ch->remote_nentries) * ch->msg_size);
-		msg->flags = 0;
-	} while (++put < (volatile s64) ch->remote_GP.put);
-}
-
-
-static void
-xpc_process_msg_IPI(struct xpc_partition *part, int ch_number)
-{
-	struct xpc_channel *ch = &part->channels[ch_number];
-	int nmsgs_sent;
-
-
-	ch->remote_GP = part->remote_GPs[ch_number];
-
-
-	/* See what, if anything, has changed for each connected channel */
-
-	xpc_msgqueue_ref(ch);
-
-	if (ch->w_remote_GP.get == ch->remote_GP.get &&
-				ch->w_remote_GP.put == ch->remote_GP.put) {
-		/* nothing changed since GPs were last pulled */
-		xpc_msgqueue_deref(ch);
-		return;
-	}
-
-	if (!(ch->flags & XPC_C_CONNECTED)){
-		xpc_msgqueue_deref(ch);
-		return;
-	}
-
-
-	/*
-	 * First check to see if messages recently sent by us have been
-	 * received by the other side. (The remote GET value will have
-	 * changed since we last looked at it.)
-	 */
-
-	if (ch->w_remote_GP.get != ch->remote_GP.get) {
-
-		/*
-		 * We need to notify any senders that want to be notified
-		 * that their sent messages have been received by their
-		 * intended recipients. We need to do this before updating
-		 * w_remote_GP.get so that we don't allocate the same message
-		 * queue entries prematurely (see xpc_allocate_msg()).
-		 */
-		if (atomic_read(&ch->n_to_notify) > 0) {
-			/*
-			 * Notify senders that messages sent have been
-			 * received and delivered by the other side.
-			 */
-			xpc_notify_senders(ch, xpcMsgDelivered,
-							ch->remote_GP.get);
-		}
-
-		/*
-		 * Clear msg->flags in previously sent messages, so that
-		 * they're ready for xpc_allocate_msg().
-		 */
-		xpc_clear_local_msgqueue_flags(ch);
-
-		ch->w_remote_GP.get = ch->remote_GP.get;
-
-		dev_dbg(xpc_chan, "w_remote_GP.get changed to %ld, partid=%d, "
-			"channel=%d\n", ch->w_remote_GP.get, ch->partid,
-			ch->number);
-
-		/*
-		 * If anyone was waiting for message queue entries to become
-		 * available, wake them up.
-		 */
-		if (atomic_read(&ch->n_on_msg_allocate_wq) > 0) {
-			wake_up(&ch->msg_allocate_wq);
-		}
-	}
-
-
-	/*
-	 * Now check for newly sent messages by the other side. (The remote
-	 * PUT value will have changed since we last looked at it.)
-	 */
-
-	if (ch->w_remote_GP.put != ch->remote_GP.put) {
-		/*
-		 * Clear msg->flags in previously received messages, so that
-		 * they're ready for xpc_get_deliverable_msg().
-		 */
-		xpc_clear_remote_msgqueue_flags(ch);
-
-		ch->w_remote_GP.put = ch->remote_GP.put;
-
-		dev_dbg(xpc_chan, "w_remote_GP.put changed to %ld, partid=%d, "
-			"channel=%d\n", ch->w_remote_GP.put, ch->partid,
-			ch->number);
-
-		nmsgs_sent = ch->w_remote_GP.put - ch->w_local_GP.get;
-		if (nmsgs_sent > 0) {
-			dev_dbg(xpc_chan, "msgs waiting to be copied and "
-				"delivered=%d, partid=%d, channel=%d\n",
-				nmsgs_sent, ch->partid, ch->number);
-
-			if (ch->flags & XPC_C_CONNECTEDCALLOUT_MADE) {
-				xpc_activate_kthreads(ch, nmsgs_sent);
-			}
-		}
-	}
-
-	xpc_msgqueue_deref(ch);
-}
-
-
-void
-xpc_process_channel_activity(struct xpc_partition *part)
-{
-	unsigned long irq_flags;
-	u64 IPI_amo, IPI_flags;
-	struct xpc_channel *ch;
-	int ch_number;
-	u32 ch_flags;
-
-
-	IPI_amo = xpc_get_IPI_flags(part);
-
-	/*
-	 * Initiate channel connections for registered channels.
-	 *
-	 * For each connected channel that has pending messages activate idle
-	 * kthreads and/or create new kthreads as needed.
-	 */
-
-	for (ch_number = 0; ch_number < part->nchannels; ch_number++) {
-		ch = &part->channels[ch_number];
-
-
-		/*
-		 * Process any open or close related IPI flags, and then deal
-		 * with connecting or disconnecting the channel as required.
-		 */
-
-		IPI_flags = XPC_GET_IPI_FLAGS(IPI_amo, ch_number);
-
-		if (XPC_ANY_OPENCLOSE_IPI_FLAGS_SET(IPI_flags)) {
-			xpc_process_openclose_IPI(part, ch_number, IPI_flags);
-		}
-
-		ch_flags = ch->flags;	/* need an atomic snapshot of flags */
-
-		if (ch_flags & XPC_C_DISCONNECTING) {
-			spin_lock_irqsave(&ch->lock, irq_flags);
-			xpc_process_disconnect(ch, &irq_flags);
-			spin_unlock_irqrestore(&ch->lock, irq_flags);
-			continue;
-		}
-
-		if (part->act_state == XPC_P_DEACTIVATING) {
-			continue;
-		}
-
-		if (!(ch_flags & XPC_C_CONNECTED)) {
-			if (!(ch_flags & XPC_C_OPENREQUEST)) {
-				DBUG_ON(ch_flags & XPC_C_SETUP);
-				(void) xpc_connect_channel(ch);
-			} else {
-				spin_lock_irqsave(&ch->lock, irq_flags);
-				xpc_process_connect(ch, &irq_flags);
-				spin_unlock_irqrestore(&ch->lock, irq_flags);
-			}
-			continue;
-		}
-
-
-		/*
-		 * Process any message related IPI flags, this may involve the
-		 * activation of kthreads to deliver any pending messages sent
-		 * from the other partition.
-		 */
-
-		if (XPC_ANY_MSG_IPI_FLAGS_SET(IPI_flags)) {
-			xpc_process_msg_IPI(part, ch_number);
-		}
-	}
-}
-
-
-/*
- * XPC's heartbeat code calls this function to inform XPC that a partition is
- * going down.  XPC responds by tearing down the XPartition Communication
- * infrastructure used for the just downed partition.
- *
- * XPC's heartbeat code will never call this function and xpc_partition_up()
- * at the same time. Nor will it ever make multiple calls to either function
- * at the same time.
- */
-void
-xpc_partition_going_down(struct xpc_partition *part, enum xpc_retval reason)
-{
-	unsigned long irq_flags;
-	int ch_number;
-	struct xpc_channel *ch;
-
-
-	dev_dbg(xpc_chan, "deactivating partition %d, reason=%d\n",
-		XPC_PARTID(part), reason);
-
-	if (!xpc_part_ref(part)) {
-		/* infrastructure for this partition isn't currently set up */
-		return;
-	}
-
-
-	/* disconnect channels associated with the partition going down */
-
-	for (ch_number = 0; ch_number < part->nchannels; ch_number++) {
-		ch = &part->channels[ch_number];
-
-		xpc_msgqueue_ref(ch);
-		spin_lock_irqsave(&ch->lock, irq_flags);
-
-		XPC_DISCONNECT_CHANNEL(ch, reason, &irq_flags);
-
-		spin_unlock_irqrestore(&ch->lock, irq_flags);
-		xpc_msgqueue_deref(ch);
-	}
-
-	xpc_wakeup_channel_mgr(part);
-
-	xpc_part_deref(part);
-}
-
-
-/*
- * Teardown the infrastructure necessary to support XPartition Communication
- * between the specified remote partition and the local one.
- */
-void
-xpc_teardown_infrastructure(struct xpc_partition *part)
-{
-	partid_t partid = XPC_PARTID(part);
-
-
-	/*
-	 * We start off by making this partition inaccessible to local
-	 * processes by marking it as no longer setup. Then we make it
-	 * inaccessible to remote processes by clearing the XPC per partition
-	 * specific variable's magic # (which indicates that these variables
-	 * are no longer valid) and by ignoring all XPC notify IPIs sent to
-	 * this partition.
-	 */
-
-	DBUG_ON(atomic_read(&part->nchannels_engaged) != 0);
-	DBUG_ON(atomic_read(&part->nchannels_active) != 0);
-	DBUG_ON(part->setup_state != XPC_P_SETUP);
-	part->setup_state = XPC_P_WTEARDOWN;
-
-	xpc_vars_part[partid].magic = 0;
-
-
-	free_irq(SGI_XPC_NOTIFY, (void *) (u64) partid);
-
-
-	/*
-	 * Before proceeding with the teardown we have to wait until all
-	 * existing references cease.
-	 */
-	wait_event(part->teardown_wq, (atomic_read(&part->references) == 0));
-
-
-	/* now we can begin tearing down the infrastructure */
-
-	part->setup_state = XPC_P_TORNDOWN;
-
-	/* in case we've still got outstanding timers registered... */
-	del_timer_sync(&part->dropped_IPI_timer);
-
-	kfree(part->remote_openclose_args_base);
-	part->remote_openclose_args = NULL;
-	kfree(part->local_openclose_args_base);
-	part->local_openclose_args = NULL;
-	kfree(part->remote_GPs_base);
-	part->remote_GPs = NULL;
-	kfree(part->local_GPs_base);
-	part->local_GPs = NULL;
-	kfree(part->channels);
-	part->channels = NULL;
-	part->local_IPI_amo_va = NULL;
-}
-
-
-/*
- * Called by XP at the time of channel connection registration to cause
- * XPC to establish connections to all currently active partitions.
- */
-void
-xpc_initiate_connect(int ch_number)
-{
-	partid_t partid;
-	struct xpc_partition *part;
-	struct xpc_channel *ch;
-
-
-	DBUG_ON(ch_number < 0 || ch_number >= XPC_NCHANNELS);
-
-	for (partid = 1; partid < XP_MAX_PARTITIONS; partid++) {
-		part = &xpc_partitions[partid];
-
-		if (xpc_part_ref(part)) {
-			ch = &part->channels[ch_number];
-
-			/*
-			 * Initiate the establishment of a connection on the
-			 * newly registered channel to the remote partition.
-			 */
-			xpc_wakeup_channel_mgr(part);
-			xpc_part_deref(part);
-		}
-	}
-}
-
-
-void
-xpc_connected_callout(struct xpc_channel *ch)
-{
-	/* let the registerer know that a connection has been established */
-
-	if (ch->func != NULL) {
-		dev_dbg(xpc_chan, "ch->func() called, reason=xpcConnected, "
-			"partid=%d, channel=%d\n", ch->partid, ch->number);
-
-		ch->func(xpcConnected, ch->partid, ch->number,
-				(void *) (u64) ch->local_nentries, ch->key);
-
-		dev_dbg(xpc_chan, "ch->func() returned, reason=xpcConnected, "
-			"partid=%d, channel=%d\n", ch->partid, ch->number);
-	}
-}
-
-
-/*
- * Called by XP at the time of channel connection unregistration to cause
- * XPC to teardown all current connections for the specified channel.
- *
- * Before returning xpc_initiate_disconnect() will wait until all connections
- * on the specified channel have been closed/torndown. So the caller can be
- * assured that they will not be receiving any more callouts from XPC to the
- * function they registered via xpc_connect().
- *
- * Arguments:
- *
- *	ch_number - channel # to unregister.
- */
-void
-xpc_initiate_disconnect(int ch_number)
-{
-	unsigned long irq_flags;
-	partid_t partid;
-	struct xpc_partition *part;
-	struct xpc_channel *ch;
-
-
-	DBUG_ON(ch_number < 0 || ch_number >= XPC_NCHANNELS);
-
-	/* initiate the channel disconnect for every active partition */
-	for (partid = 1; partid < XP_MAX_PARTITIONS; partid++) {
-		part = &xpc_partitions[partid];
-
-		if (xpc_part_ref(part)) {
-			ch = &part->channels[ch_number];
-			xpc_msgqueue_ref(ch);
-
-			spin_lock_irqsave(&ch->lock, irq_flags);
-
-			if (!(ch->flags & XPC_C_DISCONNECTED)) {
-				ch->flags |= XPC_C_WDISCONNECT;
-
-				XPC_DISCONNECT_CHANNEL(ch, xpcUnregistering,
-								&irq_flags);
-			}
-
-			spin_unlock_irqrestore(&ch->lock, irq_flags);
-
-			xpc_msgqueue_deref(ch);
-			xpc_part_deref(part);
-		}
-	}
-
-	xpc_disconnect_wait(ch_number);
-}
-
-
-/*
- * To disconnect a channel, and reflect it back to all who may be waiting.
- *
- * An OPEN is not allowed until XPC_C_DISCONNECTING is cleared by
- * xpc_process_disconnect(), and if set, XPC_C_WDISCONNECT is cleared by
- * xpc_disconnect_wait().
- *
- * THE CHANNEL IS TO BE LOCKED BY THE CALLER AND WILL REMAIN LOCKED UPON RETURN.
- */
-void
-xpc_disconnect_channel(const int line, struct xpc_channel *ch,
-			enum xpc_retval reason, unsigned long *irq_flags)
-{
-	u32 channel_was_connected = (ch->flags & XPC_C_CONNECTED);
-
-
-	DBUG_ON(!spin_is_locked(&ch->lock));
-
-	if (ch->flags & (XPC_C_DISCONNECTING | XPC_C_DISCONNECTED)) {
-		return;
-	}
-	DBUG_ON(!(ch->flags & (XPC_C_CONNECTING | XPC_C_CONNECTED)));
-
-	dev_dbg(xpc_chan, "reason=%d, line=%d, partid=%d, channel=%d\n",
-		reason, line, ch->partid, ch->number);
-
-	XPC_SET_REASON(ch, reason, line);
-
-	ch->flags |= (XPC_C_CLOSEREQUEST | XPC_C_DISCONNECTING);
-	/* some of these may not have been set */
-	ch->flags &= ~(XPC_C_OPENREQUEST | XPC_C_OPENREPLY |
-			XPC_C_ROPENREQUEST | XPC_C_ROPENREPLY |
-			XPC_C_CONNECTING | XPC_C_CONNECTED);
-
-	xpc_IPI_send_closerequest(ch, irq_flags);
-
-	if (channel_was_connected) {
-		ch->flags |= XPC_C_WASCONNECTED;
-	}
-
-	spin_unlock_irqrestore(&ch->lock, *irq_flags);
-
-	/* wake all idle kthreads so they can exit */
-	if (atomic_read(&ch->kthreads_idle) > 0) {
-		wake_up_all(&ch->idle_wq);
-
-	} else if ((ch->flags & XPC_C_CONNECTEDCALLOUT_MADE) &&
-			!(ch->flags & XPC_C_DISCONNECTINGCALLOUT)) {
-		/* start a kthread that will do the xpcDisconnecting callout */
-		xpc_create_kthreads(ch, 1, 1);
-	}
-
-	/* wake those waiting to allocate an entry from the local msg queue */
-	if (atomic_read(&ch->n_on_msg_allocate_wq) > 0) {
-		wake_up(&ch->msg_allocate_wq);
-	}
-
-	spin_lock_irqsave(&ch->lock, *irq_flags);
-}
-
-
-void
-xpc_disconnect_callout(struct xpc_channel *ch, enum xpc_retval reason)
-{
-	/*
-	 * Let the channel's registerer know that the channel is being
-	 * disconnected. We don't want to do this if the registerer was never
-	 * informed of a connection being made.
-	 */
-
-	if (ch->func != NULL) {
-		dev_dbg(xpc_chan, "ch->func() called, reason=%d, partid=%d, "
-			"channel=%d\n", reason, ch->partid, ch->number);
-
-		ch->func(reason, ch->partid, ch->number, NULL, ch->key);
-
-		dev_dbg(xpc_chan, "ch->func() returned, reason=%d, partid=%d, "
-			"channel=%d\n", reason, ch->partid, ch->number);
-	}
-}
-
-
-/*
- * Wait for a message entry to become available for the specified channel,
- * but don't wait any longer than 1 jiffy.
- */
-static enum xpc_retval
-xpc_allocate_msg_wait(struct xpc_channel *ch)
-{
-	enum xpc_retval ret;
-
-
-	if (ch->flags & XPC_C_DISCONNECTING) {
-		DBUG_ON(ch->reason == xpcInterrupted);  // >>> Is this true?
-		return ch->reason;
-	}
-
-	atomic_inc(&ch->n_on_msg_allocate_wq);
-	ret = interruptible_sleep_on_timeout(&ch->msg_allocate_wq, 1);
-	atomic_dec(&ch->n_on_msg_allocate_wq);
-
-	if (ch->flags & XPC_C_DISCONNECTING) {
-		ret = ch->reason;
-		DBUG_ON(ch->reason == xpcInterrupted);  // >>> Is this true?
-	} else if (ret == 0) {
-		ret = xpcTimeout;
-	} else {
-		ret = xpcInterrupted;
-	}
-
-	return ret;
-}
-
-
-/*
- * Allocate an entry for a message from the message queue associated with the
- * specified channel.
- */
-static enum xpc_retval
-xpc_allocate_msg(struct xpc_channel *ch, u32 flags,
-			struct xpc_msg **address_of_msg)
-{
-	struct xpc_msg *msg;
-	enum xpc_retval ret;
-	s64 put;
-
-
-	/* this reference will be dropped in xpc_send_msg() */
-	xpc_msgqueue_ref(ch);
-
-	if (ch->flags & XPC_C_DISCONNECTING) {
-		xpc_msgqueue_deref(ch);
-		return ch->reason;
-	}
-	if (!(ch->flags & XPC_C_CONNECTED)) {
-		xpc_msgqueue_deref(ch);
-		return xpcNotConnected;
-	}
-
-
-	/*
-	 * Get the next available message entry from the local message queue.
-	 * If none are available, we'll make sure that we grab the latest
-	 * GP values.
-	 */
-	ret = xpcTimeout;
-
-	while (1) {
-
-		put = (volatile s64) ch->w_local_GP.put;
-		if (put - (volatile s64) ch->w_remote_GP.get <
-							ch->local_nentries) {
-
-			/* There are available message entries. We need to try
-			 * to secure one for ourselves. We'll do this by trying
-			 * to increment w_local_GP.put as long as someone else
-			 * doesn't beat us to it. If they do, we'll have to
-			 * try again.
-		 	 */
-			if (cmpxchg(&ch->w_local_GP.put, put, put + 1) ==
-									put) {
-				/* we got the entry referenced by put */
-				break;
-			}
-			continue;	/* try again */
-		}
-
-
-		/*
-		 * There aren't any available msg entries at this time.
-		 *
-		 * In waiting for a message entry to become available,
-		 * we set a timeout in case the other side is not
-		 * sending completion IPIs. This lets us fake an IPI
-		 * that will cause the IPI handler to fetch the latest
-		 * GP values as if an IPI was sent by the other side.
-		 */
-		if (ret == xpcTimeout) {
-			xpc_IPI_send_local_msgrequest(ch);
-		}
-
-		if (flags & XPC_NOWAIT) {
-			xpc_msgqueue_deref(ch);
-			return xpcNoWait;
-		}
-
-		ret = xpc_allocate_msg_wait(ch);
-		if (ret != xpcInterrupted && ret != xpcTimeout) {
-			xpc_msgqueue_deref(ch);
-			return ret;
-		}
-	}
-
-
-	/* get the message's address and initialize it */
-	msg = (struct xpc_msg *) ((u64) ch->local_msgqueue +
-				(put % ch->local_nentries) * ch->msg_size);
-
-
-	DBUG_ON(msg->flags != 0);
-	msg->number = put;
-
-	dev_dbg(xpc_chan, "w_local_GP.put changed to %ld; msg=0x%p, "
-		"msg_number=%ld, partid=%d, channel=%d\n", put + 1,
-		(void *) msg, msg->number, ch->partid, ch->number);
-
-	*address_of_msg = msg;
-
-	return xpcSuccess;
-}
-
-
-/*
- * Allocate an entry for a message from the message queue associated with the
- * specified channel. NOTE that this routine can sleep waiting for a message
- * entry to become available. To not sleep, pass in the XPC_NOWAIT flag.
- *
- * Arguments:
- *
- *	partid - ID of partition to which the channel is connected.
- *	ch_number - channel #.
- *	flags - see xpc.h for valid flags.
- *	payload - address of the allocated payload area pointer (filled in on
- * 	          return) in which the user-defined message is constructed.
- */
-enum xpc_retval
-xpc_initiate_allocate(partid_t partid, int ch_number, u32 flags, void **payload)
-{
-	struct xpc_partition *part = &xpc_partitions[partid];
-	enum xpc_retval ret = xpcUnknownReason;
-	struct xpc_msg *msg = NULL;
-
-
-	DBUG_ON(partid <= 0 || partid >= XP_MAX_PARTITIONS);
-	DBUG_ON(ch_number < 0 || ch_number >= part->nchannels);
-
-	*payload = NULL;
-
-	if (xpc_part_ref(part)) {
-		ret = xpc_allocate_msg(&part->channels[ch_number], flags, &msg);
-		xpc_part_deref(part);
-
-		if (msg != NULL) {
-			*payload = &msg->payload;
-		}
-	}
-
-	return ret;
-}
-
-
-/*
- * Now we actually send the messages that are ready to be sent by advancing
- * the local message queue's Put value and then send an IPI to the recipient
- * partition.
- */
-static void
-xpc_send_msgs(struct xpc_channel *ch, s64 initial_put)
-{
-	struct xpc_msg *msg;
-	s64 put = initial_put + 1;
-	int send_IPI = 0;
-
-
-	while (1) {
-
-		while (1) {
-			if (put == (volatile s64) ch->w_local_GP.put) {
-				break;
-			}
-
-			msg = (struct xpc_msg *) ((u64) ch->local_msgqueue +
-			       (put % ch->local_nentries) * ch->msg_size);
-
-			if (!(msg->flags & XPC_M_READY)) {
-				break;
-			}
-
-			put++;
-		}
-
-		if (put == initial_put) {
-			/* nothing's changed */
-			break;
-		}
-
-		if (cmpxchg_rel(&ch->local_GP->put, initial_put, put) !=
-								initial_put) {
-			/* someone else beat us to it */
-			DBUG_ON((volatile s64) ch->local_GP->put < initial_put);
-			break;
-		}
-
-		/* we just set the new value of local_GP->put */
-
-		dev_dbg(xpc_chan, "local_GP->put changed to %ld, partid=%d, "
-			"channel=%d\n", put, ch->partid, ch->number);
-
-		send_IPI = 1;
-
-		/*
-		 * We need to ensure that the message referenced by
-		 * local_GP->put is not XPC_M_READY or that local_GP->put
-		 * equals w_local_GP.put, so we'll go have a look.
-		 */
-		initial_put = put;
-	}
-
-	if (send_IPI) {
-		xpc_IPI_send_msgrequest(ch);
-	}
-}
-
-
-/*
- * Common code that does the actual sending of the message by advancing the
- * local message queue's Put value and sends an IPI to the partition the
- * message is being sent to.
- */
-static enum xpc_retval
-xpc_send_msg(struct xpc_channel *ch, struct xpc_msg *msg, u8 notify_type,
-			xpc_notify_func func, void *key)
-{
-	enum xpc_retval ret = xpcSuccess;
-	struct xpc_notify *notify = notify;
-	s64 put, msg_number = msg->number;
-
-
-	DBUG_ON(notify_type == XPC_N_CALL && func == NULL);
-	DBUG_ON((((u64) msg - (u64) ch->local_msgqueue) / ch->msg_size) !=
-					msg_number % ch->local_nentries);
-	DBUG_ON(msg->flags & XPC_M_READY);
-
-	if (ch->flags & XPC_C_DISCONNECTING) {
-		/* drop the reference grabbed in xpc_allocate_msg() */
-		xpc_msgqueue_deref(ch);
-		return ch->reason;
-	}
-
-	if (notify_type != 0) {
-		/*
-		 * Tell the remote side to send an ACK interrupt when the
-		 * message has been delivered.
-		 */
-		msg->flags |= XPC_M_INTERRUPT;
-
-		atomic_inc(&ch->n_to_notify);
-
-		notify = &ch->notify_queue[msg_number % ch->local_nentries];
-		notify->func = func;
-		notify->key = key;
-		notify->type = notify_type;
-
-		// >>> is a mb() needed here?
-
-		if (ch->flags & XPC_C_DISCONNECTING) {
-			/*
-			 * An error occurred between our last error check and
-			 * this one. We will try to clear the type field from
-			 * the notify entry. If we succeed then
-			 * xpc_disconnect_channel() didn't already process
-			 * the notify entry.
-			 */
-			if (cmpxchg(&notify->type, notify_type, 0) ==
-								notify_type) {
-				atomic_dec(&ch->n_to_notify);
-				ret = ch->reason;
-			}
-
-			/* drop the reference grabbed in xpc_allocate_msg() */
-			xpc_msgqueue_deref(ch);
-			return ret;
-		}
-	}
-
-	msg->flags |= XPC_M_READY;
-
-	/*
-	 * The preceding store of msg->flags must occur before the following
-	 * load of ch->local_GP->put.
-	 */
-	mb();
-
-	/* see if the message is next in line to be sent, if so send it */
-
-	put = ch->local_GP->put;
-	if (put == msg_number) {
-		xpc_send_msgs(ch, put);
-	}
-
-	/* drop the reference grabbed in xpc_allocate_msg() */
-	xpc_msgqueue_deref(ch);
-	return ret;
-}
-
-
-/*
- * Send a message previously allocated using xpc_initiate_allocate() on the
- * specified channel connected to the specified partition.
- *
- * This routine will not wait for the message to be received, nor will
- * notification be given when it does happen. Once this routine has returned
- * the message entry allocated via xpc_initiate_allocate() is no longer
- * accessable to the caller.
- *
- * This routine, although called by users, does not call xpc_part_ref() to
- * ensure that the partition infrastructure is in place. It relies on the
- * fact that we called xpc_msgqueue_ref() in xpc_allocate_msg().
- *
- * Arguments:
- *
- *	partid - ID of partition to which the channel is connected.
- *	ch_number - channel # to send message on.
- *	payload - pointer to the payload area allocated via
- *			xpc_initiate_allocate().
- */
-enum xpc_retval
-xpc_initiate_send(partid_t partid, int ch_number, void *payload)
-{
-	struct xpc_partition *part = &xpc_partitions[partid];
-	struct xpc_msg *msg = XPC_MSG_ADDRESS(payload);
-	enum xpc_retval ret;
-
-
-	dev_dbg(xpc_chan, "msg=0x%p, partid=%d, channel=%d\n", (void *) msg,
-		partid, ch_number);
-
-	DBUG_ON(partid <= 0 || partid >= XP_MAX_PARTITIONS);
-	DBUG_ON(ch_number < 0 || ch_number >= part->nchannels);
-	DBUG_ON(msg == NULL);
-
-	ret = xpc_send_msg(&part->channels[ch_number], msg, 0, NULL, NULL);
-
-	return ret;
-}
-
-
-/*
- * Send a message previously allocated using xpc_initiate_allocate on the
- * specified channel connected to the specified partition.
- *
- * This routine will not wait for the message to be sent. Once this routine
- * has returned the message entry allocated via xpc_initiate_allocate() is no
- * longer accessable to the caller.
- *
- * Once the remote end of the channel has received the message, the function
- * passed as an argument to xpc_initiate_send_notify() will be called. This
- * allows the sender to free up or re-use any buffers referenced by the
- * message, but does NOT mean the message has been processed at the remote
- * end by a receiver.
- *
- * If this routine returns an error, the caller's function will NOT be called.
- *
- * This routine, although called by users, does not call xpc_part_ref() to
- * ensure that the partition infrastructure is in place. It relies on the
- * fact that we called xpc_msgqueue_ref() in xpc_allocate_msg().
- *
- * Arguments:
- *
- *	partid - ID of partition to which the channel is connected.
- *	ch_number - channel # to send message on.
- *	payload - pointer to the payload area allocated via
- *			xpc_initiate_allocate().
- *	func - function to call with asynchronous notification of message
- *		  receipt. THIS FUNCTION MUST BE NON-BLOCKING.
- *	key - user-defined key to be passed to the function when it's called.
- */
-enum xpc_retval
-xpc_initiate_send_notify(partid_t partid, int ch_number, void *payload,
-				xpc_notify_func func, void *key)
-{
-	struct xpc_partition *part = &xpc_partitions[partid];
-	struct xpc_msg *msg = XPC_MSG_ADDRESS(payload);
-	enum xpc_retval ret;
-
-
-	dev_dbg(xpc_chan, "msg=0x%p, partid=%d, channel=%d\n", (void *) msg,
-		partid, ch_number);
-
-	DBUG_ON(partid <= 0 || partid >= XP_MAX_PARTITIONS);
-	DBUG_ON(ch_number < 0 || ch_number >= part->nchannels);
-	DBUG_ON(msg == NULL);
-	DBUG_ON(func == NULL);
-
-	ret = xpc_send_msg(&part->channels[ch_number], msg, XPC_N_CALL,
-								func, key);
-	return ret;
-}
-
-
-static struct xpc_msg *
-xpc_pull_remote_msg(struct xpc_channel *ch, s64 get)
-{
-	struct xpc_partition *part = &xpc_partitions[ch->partid];
-	struct xpc_msg *remote_msg, *msg;
-	u32 msg_index, nmsgs;
-	u64 msg_offset;
-	enum xpc_retval ret;
-
-
-	if (mutex_lock_interruptible(&ch->msg_to_pull_mutex) != 0) {
-		/* we were interrupted by a signal */
-		return NULL;
-	}
-
-	while (get >= ch->next_msg_to_pull) {
-
-		/* pull as many messages as are ready and able to be pulled */
-
-		msg_index = ch->next_msg_to_pull % ch->remote_nentries;
-
-		DBUG_ON(ch->next_msg_to_pull >=
-					(volatile s64) ch->w_remote_GP.put);
-		nmsgs =  (volatile s64) ch->w_remote_GP.put -
-						ch->next_msg_to_pull;
-		if (msg_index + nmsgs > ch->remote_nentries) {
-			/* ignore the ones that wrap the msg queue for now */
-			nmsgs = ch->remote_nentries - msg_index;
-		}
-
-		msg_offset = msg_index * ch->msg_size;
-		msg = (struct xpc_msg *) ((u64) ch->remote_msgqueue +
-								msg_offset);
-		remote_msg = (struct xpc_msg *) (ch->remote_msgqueue_pa +
-								msg_offset);
-
-		if ((ret = xpc_pull_remote_cachelines(part, msg, remote_msg,
-				nmsgs * ch->msg_size)) != xpcSuccess) {
-
-			dev_dbg(xpc_chan, "failed to pull %d msgs starting with"
-				" msg %ld from partition %d, channel=%d, "
-				"ret=%d\n", nmsgs, ch->next_msg_to_pull,
-				ch->partid, ch->number, ret);
-
-			XPC_DEACTIVATE_PARTITION(part, ret);
-
-			mutex_unlock(&ch->msg_to_pull_mutex);
-			return NULL;
-		}
-
-		mb();	/* >>> this may not be needed, we're not sure */
-
-		ch->next_msg_to_pull += nmsgs;
-	}
-
-	mutex_unlock(&ch->msg_to_pull_mutex);
-
-	/* return the message we were looking for */
-	msg_offset = (get % ch->remote_nentries) * ch->msg_size;
-	msg = (struct xpc_msg *) ((u64) ch->remote_msgqueue + msg_offset);
-
-	return msg;
-}
-
-
-/*
- * Get a message to be delivered.
- */
-static struct xpc_msg *
-xpc_get_deliverable_msg(struct xpc_channel *ch)
-{
-	struct xpc_msg *msg = NULL;
-	s64 get;
-
-
-	do {
-		if ((volatile u32) ch->flags & XPC_C_DISCONNECTING) {
-			break;
-		}
-
-		get = (volatile s64) ch->w_local_GP.get;
-		if (get == (volatile s64) ch->w_remote_GP.put) {
-			break;
-		}
-
-		/* There are messages waiting to be pulled and delivered.
-		 * We need to try to secure one for ourselves. We'll do this
-		 * by trying to increment w_local_GP.get and hope that no one
-		 * else beats us to it. If they do, we'll we'll simply have
-		 * to try again for the next one.
-	 	 */
-
-		if (cmpxchg(&ch->w_local_GP.get, get, get + 1) == get) {
-			/* we got the entry referenced by get */
-
-			dev_dbg(xpc_chan, "w_local_GP.get changed to %ld, "
-				"partid=%d, channel=%d\n", get + 1,
-				ch->partid, ch->number);
-
-			/* pull the message from the remote partition */
-
-			msg = xpc_pull_remote_msg(ch, get);
-
-			DBUG_ON(msg != NULL && msg->number != get);
-			DBUG_ON(msg != NULL && (msg->flags & XPC_M_DONE));
-			DBUG_ON(msg != NULL && !(msg->flags & XPC_M_READY));
-
-			break;
-		}
-
-	} while (1);
-
-	return msg;
-}
-
-
-/*
- * Deliver a message to its intended recipient.
- */
-void
-xpc_deliver_msg(struct xpc_channel *ch)
-{
-	struct xpc_msg *msg;
-
-
-	if ((msg = xpc_get_deliverable_msg(ch)) != NULL) {
-
-		/*
-		 * This ref is taken to protect the payload itself from being
-		 * freed before the user is finished with it, which the user
-		 * indicates by calling xpc_initiate_received().
-		 */
-		xpc_msgqueue_ref(ch);
-
-		atomic_inc(&ch->kthreads_active);
-
-		if (ch->func != NULL) {
-			dev_dbg(xpc_chan, "ch->func() called, msg=0x%p, "
-				"msg_number=%ld, partid=%d, channel=%d\n",
-				(void *) msg, msg->number, ch->partid,
-				ch->number);
-
-			/* deliver the message to its intended recipient */
-			ch->func(xpcMsgReceived, ch->partid, ch->number,
-					&msg->payload, ch->key);
-
-			dev_dbg(xpc_chan, "ch->func() returned, msg=0x%p, "
-				"msg_number=%ld, partid=%d, channel=%d\n",
-				(void *) msg, msg->number, ch->partid,
-				ch->number);
-		}
-
-		atomic_dec(&ch->kthreads_active);
-	}
-}
-
-
-/*
- * Now we actually acknowledge the messages that have been delivered and ack'd
- * by advancing the cached remote message queue's Get value and if requested
- * send an IPI to the message sender's partition.
- */
-static void
-xpc_acknowledge_msgs(struct xpc_channel *ch, s64 initial_get, u8 msg_flags)
-{
-	struct xpc_msg *msg;
-	s64 get = initial_get + 1;
-	int send_IPI = 0;
-
-
-	while (1) {
-
-		while (1) {
-			if (get == (volatile s64) ch->w_local_GP.get) {
-				break;
-			}
-
-			msg = (struct xpc_msg *) ((u64) ch->remote_msgqueue +
-			       (get % ch->remote_nentries) * ch->msg_size);
-
-			if (!(msg->flags & XPC_M_DONE)) {
-				break;
-			}
-
-			msg_flags |= msg->flags;
-			get++;
-		}
-
-		if (get == initial_get) {
-			/* nothing's changed */
-			break;
-		}
-
-		if (cmpxchg_rel(&ch->local_GP->get, initial_get, get) !=
-								initial_get) {
-			/* someone else beat us to it */
-			DBUG_ON((volatile s64) ch->local_GP->get <=
-								initial_get);
-			break;
-		}
-
-		/* we just set the new value of local_GP->get */
-
-		dev_dbg(xpc_chan, "local_GP->get changed to %ld, partid=%d, "
-			"channel=%d\n", get, ch->partid, ch->number);
-
-		send_IPI = (msg_flags & XPC_M_INTERRUPT);
-
-		/*
-		 * We need to ensure that the message referenced by
-		 * local_GP->get is not XPC_M_DONE or that local_GP->get
-		 * equals w_local_GP.get, so we'll go have a look.
-		 */
-		initial_get = get;
-	}
-
-	if (send_IPI) {
-		xpc_IPI_send_msgrequest(ch);
-	}
-}
-
-
-/*
- * Acknowledge receipt of a delivered message.
- *
- * If a message has XPC_M_INTERRUPT set, send an interrupt to the partition
- * that sent the message.
- *
- * This function, although called by users, does not call xpc_part_ref() to
- * ensure that the partition infrastructure is in place. It relies on the
- * fact that we called xpc_msgqueue_ref() in xpc_deliver_msg().
- *
- * Arguments:
- *
- *	partid - ID of partition to which the channel is connected.
- *	ch_number - channel # message received on.
- *	payload - pointer to the payload area allocated via
- *			xpc_initiate_allocate().
- */
-void
-xpc_initiate_received(partid_t partid, int ch_number, void *payload)
-{
-	struct xpc_partition *part = &xpc_partitions[partid];
-	struct xpc_channel *ch;
-	struct xpc_msg *msg = XPC_MSG_ADDRESS(payload);
-	s64 get, msg_number = msg->number;
-
-
-	DBUG_ON(partid <= 0 || partid >= XP_MAX_PARTITIONS);
-	DBUG_ON(ch_number < 0 || ch_number >= part->nchannels);
-
-	ch = &part->channels[ch_number];
-
-	dev_dbg(xpc_chan, "msg=0x%p, msg_number=%ld, partid=%d, channel=%d\n",
-		(void *) msg, msg_number, ch->partid, ch->number);
-
-	DBUG_ON((((u64) msg - (u64) ch->remote_msgqueue) / ch->msg_size) !=
-					msg_number % ch->remote_nentries);
-	DBUG_ON(msg->flags & XPC_M_DONE);
-
-	msg->flags |= XPC_M_DONE;
-
-	/*
-	 * The preceding store of msg->flags must occur before the following
-	 * load of ch->local_GP->get.
-	 */
-	mb();
-
-	/*
-	 * See if this message is next in line to be acknowledged as having
-	 * been delivered.
-	 */
-	get = ch->local_GP->get;
-	if (get == msg_number) {
-		xpc_acknowledge_msgs(ch, get, msg->flags);
-	}
-
-	/* the call to xpc_msgqueue_ref() was done by xpc_deliver_msg()  */
-	xpc_msgqueue_deref(ch);
-}
-
diff --git a/arch/ia64/sn/kernel/xpc_main.c b/arch/ia64/sn/kernel/xpc_main.c
deleted file mode 100644
index 9e0b164..0000000
--- a/arch/ia64/sn/kernel/xpc_main.c
+++ /dev/null
@@ -1,1431 +0,0 @@
-/*
- * This file is subject to the terms and conditions of the GNU General Public
- * License.  See the file "COPYING" in the main directory of this archive
- * for more details.
- *
- * Copyright (c) 2004-2007 Silicon Graphics, Inc.  All Rights Reserved.
- */
-
-
-/*
- * Cross Partition Communication (XPC) support - standard version.
- *
- *	XPC provides a message passing capability that crosses partition
- *	boundaries. This module is made up of two parts:
- *
- *	    partition	This part detects the presence/absence of other
- *			partitions. It provides a heartbeat and monitors
- *			the heartbeats of other partitions.
- *
- *	    channel	This part manages the channels and sends/receives
- *			messages across them to/from other partitions.
- *
- *	There are a couple of additional functions residing in XP, which
- *	provide an interface to XPC for its users.
- *
- *
- *	Caveats:
- *
- *	  . We currently have no way to determine which nasid an IPI came
- *	    from. Thus, xpc_IPI_send() does a remote AMO write followed by
- *	    an IPI. The AMO indicates where data is to be pulled from, so
- *	    after the IPI arrives, the remote partition checks the AMO word.
- *	    The IPI can actually arrive before the AMO however, so other code
- *	    must periodically check for this case. Also, remote AMO operations
- *	    do not reliably time out. Thus we do a remote PIO read solely to
- *	    know whether the remote partition is down and whether we should
- *	    stop sending IPIs to it. This remote PIO read operation is set up
- *	    in a special nofault region so SAL knows to ignore (and cleanup)
- *	    any errors due to the remote AMO write, PIO read, and/or PIO
- *	    write operations.
- *
- *	    If/when new hardware solves this IPI problem, we should abandon
- *	    the current approach.
- *
- */
-
-
-#include <linux/kernel.h>
-#include <linux/module.h>
-#include <linux/init.h>
-#include <linux/sched.h>
-#include <linux/syscalls.h>
-#include <linux/cache.h>
-#include <linux/interrupt.h>
-#include <linux/delay.h>
-#include <linux/reboot.h>
-#include <linux/completion.h>
-#include <linux/kdebug.h>
-#include <asm/sn/intr.h>
-#include <asm/sn/sn_sal.h>
-#include <asm/uaccess.h>
-#include <asm/sn/xpc.h>
-
-
-/* define two XPC debug device structures to be used with dev_dbg() et al */
-
-struct device_driver xpc_dbg_name = {
-	.name = "xpc"
-};
-
-struct device xpc_part_dbg_subname = {
-	.bus_id = {0},		/* set to "part" at xpc_init() time */
-	.driver = &xpc_dbg_name
-};
-
-struct device xpc_chan_dbg_subname = {
-	.bus_id = {0},		/* set to "chan" at xpc_init() time */
-	.driver = &xpc_dbg_name
-};
-
-struct device *xpc_part = &xpc_part_dbg_subname;
-struct device *xpc_chan = &xpc_chan_dbg_subname;
-
-
-static int xpc_kdebug_ignore;
-
-
-/* systune related variables for /proc/sys directories */
-
-static int xpc_hb_interval = XPC_HB_DEFAULT_INTERVAL;
-static int xpc_hb_min_interval = 1;
-static int xpc_hb_max_interval = 10;
-
-static int xpc_hb_check_interval = XPC_HB_CHECK_DEFAULT_INTERVAL;
-static int xpc_hb_check_min_interval = 10;
-static int xpc_hb_check_max_interval = 120;
-
-int xpc_disengage_request_timelimit = XPC_DISENGAGE_REQUEST_DEFAULT_TIMELIMIT;
-static int xpc_disengage_request_min_timelimit = 0;
-static int xpc_disengage_request_max_timelimit = 120;
-
-static ctl_table xpc_sys_xpc_hb_dir[] = {
-	{
-		.ctl_name 	= CTL_UNNUMBERED,
-		.procname	= "hb_interval",
-		.data		= &xpc_hb_interval,
-		.maxlen		= sizeof(int),
-		.mode		= 0644,
-		.proc_handler	= &proc_dointvec_minmax,
-		.strategy	= &sysctl_intvec,
-		.extra1		= &xpc_hb_min_interval,
-		.extra2		= &xpc_hb_max_interval
-	},
-	{
-		.ctl_name	= CTL_UNNUMBERED,
-		.procname	= "hb_check_interval",
-		.data		= &xpc_hb_check_interval,
-		.maxlen		= sizeof(int),
-		.mode		= 0644,
-		.proc_handler	= &proc_dointvec_minmax,
-		.strategy	= &sysctl_intvec,
-		.extra1		= &xpc_hb_check_min_interval,
-		.extra2		= &xpc_hb_check_max_interval
-	},
-	{}
-};
-static ctl_table xpc_sys_xpc_dir[] = {
-	{
-		.ctl_name	= CTL_UNNUMBERED,
-		.procname	= "hb",
-		.mode		= 0555,
-		.child		= xpc_sys_xpc_hb_dir
-	},
-	{
-		.ctl_name	= CTL_UNNUMBERED,
-		.procname	= "disengage_request_timelimit",
-		.data		= &xpc_disengage_request_timelimit,
-		.maxlen		= sizeof(int),
-		.mode		= 0644,
-		.proc_handler	= &proc_dointvec_minmax,
-		.strategy	= &sysctl_intvec,
-		.extra1		= &xpc_disengage_request_min_timelimit,
-		.extra2		= &xpc_disengage_request_max_timelimit
-	},
-	{}
-};
-static ctl_table xpc_sys_dir[] = {
-	{
-		.ctl_name	= CTL_UNNUMBERED,
-		.procname	= "xpc",
-		.mode		= 0555,
-		.child		= xpc_sys_xpc_dir
-	},
-	{}
-};
-static struct ctl_table_header *xpc_sysctl;
-
-/* non-zero if any remote partition disengage request was timed out */
-int xpc_disengage_request_timedout;
-
-/* #of IRQs received */
-static atomic_t xpc_act_IRQ_rcvd;
-
-/* IRQ handler notifies this wait queue on receipt of an IRQ */
-static DECLARE_WAIT_QUEUE_HEAD(xpc_act_IRQ_wq);
-
-static unsigned long xpc_hb_check_timeout;
-
-/* notification that the xpc_hb_checker thread has exited */
-static DECLARE_COMPLETION(xpc_hb_checker_exited);
-
-/* notification that the xpc_discovery thread has exited */
-static DECLARE_COMPLETION(xpc_discovery_exited);
-
-
-static struct timer_list xpc_hb_timer;
-
-
-static void xpc_kthread_waitmsgs(struct xpc_partition *, struct xpc_channel *);
-
-
-static int xpc_system_reboot(struct notifier_block *, unsigned long, void *);
-static struct notifier_block xpc_reboot_notifier = {
-	.notifier_call = xpc_system_reboot,
-};
-
-static int xpc_system_die(struct notifier_block *, unsigned long, void *);
-static struct notifier_block xpc_die_notifier = {
-	.notifier_call = xpc_system_die,
-};
-
-
-/*
- * Timer function to enforce the timelimit on the partition disengage request.
- */
-static void
-xpc_timeout_partition_disengage_request(unsigned long data)
-{
-	struct xpc_partition *part = (struct xpc_partition *) data;
-
-
-	DBUG_ON(time_before(jiffies, part->disengage_request_timeout));
-
-	(void) xpc_partition_disengaged(part);
-
-	DBUG_ON(part->disengage_request_timeout != 0);
-	DBUG_ON(xpc_partition_engaged(1UL << XPC_PARTID(part)) != 0);
-}
-
-
-/*
- * Notify the heartbeat check thread that an IRQ has been received.
- */
-static irqreturn_t
-xpc_act_IRQ_handler(int irq, void *dev_id)
-{
-	atomic_inc(&xpc_act_IRQ_rcvd);
-	wake_up_interruptible(&xpc_act_IRQ_wq);
-	return IRQ_HANDLED;
-}
-
-
-/*
- * Timer to produce the heartbeat.  The timer structures function is
- * already set when this is initially called.  A tunable is used to
- * specify when the next timeout should occur.
- */
-static void
-xpc_hb_beater(unsigned long dummy)
-{
-	xpc_vars->heartbeat++;
-
-	if (time_after_eq(jiffies, xpc_hb_check_timeout)) {
-		wake_up_interruptible(&xpc_act_IRQ_wq);
-	}
-
-	xpc_hb_timer.expires = jiffies + (xpc_hb_interval * HZ);
-	add_timer(&xpc_hb_timer);
-}
-
-
-/*
- * This thread is responsible for nearly all of the partition
- * activation/deactivation.
- */
-static int
-xpc_hb_checker(void *ignore)
-{
-	int last_IRQ_count = 0;
-	int new_IRQ_count;
-	int force_IRQ=0;
-
-
-	/* this thread was marked active by xpc_hb_init() */
-
-	daemonize(XPC_HB_CHECK_THREAD_NAME);
-
-	set_cpus_allowed(current, cpumask_of_cpu(XPC_HB_CHECK_CPU));
-
-	/* set our heartbeating to other partitions into motion */
-	xpc_hb_check_timeout = jiffies + (xpc_hb_check_interval * HZ);
-	xpc_hb_beater(0);
-
-	while (!(volatile int) xpc_exiting) {
-
-		dev_dbg(xpc_part, "woke up with %d ticks rem; %d IRQs have "
-			"been received\n",
-			(int) (xpc_hb_check_timeout - jiffies),
-			atomic_read(&xpc_act_IRQ_rcvd) - last_IRQ_count);
-
-
-		/* checking of remote heartbeats is skewed by IRQ handling */
-		if (time_after_eq(jiffies, xpc_hb_check_timeout)) {
-			dev_dbg(xpc_part, "checking remote heartbeats\n");
-			xpc_check_remote_hb();
-
-			/*
-			 * We need to periodically recheck to ensure no
-			 * IPI/AMO pairs have been missed.  That check
-			 * must always reset xpc_hb_check_timeout.
-			 */
-			force_IRQ = 1;
-		}
-
-
-		/* check for outstanding IRQs */
-		new_IRQ_count = atomic_read(&xpc_act_IRQ_rcvd);
-		if (last_IRQ_count < new_IRQ_count || force_IRQ != 0) {
-			force_IRQ = 0;
-
-			dev_dbg(xpc_part, "found an IRQ to process; will be "
-				"resetting xpc_hb_check_timeout\n");
-
-			last_IRQ_count += xpc_identify_act_IRQ_sender();
-			if (last_IRQ_count < new_IRQ_count) {
-				/* retry once to help avoid missing AMO */
-				(void) xpc_identify_act_IRQ_sender();
-			}
-			last_IRQ_count = new_IRQ_count;
-
-			xpc_hb_check_timeout = jiffies +
-					   (xpc_hb_check_interval * HZ);
-		}
-
-		/* wait for IRQ or timeout */
-		(void) wait_event_interruptible(xpc_act_IRQ_wq,
-			    (last_IRQ_count < atomic_read(&xpc_act_IRQ_rcvd) ||
-					time_after_eq(jiffies, xpc_hb_check_timeout) ||
-						(volatile int) xpc_exiting));
-	}
-
-	dev_dbg(xpc_part, "heartbeat checker is exiting\n");
-
-
-	/* mark this thread as having exited */
-	complete(&xpc_hb_checker_exited);
-	return 0;
-}
-
-
-/*
- * This thread will attempt to discover other partitions to activate
- * based on info provided by SAL. This new thread is short lived and
- * will exit once discovery is complete.
- */
-static int
-xpc_initiate_discovery(void *ignore)
-{
-	daemonize(XPC_DISCOVERY_THREAD_NAME);
-
-	xpc_discovery();
-
-	dev_dbg(xpc_part, "discovery thread is exiting\n");
-
-	/* mark this thread as having exited */
-	complete(&xpc_discovery_exited);
-	return 0;
-}
-
-
-/*
- * Establish first contact with the remote partititon. This involves pulling
- * the XPC per partition variables from the remote partition and waiting for
- * the remote partition to pull ours.
- */
-static enum xpc_retval
-xpc_make_first_contact(struct xpc_partition *part)
-{
-	enum xpc_retval ret;
-
-
-	while ((ret = xpc_pull_remote_vars_part(part)) != xpcSuccess) {
-		if (ret != xpcRetry) {
-			XPC_DEACTIVATE_PARTITION(part, ret);
-			return ret;
-		}
-
-		dev_dbg(xpc_chan, "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_DEACTIVATING) {
-			return part->reason;
-		}
-	}
-
-	return xpc_mark_partition_active(part);
-}
-
-
-/*
- * The first kthread assigned to a newly activated partition is the one
- * created by XPC HB with which it calls xpc_partition_up(). XPC hangs on to
- * that kthread until the partition is brought down, at which time that kthread
- * returns back to XPC HB. (The return of that kthread will signify to XPC HB
- * that XPC has dismantled all communication infrastructure for the associated
- * partition.) This kthread becomes the channel manager for that partition.
- *
- * Each active partition has a channel manager, who, besides connecting and
- * disconnecting channels, will ensure that each of the partition's connected
- * channels has the required number of assigned kthreads to get the work done.
- */
-static void
-xpc_channel_mgr(struct xpc_partition *part)
-{
-	while (part->act_state != XPC_P_DEACTIVATING ||
-			atomic_read(&part->nchannels_active) > 0 ||
-					!xpc_partition_disengaged(part)) {
-
-		xpc_process_channel_activity(part);
-
-
-		/*
-		 * Wait until we've been requested to activate kthreads or
-		 * all of the channel's message queues have been torn down or
-		 * a signal is pending.
-		 *
-		 * The channel_mgr_requests is set to 1 after being awakened,
-		 * This is done to prevent the channel mgr from making one pass
-		 * through the loop for each request, since he will
-		 * be servicing all the requests in one pass. The reason it's
-		 * set to 1 instead of 0 is so that other kthreads will know
-		 * that the channel mgr is running and won't bother trying to
-		 * wake him up.
-		 */
-		atomic_dec(&part->channel_mgr_requests);
-		(void) wait_event_interruptible(part->channel_mgr_wq,
-				(atomic_read(&part->channel_mgr_requests) > 0 ||
-				(volatile u64) part->local_IPI_amo != 0 ||
-				((volatile u8) part->act_state ==
-							XPC_P_DEACTIVATING &&
-				atomic_read(&part->nchannels_active) == 0 &&
-				xpc_partition_disengaged(part))));
-		atomic_set(&part->channel_mgr_requests, 1);
-
-		// >>> Does it need to wakeup periodically as well? In case we
-		// >>> miscalculated the #of kthreads to wakeup or create?
-	}
-}
-
-
-/*
- * When XPC HB determines that a partition has come up, it will create a new
- * kthread and that kthread will call this function to attempt to set up the
- * basic infrastructure used for Cross Partition Communication with the newly
- * upped partition.
- *
- * The kthread that was created by XPC HB and which setup the XPC
- * infrastructure will remain assigned to the partition until the partition
- * goes down. At which time the kthread will teardown the XPC infrastructure
- * and then exit.
- *
- * XPC HB will put the remote partition's XPC per partition specific variables
- * physical address into xpc_partitions[partid].remote_vars_part_pa prior to
- * calling xpc_partition_up().
- */
-static void
-xpc_partition_up(struct xpc_partition *part)
-{
-	DBUG_ON(part->channels != NULL);
-
-	dev_dbg(xpc_chan, "activating partition %d\n", XPC_PARTID(part));
-
-	if (xpc_setup_infrastructure(part) != xpcSuccess) {
-		return;
-	}
-
-	/*
-	 * The kthread that XPC HB called us with will become the
-	 * channel manager for this partition. It will not return
-	 * back to XPC HB until the partition's XPC infrastructure
-	 * has been dismantled.
-	 */
-
-	(void) xpc_part_ref(part);	/* this will always succeed */
-
-	if (xpc_make_first_contact(part) == xpcSuccess) {
-		xpc_channel_mgr(part);
-	}
-
-	xpc_part_deref(part);
-
-	xpc_teardown_infrastructure(part);
-}
-
-
-static int
-xpc_activating(void *__partid)
-{
-	partid_t partid = (u64) __partid;
-	struct xpc_partition *part = &xpc_partitions[partid];
-	unsigned long irq_flags;
-	struct sched_param param = { .sched_priority = MAX_RT_PRIO - 1 };
-	int ret;
-
-
-	DBUG_ON(partid <= 0 || partid >= XP_MAX_PARTITIONS);
-
-	spin_lock_irqsave(&part->act_lock, irq_flags);
-
-	if (part->act_state == XPC_P_DEACTIVATING) {
-		part->act_state = XPC_P_INACTIVE;
-		spin_unlock_irqrestore(&part->act_lock, irq_flags);
-		part->remote_rp_pa = 0;
-		return 0;
-	}
-
-	/* indicate the thread is activating */
-	DBUG_ON(part->act_state != XPC_P_ACTIVATION_REQ);
-	part->act_state = XPC_P_ACTIVATING;
-
-	XPC_SET_REASON(part, 0, 0);
-	spin_unlock_irqrestore(&part->act_lock, irq_flags);
-
-	dev_dbg(xpc_part, "bringing partition %d up\n", partid);
-
-	daemonize("xpc%02d", partid);
-
-	/*
-	 * This thread needs to run at a realtime priority to prevent a
-	 * significant performance degradation.
-	 */
-	ret = sched_setscheduler(current, SCHED_FIFO, &param);
-	if (ret != 0) {
-		dev_warn(xpc_part, "unable to set pid %d to a realtime "
-			"priority, ret=%d\n", current->pid, ret);
-	}
-
-	/* allow this thread and its children to run on any CPU */
-	set_cpus_allowed(current, CPU_MASK_ALL);
-
-	/*
-	 * Register the remote partition's AMOs with SAL so it can handle
-	 * and cleanup errors within that address range should the remote
-	 * partition go down. We don't unregister this range because it is
-	 * difficult to tell when outstanding writes to the remote partition
-	 * are finished and thus when it is safe to unregister. This should
-	 * not result in wasted space in the SAL xp_addr_region table because
-	 * we should get the same page for remote_amos_page_pa after module
-	 * reloads and system reboots.
-	 */
-	if (sn_register_xp_addr_region(part->remote_amos_page_pa,
-							PAGE_SIZE, 1) < 0) {
-		dev_warn(xpc_part, "xpc_partition_up(%d) failed to register "
-			"xp_addr region\n", partid);
-
-		spin_lock_irqsave(&part->act_lock, irq_flags);
-		part->act_state = XPC_P_INACTIVE;
-		XPC_SET_REASON(part, xpcPhysAddrRegFailed, __LINE__);
-		spin_unlock_irqrestore(&part->act_lock, irq_flags);
-		part->remote_rp_pa = 0;
-		return 0;
-	}
-
-	xpc_allow_hb(partid, xpc_vars);
-	xpc_IPI_send_activated(part);
-
-
-	/*
-	 * xpc_partition_up() holds this thread and marks this partition as
-	 * XPC_P_ACTIVE by calling xpc_hb_mark_active().
-	 */
-	(void) xpc_partition_up(part);
-
-	xpc_disallow_hb(partid, xpc_vars);
-	xpc_mark_partition_inactive(part);
-
-	if (part->reason == xpcReactivating) {
-		/* interrupting ourselves results in activating partition */
-		xpc_IPI_send_reactivate(part);
-	}
-
-	return 0;
-}
-
-
-void
-xpc_activate_partition(struct xpc_partition *part)
-{
-	partid_t partid = XPC_PARTID(part);
-	unsigned long irq_flags;
-	pid_t pid;
-
-
-	spin_lock_irqsave(&part->act_lock, irq_flags);
-
-	DBUG_ON(part->act_state != XPC_P_INACTIVE);
-
-	part->act_state = XPC_P_ACTIVATION_REQ;
-	XPC_SET_REASON(part, xpcCloneKThread, __LINE__);
-
-	spin_unlock_irqrestore(&part->act_lock, irq_flags);
-
-	pid = kernel_thread(xpc_activating, (void *) ((u64) partid), 0);
-
-	if (unlikely(pid <= 0)) {
-		spin_lock_irqsave(&part->act_lock, irq_flags);
-		part->act_state = XPC_P_INACTIVE;
-		XPC_SET_REASON(part, xpcCloneKThreadFailed, __LINE__);
-		spin_unlock_irqrestore(&part->act_lock, irq_flags);
-	}
-}
-
-
-/*
- * Handle the receipt of a SGI_XPC_NOTIFY IRQ by seeing whether the specified
- * partition actually sent it. Since SGI_XPC_NOTIFY IRQs may be shared by more
- * than one partition, we use an AMO_t structure per partition to indicate
- * whether a partition has sent an IPI or not.  >>> If it has, then wake up the
- * associated kthread to handle it.
- *
- * All SGI_XPC_NOTIFY IRQs received by XPC are the result of IPIs sent by XPC
- * running on other partitions.
- *
- * Noteworthy Arguments:
- *
- *	irq - Interrupt ReQuest number. NOT USED.
- *
- *	dev_id - partid of IPI's potential sender.
- */
-irqreturn_t
-xpc_notify_IRQ_handler(int irq, void *dev_id)
-{
-	partid_t partid = (partid_t) (u64) dev_id;
-	struct xpc_partition *part = &xpc_partitions[partid];
-
-
-	DBUG_ON(partid <= 0 || partid >= XP_MAX_PARTITIONS);
-
-	if (xpc_part_ref(part)) {
-		xpc_check_for_channel_activity(part);
-
-		xpc_part_deref(part);
-	}
-	return IRQ_HANDLED;
-}
-
-
-/*
- * Check to see if xpc_notify_IRQ_handler() dropped any IPIs on the floor
- * because the write to their associated IPI amo completed after the IRQ/IPI
- * was received.
- */
-void
-xpc_dropped_IPI_check(struct xpc_partition *part)
-{
-	if (xpc_part_ref(part)) {
-		xpc_check_for_channel_activity(part);
-
-		part->dropped_IPI_timer.expires = jiffies +
-							XPC_P_DROPPED_IPI_WAIT;
-		add_timer(&part->dropped_IPI_timer);
-		xpc_part_deref(part);
-	}
-}
-
-
-void
-xpc_activate_kthreads(struct xpc_channel *ch, int needed)
-{
-	int idle = atomic_read(&ch->kthreads_idle);
-	int assigned = atomic_read(&ch->kthreads_assigned);
-	int wakeup;
-
-
-	DBUG_ON(needed <= 0);
-
-	if (idle > 0) {
-		wakeup = (needed > idle) ? idle : needed;
-		needed -= wakeup;
-
-		dev_dbg(xpc_chan, "wakeup %d idle kthreads, partid=%d, "
-			"channel=%d\n", wakeup, ch->partid, ch->number);
-
-		/* only wakeup the requested number of kthreads */
-		wake_up_nr(&ch->idle_wq, wakeup);
-	}
-
-	if (needed <= 0) {
-		return;
-	}
-
-	if (needed + assigned > ch->kthreads_assigned_limit) {
-		needed = ch->kthreads_assigned_limit - assigned;
-		// >>>should never be less than 0
-		if (needed <= 0) {
-			return;
-		}
-	}
-
-	dev_dbg(xpc_chan, "create %d new kthreads, partid=%d, channel=%d\n",
-		needed, ch->partid, ch->number);
-
-	xpc_create_kthreads(ch, needed, 0);
-}
-
-
-/*
- * This function is where XPC's kthreads wait for messages to deliver.
- */
-static void
-xpc_kthread_waitmsgs(struct xpc_partition *part, struct xpc_channel *ch)
-{
-	do {
-		/* deliver messages to their intended recipients */
-
-		while ((volatile s64) ch->w_local_GP.get <
-				(volatile s64) ch->w_remote_GP.put &&
-					!((volatile u32) ch->flags &
-						XPC_C_DISCONNECTING)) {
-			xpc_deliver_msg(ch);
-		}
-
-		if (atomic_inc_return(&ch->kthreads_idle) >
-						ch->kthreads_idle_limit) {
-			/* too many idle kthreads on this channel */
-			atomic_dec(&ch->kthreads_idle);
-			break;
-		}
-
-		dev_dbg(xpc_chan, "idle kthread calling "
-			"wait_event_interruptible_exclusive()\n");
-
-		(void) wait_event_interruptible_exclusive(ch->idle_wq,
-				((volatile s64) ch->w_local_GP.get <
-					(volatile s64) ch->w_remote_GP.put ||
-				((volatile u32) ch->flags &
-						XPC_C_DISCONNECTING)));
-
-		atomic_dec(&ch->kthreads_idle);
-
-	} while (!((volatile u32) ch->flags & XPC_C_DISCONNECTING));
-}
-
-
-static int
-xpc_daemonize_kthread(void *args)
-{
-	partid_t partid = XPC_UNPACK_ARG1(args);
-	u16 ch_number = XPC_UNPACK_ARG2(args);
-	struct xpc_partition *part = &xpc_partitions[partid];
-	struct xpc_channel *ch;
-	int n_needed;
-	unsigned long irq_flags;
-
-
-	daemonize("xpc%02dc%d", partid, ch_number);
-
-	dev_dbg(xpc_chan, "kthread starting, partid=%d, channel=%d\n",
-		partid, ch_number);
-
-	ch = &part->channels[ch_number];
-
-	if (!(ch->flags & XPC_C_DISCONNECTING)) {
-
-		/* let registerer know that connection has been established */
-
-		spin_lock_irqsave(&ch->lock, irq_flags);
-		if (!(ch->flags & XPC_C_CONNECTEDCALLOUT)) {
-			ch->flags |= XPC_C_CONNECTEDCALLOUT;
-			spin_unlock_irqrestore(&ch->lock, irq_flags);
-
-			xpc_connected_callout(ch);
-
-			spin_lock_irqsave(&ch->lock, irq_flags);
-			ch->flags |= XPC_C_CONNECTEDCALLOUT_MADE;
-			spin_unlock_irqrestore(&ch->lock, irq_flags);
-
-			/*
-			 * It is possible that while the callout was being
-			 * made that the remote partition sent some messages.
-			 * If that is the case, we may need to activate
-			 * additional kthreads to help deliver them. We only
-			 * need one less than total #of messages to deliver.
-			 */
-			n_needed = ch->w_remote_GP.put - ch->w_local_GP.get - 1;
-			if (n_needed > 0 &&
-					!(ch->flags & XPC_C_DISCONNECTING)) {
-				xpc_activate_kthreads(ch, n_needed);
-			}
-		} else {
-			spin_unlock_irqrestore(&ch->lock, irq_flags);
-		}
-
-		xpc_kthread_waitmsgs(part, ch);
-	}
-
-	/* let registerer know that connection is disconnecting */
-
-	spin_lock_irqsave(&ch->lock, irq_flags);
-	if ((ch->flags & XPC_C_CONNECTEDCALLOUT_MADE) &&
-			!(ch->flags & XPC_C_DISCONNECTINGCALLOUT)) {
-		ch->flags |= XPC_C_DISCONNECTINGCALLOUT;
-		spin_unlock_irqrestore(&ch->lock, irq_flags);
-
-		xpc_disconnect_callout(ch, xpcDisconnecting);
-
-		spin_lock_irqsave(&ch->lock, irq_flags);
-		ch->flags |= XPC_C_DISCONNECTINGCALLOUT_MADE;
-	}
-	spin_unlock_irqrestore(&ch->lock, irq_flags);
-
-	if (atomic_dec_return(&ch->kthreads_assigned) == 0) {
-		if (atomic_dec_return(&part->nchannels_engaged) == 0) {
-			xpc_mark_partition_disengaged(part);
-			xpc_IPI_send_disengage(part);
-		}
-	}
-
-	xpc_msgqueue_deref(ch);
-
-	dev_dbg(xpc_chan, "kthread exiting, partid=%d, channel=%d\n",
-		partid, ch_number);
-
-	xpc_part_deref(part);
-	return 0;
-}
-
-
-/*
- * For each partition that XPC has established communications with, there is
- * a minimum of one kernel thread assigned to perform any operation that
- * may potentially sleep or block (basically the callouts to the asynchronous
- * functions registered via xpc_connect()).
- *
- * Additional kthreads are created and destroyed by XPC as the workload
- * demands.
- *
- * A kthread is assigned to one of the active channels that exists for a given
- * partition.
- */
-void
-xpc_create_kthreads(struct xpc_channel *ch, int needed,
-			int ignore_disconnecting)
-{
-	unsigned long irq_flags;
-	pid_t pid;
-	u64 args = XPC_PACK_ARGS(ch->partid, ch->number);
-	struct xpc_partition *part = &xpc_partitions[ch->partid];
-
-
-	while (needed-- > 0) {
-
-		/*
-		 * The following is done on behalf of the newly created
-		 * kthread. That kthread is responsible for doing the
-		 * counterpart to the following before it exits.
-		 */
-		if (ignore_disconnecting) {
-			if (!atomic_inc_not_zero(&ch->kthreads_assigned)) {
-				/* kthreads assigned had gone to zero */
-				BUG_ON(!(ch->flags &
-					XPC_C_DISCONNECTINGCALLOUT_MADE));
-				break;
-			}
-
-		} else if (ch->flags & XPC_C_DISCONNECTING) {
-			break;
-
-		} else if (atomic_inc_return(&ch->kthreads_assigned) == 1) {
-			if (atomic_inc_return(&part->nchannels_engaged) == 1)
-				xpc_mark_partition_engaged(part);
-		}
-		(void) xpc_part_ref(part);
-		xpc_msgqueue_ref(ch);
-
-		pid = kernel_thread(xpc_daemonize_kthread, (void *) args, 0);
-		if (pid < 0) {
-			/* the fork failed */
-
-			/*
-			 * NOTE: if (ignore_disconnecting &&
-			 * !(ch->flags & XPC_C_DISCONNECTINGCALLOUT)) is true,
-			 * then we'll deadlock if all other kthreads assigned
-			 * to this channel are blocked in the channel's
-			 * registerer, because the only thing that will unblock
-			 * them is the xpcDisconnecting callout that this
-			 * failed kernel_thread would have made.
-			 */
-
-			if (atomic_dec_return(&ch->kthreads_assigned) == 0 &&
-			    atomic_dec_return(&part->nchannels_engaged) == 0) {
-				xpc_mark_partition_disengaged(part);
-				xpc_IPI_send_disengage(part);
-			}
-			xpc_msgqueue_deref(ch);
-			xpc_part_deref(part);
-
-			if (atomic_read(&ch->kthreads_assigned) <
-						ch->kthreads_idle_limit) {
-				/*
-				 * Flag this as an error only if we have an
-				 * insufficient #of kthreads for the channel
-				 * to function.
-				 */
-				spin_lock_irqsave(&ch->lock, irq_flags);
-				XPC_DISCONNECT_CHANNEL(ch, xpcLackOfResources,
-								&irq_flags);
-				spin_unlock_irqrestore(&ch->lock, irq_flags);
-			}
-			break;
-		}
-
-		ch->kthreads_created++;	// >>> temporary debug only!!!
-	}
-}
-
-
-void
-xpc_disconnect_wait(int ch_number)
-{
-	unsigned long irq_flags;
-	partid_t partid;
-	struct xpc_partition *part;
-	struct xpc_channel *ch;
-	int wakeup_channel_mgr;
-
-
-	/* now wait for all callouts to the caller's function to cease */
-	for (partid = 1; partid < XP_MAX_PARTITIONS; partid++) {
-		part = &xpc_partitions[partid];
-
-		if (!xpc_part_ref(part)) {
-			continue;
-		}
-
-		ch = &part->channels[ch_number];
-
-		if (!(ch->flags & XPC_C_WDISCONNECT)) {
-			xpc_part_deref(part);
-			continue;
-		}
-
-		wait_for_completion(&ch->wdisconnect_wait);
-
-		spin_lock_irqsave(&ch->lock, irq_flags);
-		DBUG_ON(!(ch->flags & XPC_C_DISCONNECTED));
-		wakeup_channel_mgr = 0;
-
-		if (ch->delayed_IPI_flags) {
-			if (part->act_state != XPC_P_DEACTIVATING) {
-				spin_lock(&part->IPI_lock);
-				XPC_SET_IPI_FLAGS(part->local_IPI_amo,
-					ch->number, ch->delayed_IPI_flags);
-				spin_unlock(&part->IPI_lock);
-				wakeup_channel_mgr = 1;
-			}
-			ch->delayed_IPI_flags = 0;
-		}
-
-		ch->flags &= ~XPC_C_WDISCONNECT;
-		spin_unlock_irqrestore(&ch->lock, irq_flags);
-
-		if (wakeup_channel_mgr) {
-			xpc_wakeup_channel_mgr(part);
-		}
-
-		xpc_part_deref(part);
-	}
-}
-
-
-static void
-xpc_do_exit(enum xpc_retval reason)
-{
-	partid_t partid;
-	int active_part_count, printed_waiting_msg = 0;
-	struct xpc_partition *part;
-	unsigned long printmsg_time, disengage_request_timeout = 0;
-
-
-	/* a 'rmmod XPC' and a 'reboot' cannot both end up here together */
-	DBUG_ON(xpc_exiting == 1);
-
-	/*
-	 * Let the heartbeat checker thread and the discovery thread
-	 * (if one is running) know that they should exit. Also wake up
-	 * the heartbeat checker thread in case it's sleeping.
-	 */
-	xpc_exiting = 1;
-	wake_up_interruptible(&xpc_act_IRQ_wq);
-
-	/* ignore all incoming interrupts */
-	free_irq(SGI_XPC_ACTIVATE, NULL);
-
-	/* wait for the discovery thread to exit */
-	wait_for_completion(&xpc_discovery_exited);
-
-	/* wait for the heartbeat checker thread to exit */
-	wait_for_completion(&xpc_hb_checker_exited);
-
-
-	/* sleep for a 1/3 of a second or so */
-	(void) msleep_interruptible(300);
-
-
-	/* wait for all partitions to become inactive */
-
-	printmsg_time = jiffies + (XPC_DISENGAGE_PRINTMSG_INTERVAL * HZ);
-	xpc_disengage_request_timedout = 0;
-
-	do {
-		active_part_count = 0;
-
-		for (partid = 1; partid < XP_MAX_PARTITIONS; partid++) {
-			part = &xpc_partitions[partid];
-
-			if (xpc_partition_disengaged(part) &&
-					part->act_state == XPC_P_INACTIVE) {
-				continue;
-			}
-
-			active_part_count++;
-
-			XPC_DEACTIVATE_PARTITION(part, reason);
-
-			if (part->disengage_request_timeout >
-						disengage_request_timeout) {
-				disengage_request_timeout =
-						part->disengage_request_timeout;
-			}
-		}
-
-		if (xpc_partition_engaged(-1UL)) {
-			if (time_after(jiffies, printmsg_time)) {
-				dev_info(xpc_part, "waiting for remote "
-					"partitions to disengage, timeout in "
-					"%ld seconds\n",
-					(disengage_request_timeout - jiffies)
-									/ HZ);
-				printmsg_time = jiffies +
-					(XPC_DISENGAGE_PRINTMSG_INTERVAL * HZ);
-				printed_waiting_msg = 1;
-			}
-
-		} else if (active_part_count > 0) {
-			if (printed_waiting_msg) {
-				dev_info(xpc_part, "waiting for local partition"
-					" to disengage\n");
-				printed_waiting_msg = 0;
-			}
-
-		} else {
-			if (!xpc_disengage_request_timedout) {
-				dev_info(xpc_part, "all partitions have "
-					"disengaged\n");
-			}
-			break;
-		}
-
-		/* sleep for a 1/3 of a second or so */
-		(void) msleep_interruptible(300);
-
-	} while (1);
-
-	DBUG_ON(xpc_partition_engaged(-1UL));
-
-
-	/* indicate to others that our reserved page is uninitialized */
-	xpc_rsvd_page->vars_pa = 0;
-
-	/* now it's time to eliminate our heartbeat */
-	del_timer_sync(&xpc_hb_timer);
-	DBUG_ON(xpc_vars->heartbeating_to_mask != 0);
-
-	if (reason == xpcUnloading) {
-		/* take ourselves off of the reboot_notifier_list */
-		(void) unregister_reboot_notifier(&xpc_reboot_notifier);
-
-		/* take ourselves off of the die_notifier list */
-		(void) unregister_die_notifier(&xpc_die_notifier);
-	}
-
-	/* close down protections for IPI operations */
-	xpc_restrict_IPI_ops();
-
-
-	/* clear the interface to XPC's functions */
-	xpc_clear_interface();
-
-	if (xpc_sysctl) {
-		unregister_sysctl_table(xpc_sysctl);
-	}
-
-	kfree(xpc_remote_copy_buffer_base);
-}
-
-
-/*
- * This function is called when the system is being rebooted.
- */
-static int
-xpc_system_reboot(struct notifier_block *nb, unsigned long event, void *unused)
-{
-	enum xpc_retval reason;
-
-
-	switch (event) {
-	case SYS_RESTART:
-		reason = xpcSystemReboot;
-		break;
-	case SYS_HALT:
-		reason = xpcSystemHalt;
-		break;
-	case SYS_POWER_OFF:
-		reason = xpcSystemPoweroff;
-		break;
-	default:
-		reason = xpcSystemGoingDown;
-	}
-
-	xpc_do_exit(reason);
-	return NOTIFY_DONE;
-}
-
-
-/*
- * Notify other partitions to disengage from all references to our memory.
- */
-static void
-xpc_die_disengage(void)
-{
-	struct xpc_partition *part;
-	partid_t partid;
-	unsigned long engaged;
-	long time, printmsg_time, disengage_request_timeout;
-
-
-	/* keep xpc_hb_checker thread from doing anything (just in case) */
-	xpc_exiting = 1;
-
-	xpc_vars->heartbeating_to_mask = 0;  /* indicate we're deactivated */
-
-	for (partid = 1; partid < XP_MAX_PARTITIONS; partid++) {
-		part = &xpc_partitions[partid];
-
-		if (!XPC_SUPPORTS_DISENGAGE_REQUEST(part->
-							remote_vars_version)) {
-
-			/* just in case it was left set by an earlier XPC */
-			xpc_clear_partition_engaged(1UL << partid);
-			continue;
-		}
-
-		if (xpc_partition_engaged(1UL << partid) ||
-					part->act_state != XPC_P_INACTIVE) {
-			xpc_request_partition_disengage(part);
-			xpc_mark_partition_disengaged(part);
-			xpc_IPI_send_disengage(part);
-		}
-	}
-
-	time = rtc_time();
-	printmsg_time = time +
-		(XPC_DISENGAGE_PRINTMSG_INTERVAL * sn_rtc_cycles_per_second);
-	disengage_request_timeout = time +
-		(xpc_disengage_request_timelimit * sn_rtc_cycles_per_second);
-
-	/* wait for all other partitions to disengage from us */
-
-	while (1) {
-		engaged = xpc_partition_engaged(-1UL);
-		if (!engaged) {
-			dev_info(xpc_part, "all partitions have disengaged\n");
-			break;
-		}
-
-		time = rtc_time();
-		if (time >= disengage_request_timeout) {
-			for (partid = 1; partid < XP_MAX_PARTITIONS; partid++) {
-				if (engaged & (1UL << partid)) {
-					dev_info(xpc_part, "disengage from "
-						"remote partition %d timed "
-						"out\n", partid);
-				}
-			}
-			break;
-		}
-
-		if (time >= printmsg_time) {
-			dev_info(xpc_part, "waiting for remote partitions to "
-				"disengage, timeout in %ld seconds\n",
-				(disengage_request_timeout - time) /
-						sn_rtc_cycles_per_second);
-			printmsg_time = time +
-					(XPC_DISENGAGE_PRINTMSG_INTERVAL *
-						sn_rtc_cycles_per_second);
-		}
-	}
-}
-
-
-/*
- * This function is called when the system is being restarted or halted due
- * to some sort of system failure. If this is the case we need to notify the
- * other partitions to disengage from all references to our memory.
- * This function can also be called when our heartbeater could be offlined
- * for a time. In this case we need to notify other partitions to not worry
- * about the lack of a heartbeat.
- */
-static int
-xpc_system_die(struct notifier_block *nb, unsigned long event, void *unused)
-{
-	switch (event) {
-	case DIE_MACHINE_RESTART:
-	case DIE_MACHINE_HALT:
-		xpc_die_disengage();
-		break;
-
-	case DIE_KDEBUG_ENTER:
-		/* Should lack of heartbeat be ignored by other partitions? */
-		if (!xpc_kdebug_ignore) {
-			break;
-		}
-		/* fall through */
-	case DIE_MCA_MONARCH_ENTER:
-	case DIE_INIT_MONARCH_ENTER:
-		xpc_vars->heartbeat++;
-		xpc_vars->heartbeat_offline = 1;
-		break;
-
-	case DIE_KDEBUG_LEAVE:
-		/* Is lack of heartbeat being ignored by other partitions? */
-		if (!xpc_kdebug_ignore) {
-			break;
-		}
-		/* fall through */
-	case DIE_MCA_MONARCH_LEAVE:
-	case DIE_INIT_MONARCH_LEAVE:
-		xpc_vars->heartbeat++;
-		xpc_vars->heartbeat_offline = 0;
-		break;
-	}
-
-	return NOTIFY_DONE;
-}
-
-
-int __init
-xpc_init(void)
-{
-	int ret;
-	partid_t partid;
-	struct xpc_partition *part;
-	pid_t pid;
-	size_t buf_size;
-
-
-	if (!ia64_platform_is("sn2")) {
-		return -ENODEV;
-	}
-
-
-	buf_size = max(XPC_RP_VARS_SIZE,
-				XPC_RP_HEADER_SIZE + XP_NASID_MASK_BYTES);
-	xpc_remote_copy_buffer = xpc_kmalloc_cacheline_aligned(buf_size,
-				     GFP_KERNEL, &xpc_remote_copy_buffer_base);
-	if (xpc_remote_copy_buffer == NULL)
-		return -ENOMEM;
-
-	snprintf(xpc_part->bus_id, BUS_ID_SIZE, "part");
-	snprintf(xpc_chan->bus_id, BUS_ID_SIZE, "chan");
-
-	xpc_sysctl = register_sysctl_table(xpc_sys_dir);
-
-	/*
-	 * The first few fields of each entry of xpc_partitions[] need to
-	 * be initialized now so that calls to xpc_connect() and
-	 * xpc_disconnect() can be made prior to the activation of any remote
-	 * partition. NOTE THAT NONE OF THE OTHER FIELDS BELONGING TO THESE
-	 * ENTRIES ARE MEANINGFUL UNTIL AFTER AN ENTRY'S CORRESPONDING
-	 * PARTITION HAS BEEN ACTIVATED.
-	 */
-	for (partid = 1; partid < XP_MAX_PARTITIONS; partid++) {
-		part = &xpc_partitions[partid];
-
-		DBUG_ON((u64) part != L1_CACHE_ALIGN((u64) part));
-
-		part->act_IRQ_rcvd = 0;
-		spin_lock_init(&part->act_lock);
-		part->act_state = XPC_P_INACTIVE;
-		XPC_SET_REASON(part, 0, 0);
-
-		init_timer(&part->disengage_request_timer);
-		part->disengage_request_timer.function =
-				xpc_timeout_partition_disengage_request;
-		part->disengage_request_timer.data = (unsigned long) part;
-
-		part->setup_state = XPC_P_UNSET;
-		init_waitqueue_head(&part->teardown_wq);
-		atomic_set(&part->references, 0);
-	}
-
-	/*
-	 * Open up protections for IPI operations (and AMO operations on
-	 * Shub 1.1 systems).
-	 */
-	xpc_allow_IPI_ops();
-
-	/*
-	 * Interrupts being processed will increment this atomic variable and
-	 * awaken the heartbeat thread which will process the interrupts.
-	 */
-	atomic_set(&xpc_act_IRQ_rcvd, 0);
-
-	/*
-	 * This is safe to do before the xpc_hb_checker thread has started
-	 * because the handler releases a wait queue.  If an interrupt is
-	 * received before the thread is waiting, it will not go to sleep,
-	 * but rather immediately process the interrupt.
-	 */
-	ret = request_irq(SGI_XPC_ACTIVATE, xpc_act_IRQ_handler, 0,
-							"xpc hb", NULL);
-	if (ret != 0) {
-		dev_err(xpc_part, "can't register ACTIVATE IRQ handler, "
-			"errno=%d\n", -ret);
-
-		xpc_restrict_IPI_ops();
-
-		if (xpc_sysctl) {
-			unregister_sysctl_table(xpc_sysctl);
-		}
-
-		kfree(xpc_remote_copy_buffer_base);
-		return -EBUSY;
-	}
-
-	/*
-	 * Fill the partition reserved page with the information needed by
-	 * other partitions to discover we are alive and establish initial
-	 * communications.
-	 */
-	xpc_rsvd_page = xpc_rsvd_page_init();
-	if (xpc_rsvd_page == NULL) {
-		dev_err(xpc_part, "could not setup our reserved page\n");
-
-		free_irq(SGI_XPC_ACTIVATE, NULL);
-		xpc_restrict_IPI_ops();
-
-		if (xpc_sysctl) {
-			unregister_sysctl_table(xpc_sysctl);
-		}
-
-		kfree(xpc_remote_copy_buffer_base);
-		return -EBUSY;
-	}
-
-
-	/* add ourselves to the reboot_notifier_list */
-	ret = register_reboot_notifier(&xpc_reboot_notifier);
-	if (ret != 0) {
-		dev_warn(xpc_part, "can't register reboot notifier\n");
-	}
-
-	/* add ourselves to the die_notifier list */
-	ret = register_die_notifier(&xpc_die_notifier);
-	if (ret != 0) {
-		dev_warn(xpc_part, "can't register die notifier\n");
-	}
-
-	init_timer(&xpc_hb_timer);
-	xpc_hb_timer.function = xpc_hb_beater;
-
-	/*
-	 * The real work-horse behind xpc.  This processes incoming
-	 * interrupts and monitors remote heartbeats.
-	 */
-	pid = kernel_thread(xpc_hb_checker, NULL, 0);
-	if (pid < 0) {
-		dev_err(xpc_part, "failed while forking hb check thread\n");
-
-		/* indicate to others that our reserved page is uninitialized */
-		xpc_rsvd_page->vars_pa = 0;
-
-		/* take ourselves off of the reboot_notifier_list */
-		(void) unregister_reboot_notifier(&xpc_reboot_notifier);
-
-		/* take ourselves off of the die_notifier list */
-		(void) unregister_die_notifier(&xpc_die_notifier);
-
-		del_timer_sync(&xpc_hb_timer);
-		free_irq(SGI_XPC_ACTIVATE, NULL);
-		xpc_restrict_IPI_ops();
-
-		if (xpc_sysctl) {
-			unregister_sysctl_table(xpc_sysctl);
-		}
-
-		kfree(xpc_remote_copy_buffer_base);
-		return -EBUSY;
-	}
-
-
-	/*
-	 * Startup a thread that will attempt to discover other partitions to
-	 * activate based on info provided by SAL. This new thread is short
-	 * lived and will exit once discovery is complete.
-	 */
-	pid = kernel_thread(xpc_initiate_discovery, NULL, 0);
-	if (pid < 0) {
-		dev_err(xpc_part, "failed while forking discovery thread\n");
-
-		/* mark this new thread as a non-starter */
-		complete(&xpc_discovery_exited);
-
-		xpc_do_exit(xpcUnloading);
-		return -EBUSY;
-	}
-
-
-	/* set the interface to point at XPC's functions */
-	xpc_set_interface(xpc_initiate_connect, xpc_initiate_disconnect,
-			  xpc_initiate_allocate, xpc_initiate_send,
-			  xpc_initiate_send_notify, xpc_initiate_received,
-			  xpc_initiate_partid_to_nasids);
-
-	return 0;
-}
-module_init(xpc_init);
-
-
-void __exit
-xpc_exit(void)
-{
-	xpc_do_exit(xpcUnloading);
-}
-module_exit(xpc_exit);
-
-
-MODULE_AUTHOR("Silicon Graphics, Inc.");
-MODULE_DESCRIPTION("Cross Partition Communication (XPC) support");
-MODULE_LICENSE("GPL");
-
-module_param(xpc_hb_interval, int, 0);
-MODULE_PARM_DESC(xpc_hb_interval, "Number of seconds between "
-		"heartbeat increments.");
-
-module_param(xpc_hb_check_interval, int, 0);
-MODULE_PARM_DESC(xpc_hb_check_interval, "Number of seconds between "
-		"heartbeat checks.");
-
-module_param(xpc_disengage_request_timelimit, int, 0);
-MODULE_PARM_DESC(xpc_disengage_request_timelimit, "Number of seconds to wait "
-		"for disengage request to complete.");
-
-module_param(xpc_kdebug_ignore, int, 0);
-MODULE_PARM_DESC(xpc_kdebug_ignore, "Should lack of heartbeat be ignored by "
-		"other partitions when dropping into kdebug.");
-
diff --git a/arch/ia64/sn/kernel/xpc_partition.c b/arch/ia64/sn/kernel/xpc_partition.c
deleted file mode 100644
index 9e97c26..0000000
--- a/arch/ia64/sn/kernel/xpc_partition.c
+++ /dev/null
@@ -1,1239 +0,0 @@
-/*
- * This file is subject to the terms and conditions of the GNU General Public
- * License.  See the file "COPYING" in the main directory of this archive
- * for more details.
- *
- * Copyright (c) 2004-2006 Silicon Graphics, Inc.  All Rights Reserved.
- */
-
-
-/*
- * Cross Partition Communication (XPC) partition support.
- *
- *	This is the part of XPC that detects the presence/absence of
- *	other partitions. It provides a heartbeat and monitors the
- *	heartbeats of other partitions.
- *
- */
-
-
-#include <linux/kernel.h>
-#include <linux/sysctl.h>
-#include <linux/cache.h>
-#include <linux/mmzone.h>
-#include <linux/nodemask.h>
-#include <asm/uncached.h>
-#include <asm/sn/bte.h>
-#include <asm/sn/intr.h>
-#include <asm/sn/sn_sal.h>
-#include <asm/sn/nodepda.h>
-#include <asm/sn/addrs.h>
-#include <asm/sn/xpc.h>
-
-
-/* XPC is exiting flag */
-int xpc_exiting;
-
-
-/* SH_IPI_ACCESS shub register value on startup */
-static u64 xpc_sh1_IPI_access;
-static u64 xpc_sh2_IPI_access0;
-static u64 xpc_sh2_IPI_access1;
-static u64 xpc_sh2_IPI_access2;
-static u64 xpc_sh2_IPI_access3;
-
-
-/* original protection values for each node */
-u64 xpc_prot_vec[MAX_NUMNODES];
-
-
-/* this partition's reserved page pointers */
-struct xpc_rsvd_page *xpc_rsvd_page;
-static u64 *xpc_part_nasids;
-static u64 *xpc_mach_nasids;
-struct xpc_vars *xpc_vars;
-struct xpc_vars_part *xpc_vars_part;
-
-static int xp_nasid_mask_bytes;	/* actual size in bytes of nasid mask */
-static int xp_nasid_mask_words;	/* actual size in words of nasid mask */
-
-
-/*
- * For performance reasons, each entry of xpc_partitions[] is cacheline
- * aligned. And xpc_partitions[] is padded with an additional entry at the
- * end so that the last legitimate entry doesn't share its cacheline with
- * another variable.
- */
-struct xpc_partition xpc_partitions[XP_MAX_PARTITIONS + 1];
-
-
-/*
- * Generic buffer used to store a local copy of portions of a remote
- * partition's reserved page (either its header and part_nasids mask,
- * or its vars).
- */
-char *xpc_remote_copy_buffer;
-void *xpc_remote_copy_buffer_base;
-
-
-/*
- * Guarantee that the kmalloc'd memory is cacheline aligned.
- */
-void *
-xpc_kmalloc_cacheline_aligned(size_t size, gfp_t flags, void **base)
-{
-	/* see if kmalloc will give us cachline aligned memory by default */
-	*base = kmalloc(size, flags);
-	if (*base == NULL) {
-		return NULL;
-	}
-	if ((u64) *base == L1_CACHE_ALIGN((u64) *base)) {
-		return *base;
-	}
-	kfree(*base);
-
-	/* nope, we'll have to do it ourselves */
-	*base = kmalloc(size + L1_CACHE_BYTES, flags);
-	if (*base == NULL) {
-		return NULL;
-	}
-	return (void *) L1_CACHE_ALIGN((u64) *base);
-}
-
-
-/*
- * Given a nasid, get the physical address of the  partition's reserved page
- * for that nasid. This function returns 0 on any error.
- */
-static u64
-xpc_get_rsvd_page_pa(int nasid)
-{
-	bte_result_t bte_res;
-	s64 status;
-	u64 cookie = 0;
-	u64 rp_pa = nasid;	/* seed with nasid */
-	u64 len = 0;
-	u64 buf = buf;
-	u64 buf_len = 0;
-	void *buf_base = NULL;
-
-
-	while (1) {
-
-		status = sn_partition_reserved_page_pa(buf, &cookie, &rp_pa,
-								&len);
-
-		dev_dbg(xpc_part, "SAL returned with status=%li, cookie="
-			"0x%016lx, address=0x%016lx, len=0x%016lx\n",
-			status, cookie, rp_pa, len);
-
-		if (status != SALRET_MORE_PASSES) {
-			break;
-		}
-
-		if (L1_CACHE_ALIGN(len) > buf_len) {
-			kfree(buf_base);
-			buf_len = L1_CACHE_ALIGN(len);
-			buf = (u64) xpc_kmalloc_cacheline_aligned(buf_len,
-							GFP_KERNEL, &buf_base);
-			if (buf_base == NULL) {
-				dev_err(xpc_part, "unable to kmalloc "
-					"len=0x%016lx\n", buf_len);
-				status = SALRET_ERROR;
-				break;
-			}
-		}
-
-		bte_res = xp_bte_copy(rp_pa, buf, buf_len,
-					(BTE_NOTIFY | BTE_WACQUIRE), NULL);
-		if (bte_res != BTE_SUCCESS) {
-			dev_dbg(xpc_part, "xp_bte_copy failed %i\n", bte_res);
-			status = SALRET_ERROR;
-			break;
-		}
-	}
-
-	kfree(buf_base);
-
-	if (status != SALRET_OK) {
-		rp_pa = 0;
-	}
-	dev_dbg(xpc_part, "reserved page at phys address 0x%016lx\n", rp_pa);
-	return rp_pa;
-}
-
-
-/*
- * Fill the partition reserved page with the information needed by
- * other partitions to discover we are alive and establish initial
- * communications.
- */
-struct xpc_rsvd_page *
-xpc_rsvd_page_init(void)
-{
-	struct xpc_rsvd_page *rp;
-	AMO_t *amos_page;
-	u64 rp_pa, nasid_array = 0;
-	int i, ret;
-
-
-	/* get the local reserved page's address */
-
-	preempt_disable();
-	rp_pa = xpc_get_rsvd_page_pa(cpuid_to_nasid(smp_processor_id()));
-	preempt_enable();
-	if (rp_pa == 0) {
-		dev_err(xpc_part, "SAL failed to locate the reserved page\n");
-		return NULL;
-	}
-	rp = (struct xpc_rsvd_page *) __va(rp_pa);
-
-	if (rp->partid != sn_partition_id) {
-		dev_err(xpc_part, "the reserved page's partid of %d should be "
-			"%d\n", rp->partid, sn_partition_id);
-		return NULL;
-	}
-
-	rp->version = XPC_RP_VERSION;
-
-	/* establish the actual sizes of the nasid masks */
-	if (rp->SAL_version == 1) {
-		/* SAL_version 1 didn't set the nasids_size field */
-		rp->nasids_size = 128;
-	}
-	xp_nasid_mask_bytes = rp->nasids_size;
-	xp_nasid_mask_words = xp_nasid_mask_bytes / 8;
-
-	/* setup the pointers to the various items in the reserved page */
-	xpc_part_nasids = XPC_RP_PART_NASIDS(rp);
-	xpc_mach_nasids = XPC_RP_MACH_NASIDS(rp);
-	xpc_vars = XPC_RP_VARS(rp);
-	xpc_vars_part = XPC_RP_VARS_PART(rp);
-
-	/*
-	 * Before clearing xpc_vars, see if a page of AMOs had been previously
-	 * allocated. If not we'll need to allocate one and set permissions
-	 * so that cross-partition AMOs are allowed.
-	 *
-	 * The allocated AMO page needs MCA reporting to remain disabled after
-	 * XPC has unloaded.  To make this work, we keep a copy of the pointer
-	 * to this page (i.e., amos_page) in the struct xpc_vars structure,
-	 * which is pointed to by the reserved page, and re-use that saved copy
-	 * on subsequent loads of XPC. This AMO page is never freed, and its
-	 * memory protections are never restricted.
-	 */
-	if ((amos_page = xpc_vars->amos_page) == NULL) {
-		amos_page = (AMO_t *) TO_AMO(uncached_alloc_page(0));
-		if (amos_page == NULL) {
-			dev_err(xpc_part, "can't allocate page of AMOs\n");
-			return NULL;
-		}
-
-		/*
-		 * Open up AMO-R/W to cpu.  This is done for Shub 1.1 systems
-		 * when xpc_allow_IPI_ops() is called via xpc_hb_init().
-		 */
-		if (!enable_shub_wars_1_1()) {
-			ret = sn_change_memprotect(ia64_tpa((u64) amos_page),
-					PAGE_SIZE, SN_MEMPROT_ACCESS_CLASS_1,
-					&nasid_array);
-			if (ret != 0) {
-				dev_err(xpc_part, "can't change memory "
-					"protections\n");
-				uncached_free_page(__IA64_UNCACHED_OFFSET |
-						   TO_PHYS((u64) amos_page));
-				return NULL;
-			}
-		}
-	} else if (!IS_AMO_ADDRESS((u64) amos_page)) {
-		/*
-		 * EFI's XPBOOT can also set amos_page in the reserved page,
-		 * but it happens to leave it as an uncached physical address
-		 * and we need it to be an uncached virtual, so we'll have to
-		 * convert it.
-		 */
-		if (!IS_AMO_PHYS_ADDRESS((u64) amos_page)) {
-			dev_err(xpc_part, "previously used amos_page address "
-				"is bad = 0x%p\n", (void *) amos_page);
-			return NULL;
-		}
-		amos_page = (AMO_t *) TO_AMO((u64) amos_page);
-	}
-
-	/* clear xpc_vars */
-	memset(xpc_vars, 0, sizeof(struct xpc_vars));
-
-	xpc_vars->version = XPC_V_VERSION;
-	xpc_vars->act_nasid = cpuid_to_nasid(0);
-	xpc_vars->act_phys_cpuid = cpu_physical_id(0);
-	xpc_vars->vars_part_pa = __pa(xpc_vars_part);
-	xpc_vars->amos_page_pa = ia64_tpa((u64) amos_page);
-	xpc_vars->amos_page = amos_page;  /* save for next load of XPC */
-
-
-	/* clear xpc_vars_part */
-	memset((u64 *) xpc_vars_part, 0, sizeof(struct xpc_vars_part) *
-							XP_MAX_PARTITIONS);
-
-	/* initialize the activate IRQ related AMO variables */
-	for (i = 0; i < xp_nasid_mask_words; i++) {
-		(void) xpc_IPI_init(XPC_ACTIVATE_IRQ_AMOS + i);
-	}
-
-	/* initialize the engaged remote partitions related AMO variables */
-	(void) xpc_IPI_init(XPC_ENGAGED_PARTITIONS_AMO);
-	(void) xpc_IPI_init(XPC_DISENGAGE_REQUEST_AMO);
-
-	/* timestamp of when reserved page was setup by XPC */
-	rp->stamp = CURRENT_TIME;
-
-	/*
-	 * This signifies to the remote partition that our reserved
-	 * page is initialized.
-	 */
-	rp->vars_pa = __pa(xpc_vars);
-
-	return rp;
-}
-
-
-/*
- * Change protections to allow IPI operations (and AMO operations on
- * Shub 1.1 systems).
- */
-void
-xpc_allow_IPI_ops(void)
-{
-	int node;
-	int nasid;
-
-
-	// >>> Change SH_IPI_ACCESS code to use SAL call once it is available.
-
-	if (is_shub2()) {
-		xpc_sh2_IPI_access0 =
-			(u64) HUB_L((u64 *) LOCAL_MMR_ADDR(SH2_IPI_ACCESS0));
-		xpc_sh2_IPI_access1 =
-			(u64) HUB_L((u64 *) LOCAL_MMR_ADDR(SH2_IPI_ACCESS1));
-		xpc_sh2_IPI_access2 =
-			(u64) HUB_L((u64 *) LOCAL_MMR_ADDR(SH2_IPI_ACCESS2));
-		xpc_sh2_IPI_access3 =
-			(u64) HUB_L((u64 *) LOCAL_MMR_ADDR(SH2_IPI_ACCESS3));
-
-		for_each_online_node(node) {
-			nasid = cnodeid_to_nasid(node);
-			HUB_S((u64 *) GLOBAL_MMR_ADDR(nasid, SH2_IPI_ACCESS0),
-								-1UL);
-			HUB_S((u64 *) GLOBAL_MMR_ADDR(nasid, SH2_IPI_ACCESS1),
-								-1UL);
-			HUB_S((u64 *) GLOBAL_MMR_ADDR(nasid, SH2_IPI_ACCESS2),
-								-1UL);
-			HUB_S((u64 *) GLOBAL_MMR_ADDR(nasid, SH2_IPI_ACCESS3),
-								-1UL);
-		}
-
-	} else {
-		xpc_sh1_IPI_access =
-			(u64) HUB_L((u64 *) LOCAL_MMR_ADDR(SH1_IPI_ACCESS));
-
-		for_each_online_node(node) {
-			nasid = cnodeid_to_nasid(node);
-			HUB_S((u64 *) GLOBAL_MMR_ADDR(nasid, SH1_IPI_ACCESS),
-								-1UL);
-
-			/*
-			 * Since the BIST collides with memory operations on
-			 * SHUB 1.1 sn_change_memprotect() cannot be used.
-			 */
-			if (enable_shub_wars_1_1()) {
-				/* open up everything */
-				xpc_prot_vec[node] = (u64) HUB_L((u64 *)
-						GLOBAL_MMR_ADDR(nasid,
-						SH1_MD_DQLP_MMR_DIR_PRIVEC0));
-				HUB_S((u64 *) GLOBAL_MMR_ADDR(nasid,
-						SH1_MD_DQLP_MMR_DIR_PRIVEC0),
-								-1UL);
-				HUB_S((u64 *) GLOBAL_MMR_ADDR(nasid,
-						SH1_MD_DQRP_MMR_DIR_PRIVEC0),
-								-1UL);
-			}
-		}
-	}
-}
-
-
-/*
- * Restrict protections to disallow IPI operations (and AMO operations on
- * Shub 1.1 systems).
- */
-void
-xpc_restrict_IPI_ops(void)
-{
-	int node;
-	int nasid;
-
-
-	// >>> Change SH_IPI_ACCESS code to use SAL call once it is available.
-
-	if (is_shub2()) {
-
-		for_each_online_node(node) {
-			nasid = cnodeid_to_nasid(node);
-			HUB_S((u64 *) GLOBAL_MMR_ADDR(nasid, SH2_IPI_ACCESS0),
-							xpc_sh2_IPI_access0);
-			HUB_S((u64 *) GLOBAL_MMR_ADDR(nasid, SH2_IPI_ACCESS1),
-							xpc_sh2_IPI_access1);
-			HUB_S((u64 *) GLOBAL_MMR_ADDR(nasid, SH2_IPI_ACCESS2),
-							xpc_sh2_IPI_access2);
-			HUB_S((u64 *) GLOBAL_MMR_ADDR(nasid, SH2_IPI_ACCESS3),
-							xpc_sh2_IPI_access3);
-		}
-
-	} else {
-
-		for_each_online_node(node) {
-			nasid = cnodeid_to_nasid(node);
-			HUB_S((u64 *) GLOBAL_MMR_ADDR(nasid, SH1_IPI_ACCESS),
-							xpc_sh1_IPI_access);
-
-			if (enable_shub_wars_1_1()) {
-				HUB_S((u64 *) GLOBAL_MMR_ADDR(nasid,
-						SH1_MD_DQLP_MMR_DIR_PRIVEC0),
-							xpc_prot_vec[node]);
-				HUB_S((u64 *) GLOBAL_MMR_ADDR(nasid,
-						SH1_MD_DQRP_MMR_DIR_PRIVEC0),
-							xpc_prot_vec[node]);
-			}
-		}
-	}
-}
-
-
-/*
- * At periodic intervals, scan through all active partitions and ensure
- * their heartbeat is still active.  If not, the partition is deactivated.
- */
-void
-xpc_check_remote_hb(void)
-{
-	struct xpc_vars *remote_vars;
-	struct xpc_partition *part;
-	partid_t partid;
-	bte_result_t bres;
-
-
-	remote_vars = (struct xpc_vars *) xpc_remote_copy_buffer;
-
-	for (partid = 1; partid < XP_MAX_PARTITIONS; partid++) {
-
-		if (xpc_exiting) {
-			break;
-		}
-
-		if (partid == sn_partition_id) {
-			continue;
-		}
-
-		part = &xpc_partitions[partid];
-
-		if (part->act_state == XPC_P_INACTIVE ||
-				part->act_state == XPC_P_DEACTIVATING) {
-			continue;
-		}
-
-		/* pull the remote_hb cache line */
-		bres = xp_bte_copy(part->remote_vars_pa,
-					(u64) remote_vars,
-					XPC_RP_VARS_SIZE,
-					(BTE_NOTIFY | BTE_WACQUIRE), NULL);
-		if (bres != BTE_SUCCESS) {
-			XPC_DEACTIVATE_PARTITION(part,
-						xpc_map_bte_errors(bres));
-			continue;
-		}
-
-		dev_dbg(xpc_part, "partid = %d, heartbeat = %ld, last_heartbeat"
-			" = %ld, heartbeat_offline = %ld, HB_mask = 0x%lx\n",
-			partid, remote_vars->heartbeat, part->last_heartbeat,
-			remote_vars->heartbeat_offline,
-			remote_vars->heartbeating_to_mask);
-
-		if (((remote_vars->heartbeat == part->last_heartbeat) &&
-			(remote_vars->heartbeat_offline == 0)) ||
-			     !xpc_hb_allowed(sn_partition_id, remote_vars)) {
-
-			XPC_DEACTIVATE_PARTITION(part, xpcNoHeartbeat);
-			continue;
-		}
-
-		part->last_heartbeat = remote_vars->heartbeat;
-	}
-}
-
-
-/*
- * Get a copy of a portion of the remote partition's rsvd page.
- *
- * remote_rp points to a buffer that is cacheline aligned for BTE copies and
- * is large enough to contain a copy of their reserved page header and
- * part_nasids mask.
- */
-static enum xpc_retval
-xpc_get_remote_rp(int nasid, u64 *discovered_nasids,
-		struct xpc_rsvd_page *remote_rp, u64 *remote_rp_pa)
-{
-	int bres, i;
-
-
-	/* get the reserved page's physical address */
-
-	*remote_rp_pa = xpc_get_rsvd_page_pa(nasid);
-	if (*remote_rp_pa == 0) {
-		return xpcNoRsvdPageAddr;
-	}
-
-
-	/* pull over the reserved page header and part_nasids mask */
-	bres = xp_bte_copy(*remote_rp_pa, (u64) remote_rp,
-				XPC_RP_HEADER_SIZE + xp_nasid_mask_bytes,
-				(BTE_NOTIFY | BTE_WACQUIRE), NULL);
-	if (bres != BTE_SUCCESS) {
-		return xpc_map_bte_errors(bres);
-	}
-
-
-	if (discovered_nasids != NULL) {
-		u64 *remote_part_nasids = XPC_RP_PART_NASIDS(remote_rp);
-
-
-		for (i = 0; i < xp_nasid_mask_words; i++) {
-			discovered_nasids[i] |= remote_part_nasids[i];
-		}
-	}
-
-
-	/* check that the partid is for another partition */
-
-	if (remote_rp->partid < 1 ||
-				remote_rp->partid > (XP_MAX_PARTITIONS - 1)) {
-		return xpcInvalidPartid;
-	}
-
-	if (remote_rp->partid == sn_partition_id) {
-		return xpcLocalPartid;
-	}
-
-
-	if (XPC_VERSION_MAJOR(remote_rp->version) !=
-					XPC_VERSION_MAJOR(XPC_RP_VERSION)) {
-		return xpcBadVersion;
-	}
-
-	return xpcSuccess;
-}
-
-
-/*
- * Get a copy of the remote partition's XPC variables from the reserved page.
- *
- * remote_vars points to a buffer that is cacheline aligned for BTE copies and
- * assumed to be of size XPC_RP_VARS_SIZE.
- */
-static enum xpc_retval
-xpc_get_remote_vars(u64 remote_vars_pa, struct xpc_vars *remote_vars)
-{
-	int bres;
-
-
-	if (remote_vars_pa == 0) {
-		return xpcVarsNotSet;
-	}
-
-	/* pull over the cross partition variables */
-	bres = xp_bte_copy(remote_vars_pa, (u64) remote_vars, XPC_RP_VARS_SIZE,
-				(BTE_NOTIFY | BTE_WACQUIRE), NULL);
-	if (bres != BTE_SUCCESS) {
-		return xpc_map_bte_errors(bres);
-	}
-
-	if (XPC_VERSION_MAJOR(remote_vars->version) !=
-					XPC_VERSION_MAJOR(XPC_V_VERSION)) {
-		return xpcBadVersion;
-	}
-
-	return xpcSuccess;
-}
-
-
-/*
- * Update the remote partition's info.
- */
-static void
-xpc_update_partition_info(struct xpc_partition *part, u8 remote_rp_version,
-		struct timespec *remote_rp_stamp, u64 remote_rp_pa,
-		u64 remote_vars_pa, struct xpc_vars *remote_vars)
-{
-	part->remote_rp_version = remote_rp_version;
-	dev_dbg(xpc_part, "  remote_rp_version = 0x%016x\n",
-		part->remote_rp_version);
-
-	part->remote_rp_stamp = *remote_rp_stamp;
-	dev_dbg(xpc_part, "  remote_rp_stamp (tv_sec = 0x%lx tv_nsec = 0x%lx\n",
-		part->remote_rp_stamp.tv_sec, part->remote_rp_stamp.tv_nsec);
-
-	part->remote_rp_pa = remote_rp_pa;
-	dev_dbg(xpc_part, "  remote_rp_pa = 0x%016lx\n", part->remote_rp_pa);
-
-	part->remote_vars_pa = remote_vars_pa;
-	dev_dbg(xpc_part, "  remote_vars_pa = 0x%016lx\n",
-		part->remote_vars_pa);
-
-	part->last_heartbeat = remote_vars->heartbeat;
-	dev_dbg(xpc_part, "  last_heartbeat = 0x%016lx\n",
-		part->last_heartbeat);
-
-	part->remote_vars_part_pa = remote_vars->vars_part_pa;
-	dev_dbg(xpc_part, "  remote_vars_part_pa = 0x%016lx\n",
-		part->remote_vars_part_pa);
-
-	part->remote_act_nasid = remote_vars->act_nasid;
-	dev_dbg(xpc_part, "  remote_act_nasid = 0x%x\n",
-		part->remote_act_nasid);
-
-	part->remote_act_phys_cpuid = remote_vars->act_phys_cpuid;
-	dev_dbg(xpc_part, "  remote_act_phys_cpuid = 0x%x\n",
-		part->remote_act_phys_cpuid);
-
-	part->remote_amos_page_pa = remote_vars->amos_page_pa;
-	dev_dbg(xpc_part, "  remote_amos_page_pa = 0x%lx\n",
-		part->remote_amos_page_pa);
-
-	part->remote_vars_version = remote_vars->version;
-	dev_dbg(xpc_part, "  remote_vars_version = 0x%x\n",
-		part->remote_vars_version);
-}
-
-
-/*
- * Prior code has determined the nasid which generated an IPI.  Inspect
- * that nasid to determine if its partition needs to be activated or
- * deactivated.
- *
- * A partition is consider "awaiting activation" if our partition
- * flags indicate it is not active and it has a heartbeat.  A
- * partition is considered "awaiting deactivation" if our partition
- * flags indicate it is active but it has no heartbeat or it is not
- * sending its heartbeat to us.
- *
- * To determine the heartbeat, the remote nasid must have a properly
- * initialized reserved page.
- */
-static void
-xpc_identify_act_IRQ_req(int nasid)
-{
-	struct xpc_rsvd_page *remote_rp;
-	struct xpc_vars *remote_vars;
-	u64 remote_rp_pa;
-	u64 remote_vars_pa;
-	int remote_rp_version;
-	int reactivate = 0;
-	int stamp_diff;
-	struct timespec remote_rp_stamp = { 0, 0 };
-	partid_t partid;
-	struct xpc_partition *part;
-	enum xpc_retval ret;
-
-
-	/* pull over the reserved page structure */
-
-	remote_rp = (struct xpc_rsvd_page *) xpc_remote_copy_buffer;
-
-	ret = xpc_get_remote_rp(nasid, NULL, remote_rp, &remote_rp_pa);
-	if (ret != xpcSuccess) {
-		dev_warn(xpc_part, "unable to get reserved page from nasid %d, "
-			"which sent interrupt, reason=%d\n", nasid, ret);
-		return;
-	}
-
-	remote_vars_pa = remote_rp->vars_pa;
-	remote_rp_version = remote_rp->version;
-	if (XPC_SUPPORTS_RP_STAMP(remote_rp_version)) {
-		remote_rp_stamp = remote_rp->stamp;
-	}
-	partid = remote_rp->partid;
-	part = &xpc_partitions[partid];
-
-
-	/* pull over the cross partition variables */
-
-	remote_vars = (struct xpc_vars *) xpc_remote_copy_buffer;
-
-	ret = xpc_get_remote_vars(remote_vars_pa, remote_vars);
-	if (ret != xpcSuccess) {
-
-		dev_warn(xpc_part, "unable to get XPC variables from nasid %d, "
-			"which sent interrupt, reason=%d\n", nasid, ret);
-
-		XPC_DEACTIVATE_PARTITION(part, ret);
-		return;
-	}
-
-
-	part->act_IRQ_rcvd++;
-
-	dev_dbg(xpc_part, "partid for nasid %d is %d; IRQs = %d; HB = "
-		"%ld:0x%lx\n", (int) nasid, (int) partid, part->act_IRQ_rcvd,
-		remote_vars->heartbeat, remote_vars->heartbeating_to_mask);
-
-	if (xpc_partition_disengaged(part) &&
-					part->act_state == XPC_P_INACTIVE) {
-
-		xpc_update_partition_info(part, remote_rp_version,
-					&remote_rp_stamp, remote_rp_pa,
-					remote_vars_pa, remote_vars);
-
-		if (XPC_SUPPORTS_DISENGAGE_REQUEST(part->remote_vars_version)) {
-			if (xpc_partition_disengage_requested(1UL << partid)) {
-				/*
-				 * Other side is waiting on us to disengage,
-				 * even though we already have.
-				 */
-				return;
-			}
-		} else {
-			/* other side doesn't support disengage requests */
-			xpc_clear_partition_disengage_request(1UL << partid);
-		}
-
-		xpc_activate_partition(part);
-		return;
-	}
-
-	DBUG_ON(part->remote_rp_version == 0);
-	DBUG_ON(part->remote_vars_version == 0);
-
-	if (!XPC_SUPPORTS_RP_STAMP(part->remote_rp_version)) {
-		DBUG_ON(XPC_SUPPORTS_DISENGAGE_REQUEST(part->
-							remote_vars_version));
-
-		if (!XPC_SUPPORTS_RP_STAMP(remote_rp_version)) {
-			DBUG_ON(XPC_SUPPORTS_DISENGAGE_REQUEST(remote_vars->
-								version));
-			/* see if the other side rebooted */
-			if (part->remote_amos_page_pa ==
-				remote_vars->amos_page_pa &&
-					xpc_hb_allowed(sn_partition_id,
-								remote_vars)) {
-				/* doesn't look that way, so ignore the IPI */
-				return;
-			}
-		}
-
-		/*
-		 * Other side rebooted and previous XPC didn't support the
-		 * disengage request, so we don't need to do anything special.
-		 */
-
-		xpc_update_partition_info(part, remote_rp_version,
-						&remote_rp_stamp, remote_rp_pa,
-						remote_vars_pa, remote_vars);
-		part->reactivate_nasid = nasid;
-		XPC_DEACTIVATE_PARTITION(part, xpcReactivating);
-		return;
-	}
-
-	DBUG_ON(!XPC_SUPPORTS_DISENGAGE_REQUEST(part->remote_vars_version));
-
-	if (!XPC_SUPPORTS_RP_STAMP(remote_rp_version)) {
-		DBUG_ON(!XPC_SUPPORTS_DISENGAGE_REQUEST(remote_vars->version));
-
-		/*
-		 * Other side rebooted and previous XPC did support the
-		 * disengage request, but the new one doesn't.
-		 */
-
-		xpc_clear_partition_engaged(1UL << partid);
-		xpc_clear_partition_disengage_request(1UL << partid);
-
-		xpc_update_partition_info(part, remote_rp_version,
-						&remote_rp_stamp, remote_rp_pa,
-						remote_vars_pa, remote_vars);
-		reactivate = 1;
-
-	} else {
-		DBUG_ON(!XPC_SUPPORTS_DISENGAGE_REQUEST(remote_vars->version));
-
-		stamp_diff = xpc_compare_stamps(&part->remote_rp_stamp,
-							&remote_rp_stamp);
-		if (stamp_diff != 0) {
-			DBUG_ON(stamp_diff >= 0);
-
-			/*
-			 * Other side rebooted and the previous XPC did support
-			 * the disengage request, as does the new one.
-			 */
-
-			DBUG_ON(xpc_partition_engaged(1UL << partid));
-			DBUG_ON(xpc_partition_disengage_requested(1UL <<
-								partid));
-
-			xpc_update_partition_info(part, remote_rp_version,
-						&remote_rp_stamp, remote_rp_pa,
-						remote_vars_pa, remote_vars);
-			reactivate = 1;
-		}
-	}
-
-	if (part->disengage_request_timeout > 0 &&
-					!xpc_partition_disengaged(part)) {
-		/* still waiting on other side to disengage from us */
-		return;
-	}
-
-	if (reactivate) {
-		part->reactivate_nasid = nasid;
-		XPC_DEACTIVATE_PARTITION(part, xpcReactivating);
-
-	} else if (XPC_SUPPORTS_DISENGAGE_REQUEST(part->remote_vars_version) &&
-			xpc_partition_disengage_requested(1UL << partid)) {
-		XPC_DEACTIVATE_PARTITION(part, xpcOtherGoingDown);
-	}
-}
-
-
-/*
- * Loop through the activation AMO variables and process any bits
- * which are set.  Each bit indicates a nasid sending a partition
- * activation or deactivation request.
- *
- * Return #of IRQs detected.
- */
-int
-xpc_identify_act_IRQ_sender(void)
-{
-	int word, bit;
-	u64 nasid_mask;
-	u64 nasid;			/* remote nasid */
-	int n_IRQs_detected = 0;
-	AMO_t *act_amos;
-
-
-	act_amos = xpc_vars->amos_page + XPC_ACTIVATE_IRQ_AMOS;
-
-
-	/* scan through act AMO variable looking for non-zero entries */
-	for (word = 0; word < xp_nasid_mask_words; word++) {
-
-		if (xpc_exiting) {
-			break;
-		}
-
-		nasid_mask = xpc_IPI_receive(&act_amos[word]);
-		if (nasid_mask == 0) {
-			/* no IRQs from nasids in this variable */
-			continue;
-		}
-
-		dev_dbg(xpc_part, "AMO[%d] gave back 0x%lx\n", word,
-			nasid_mask);
-
-
-		/*
-		 * If this nasid has been added to the machine since
-		 * our partition was reset, this will retain the
-		 * remote nasid in our reserved pages machine mask.
-		 * This is used in the event of module reload.
-		 */
-		xpc_mach_nasids[word] |= nasid_mask;
-
-
-		/* locate the nasid(s) which sent interrupts */
-
-		for (bit = 0; bit < (8 * sizeof(u64)); bit++) {
-			if (nasid_mask & (1UL << bit)) {
-				n_IRQs_detected++;
-				nasid = XPC_NASID_FROM_W_B(word, bit);
-				dev_dbg(xpc_part, "interrupt from nasid %ld\n",
-					nasid);
-				xpc_identify_act_IRQ_req(nasid);
-			}
-		}
-	}
-	return n_IRQs_detected;
-}
-
-
-/*
- * See if the other side has responded to a partition disengage request
- * from us.
- */
-int
-xpc_partition_disengaged(struct xpc_partition *part)
-{
-	partid_t partid = XPC_PARTID(part);
-	int disengaged;
-
-
-	disengaged = (xpc_partition_engaged(1UL << partid) == 0);
-	if (part->disengage_request_timeout) {
-		if (!disengaged) {
-			if (time_before(jiffies, part->disengage_request_timeout)) {
-				/* timelimit hasn't been reached yet */
-				return 0;
-			}
-
-			/*
-			 * Other side hasn't responded to our disengage
-			 * request in a timely fashion, so assume it's dead.
-			 */
-
-			dev_info(xpc_part, "disengage from remote partition %d "
-				"timed out\n", partid);
-			xpc_disengage_request_timedout = 1;
-			xpc_clear_partition_engaged(1UL << partid);
-			disengaged = 1;
-		}
-		part->disengage_request_timeout = 0;
-
-		/* cancel the timer function, provided it's not us */
-		if (!in_interrupt()) {
-			del_singleshot_timer_sync(&part->
-						      disengage_request_timer);
-		}
-
-		DBUG_ON(part->act_state != XPC_P_DEACTIVATING &&
-					part->act_state != XPC_P_INACTIVE);
-		if (part->act_state != XPC_P_INACTIVE) {
-			xpc_wakeup_channel_mgr(part);
-		}
-
-		if (XPC_SUPPORTS_DISENGAGE_REQUEST(part->remote_vars_version)) {
-			xpc_cancel_partition_disengage_request(part);
-		}
-	}
-	return disengaged;
-}
-
-
-/*
- * Mark specified partition as active.
- */
-enum xpc_retval
-xpc_mark_partition_active(struct xpc_partition *part)
-{
-	unsigned long irq_flags;
-	enum xpc_retval ret;
-
-
-	dev_dbg(xpc_part, "setting partition %d to ACTIVE\n", XPC_PARTID(part));
-
-	spin_lock_irqsave(&part->act_lock, irq_flags);
-	if (part->act_state == XPC_P_ACTIVATING) {
-		part->act_state = XPC_P_ACTIVE;
-		ret = xpcSuccess;
-	} else {
-		DBUG_ON(part->reason == xpcSuccess);
-		ret = part->reason;
-	}
-	spin_unlock_irqrestore(&part->act_lock, irq_flags);
-
-	return ret;
-}
-
-
-/*
- * Notify XPC that the partition is down.
- */
-void
-xpc_deactivate_partition(const int line, struct xpc_partition *part,
-				enum xpc_retval reason)
-{
-	unsigned long irq_flags;
-
-
-	spin_lock_irqsave(&part->act_lock, irq_flags);
-
-	if (part->act_state == XPC_P_INACTIVE) {
-		XPC_SET_REASON(part, reason, line);
-		spin_unlock_irqrestore(&part->act_lock, irq_flags);
-		if (reason == xpcReactivating) {
-			/* we interrupt ourselves to reactivate partition */
-			xpc_IPI_send_reactivate(part);
-		}
-		return;
-	}
-	if (part->act_state == XPC_P_DEACTIVATING) {
-		if ((part->reason == xpcUnloading && reason != xpcUnloading) ||
-					reason == xpcReactivating) {
-			XPC_SET_REASON(part, reason, line);
-		}
-		spin_unlock_irqrestore(&part->act_lock, irq_flags);
-		return;
-	}
-
-	part->act_state = XPC_P_DEACTIVATING;
-	XPC_SET_REASON(part, reason, line);
-
-	spin_unlock_irqrestore(&part->act_lock, irq_flags);
-
-	if (XPC_SUPPORTS_DISENGAGE_REQUEST(part->remote_vars_version)) {
-		xpc_request_partition_disengage(part);
-		xpc_IPI_send_disengage(part);
-
-		/* set a timelimit on the disengage request */
-		part->disengage_request_timeout = jiffies +
-					(xpc_disengage_request_timelimit * HZ);
-		part->disengage_request_timer.expires =
-					part->disengage_request_timeout;
-		add_timer(&part->disengage_request_timer);
-	}
-
-	dev_dbg(xpc_part, "bringing partition %d down, reason = %d\n",
-		XPC_PARTID(part), reason);
-
-	xpc_partition_going_down(part, reason);
-}
-
-
-/*
- * Mark specified partition as inactive.
- */
-void
-xpc_mark_partition_inactive(struct xpc_partition *part)
-{
-	unsigned long irq_flags;
-
-
-	dev_dbg(xpc_part, "setting partition %d to INACTIVE\n",
-		XPC_PARTID(part));
-
-	spin_lock_irqsave(&part->act_lock, irq_flags);
-	part->act_state = XPC_P_INACTIVE;
-	spin_unlock_irqrestore(&part->act_lock, irq_flags);
-	part->remote_rp_pa = 0;
-}
-
-
-/*
- * SAL has provided a partition and machine mask.  The partition mask
- * contains a bit for each even nasid in our partition.  The machine
- * mask contains a bit for each even nasid in the entire machine.
- *
- * Using those two bit arrays, we can determine which nasids are
- * known in the machine.  Each should also have a reserved page
- * initialized if they are available for partitioning.
- */
-void
-xpc_discovery(void)
-{
-	void *remote_rp_base;
-	struct xpc_rsvd_page *remote_rp;
-	struct xpc_vars *remote_vars;
-	u64 remote_rp_pa;
-	u64 remote_vars_pa;
-	int region;
-	int region_size;
-	int max_regions;
-	int nasid;
-	struct xpc_rsvd_page *rp;
-	partid_t partid;
-	struct xpc_partition *part;
-	u64 *discovered_nasids;
-	enum xpc_retval ret;
-
-
-	remote_rp = xpc_kmalloc_cacheline_aligned(XPC_RP_HEADER_SIZE +
-						xp_nasid_mask_bytes,
-						GFP_KERNEL, &remote_rp_base);
-	if (remote_rp == NULL) {
-		return;
-	}
-	remote_vars = (struct xpc_vars *) remote_rp;
-
-
-	discovered_nasids = kzalloc(sizeof(u64) * xp_nasid_mask_words,
-							GFP_KERNEL);
-	if (discovered_nasids == NULL) {
-		kfree(remote_rp_base);
-		return;
-	}
-
-	rp = (struct xpc_rsvd_page *) xpc_rsvd_page;
-
-	/*
-	 * The term 'region' in this context refers to the minimum number of
-	 * nodes that can comprise an access protection grouping. The access
-	 * protection is in regards to memory, IOI and IPI.
-	 */
-	max_regions = 64;
-	region_size = sn_region_size;
-
-	switch (region_size) {
-	case 128:
-		max_regions *= 2;
-	case 64:
-		max_regions *= 2;
-	case 32:
-		max_regions *= 2;
-		region_size = 16;
-		DBUG_ON(!is_shub2());
-	}
-
-	for (region = 0; region < max_regions; region++) {
-
-		if ((volatile int) xpc_exiting) {
-			break;
-		}
-
-		dev_dbg(xpc_part, "searching region %d\n", region);
-
-		for (nasid = (region * region_size * 2);
-		     nasid < ((region + 1) * region_size * 2);
-		     nasid += 2) {
-
-			if ((volatile int) xpc_exiting) {
-				break;
-			}
-
-			dev_dbg(xpc_part, "checking nasid %d\n", nasid);
-
-
-			if (XPC_NASID_IN_ARRAY(nasid, xpc_part_nasids)) {
-				dev_dbg(xpc_part, "PROM indicates Nasid %d is "
-					"part of the local partition; skipping "
-					"region\n", nasid);
-				break;
-			}
-
-			if (!(XPC_NASID_IN_ARRAY(nasid, xpc_mach_nasids))) {
-				dev_dbg(xpc_part, "PROM indicates Nasid %d was "
-					"not on Numa-Link network at reset\n",
-					nasid);
-				continue;
-			}
-
-			if (XPC_NASID_IN_ARRAY(nasid, discovered_nasids)) {
-				dev_dbg(xpc_part, "Nasid %d is part of a "
-					"partition which was previously "
-					"discovered\n", nasid);
-				continue;
-			}
-
-
-			/* pull over the reserved page structure */
-
-			ret = xpc_get_remote_rp(nasid, discovered_nasids,
-					      remote_rp, &remote_rp_pa);
-			if (ret != xpcSuccess) {
-				dev_dbg(xpc_part, "unable to get reserved page "
-					"from nasid %d, reason=%d\n", nasid,
-					ret);
-
-				if (ret == xpcLocalPartid) {
-					break;
-				}
-				continue;
-			}
-
-			remote_vars_pa = remote_rp->vars_pa;
-
-			partid = remote_rp->partid;
-			part = &xpc_partitions[partid];
-
-
-			/* pull over the cross partition variables */
-
-			ret = xpc_get_remote_vars(remote_vars_pa, remote_vars);
-			if (ret != xpcSuccess) {
-				dev_dbg(xpc_part, "unable to get XPC variables "
-					"from nasid %d, reason=%d\n", nasid,
-					ret);
-
-				XPC_DEACTIVATE_PARTITION(part, ret);
-				continue;
-			}
-
-			if (part->act_state != XPC_P_INACTIVE) {
-				dev_dbg(xpc_part, "partition %d on nasid %d is "
-					"already activating\n", partid, nasid);
-				break;
-			}
-
-			/*
-			 * Register the remote partition's AMOs with SAL so it
-			 * can handle and cleanup errors within that address
-			 * range should the remote partition go down. We don't
-			 * unregister this range because it is difficult to
-			 * tell when outstanding writes to the remote partition
-			 * are finished and thus when it is thus safe to
-			 * unregister. This should not result in wasted space
-			 * in the SAL xp_addr_region table because we should
-			 * get the same page for remote_act_amos_pa after
-			 * module reloads and system reboots.
-			 */
-			if (sn_register_xp_addr_region(
-					    remote_vars->amos_page_pa,
-							PAGE_SIZE, 1) < 0) {
-				dev_dbg(xpc_part, "partition %d failed to "
-					"register xp_addr region 0x%016lx\n",
-					partid, remote_vars->amos_page_pa);
-
-				XPC_SET_REASON(part, xpcPhysAddrRegFailed,
-						__LINE__);
-				break;
-			}
-
-			/*
-			 * The remote nasid is valid and available.
-			 * Send an interrupt to that nasid to notify
-			 * it that we are ready to begin activation.
-			 */
-			dev_dbg(xpc_part, "sending an interrupt to AMO 0x%lx, "
-				"nasid %d, phys_cpuid 0x%x\n",
-				remote_vars->amos_page_pa,
-				remote_vars->act_nasid,
-				remote_vars->act_phys_cpuid);
-
-			if (XPC_SUPPORTS_DISENGAGE_REQUEST(remote_vars->
-								version)) {
-				part->remote_amos_page_pa =
-						remote_vars->amos_page_pa;
-				xpc_mark_partition_disengaged(part);
-				xpc_cancel_partition_disengage_request(part);
-			}
-			xpc_IPI_send_activate(remote_vars);
-		}
-	}
-
-	kfree(discovered_nasids);
-	kfree(remote_rp_base);
-}
-
-
-/*
- * Given a partid, get the nasids owned by that partition from the
- * remote partition's reserved page.
- */
-enum xpc_retval
-xpc_initiate_partid_to_nasids(partid_t partid, void *nasid_mask)
-{
-	struct xpc_partition *part;
-	u64 part_nasid_pa;
-	int bte_res;
-
-
-	part = &xpc_partitions[partid];
-	if (part->remote_rp_pa == 0) {
-		return xpcPartitionDown;
-	}
-
-	memset(nasid_mask, 0, XP_NASID_MASK_BYTES);
-
-	part_nasid_pa = (u64) XPC_RP_PART_NASIDS(part->remote_rp_pa);
-
-	bte_res = xp_bte_copy(part_nasid_pa, (u64) nasid_mask,
-			xp_nasid_mask_bytes, (BTE_NOTIFY | BTE_WACQUIRE), NULL);
-
-	return xpc_map_bte_errors(bte_res);
-}
-
diff --git a/arch/ia64/sn/kernel/xpnet.c b/arch/ia64/sn/kernel/xpnet.c
deleted file mode 100644
index a5df672..0000000
--- a/arch/ia64/sn/kernel/xpnet.c
+++ /dev/null
@@ -1,718 +0,0 @@
-/*
- * This file is subject to the terms and conditions of the GNU General Public
- * License.  See the file "COPYING" in the main directory of this archive
- * for more details.
- *
- * Copyright (C) 1999,2001-2005 Silicon Graphics, Inc. All rights reserved.
- */
-
-
-/*
- * Cross Partition Network Interface (XPNET) support
- *
- *	XPNET provides a virtual network layered on top of the Cross
- *	Partition communication layer.
- *
- *	XPNET provides direct point-to-point and broadcast-like support
- *	for an ethernet-like device.  The ethernet broadcast medium is
- *	replaced with a point-to-point message structure which passes
- *	pointers to a DMA-capable block that a remote partition should
- *	retrieve and pass to the upper level networking layer.
- *
- */
-
-
-#include <linux/module.h>
-#include <linux/kernel.h>
-#include <linux/init.h>
-#include <linux/ioport.h>
-#include <linux/netdevice.h>
-#include <linux/etherdevice.h>
-#include <linux/delay.h>
-#include <linux/ethtool.h>
-#include <linux/mii.h>
-#include <linux/smp.h>
-#include <linux/string.h>
-#include <asm/sn/bte.h>
-#include <asm/sn/io.h>
-#include <asm/sn/sn_sal.h>
-#include <asm/types.h>
-#include <asm/atomic.h>
-#include <asm/sn/xp.h>
-
-
-/*
- * The message payload transferred by XPC.
- *
- * buf_pa is the physical address where the DMA should pull from.
- *
- * NOTE: for performance reasons, buf_pa should _ALWAYS_ begin on a
- * cacheline boundary.  To accomplish this, we record the number of
- * bytes from the beginning of the first cacheline to the first useful
- * byte of the skb (leadin_ignore) and the number of bytes from the
- * last useful byte of the skb to the end of the last cacheline
- * (tailout_ignore).
- *
- * size is the number of bytes to transfer which includes the skb->len
- * (useful bytes of the senders skb) plus the leadin and tailout
- */
-struct xpnet_message {
-	u16 version;		/* Version for this message */
-	u16 embedded_bytes;	/* #of bytes embedded in XPC message */
-	u32 magic;		/* Special number indicating this is xpnet */
-	u64 buf_pa;		/* phys address of buffer to retrieve */
-	u32 size;		/* #of bytes in buffer */
-	u8 leadin_ignore;	/* #of bytes to ignore at the beginning */
-	u8 tailout_ignore;	/* #of bytes to ignore at the end */
-	unsigned char data;	/* body of small packets */
-};
-
-/*
- * Determine the size of our message, the cacheline aligned size,
- * and then the number of message will request from XPC.
- *
- * XPC expects each message to exist in an individual cacheline.
- */
-#define XPNET_MSG_SIZE		(L1_CACHE_BYTES - XPC_MSG_PAYLOAD_OFFSET)
-#define XPNET_MSG_DATA_MAX	\
-		(XPNET_MSG_SIZE - (u64)(&((struct xpnet_message *)0)->data))
-#define XPNET_MSG_ALIGNED_SIZE	(L1_CACHE_ALIGN(XPNET_MSG_SIZE))
-#define XPNET_MSG_NENTRIES	(PAGE_SIZE / XPNET_MSG_ALIGNED_SIZE)
-
-
-#define XPNET_MAX_KTHREADS	(XPNET_MSG_NENTRIES + 1)
-#define XPNET_MAX_IDLE_KTHREADS	(XPNET_MSG_NENTRIES + 1)
-
-/*
- * Version number of XPNET implementation. XPNET can always talk to versions
- * with same major #, and never talk to versions with a different version.
- */
-#define _XPNET_VERSION(_major, _minor)	(((_major) << 4) | (_minor))
-#define XPNET_VERSION_MAJOR(_v)		((_v) >> 4)
-#define XPNET_VERSION_MINOR(_v)		((_v) & 0xf)
-
-#define	XPNET_VERSION _XPNET_VERSION(1,0)		/* version 1.0 */
-#define	XPNET_VERSION_EMBED _XPNET_VERSION(1,1)		/* version 1.1 */
-#define XPNET_MAGIC	0x88786984 /* "XNET" */
-
-#define XPNET_VALID_MSG(_m)						     \
-   ((XPNET_VERSION_MAJOR(_m->version) == XPNET_VERSION_MAJOR(XPNET_VERSION)) \
-    && (msg->magic == XPNET_MAGIC))
-
-#define XPNET_DEVICE_NAME		"xp0"
-
-
-/*
- * When messages are queued with xpc_send_notify, a kmalloc'd buffer
- * of the following type is passed as a notification cookie.  When the
- * notification function is called, we use the cookie to decide
- * whether all outstanding message sends have completed.  The skb can
- * then be released.
- */
-struct xpnet_pending_msg {
-	struct list_head free_list;
-	struct sk_buff *skb;
-	atomic_t use_count;
-};
-
-/* driver specific structure pointed to by the device structure */
-struct xpnet_dev_private {
-	struct net_device_stats stats;
-};
-
-struct net_device *xpnet_device;
-
-/*
- * When we are notified of other partitions activating, we add them to
- * our bitmask of partitions to which we broadcast.
- */
-static u64 xpnet_broadcast_partitions;
-/* protect above */
-static DEFINE_SPINLOCK(xpnet_broadcast_lock);
-
-/*
- * Since the Block Transfer Engine (BTE) is being used for the transfer
- * and it relies upon cache-line size transfers, we need to reserve at
- * least one cache-line for head and tail alignment.  The BTE is
- * limited to 8MB transfers.
- *
- * Testing has shown that changing MTU to greater than 64KB has no effect
- * on TCP as the two sides negotiate a Max Segment Size that is limited
- * to 64K.  Other protocols May use packets greater than this, but for
- * now, the default is 64KB.
- */
-#define XPNET_MAX_MTU (0x800000UL - L1_CACHE_BYTES)
-/* 32KB has been determined to be the ideal */
-#define XPNET_DEF_MTU (0x8000UL)
-
-
-/*
- * The partition id is encapsulated in the MAC address.  The following
- * define locates the octet the partid is in.
- */
-#define XPNET_PARTID_OCTET	1
-#define XPNET_LICENSE_OCTET	2
-
-
-/*
- * Define the XPNET debug device structure that is to be used with dev_dbg(),
- * dev_err(), dev_warn(), and dev_info().
- */
-struct device_driver xpnet_dbg_name = {
-	.name = "xpnet"
-};
-
-struct device xpnet_dbg_subname = {
-	.bus_id = {0},			/* set to "" */
-	.driver = &xpnet_dbg_name
-};
-
-struct device *xpnet = &xpnet_dbg_subname;
-
-/*
- * Packet was recevied by XPC and forwarded to us.
- */
-static void
-xpnet_receive(partid_t partid, int channel, struct xpnet_message *msg)
-{
-	struct sk_buff *skb;
-	bte_result_t bret;
-	struct xpnet_dev_private *priv =
-		(struct xpnet_dev_private *) xpnet_device->priv;
-
-
-	if (!XPNET_VALID_MSG(msg)) {
-		/*
-		 * Packet with a different XPC version.  Ignore.
-		 */
-		xpc_received(partid, channel, (void *) msg);
-
-		priv->stats.rx_errors++;
-
-		return;
-	}
-	dev_dbg(xpnet, "received 0x%lx, %d, %d, %d\n", msg->buf_pa, msg->size,
-		msg->leadin_ignore, msg->tailout_ignore);
-
-
-	/* reserve an extra cache line */
-	skb = dev_alloc_skb(msg->size + L1_CACHE_BYTES);
-	if (!skb) {
-		dev_err(xpnet, "failed on dev_alloc_skb(%d)\n",
-			msg->size + L1_CACHE_BYTES);
-
-		xpc_received(partid, channel, (void *) msg);
-
-		priv->stats.rx_errors++;
-
-		return;
-	}
-
-	/*
-	 * The allocated skb has some reserved space.
-	 * In order to use bte_copy, we need to get the
-	 * skb->data pointer moved forward.
-	 */
-	skb_reserve(skb, (L1_CACHE_BYTES - ((u64)skb->data &
-					    (L1_CACHE_BYTES - 1)) +
-			  msg->leadin_ignore));
-
-	/*
-	 * Update the tail pointer to indicate data actually
-	 * transferred.
-	 */
-	skb_put(skb, (msg->size - msg->leadin_ignore - msg->tailout_ignore));
-
-	/*
-	 * Move the data over from the other side.
-	 */
-	if ((XPNET_VERSION_MINOR(msg->version) == 1) &&
-						(msg->embedded_bytes != 0)) {
-		dev_dbg(xpnet, "copying embedded message. memcpy(0x%p, 0x%p, "
-			"%lu)\n", skb->data, &msg->data,
-			(size_t) msg->embedded_bytes);
-
-		skb_copy_to_linear_data(skb, &msg->data, (size_t)msg->embedded_bytes);
-	} else {
-		dev_dbg(xpnet, "transferring buffer to the skb->data area;\n\t"
-			"bte_copy(0x%p, 0x%p, %hu)\n", (void *)msg->buf_pa,
-			(void *)__pa((u64)skb->data & ~(L1_CACHE_BYTES - 1)),
-			msg->size);
-
-		bret = bte_copy(msg->buf_pa,
-				__pa((u64)skb->data & ~(L1_CACHE_BYTES - 1)),
-				msg->size, (BTE_NOTIFY | BTE_WACQUIRE), NULL);
-
-		if (bret != BTE_SUCCESS) {
-			// >>> Need better way of cleaning skb.  Currently skb
-			// >>> appears in_use and we can't just call
-			// >>> dev_kfree_skb.
-			dev_err(xpnet, "bte_copy(0x%p, 0x%p, 0x%hx) returned "
-				"error=0x%x\n", (void *)msg->buf_pa,
-				(void *)__pa((u64)skb->data &
-							~(L1_CACHE_BYTES - 1)),
-				msg->size, bret);
-
-			xpc_received(partid, channel, (void *) msg);
-
-			priv->stats.rx_errors++;
-
-			return;
-		}
-	}
-
-	dev_dbg(xpnet, "<skb->head=0x%p skb->data=0x%p skb->tail=0x%p "
-		"skb->end=0x%p skb->len=%d\n", (void *) skb->head,
-		(void *)skb->data, skb_tail_pointer(skb), skb_end_pointer(skb),
-		skb->len);
-
-	skb->protocol = eth_type_trans(skb, xpnet_device);
-	skb->ip_summed = CHECKSUM_UNNECESSARY;
-
-	dev_dbg(xpnet, "passing skb to network layer\n"
-		KERN_DEBUG "\tskb->head=0x%p skb->data=0x%p skb->tail=0x%p "
-		"skb->end=0x%p skb->len=%d\n",
-		(void *)skb->head, (void *)skb->data, skb_tail_pointer(skb),
-		skb_end_pointer(skb), skb->len);
-
-
-	xpnet_device->last_rx = jiffies;
-	priv->stats.rx_packets++;
-	priv->stats.rx_bytes += skb->len + ETH_HLEN;
-
-	netif_rx_ni(skb);
-	xpc_received(partid, channel, (void *) msg);
-}
-
-
-/*
- * This is the handler which XPC calls during any sort of change in
- * state or message reception on a connection.
- */
-static void
-xpnet_connection_activity(enum xpc_retval reason, partid_t partid, int channel,
-			  void *data, void *key)
-{
-	long bp;
-
-
-	DBUG_ON(partid <= 0 || partid >= XP_MAX_PARTITIONS);
-	DBUG_ON(channel != XPC_NET_CHANNEL);
-
-	switch(reason) {
-	case xpcMsgReceived:	/* message received */
-		DBUG_ON(data == NULL);
-
-		xpnet_receive(partid, channel, (struct xpnet_message *) data);
-		break;
-
-	case xpcConnected:	/* connection completed to a partition */
-		spin_lock_bh(&xpnet_broadcast_lock);
-		xpnet_broadcast_partitions |= 1UL << (partid -1 );
-		bp = xpnet_broadcast_partitions;
-		spin_unlock_bh(&xpnet_broadcast_lock);
-
-		netif_carrier_on(xpnet_device);
-
-		dev_dbg(xpnet, "%s connection created to partition %d; "
-			"xpnet_broadcast_partitions=0x%lx\n",
-			xpnet_device->name, partid, bp);
-		break;
-
-	default:
-		spin_lock_bh(&xpnet_broadcast_lock);
-		xpnet_broadcast_partitions &= ~(1UL << (partid -1 ));
-		bp = xpnet_broadcast_partitions;
-		spin_unlock_bh(&xpnet_broadcast_lock);
-
-		if (bp == 0) {
-			netif_carrier_off(xpnet_device);
-		}
-
-		dev_dbg(xpnet, "%s disconnected from partition %d; "
-			"xpnet_broadcast_partitions=0x%lx\n",
-			xpnet_device->name, partid, bp);
-		break;
-
-	}
-}
-
-
-static int
-xpnet_dev_open(struct net_device *dev)
-{
-	enum xpc_retval ret;
-
-
-	dev_dbg(xpnet, "calling xpc_connect(%d, 0x%p, NULL, %ld, %ld, %ld, "
-		"%ld)\n", XPC_NET_CHANNEL, xpnet_connection_activity,
-		XPNET_MSG_SIZE, XPNET_MSG_NENTRIES, XPNET_MAX_KTHREADS,
-		XPNET_MAX_IDLE_KTHREADS);
-
-	ret = xpc_connect(XPC_NET_CHANNEL, xpnet_connection_activity, NULL,
-			  XPNET_MSG_SIZE, XPNET_MSG_NENTRIES,
-			  XPNET_MAX_KTHREADS, XPNET_MAX_IDLE_KTHREADS);
-	if (ret != xpcSuccess) {
-		dev_err(xpnet, "ifconfig up of %s failed on XPC connect, "
-			"ret=%d\n", dev->name, ret);
-
-		return -ENOMEM;
-	}
-
-	dev_dbg(xpnet, "ifconfig up of %s; XPC connected\n", dev->name);
-
-	return 0;
-}
-
-
-static int
-xpnet_dev_stop(struct net_device *dev)
-{
-	xpc_disconnect(XPC_NET_CHANNEL);
-
-	dev_dbg(xpnet, "ifconfig down of %s; XPC disconnected\n", dev->name);
-
-	return 0;
-}
-
-
-static int
-xpnet_dev_change_mtu(struct net_device *dev, int new_mtu)
-{
-	/* 68 comes from min TCP+IP+MAC header */
-	if ((new_mtu < 68) || (new_mtu > XPNET_MAX_MTU)) {
-		dev_err(xpnet, "ifconfig %s mtu %d failed; value must be "
-			"between 68 and %ld\n", dev->name, new_mtu,
-			XPNET_MAX_MTU);
-		return -EINVAL;
-	}
-
-	dev->mtu = new_mtu;
-	dev_dbg(xpnet, "ifconfig %s mtu set to %d\n", dev->name, new_mtu);
-	return 0;
-}
-
-
-/*
- * Required for the net_device structure.
- */
-static int
-xpnet_dev_set_config(struct net_device *dev, struct ifmap *new_map)
-{
-	return 0;
-}
-
-
-/*
- * Return statistics to the caller.
- */
-static struct net_device_stats *
-xpnet_dev_get_stats(struct net_device *dev)
-{
-	struct xpnet_dev_private *priv;
-
-
-	priv = (struct xpnet_dev_private *) dev->priv;
-
-	return &priv->stats;
-}
-
-
-/*
- * Notification that the other end has received the message and
- * DMA'd the skb information.  At this point, they are done with
- * our side.  When all recipients are done processing, we
- * release the skb and then release our pending message structure.
- */
-static void
-xpnet_send_completed(enum xpc_retval reason, partid_t partid, int channel,
-			void *__qm)
-{
-	struct xpnet_pending_msg *queued_msg =
-		(struct xpnet_pending_msg *) __qm;
-
-
-	DBUG_ON(queued_msg == NULL);
-
-	dev_dbg(xpnet, "message to %d notified with reason %d\n",
-		partid, reason);
-
-	if (atomic_dec_return(&queued_msg->use_count) == 0) {
-		dev_dbg(xpnet, "all acks for skb->head=-x%p\n",
-			(void *) queued_msg->skb->head);
-
-		dev_kfree_skb_any(queued_msg->skb);
-		kfree(queued_msg);
-	}
-}
-
-
-/*
- * Network layer has formatted a packet (skb) and is ready to place it
- * "on the wire".  Prepare and send an xpnet_message to all partitions
- * which have connected with us and are targets of this packet.
- *
- * MAC-NOTE:  For the XPNET driver, the MAC address contains the
- * destination partition_id.  If the destination partition id word
- * is 0xff, this packet is to broadcast to all partitions.
- */
-static int
-xpnet_dev_hard_start_xmit(struct sk_buff *skb, struct net_device *dev)
-{
-	struct xpnet_pending_msg *queued_msg;
-	enum xpc_retval ret;
-	struct xpnet_message *msg;
-	u64 start_addr, end_addr;
-	long dp;
-	u8 second_mac_octet;
-	partid_t dest_partid;
-	struct xpnet_dev_private *priv;
-	u16 embedded_bytes;
-
-
-	priv = (struct xpnet_dev_private *) dev->priv;
-
-
-	dev_dbg(xpnet, ">skb->head=0x%p skb->data=0x%p skb->tail=0x%p "
-		"skb->end=0x%p skb->len=%d\n", (void *) skb->head,
-		(void *)skb->data, skb_tail_pointer(skb), skb_end_pointer(skb),
-		skb->len);
-
-
-	/*
-	 * The xpnet_pending_msg tracks how many outstanding
-	 * xpc_send_notifies are relying on this skb.  When none
-	 * remain, release the skb.
-	 */
-	queued_msg = kmalloc(sizeof(struct xpnet_pending_msg), GFP_ATOMIC);
-	if (queued_msg == NULL) {
-		dev_warn(xpnet, "failed to kmalloc %ld bytes; dropping "
-			"packet\n", sizeof(struct xpnet_pending_msg));
-
-		priv->stats.tx_errors++;
-
-		return -ENOMEM;
-	}
-
-
-	/* get the beginning of the first cacheline and end of last */
-	start_addr = ((u64) skb->data & ~(L1_CACHE_BYTES - 1));
-	end_addr = L1_CACHE_ALIGN((u64)skb_tail_pointer(skb));
-
-	/* calculate how many bytes to embed in the XPC message */
-	embedded_bytes = 0;
-	if (unlikely(skb->len <= XPNET_MSG_DATA_MAX)) {
-		/* skb->data does fit so embed */
-		embedded_bytes = skb->len;
-	}
-
-
-	/*
-	 * Since the send occurs asynchronously, we set the count to one
-	 * and begin sending.  Any sends that happen to complete before
-	 * we are done sending will not free the skb.  We will be left
-	 * with that task during exit.  This also handles the case of
-	 * a packet destined for a partition which is no longer up.
-	 */
-	atomic_set(&queued_msg->use_count, 1);
-	queued_msg->skb = skb;
-
-
-	second_mac_octet = skb->data[XPNET_PARTID_OCTET];
-	if (second_mac_octet == 0xff) {
-		/* we are being asked to broadcast to all partitions */
-		dp = xpnet_broadcast_partitions;
-	} else if (second_mac_octet != 0) {
-		dp = xpnet_broadcast_partitions &
-					(1UL << (second_mac_octet - 1));
-	} else {
-		/* 0 is an invalid partid.  Ignore */
-		dp = 0;
-	}
-	dev_dbg(xpnet, "destination Partitions mask (dp) = 0x%lx\n", dp);
-
-	/*
-	 * If we wanted to allow promiscuous mode to work like an
-	 * unswitched network, this would be a good point to OR in a
-	 * mask of partitions which should be receiving all packets.
-	 */
-
-	/*
-	 * Main send loop.
-	 */
-	for (dest_partid = 1; dp && dest_partid < XP_MAX_PARTITIONS;
-	     dest_partid++) {
-
-
-		if (!(dp & (1UL << (dest_partid - 1)))) {
-			/* not destined for this partition */
-			continue;
-		}
-
-		/* remove this partition from the destinations mask */
-		dp &= ~(1UL << (dest_partid - 1));
-
-
-		/* found a partition to send to */
-
-		ret = xpc_allocate(dest_partid, XPC_NET_CHANNEL,
-				   XPC_NOWAIT, (void **)&msg);
-		if (unlikely(ret != xpcSuccess)) {
-			continue;
-		}
-
-		msg->embedded_bytes = embedded_bytes;
-		if (unlikely(embedded_bytes != 0)) {
-			msg->version = XPNET_VERSION_EMBED;
-			dev_dbg(xpnet, "calling memcpy(0x%p, 0x%p, 0x%lx)\n",
-				&msg->data, skb->data, (size_t) embedded_bytes);
-			skb_copy_from_linear_data(skb, &msg->data,
-						  (size_t)embedded_bytes);
-		} else {
-			msg->version = XPNET_VERSION;
-		}
-		msg->magic = XPNET_MAGIC;
-		msg->size = end_addr - start_addr;
-		msg->leadin_ignore = (u64) skb->data - start_addr;
-		msg->tailout_ignore = end_addr - (u64)skb_tail_pointer(skb);
-		msg->buf_pa = __pa(start_addr);
-
-		dev_dbg(xpnet, "sending XPC message to %d:%d\n"
-			KERN_DEBUG "msg->buf_pa=0x%lx, msg->size=%u, "
-			"msg->leadin_ignore=%u, msg->tailout_ignore=%u\n",
-			dest_partid, XPC_NET_CHANNEL, msg->buf_pa, msg->size,
-			msg->leadin_ignore, msg->tailout_ignore);
-
-
-		atomic_inc(&queued_msg->use_count);
-
-		ret = xpc_send_notify(dest_partid, XPC_NET_CHANNEL, msg,
-				      xpnet_send_completed, queued_msg);
-		if (unlikely(ret != xpcSuccess)) {
-			atomic_dec(&queued_msg->use_count);
-			continue;
-		}
-
-	}
-
-	if (atomic_dec_return(&queued_msg->use_count) == 0) {
-		dev_dbg(xpnet, "no partitions to receive packet destined for "
-			"%d\n", dest_partid);
-
-
-		dev_kfree_skb(skb);
-		kfree(queued_msg);
-	}
-
-	priv->stats.tx_packets++;
-	priv->stats.tx_bytes += skb->len;
-
-	return 0;
-}
-
-
-/*
- * Deal with transmit timeouts coming from the network layer.
- */
-static void
-xpnet_dev_tx_timeout (struct net_device *dev)
-{
-	struct xpnet_dev_private *priv;
-
-
-	priv = (struct xpnet_dev_private *) dev->priv;
-
-	priv->stats.tx_errors++;
-	return;
-}
-
-
-static int __init
-xpnet_init(void)
-{
-	int i;
-	u32 license_num;
-	int result = -ENOMEM;
-
-
-	if (!ia64_platform_is("sn2")) {
-		return -ENODEV;
-	}
-
-	dev_info(xpnet, "registering network device %s\n", XPNET_DEVICE_NAME);
-
-	/*
-	 * use ether_setup() to init the majority of our device
-	 * structure and then override the necessary pieces.
-	 */
-	xpnet_device = alloc_netdev(sizeof(struct xpnet_dev_private),
-				    XPNET_DEVICE_NAME, ether_setup);
-	if (xpnet_device == NULL) {
-		return -ENOMEM;
-	}
-
-	netif_carrier_off(xpnet_device);
-
-	xpnet_device->mtu = XPNET_DEF_MTU;
-	xpnet_device->change_mtu = xpnet_dev_change_mtu;
-	xpnet_device->open = xpnet_dev_open;
-	xpnet_device->get_stats = xpnet_dev_get_stats;
-	xpnet_device->stop = xpnet_dev_stop;
-	xpnet_device->hard_start_xmit = xpnet_dev_hard_start_xmit;
-	xpnet_device->tx_timeout = xpnet_dev_tx_timeout;
-	xpnet_device->set_config = xpnet_dev_set_config;
-
-	/*
-	 * Multicast assumes the LSB of the first octet is set for multicast
-	 * MAC addresses.  We chose the first octet of the MAC to be unlikely
-	 * to collide with any vendor's officially issued MAC.
-	 */
-	xpnet_device->dev_addr[0] = 0xfe;
-	xpnet_device->dev_addr[XPNET_PARTID_OCTET] = sn_partition_id;
-	license_num = sn_partition_serial_number_val();
-	for (i = 3; i >= 0; i--) {
-		xpnet_device->dev_addr[XPNET_LICENSE_OCTET + i] =
-							license_num & 0xff;
-		license_num = license_num >> 8;
-	}
-
-	/*
-	 * ether_setup() sets this to a multicast device.  We are
-	 * really not supporting multicast at this time.
-	 */
-	xpnet_device->flags &= ~IFF_MULTICAST;
-
-	/*
-	 * No need to checksum as it is a DMA transfer.  The BTE will
-	 * report an error if the data is not retrievable and the
-	 * packet will be dropped.
-	 */
-	xpnet_device->features = NETIF_F_NO_CSUM;
-
-	result = register_netdev(xpnet_device);
-	if (result != 0) {
-		free_netdev(xpnet_device);
-	}
-
-	return result;
-}
-module_init(xpnet_init);
-
-
-static void __exit
-xpnet_exit(void)
-{
-	dev_info(xpnet, "unregistering network device %s\n",
-		xpnet_device[0].name);
-
-	unregister_netdev(xpnet_device);
-
-	free_netdev(xpnet_device);
-}
-module_exit(xpnet_exit);
-
-
-MODULE_AUTHOR("Silicon Graphics, Inc.");
-MODULE_DESCRIPTION("Cross Partition Network adapter (XPNET)");
-MODULE_LICENSE("GPL");
-