xref: /illumos-gate/usr/src/uts/common/io/rsm/rsmka_pathmanager.c (revision b97d6ca7)

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */
/*
 * Copyright 2004 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 * Copyright 2012 Milan Jurik. All rights reserved.
 */

/*
 * This module provides for the management of interconnect adapters
 * inter-node connections (aka paths), and IPC.  Adapter descriptors are
 * maintained on a linked list; one list per adapter devname.  Each
 * adapter descriptor heads a linked list of path descriptors.  There is
 * also a linked list of ipc_info descriptors; one for each node.  Each
 * ipc_info descriptor heads a circular list of ipc tokens (the tokens are
 * embedded within a path descriptor). The tokens are used in round robin
 * fashion.
 *
 *
 * The exported interface consists of the following functions:
 *	- rsmka_add_adapter
 *	- rsmka_remove_adapter
 *
 *      [add_path and remove_path only called for current adapters]
 *	- rsmka_add_path
 *	- rsmka_remove_path	[a path down request is implicit]
 *
 *	- rsmka_path_up           [called at clock ipl for Sun Cluster]
 *	- rsmka_path_down         [called at clock ipl for Sun Cluster]
 *	- rsmka_disconnect_node   [called at clock ipl for Sun Cluster;
 *				treat like path-down for all node paths;
 *				can be before node_alive; always before
 *				node_died.]
 *
 *	[node_alive and node_died are always paired]
 *	- rsmka_node_alive   called after the first cluster path is up
 *                           for this node
 *	- rsmka_node_died
 *
 *      [set the local node id]
 *      - rsmka_set_my_nodeid    called to set the variable my_nodeid to the
 *                           local node id
 *
 * Processing for these functions is setup as a state machine supported
 * by the data structures described above.
 *
 * For Sun Cluster these are called from the Path-Manager/Kernel-Agent
 * Interface (rsmka_pm_interface.cc).
 *
 * The functions rsm_path_up, rsm_path_down, and rsm_disconnect_node are
 * called at clock interrupt level from the Path-Manager/Kernel-Agent
 * Interface which precludes sleeping; so these functions may (optionally)
 * defer processing to an independent thread running at normal ipl.
 *
 *
 * lock definitions:
 *
 *	(mutex) work_queue.work_mutex
 *			protects linked list of work tokens and used
 *			with cv_wait/cv_signal thread synchronization.
 *			No other locks acquired when held.
 *
 *	(mutex) adapter_listhead_base.listlock
 *			protects linked list of adapter listheads
 *			Always acquired before listhead->mutex
 *
 *
 *	(mutex) ipc_info_lock
 *			protects ipc_info list and sendq token lists
 *			Always acquired before listhead->mutex
 *
 *      (mutex) listhead->mutex
 *			protects adapter listhead, linked list of
 *			adapters, and linked list of paths.
 *
 *      (mutex) path->mutex
 *			protects the path descriptor.
 *			work_queue.work_mutex may be acquired when holding
 *			this lock.
 *
 *	(mutex) adapter->mutex
 *			protects adapter descriptor contents.  used
 *			mainly for ref_cnt update.
 */

#include <sys/param.h>
#include <sys/kmem.h>
#include <sys/errno.h>
#include <sys/time.h>
#include <sys/devops.h>
#include <sys/ddi_impldefs.h>
#include <sys/cmn_err.h>
#include <sys/ddi.h>
#include <sys/sunddi.h>
#include <sys/proc.h>
#include <sys/thread.h>
#include <sys/taskq.h>
#include <sys/callb.h>

#include <sys/rsm/rsm.h>
#include <rsm_in.h>
#include <sys/rsm/rsmka_path_int.h>

extern void _cplpl_init();
extern void _cplpl_fini();
extern pri_t maxclsyspri;
extern int   rsm_hash_size;

extern rsm_node_id_t my_nodeid;
extern rsmhash_table_t rsm_import_segs;
extern rsm_intr_hand_ret_t rsm_srv_func(rsm_controller_object_t *,
    rsm_intr_q_op_t, rsm_addr_t, void *, size_t, rsm_intr_hand_arg_t);
extern void rsmseg_unload(rsmseg_t *);
extern void rsm_suspend_complete(rsm_node_id_t src_node, int flag);
extern int rsmipc_send_controlmsg(path_t *path, int msgtype);
extern void rsmka_path_monitor_initialize();
extern void rsmka_path_monitor_terminate();

extern adapter_t loopback_adapter;
/*
 * Lint errors and warnings are displayed; informational messages
 * are suppressed.
 */
/* lint -w2 */


/*
 * macros SQ_TOKEN_TO_PATH and WORK_TOKEN_TO_PATH use a null pointer
 * for computational purposes.  Ignore the lint warning.
 */
/* lint -save -e413 */
/* FUNCTION PROTOTYPES */
static adapter_t *init_adapter(char *, int, rsm_addr_t,
    rsm_controller_handle_t, rsm_ops_t *, srv_handler_arg_t *);
adapter_t *rsmka_lookup_adapter(char *, int);
static ipc_info_t *lookup_ipc_info(rsm_node_id_t);
static ipc_info_t *init_ipc_info(rsm_node_id_t, boolean_t);
static path_t *lookup_path(char *, int, rsm_node_id_t, rsm_addr_t);
static void pathup_to_pathactive(ipc_info_t *, rsm_node_id_t);
static void path_importer_disconnect(path_t *);
boolean_t rsmka_do_path_active(path_t *, int);
static boolean_t do_path_up(path_t *, int);
static void do_path_down(path_t *, int);
static void enqueue_work(work_token_t *);
static boolean_t cancel_work(work_token_t *);
static void link_path(path_t *);
static void destroy_path(path_t *);
static void link_sendq_token(sendq_token_t *, rsm_node_id_t);
static void unlink_sendq_token(sendq_token_t *, rsm_node_id_t);
boolean_t rsmka_check_node_alive(rsm_node_id_t);
static void do_deferred_work(caddr_t);
static int create_ipc_sendq(path_t *);
static void destroy_ipc_info(ipc_info_t *);
void rsmka_pathmanager_cleanup();
void rsmka_release_adapter(adapter_t *);

kt_did_t rsm_thread_id;
int rsmka_terminate_workthread_loop = 0;

static struct adapter_listhead_list adapter_listhead_base;
static work_queue_t work_queue;

/* protect ipc_info descriptor manipulation */
static kmutex_t ipc_info_lock;

static ipc_info_t *ipc_info_head = NULL;

static int category = RSM_PATH_MANAGER | RSM_KERNEL_AGENT;

/* for synchronization with rsmipc_send() in rsm.c */
kmutex_t ipc_info_cvlock;
kcondvar_t ipc_info_cv;



/*
 * RSMKA PATHMANAGER INITIALIZATION AND CLEANUP ROUTINES
 *
 */


/*
 * Called from the rsm module (rsm.c)  _init() routine
 */
void
rsmka_pathmanager_init()
{
	kthread_t *tp;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_pathmanager_init enter\n"));

	/* initialization for locks and condition variables  */
	mutex_init(&work_queue.work_mutex, NULL, MUTEX_DEFAULT, NULL);
	mutex_init(&ipc_info_lock, NULL, MUTEX_DEFAULT, NULL);
	mutex_init(&ipc_info_cvlock, NULL, MUTEX_DEFAULT, NULL);
	mutex_init(&adapter_listhead_base.listlock, NULL,
	    MUTEX_DEFAULT, NULL);

	cv_init(&work_queue.work_cv, NULL, CV_DEFAULT, NULL);
	cv_init(&ipc_info_cv, NULL, CV_DEFAULT, NULL);

	tp = thread_create(NULL, 0, do_deferred_work, NULL, 0, &p0,
	    TS_RUN, maxclsyspri);
	rsm_thread_id = tp->t_did;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_pathmanager_init done\n"));
}

void
rsmka_pathmanager_cleanup()
{
	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_pathmanager_cleanup enter\n"));

	ASSERT(work_queue.head == NULL);

	/*
	 * In processing the remove path callbacks from the path monitor
	 * object, all deferred work will have been completed. So
	 * awaken the deferred work thread to give it a chance to exit
	 * the loop.
	 */
	mutex_enter(&work_queue.work_mutex);
	rsmka_terminate_workthread_loop++;
	cv_signal(&work_queue.work_cv);
	mutex_exit(&work_queue.work_mutex);

	/*
	 * Wait for the deferred work thread to exit before
	 * destroying the locks and cleaning up other data
	 * structures.
	 */
	if (rsm_thread_id)
		thread_join(rsm_thread_id);

	/*
	 * Destroy locks & condition variables
	 */
	mutex_destroy(&work_queue.work_mutex);
	cv_destroy(&work_queue.work_cv);

	mutex_enter(&ipc_info_lock);
	while (ipc_info_head)
		destroy_ipc_info(ipc_info_head);
	mutex_exit(&ipc_info_lock);

	mutex_destroy(&ipc_info_lock);

	mutex_destroy(&ipc_info_cvlock);
	cv_destroy(&ipc_info_cv);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_pathmanager_cleanup done\n"));

}

void
rsmka_set_my_nodeid(rsm_node_id_t local_nodeid)
{
	my_nodeid = local_nodeid;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsm: node %d \n", my_nodeid));

}

/*
 * DEFERRED WORK THREAD AND WORK QUEUE SUPPORT ROUTINES
 *
 */

/*
 * This function is the executable code of the thread which handles
 * deferred work.  Work is deferred when a function is called at
 * clock ipl and processing may require blocking.
 *
 *
 * The thread is created by a call to taskq_create in rsmka_pathmanager_init.
 * After creation, a call to taskq_dispatch causes this function to
 * execute.  It loops forever - blocked until work is enqueued from
 * rsmka_do_path_active, do_path_down, or rsmka_disconnect_node.
 * rsmka_pathmanager_cleanup (called from _fini) will
 * set rsmka_terminate_workthread_loop and the task processing will
 * terminate.
 */
