Index: exec/cfg.c
===================================================================
--- exec/cfg.c	(revision 1498)
+++ exec/cfg.c	(revision 1499)
@@ -268,7 +268,7 @@
 		res_lib_cfg_ringreenable.header.id = MESSAGE_RES_CFG_RINGREENABLE;
 		res_lib_cfg_ringreenable.header.size = sizeof (struct res_lib_cfg_ringreenable);
 		res_lib_cfg_ringreenable.header.error = SA_AIS_OK;
-		openais_conn_send_response (
+		openais_response_send (
 			req_exec_cfg_ringreenable->source.conn,
 			&res_lib_cfg_ringreenable,
 			sizeof (struct res_lib_cfg_ringreenable));
@@ -312,7 +312,7 @@
 		strcpy ((char *)&res_lib_cfg_ringstatusget.interface_name[i],
 			totem_ip_string);
 	}
-	openais_conn_send_response (
+	openais_response_send (
 		conn,
 		&res_lib_cfg_ringstatusget,
 		sizeof (struct res_lib_cfg_ringstatusget));
Index: exec/evs.c
===================================================================
--- exec/evs.c	(revision 1498)
+++ exec/evs.c	(revision 1499)
@@ -244,7 +244,8 @@
 	 */
 	for (list = confchg_notify.next; list != &confchg_notify; list = list->next) {
 		evs_pd = list_entry (list, struct evs_pd, list);
-		openais_conn_send_response (evs_pd->conn,
+		openais_dispatch_send (
+			evs_pd->conn,
 			&res_evs_confchg_callback,
 			sizeof (res_evs_confchg_callback));
 	}
@@ -262,7 +263,9 @@
 	list_init (&evs_pd->list);
 	list_add (&evs_pd->list, &confchg_notify);
 
-	openais_conn_send_response (conn, &res_evs_confchg_callback,
+	openais_dispatch_send (
+		conn,
+		&res_evs_confchg_callback,
 		sizeof (res_evs_confchg_callback));
 
 	return (0);
@@ -308,7 +311,9 @@
 	res_lib_evs_join.header.id = MESSAGE_RES_EVS_JOIN;
 	res_lib_evs_join.header.error = error;
 
-	openais_conn_send_response (conn, &res_lib_evs_join,
+	openais_response_send (
+		conn,
+		&res_lib_evs_join,
 		sizeof (struct res_lib_evs_join));
 }
 
@@ -354,7 +359,9 @@
 	res_lib_evs_leave.header.id = MESSAGE_RES_EVS_LEAVE;
 	res_lib_evs_leave.header.error = error;
 
-	openais_conn_send_response (conn, &res_lib_evs_leave,
+	openais_response_send (
+		conn,
+		&res_lib_evs_leave,
 		sizeof (struct res_lib_evs_leave));
 }
 
@@ -397,7 +404,9 @@
 	res_lib_evs_mcast_joined.header.id = MESSAGE_RES_EVS_MCAST_JOINED;
 	res_lib_evs_mcast_joined.header.error = error;
 
-	openais_conn_send_response (conn, &res_lib_evs_mcast_joined,
+	openais_response_send (
+		conn,
+		&res_lib_evs_mcast_joined,
 		sizeof (struct res_lib_evs_mcast_joined));
 }
 
@@ -443,7 +452,9 @@
 	res_lib_evs_mcast_groups.header.id = MESSAGE_RES_EVS_MCAST_GROUPS;
 	res_lib_evs_mcast_groups.header.error = error;
 
-	openais_conn_send_response (conn, &res_lib_evs_mcast_groups,
+	openais_response_send (
+		conn,
+		&res_lib_evs_mcast_groups,
 		sizeof (struct res_lib_evs_mcast_groups));
 }
 
@@ -462,7 +473,9 @@
 	res_lib_evs_membership_get.member_list_entries =
 		res_evs_confchg_callback.member_list_entries;
 
-	openais_conn_send_response (conn, &res_lib_evs_membership_get,
+	openais_response_send (
+		conn,
+		&res_lib_evs_membership_get,
 		sizeof (struct res_lib_evs_membership_get));
 }
 
@@ -487,8 +500,10 @@
 	int i, j;
 	struct evs_pd *evs_pd;
 
-	res_evs_deliver_callback.header.size = sizeof (struct res_evs_deliver_callback) +
-		req_exec_evs_mcast->msg_len;
+	res_evs_deliver_callback.header.size =
+		sizeof (struct res_evs_deliver_callback) +
+			req_exec_evs_mcast->msg_len;
+
 	res_evs_deliver_callback.header.id = MESSAGE_RES_EVS_DELIVER_CALLBACK;
 	res_evs_deliver_callback.header.error = SA_AIS_OK;
 	res_evs_deliver_callback.msglen = req_exec_evs_mcast->msg_len;
@@ -517,9 +532,13 @@
 
 		if (found) {
 			res_evs_deliver_callback.local_nodeid = nodeid;
-			openais_conn_send_response (evs_pd->conn, &res_evs_deliver_callback,
+			openais_dispatch_send (
+				evs_pd->conn,
+				&res_evs_deliver_callback,
 				sizeof (struct res_evs_deliver_callback));
-			openais_conn_send_response (evs_pd->conn, msg_addr,
+			openais_dispatch_send (
+				evs_pd->conn,
+				msg_addr,
 				req_exec_evs_mcast->msg_len);
 		}
 	}
Index: exec/cpg.c
===================================================================
--- exec/cpg.c	(revision 1498)
+++ exec/cpg.c	(revision 1499)
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2006-2007 Red Hat, Inc.
+ * Copyright (c) 2006 Red Hat, Inc.
  * Copyright (c) 2006 Sun Microsystems, Inc.
  *
  * All rights reserved.
@@ -32,6 +32,9 @@
  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
  * THE POSSIBILITY OF SUCH DAMAGE.
  */
+#ifndef OPENAIS_BSD
+#include <alloca.h>
+#endif
 #include <sys/types.h>
 #include <sys/socket.h>
 #include <sys/un.h>
@@ -60,8 +63,11 @@
 #include "totempg.h"
 #include "totemip.h"
 #include "main.h"
+#include "flow.h"
+#include "tlist.h"
 #include "ipc.h"
 #include "mempool.h"
+#include "objdb.h"
 #include "service.h"
 #include "jhash.h"
 #include "swab.h"
@@ -255,7 +261,7 @@
 };
 
 struct openais_service_handler cpg_service_handler = {
-	.name				        = (unsigned char*)"openais cluster closed process group service v1.01",
+	.name				        = (unsigned char *)"openais cluster closed process group service v1.01",
 	.id					= CPG_SERVICE,
 	.private_data_size			= sizeof (struct process_info),
 	.flow_control				= OPENAIS_FLOW_CONTROL_REQUIRED,
@@ -434,14 +440,14 @@
 	}
 
 	if (conn) {
-		openais_conn_send_response(conn, buf, size);
+		openais_dispatch_send(conn, buf, size);
 	}
 	else {
 		/* Send it to all listeners */
 		for (iter = gi->members.next, tmp=iter->next; iter != &gi->members; iter = tmp, tmp=iter->next) {
 			struct process_info *pi = list_entry(iter, struct process_info, list);
 			if (pi->trackerconn && (pi->flags & PI_FLAG_MEMBER)) {
-				if (openais_conn_send_response(pi->trackerconn, buf, size) == -1) {
+				if (openais_dispatch_send(pi->trackerconn, buf, size) == -1) {
 					// Error ??
 				}
 			}
@@ -477,14 +483,17 @@
 	struct group_info *gi = pi->group;
 	mar_cpg_address_t notify_info;
 
-	log_printf(LOG_LEVEL_DEBUG, "exit_fn for conn=%p\n", conn);
-
 	if (gi) {
 		notify_info.pid = pi->pid;
 		notify_info.nodeid = totempg_my_nodeid_get();
 		notify_info.reason = CONFCHG_CPG_REASON_PROCDOWN;
 		cpg_node_joinleave_send(gi, pi, MESSAGE_REQ_EXEC_CPG_PROCLEAVE, CONFCHG_CPG_REASON_PROCDOWN);
 		list_del(&pi->list);
+		openais_ipc_flow_control_destroy (
+			conn,
+			CPG_SERVICE,
+			(unsigned char *)gi->group_name.value,
+			(unsigned int)gi->group_name.length);
 	}
 	return (0);
 }
@@ -531,7 +540,7 @@
 	req_exec_cpg_procjoin.header.size = sizeof(req_exec_cpg_procjoin);
 	req_exec_cpg_procjoin.header.id = SERVICE_ID_MAKE(CPG_SERVICE, fn);
 
-	req_exec_cpg_iovec.iov_base = &req_exec_cpg_procjoin;
+	req_exec_cpg_iovec.iov_base = (char *)&req_exec_cpg_procjoin;
 	req_exec_cpg_iovec.iov_len = sizeof(req_exec_cpg_procjoin);
 
 	result = totempg_groups_mcast_joined (openais_group_handle, &req_exec_cpg_iovec, 1, TOTEMPG_AGREED);
@@ -544,15 +553,17 @@
 	struct list_head *remlist)
 {
 	int i;
-	struct list_head *iter, *iter2, *tmp;
+	struct list_head *iter, *iter2;
 	struct process_info *pi;
 	struct group_info *gi;
 
 	for (i=0; i < GROUP_HASH_SIZE; i++) {
-		for (iter = group_lists[i].next; iter != &group_lists[i]; iter = iter->next) {
+		for (iter = group_lists[i].next; iter != &group_lists[i];) {
 			gi = list_entry(iter, struct group_info, list);
-			for (iter2 = gi->members.next, tmp = iter2->next; iter2 != &gi->members; iter2 = tmp, tmp = iter2->next) {
+			iter = iter->next;
+			for (iter2 = gi->members.next; iter2 != &gi->members;) {
 				pi = list_entry(iter2, struct process_info, list);
+				iter2 = iter2->next;
 
 				if (pi->nodeid == nodeid) {
 
@@ -643,7 +654,7 @@
 
 	/* Don't send this message until we get the final configuration message */
 	if (configuration_type == TOTEM_CONFIGURATION_REGULAR && req_exec_cpg_downlist.left_nodes) {
-		req_exec_cpg_iovec.iov_base = &req_exec_cpg_downlist;
+		req_exec_cpg_iovec.iov_base = (char *)&req_exec_cpg_downlist;
 		req_exec_cpg_iovec.iov_len = req_exec_cpg_downlist.header.size;
 
 		totempg_groups_mcast_joined (openais_group_handle, &req_exec_cpg_iovec, 1, TOTEMPG_AGREED);
@@ -816,7 +827,7 @@
 	struct req_exec_cpg_procjoin *req_exec_cpg_procjoin = (struct req_exec_cpg_procjoin *)message;
 	struct group_info *gi;
 	struct process_info *pi;
-	struct list_head *iter;
+	volatile struct list_head *iter;
 	mar_cpg_address_t notify_info;
 
 	log_printf(LOG_LEVEL_DEBUG, "got procleave message from cluster node %d\n", nodeid);
@@ -833,19 +844,28 @@
 			    1, &notify_info,
 			    MESSAGE_RES_CPG_CONFCHG_CALLBACK);
 
-        /* Find the node/PID to remove */
-	for (iter = gi->members.next; iter != &gi->members; iter = iter->next) {
+        /*
+	 * Find the node/PID to remove
+	 */
+	for (iter = gi->members.next; iter != &gi->members;) {
 		pi = list_entry(iter, struct process_info, list);
+
+		iter = iter->next;
+
 		if (pi->pid == req_exec_cpg_procjoin->pid &&
 		    pi->nodeid == nodeid) {
 
+			if (list_empty(&gi->members)) {
+				remove_group(gi);
+			}
+			
 			list_del(&pi->list);
-			if (!pi->conn)
+			if (pi->conn) {
+				openais_conn_info_refcnt_dec(pi->conn);
+			} else {
 				free(pi);
+			}
 
-			if (list_empty(&gi->members)) {
-				remove_group(gi);
-			}
 			break;
 		}
 	}
@@ -904,6 +924,7 @@
 		openais_ipc_flow_control_local_decrement (req_exec_cpg_mcast->source.conn);
 		process_info = (struct process_info *)openais_conn_private_data_get (req_exec_cpg_mcast->source.conn);
 		res_lib_cpg_mcast->flow_control_state = process_info->flow_control_state;
+		openais_conn_info_refcnt_dec (req_exec_cpg_mcast->source.conn);
 	}
 	memcpy(&res_lib_cpg_mcast->group_name, &gi->group_name,
 		sizeof(mar_cpg_name_t));
@@ -914,7 +935,7 @@
 	for (iter = gi->members.next; iter != &gi->members; iter = iter->next) {
 		struct process_info *pi = list_entry(iter, struct process_info, list);
 		if (pi->trackerconn) {
-			openais_conn_send_response(
+			openais_dispatch_send (
 				pi->trackerconn,
 				buf,
 				res_lib_cpg_mcast->header.size);
@@ -993,6 +1014,7 @@
 	struct process_info *pi = (struct process_info *)openais_conn_private_data_get (conn);
 	pi->conn = conn;
 
+	openais_conn_info_refcnt_inc (conn);
 	log_printf(LOG_LEVEL_DEBUG, "lib_init_fn: conn=%p, pi=%p\n", conn, pi);
 	return (0);
 }
@@ -1041,7 +1063,7 @@
 	res_lib_cpg_join.header.size = sizeof(res_lib_cpg_join);
 	res_lib_cpg_join.header.id = MESSAGE_RES_CPG_JOIN;
 	res_lib_cpg_join.header.error = error;
-	openais_conn_send_response(conn, &res_lib_cpg_join, sizeof(res_lib_cpg_join));
+	openais_response_send(conn, &res_lib_cpg_join, sizeof(res_lib_cpg_join));
 }
 
 /* Leave message from the library */
@@ -1052,8 +1074,6 @@
 	struct group_info *gi;
 	SaAisErrorT error = SA_AIS_OK;
 
-	log_printf(LOG_LEVEL_DEBUG, "got leave request on %p\n", conn);
-
 	if (!pi || !pi->pid || !pi->group) {
 		error = SA_AIS_ERR_INVALID_PARAM;
 		goto leave_ret;
@@ -1076,7 +1096,7 @@
 	res_lib_cpg_leave.header.size = sizeof(res_lib_cpg_leave);
 	res_lib_cpg_leave.header.id = MESSAGE_RES_CPG_LEAVE;
 	res_lib_cpg_leave.header.error = error;
-	openais_conn_send_response(conn, &res_lib_cpg_leave, sizeof(res_lib_cpg_leave));
+	openais_response_send(conn, &res_lib_cpg_leave, sizeof(res_lib_cpg_leave));
 }
 
 /* Mcast message from the library */
@@ -1099,7 +1119,7 @@
 		res_lib_cpg_mcast.header.id = MESSAGE_RES_CPG_MCAST;
 		res_lib_cpg_mcast.header.error = SA_AIS_ERR_ACCESS; /* TODO Better error code ?? */
 		res_lib_cpg_mcast.flow_control_state = CPG_FLOW_CONTROL_DISABLED;
-		openais_conn_send_response(conn, &res_lib_cpg_mcast,
+		openais_response_send(conn, &res_lib_cpg_mcast,
 			sizeof(res_lib_cpg_mcast));
 		return;
 	}
@@ -1107,15 +1127,16 @@
 	req_exec_cpg_mcast.header.size = sizeof(req_exec_cpg_mcast) + msglen;
 	req_exec_cpg_mcast.header.id = SERVICE_ID_MAKE(CPG_SERVICE,
 		MESSAGE_REQ_EXEC_CPG_MCAST);
+	openais_conn_info_refcnt_inc (conn);
 	req_exec_cpg_mcast.pid = pi->pid;
 	req_exec_cpg_mcast.msglen = msglen;
 	message_source_set (&req_exec_cpg_mcast.source, conn);
 	memcpy(&req_exec_cpg_mcast.group_name, &gi->group_name,
 		sizeof(mar_cpg_name_t));
 
-	req_exec_cpg_iovec[0].iov_base = &req_exec_cpg_mcast;
+	req_exec_cpg_iovec[0].iov_base = (char *)&req_exec_cpg_mcast;
 	req_exec_cpg_iovec[0].iov_len = sizeof(req_exec_cpg_mcast);
-	req_exec_cpg_iovec[1].iov_base = &req_lib_cpg_mcast->message;
+	req_exec_cpg_iovec[1].iov_base = (char *)&req_lib_cpg_mcast->message;
 	req_exec_cpg_iovec[1].iov_len = msglen;
 
 	// TODO: guarantee type...
@@ -1126,7 +1147,7 @@
 	res_lib_cpg_mcast.header.id = MESSAGE_RES_CPG_MCAST;
 	res_lib_cpg_mcast.header.error = SA_AIS_OK;
 	res_lib_cpg_mcast.flow_control_state = pi->flow_control_state;
-	openais_conn_send_response(conn, &res_lib_cpg_mcast,
+	openais_response_send(conn, &res_lib_cpg_mcast,
 		sizeof(res_lib_cpg_mcast));
 }
 
@@ -1140,7 +1161,7 @@
 		res.size = sizeof(res);
 		res.id = MESSAGE_RES_CPG_MEMBERSHIP;
 		res.error = SA_AIS_ERR_ACCESS; /* TODO Better error code */
-		openais_conn_send_response(conn, &res, sizeof(res));
+		openais_response_send(conn, &res, sizeof(res));
 		return;
 	}
 
@@ -1154,7 +1175,6 @@
 	struct res_lib_cpg_trackstart res_lib_cpg_trackstart;
 	struct group_info *gi;
 	struct process_info *otherpi;
-	void *otherconn;
 	SaAisErrorT error = SA_AIS_OK;
 
 	log_printf(LOG_LEVEL_DEBUG, "got trackstart request on %p\n", conn);
@@ -1166,7 +1186,6 @@
 	}
 
 	/* Find the partner connection and add us to it's process_info struct */
-	otherconn = openais_conn_partner_get (conn);
 	otherpi = (struct process_info *)openais_conn_private_data_get (conn);
 	otherpi->trackerconn = conn;
 
@@ -1174,7 +1193,7 @@
 	res_lib_cpg_trackstart.header.size = sizeof(res_lib_cpg_trackstart);
 	res_lib_cpg_trackstart.header.id = MESSAGE_RES_CPG_TRACKSTART;
 	res_lib_cpg_trackstart.header.error = SA_AIS_OK;
-	openais_conn_send_response(conn, &res_lib_cpg_trackstart, sizeof(res_lib_cpg_trackstart));
+	openais_response_send(conn, &res_lib_cpg_trackstart, sizeof(res_lib_cpg_trackstart));
 }
 
 static void message_handler_req_lib_cpg_trackstop (void *conn, void *message)
@@ -1182,7 +1201,6 @@
 	struct req_lib_cpg_trackstop *req_lib_cpg_trackstop = (struct req_lib_cpg_trackstop *)message;
 	struct res_lib_cpg_trackstop res_lib_cpg_trackstop;
 	struct process_info *otherpi;
-	void *otherconn;
 	struct group_info *gi;
 	SaAisErrorT error = SA_AIS_OK;
 
@@ -1195,7 +1213,6 @@
 	}
 
 	/* Find the partner connection and add us to it's process_info struct */
-	otherconn = openais_conn_partner_get (conn);
 	otherpi = (struct process_info *)openais_conn_private_data_get (conn);
 	otherpi->trackerconn = NULL;
 
@@ -1203,7 +1220,7 @@
 	res_lib_cpg_trackstop.header.size = sizeof(res_lib_cpg_trackstop);
 	res_lib_cpg_trackstop.header.id = MESSAGE_RES_CPG_TRACKSTOP;
 	res_lib_cpg_trackstop.header.error = SA_AIS_OK;
-	openais_conn_send_response(conn, &res_lib_cpg_trackstop.header, sizeof(res_lib_cpg_trackstop));
+	openais_response_send(conn, &res_lib_cpg_trackstop.header, sizeof(res_lib_cpg_trackstop));
 }
 
 static void message_handler_req_lib_cpg_local_get (void *conn, void *message)
@@ -1213,8 +1230,8 @@
 	res_lib_cpg_local_get.header.size = sizeof(res_lib_cpg_local_get);
 	res_lib_cpg_local_get.header.id = MESSAGE_RES_CPG_LOCAL_GET;
 	res_lib_cpg_local_get.header.error = SA_AIS_OK;
-	res_lib_cpg_local_get.local_nodeid = totempg_my_nodeid_get();
+	res_lib_cpg_local_get.local_nodeid = totempg_my_nodeid_get ();
 
-	openais_conn_send_response(conn, &res_lib_cpg_local_get,
+	openais_response_send(conn, &res_lib_cpg_local_get,
 		sizeof(res_lib_cpg_local_get));
 }
Index: exec/clm.c
===================================================================
--- exec/clm.c	(revision 1498)
+++ exec/clm.c	(revision 1499)
@@ -407,7 +407,7 @@
 		/*
 		 * Send notifications to all CLM listeners
 		 */
-		openais_conn_send_response (
+		openais_dispatch_send (
 			clm_pd->conn,
 			&res_lib_clm_clustertrack,
 			sizeof (struct res_lib_clm_clustertrack));
@@ -667,14 +667,16 @@
 		list_add (&clm_pd->list, &library_notification_send_listhead);
 	}
 
-	openais_conn_send_response (conn, &res_lib_clm_clustertrack,
+	openais_response_send (
+		conn,
+		&res_lib_clm_clustertrack,
 		sizeof (struct res_lib_clm_clustertrack));
 
 	if (req_lib_clm_clustertrack->return_in_callback) {
 		res_lib_clm_clustertrack.header.id = MESSAGE_RES_CLM_TRACKCALLBACK;
 
-		openais_conn_send_response (
-			openais_conn_partner_get (conn),
+		openais_dispatch_send (
+			conn,
 			&res_lib_clm_clustertrack,
 			sizeof (struct res_lib_clm_clustertrack));
 	}
@@ -699,7 +701,9 @@
 	list_del (&clm_pd->list);
 	list_init (&clm_pd->list);
 
-	openais_conn_send_response (conn, &res_lib_clm_trackstop,
+	openais_response_send (
+		conn,
+		&res_lib_clm_trackstop,
 		sizeof (struct res_lib_clm_trackstop));
 }
 
@@ -735,7 +739,11 @@
 	if (valid) {
 		memcpy (&res_clm_nodeget.cluster_node, cluster_node, sizeof (mar_clm_cluster_node_t));
 	}
-	openais_conn_send_response (conn, &res_clm_nodeget, sizeof (struct res_clm_nodeget));
+
+	openais_response_send (
+		conn,
+		&res_clm_nodeget,
+		sizeof (struct res_clm_nodeget));
 }
 
 static void message_handler_req_lib_clm_nodegetasync (void *conn, void *msg)
@@ -770,7 +778,9 @@
 	res_clm_nodegetasync.header.id = MESSAGE_RES_CLM_NODEGETASYNC;
 	res_clm_nodegetasync.header.error = SA_AIS_OK;
 
-	openais_conn_send_response (conn, &res_clm_nodegetasync,
+	openais_response_send (
+		conn,
+		&res_clm_nodegetasync,
 		sizeof (struct res_clm_nodegetasync));
 
 	/*
@@ -784,7 +794,8 @@
 		memcpy (&res_clm_nodegetcallback.cluster_node, cluster_node,
 			sizeof (mar_clm_cluster_node_t));
 	}
-	openais_conn_send_response (openais_conn_partner_get (conn),
+	openais_dispatch_send (
+		conn,
 		&res_clm_nodegetcallback,
 		sizeof (struct res_clm_nodegetcallback));
 }
Index: exec/ipc.c
===================================================================
--- exec/ipc.c	(revision 1498)
+++ exec/ipc.c	(revision 1499)
@@ -1,5 +1,4 @@
 /*
- * Copyright (c) 2002-2006 MontaVista Software, Inc.
  * Copyright (c) 2006-2007 Red Hat, Inc.
  *
  * All rights reserved.
@@ -54,7 +53,11 @@
 #include <signal.h>
 #include <sched.h>
 #include <time.h>
+#if defined(OPENAIS_SOLARIS) && defined(HAVE_GETPEERUCRED)
+#include <ucred.h>
+#endif
 
+#include "swab.h"
 #include "../include/saAis.h"
 #include "../include/list.h"
 #include "../include/queue.h"
@@ -66,6 +69,8 @@
 #include "mainconfig.h"
 #include "totemconfig.h"
 #include "main.h"
+#include "flow.h"
+#include "tlist.h"
 #include "ipc.h"
 #include "flow.h"
 #include "service.h"
@@ -79,6 +84,10 @@
 
 #include "util.h"
 
+#ifdef OPENAIS_SOLARIS
+#define MSG_NOSIGNAL 0
+#endif
+
 #define SERVER_BACKLOG 5
 
 /*
@@ -96,8 +105,10 @@
 
 static totempg_groups_handle ipc_handle;
 
-DECLARE_LIST_INIT (conn_info_list_head);
+static pthread_mutex_t conn_io_list_mutex = PTHREAD_MUTEX_INITIALIZER;
 
+DECLARE_LIST_INIT (conn_io_list_head);
+
 static void (*ipc_serialize_lock_fn) (void);
 
 static void (*ipc_serialize_unlock_fn) (void);
@@ -107,148 +118,282 @@
 	size_t mlen;
 };
 
-enum conn_state {
-	CONN_STATE_ACTIVE,
-	CONN_STATE_SECURITY,
-	CONN_STATE_REQUESTED,
-	CONN_STATE_CLOSED,
-	CONN_STATE_DISCONNECTED
+enum conn_io_state {
+	CONN_IO_STATE_INITIALIZING,
+	CONN_IO_STATE_AUTHENTICATED,
+	CONN_IO_STATE_INIT_FAILED
 };
 
+enum conn_info_state {
+	CONN_INFO_STATE_INITIALIZING,
+	CONN_INFO_STATE_ACTIVE,
+	CONN_INFO_STATE_DISCONNECT_REQUESTED,
+	CONN_INFO_STATE_DISCONNECTED
+};
+
+struct conn_info;
+
+struct conn_io {
+	int fd;				/* File descriptor  */
+	unsigned int events;		/* events polled for by file descriptor */
+	pthread_t thread;		/* thread identifier */
+	pthread_attr_t thread_attr;	/* thread attribute */
+	char *inb;			/* Input buffer for non-blocking reads */
+	int inb_nextheader;		/* Next message header starts here */
+	int inb_start;			/* Start location of input buffer */
+	int inb_inuse;			/* Bytes currently stored in input buffer */
+	struct queue outq;		/* Circular queue for outgoing requests */
+	int byte_start;			/* Byte to start sending from in head of queue */
+	unsigned int fcc;		/* flow control local count */
+	enum conn_io_state state;	/* state of this conn_io connection */
+	struct conn_info *conn_info;	/* connection information combining multiple conn_io structs */
+	unsigned int refcnt;		/* reference count for conn_io data structure */
+	pthread_mutex_t mutex;
+	unsigned int service;
+	struct list_head list;
+};
+
+
 struct conn_info {
-	int fd;			/* File descriptor  */
-	unsigned int events;	/* events polled for by file descriptor */
-	enum conn_state state;	/* State of this connection */
-	pthread_t thread;	/* thread identifier */
-	pthread_attr_t thread_attr;	/* thread attribute */
-	char *inb;		/* Input buffer for non-blocking reads */
-	int inb_nextheader;	/* Next message header starts here */
-	int inb_start;		/* Start location of input buffer */
-	int inb_inuse;		/* Bytes currently stored in input buffer */
-	struct queue outq;	/* Circular queue for outgoing requests */
-	int byte_start;		/* Byte to start sending from in head of queue */
-	enum service_types service;/* Type of service so dispatch knows how to route message */
-	int authenticated;	/* Is this connection authenticated? */
-	void *private_data;	/* library connection private data */
-	struct conn_info *conn_info_partner;	/* partner connection dispatch<->response */
+	enum conn_info_state state;			/* State of this connection */
+	enum service_types service;		/* Type of service so dispatch knows how to route message */
+	void *private_data;			/* library connection private data */
 	unsigned int flow_control_handle;	/* flow control identifier */
 	unsigned int flow_control_enabled;	/* flow control enabled bit */
-	unsigned int flow_control_local_count;	/* flow control local count */
 	enum openais_flow_control flow_control;	/* Does this service use IPC flow control */
 	pthread_mutex_t flow_control_mutex;
+	unsigned int flow_control_local_count;		/* flow control local count */
         int (*lib_exit_fn) (void *conn);
-	struct timerlist timerlist;
 	pthread_mutex_t mutex;
-	pthread_mutex_t *shared_mutex;
-	struct list_head list;
+	struct conn_io *conn_io_response;
+	struct conn_io *conn_io_dispatch;
+	unsigned int refcnt;
 };
 
-static void *prioritized_poll_thread (void *conn);
-static int conn_info_outq_flush (struct conn_info *conn_info);
-static void libais_deliver (struct conn_info *conn_info);
-static void ipc_flow_control (struct conn_info *conn_info);
+static void *prioritized_poll_thread (void *conn_io_in);
+static int conn_io_outq_flush (struct conn_io *conn_io);
+static void conn_io_deliver (struct conn_io *conn_io);
+//static void ipc_flow_control (struct conn_info *conn_info);
+static inline void conn_info_destroy (struct conn_info *conn_info);
+static void conn_io_destroy (struct conn_io *conn_io);
+static int conn_io_send (struct conn_io *conn_io, void *msg, int mlen); 
+static inline struct conn_info *conn_info_create (void);
+static int conn_io_found (struct conn_io *conn_io_to_match);
 
+static int response_init_send (struct conn_io *conn_io, void *message);
+static int dispatch_init_send (struct conn_io *conn_io, void *message);
+
  /*
   * IPC Initializers
   */
 
-static int response_init_send_response (
-	struct conn_info *conn_info,
-	void *message);
-static int dispatch_init_send_response (
-	struct conn_info *conn_info,
-	void *message);
+static int conn_io_refcnt_value (struct conn_io *conn_io)
+{
+	unsigned int refcnt;
 
-static int (*ais_init_service[]) (struct conn_info *conn_info, void *message) = {
-	response_init_send_response,
-	dispatch_init_send_response
+	pthread_mutex_lock (&conn_io->mutex);
+	refcnt = conn_io->refcnt;
+	pthread_mutex_unlock (&conn_io->mutex);
+
+	return (refcnt);
+}
+
+static void conn_io_refcnt_inc (struct conn_io *conn_io)
+{
+	pthread_mutex_lock (&conn_io->mutex);
+	conn_io->refcnt += 1;
+	pthread_mutex_unlock (&conn_io->mutex);
+}
+
+static int conn_io_refcnt_dec (struct conn_io *conn_io)
+{
+	unsigned int refcnt;
+
+	pthread_mutex_lock (&conn_io->mutex);
+	conn_io->refcnt -= 1;
+	refcnt = conn_io->refcnt;
+	pthread_mutex_unlock (&conn_io->mutex);
+
+	return (refcnt);
+}
+
+static void conn_info_refcnt_inc (struct conn_info *conn_info)
+{
+	/*
+	 * Connection not fully initialized yet
+	 */
+	if (conn_info == NULL) {
+		return;
+	}
+	pthread_mutex_lock (&conn_info->mutex);
+	conn_info->refcnt += 1;
+	pthread_mutex_unlock (&conn_info->mutex);
+}
+
+static void conn_info_refcnt_dec (struct conn_info *conn_info)
+{
+	int refcnt;
+
+	/*
+	 * Connection not fully initialized yet
+	 */
+	if (conn_info == NULL) {
+		return;
+	}
+	pthread_mutex_lock (&conn_info->mutex);
+	conn_info->refcnt -= 1;
+	refcnt = conn_info->refcnt;
+	assert (refcnt >= 0);
+	pthread_mutex_unlock (&conn_info->mutex);
+
+	if (refcnt == 0) {
+		conn_info_destroy (conn_info);
+	}
+}
+
+void openais_conn_info_refcnt_dec (void *conn)
+{
+	struct conn_info *conn_info = (struct conn_info *)conn;
+
+	conn_info_refcnt_dec (conn_info);
+}
+
+void openais_conn_info_refcnt_inc (void *conn)
+{
+	struct conn_info *conn_info = (struct conn_info *)conn;
+
+	conn_info_refcnt_inc (conn_info);
+}
+	
+static int (*ais_init_service[]) (struct conn_io *conn_io, void *message) = {
+	response_init_send,
+	dispatch_init_send
 };
 
-static void libais_disconnect_security (struct conn_info *conn_info)
+static void disconnect_request (struct conn_info *conn_info)
 {
-	conn_info->state = CONN_STATE_SECURITY;
-	close (conn_info->fd);
+unsigned int res;
+	/*
+	 * connection not fully active yet
+	 */
+	if (conn_info == NULL) {
+		return;
+	}
+	/*
+	 * We only want to decrement the reference count on these two
+	 * conn_io contexts one time
+	 */
+	if (conn_info->state != CONN_INFO_STATE_ACTIVE) {
+		return;
+	}
+	res = conn_io_refcnt_dec (conn_info->conn_io_response);
+	res = conn_io_refcnt_dec (conn_info->conn_io_dispatch);
+	conn_info->state = CONN_INFO_STATE_DISCONNECT_REQUESTED;
 }
 
-static int response_init_send_response (
-	struct conn_info *conn_info,
+static int response_init_send (
+	struct conn_io *conn_io,
 	void *message)
 {
 	SaAisErrorT error = SA_AIS_ERR_ACCESS;
-	uintptr_t cinfo = (uintptr_t)conn_info;
+	uintptr_t cinfo = (uintptr_t)conn_io;
 	mar_req_lib_response_init_t *req_lib_response_init = (mar_req_lib_response_init_t *)message;
 	mar_res_lib_response_init_t res_lib_response_init;
 
-	if (conn_info->authenticated) {
-		conn_info->service = req_lib_response_init->resdis_header.service;
+	if (conn_io->state == CONN_IO_STATE_AUTHENTICATED) {
 		error = SA_AIS_OK;
+		conn_io->service = req_lib_response_init->resdis_header.service;
 	}
 	res_lib_response_init.header.size = sizeof (mar_res_lib_response_init_t);
 	res_lib_response_init.header.id = MESSAGE_RES_INIT;
 	res_lib_response_init.header.error = error;
 	res_lib_response_init.conn_info = (mar_uint64_t)cinfo;
 
-	openais_conn_send_response (
-		conn_info,
+	conn_io_send (
+		conn_io,
 		&res_lib_response_init,
 		sizeof (res_lib_response_init));
 
 	if (error == SA_AIS_ERR_ACCESS) {
-		libais_disconnect_security (conn_info);
+		conn_io_destroy (conn_io);
 		return (-1);
 	}
+
 	return (0);
 }
 
-static int dispatch_init_send_response (
-	struct conn_info *conn_info,
+/*
+ * This is called iwth ipc_serialize_lock_fn() called
+ * Therefore there are no races with the destruction of the conn_io
+ * data structure
+ */
+static int dispatch_init_send (
+	struct conn_io *conn_io,
 	void *message)
 {
 	SaAisErrorT error = SA_AIS_ERR_ACCESS;
 	uintptr_t cinfo;
 	mar_req_lib_dispatch_init_t *req_lib_dispatch_init = (mar_req_lib_dispatch_init_t *)message;
 	mar_res_lib_dispatch_init_t res_lib_dispatch_init;
-	struct conn_info *msg_conn_info;
+	struct conn_io *msg_conn_io;
+	struct conn_info *conn_info;
+	unsigned int service;
 
-	if (conn_info->authenticated) {
-		conn_info->service = req_lib_dispatch_init->resdis_header.service;
-		if (!ais_service[req_lib_dispatch_init->resdis_header.service])
+	service = req_lib_dispatch_init->resdis_header.service;
+	cinfo = (uintptr_t)req_lib_dispatch_init->conn_info;
+	msg_conn_io = (struct conn_io *)cinfo;
+
+	/*
+	 * The response IPC connection has disconnected already for
+	 * some reason and is no longer referenceable in the system
+	 */
+	if (conn_io->state == CONN_IO_STATE_AUTHENTICATED) {
+		/*
+		 * If the response conn_io isn't found, it disconnected.
+		 * Hence, a full connection cannot be made and this connection
+		 * should be aborted by the poll thread
+		 */
+		if (conn_io_found (msg_conn_io) == 0) {
+			error = SA_AIS_ERR_TRY_AGAIN;
+			conn_io->state = CONN_IO_STATE_INIT_FAILED;
+		} else
+		/*
+		 * If no service is found for the requested library service,
+		 * the proper service handler isn't loaded and this connection
+		 * should be aborted by the poll thread
+		 */
+		if (ais_service[service] == NULL) {
 			error = SA_AIS_ERR_NOT_SUPPORTED;
-		else
+			conn_io->state = CONN_IO_STATE_INIT_FAILED;
+		} else {
 			error = SA_AIS_OK;
+		}
 
-		cinfo = (uintptr_t)req_lib_dispatch_init->conn_info;
-		conn_info->conn_info_partner = (struct conn_info *)cinfo;
-
-		/* temporary fix for memory leak
+		/*
+		 * The response and dispatch conn_io structures are available.
+		 * Attempt to allocate the appropriate memory for the private
+		 * data area
 		 */
-		pthread_mutex_destroy (conn_info->conn_info_partner->shared_mutex);
-		free (conn_info->conn_info_partner->shared_mutex);
-		
-		conn_info->conn_info_partner->shared_mutex = conn_info->shared_mutex;
-
-		list_add (&conn_info_list_head, &conn_info->list);
-		list_add (&conn_info_list_head, &conn_info->conn_info_partner->list);
-
-		msg_conn_info = (struct conn_info *)cinfo;
-		msg_conn_info->conn_info_partner = conn_info;
-
 		if (error == SA_AIS_OK) {
 			int private_data_size;
 
-			private_data_size = ais_service[req_lib_dispatch_init->resdis_header.service]->private_data_size;
+			conn_info = conn_info_create ();
+			private_data_size = ais_service[service]->private_data_size;
 			if (private_data_size) {
 				conn_info->private_data = malloc (private_data_size);
 
-				conn_info->conn_info_partner->private_data = conn_info->private_data;
+				/*
+				 * No private data could be allocated so
+				 * request the poll thread to abort
+				 */
 				if (conn_info->private_data == NULL) {
+					conn_io->state = CONN_IO_STATE_INIT_FAILED;
 					error = SA_AIS_ERR_NO_MEMORY;
 				} else {
 					memset (conn_info->private_data, 0, private_data_size);
 				}
 			} else {
 				conn_info->private_data = NULL;
-				conn_info->conn_info_partner->private_data = NULL;
 			}
 		}
 	}
@@ -257,338 +402,340 @@
 	res_lib_dispatch_init.header.id = MESSAGE_RES_INIT;
 	res_lib_dispatch_init.header.error = error;
 
-	openais_conn_send_response (
-		conn_info,
-		&res_lib_dispatch_init,
-		sizeof (res_lib_dispatch_init));
+	if (error != SA_AIS_OK) {
+		conn_io_send (
+			conn_io,
+			&res_lib_dispatch_init,
+			sizeof (res_lib_dispatch_init));
 
-	if (error == SA_AIS_ERR_ACCESS) {
-		libais_disconnect_security (conn_info);
 		return (-1);
 	}
-	if (error != SA_AIS_OK) {
-		return (-1);
-	}
 
-	conn_info->state = CONN_STATE_ACTIVE;
-	conn_info->conn_info_partner->state = CONN_STATE_ACTIVE;
-	conn_info->lib_exit_fn = ais_service[conn_info->service]->lib_exit_fn;
+	/*
+	 * connect both dispatch and response conn_ios into the conn_info
+	 * data structure
+	 */
+	conn_info->state = CONN_INFO_STATE_ACTIVE;
+	conn_info->lib_exit_fn = ais_service[service]->lib_exit_fn;
+	conn_info->conn_io_response = msg_conn_io;
+	conn_info->conn_io_response->conn_info = conn_info;
+	conn_info->conn_io_dispatch = conn_io;
+	conn_info->service = service;
+	conn_io->service = service;
+	conn_io->conn_info = conn_info;
 	ais_service[conn_info->service]->lib_init_fn (conn_info);
 
 	conn_info->flow_control = ais_service[conn_info->service]->flow_control;
-	conn_info->conn_info_partner->flow_control = ais_service[conn_info->service]->flow_control;
 	if (ais_service[conn_info->service]->flow_control == OPENAIS_FLOW_CONTROL_REQUIRED) {
 		openais_flow_control_ipc_init (
 			&conn_info->flow_control_handle,
 			conn_info->service);
 
 	}
+
+	/*
+	 * Tell the library the IPC connections are configured
+	 */
+	conn_io_send (
+		conn_io,
+		&res_lib_dispatch_init,
+		sizeof (res_lib_dispatch_init));
 	return (0);
 }
 
 /*
  * Create a connection data structure
  */
-static inline unsigned int conn_info_create (int fd) {
+static inline struct conn_info *conn_info_create (void)
+{
 	struct conn_info *conn_info;
-	int res;
 
 	conn_info = malloc (sizeof (struct conn_info));
 	if (conn_info == 0) {
-		return (ENOMEM);
+		return (NULL);
 	}
 
 	memset (conn_info, 0, sizeof (struct conn_info));
 
-	res = queue_init (&conn_info->outq, SIZEQUEUE,
+	conn_info->refcnt = 2;
+	pthread_mutex_init (&conn_info->mutex, NULL);
+	conn_info->state = CONN_INFO_STATE_INITIALIZING;
+
+	return (conn_info);
+}
+
+static inline void conn_info_destroy (struct conn_info *conn_info)
+{
+	if (conn_info->private_data) {
+		free (conn_info->private_data);
+	}
+	pthread_mutex_destroy (&conn_info->mutex);
+	free (conn_info);
+}
+
+static int conn_io_create (int fd)
+{
+	int res;
+	struct conn_io *conn_io;
+
+	conn_io = malloc (sizeof (struct conn_io));
+	if (conn_io == NULL) {
+		return (-1);
+	}
+	memset (conn_io, 0, sizeof (struct conn_io));
+
+	res = queue_init (&conn_io->outq, SIZEQUEUE,
 		sizeof (struct outq_item));
 	if (res != 0) {
-		free (conn_info);
-		return (ENOMEM);
+		return (-1);
 	}
-	conn_info->inb = malloc (sizeof (char) * SIZEINB);
-	if (conn_info->inb == NULL) {
-		queue_free (&conn_info->outq);
-		free (conn_info);
-		return (ENOMEM);
+
+	conn_io->inb = malloc (sizeof (char) * SIZEINB);
+	if (conn_io->inb == NULL) {
+		queue_free (&conn_io->outq);
+		return (-1);
 	}
-	conn_info->shared_mutex = malloc (sizeof (pthread_mutex_t));
-	if (conn_info->shared_mutex == NULL) {
-		free (conn_info->inb);
-		queue_free (&conn_info->outq);
-		free (conn_info);
-		return (ENOMEM);
-	}
 
-	pthread_mutex_init (&conn_info->mutex, NULL);
-	pthread_mutex_init (&conn_info->flow_control_mutex, NULL);
-	pthread_mutex_init (conn_info->shared_mutex, NULL);
+	conn_io->fd = fd;
+	conn_io->events = POLLIN|POLLNVAL;
+	conn_io->refcnt = 1;
+	conn_io->service = SOCKET_SERVICE_INIT;
+	conn_io->state = CONN_IO_STATE_INITIALIZING;
 
-	list_init (&conn_info->list);
-	conn_info->state = CONN_STATE_ACTIVE;
-	conn_info->fd = fd;
-	conn_info->events = POLLIN|POLLNVAL;
-	conn_info->service = SOCKET_SERVICE_INIT;
+	pthread_attr_init (&conn_io->thread_attr);
 
-	pthread_attr_init (&conn_info->thread_attr);
-/*
- * IA64 needs more stack space then other arches
- */
+	pthread_mutex_init (&conn_io->mutex, NULL);
+
+	/*
+	 * IA64 needs more stack space then other arches
+	 */
 #if defined(__ia64__)
-	pthread_attr_setstacksize (&conn_info->thread_attr, 400000);
+	pthread_attr_setstacksize (&conn_io->thread_attr, 400000);
 #else
-	pthread_attr_setstacksize (&conn_info->thread_attr, 200000);
+	pthread_attr_setstacksize (&conn_io->thread_attr, 200000);
 #endif
 
-	pthread_attr_setdetachstate (&conn_info->thread_attr, PTHREAD_CREATE_DETACHED);
-	res = pthread_create (&conn_info->thread, &conn_info->thread_attr,
-		prioritized_poll_thread, conn_info);
+	pthread_attr_setdetachstate (&conn_io->thread_attr, PTHREAD_CREATE_DETACHED);
+
+	res = pthread_create (&conn_io->thread, &conn_io->thread_attr,
+		prioritized_poll_thread, conn_io);
+
+	list_init (&conn_io->list);
+
+	pthread_mutex_lock (&conn_io_list_mutex);
+	list_add (&conn_io->list, &conn_io_list_head);
+	pthread_mutex_unlock (&conn_io_list_mutex);
 	return (res);
 }
 
-static void conn_info_destroy (struct conn_info *conn_info)
+static void conn_io_destroy (struct conn_io *conn_io)
 {
 	struct outq_item *outq_item;
 
 	/*
 	 * Free the outq queued items
 	 */
-	while (!queue_is_empty (&conn_info->outq)) {
-		outq_item = queue_item_get (&conn_info->outq);
+	while (!queue_is_empty (&conn_io->outq)) {
+		outq_item = queue_item_get (&conn_io->outq);
 		free (outq_item->msg);
-		queue_item_remove (&conn_info->outq);
+		queue_item_remove (&conn_io->outq);
 	}
 
-	queue_free (&conn_info->outq);
-	free (conn_info->inb);
-	if (conn_info->conn_info_partner) {
-		conn_info->conn_info_partner->conn_info_partner = NULL;
-	}
+	queue_free (&conn_io->outq);
+	free (conn_io->inb);
+	close (conn_io->fd);
+	pthread_mutex_lock (&conn_io_list_mutex);
+	list_del (&conn_io->list);
+	pthread_mutex_unlock (&conn_io_list_mutex);
 	
-	pthread_attr_destroy (&conn_info->thread_attr);
-	pthread_mutex_destroy (&conn_info->mutex);
-	pthread_mutex_destroy (&conn_info->flow_control_mutex);
-
-	list_del (&conn_info->list);
-	free (conn_info);
+	pthread_attr_destroy (&conn_io->thread_attr);
+	pthread_mutex_destroy (&conn_io->mutex);
+	free (conn_io);
 }
 
-static int libais_connection_active (struct conn_info *conn_info)
+static int conn_io_found (struct conn_io *conn_io_to_match)
 {
-	return (conn_info->state == CONN_STATE_ACTIVE);
-}
+	struct list_head *list;
+	struct conn_io *conn_io;
 
-static void libais_disconnect_request (struct conn_info *conn_info)
-{
-	if (conn_info->state == CONN_STATE_ACTIVE) {
-		conn_info->state = CONN_STATE_REQUESTED;
-		conn_info->conn_info_partner->state = CONN_STATE_REQUESTED;
+	for (list = conn_io_list_head.next; list != &conn_io_list_head;
+		list = list->next) {
+	
+		conn_io = list_entry (list, struct conn_io, list);
+		if (conn_io == conn_io_to_match) {
+			return (1);
+		}
 	}
-}
 
-static int libais_disconnect (struct conn_info *conn_info)
-{
-	int res = 0;
-
-	assert (conn_info->state != CONN_STATE_ACTIVE);
-
-	if (conn_info->state == CONN_STATE_DISCONNECTED) {
-		assert (0);
-	}
-
-	/*
-	 * Close active connections
-	 */
-	if (conn_info->state == CONN_STATE_ACTIVE || conn_info->state == CONN_STATE_REQUESTED) {
-		close (conn_info->fd);
-		conn_info->state = CONN_STATE_CLOSED;
-		close (conn_info->conn_info_partner->fd);
-		conn_info->conn_info_partner->state = CONN_STATE_CLOSED;
-	}
-
-	/*
-	 * Note we will only call the close operation once on the first time
-	 * one of the connections is closed
-	 */	
-	if (conn_info->state == CONN_STATE_CLOSED) {
-		if (conn_info->lib_exit_fn) {
-			res = conn_info->lib_exit_fn (conn_info);
-		}
-		if (res == -1) {
-			return (-1);
-		}
-		if (conn_info->conn_info_partner->lib_exit_fn) {
-			res = conn_info->conn_info_partner->lib_exit_fn (conn_info);
-		}
-		if (res == -1) {
-			return (-1);
-		}
-	}
-	conn_info->state = CONN_STATE_DISCONNECTED;
-	conn_info->conn_info_partner->state = CONN_STATE_DISCONNECTED;
-	if (conn_info->flow_control_enabled == 1) {
-		openais_flow_control_disable (conn_info->flow_control_handle);
-	}
 	return (0);
 }
 
-static inline void conn_info_mutex_lock (
-	struct conn_info *conn_info,
-	unsigned int service)
-{
-	if (service == SOCKET_SERVICE_INIT) {
-		pthread_mutex_lock (&conn_info->mutex);
-	} else {
-		pthread_mutex_lock (conn_info->shared_mutex);
-	}
-}
-static inline void conn_info_mutex_unlock (
-	struct conn_info *conn_info,
-	unsigned int service)
-{
-	if (service == SOCKET_SERVICE_INIT) {
-		pthread_mutex_unlock (&conn_info->mutex);
-	} else {
-		pthread_mutex_unlock (conn_info->shared_mutex);
-	}
-}
-
 /*
  * This thread runs in a specific thread priority mode to handle
- * I/O requests from the library
+ * I/O requests from or to the library
  */
-static void *prioritized_poll_thread (void *conn)
+static void *prioritized_poll_thread (void *conn_io_in)
 {
-	struct conn_info *conn_info = (struct conn_info *)conn;
+	struct conn_io *conn_io = (struct conn_io *)conn_io_in;
+	struct conn_info *conn_info = NULL;
 	struct pollfd ufd;
 	int fds;
 	struct sched_param sched_param;
 	int res;
-	pthread_mutex_t *rel_mutex;
-	unsigned int service;
-	struct conn_info *cinfo_partner;
-	void *private_data;
 
-#if defined(OPENAIS_BSD) || defined(OPENAIS_LINUX)
-	res = sched_get_priority_max (SCHED_RR);
-	if (res != -1) {
-		sched_param.sched_priority = res;
-		res = pthread_setschedparam (conn_info->thread, SCHED_RR, &sched_param);
-		if (res == -1) {
-			log_printf (LOG_LEVEL_WARNING, "Could not set SCHED_RR at priority %d: %s\n",
-				sched_param.sched_priority, strerror (errno));
-		}
-	} else
-		log_printf (LOG_LEVEL_WARNING, "Could not get maximum scheduler priority: %s\n", strerror (errno));
-#else
-	log_printf(LOG_LEVEL_WARNING, "Scheduler priority left to default value (no OS support)\n");
-#endif
+	sched_param.sched_priority = 1;
+//	res = pthread_setschedparam (conn_io->thread, SCHED_RR, &sched_param);
 
-	ufd.fd = conn_info->fd;
+	ufd.fd = conn_io->fd;
 	for (;;) {
 retry_poll:
-		service = conn_info->service;
-		ufd.events = conn_info->events;
+		conn_info = conn_io->conn_info;
+		conn_io_refcnt_inc (conn_io);
+		conn_info_refcnt_inc (conn_info);
+
+		ufd.events = conn_io->events;
 		ufd.revents = 0;
 		fds = poll (&ufd, 1, -1);
-		
-		conn_info_mutex_lock (conn_info, service);
-		
-		switch (conn_info->state) {
-		case CONN_STATE_SECURITY:
-			conn_info_mutex_unlock (conn_info, service);
-			pthread_mutex_destroy (conn_info->shared_mutex);
-			free (conn_info->shared_mutex);
-			conn_info_destroy (conn);
-			pthread_exit (0);
-			break;
-
-		case CONN_STATE_REQUESTED:
-		case CONN_STATE_CLOSED:
-			res = libais_disconnect (conn);
-			if (res != 0) {
-				conn_info_mutex_unlock (conn_info, service);
-				goto retry_poll;
-			}
-			break;
-
-		case CONN_STATE_DISCONNECTED:
-			rel_mutex = conn_info->shared_mutex;
-			private_data = conn_info->private_data;
-			cinfo_partner = conn_info->conn_info_partner;
-			conn_info_destroy (conn);
-			if (service == SOCKET_SERVICE_INIT) {
-				pthread_mutex_unlock (&conn_info->mutex);
-			} else {
-				pthread_mutex_unlock (rel_mutex);
-			}
-			if (cinfo_partner == NULL) {
-				pthread_mutex_destroy (rel_mutex);
-				free (rel_mutex);
-				free (private_data);
-			}
-			pthread_exit (0);
-			/*
-			 * !! NOTE !! this is the exit point for this thread
-			 */
-			break;
-
-		default:
-			break;
-		}
-
 		if (fds == -1) {
-			conn_info_mutex_unlock (conn_info, service);
+			conn_io_refcnt_dec (conn_io);
+			conn_info_refcnt_dec (conn_info);
 			goto retry_poll;
 		}
 
 		ipc_serialize_lock_fn ();
 
 		if (fds == 1 && ufd.revents) {
-			if (ufd.revents & (POLLERR|POLLHUP)) {
+			if ((ufd.revents & (POLLERR|POLLHUP)) ||
+				(conn_info &&
+					conn_info->state == CONN_INFO_STATE_DISCONNECT_REQUESTED)) {
+				disconnect_request (conn_info);
 
-				libais_disconnect_request (conn_info);
-
-				conn_info_mutex_unlock (conn_info, service);
+				conn_io_refcnt_dec (conn_io);
+				conn_info_refcnt_dec (conn_info);
 				ipc_serialize_unlock_fn ();
-				continue;
+				break; /* from for */
 			}
 			
 			if (ufd.revents & POLLOUT) {
-				conn_info_outq_flush (conn_info);
+				conn_io_outq_flush (conn_io);
 			}
 
 			if ((ufd.revents & POLLIN) == POLLIN) {
-				libais_deliver (conn_info);
+				conn_io_deliver (conn_io);
 			}
 
-			ipc_flow_control (conn_info);
+			/*
+			 * IPC initializiation failed because response fd
+			 * disconnected before it was linked to dispatch fd
+			 */
+			if (conn_io->state == CONN_IO_STATE_INIT_FAILED) {
+				conn_io_destroy (conn_io);
+				conn_info_refcnt_dec (conn_info);
+				ipc_serialize_unlock_fn ();
+				pthread_exit (0);
+			}
+			/*
+			 * IPC initializiation failed because response fd
+			 * disconnected before it was linked to dispatch fd
+			 */
+			if (conn_io->state == CONN_IO_STATE_INIT_FAILED) {
+				break;
+			}
 
+//			ipc_flow_control (conn_info);
+
 		}
 
 		ipc_serialize_unlock_fn ();
-		conn_info_mutex_unlock (conn_info, service);
+
+		conn_io_refcnt_dec (conn_io);
+		conn_info_refcnt_dec (conn_info);
 	}
 
+	ipc_serialize_lock_fn ();
+
 	/*
+	 * IPC initializiation failed because response fd
+	 * disconnected before it was linked to dispatch fd
+	 */
+	if (conn_io->conn_info == NULL || conn_io->state == CONN_IO_STATE_INIT_FAILED) {
+		conn_io_destroy (conn_io);
+		conn_info_refcnt_dec (conn_info);
+		ipc_serialize_unlock_fn ();
+		pthread_exit (0);
+	}
+
+	conn_info = conn_io->conn_info;
+
+	/*
+	 * This is the response conn_io
+	 */
+	if (conn_info->conn_io_response == conn_io) {
+		for (;;) {
+			if (conn_io_refcnt_value (conn_io) == 0) {
+				conn_io->conn_info = NULL;
+				conn_io_destroy (conn_io);
+				conn_info_refcnt_dec (conn_info);
+				ipc_serialize_unlock_fn ();
+				pthread_exit (0);
+			}
+		usleep (1000);
+	printf ("sleep 1\n");
+		}
+	} /* response conn_io */
+
+	/*
+	 * This is the dispatch conn_io
+	 */
+	if (conn_io->conn_info->conn_io_dispatch == conn_io) {
+		for (;;) {
+			if (conn_io_refcnt_value (conn_io) == 0) {
+				res = 0; // TODO
+				/*
+				 * Execute the library exit function
+				 */
+				if (conn_io->conn_info->lib_exit_fn) {
+					res = conn_io->conn_info->lib_exit_fn (conn_info);
+				}
+				if (res == 0) {
+					if (conn_io->conn_info->flow_control_enabled == 1) {
+//						openais_flow_control_disable (
+//							conn_info->flow_control_handle);
+					}
+					conn_io->conn_info = NULL;
+					conn_io_destroy (conn_io);
+					conn_info_refcnt_dec (conn_info);
+					ipc_serialize_unlock_fn ();
+					pthread_exit (0);
+				}
+			} /* refcnt == 0 */
+			usleep (1000);
+	printf ("sleep 2\n");
+		} /* for (;;) */
+	} /* dispatch conn_io */
+
+	/*
 	 * This code never reached
 	 */
 	return (0);
 }
 
-#if defined(OPENAIS_LINUX)
+#if defined(OPENAIS_LINUX) || defined(OPENAIS_SOLARIS)
 /* SUN_LEN is broken for abstract namespace
  */
 #define AIS_SUN_LEN(a) sizeof(*(a))
+#else
+#define AIS_SUN_LEN(a) SUN_LEN(a)
+#endif
 
+#if defined(OPENAIS_LINUX)
 char *socketname = "libais.socket";
 #else
-#define AIS_SUN_LEN(a) SUN_LEN(a)
-
 char *socketname = "/var/run/libais.socket";
 #endif
 
 
+#ifdef COMPILOE_OUT
 static void ipc_flow_control (struct conn_info *conn_info)
 {
 	unsigned int entries_used;
@@ -646,8 +793,9 @@
 		}
 	}
 }
+#endif
 
-static int conn_info_outq_flush (struct conn_info *conn_info) {
+static int conn_io_outq_flush (struct conn_io *conn_io) {
 	struct queue *outq;
 	int res = 0;
 	struct outq_item *queue_item;
@@ -655,46 +803,51 @@
 	struct iovec iov_send;
 	char *msg_addr;
 
-	if (!libais_connection_active (conn_info)) {
-		return (-1);
-	}
-	outq = &conn_info->outq;
+	outq = &conn_io->outq;
 
 	msg_send.msg_iov = &iov_send;
 	msg_send.msg_name = 0;
 	msg_send.msg_namelen = 0;
 	msg_send.msg_iovlen = 1;
+#ifndef OPENAIS_SOLARIS
 	msg_send.msg_control = 0;
 	msg_send.msg_controllen = 0;
 	msg_send.msg_flags = 0;
+#else
+	msg_send.msg_accrights = 0;
+	msg_send.msg_accrightslen = 0;
+#endif
 
+	pthread_mutex_lock (&conn_io->mutex);
 	while (!queue_is_empty (outq)) {
 		queue_item = queue_item_get (outq);
 		msg_addr = (char *)queue_item->msg;
-		msg_addr = &msg_addr[conn_info->byte_start];
+		msg_addr = &msg_addr[conn_io->byte_start];
 
 		iov_send.iov_base = msg_addr;
-		iov_send.iov_len = queue_item->mlen - conn_info->byte_start;
+		iov_send.iov_len = queue_item->mlen - conn_io->byte_start;
 
 retry_sendmsg:
-		res = sendmsg (conn_info->fd, &msg_send, MSG_NOSIGNAL);
+		res = sendmsg (conn_io->fd, &msg_send, MSG_NOSIGNAL);
 		if (res == -1 && errno == EINTR) {
 			goto retry_sendmsg;
 		}
 		if (res == -1 && errno == EAGAIN) {
+			pthread_mutex_unlock (&conn_io->mutex);
 			return (0);
 		}
 		if (res == -1 && errno == EPIPE) {
-			libais_disconnect_request (conn_info);
+			disconnect_request (conn_io->conn_info);
+			pthread_mutex_unlock (&conn_io->mutex);
 			return (0);
 		}
 		if (res == -1) {
-			printf ("ERRNO is %d\n", errno);
 			assert (0); /* some other unhandled error here */
 		}
-		if (res + conn_info->byte_start != queue_item->mlen) {
-			conn_info->byte_start += res;
+		if (res + conn_io->byte_start != queue_item->mlen) {
+			conn_io->byte_start += res;
 
+			pthread_mutex_unlock (&conn_io->mutex);
 			return (0);
 		}
 
@@ -702,14 +855,15 @@
 		 * Message sent, try sending another message
 		 */
 		queue_item_remove (outq);
-		conn_info->byte_start = 0;
+		conn_io->byte_start = 0;
 		free (queue_item->msg);
 	} /* while queue not empty */
 
 	if (queue_is_empty (outq)) {
-		conn_info->events = POLLIN|POLLNVAL;
+		conn_io->events = POLLIN|POLLNVAL;
 	}
 
+	pthread_mutex_unlock (&conn_io->mutex);
 	return (0);
 }
 
@@ -720,7 +874,7 @@
 	char buf[4096];
 };
 
-static void libais_deliver (struct conn_info *conn_info)
+static void conn_io_deliver (struct conn_io *conn_io)
 {
 	int res;
 	mar_req_header_t *header;
@@ -732,9 +886,6 @@
 	char cmsg_cred[CMSG_SPACE (sizeof (struct ucred))];
 	struct ucred *cred;
 	int on = 0;
-#else
-	uid_t euid;
-	gid_t egid;
 #endif
 	int send_ok = 0;
 	int send_ok_joined = 0;
@@ -745,9 +896,10 @@
 	msg_recv.msg_iovlen = 1;
 	msg_recv.msg_name = 0;
 	msg_recv.msg_namelen = 0;
+#ifndef OPENAIS_SOLARIS
 	msg_recv.msg_flags = 0;
 
-	if (conn_info->authenticated) {
+	if (conn_io->state == CONN_IO_STATE_AUTHENTICATED) {
 		msg_recv.msg_control = 0;
 		msg_recv.msg_controllen = 0;
 	} else {
@@ -756,24 +908,54 @@
 		msg_recv.msg_controllen = sizeof (cmsg_cred);
 #else
 		euid = -1; egid = -1;
-		if (getpeereid(conn_info->fd, &euid, &egid) != -1 &&
+		if (getpeereid(conn_io->fd, &euid, &egid) != -1 &&
 		    (euid == 0 || egid == g_gid_valid)) {
-				conn_info->authenticated = 1;
+				conn_io->state = CONN_IO_STATE_AUTHENTICATED;
 		}
-		if (conn_info->authenticated == 0) {
+		if (conn_io->state == CONN_IO_STATE_INITIALIZING) {
 			log_printf (LOG_LEVEL_SECURITY, "Connection not authenticated because gid is %d, expecting %d\n", egid, g_gid_valid);
 		}
 #endif
 	}
 
-	iov_recv.iov_base = &conn_info->inb[conn_info->inb_start];
-	iov_recv.iov_len = (SIZEINB) - conn_info->inb_start;
-	if (conn_info->inb_inuse == SIZEINB) {
+	#else /* OPENAIS_SOLARIS */
+	msg_recv.msg_accrights = 0;
+	msg_recv.msg_accrightslen = 0;
+
+
+	if (! conn_info->authenticated) {
+	#ifdef HAVE_GETPEERUCRED
+	ucred_t *uc;
+	uid_t euid = -1;
+	gid_t egid = -1;
+	if (getpeerucred(conn_info->fd, &uc) == 0) {
+	euid = ucred_geteuid(uc);
+	egid = ucred_getegid(uc);
+	if ((euid == 0) || (egid == g_gid_valid)) {
+		conn_info->authenticated = 1;
+	}
+		ucred_free(uc);
+	}
+	if (conn_info->authenticated == 0) {
+		log_printf (LOG_LEVEL_SECURITY, "Connection not authenticated because gid is %d, expecting %d\n", (int)egid, g_gid_valid);
+ 		}
+ #else
+ 		log_printf (LOG_LEVEL_SECURITY, "Connection not authenticated "
+ 			"because platform does not support "
+ 			"authentication with sockets, continuing "
+ 			"with a fake authentication\n");
+ 		conn_info->authenticated = 1;
+ #endif
+ 	}
+ #endif
+	iov_recv.iov_base = &conn_io->inb[conn_io->inb_start];
+	iov_recv.iov_len = (SIZEINB) - conn_io->inb_start;
+	if (conn_io->inb_inuse == SIZEINB) {
 		return;
 	}
 
 retry_recv:
-	res = recvmsg (conn_info->fd, &msg_recv, MSG_NOSIGNAL);
+	res = recvmsg (conn_io->fd, &msg_recv, MSG_NOSIGNAL);
 	if (res == -1 && errno == EINTR) {
 		goto retry_recv;
 	} else
@@ -794,17 +976,17 @@
 	 * Authenticate if this connection has not been authenticated
 	 */
 #ifdef OPENAIS_LINUX
-	if (conn_info->authenticated == 0) {
+	if (conn_io->state == CONN_IO_STATE_INITIALIZING) {
 		cmsg = CMSG_FIRSTHDR (&msg_recv);
 		assert (cmsg);
 		cred = (struct ucred *)CMSG_DATA (cmsg);
 		if (cred) {
 			if (cred->uid == 0 || cred->gid == g_gid_valid) {
-				setsockopt(conn_info->fd, SOL_SOCKET, SO_PASSCRED, &on, sizeof (on));
-				conn_info->authenticated = 1;
+				setsockopt(conn_io->fd, SOL_SOCKET, SO_PASSCRED, &on, sizeof (on));
+				conn_io->state = CONN_IO_STATE_AUTHENTICATED;
 			}
 		}
-		if (conn_info->authenticated == 0) {
+		if (conn_io->state == CONN_IO_STATE_INITIALIZING) {
 			log_printf (LOG_LEVEL_SECURITY, "Connection not authenticated because gid is %d, expecting %d\n", cred->gid, g_gid_valid);
 		}
 	}
@@ -813,23 +995,23 @@
 	 * Dispatch all messages received in recvmsg that can be dispatched
 	 * sizeof (mar_req_header_t) needed at minimum to do any processing
 	 */
-	conn_info->inb_inuse += res;
-	conn_info->inb_start += res;
+	conn_io->inb_inuse += res;
+	conn_io->inb_start += res;
 
-	while (conn_info->inb_inuse >= sizeof (mar_req_header_t) && res != -1) {
-		header = (mar_req_header_t *)&conn_info->inb[conn_info->inb_start - conn_info->inb_inuse];
+	while (conn_io->inb_inuse >= sizeof (mar_req_header_t) && res != -1) {
+		header = (mar_req_header_t *)&conn_io->inb[conn_io->inb_start - conn_io->inb_inuse];
 
-		if (header->size > conn_info->inb_inuse) {
+		if (header->size > conn_io->inb_inuse) {
 			break;
 		}
-		service = conn_info->service;
+		service = conn_io->service;
 
 		/*
 		 * If this service is in init phase, initialize service
 		 * else handle message using service service
 		 */
-		if (service == SOCKET_SERVICE_INIT) {
-			res = ais_init_service[header->id] (conn_info, header);
+		if (conn_io->service == SOCKET_SERVICE_INIT) {
+			res = ais_init_service[header->id] (conn_io, header);
 		} else  {
 			/*
 			 * Not an init service, but a standard service
@@ -846,7 +1028,7 @@
 			 * to queue a message, otherwise tell the library we are busy and to
 			 * try again later
 			 */
-			send_ok_joined_iovec.iov_base = header;
+			send_ok_joined_iovec.iov_base = (char *)header;
 			send_ok_joined_iovec.iov_len = header->size;
 			send_ok_joined = totempg_groups_send_ok_joined (openais_group_handle,
 				&send_ok_joined_iovec, 1);
@@ -859,7 +1041,7 @@
 				(sync_in_process() == 0)));
 
 			if (send_ok) {
-				ais_service[service]->lib_service[header->id].lib_handler_fn(conn_info, header);
+				ais_service[service]->lib_service[header->id].lib_handler_fn(conn_io->conn_info, header);
 			} else {
 
 				/*
@@ -870,33 +1052,33 @@
 				res_overlay.header.id =
 					ais_service[service]->lib_service[header->id].response_id;
 				res_overlay.header.error = SA_AIS_ERR_TRY_AGAIN;
-				openais_conn_send_response (
-					conn_info,
+				conn_io_send (
+					conn_io,
 					&res_overlay,
 					res_overlay.header.size);
 			}
 		}
-		conn_info->inb_inuse -= header->size;
+		conn_io->inb_inuse -= header->size;
 	} /* while */
 
-	if (conn_info->inb_inuse == 0) {
-		conn_info->inb_start = 0;
+	if (conn_io->inb_inuse == 0) {
+		conn_io->inb_start = 0;
 	} else
-// BUG	if (connections[conn_info->fd].inb_start + connections[conn_info->fd].inb_inuse >= SIZEINB) {
-	if (conn_info->inb_start >= SIZEINB) {
+// BUG	if (connections[conn_io->fd].inb_start + connections[conn_io->fd].inb_inuse >= SIZEINB) {
+	if (conn_io->inb_start >= SIZEINB) {
 		/*
 		 * If in buffer is full, move it back to start
 		 */
-		memmove (conn_info->inb,
-			&conn_info->inb[conn_info->inb_start - conn_info->inb_inuse],
-			sizeof (char) * conn_info->inb_inuse);
-		conn_info->inb_start = conn_info->inb_inuse;
+		memmove (conn_io->inb,
+			&conn_io->inb[conn_io->inb_start - conn_io->inb_inuse],
+			sizeof (char) * conn_io->inb_inuse);
+		conn_io->inb_start = conn_io->inb_inuse;
 	}
 
 	return;
 }
 
-static int poll_handler_libais_accept (
+static int poll_handler_accept (
 	poll_handle handle,
 	int fd,
 	int revent,
@@ -944,7 +1126,7 @@
 
 	log_printf (LOG_LEVEL_DEBUG, "connection received from libais client %d.\n", new_fd);
 
-	res = conn_info_create (new_fd);
+	res = conn_io_create (new_fd);
 	if (res != 0) {
 		close (new_fd);
 	}
@@ -994,8 +1176,6 @@
 	struct sockaddr_un un_addr;
 	int res;
 
-	log_init ("IPC");
-
 	ipc_serialize_lock_fn = serialize_lock_fn;
 
 	ipc_serialize_unlock_fn = serialize_unlock_fn;
@@ -1009,7 +1189,7 @@
 		openais_exit_error (AIS_DONE_LIBAIS_SOCKET);
 	};
 
-	totemip_nosigpipe(libais_server_fd);
+	totemip_nosigpipe (libais_server_fd);
 	res = fcntl (libais_server_fd, F_SETFL, O_NONBLOCK);
 	if (res == -1) {
 		log_printf (LOG_LEVEL_ERROR, "Could not set non-blocking operation on server socket: %s\n", strerror (errno));
@@ -1041,7 +1221,7 @@
          * Setup libais connection dispatch routine
          */
         poll_dispatch_add (aisexec_poll_handle, libais_server_fd,
-                POLLIN, 0, poll_handler_libais_accept);
+                POLLIN, 0, poll_handler_accept);
 
 	g_gid_valid = gid_valid;
 
@@ -1063,33 +1243,14 @@
 {
 	struct conn_info *conn_info = (struct conn_info *)conn;
 
-	if (conn != NULL) {
-		return ((void *)conn_info->private_data);
-	} else {
-		return NULL;
-	}
+	return (conn_info->private_data);
 }
 
-/*
- * Get the conn info partner connection
- */
-void *openais_conn_partner_get (void *conn)
-{
-	struct conn_info *conn_info = (struct conn_info *)conn;
-
-	if (conn != NULL) {
-		return ((void *)conn_info->conn_info_partner);
-	} else {
-		return NULL;
-	}
-}
-
-int openais_conn_send_response (
-	void *conn,
+static int conn_io_send (
+	struct conn_io *conn_io,
 	void *msg,
 	int mlen)
 {
-	struct queue *outq;
 	char *cmsg;
 	int res = 0;
 	int queue_empty;
@@ -1098,47 +1259,47 @@
 	struct msghdr msg_send;
 	struct iovec iov_send;
 	char *msg_addr;
-	struct conn_info *conn_info = (struct conn_info *)conn;
 
-	if (conn_info == NULL) {
-		return -1;
+	if (conn_io == NULL) {
+		assert (0);
 	}
 
-	if (!libais_connection_active (conn_info)) {
-		return (-1);
-	}
+//	ipc_flow_control (conn_info);
 
-	ipc_flow_control (conn_info);
-
-	outq = &conn_info->outq;
-
 	msg_send.msg_iov = &iov_send;
 	msg_send.msg_name = 0;
 	msg_send.msg_namelen = 0;
 	msg_send.msg_iovlen = 1;
+#ifndef OPENAIS_SOLARIS
 	msg_send.msg_control = 0;
 	msg_send.msg_controllen = 0;
 	msg_send.msg_flags = 0;
+#else
+	msg_send.msg_accrights = 0;
+	msg_send.msg_accrightslen = 0;
+#endif
 
-	if (queue_is_full (outq)) {
+	pthread_mutex_lock (&conn_io->mutex);
+	if (queue_is_full (&conn_io->outq)) {
 		/*
 		 * Start a disconnect if we have not already started one
 		 * and report that the outgoing queue is full
 		 */
 		log_printf (LOG_LEVEL_ERROR, "Library queue is full, disconnecting library connection.\n");
-		libais_disconnect_request (conn_info);
+		disconnect_request (conn_io->conn_info);
+		pthread_mutex_unlock (&conn_io->mutex);
 		return (-1);
 	}
-	while (!queue_is_empty (outq)) {
-		queue_item = queue_item_get (outq);
+	while (!queue_is_empty (&conn_io->outq)) {
+		queue_item = queue_item_get (&conn_io->outq);
 		msg_addr = (char *)queue_item->msg;
-		msg_addr = &msg_addr[conn_info->byte_start];
+		msg_addr = &msg_addr[conn_io->byte_start];
 
 		iov_send.iov_base = msg_addr;
-		iov_send.iov_len = queue_item->mlen - conn_info->byte_start;
+		iov_send.iov_len = queue_item->mlen - conn_io->byte_start;
 
 retry_sendmsg:
-		res = sendmsg (conn_info->fd, &msg_send, MSG_NOSIGNAL);
+		res = sendmsg (conn_io->fd, &msg_send, MSG_NOSIGNAL);
 		if (res == -1 && errno == EINTR) {
 			goto retry_sendmsg;
 		}
@@ -1146,29 +1307,30 @@
 			break; /* outgoing kernel queue full */
 		}
 		if (res == -1 && errno == EPIPE) {
-			libais_disconnect_request (conn_info);
+			disconnect_request (conn_io->conn_info);
+			pthread_mutex_unlock (&conn_io->mutex);
 			return (0);
 		}
 		if (res == -1) {
-			assert (0);
+//			assert (0);
 			break; /* some other error, stop trying to send message */
 		}
-		if (res + conn_info->byte_start != queue_item->mlen) {
-			conn_info->byte_start += res;
+		if (res + conn_io->byte_start != queue_item->mlen) {
+			conn_io->byte_start += res;
 			break;
 		}
 
 		/*
 		 * Message sent, try sending another message
 		 */
-		queue_item_remove (outq);
-		conn_info->byte_start = 0;
+		queue_item_remove (&conn_io->outq);
+		conn_io->byte_start = 0;
 		free (queue_item->msg);
 	} /* while queue not empty */
 
 	res = -1;
 
-	queue_empty = queue_is_empty (outq);
+	queue_empty = queue_is_empty (&conn_io->outq);
 	/*
 	 * Send request message
 	 */
@@ -1177,21 +1339,21 @@
 		iov_send.iov_base = msg;
 		iov_send.iov_len = mlen;
 retry_sendmsg_two:
-		res = sendmsg (conn_info->fd, &msg_send, MSG_NOSIGNAL);
+		res = sendmsg (conn_io->fd, &msg_send, MSG_NOSIGNAL);
 		if (res == -1 && errno == EINTR) {
 			goto retry_sendmsg_two;
 		}
 		if (res == -1 && errno == EAGAIN) {
-			conn_info->byte_start = 0;
-			conn_info->events = POLLIN|POLLNVAL;
+			conn_io->byte_start = 0;
+			conn_io->events = POLLIN|POLLNVAL;
 		}
 		if (res != -1) {
 			if (res != mlen) {
-				conn_info->byte_start += res;
+				conn_io->byte_start += res;
 				res = -1;
 			} else {
-				conn_info->byte_start = 0;
-				conn_info->events = POLLIN|POLLNVAL;
+				conn_io->byte_start = 0;
+				conn_io->events = POLLIN|POLLNVAL;
 			}
 		}
 	}
@@ -1203,21 +1365,23 @@
 		cmsg = malloc (mlen);
 		if (cmsg == 0) {
 			log_printf (LOG_LEVEL_ERROR, "Library queue couldn't allocate a message, disconnecting library connection.\n");
-			libais_disconnect_request (conn_info);
+			disconnect_request (conn_io->conn_info);
+			pthread_mutex_unlock (&conn_io->mutex);
 			return (-1);
 		}
 		queue_item_out.msg = cmsg;
 		queue_item_out.mlen = mlen;
 		memcpy (cmsg, msg, mlen);
-		queue_item_add (outq, &queue_item_out);
+		queue_item_add (&conn_io->outq, &queue_item_out);
 
 		/*
 		 * Send a pthread_kill to interrupt the poll syscall
 		 * and start a new poll operation in the thread
 		 */
-		conn_info->events = POLLIN|POLLOUT|POLLNVAL;
-		pthread_kill (conn_info->thread, SIGUSR1);
+		conn_io->events = POLLIN|POLLOUT|POLLNVAL;
+		pthread_kill (conn_io->thread, SIGUSR1);
 	}
+	pthread_mutex_unlock (&conn_io->mutex);
 	return (0);
 }
 
@@ -1238,7 +1402,6 @@
 		id_len,
 		flow_control_state_set_fn,
 		context);	
-	conn_info->conn_info_partner->flow_control_handle = conn_info->flow_control_handle;
 }
 
 void openais_ipc_flow_control_destroy (
@@ -1279,3 +1442,18 @@
 
 	pthread_mutex_unlock (&conn_info->flow_control_mutex);
 }
+
+
+int openais_response_send (void *conn, void *msg, int mlen)
+{
+	struct conn_info *conn_info = (struct conn_info *)conn;
+	
+	return (conn_io_send (conn_info->conn_io_response, msg, mlen));
+}
+
+int openais_dispatch_send (void *conn, void *msg, int mlen)
+{
+	struct conn_info *conn_info = (struct conn_info *)conn;
+	
+	return (conn_io_send (conn_info->conn_io_dispatch, msg, mlen));
+}
Index: exec/ipc.h
===================================================================
--- exec/ipc.h	(revision 1498)
+++ exec/ipc.h	(revision 1499)
@@ -42,12 +42,16 @@
 
 extern int message_source_is_local (mar_message_source_t *source);
 
-extern void *openais_conn_partner_get (void *conn);
-
 extern void *openais_conn_private_data_get (void *conn);
 
-extern int openais_conn_send_response (void *conn, void *msg, int mlen);
+extern int openais_response_send (void *conn, void *msg, int mlen);
 
+extern int openais_dispatch_send (void *conn, void *msg, int mlen);
+
+extern void openais_conn_info_refcnt_dec (void *conn);
+
+extern void openais_conn_info_refcnt_inc (void *conn);
+
 extern void openais_ipc_init (
         void (*serialize_lock_fn) (void),
         void (*serialize_unlock_fn) (void),
Index: exec/amfcomp.c
===================================================================
--- exec/amfcomp.c	(revision 1498)
+++ exec/amfcomp.c	(revision 1499)
@@ -547,8 +547,8 @@
 		AMF_RESPONSE_COMPONENTTERMINATECALLBACK,
 		component_terminate_callback_data);
 				        
-	openais_conn_send_response (
-		openais_conn_partner_get (comp->conn),
+	openais_dispatch_send (
+		comp->conn,
 		&res_lib,
 		sizeof (struct res_lib_amf_componentterminatecallback));
 
@@ -817,8 +817,8 @@
 
 	res_lib_amf_csiremovecallback.csiFlags = 0;
 				        
-	openais_conn_send_response (
-		openais_conn_partner_get (comp->conn),
+	openais_dispatch_send (
+		comp->conn,
 		&res_lib_amf_csiremovecallback,
 		sizeof (struct res_lib_amf_csiremovecallback));
 }
@@ -1011,8 +1011,8 @@
 
 	TRACE8 ("sending healthcheck request to component %s",
 			res_lib.compName.value);
-	openais_conn_send_response (
-		openais_conn_partner_get (healthcheck->comp->conn),
+	openais_dispatch_send (
+		healthcheck->comp->conn,
 		&res_lib, sizeof (struct res_lib_amf_healthcheckcallback));
 }
 
@@ -1117,8 +1117,7 @@
 	res_lib->invocation =
 		invocation_create (AMF_RESPONSE_CSISETCALLBACK, csi_assignment);
 
-	openais_conn_send_response (
-		openais_conn_partner_get (comp->conn), res_lib,	res_lib->header.size);
+	openais_dispatch_send (comp->conn, res_lib, res_lib->header.size);
 
 	free(p);
 }
@@ -1154,7 +1153,7 @@
 		res_lib.header.size = sizeof (struct res_lib_amf_componenterrorreport);
 		res_lib.header.id = MESSAGE_RES_AMF_COMPONENTERRORREPORT;
 		res_lib.header.error = SA_AIS_OK;
-		openais_conn_send_response (comp->conn, &res_lib, sizeof (res_lib));
+		openais_dispatch_send (comp->conn, &res_lib, sizeof (res_lib));
 	}
 
 	/* report to SU and let it handle the problem */
Index: exec/evt.c
===================================================================
--- exec/evt.c	(revision 1498)
+++ exec/evt.c	(revision 1499)
@@ -1853,8 +1853,10 @@
 		res.evd_head.size = sizeof(res);
 		res.evd_head.id = MESSAGE_RES_EVT_AVAILABLE;
 		res.evd_head.error = SA_AIS_OK;
-		openais_conn_send_response(openais_conn_partner_get(conn),
-				&res, sizeof(res));
+		openais_dispatch_send (
+				conn,
+				&res,
+				sizeof(res));
 	}
 
 }
@@ -2265,7 +2267,7 @@
 	res.ico_head.size = sizeof(res);
 	res.ico_head.id = MESSAGE_RES_EVT_OPEN_CHANNEL;
 	res.ico_head.error = error;
-	openais_conn_send_response(conn, &res, sizeof(res));
+	openais_response_send(conn, &res, sizeof(res));
 }
 
 /*
@@ -2322,7 +2324,7 @@
 	res.ico_head.size = sizeof(res);
 	res.ico_head.id = MESSAGE_RES_EVT_OPEN_CHANNEL;
 	res.ico_head.error = error;
-	openais_conn_send_response(conn, &res, sizeof(res));
+	openais_dispatch_send(conn, &res, sizeof(res));
 }
 
 
@@ -2415,7 +2417,7 @@
 	res.icc_head.size = sizeof(res);
 	res.icc_head.id = MESSAGE_RES_EVT_CLOSE_CHANNEL;
 	res.icc_head.error = ((ret == 0) ? SA_AIS_OK : SA_AIS_ERR_BAD_HANDLE);
-	openais_conn_send_response(conn, &res, sizeof(res));
+	openais_response_send(conn, &res, sizeof(res));
 }
 
 /*
@@ -2487,7 +2489,7 @@
 	res.iuc_head.size = sizeof(res);
 	res.iuc_head.id = MESSAGE_RES_EVT_UNLINK_CHANNEL;
 	res.iuc_head.error = error;
-	openais_conn_send_response(conn, &res, sizeof(res));
+	openais_response_send(conn, &res, sizeof(res));
 }
 
 /*
@@ -2591,7 +2593,7 @@
 	res.ics_head.size = sizeof(res);
 	res.ics_head.id = MESSAGE_RES_EVT_SUBSCRIBE;
 	res.ics_head.error = error;
-	openais_conn_send_response(conn, &res, sizeof(res));
+	openais_response_send(conn, &res, sizeof(res));
 
 	/*
 	 * See if an existing event with a retention time
@@ -2624,7 +2626,7 @@
 	res.ics_head.size = sizeof(res);
 	res.ics_head.id = MESSAGE_RES_EVT_SUBSCRIBE;
 	res.ics_head.error = error;
-	openais_conn_send_response(conn, &res, sizeof(res));
+	openais_response_send(conn, &res, sizeof(res));
 }
 
 /*
@@ -2691,7 +2693,7 @@
 	res.icu_head.size = sizeof(res);
 	res.icu_head.id = MESSAGE_RES_EVT_UNSUBSCRIBE;
 	res.icu_head.error = error;
-	openais_conn_send_response(conn, &res, sizeof(res));
+	openais_response_send(conn, &res, sizeof(res));
 }
 
 /*
@@ -2763,7 +2765,7 @@
 	res.iep_head.id = MESSAGE_RES_EVT_PUBLISH;
 	res.iep_head.error = error;
 	res.iep_event_id = event_id;
-	openais_conn_send_response(conn, &res, sizeof(res));
+	openais_response_send(conn, &res, sizeof(res));
 }
 
 /*
@@ -2827,7 +2829,7 @@
 	res.iec_head.size = sizeof(res);
 	res.iec_head.id = MESSAGE_RES_EVT_CLEAR_RETENTIONTIME;
 	res.iec_head.error = error;
-	openais_conn_send_response(conn, &res, sizeof(res));
+	openais_response_send(conn, &res, sizeof(res));
 
 }
 
@@ -2866,7 +2868,7 @@
 			edp->ed_event.led_head.id = MESSAGE_RES_EVT_EVENT_DATA;
 			edp->ed_event.led_head.error = SA_AIS_OK;
 			free(cel);
-			openais_conn_send_response(conn, &edp->ed_event,
+			openais_response_send(conn, &edp->ed_event,
 											edp->ed_event.led_head.size);
 			free_event_data(edp);
 			goto data_get_done;
@@ -2876,7 +2878,7 @@
 	res.led_head.size = sizeof(res.led_head);
 	res.led_head.id = MESSAGE_RES_EVT_EVENT_DATA;
 	res.led_head.error = SA_AIS_ERR_NOT_EXIST;
-	openais_conn_send_response(conn, &res, res.led_head.size);
+	openais_response_send(conn, &res, res.led_head.size);
 
 	/*
 	 * See if there are any events that the app doesn't know about
@@ -3038,7 +3040,7 @@
 	struct unlink_chan_pending *ucp;
 	struct retention_time_clear_pending *rtc;
 	struct libevt_pd *esip =
-		openais_conn_private_data_get(openais_conn_partner_get(conn));
+		openais_conn_private_data_get(conn);
 
 	log_printf(LOG_LEVEL_DEBUG, "saEvtFinalize (Event exit request)\n");
 	log_printf(LOG_LEVEL_DEBUG, "saEvtFinalize %d evts on list\n",
@@ -3431,7 +3433,7 @@
 	res.ico_head.id = MESSAGE_RES_EVT_OPEN_CHANNEL;
 	res.ico_head.error = SA_AIS_ERR_TIMEOUT;
 	ocp->ocp_invocation = OPEN_TIMED_OUT;
-	openais_conn_send_response(ocp->ocp_conn, &res, sizeof(res));
+	openais_response_send(ocp->ocp_conn, &res, sizeof(res));
 }
 
 /*
@@ -3521,15 +3523,14 @@
 		resa.ica_channel_handle = handle;
 		resa.ica_c_handle = ocp->ocp_c_handle;
 		resa.ica_invocation = ocp->ocp_invocation;
-		openais_conn_send_response(openais_conn_partner_get(ocp->ocp_conn),
-				&resa, sizeof(resa));
+		openais_dispatch_send(ocp->ocp_conn, &resa, sizeof(resa));
 	} else {
 		struct res_evt_channel_open res;
 		res.ico_head.size = sizeof(res);
 		res.ico_head.id = MESSAGE_RES_EVT_OPEN_CHANNEL;
 		res.ico_head.error = (ret == 0 ? SA_AIS_OK : SA_AIS_ERR_BAD_HANDLE);
 		res.ico_channel_handle = handle;
-		openais_conn_send_response(ocp->ocp_conn, &res, sizeof(res));
+		openais_response_send(ocp->ocp_conn, &res, sizeof(res));
 	}
 
 	if (timer_del_status == 0) {
@@ -3554,7 +3555,7 @@
 	res.iuc_head.size = sizeof(res);
 	res.iuc_head.id = MESSAGE_RES_EVT_UNLINK_CHANNEL;
 	res.iuc_head.error = SA_AIS_OK;
-	openais_conn_send_response(ucp->ucp_conn, &res, sizeof(res));
+	openais_response_send(ucp->ucp_conn, &res, sizeof(res));
 
 	free(ucp);
 }
@@ -3574,7 +3575,7 @@
 	res.iec_head.size = sizeof(res);
 	res.iec_head.id = MESSAGE_RES_EVT_CLEAR_RETENTIONTIME;
 	res.iec_head.error = ret;
-	openais_conn_send_response(rtc->rtc_conn, &res, sizeof(res));
+	openais_response_send(rtc->rtc_conn, &res, sizeof(res));
 
 	list_del(&rtc->rtc_entry);
 	free(rtc);
Index: exec/ckpt.c
===================================================================
--- exec/ckpt.c	(revision 1498)
+++ exec/ckpt.c	(revision 1499)
@@ -1401,12 +1401,12 @@
 				res_lib_ckpt_checkpointopenasync.ckpt_id = checkpoint->ckpt_id;
 			}
 
-			openais_conn_send_response (
+			openais_response_send (
 				req_exec_ckpt_checkpointopen->source.conn,
 				&res_lib_ckpt_checkpointopenasync,
 				sizeof (struct res_lib_ckpt_checkpointopenasync));
-			openais_conn_send_response (
-				openais_conn_partner_get (req_exec_ckpt_checkpointopen->source.conn),
+			openais_dispatch_send (
+				req_exec_ckpt_checkpointopen->source.conn,
 				&res_lib_ckpt_checkpointopenasync,
 				sizeof (struct res_lib_ckpt_checkpointopenasync));
 		} else {
@@ -1420,7 +1420,7 @@
 			}
 			res_lib_ckpt_checkpointopen.header.error = error;
 
-			openais_conn_send_response (
+			openais_response_send (
 				req_exec_ckpt_checkpointopen->source.conn,
 				&res_lib_ckpt_checkpointopen,
 				sizeof (struct res_lib_ckpt_checkpointopen));
@@ -1593,8 +1593,10 @@
 		res_lib_ckpt_checkpointclose.header.size = sizeof (struct res_lib_ckpt_checkpointclose);
 		res_lib_ckpt_checkpointclose.header.id = MESSAGE_RES_CKPT_CHECKPOINT_CHECKPOINTCLOSE;
 		res_lib_ckpt_checkpointclose.header.error = error;
-		openais_conn_send_response (req_exec_ckpt_checkpointclose->source.conn,
-			&res_lib_ckpt_checkpointclose, sizeof (struct res_lib_ckpt_checkpointclose));
+		openais_response_send (
+			req_exec_ckpt_checkpointclose->source.conn,
+			&res_lib_ckpt_checkpointclose,
+			sizeof (struct res_lib_ckpt_checkpointclose));
 	}
 
 	/*
@@ -1646,7 +1648,7 @@
 		res_lib_ckpt_checkpointunlink.header.size = sizeof (struct res_lib_ckpt_checkpointunlink);
 		res_lib_ckpt_checkpointunlink.header.id = MESSAGE_RES_CKPT_CHECKPOINT_CHECKPOINTUNLINK;
 		res_lib_ckpt_checkpointunlink.header.error = error;
-		openais_conn_send_response (
+		openais_response_send (
 			req_exec_ckpt_checkpointunlink->source.conn,
 			&res_lib_ckpt_checkpointunlink,
 			sizeof (struct res_lib_ckpt_checkpointunlink));
@@ -1694,7 +1696,7 @@
 		res_lib_ckpt_checkpointretentiondurationset.header.id = MESSAGE_RES_CKPT_CHECKPOINT_CHECKPOINTRETENTIONDURATIONSET;
 		res_lib_ckpt_checkpointretentiondurationset.header.error = error;
 
-		openais_conn_send_response (
+		openais_response_send (
 			req_exec_ckpt_checkpointretentiondurationset->source.conn,
 			&res_lib_ckpt_checkpointretentiondurationset,
 			sizeof (struct res_lib_ckpt_checkpointretentiondurationset));
@@ -1900,7 +1902,8 @@
 		res_lib_ckpt_sectioncreate.header.id = MESSAGE_RES_CKPT_CHECKPOINT_SECTIONCREATE;
 		res_lib_ckpt_sectioncreate.header.error = error;
 
-		openais_conn_send_response (req_exec_ckpt_sectioncreate->source.conn,
+		openais_response_send (
+			req_exec_ckpt_sectioncreate->source.conn,
 			&res_lib_ckpt_sectioncreate,
 			sizeof (struct res_lib_ckpt_sectioncreate));
 	}
@@ -1965,7 +1968,7 @@
 		res_lib_ckpt_sectiondelete.header.id = MESSAGE_RES_CKPT_CHECKPOINT_SECTIONDELETE;
 		res_lib_ckpt_sectiondelete.header.error = error;
 
-		openais_conn_send_response (
+		openais_response_send (
 			req_exec_ckpt_sectiondelete->source.conn,
 			&res_lib_ckpt_sectiondelete,
 			sizeof (struct res_lib_ckpt_sectiondelete));
@@ -2058,7 +2061,7 @@
 			 MESSAGE_RES_CKPT_CHECKPOINT_SECTIONEXPIRATIONTIMESET;
 		res_lib_ckpt_sectionexpirationtimeset.header.error = error;
 
-		openais_conn_send_response (
+		openais_response_send (
 			req_exec_ckpt_sectionexpirationtimeset->source.conn,
 			&res_lib_ckpt_sectionexpirationtimeset,
 			sizeof (struct res_lib_ckpt_sectionexpirationtimeset));
@@ -2168,7 +2171,7 @@
 			MESSAGE_RES_CKPT_CHECKPOINT_SECTIONWRITE;
 		res_lib_ckpt_sectionwrite.header.error = error;
 
-		openais_conn_send_response (
+		openais_response_send (
 			req_exec_ckpt_sectionwrite->source.conn,
 			&res_lib_ckpt_sectionwrite,
 			sizeof (struct res_lib_ckpt_sectionwrite));
@@ -2265,7 +2268,7 @@
 			MESSAGE_RES_CKPT_CHECKPOINT_SECTIONOVERWRITE;
 		res_lib_ckpt_sectionoverwrite.header.error = error;
 
-		openais_conn_send_response (
+		openais_response_send (
 			req_exec_ckpt_sectionoverwrite->source.conn,
 			&res_lib_ckpt_sectionoverwrite,
 			sizeof (struct res_lib_ckpt_sectionoverwrite));
@@ -2358,7 +2361,7 @@
 			res_lib_ckpt_sectionread.data_read = section_size;
 		}
 
-		openais_conn_send_response (
+		openais_response_send (
 			req_exec_ckpt_sectionread->source.conn,
 			&res_lib_ckpt_sectionread,
 			sizeof (struct res_lib_ckpt_sectionread));
@@ -2369,7 +2372,7 @@
 		if (error == SA_AIS_OK) {
 			char *sd;
 			sd = (char *)checkpoint_section->section_data;
-			openais_conn_send_response (
+			openais_response_send (
 				req_exec_ckpt_sectionread->source.conn,
 				&sd[req_exec_ckpt_sectionread->data_offset],
 				section_size);
@@ -2578,7 +2581,7 @@
 	res_lib_ckpt_activereplicaset.header.id = MESSAGE_RES_CKPT_ACTIVEREPLICASET;
 	res_lib_ckpt_activereplicaset.header.error = error;
 
-	openais_conn_send_response (
+	openais_response_send (
 		conn,
 		&res_lib_ckpt_activereplicaset,
 		sizeof (struct res_lib_ckpt_activereplicaset));
@@ -2641,7 +2644,7 @@
 		res_lib_ckpt_checkpointstatusget.header.id = MESSAGE_RES_CKPT_CHECKPOINT_CHECKPOINTSTATUSGET;
 		res_lib_ckpt_checkpointstatusget.header.error = SA_AIS_ERR_NOT_EXIST;
 	}
-	openais_conn_send_response (
+	openais_response_send (
 		conn,
 		&res_lib_ckpt_checkpointstatusget,
 		sizeof (struct res_lib_ckpt_checkpointstatusget));
@@ -2967,7 +2970,7 @@
 	res_lib_ckpt_checkpointsynchronize.header.size = sizeof (struct res_lib_ckpt_checkpointsynchronize);
 	res_lib_ckpt_checkpointsynchronize.header.id = MESSAGE_RES_CKPT_CHECKPOINT_CHECKPOINTSYNCHRONIZE;
 
-	openais_conn_send_response (
+	openais_response_send (
 		conn,
 		&res_lib_ckpt_checkpointsynchronize,
 		sizeof (struct res_lib_ckpt_checkpointsynchronize));
@@ -2998,13 +3001,13 @@
 	res_lib_ckpt_checkpointsynchronizeasync.header.id = MESSAGE_RES_CKPT_CHECKPOINT_CHECKPOINTSYNCHRONIZEASYNC;
 	res_lib_ckpt_checkpointsynchronizeasync.invocation = req_lib_ckpt_checkpointsynchronizeasync->invocation;
 
-	openais_conn_send_response (
+	openais_response_send (
 		conn,
 		&res_lib_ckpt_checkpointsynchronizeasync,
 		sizeof (struct res_lib_ckpt_checkpointsynchronizeasync));
 
-	openais_conn_send_response (
-		openais_conn_partner_get (conn),
+	openais_dispatch_send (
+		conn,
 		&res_lib_ckpt_checkpointsynchronizeasync,
 		sizeof (struct res_lib_ckpt_checkpointsynchronizeasync));
 }
@@ -3133,7 +3136,7 @@
 	res_lib_ckpt_sectioniterationinitialize.max_section_id_size =
 		checkpoint->checkpoint_creation_attributes.max_section_id_size;
 
-	openais_conn_send_response (
+	openais_response_send (
 		conn,
 		&res_lib_ckpt_sectioniterationinitialize,
 		sizeof (struct res_lib_ckpt_sectioniterationinitialize));
@@ -3174,7 +3177,7 @@
 	res_lib_ckpt_sectioniterationfinalize.header.id = MESSAGE_RES_CKPT_SECTIONITERATIONFINALIZE;
 	res_lib_ckpt_sectioniterationfinalize.header.error = error;
 
-	openais_conn_send_response (
+	openais_response_send (
 		conn,
 		&res_lib_ckpt_sectioniterationfinalize,
 		sizeof (struct res_lib_ckpt_sectioniterationfinalize));
@@ -3262,13 +3265,13 @@
 	res_lib_ckpt_sectioniterationnext.header.id = MESSAGE_RES_CKPT_SECTIONITERATIONNEXT;
 	res_lib_ckpt_sectioniterationnext.header.error = error;
 
-	openais_conn_send_response (
+	openais_response_send (
 		conn,
 		&res_lib_ckpt_sectioniterationnext,
 		sizeof (struct res_lib_ckpt_sectioniterationnext));
 
 	if (error == SA_AIS_OK) {
-		openais_conn_send_response (
+		openais_response_send (
 			conn,
 			checkpoint_section->section_descriptor.section_id.id,
 			checkpoint_section->section_descriptor.section_id.id_len);
Index: exec/amf.c
===================================================================
--- exec/amf.c	(revision 1498)
+++ exec/amf.c	(revision 1499)
@@ -480,7 +480,7 @@
 		res_lib.header.id = MESSAGE_RES_AMF_COMPONENTREGISTER;
 		res_lib.header.size = sizeof (struct res_lib_amf_componentregister);
 		res_lib.header.error = error;
-		openais_conn_send_response (
+		openais_response_send (
 			comp->conn, &res_lib, sizeof (struct res_lib_amf_componentregister));
 	}
 }
@@ -544,7 +544,7 @@
 		res_lib.header.id = MESSAGE_RES_AMF_RESPONSE;
 		res_lib.header.size = sizeof (struct res_lib_amf_response);
 		res_lib.header.error = retval;
-		openais_conn_send_response (comp->conn, &res_lib, sizeof (res_lib));
+		openais_response_send (comp->conn, &res_lib, sizeof (res_lib));
 	}
 }
 
@@ -583,7 +583,7 @@
 		res_lib.header.id = MESSAGE_RES_AMF_COMPONENTREGISTER;
 		res_lib.header.size = sizeof (struct res_lib_amf_componentregister);
 		res_lib.header.error = SA_AIS_ERR_INVALID_PARAM;
-		openais_conn_send_response (
+		openais_response_send (
 			conn, &res_lib, sizeof (struct res_lib_amf_componentregister));
 	}
 }
@@ -658,12 +658,13 @@
 	res_lib.header.id = MESSAGE_RES_AMF_HEALTHCHECKSTART;
 	res_lib.header.size = sizeof (res_lib);
 	res_lib.header.error = error;
-	openais_conn_send_response (conn, &res_lib,
+	openais_response_send (conn, &res_lib,
 		sizeof (struct res_lib_amf_healthcheckstart));
 }
 
 static void message_handler_req_lib_amf_healthcheckconfirm (
-	void *conn,	void *msg)
+	void *conn,
+	void *msg)
 {
 	struct req_lib_amf_healthcheckconfirm *req_lib = msg;
 	struct res_lib_amf_healthcheckconfirm res_lib;
@@ -683,7 +684,7 @@
 	res_lib.header.id = MESSAGE_RES_AMF_HEALTHCHECKCONFIRM;
 	res_lib.header.size = sizeof (res_lib);
 	res_lib.header.error = error;
-	openais_conn_send_response (conn, &res_lib, sizeof (res_lib));
+	openais_response_send (conn, &res_lib, sizeof (res_lib));
 }
 
 static void message_handler_req_lib_amf_healthcheckstop (
@@ -706,7 +707,7 @@
 	res_lib.header.id = MESSAGE_RES_AMF_HEALTHCHECKSTOP;
 	res_lib.header.size = sizeof (res_lib);
 	res_lib.header.error = error;
-	openais_conn_send_response (conn, &res_lib, sizeof (res_lib));
+	openais_response_send (conn, &res_lib, sizeof (res_lib));
 }
 
 static void message_handler_req_lib_amf_hastateget (void *conn, void *msg)
@@ -732,7 +733,7 @@
 	res_lib.header.size = sizeof (struct res_lib_amf_hastateget);
 	res_lib.header.error = error;
 
-	openais_conn_send_response (conn, &res_lib,
+	openais_response_send (conn, &res_lib,
 		sizeof (struct res_lib_amf_hastateget));
 }
 
@@ -788,7 +789,7 @@
 	if (amfProtectionGroup) {
 		res_lib_amf_protectiongrouptrack.header.error = SA_AIS_OK;
 	}
-	openais_conn_send_response (conn, &res_lib_amf_protectiongrouptrack,
+	openais_response_send (conn, &res_lib_amf_protectiongrouptrack,
 		sizeof (struct res_lib_amf_protectiongrouptrack));
 
 	if (amfProtectionGroup &&
@@ -848,7 +849,7 @@
 	if (track) {
 		res_lib_amf_protectiongrouptrackstop.header.error = SA_AIS_OK;
 	}
-	openais_conn_send_response (conn, &res_lib_amf_protectiongrouptrackstop,
+	openais_response_send (conn, &res_lib_amf_protectiongrouptrackstop,
 		sizeof (struct res_lib_amf_protectiongrouptrackstop));
 
 #endif
@@ -896,7 +897,7 @@
 		res_lib.header.size = sizeof (struct res_lib_amf_componenterrorreport);
 		res_lib.header.id = MESSAGE_RES_AMF_COMPONENTERRORREPORT;
 		res_lib.header.error = SA_AIS_ERR_NOT_EXIST;
-		openais_conn_send_response (conn, &res_lib,
+		openais_response_send (conn, &res_lib,
 			sizeof (struct res_lib_amf_componenterrorreport));
 	}
 }
@@ -963,7 +964,7 @@
 		res_lib.header.id = MESSAGE_RES_AMF_RESPONSE;
 		res_lib.header.size = sizeof (struct res_lib_amf_response);
 		res_lib.header.error = retval;
-		openais_conn_send_response (conn, &res_lib, sizeof (res_lib));
+		openais_response_send (conn, &res_lib, sizeof (res_lib));
 	}
 }
 
Index: exec/lck.c
===================================================================
--- exec/lck.c	(revision 1498)
+++ exec/lck.c	(revision 1499)
@@ -719,12 +719,12 @@
 				&req_exec_lck_resourceopen->source,
 				sizeof (mar_message_source_t));
 
-			openais_conn_send_response (
+			openais_response_send (
 				req_exec_lck_resourceopen->source.conn,
 				&res_lib_lck_resourceopenasync,
 				sizeof (struct res_lib_lck_resourceopenasync));
-			openais_conn_send_response (
-				openais_conn_partner_get (req_exec_lck_resourceopen->source.conn),
+			openais_dispatch_send (
+				req_exec_lck_resourceopen->source.conn,
 				&res_lib_lck_resourceopenasync,
 				sizeof (struct res_lib_lck_resourceopenasync));
 		} else {
@@ -738,7 +738,7 @@
 				&req_exec_lck_resourceopen->source,
 				sizeof (mar_message_source_t));
 
-			openais_conn_send_response (req_exec_lck_resourceopen->source.conn,
+			openais_response_send (req_exec_lck_resourceopen->source.conn,
 				&res_lib_lck_resourceopen,
 				sizeof (struct res_lib_lck_resourceopen));
 		}
@@ -774,7 +774,7 @@
 		res_lib_lck_resourceclose.header.size = sizeof (struct res_lib_lck_resourceclose);
 		res_lib_lck_resourceclose.header.id = MESSAGE_RES_LCK_RESOURCECLOSE;
 		res_lib_lck_resourceclose.header.error = error;
-		openais_conn_send_response (
+		openais_response_send (
 			req_exec_lck_resourceclose->source.conn,
 			&res_lib_lck_resourceclose, sizeof (struct res_lib_lck_resourceclose));
 	}
@@ -801,8 +801,8 @@
 		res_lib_lck_lockwaitercallback.mode_held = SA_LCK_PR_LOCK_MODE;
 	}
 
-	openais_conn_send_response (
-		openais_conn_partner_get (resource_lock->callback_source.conn),
+	openais_dispatch_send (
+		resource_lock->callback_source.conn,
 		&res_lib_lck_lockwaitercallback,
 		sizeof (struct res_lib_lck_lockwaitercallback));
 }
@@ -837,8 +837,8 @@
 			res_lib_lck_resourcelockasync.lockStatus = resource_lock->lock_status;
 			res_lib_lck_resourcelockasync.invocation = resource_lock->invocation;
 			res_lib_lck_resourcelockasync.lockId = resource_lock->lock_id;
-			openais_conn_send_response (
-				openais_conn_partner_get (source->conn),
+			openais_dispatch_send (
+				source->conn,
 				&res_lib_lck_resourcelockasync,
 				sizeof (struct res_lib_lck_resourcelockasync));
 		}
@@ -861,7 +861,7 @@
 			res_lib_lck_resourcelock.header.error = error;
 			res_lib_lck_resourcelock.resource_lock = (void *)resource_lock;
 			res_lib_lck_resourcelock.lockStatus = resource_lock->lock_status;
-			openais_conn_send_response (source->conn,
+			openais_response_send (source->conn,
 				&res_lib_lck_resourcelock,
 				sizeof (struct res_lib_lck_resourcelock));
 		}
@@ -1133,14 +1133,11 @@
 		 * Deliver async response to library
 		 */
 		req_exec_lck_resourcelock->source.conn =
-			openais_conn_partner_get (req_exec_lck_resourcelock->source.conn);
+			req_exec_lck_resourcelock->source.conn;
 		resource_lock_async_deliver (
 			&req_exec_lck_resourcelock->source,
 			resource_lock,
 			SA_AIS_OK);
-// TODO why is this twice ?
-		req_exec_lck_resourcelock->source.conn =
-			openais_conn_partner_get (req_exec_lck_resourcelock->source.conn);
 	}
 
 error_exit:
@@ -1184,11 +1181,11 @@
 			res_lib_lck_resourceunlockasync.invocation =
 				req_exec_lck_resourceunlock->invocation;
 
-			openais_conn_send_response (
-				openais_conn_partner_get(req_exec_lck_resourceunlock->source.conn),
+			openais_dispatch_send (
+				req_exec_lck_resourceunlock->source.conn,
 				&res_lib_lck_resourceunlockasync,
 				sizeof (struct res_lib_lck_resourceunlockasync));
-			openais_conn_send_response (
+			openais_response_send (
 				resource_lock->callback_source.conn,
 				&res_lib_lck_resourceunlockasync,
 				sizeof (struct res_lib_lck_resourceunlockasync));
@@ -1196,8 +1193,10 @@
 			res_lib_lck_resourceunlock.header.size = sizeof (struct res_lib_lck_resourceunlock);
 			res_lib_lck_resourceunlock.header.id = MESSAGE_RES_LCK_RESOURCEUNLOCK;
 			res_lib_lck_resourceunlock.header.error = error;
-			openais_conn_send_response (req_exec_lck_resourceunlock->source.conn,
-				&res_lib_lck_resourceunlock, sizeof (struct res_lib_lck_resourceunlock));
+			openais_response_send (
+				req_exec_lck_resourceunlock->source.conn,
+				&res_lib_lck_resourceunlock,
+				sizeof (struct res_lib_lck_resourceunlock));
 		}
 	}
 }
@@ -1253,8 +1252,10 @@
 		res_lib_lck_lockpurge.header.size = sizeof (struct res_lib_lck_lockpurge);
 		res_lib_lck_lockpurge.header.id = MESSAGE_RES_LCK_LOCKPURGE;
 		res_lib_lck_lockpurge.header.error = error;
-		openais_conn_send_response (req_exec_lck_lockpurge->source.conn,
-			&res_lib_lck_lockpurge, sizeof (struct res_lib_lck_lockpurge));
+		openais_response_send (
+			req_exec_lck_lockpurge->source.conn,
+			&res_lib_lck_lockpurge,
+			sizeof (struct res_lib_lck_lockpurge));
 	}
 }
 
@@ -1366,7 +1367,8 @@
 		res_lib_lck_resourceclose.header.id = MESSAGE_RES_LCK_RESOURCECLOSE;
 		res_lib_lck_resourceclose.header.error = SA_AIS_ERR_NOT_EXIST;
 
-		openais_conn_send_response (conn,
+		openais_response_send (
+			conn,
 			&res_lib_lck_resourceclose,
 			sizeof (struct res_lib_lck_resourceclose));
 	}
Index: exec/msg.c
===================================================================
--- exec/msg.c	(revision 1498)
+++ exec/msg.c	(revision 1499)
@@ -807,12 +807,12 @@
 				&req_exec_msg_queueopen->source,
 				sizeof (mar_message_source_t));
 
-			openais_conn_send_response (
+			openais_response_send (
 				req_exec_msg_queueopen->source.conn,
 				&res_lib_msg_queueopenasync,
 				sizeof (struct res_lib_msg_queueopenasync));
-			openais_conn_send_response (
-				openais_conn_partner_get (req_exec_msg_queueopen->source.conn),
+			openais_dispatch_send (
+				req_exec_msg_queueopen->source.conn,
 				&res_lib_msg_queueopenasync,
 				sizeof (struct res_lib_msg_queueopenasync));
 		} else {
@@ -826,7 +826,7 @@
 				&req_exec_msg_queueopen->source,
 				sizeof (mar_message_source_t));
 
-			openais_conn_send_response (
+			openais_dispatch_send (
 				req_exec_msg_queueopen->source.conn,
 				&res_lib_msg_queueopen,
 				sizeof (struct res_lib_msg_queueopen));
@@ -865,8 +865,10 @@
 		res_lib_msg_queueclose.header.size = sizeof (struct res_lib_msg_queueclose);
 		res_lib_msg_queueclose.header.id = MESSAGE_RES_MSG_QUEUECLOSE;
 		res_lib_msg_queueclose.header.error = error;
-		openais_conn_send_response (req_exec_msg_queueclose->source.conn,
-			&res_lib_msg_queueclose, sizeof (struct res_lib_msg_queueclose));
+		openais_dispatch_send (
+			req_exec_msg_queueclose->source.conn,
+			&res_lib_msg_queueclose,
+			sizeof (struct res_lib_msg_queueclose));
 	}
 }
 
@@ -927,7 +929,7 @@
 		res_lib_msg_queuegroupcreate.header.id = MESSAGE_RES_MSG_QUEUEGROUPCREATE;
 		res_lib_msg_queuegroupcreate.header.error = error;
 
-		openais_conn_send_response (
+		openais_dispatch_send (
 			req_exec_msg_queuegroupcreate->source.conn,
 			&res_lib_msg_queuegroupcreate,
 			sizeof (struct res_lib_msg_queuegroupcreate));
@@ -974,7 +976,7 @@
 		res_lib_msg_queuegroupinsert.header.id = MESSAGE_RES_MSG_QUEUEGROUPCREATE;
 		res_lib_msg_queuegroupinsert.header.error = error;
 
-		openais_conn_send_response (
+		openais_dispatch_send (
 			req_exec_msg_queuegroupinsert->source.conn,
 			&res_lib_msg_queuegroupinsert,
 			sizeof (struct res_lib_msg_queuegroupinsert));
@@ -1018,7 +1020,7 @@
 		res_lib_msg_queuegroupremove.header.id = MESSAGE_RES_MSG_QUEUEGROUPCREATE;
 		res_lib_msg_queuegroupremove.header.error = error;
 
-		openais_conn_send_response (
+		openais_dispatch_send (
 			req_exec_msg_queuegroupremove->source.conn,
 			&res_lib_msg_queuegroupremove,
 			sizeof (struct res_lib_msg_queuegroupremove));
@@ -1049,7 +1051,7 @@
 		res_lib_msg_queuegroupdelete.header.id = MESSAGE_RES_MSG_QUEUEGROUPCREATE;
 		res_lib_msg_queuegroupdelete.header.error = error;
 
-		openais_conn_send_response (
+		openais_dispatch_send (
 			req_exec_msg_queuegroupdelete->source.conn,
 			&res_lib_msg_queuegroupdelete,
 			sizeof (struct res_lib_msg_queuegroupdelete));
Index: lib/cpg.c
===================================================================
--- lib/cpg.c	(revision 1498)
+++ lib/cpg.c	(revision 1499)
@@ -107,8 +107,7 @@
 		goto error_destroy;
 	}
 
-	error = saServiceConnect (&cpg_inst->dispatch_fd,
-				     &cpg_inst->response_fd,
+	error = saServiceConnect (&cpg_inst->response_fd, &cpg_inst->dispatch_fd,
 		CPG_SERVICE);
 	if (error != SA_AIS_OK) {
 		goto error_put_destroy;
@@ -461,7 +460,7 @@
 	iov[0].iov_base = &req_lib_cpg_trackstart;
 	iov[0].iov_len = sizeof (struct req_lib_cpg_trackstart);
 
-	error = saSendMsgReceiveReply (cpg_inst->dispatch_fd, iov, 1,
+	error = saSendMsgReceiveReply (cpg_inst->response_fd, iov, 1,
 		&res_lib_cpg_trackstart, sizeof (struct res_lib_cpg_trackstart));
 
 	if (error != SA_AIS_OK) {
Index: README.devmap
===================================================================
--- README.devmap	(revision 1498)
+++ README.devmap	(revision 1499)
@@ -1221,3 +1221,91 @@
 for other people.
 
 Have fun!
+
+---
+IPC state machine
+
+lib_exit_fn may not use openais_response_send or openais_dispatch_send
+
+state INITIALIZING
+------------------
+receive response end of a library request
+	create conn_io data structure
+	if the connection's UID/GID is invalid
+		conn_io_disconnect
+	send response to conn_io with conn_io's address
+	set conn_io refcnt to 1
+
+receive dispatch end of a library connection with response end conn_io address
+	find matching connection
+	if the connection's UID/GID is invalid
+		conn_io_disconnect
+	send reponse to conn_io
+	create conn_info data structure
+	if conn_io response end is valid
+		store dispatch end of conn_io into conn_info data structure
+		store response conn_io into conn_info data structure
+	set conn_io refcnt to 1
+	call lib_init_fn for service type
+	set conn_info refcnt to 1
+	set state to ACTIVE
+
+event response disconnects and conn_info not bound
+	disconnect connection
+event receive connects and response connection not found
+	disconnect dispatch
+event no dispatch connection within 5 seconds
+	disconnect dispatch
+event disconnect_request
+	set state to DISCONNECT_REQUESTED
+	decrement response conn_io refcnt by 1
+	decrement dispatch conn_io refcnt by 1
+
+state ACTIVE
+------------
+do {
+	increment conn_io refcnt by 1
+	poll
+	if invalid poll set state internal_disconnect_request
+	dispatch library handler functions
+	flush any output that can be flushed
+	decrement conn_io refcnt by 1
+	if state set to DISCONNECT_REQUESTED
+		execute algorithm for disconnect requested state
+}
+
+event internal_disconnect_request
+	set state to DISCONNECT_REQUESTED
+	decrement response conn_io refcnt by 1
+	decrement dispatch conn_io refcnt by 1
+event openais_conn_refcnt_increment
+	increase conn_info reference count by 1
+event openais_conn_refcnt_decrement
+	decrease conn_info reference count by 1
+
+state DISCONNECT_REQUESTED
+--------------------------
+if this is the response conn_io data
+do {
+	if response conn_io refcnt is 0
+		destroy conn_io data structure
+		decrement conn_info reference count
+		exit thread
+	sleep configurable short duration
+}
+
+if this is the dispatch conn_io data
+do {
+	if dispatch conn_io refcnt is 0
+		call lib_exit_fn
+		if successful
+			destroy conn_io data structure
+			decrement conn_info reference count
+			exit thread
+	sleep configurable short duration
+}
+
+when conn_info reference count equals 0
+	free conn_info data structure
+
+
