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