static void
do_deferred_work(caddr_t arg /*ARGSUSED*/)
{

	adapter_t 			*adapter;
	path_t				*path;
	work_token_t			*work_token;
	int				work_opcode;
	rsm_send_q_handle_t		sendq_handle;
	int				error;
	timespec_t			tv;
	callb_cpr_t			cprinfo;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "do_deferred_work enter\n"));

	CALLB_CPR_INIT(&cprinfo, &work_queue.work_mutex, callb_generic_cpr,
	    "rsm_deferred_work");

	for (;;) {
		mutex_enter(&work_queue.work_mutex);

		if (rsmka_terminate_workthread_loop) {
			goto exit;
		}

		/* When there is no work to do, block here */
		while (work_queue.head == NULL) {
			/* Since no work to do, Safe to CPR */
			CALLB_CPR_SAFE_BEGIN(&cprinfo);
			cv_wait(&work_queue.work_cv, &work_queue.work_mutex);
			CALLB_CPR_SAFE_END(&cprinfo, &work_queue.work_mutex);

			if (rsmka_terminate_workthread_loop) {
				goto exit;
			}
		}

		/*
		 * Remove a work token and begin work
		 */
		work_token = work_queue.head;
		work_queue.head = work_token->next;
		if (work_queue.tail == work_token)
			work_queue.tail = NULL;

		work_opcode = work_token->opcode;
		path = WORK_TOKEN_TO_PATH(work_token, work_opcode -1);
		work_token->next = NULL;
		mutex_exit(&work_queue.work_mutex);


		switch (work_opcode) {
		case RSMKA_IPC_UP:
			DBG_PRINTF((category, RSM_DEBUG,
			    "do_deferred_work:up,  path = %lx\n", path));
			error = create_ipc_sendq(path);
			mutex_enter(&path->mutex);
			if (path->state != RSMKA_PATH_UP) {
				/*
				 * path state has changed, if sendq was created,
				 * destroy it and return. Don't need to worry
				 * about sendq ref_cnt since no one starts
				 * using the sendq till path state becomes
				 * active
				 */
				if (error == RSM_SUCCESS) {
					sendq_handle = path->sendq_token.
					    rsmpi_sendq_handle;
					path->sendq_token.rsmpi_sendq_handle =
					    NULL;
					adapter = path->local_adapter;
					mutex_exit(&path->mutex);

					if (sendq_handle != NULL) {
						adapter->rsmpi_ops->
						    rsm_sendq_destroy(
						    sendq_handle);
					}
					mutex_enter(&path->mutex);
				}
				/* free up work token */
				work_token->opcode = 0;

				/*
				 * decrement reference count for the path
				 * descriptor and signal for synchronization
				 * with rsmka_remove_path. PATH_HOLD_NOLOCK was
				 * done by rsmka_path_up.
				 */
				PATH_RELE_NOLOCK(path);
				mutex_exit(&path->mutex);
				break;
			}

			if (error == RSM_SUCCESS) {
				DBG_PRINTF((category, RSM_DEBUG,
				    "do_deferred_work:success on up\n"));
				/* clear flag since sendq_create succeeded */
				path->flags &= ~RSMKA_SQCREATE_PENDING;
				path->state = RSMKA_PATH_ACTIVE;

				/*
				 * now that path is active we send the
				 * RSMIPC_MSG_SQREADY to the remote endpoint
				 */
				path->procmsg_cnt = 0;
				path->sendq_token.msgbuf_avail = 0;

				/* Calculate local incarnation number */
				gethrestime(&tv);
				if (tv.tv_sec == RSM_UNKNOWN_INCN)
					tv.tv_sec = 1;
				path->local_incn = (int64_t)tv.tv_sec;

				/*
				 * if send fails here its due to some
				 * non-transient error because QUEUE_FULL is
				 * not possible here since we are the first
				 * message on this sendq. The error will cause
				 * the path to go down anyways, so ignore
				 * the return value.
				 */
				(void) rsmipc_send_controlmsg(path,
				    RSMIPC_MSG_SQREADY);
				/* wait for SQREADY_ACK message */
				path->flags |= RSMKA_WAIT_FOR_SQACK;
			} else {
				/*
				 * sendq create failed possibly because
				 * the remote end is not yet ready eg.
				 * handler not registered, set a flag
				 * so that when there is an indication
				 * that the remote end is ready
				 * rsmka_do_path_active will be retried.
				 */
				path->flags |= RSMKA_SQCREATE_PENDING;
			}

			/* free up work token */
			work_token->opcode = 0;

			/*
			 * decrement reference count for the path
			 * descriptor and signal for synchronization with
			 * rsmka_remove_path. PATH_HOLD_NOLOCK was done
			 * by rsmka_path_up.
			 */
			PATH_RELE_NOLOCK(path);
			mutex_exit(&path->mutex);

			break;
		case RSMKA_IPC_DOWN:
			DBG_PRINTF((category, RSM_DEBUG,
			    "do_deferred_work:down, path = %lx\n", path));

			/*
			 * Unlike the processing of path_down in the case
			 * where the RSMKA_NO_SLEEP flag is not set, here,
			 * the state of the path is changed directly to
			 * RSMKA_PATH_DOWN. This is because in this case
			 * where the RSMKA_NO_SLEEP flag is set, any other
			 * calls referring this path will just queue up
			 * and will be processed only after the path
			 * down processing has completed.
			 */
			mutex_enter(&path->mutex);
			path->state = RSMKA_PATH_DOWN;
			/*
			 * clear the WAIT_FOR_SQACK flag since path is down.
			 */
			path->flags &= ~RSMKA_WAIT_FOR_SQACK;

			/*
			 * this wakes up any thread waiting to receive credits
			 * in rsmipc_send to tell it that the path is down
			 * thus releasing the sendq.
			 */
			cv_broadcast(&path->sendq_token.sendq_cv);

			mutex_exit(&path->mutex);

			/* drain the messages from the receive msgbuf */
			taskq_wait(path->recv_taskq);

			/*
			 * The path_importer_disconnect function has to
			 * be called after releasing the mutex on the path
			 * in order to avoid any recursive mutex enter panics
			 */
			path_importer_disconnect(path);
			DBG_PRINTF((category, RSM_DEBUG,
			    "do_deferred_work: success on down\n"));
			/*
			 * decrement reference count for the path
			 * descriptor and signal for synchronization with
			 * rsmka_remove_path. PATH_HOLD_NOLOCK was done
			 * by rsmka_path_down.
			 */
			mutex_enter(&path->mutex);

#ifdef DEBUG
			/*
			 * Some IPC messages left in the recv_buf,
			 * they'll be dropped
			 */
			if (path->msgbuf_cnt != 0)
				cmn_err(CE_NOTE,
				    "path=%lx msgbuf_cnt != 0\n",
				    (uintptr_t)path);
#endif

			/*
			 * Don't want to destroy a send queue when a token
			 * has been acquired; so wait 'til the token is
			 * no longer referenced (with a cv_wait).
			 */
			while (path->sendq_token.ref_cnt != 0)
				cv_wait(&path->sendq_token.sendq_cv,
				    &path->mutex);

			sendq_handle = path->sendq_token.rsmpi_sendq_handle;
			path->sendq_token.rsmpi_sendq_handle = NULL;

			/* destroy the send queue and release the handle */
			if (sendq_handle != NULL) {
				adapter = path->local_adapter;
				adapter->rsmpi_ops->rsm_sendq_destroy(
				    sendq_handle);
			}

			work_token->opcode = 0;
			PATH_RELE_NOLOCK(path);
			mutex_exit(&path->mutex);
			break;
		default:
			DBG_PRINTF((category, RSM_DEBUG,
			    "do_deferred_work: bad work token opcode\n"));
			break;
		}
	}

exit:
	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "do_deferred_work done\n"));
	/*
	 * CALLB_CPR_EXIT does a mutex_exit for
	 * the work_queue.work_mutex
	 */
	CALLB_CPR_EXIT(&cprinfo);
}

/*
 * Work is inserted at the tail of the list and processed from the
 * head of the list.
 */
static void
enqueue_work(work_token_t *token)
{
	work_token_t	*tail_token;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "enqueue_work enter\n"));

	ASSERT(MUTEX_HELD(&work_queue.work_mutex));

	token->next = NULL;
	if (work_queue.head == NULL) {
		work_queue.head = work_queue.tail = token;
	} else {
		tail_token = work_queue.tail;
		work_queue.tail = tail_token->next = token;
	}

	/* wake up deferred work thread */
	cv_signal(&work_queue.work_cv);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "enqueue_work done\n"));
}


/*
 * If the work_token is found on the work queue, the work is cancelled
 * by removing the token from the work queue.
 *
 * Return true if a work_token was found and cancelled, otherwise return false
 *
 * enqueue_work increments the path refcnt to make sure that the path doesn't
 * go away, callers of cancel_work need to decrement the refcnt of the path to
 * which this work_token belongs if a work_token is found in the work_queue
 * and cancelled ie. when the return value is B_TRUE.
 */
static boolean_t
cancel_work(work_token_t *work_token)
{
	work_token_t	*current_token;
	work_token_t	*prev_token = NULL;
	boolean_t	cancelled = B_FALSE;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "cancel_work enter\n"));

	ASSERT(MUTEX_HELD(&work_queue.work_mutex));


	current_token = work_queue.head;
	while (current_token != NULL) {
		if (current_token == work_token) {
			if (work_token == work_queue.head)
				work_queue.head = work_token->next;
			else
				prev_token->next = work_token->next;
			if (work_token == work_queue.tail)
				work_queue.tail = prev_token;

			current_token->opcode = 0;
			current_token->next = NULL;
			/* found and cancelled work */
			cancelled = B_TRUE;
			DBG_PRINTF((category, RSM_DEBUG,
			    "cancelled_work = 0x%p\n", work_token));
			break;
		}
		prev_token = current_token;
		current_token = current_token->next;
	}

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "cancel_work done\n"));
	return (cancelled);
}

/*
 * EXTERNAL INTERFACES
 *
 * For Galileo Clustering, these routine are called from
 * rsmka_pm_interface.cc
 *
 */

/*
 *
 * If the adapter is supported by rsmpi then initialize an adapter descriptor
 * and link it to the list of adapters.  The adapter attributes are obtained
 * from rsmpi and stored in the descriptor.  Finally, a service handler
 * for incoming ipc on this adapter is registered with rsmpi.
 * A pointer for the adapter descriptor is returned as a cookie to the
 * caller.  The cookie may be use with subsequent calls to save the time of
 * adapter descriptor lookup.
 *
 * The adapter descriptor maintains a reference count which is intialized
 * to 1 and incremented on lookups; when a cookie is used in place of
 * a lookup, an explicit ADAPTER_HOLD is required.
 */

void *
rsmka_add_adapter(char *name, int instance, rsm_addr_t hwaddr)
{
	adapter_t		*adapter;
	rsm_controller_object_t	rsmpi_adapter_object;
	rsm_controller_handle_t	rsmpi_adapter_handle;
	rsm_ops_t		*rsmpi_ops_vector;
	int			adapter_is_supported;
	rsm_controller_attr_t	*attr;
	srv_handler_arg_t	*srv_hdlr_argp;
	int result;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsmka_add_adapter enter\n"));

	DBG_PRINTF((category, RSM_DEBUG,
	    "rsmka_add_adapter: name = %s instance = %d hwaddr = %llx \n",
	    name, instance, hwaddr));

	/* verify name length */
	if (strlen(name) >= MAXNAMELEN) {
		DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
		    "rsmka_add_adapter done: name too long\n"));
		return (NULL);
	}


	/* Check if rsmpi supports this adapter type */
	adapter_is_supported = rsm_get_controller(name, instance,
	    &rsmpi_adapter_object, RSM_VERSION);

	if (adapter_is_supported != RSM_SUCCESS) {
		DBG_PRINTF((category, RSM_ERR,
		    "rsmka_add_adapter done: adapter not supported\n"));
		return (NULL);
	}

	rsmpi_adapter_handle = rsmpi_adapter_object.handle;
	rsmpi_ops_vector = rsmpi_adapter_object.ops;

	/* Get adapter attributes */
	result = rsm_get_controller_attr(rsmpi_adapter_handle, &attr);
	if (result != RSM_SUCCESS) {
		DBG_PRINTF((category, RSM_ERR,
		    "rsm: get_controller_attr(%d) Failed %x\n",
		    instance, result));
		(void) rsm_release_controller(name, instance,
		    &rsmpi_adapter_object);
		return (NULL);
	}

	DBG_PRINTF((category, RSM_DEBUG,
	    "rsmka_add_adapter: register service offset = %d\n", hwaddr));

	/*
	 * create a srv_handler_arg_t object, initialize it and register
	 * it along with rsm_srv_func. This get passed as the
	 * rsm_intr_hand_arg_t when the handler gets invoked.
	 */
	srv_hdlr_argp = kmem_zalloc(sizeof (srv_handler_arg_t), KM_SLEEP);

	(void) strcpy(srv_hdlr_argp->adapter_name, name);
	srv_hdlr_argp->adapter_instance = instance;
	srv_hdlr_argp->adapter_hwaddr = hwaddr;

	/* Have rsmpi register the ipc receive handler for this adapter */
	/*
	 * Currently, we need to pass in a separate service identifier for
	 * each adapter. In order to obtain a unique service identifier
	 * value for an adapter, we add the hardware address of the
	 * adapter to the base service identifier(RSM_SERVICE which is
	 * defined as RSM_INTR_T_KA as per the RSMPI specification).
	 * NOTE: This may result in using some of the service identifier
	 * values defined for RSM_INTR_T_XPORT(the Sun Cluster Transport).
	 */
	result = rsmpi_ops_vector->rsm_register_handler(
	    rsmpi_adapter_handle, &rsmpi_adapter_object,
	    RSM_SERVICE+(uint_t)hwaddr, rsm_srv_func,
	    (rsm_intr_hand_arg_t)srv_hdlr_argp, NULL, 0);

	if (result != RSM_SUCCESS) {
		DBG_PRINTF((category, RSM_ERR,
		    "rsmka_add_adapter done: rsm_register_handler"
		    "failed %d\n",
		    instance));
		return (NULL);
	}

	/* Initialize an adapter descriptor and add it to the adapter list */
	adapter = init_adapter(name, instance, hwaddr,
	    rsmpi_adapter_handle, rsmpi_ops_vector, srv_hdlr_argp);

	/* Copy over the attributes from the pointer returned to us */
	adapter->rsm_attr = *attr;

	/*
	 * With the addition of the topology obtainment interface, applications
	 * now get the local nodeid from the topology data structure.
	 *
	 * adapter->rsm_attr.attr_node_id = my_nodeid;
	 */
	DBG_PRINTF((category, RSM_ERR,
	    "rsmka_add_adapter: adapter = %lx\n", adapter));

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsmka_add_adapter done\n"));

	/* return adapter pointer as a cookie for later fast access */
	return ((void *)adapter);
}


/*
 * Unlink the adapter descriptor and call rsmka_release_adapter which
 * will decrement the reference count and possibly free the desriptor.
 */
boolean_t
rsmka_remove_adapter(char *name, uint_t instance, void *cookie, int flags)
{
	adapter_t		*adapter;
	adapter_listhead_t	*listhead;
	adapter_t		*prev, *current;
	rsm_controller_object_t	rsm_cntl_obj;


	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_remove_adapter enter\n"));

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_remove_adapter: cookie = %lx\n", cookie));

	if (flags & RSMKA_USE_COOKIE) {
		adapter = (adapter_t *)cookie;
	} else {
		adapter = rsmka_lookup_adapter(name, instance);
		/*
		 * rsmka_lookup_adapter increments the ref_cnt; need
		 * to decrement here to get true count
		 */
		ADAPTER_RELE(adapter);
	}
	ASSERT(adapter->next_path == NULL);

	listhead = adapter->listhead;

	mutex_enter(&adapter_listhead_base.listlock);

	mutex_enter(&listhead->mutex);

	/* find the adapter in the list and remove it */
	prev = NULL;
	current = listhead->next_adapter;
	while (current != NULL) {
		if (adapter->instance == current->instance) {
			break;
		} else {
			prev = current;
			current = current->next;
		}
	}
	ASSERT(current != NULL);

	if (prev == NULL)
		listhead->next_adapter = current->next;
	else
		prev->next = current->next;

	listhead->adapter_count--;

	mutex_exit(&listhead->mutex);

	mutex_exit(&adapter_listhead_base.listlock);

	mutex_enter(&current->mutex);

	/*
	 * unregister the handler
	 */
	current->rsmpi_ops->rsm_unregister_handler(current->rsmpi_handle,
	    RSM_SERVICE+current->hwaddr, rsm_srv_func,
	    (rsm_intr_hand_arg_t)current->hdlr_argp);

	DBG_PRINTF((category, RSM_DEBUG, "rsmka_remove_adapter: unreg hdlr "
	    ":adapter=%lx, hwaddr=%lx\n", current, current->hwaddr));

	rsm_cntl_obj.handle = current->rsmpi_handle;
	rsm_cntl_obj.ops = current->rsmpi_ops;

	(void) rsm_release_controller(current->listhead->adapter_devname,
	    current->instance, &rsm_cntl_obj);

	mutex_exit(&current->mutex);

	rsmka_release_adapter(current);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_remove_adapter done\n"));

	return (B_TRUE);
}

/*
 * An adapter descriptor will exist from an earlier add_adapter. This
 * function does:
 *		initialize the path descriptor
 *		initialize the ipc descriptor (it may already exist)
 *		initialize and link a sendq token for this path
 */
void *
rsmka_add_path(char *adapter_name, int adapter_instance,
    rsm_node_id_t remote_node,
    rsm_addr_t remote_hwaddr, int rem_adapt_instance,
    void *cookie, int flags)
{

	path_t			*path;
	adapter_t		*adapter;
	char			tq_name[TASKQ_NAMELEN];

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsmka_add_path enter\n"));

	/* allocate new path descriptor */
	path = kmem_zalloc(sizeof (path_t), KM_SLEEP);

	if (flags & RSMKA_USE_COOKIE) {
		adapter = (adapter_t *)cookie;
		ADAPTER_HOLD(adapter);
	} else {
		adapter = rsmka_lookup_adapter(adapter_name, adapter_instance);
	}

	DBG_PRINTF((category, RSM_DEBUG,
	    "rsmka_add_path: adapter = %lx\n", adapter));

	/*
	 * initialize path descriptor
	 * don't need to increment adapter reference count because
	 * it can't be removed if paths exist for it.
	 */
	mutex_init(&path->mutex, NULL, MUTEX_DEFAULT, NULL);

	PATH_HOLD(path);
	path->state = RSMKA_PATH_DOWN;
	path->remote_node = remote_node;
	path->remote_hwaddr = remote_hwaddr;
	path->remote_devinst = rem_adapt_instance;
	path->local_adapter = adapter;

	/* taskq is for sendq on adapter with remote_hwaddr on remote_node */
	(void) snprintf(tq_name, sizeof (tq_name), "%x_%llx",
	    remote_node, (unsigned long long) remote_hwaddr);

	path->recv_taskq = taskq_create_instance(tq_name, adapter_instance,
	    RSMKA_ONE_THREAD, maxclsyspri, RSMIPC_MAX_MESSAGES,
	    RSMIPC_MAX_MESSAGES, TASKQ_PREPOPULATE);

	/* allocate the message buffer array */
	path->msgbuf_queue = (msgbuf_elem_t *)kmem_zalloc(
	    RSMIPC_MAX_MESSAGES * sizeof (msgbuf_elem_t), KM_SLEEP);

	/*
	 * init cond variables for synch with rsmipc_send()
	 * and rsmka_remove_path
	 */
	cv_init(&path->sendq_token.sendq_cv, NULL, CV_DEFAULT, NULL);
	cv_init(&path->hold_cv, NULL, CV_DEFAULT, NULL);

	/* link path descriptor on adapter path list */
	link_path(path);

	/* link the path sendq token on the ipc_info token list */
	link_sendq_token(&path->sendq_token, remote_node);

	/* ADAPTER_HOLD done above by rsmka_lookup_adapter */
	ADAPTER_RELE(adapter);

	DBG_PRINTF((category, RSM_DEBUG, "rsmka_add_path: path = %lx\n", path));

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsmka_add_path done\n"));
	return ((void *)path);
}

/*
 * Wait for the path descriptor reference count to become zero then
 * directly call path down processing.  Finally, unlink the sendq token and
 * free the path descriptor memory.
 *
 * Note: lookup_path locks the path and increments the path hold count
 */
void
rsmka_remove_path(char *adapter_name, int instance, rsm_node_id_t remote_node,
    rsm_addr_t remote_hwaddr, void *path_cookie, int flags)
{
	path_t		*path;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsmka_remove_path enter\n"));

	if (flags & RSMKA_USE_COOKIE) {
		path = (path_t *)path_cookie;
		mutex_enter(&path->mutex);
	} else {
		path = lookup_path(adapter_name, instance,  remote_node,
		    remote_hwaddr);

		/*
		 * remember, lookup_path increments the reference
		 * count - so decrement now so we can get to zero
		 */
		PATH_RELE_NOLOCK(path);
	}

	DBG_PRINTF((category, RSM_DEBUG,
	    "rsmka_remove_path: path = %lx\n", path));

	while (path->state == RSMKA_PATH_GOING_DOWN)
		cv_wait(&path->hold_cv, &path->mutex);

	/* attempt to cancel any possibly pending work */
	mutex_enter(&work_queue.work_mutex);
	if (cancel_work(&path->work_token[RSMKA_IPC_UP_INDEX])) {
		PATH_RELE_NOLOCK(path);
	}
	if (cancel_work(&path->work_token[RSMKA_IPC_DOWN_INDEX])) {
		PATH_RELE_NOLOCK(path);
	}
	mutex_exit(&work_queue.work_mutex);

	/*
	 * The path descriptor ref cnt was set to 1 initially when
	 * the path was added.  So we need to do a decrement here to
	 * balance that.
	 */
	PATH_RELE_NOLOCK(path);

	switch (path->state) {
	case RSMKA_PATH_UP:
		/* clear the flag */
		path->flags &= ~RSMKA_SQCREATE_PENDING;
		path->state = RSMKA_PATH_DOWN;
		break;
	case RSMKA_PATH_DOWN:
		break;

	case RSMKA_PATH_ACTIVE:
		/*
		 * rsmka_remove_path should not call do_path_down
		 * with the RSMKA_NO_SLEEP flag set since for
		 * this code path, the deferred work would
		 * incorrectly do a PATH_RELE_NOLOCK.
		 */
		do_path_down(path, 0);
		break;
	default:
		mutex_exit(&path->mutex);
		DBG_PRINTF((category, RSM_ERR,
		    "rsm_remove_path: invalid path state %d\n",
		    path->state));
		return;

	}

	/*
	 * wait for all references to the path to be released. If a thread
	 * was waiting to receive credits do_path_down should wake it up
	 * since the path is going down and that will cause the sleeping
	 * thread to release its hold on the path.
	 */
	while (path->ref_cnt != 0) {
		cv_wait(&path->hold_cv, &path->mutex);
	}

	mutex_exit(&path->mutex);

	/*
	 * remove from ipc token list
	 * NOTE: use the remote_node value from the path structure
	 * since for RSMKA_USE_COOKIE being set, the remote_node
	 * value passed into rsmka_remove_path is 0.
	 */
	unlink_sendq_token(&path->sendq_token, path->remote_node);

	/* unlink from adapter path list and free path descriptor */
	destroy_path(path);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsmka_remove_path done\n"));
}

/*
 *
 * LOCKING:
 * lookup_path locks the path and increments the path hold count. If the remote
 * node is not in the alive state, do_path_up will release the lock and
 * decrement the hold count.  Otherwise rsmka_do_path_active will release the
 * lock prior to waking up the work thread.
 *
 * REF_CNT:
 * The path descriptor ref_cnt is incremented here; it will be decremented
 * when path up processing is completed in do_path_up or by the work thread
 * if the path up is deferred.
 *
 */
boolean_t
rsmka_path_up(char *adapter_name, uint_t adapter_instance,
    rsm_node_id_t remote_node, rsm_addr_t remote_hwaddr,
    void *path_cookie, int flags)
{

	path_t			*path;
	boolean_t		rval = B_TRUE;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsmka_path_up enter\n"));

	if (flags & RSMKA_USE_COOKIE) {
		path = (path_t *)path_cookie;
		mutex_enter(&path->mutex);
		PATH_HOLD_NOLOCK(path);
	} else {
		path = lookup_path(adapter_name, adapter_instance,
		    remote_node, remote_hwaddr);
	}

	while (path->state == RSMKA_PATH_GOING_DOWN)
		cv_wait(&path->hold_cv, &path->mutex);

	DBG_PRINTF((category, RSM_DEBUG, "rsmka_path_up: path = %lx\n", path));
	rval = do_path_up(path, flags);
	mutex_exit(&path->mutex);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsmka_path_up done\n"));
	return (rval);
}

/*
 *
 * LOCKING:
 * lookup_path locks the path and increments the path hold count. If the
 * current state is ACTIVE the path lock is release prior to waking up
 * the work thread in do_path_down .  The work thread will decrement the hold
 * count when the work for this is finished.
 *
 *
 * REF_CNT:
 * The path descriptor ref_cnt is incremented here; it will be decremented
 * when path down processing is completed in do_path_down or by the work thread
 * if the path down is deferred.
 *
 */
boolean_t
rsmka_path_down(char *adapter_devname, int instance, rsm_node_id_t remote_node,
    rsm_addr_t remote_hwaddr,  void *path_cookie, int flags)
{
	path_t			*path;
	boolean_t		rval = B_TRUE;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsmka_path_down enter\n"));

	if (flags & RSMKA_USE_COOKIE) {
		path = (path_t *)path_cookie;
		mutex_enter(&path->mutex);
		PATH_HOLD_NOLOCK(path);
	} else {
		path = lookup_path(adapter_devname, instance, remote_node,
		    remote_hwaddr);
	}

	while (path->state == RSMKA_PATH_GOING_DOWN)
		cv_wait(&path->hold_cv, &path->mutex);

	DBG_PRINTF((category, RSM_DEBUG,
	    "rsmka_path_down: path = %lx\n", path));

	switch (path->state) {
	case RSMKA_PATH_UP:
		/* clear the flag */
		path->flags &= ~RSMKA_SQCREATE_PENDING;
		path->state = RSMKA_PATH_GOING_DOWN;
		mutex_exit(&path->mutex);

		/*
		 * release path->mutex since enqueued tasks acquire it.
		 * Drain all the enqueued tasks.
		 */
		taskq_wait(path->recv_taskq);

		mutex_enter(&path->mutex);
		path->state = RSMKA_PATH_DOWN;
		PATH_RELE_NOLOCK(path);
		break;
	case RSMKA_PATH_DOWN:
		PATH_RELE_NOLOCK(path);
		break;
	case RSMKA_PATH_ACTIVE:
		do_path_down(path, flags);
		/*
		 * Need to release the path refcnt. Either done in do_path_down
		 * or do_deferred_work for RSMKA_NO_SLEEP being set. Has to be
		 * done here for RSMKA_NO_SLEEP not set.
		 */
		if (!(flags & RSMKA_NO_SLEEP))
			PATH_RELE_NOLOCK(path);
		break;
	default:
		DBG_PRINTF((category, RSM_ERR,
		    "rsm_path_down: invalid path state %d\n", path->state));
		rval = B_FALSE;
	}

	mutex_exit(&path->mutex);
	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsmka_path_down done\n"));
	return (rval);
}


/*
 * Paths cannot become active until node_is_alive is marked true
 * in the ipc_info descriptor for the node
 *
 * In the event this is called before any paths have been added,
 * init_ipc_info if called here.
 *
 */
boolean_t
rsmka_node_alive(rsm_node_id_t remote_node)
{
	ipc_info_t *ipc_info;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsmka_node_alive enter\n"));

	DBG_PRINTF((category, RSM_DEBUG,
	    "rsmka_node_alive: remote_node = %x\n", remote_node));

	ipc_info = lookup_ipc_info(remote_node);

	if (ipc_info == NULL) {
		ipc_info = init_ipc_info(remote_node, B_TRUE);
		DBG_PRINTF((category, RSM_DEBUG,
		    "rsmka_node_alive: new ipc_info = %lx\n", ipc_info));
	} else {
		ASSERT(ipc_info->node_is_alive == B_FALSE);
		ipc_info->node_is_alive = B_TRUE;
	}

	pathup_to_pathactive(ipc_info, remote_node);

	mutex_exit(&ipc_info_lock);

	/* rsmipc_send() may be waiting for a sendq_token */
	mutex_enter(&ipc_info_cvlock);
	cv_broadcast(&ipc_info_cv);
	mutex_exit(&ipc_info_cvlock);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsmka_node_alive done\n"));

	return (B_TRUE);
}



/*
 * Paths cannot become active when node_is_alive is marked false
 * in the ipc_info descriptor for the node
 */
boolean_t
rsmka_node_died(rsm_node_id_t remote_node)
{
	ipc_info_t *ipc_info;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsmka_node_died enter\n"));

	DBG_PRINTF((category, RSM_DEBUG,
	    "rsmka_node_died: remote_node = %x\n", remote_node));

	ipc_info = lookup_ipc_info(remote_node);
	if (ipc_info == NULL)
		return (B_FALSE);

	ASSERT(ipc_info->node_is_alive == B_TRUE);
	ipc_info->node_is_alive = B_FALSE;

	rsm_suspend_complete(remote_node, RSM_SUSPEND_NODEDEAD);

	mutex_exit(&ipc_info_lock);

	/* rsmipc_send() may be waiting for a sendq_token */
	mutex_enter(&ipc_info_cvlock);
	cv_broadcast(&ipc_info_cv);
	mutex_exit(&ipc_info_cvlock);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsmka_node_died done\n"));

	return (B_TRUE);
}

/*
 * Treat like path_down for all paths for the specified remote node.
 * Always invoked before node died.
 *
 * NOTE: This routine is not called from the cluster path interface; the
 * rsmka_path_down is called directly for each path.
 */
void
rsmka_disconnect_node(rsm_node_id_t remote_node, int flags)
{
	ipc_info_t	*ipc_info;
	path_t		*path;
	sendq_token_t	*sendq_token;
	work_token_t 	*up_token;
	work_token_t 	*down_token;
	boolean_t	do_work = B_FALSE;
	boolean_t	cancelled = B_FALSE;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_disconnect_node enter\n"));

	DBG_PRINTF((category, RSM_DEBUG,
	    "rsmka_disconnect_node: node = %d\n", remote_node));

	if (flags & RSMKA_NO_SLEEP) {
		ipc_info = lookup_ipc_info(remote_node);

		sendq_token = ipc_info->token_list;

		while (sendq_token != NULL) {
			path = SQ_TOKEN_TO_PATH(sendq_token);
			PATH_HOLD(path);
			up_token = &path->work_token[RSMKA_IPC_UP_INDEX];
			down_token = &path->work_token[RSMKA_IPC_DOWN_INDEX];

			mutex_enter(&work_queue.work_mutex);

			/* if an up token is enqueued, remove it */
			cancelled = cancel_work(up_token);

			/*
			 * If the path is active and down work hasn't
			 * already been setup then down work is needed.
			 * else
			 * if up work wasn't canceled because it was
			 * already being processed then down work is needed
			 */
			if (path->state == RSMKA_PATH_ACTIVE) {
				if (down_token->opcode == 0)
					do_work = B_TRUE;
			} else
				if (up_token->opcode == RSMKA_IPC_UP)
					do_work = B_TRUE;

			if (do_work == B_TRUE) {
				down_token->opcode = RSMKA_IPC_DOWN;
				enqueue_work(down_token);
			}
			mutex_exit(&work_queue.work_mutex);

			if (do_work == B_FALSE)
				PATH_RELE(path);

			if (cancelled) {
				PATH_RELE(path);
			}
			sendq_token = sendq_token->next;
		}

		/*
		 * Now that all the work is enqueued, wakeup the work
		 * thread.
		 */
		mutex_enter(&work_queue.work_mutex);
		cv_signal(&work_queue.work_cv);
		mutex_exit(&work_queue.work_mutex);

		IPCINFO_RELE_NOLOCK(ipc_info);
		mutex_exit(&ipc_info_lock);

	} else {
		/* get locked ipc_info descriptor */
		ipc_info = lookup_ipc_info(remote_node);

		sendq_token = ipc_info->token_list;
		while (sendq_token != NULL) {
			path = SQ_TOKEN_TO_PATH(sendq_token);
			DBG_PRINTF((category, RSM_DEBUG,
			    "rsmka_disconnect_node: path_down"
			    "for path = %x\n",
			    path));
			(void) rsmka_path_down(0, 0, 0, 0,
			    path, RSMKA_USE_COOKIE);
			sendq_token = sendq_token->next;
			if (sendq_token == ipc_info->token_list)
				break;
		}
		mutex_exit(&ipc_info_lock);
	}

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_disconnect_node done\n"));
}


/*
 * Called from rsm_node_alive - if a path to a remote node is in
 * state RSMKA_PATH_UP, transition the state to RSMKA_PATH_ACTIVE with a
 * call to rsmka_do_path_active.
 *
 * REF_CNT:
 * The path descriptor ref_cnt is incremented here; it will be decremented
 * when path up processing is completed in rsmka_do_path_active or by the work
 * thread if the path up is deferred.
 */
static void
pathup_to_pathactive(ipc_info_t *ipc_info, rsm_node_id_t remote_node)
{
	path_t		*path;
	sendq_token_t	*token;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "pathup_to_pathactive enter\n"));

	remote_node = remote_node;

	ASSERT(MUTEX_HELD(&ipc_info_lock));

	token = ipc_info->token_list;
	while (token != NULL) {
		path = SQ_TOKEN_TO_PATH(token);
		mutex_enter(&path->mutex);
		if (path->state == RSMKA_PATH_UP)  {
			PATH_HOLD_NOLOCK(path);
			(void) rsmka_do_path_active(path, 0);
		}
		mutex_exit(&path->mutex);
		token = token->next;
		if (token == ipc_info->token_list)
			break;
	}

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "pathup_to_pathactive done\n"));
}

/*
 * Called from pathup_to_pathactive and do_path_up. The objective is to
 * create an ipc send queue and transition to state RSMKA_PATH_ACTIVE.
 * For the no sleep case we may need to defer the work using a token.
 *
 */
boolean_t
rsmka_do_path_active(path_t *path, int flags)
{
	work_token_t	*up_token = &path->work_token[RSMKA_IPC_UP_INDEX];
	work_token_t	*down_token = &path->work_token[RSMKA_IPC_DOWN_INDEX];
	boolean_t	do_work = B_FALSE;
	int		error;
	timespec_t	tv;
	adapter_t	*adapter;
	rsm_send_q_handle_t	sqhdl;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_do_path_active enter\n"));

	ASSERT(MUTEX_HELD(&path->mutex));

	if (flags & RSMKA_NO_SLEEP) {
		mutex_enter(&work_queue.work_mutex);

		/* if a down token is enqueued, remove it */
		if (cancel_work(down_token)) {
			PATH_RELE_NOLOCK(path);
		}

		/*
		 * If the path is not active and up work hasn't
		 * already been setup then up work is needed.
		 * else
		 * if down work wasn't canceled because it was
		 * already being processed then up work is needed
		 */
		if (path->state != RSMKA_PATH_ACTIVE) {
			if (up_token->opcode == 0)
				do_work = B_TRUE;
		} else
			if (down_token->opcode == RSMKA_IPC_DOWN)
				do_work = B_TRUE;

		if (do_work == B_TRUE) {
			up_token->opcode = RSMKA_IPC_UP;
			enqueue_work(up_token);
		}
		else
			PATH_RELE_NOLOCK(path);

		mutex_exit(&work_queue.work_mutex);
		DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
		    "rsmka_do_path_active done\n"));
		return (B_TRUE);
	} else {
		/*
		 * Drop the path lock before calling create_ipc_sendq, shouldn't
		 * hold locks across calls to RSMPI routines.
		 */
		mutex_exit(&path->mutex);

		error = create_ipc_sendq(path);

		mutex_enter(&path->mutex);
		if (path->state != RSMKA_PATH_UP) {
			/*
			 * path state has changed, if sendq was created,
			 * destroy it and return
			 */
			if (error == RSM_SUCCESS) {
				sqhdl = path->sendq_token.rsmpi_sendq_handle;
				path->sendq_token.rsmpi_sendq_handle = NULL;
				adapter = path->local_adapter;
				mutex_exit(&path->mutex);

				if (sqhdl != NULL) {
					adapter->rsmpi_ops->rsm_sendq_destroy(
					    sqhdl);
				}
				mutex_enter(&path->mutex);
			}
			PATH_RELE_NOLOCK(path);

			DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
			    "rsmka_do_path_active done: path=%lx not UP\n",
			    (uintptr_t)path));
			return (error ? B_FALSE : B_TRUE);
		}

		if (error == RSM_SUCCESS) {
			/* clear flag since sendq_create succeeded */
			path->flags &= ~RSMKA_SQCREATE_PENDING;
			path->state = RSMKA_PATH_ACTIVE;
			/*
			 * now that path is active we send the
			 * RSMIPC_MSG_SQREADY to the remote endpoint
			 */
			path->procmsg_cnt = 0;
			path->sendq_token.msgbuf_avail = 0;

			/* Calculate local incarnation number */
			gethrestime(&tv);
			if (tv.tv_sec == RSM_UNKNOWN_INCN)
				tv.tv_sec = 1;
			path->local_incn = (int64_t)tv.tv_sec;

			/*
			 * if send fails here its due to some non-transient
			 * error because QUEUE_FULL is not possible here since
			 * we are the first message on this sendq. The error
			 * will cause the path to go down anyways so ignore
			 * the return value
			 */
			(void) rsmipc_send_controlmsg(path, RSMIPC_MSG_SQREADY);
			/* wait for SQREADY_ACK message */
			path->flags |= RSMKA_WAIT_FOR_SQACK;

			DBG_PRINTF((category, RSM_DEBUG,
			    "rsmka_do_path_active success\n"));
		} else {
			/*
			 * sendq create failed possibly because
			 * the remote end is not yet ready eg.
			 * handler not registered, set a flag
			 * so that when there is an indication
			 * that the remote end is ready rsmka_do_path_active
			 * will be retried.
			 */
			path->flags |= RSMKA_SQCREATE_PENDING;
		}

		PATH_RELE_NOLOCK(path);

		DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
		    "rsmka_do_path_active done\n"));
		return (error ? B_FALSE : B_TRUE);
	}

}

/*
 * Called from rsm_path_up.
 * If the remote node state is "alive" then call rsmka_do_path_active
 * otherwise just transition path state to RSMKA_PATH_UP.
 */
static boolean_t
do_path_up(path_t *path, int flags)
{
	boolean_t	rval;
	boolean_t	node_alive;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "do_path_up enter\n"));

	ASSERT(MUTEX_HELD(&path->mutex));

	/* path moved to ACTIVE by rsm_sqcreateop_callback - just return */
	if (path->state == RSMKA_PATH_ACTIVE) {
		DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
		    "do_path_up done: already ACTIVE\n"));
		PATH_RELE_NOLOCK(path);
		return (B_TRUE);
	}

	path->state = RSMKA_PATH_UP;

	/* initialize the receive msgbuf counters */
	path->msgbuf_head = 0;
	path->msgbuf_tail = RSMIPC_MAX_MESSAGES - 1;
	path->msgbuf_cnt = 0;
	path->procmsg_cnt = 0;
	/*
	 * rsmka_check_node_alive acquires ipc_info_lock, in order to maintain
	 * correct lock ordering drop the path lock before calling it.
	 */
	mutex_exit(&path->mutex);

	node_alive = rsmka_check_node_alive(path->remote_node);

	mutex_enter(&path->mutex);
	if (node_alive == B_TRUE)
		rval = rsmka_do_path_active(path, flags);
	else {
		PATH_RELE_NOLOCK(path);
		rval = B_TRUE;
	}

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "do_path_up done\n"));
	return (rval);
}



/*
 * Called from rsm_remove_path, rsm_path_down, deferred_work.
 * Destroy the send queue on this path.
 * Disconnect segments being imported from the remote node
 * Disconnect segments being imported by the remote node
 *
 */
static void
do_path_down(path_t *path, int flags)
{
	work_token_t *up_token = &path->work_token[RSMKA_IPC_UP_INDEX];
	work_token_t *down_token = &path->work_token[RSMKA_IPC_DOWN_INDEX];
	boolean_t do_work = B_FALSE;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "do_path_down enter\n"));

	ASSERT(MUTEX_HELD(&path->mutex));

	if (flags & RSMKA_NO_SLEEP) {
		mutex_enter(&work_queue.work_mutex);
		DBG_PRINTF((category, RSM_DEBUG,
		    "do_path_down: after work_mutex\n"));

		/* if an up token is enqueued, remove it */
		if (cancel_work(up_token)) {
			PATH_RELE_NOLOCK(path);
		}

		/*
		 * If the path is active and down work hasn't
		 * already been setup then down work is needed.
		 * else
		 * if up work wasn't canceled because it was
		 * already being processed then down work is needed
		 */
		if (path->state == RSMKA_PATH_ACTIVE) {
			if (down_token->opcode == 0)
				do_work = B_TRUE;
		} else
			if (up_token->opcode == RSMKA_IPC_UP)
				do_work = B_TRUE;

		if (do_work == B_TRUE) {
			down_token->opcode = RSMKA_IPC_DOWN;
			enqueue_work(down_token);
		} else
			PATH_RELE_NOLOCK(path);


		mutex_exit(&work_queue.work_mutex);

	} else {

		/*
		 * Change state of the path to RSMKA_PATH_GOING_DOWN and
		 * release the path mutex. Any other thread referring
		 * this path would cv_wait till the state of the path
		 * remains RSMKA_PATH_GOING_DOWN.
		 * On completing the path down processing, change the
		 * state of RSMKA_PATH_DOWN indicating that the path
		 * is indeed down.
		 */
		path->state = RSMKA_PATH_GOING_DOWN;

		/*
		 * clear the WAIT_FOR_SQACK flag since path is going down.
		 */
		path->flags &= ~RSMKA_WAIT_FOR_SQACK;

		/*
		 * this wakes up any thread waiting to receive credits
		 * in rsmipc_send to tell it that the path is going down
		 */
		cv_broadcast(&path->sendq_token.sendq_cv);

		mutex_exit(&path->mutex);

		/*
		 * drain the messages from the receive msgbuf, the
		 * tasks in the taskq_thread acquire the path->mutex
		 * so we drop the path mutex before taskq_wait.
		 */
		taskq_wait(path->recv_taskq);

		/*
		 * Disconnect segments being imported from the remote node
		 * The path_importer_disconnect function needs to be called
		 * only after releasing the mutex on the path. This is to
		 * avoid a recursive mutex enter when doing the
		 * rsmka_get_sendq_token.
		 */
		path_importer_disconnect(path);

		/*
		 * Get the path mutex, change the state of the path to
		 * RSMKA_PATH_DOWN since the path down processing has
		 * completed and cv_signal anyone who was waiting since
		 * the state was RSMKA_PATH_GOING_DOWN.
		 * NOTE: Do not do a mutex_exit here. We entered this
		 * routine with the path lock held by the caller. The
		 * caller eventually releases the path lock by doing a
		 * mutex_exit.
		 */
		mutex_enter(&path->mutex);

#ifdef DEBUG
		/*
		 * Some IPC messages left in the recv_buf,
		 * they'll be dropped
		 */
		if (path->msgbuf_cnt != 0)
			cmn_err(CE_NOTE, "path=%lx msgbuf_cnt != 0\n",
			    (uintptr_t)path);
#endif
		while (path->sendq_token.ref_cnt != 0)
			cv_wait(&path->sendq_token.sendq_cv,
			    &path->mutex);

		/* release the rsmpi handle */
		if (path->sendq_token.rsmpi_sendq_handle != NULL)
			path->local_adapter->rsmpi_ops->rsm_sendq_destroy(
			    path->sendq_token.rsmpi_sendq_handle);

		path->sendq_token.rsmpi_sendq_handle = NULL;

		path->state = RSMKA_PATH_DOWN;

		cv_signal(&path->hold_cv);

	}

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "do_path_down done\n"));

}

/*
 * Search through the list of imported segments for segments using this path
 * and unload the memory mappings for each one.  The application will
 * get an error return when a barrier close is invoked.
 * NOTE: This function has to be called only after releasing the mutex on
 * the path. This is to avoid any recursive mutex panics on the path mutex
 * since the path_importer_disconnect function would end up calling
 * rsmka_get_sendq_token which requires the path mutex.
 */

static void
path_importer_disconnect(path_t *path)
{
	int i;
	adapter_t *adapter = path->local_adapter;
	rsm_node_id_t remote_node = path->remote_node;
	rsmresource_t		*p = NULL;
	rsmseg_t *seg;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "path_importer_disconnect enter\n"));

	rw_enter(&rsm_import_segs.rsmhash_rw, RW_READER);

	if (rsm_import_segs.bucket != NULL) {
		for (i = 0; i < rsm_hash_size; i++) {
			p = rsm_import_segs.bucket[i];
			for (; p; p = p->rsmrc_next) {
				if ((p->rsmrc_node == remote_node) &&
				    (p->rsmrc_adapter == adapter)) {
					seg = (rsmseg_t *)p;
			/*
			 * In order to make rsmseg_unload and
			 * path_importer_disconnect thread safe, acquire the
			 * segment lock here. rsmseg_unload is responsible for
			 * releasing the lock. rsmseg_unload releases the lock
			 * just before a call to rsmipc_send or in case of an
			 * early exit which occurs if the segment was in the
			 * state RSM_STATE_CONNECTING or RSM_STATE_NEW.
			 */
					rsmseglock_acquire(seg);
					seg->s_flags |= RSM_FORCE_DISCONNECT;
					rsmseg_unload(seg);
				}
			}
		}
	}
	rw_exit(&rsm_import_segs.rsmhash_rw);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "path_importer_disconnect done\n"));
}




/*
 *
 * ADAPTER UTILITY FUNCTIONS
 *
 */



/*
 * Allocate new adapter list head structure and add it to the beginning of
 * the list of adapter list heads.  There is one list for each adapter
 * device name (or type).
 */
static adapter_listhead_t *
init_listhead(char *name)
{
	adapter_listhead_t *listhead;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "init_listhead enter\n"));

	/* allocation and initialization */
	listhead = kmem_zalloc(sizeof (adapter_listhead_t), KM_SLEEP);
	mutex_init(&listhead->mutex, NULL, MUTEX_DEFAULT, NULL);
	(void) strcpy(listhead->adapter_devname, name);

	/* link into list of listheads */
	mutex_enter(&adapter_listhead_base.listlock);
	if (adapter_listhead_base.next == NULL) {
		adapter_listhead_base.next = listhead;
		listhead->next_listhead = NULL;
	} else {
		listhead->next_listhead = adapter_listhead_base.next;
		adapter_listhead_base.next = listhead;
	}
	mutex_exit(&adapter_listhead_base.listlock);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "init_listhead done\n"));

	return (listhead);
}


/*
 * Search the list of adapter list heads for a match on name.
 *
 */
static adapter_listhead_t *
lookup_adapter_listhead(char *name)
{
	adapter_listhead_t *listhead;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "lookup_adapter_listhead enter\n"));

	mutex_enter(&adapter_listhead_base.listlock);
	listhead = adapter_listhead_base.next;
	while (listhead != NULL) {
		if (strcmp(name, listhead->adapter_devname) == 0)
			break;
		listhead = listhead->next_listhead;
	}
	mutex_exit(&adapter_listhead_base.listlock);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "lookup_adapter_listhead done\n"));

	return (listhead);
}


/*
 * Get the adapter list head corresponding to devname and search for
 * an adapter descriptor with a match on the instance number. If
 * successful, increment the descriptor reference count and return
 * the descriptor pointer to the caller.
 *
 */
adapter_t *
rsmka_lookup_adapter(char *devname, int instance)
{
	adapter_listhead_t *listhead;
	adapter_t *current = NULL;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_lookup_adapter enter\n"));

	listhead = lookup_adapter_listhead(devname);
	if (listhead != NULL) {
		mutex_enter(&listhead->mutex);

		current = listhead->next_adapter;
		while (current != NULL) {
			if (current->instance == instance) {
				ADAPTER_HOLD(current);
				break;
			} else
				current = current->next;
		}

		mutex_exit(&listhead->mutex);
	}

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_lookup_adapter done\n"));

	return (current);
}

/*
 * Called from rsmka_remove_adapter or rsmseg_free.
 * rsm_bind() and rsm_connect() store the adapter pointer returned
 * from rsmka_getadapter.  The pointer is kept in the segment descriptor.
 * When the segment is freed, this routine is called by rsmseg_free to decrement
 * the adapter descriptor reference count and possibly free the
 * descriptor.
 */
void
rsmka_release_adapter(adapter_t *adapter)
{
	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_release_adapter enter\n"));

	if (adapter == &loopback_adapter) {
		DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
		    "rsmka_release_adapter done\n"));
		return;
	}

	mutex_enter(&adapter->mutex);

	/* decrement reference count */
	ADAPTER_RELE_NOLOCK(adapter);

	/*
	 * if the adapter descriptor reference count is equal to the
	 * initialization value of one, then the descriptor has been
	 * unlinked and can now be freed.
	 */
	if (adapter->ref_cnt == 1) {
		mutex_exit(&adapter->mutex);

		mutex_destroy(&adapter->mutex);
		kmem_free(adapter->hdlr_argp, sizeof (srv_handler_arg_t));
		kmem_free(adapter, sizeof (adapter_t));
	}
	else
		mutex_exit(&adapter->mutex);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_release_adapter done\n"));

}



/*
 * Singly linked list. Add to the front.
 */
static void
link_adapter(adapter_t *adapter)
{

	adapter_listhead_t *listhead;
	adapter_t *current;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "link_adapter enter\n"));

	mutex_enter(&adapter_listhead_base.listlock);

	mutex_enter(&adapter->listhead->mutex);

	listhead = adapter->listhead;
	current = listhead->next_adapter;
	listhead->next_adapter = adapter;
	adapter->next = current;
	ADAPTER_HOLD(adapter);

	adapter->listhead->adapter_count++;

	mutex_exit(&adapter->listhead->mutex);

	mutex_exit(&adapter_listhead_base.listlock);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "link_adapter done\n"));
}


/*
 * Return adapter descriptor
 *
 * lookup_adapter_listhead returns with the the list of adapter listheads
 * locked.  After adding the adapter descriptor, the adapter listhead list
 * lock is dropped.
 */
static adapter_t *
init_adapter(char *name, int instance, rsm_addr_t hwaddr,
    rsm_controller_handle_t handle, rsm_ops_t *ops,
    srv_handler_arg_t *hdlr_argp)
{
	adapter_t *adapter;
	adapter_listhead_t *listhead;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "init_adapter enter\n"));

	adapter = kmem_zalloc(sizeof (adapter_t), KM_SLEEP);
	adapter->instance = instance;
	adapter->hwaddr = hwaddr;
	adapter->rsmpi_handle = handle;
	adapter->rsmpi_ops = ops;
	adapter->hdlr_argp = hdlr_argp;
	mutex_init(&adapter->mutex, NULL, MUTEX_DEFAULT, NULL);
	ADAPTER_HOLD(adapter);


	listhead = lookup_adapter_listhead(name);
	if (listhead == NULL)  {
		listhead = init_listhead(name);
	}

	adapter->listhead = listhead;

	link_adapter(adapter);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "init_adapter done\n"));

	return (adapter);
}

/*
 *
 * PATH UTILITY FUNCTIONS
 *
 */


/*
 * Search the per adapter path list for a match on remote node and
 * hwaddr.  The path ref_cnt must be greater than zero or the path
 * is in the process of being removed.
 *
 * Acquire the path lock and increment the path hold count.
 */
static path_t *
lookup_path(char *adapter_devname, int adapter_instance,
    rsm_node_id_t remote_node, rsm_addr_t hwaddr)
{
	path_t		*current;
	adapter_t	*adapter;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "lookup_path enter\n"));

	adapter = rsmka_lookup_adapter(adapter_devname, adapter_instance);
	ASSERT(adapter != NULL);

	mutex_enter(&adapter->listhead->mutex);

	/* start at the list head */
	current = adapter->next_path;

	while (current != NULL) {
		if ((current->remote_node == remote_node) &&
		    (current->remote_hwaddr == hwaddr) &&
		    (current->ref_cnt > 0))
			break;
		else
			current = current->next_path;
	}
	if (current != NULL) {
		mutex_enter(&current->mutex);
		PATH_HOLD_NOLOCK(current);
	}

	mutex_exit(&adapter->listhead->mutex);
	ADAPTER_RELE(adapter);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "lookup_path done\n"));

	return (current);
}

/*
 * This interface is similar to lookup_path but takes only the local
 * adapter name, instance and remote adapters hwaddr to identify the
 * path. This is used in the interrupt handler routines where nodeid
 * is not always available.
 */
path_t *
rsm_find_path(char *adapter_devname, int adapter_instance, rsm_addr_t hwaddr)
{
	path_t		*current;
	adapter_t	*adapter;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsm_find_path enter\n"));

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsm_find_path:adapter=%s:%d,rem=%llx\n",
	    adapter_devname, adapter_instance, hwaddr));

	adapter = rsmka_lookup_adapter(adapter_devname, adapter_instance);

	/*
	 * its possible that we are here due to an interrupt but the adapter
	 * has been removed after we received the callback.
	 */
	if (adapter == NULL)
		return (NULL);

	mutex_enter(&adapter->listhead->mutex);

	/* start at the list head */
	current = adapter->next_path;

	while (current != NULL) {
		if ((current->remote_hwaddr == hwaddr) &&
		    (current->ref_cnt > 0))
			break;
		else
			current = current->next_path;
	}
	if (current != NULL) {
		mutex_enter(&current->mutex);
		PATH_HOLD_NOLOCK(current);
	}

	mutex_exit(&adapter->listhead->mutex);

	rsmka_release_adapter(adapter);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rsm_find_path done\n"));

	return (current);
}


/*
 * Add the path to the head of the (per adapter) list of paths
 */
static void
link_path(path_t *path)
{

	adapter_t *adapter = path->local_adapter;
	path_t *first_path;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "link_path enter\n"));

	mutex_enter(&adapter_listhead_base.listlock);

	mutex_enter(&adapter->listhead->mutex);

	first_path = adapter->next_path;
	adapter->next_path = path;
	path->next_path = first_path;

	adapter->listhead->path_count++;

	mutex_exit(&adapter->listhead->mutex);

	mutex_exit(&adapter_listhead_base.listlock);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "link_path done\n"));
}

/*
 * Search the per-adapter list of paths for the specified path, beginning
 * at the head of the list.  Unlink the path and free the descriptor
 * memory.
 */
static void
destroy_path(path_t *path)
{

	adapter_t *adapter = path->local_adapter;
	path_t *prev, *current;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "destroy_path enter\n"));

	mutex_enter(&adapter_listhead_base.listlock);

	mutex_enter(&path->local_adapter->listhead->mutex);
	ASSERT(path->ref_cnt == 0);

	/* start at the list head */
	prev = NULL;
	current =  adapter->next_path;

	while (current != NULL) {
		if (path->remote_node == current->remote_node &&
		    path->remote_hwaddr == current->remote_hwaddr)
			break;
		else {
			prev = current;
			current = current->next_path;
		}
	}

	if (prev == NULL)
		adapter->next_path = current->next_path;
	else
		prev->next_path = current->next_path;

	path->local_adapter->listhead->path_count--;

	mutex_exit(&path->local_adapter->listhead->mutex);

	mutex_exit(&adapter_listhead_base.listlock);

	taskq_destroy(path->recv_taskq);

	kmem_free(path->msgbuf_queue,
	    RSMIPC_MAX_MESSAGES * sizeof (msgbuf_elem_t));

	mutex_destroy(&current->mutex);
	cv_destroy(&current->sendq_token.sendq_cv);
	cv_destroy(&path->hold_cv);
	kmem_free(current, sizeof (path_t));

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "destroy_path done\n"));
}

void
rsmka_enqueue_msgbuf(path_t *path, void *data)
{
	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_enqueue_msgbuf enter\n"));

	ASSERT(MUTEX_HELD(&path->mutex));

	ASSERT(path->msgbuf_cnt < RSMIPC_MAX_MESSAGES);

	/* increment the count and advance the tail */

	path->msgbuf_cnt++;

	if (path->msgbuf_tail == RSMIPC_MAX_MESSAGES - 1) {
		path->msgbuf_tail = 0;
	} else {
		path->msgbuf_tail++;
	}

	path->msgbuf_queue[path->msgbuf_tail].active = B_TRUE;

	bcopy(data, &(path->msgbuf_queue[path->msgbuf_tail].msg),
	    sizeof (rsmipc_request_t));

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_enqueue_msgbuf done\n"));

}

/*
 * get the head of the queue using rsmka_gethead_msgbuf and then call
 * rsmka_dequeue_msgbuf to remove it.
 */
void
rsmka_dequeue_msgbuf(path_t *path)
{
	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_dequeue_msgbuf enter\n"));

	ASSERT(MUTEX_HELD(&path->mutex));

	if (path->msgbuf_cnt == 0)
		return;

	path->msgbuf_cnt--;

	path->msgbuf_queue[path->msgbuf_head].active = B_FALSE;

	if (path->msgbuf_head == RSMIPC_MAX_MESSAGES - 1) {
		path->msgbuf_head = 0;
	} else {
		path->msgbuf_head++;
	}

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_dequeue_msgbuf done\n"));

}

msgbuf_elem_t *
rsmka_gethead_msgbuf(path_t *path)
{
	msgbuf_elem_t	*head;

	ASSERT(MUTEX_HELD(&path->mutex));

	if (path->msgbuf_cnt == 0)
		return (NULL);

	head = &path->msgbuf_queue[path->msgbuf_head];

	return (head);

}
/*
 * Called by rsm_connect which needs the hardware address of the
 * remote adapter.  A search is done through the paths for the local
 * adapter for a match on the specified remote node.
 */
rsm_node_id_t
get_remote_nodeid(adapter_t *adapter, rsm_addr_t remote_hwaddr)
{

	rsm_node_id_t remote_node;
	path_t	   *current = adapter->next_path;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "get_remote_nodeid enter\n"));

	mutex_enter(&adapter->listhead->mutex);
	while (current != NULL) {
		if (current->remote_hwaddr == remote_hwaddr) {
			remote_node = current->remote_node;
			break;
		}
		current = current->next_path;
	}

	if (current == NULL)
		remote_node = (rsm_node_id_t)-1;

	mutex_exit(&adapter->listhead->mutex);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "get_remote_nodeid done\n"));

	return (remote_node);
}

/*
 * Called by rsm_connect which needs the hardware address of the
 * remote adapter.  A search is done through the paths for the local
 * adapter for a match on the specified remote node.
 */
rsm_addr_t
get_remote_hwaddr(adapter_t *adapter, rsm_node_id_t remote_node)
{

	rsm_addr_t remote_hwaddr;
	path_t	   *current = adapter->next_path;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "get_remote_hwaddr enter\n"));

	mutex_enter(&adapter->listhead->mutex);
	while (current != NULL) {
		if (current->remote_node == remote_node) {
			remote_hwaddr = current->remote_hwaddr;
			break;
		}
		current = current->next_path;
	}
	if (current == NULL)
		remote_hwaddr = -1;
	mutex_exit(&adapter->listhead->mutex);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "get_remote_hwaddr done\n"));

	return (remote_hwaddr);
}
/*
 * IPC UTILITY FUNCTIONS
 */


/*
 * If an entry exists, return with the ipc_info_lock held
 */
static ipc_info_t *
lookup_ipc_info(rsm_node_id_t remote_node)
{
	ipc_info_t  *ipc_info;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "lookup_ipc_info enter\n"));

	mutex_enter(&ipc_info_lock);

	ipc_info = ipc_info_head;
	if (ipc_info == NULL) {
		mutex_exit(&ipc_info_lock);
		DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
		    "lookup_ipc_info done: ipc_info is NULL\n"));
		return (NULL);
	}

	while (ipc_info->remote_node != remote_node) {
		ipc_info = ipc_info->next;
		if (ipc_info == NULL) {
			DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
			    "lookup_ipc_info: ipc_info not found\n"));
			mutex_exit(&ipc_info_lock);
			break;
		}
	}

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "lookup_ipc_info done\n"));

	return (ipc_info);
}

/*
 * Create an ipc_info descriptor and return with ipc_info_lock held
 */
static ipc_info_t *
init_ipc_info(rsm_node_id_t remote_node, boolean_t state)
{
	ipc_info_t *ipc_info;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "init_ipc_info enter\n"));

	/*
	 * allocate an ipc_info descriptor and add it to a
	 * singly linked list
	 */

	ipc_info = kmem_zalloc(sizeof (ipc_info_t), KM_SLEEP);
	ipc_info->remote_node = remote_node;
	ipc_info->node_is_alive = state;

	mutex_enter(&ipc_info_lock);
	if (ipc_info_head == NULL) {
		DBG_PRINTF((category, RSM_DEBUG,
		    "init_ipc_info:ipc_info_head = %lx\n", ipc_info));
		ipc_info_head = ipc_info;
		ipc_info->next = NULL;
	} else {
		ipc_info->next = ipc_info_head;
		ipc_info_head = ipc_info;
	}

	ipc_info->remote_node = remote_node;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "init_ipc_info done\n"));

	return (ipc_info);
}

static void
destroy_ipc_info(ipc_info_t *ipc_info)
{
	ipc_info_t *current = ipc_info_head;
	ipc_info_t *prev;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "destroy_ipc_info enter\n"));

	ASSERT(MUTEX_HELD(&ipc_info_lock));

	while (current != ipc_info) {
		prev = current;
		current = current->next;
	}
	ASSERT(current != NULL);

	if (current != ipc_info_head)
		prev->next = current->next;
	else
		ipc_info_head = current->next;

	kmem_free(current, sizeof (ipc_info_t));

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "destroy_ipc_info done\n"));

}

/*
 * Sendq tokens are kept on a circular list.  If tokens A, B, C, & D are
 * on the list headed by ipc_info, then ipc_info points to A, A points to
 * D, D to C, C to B, and B to A.
 */
static void
link_sendq_token(sendq_token_t *token, rsm_node_id_t remote_node)
{
	ipc_info_t *ipc_info;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "link_sendq_token enter\n"));

	ipc_info = lookup_ipc_info(remote_node);
	if (ipc_info == NULL) {
		ipc_info = init_ipc_info(remote_node, B_FALSE);
		DBG_PRINTF((category, RSM_DEBUG,
		    "link_sendq_token: new ipc_info = %lx\n", ipc_info));
	}
	else
		DBG_PRINTF((category, RSM_DEBUG,
		    "link_sendq_token: ipc_info = %lx\n", ipc_info));

	if (ipc_info->token_list == NULL) {
		ipc_info->token_list = token;
		ipc_info->current_token = token;
		DBG_PRINTF((category, RSM_DEBUG,
		    "link_sendq_token: current = %lx\n", token));
		token->next = token;
	} else {
		DBG_PRINTF((category, RSM_DEBUG,
		    "link_sendq_token: token = %lx\n", token));
		token->next = ipc_info->token_list->next;
		ipc_info->token_list->next = token;
		ipc_info->token_list = token;
	}


	mutex_exit(&ipc_info_lock);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "link_sendq_token done\n"));

}

static void
unlink_sendq_token(sendq_token_t *token, rsm_node_id_t remote_node)
{
	sendq_token_t *prev, *start,  *current;
	ipc_info_t *ipc_info;
	path_t *path = SQ_TOKEN_TO_PATH(token);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "unlink_sendq_token enter\n"));

	ASSERT(path->ref_cnt == 0);

	ipc_info = lookup_ipc_info(remote_node);
	if (ipc_info == NULL) {
		DBG_PRINTF((category, RSM_DEBUG,
		    "ipc_info for %d not found\n", remote_node));
		return;
	}

	prev = ipc_info->token_list;
	start = current = ipc_info->token_list->next;

	for (;;) {
		if (current == token) {
			if (current->next != current) {
				DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
				    "found token, removed it\n"));
				prev->next = token->next;
				if (ipc_info->token_list == token)
					ipc_info->token_list = prev;
				ipc_info->current_token = token->next;
			} else {
				/* list will be empty  */
				DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
				    "removed token, list empty\n"));
				ipc_info->token_list = NULL;
				ipc_info->current_token = NULL;
			}
			break;
		}
		prev = current;
		current = current->next;
		if (current == start) {
			DBG_PRINTF((category, RSM_DEBUG,
			    "unlink_sendq_token: token not found\n"));
			break;
		}
	}
	mutex_exit(&ipc_info_lock);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "unlink_sendq_token done\n"));
}


void
rele_sendq_token(sendq_token_t *token)
{
	path_t *path;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rele_sendq_token enter\n"));

	path = SQ_TOKEN_TO_PATH(token);
	mutex_enter(&path->mutex);
	PATH_RELE_NOLOCK(path);
	SENDQ_TOKEN_RELE(path);
	mutex_exit(&path->mutex);

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "rele_sendq_token done\n"));

}

/*
 * A valid ipc token can only be returned if the remote node is alive.
 * Tokens are on a circular list.  Starting with the current token
 * search for a token with an endpoint in state RSM_PATH_ACTIVE.
 * rsmipc_send which calls rsmka_get_sendq_token expects that if there are
 * multiple paths available between a node-pair then consecutive calls from
 * a particular invocation of rsmipc_send will return a sendq that is
 * different from the one that was used in the previous iteration. When
 * prev_used is NULL it indicates that this is the first interation in a
 * specific rsmipc_send invocation.
 *
 * Updating the current token provides round robin selection and this
 * is done only in the first iteration ie. when prev_used is NULL
 */
sendq_token_t *
rsmka_get_sendq_token(rsm_node_id_t remote_node, sendq_token_t *prev_used)
{
	sendq_token_t *token, *first_token;
	path_t *path;
	ipc_info_t *ipc_info;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
	    "rsmka_get_sendq_token enter\n"));

	ipc_info = lookup_ipc_info(remote_node);
	if (ipc_info == NULL) {
		DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
		    "rsmka_get_sendq_token done: ipc_info is NULL\n"));
		return (NULL);
	}

	if (ipc_info->node_is_alive == B_TRUE) {
		token = first_token = ipc_info->current_token;
		if (token == NULL) {
			mutex_exit(&ipc_info_lock);
			DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
			    "rsmka_get_sendq_token done: token=NULL\n"));
			return (NULL);
		}

		for (;;) {
			path = SQ_TOKEN_TO_PATH(token);
			DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
			    "path %lx\n", path));
			mutex_enter(&path->mutex);
			if (path->state != RSMKA_PATH_ACTIVE ||
			    path->ref_cnt == 0) {
				mutex_exit(&path->mutex);
			} else {
				if (token != prev_used) {
					/* found a new token */
					break;
				}
				mutex_exit(&path->mutex);
			}

			token = token->next;
			if (token == first_token) {
				/*
				 * we didn't find a new token reuse prev_used
				 * if the corresponding path is still up
				 */
				if (prev_used) {
					path = SQ_TOKEN_TO_PATH(prev_used);
					DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
					    "path %lx\n", path));
					mutex_enter(&path->mutex);
					if (path->state != RSMKA_PATH_ACTIVE ||
					    path->ref_cnt == 0) {
						mutex_exit(&path->mutex);
					} else {
						token = prev_used;
						break;
					}
				}
				mutex_exit(&ipc_info_lock);
				DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
				    "rsmka_get_sendq_token: token=NULL\n"));
				return (NULL);
			}
		}

		PATH_HOLD_NOLOCK(path);
		SENDQ_TOKEN_HOLD(path);
		if (prev_used == NULL) {
			/* change current_token only the first time */
			ipc_info->current_token = token->next;
		}

		mutex_exit(&path->mutex);
		mutex_exit(&ipc_info_lock);

		DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
		    "rsmka_get_sendq_token done\n"));
		return (token);
	} else {
		mutex_exit(&ipc_info_lock);
		DBG_PRINTF((category, RSM_DEBUG_VERBOSE,
		    "rsmka_get_sendq_token done\n"));
		return (NULL);
	}
}



/*
 */
static int
create_ipc_sendq(path_t *path)
{
	int		rval;
	sendq_token_t	*token;
	adapter_t 	*adapter;
	int64_t		srvc_offset;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "create_ipc_sendq enter\n"));

	DBG_PRINTF((category, RSM_DEBUG, "create_ipc_sendq: path = %lx\n",
	    path));

	adapter = path->local_adapter;
	token = &path->sendq_token;

	srvc_offset = path->remote_hwaddr;

	DBG_PRINTF((category, RSM_DEBUG,
	    "create_ipc_sendq: srvc_offset = %lld\n",
	    srvc_offset));

	rval = adapter->rsmpi_ops->rsm_sendq_create(adapter->rsmpi_handle,
	    path->remote_hwaddr,
	    (rsm_intr_service_t)(RSM_SERVICE+srvc_offset),
	    (rsm_intr_pri_t)RSM_PRI, (size_t)RSM_QUEUE_SZ,
	    RSM_INTR_SEND_Q_NO_FENCE,
	    RSM_RESOURCE_SLEEP, NULL, &token->rsmpi_sendq_handle);
	if (rval == RSM_SUCCESS) {
		/* rsmipc_send() may be waiting for a sendq_token */
		mutex_enter(&ipc_info_cvlock);
		cv_broadcast(&ipc_info_cv);
		mutex_exit(&ipc_info_cvlock);
	}

	DBG_PRINTF((category, RSM_DEBUG, "create_ipc_sendq: handle = %lx\n",
	    token->rsmpi_sendq_handle));
	DBG_PRINTF((category, RSM_DEBUG, "create_ipc_sendq: rval = %d\n",
	    rval));
	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "create_ipc_sendq done\n"));

	return (rval);
}


boolean_t
rsmka_check_node_alive(rsm_node_id_t remote_node)
{
	ipc_info_t *ipc_info;

	DBG_PRINTF((category, RSM_DEBUG, "rsmka_check_node_alive enter\n"));

	ipc_info = lookup_ipc_info(remote_node);
	if (ipc_info == NULL) {
		DBG_PRINTF((category, RSM_DEBUG,
		    "rsmka_check_node_alive done: ipc_info NULL\n"));
		return (B_FALSE);
	}

	if (ipc_info->node_is_alive == B_TRUE) {
		mutex_exit(&ipc_info_lock);
		DBG_PRINTF((category, RSM_DEBUG,
		    "rsmka_check_node_alive done: node is alive\n"));
		return (B_TRUE);
	} else {
		mutex_exit(&ipc_info_lock);
		DBG_PRINTF((category, RSM_DEBUG,
		    "rsmka_check_node_alive done: node is not alive\n"));
		return (B_FALSE);
	}
}




/*
 *  TOPOLOGY IOCTL SUPPORT
 */

static uint32_t
get_topology_size(int mode)
{
	uint32_t	topology_size;
	int		pointer_area_size;
	adapter_listhead_t	*listhead;
	int		total_num_of_adapters;
	int		total_num_of_paths;

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "get_topology_size enter\n"));

	/*
	 * Find the total number of adapters and paths by adding up the
	 * individual adapter and path counts from all the listheads
	 */
	total_num_of_adapters = 0;
	total_num_of_paths = 0;
	listhead = adapter_listhead_base.next;
	while (listhead != NULL) {
		total_num_of_adapters += listhead->adapter_count;
		total_num_of_paths += listhead->path_count;
		listhead = listhead->next_listhead;
	}

#ifdef	_MULTI_DATAMODEL
	if ((mode & DATAMODEL_MASK) == DATAMODEL_ILP32)
		/*
		 * Add extra 4-bytes to make sure connections header
		 * is double-word aligned
		 */
		pointer_area_size =
		    (total_num_of_adapters + total_num_of_adapters%2) *
		    sizeof (caddr32_t);
	else
		pointer_area_size = total_num_of_adapters * sizeof (caddr_t);
#else	/* _MULTI_DATAMODEL */
	mode = mode;
	pointer_area_size = total_num_of_adapters * sizeof (caddr_t);
#endif	/* _MULTI_DATAMODEL */


	topology_size = sizeof (rsmka_topology_hdr_t) +
	    pointer_area_size +
	    (total_num_of_adapters * sizeof (rsmka_connections_hdr_t)) +
	    (total_num_of_paths * sizeof (rsmka_remote_cntlr_t));

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "get_topology_size done\n"));

	return (topology_size);
}



static void
get_topology(caddr_t arg, char *bufp, int mode)
{

	rsmka_topology_t	*tp = (rsmka_topology_t *)bufp;
	adapter_listhead_t	*listhead;
	adapter_t		*adapter;
	path_t			*path;
	int			cntlr = 0;
	rsmka_connections_t	*connection;
	rsmka_remote_cntlr_t	*rem_cntlr;
	int			total_num_of_adapters;

#ifdef	_MULTI_DATAMODEL
	rsmka_topology32_t	*tp32 = (rsmka_topology32_t *)bufp;
#else
	mode = mode;
#endif	/* _MULTI_DATAMODEL */

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "get_topology enter\n"));

	/*
	 * Find the total number of adapters by adding up the
	 * individual adapter counts from all the listheads
	 */
	total_num_of_adapters = 0;
	listhead = adapter_listhead_base.next;
	while (listhead != NULL) {
		total_num_of_adapters += listhead->adapter_count;
		listhead = listhead->next_listhead;
	}

	/* fill topology header and adjust bufp */
	tp->topology_hdr.local_nodeid = my_nodeid;
	tp->topology_hdr.local_cntlr_count = total_num_of_adapters;
	bufp = (char *)&tp->connections[0];

	/* leave room for connection pointer area */
#ifdef	_MULTI_DATAMODEL
	if ((mode & DATAMODEL_MASK) == DATAMODEL_ILP32)
		/* make sure bufp is double-word aligned */
		bufp += (total_num_of_adapters + total_num_of_adapters%2) *
		    sizeof (caddr32_t);
	else
		bufp += total_num_of_adapters * sizeof (caddr_t);
#else	/* _MULTI_DATAMODEL */
	bufp += total_num_of_adapters * sizeof (caddr_t);
#endif	/* _MULTI_DATAMODEL */

	/* fill topology from the adapter and path data */
	listhead = adapter_listhead_base.next;
	while (listhead != NULL) {
		adapter = listhead->next_adapter;
		while (adapter != NULL) {
			/* fill in user based connection pointer */
#ifdef	_MULTI_DATAMODEL
			if ((mode & DATAMODEL_MASK) == DATAMODEL_ILP32) {
				ulong_t delta = (ulong_t)bufp - (ulong_t)tp32;
				caddr32_t userbase = (caddr32_t)((ulong_t)arg &
				    0xffffffff);
				tp32->connections[cntlr++] = userbase + delta;
			} else {
				tp->connections[cntlr++] = arg +
				    (ulong_t)bufp -
				    (ulong_t)tp;
			}
#else	/* _MULTI_DATAMODEL */
				tp->connections[cntlr++] = arg +
				    (ulong_t)bufp -
				    (ulong_t)tp;
#endif	/* _MULTI_DATAMODEL */
			connection = (rsmka_connections_t *)bufp;
			(void) snprintf(connection->hdr.cntlr_name,
			    MAXNAMELEN, "%s%d",
			    listhead->adapter_devname,
			    adapter->instance);
			connection->hdr.local_hwaddr = adapter->hwaddr;
			connection->hdr.remote_cntlr_count = 0;
			bufp += sizeof (rsmka_connections_hdr_t);
			rem_cntlr = (rsmka_remote_cntlr_t *)bufp;
			path = adapter->next_path;
			while (path != NULL) {
				connection->hdr.remote_cntlr_count++;
				rem_cntlr->remote_nodeid = path->remote_node;
				(void) snprintf(rem_cntlr->remote_cntlrname,
				    MAXNAMELEN, "%s%d",
				    listhead->adapter_devname,
				    path->remote_devinst);
				rem_cntlr->remote_hwaddr = path->remote_hwaddr;
				rem_cntlr->connection_state = path->state;
				++rem_cntlr;
				path = path->next_path;
			}
			adapter = adapter->next;
			bufp = (char *)rem_cntlr;
		}
		listhead = listhead->next_listhead;
	}

	DBG_PRINTF((category, RSM_DEBUG_VERBOSE, "get_topology done\n"));

}


/*
 * Called from rsm_ioctl() in rsm.c
 * Make sure there is no possiblity of blocking while holding
 * adapter_listhead_base.lock
 */
int
rsmka_topology_ioctl(caddr_t arg, int cmd, int mode)
{
	uint32_t	topology_size;
	uint32_t 	request_size;
	char		*bufp;
	int		error = RSM_SUCCESS;
	size_t		max_toposize;

	DBG_PRINTF((category | RSM_IOCTL, RSM_DEBUG_VERBOSE,
	    "rsmka_topology_ioctl enter\n"));

	switch (cmd) {
	case RSM_IOCTL_TOPOLOGY_SIZE:
		mutex_enter(&adapter_listhead_base.listlock);
		topology_size = get_topology_size(mode);
		mutex_exit(&adapter_listhead_base.listlock);
		if (ddi_copyout((caddr_t)&topology_size,
		    (caddr_t)arg, sizeof (uint32_t), mode))
			error = RSMERR_BAD_ADDR;
		break;
	case RSM_IOCTL_TOPOLOGY_DATA:
		/*
		 * The size of the buffer which the caller has allocated
		 * is passed in.  If the size needed for the topology data
		 * is not sufficient, E2BIG is returned
		 */
		if (ddi_copyin(arg, &request_size, sizeof (uint32_t), mode)) {
			DBG_PRINTF((category | RSM_IOCTL, RSM_DEBUG_VERBOSE,
			    "rsmka_topology_ioctl done: BAD_ADDR\n"));
			return (RSMERR_BAD_ADDR);
		}
		/* calculate the max size of the topology structure */
		max_toposize = sizeof (rsmka_topology_hdr_t) +
		    RSM_MAX_CTRL * (sizeof (caddr_t) +
		    sizeof (rsmka_connections_hdr_t)) +
		    RSM_MAX_NODE * sizeof (rsmka_remote_cntlr_t);

		if (request_size > max_toposize) { /* validate request_size */
			DBG_PRINTF((category | RSM_IOCTL, RSM_DEBUG_VERBOSE,
			    "rsmka_topology_ioctl done: size too large\n"));
			return (EINVAL);
		}
		bufp = kmem_zalloc(request_size, KM_SLEEP);
		mutex_enter(&adapter_listhead_base.listlock);
		topology_size = get_topology_size(mode);
		if (request_size < topology_size) {
			kmem_free(bufp, request_size);
			mutex_exit(&adapter_listhead_base.listlock);
			DBG_PRINTF((category | RSM_IOCTL, RSM_DEBUG_VERBOSE,
			    "rsmka_topology_ioctl done: E2BIG\n"));
			return (E2BIG);
		}

		/* get the topology data and copyout to the caller */
		get_topology(arg, bufp, mode);
		mutex_exit(&adapter_listhead_base.listlock);
		if (ddi_copyout((caddr_t)bufp, (caddr_t)arg,
		    topology_size, mode))
			error = RSMERR_BAD_ADDR;

		kmem_free(bufp, request_size);
		break;
	default:
		DBG_PRINTF((category | RSM_IOCTL, RSM_DEBUG,
		    "rsmka_topology_ioctl: cmd not supported\n"));
		error = DDI_FAILURE;
	}

	DBG_PRINTF((category | RSM_IOCTL, RSM_DEBUG_VERBOSE,
	    "rsmka_topology_ioctl done: %d\n", error));
	return (error);
}