Adding upstream version 1.0.12.upstream/1.0.12upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 2767 insertions, 0 deletions

diff --git a/lib/clplumbing/ipcsocket.c b/lib/clplumbing/ipcsocket.c
new file mode 100644
index 0000000..14c3504
--- /dev/null
+++ b/lib/clplumbing/ipcsocket.c
@@ -0,0 +1,2767 @@
+/*
+ * ipcsocket unix domain socket implementation of IPC abstraction.
+ *
+ * Copyright (c) 2002 Xiaoxiang Liu <xiliu@ncsa.uiuc.edu>
+ *
+ * Stream support (c) 2004,2006 David Lee <t.d.lee@durham.ac.uk>
+ *	Note: many of the variable/function names "*socket*" should be
+ *	interpreted as having a more generic "ipc-channel-type" meaning.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ *
+ */
+
+#include <lha_internal.h>
+
+#include <clplumbing/ipc.h>
+#include <clplumbing/cl_log.h>
+#include <clplumbing/realtime.h>
+#include <clplumbing/cl_poll.h>
+
+#include <ha_msg.h>
+/* avoid including cib.h - used in gshi's "late message" code to avoid
+ *   printing insanely large messages
+ */
+#define F_CIB_CALLDATA  "cib_calldata"
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <syslog.h>
+#include <time.h>
+#include <sched.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <sys/param.h>
+#include <sys/uio.h>
+#ifdef HAVE_SYS_FILIO_H
+#	include <sys/filio.h>
+#endif
+#ifdef HAVE_SYS_SYSLIMITS_H
+#	include <sys/syslimits.h>
+#endif
+#ifdef HAVE_SYS_CRED_H
+#	include <sys/cred.h>
+#endif
+#ifdef HAVE_SYS_UCRED_H
+#	include <sys/ucred.h>
+#endif
+
+/* For 'getpeerucred()' (Solaris 10 upwards) */
+#ifdef HAVE_UCRED_H
+#	include <ucred.h>
+#endif
+
+#ifdef HAVE_SYS_SOCKET_H
+# include <sys/socket.h>
+#endif
+
+/*
+ * Normally use "socket" code.  But on some OSes alternatives may be
+ * preferred (or necessary).
+ */
+#define HB_IPC_SOCKET	1
+#define HB_IPC_STREAM	2
+/* #define HB_IPC_ANOTHER	3 */
+
+#ifndef HB_IPC_METHOD
+# if defined(SO_PEERCRED) || defined(HAVE_GETPEEREID) \
+	|| defined(SCM_CREDS) || defined(HAVE_GETPEERUCRED)
+#  define HB_IPC_METHOD	HB_IPC_SOCKET
+# elif defined(HAVE_STROPTS_H)
+#  define HB_IPC_METHOD	HB_IPC_STREAM
+# else
+#  error.  Surely we have sockets or streams...
+# endif
+#endif
+
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+# include <sys/poll.h>
+# include <netinet/in.h>
+# include <sys/un.h>
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+# include <stropts.h>
+#else
+# error "IPC type invalid"
+#endif
+
+#include <sys/ioctl.h>
+#include <unistd.h>
+#include <errno.h>
+#include <fcntl.h>
+
+#ifndef UNIX_PATH_MAX
+#	define UNIX_PATH_MAX 108
+#endif
+
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+
+# define MAX_LISTEN_NUM 128
+
+# ifndef MSG_NOSIGNAL
+#  define		MSG_NOSIGNAL	0
+# endif
+
+# ifndef AF_LOCAL
+#  define         AF_LOCAL AF_UNIX
+# endif
+
+#endif /* HB_IPC_METHOD */
+
+/***********************************************************************
+ *
+ * Determine the IPC authentication scheme...  More machine dependent than
+ * we'd like, but don't know any better way...
+ *
+ ***********************************************************************/
+#ifdef SO_PEERCRED
+#	define	USE_SO_PEERCRED
+#elif HAVE_GETPEEREID
+#	define USE_GETPEEREID
+#elif defined(SCM_CREDS)
+#	define	USE_SCM_CREDS
+#elif HAVE_GETPEERUCRED		/* e.g. Solaris 10 upwards */
+#	define USE_GETPEERUCRED
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+#	define USE_STREAM_CREDS
+#else
+#	define	USE_DUMMY_CREDS
+/* This will make it compile, but attempts to authenticate
+ * will fail.  This is a stopgap measure ;-)
+ */
+#endif
+
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+
+# ifdef USE_BINDSTAT_CREDS
+# ifndef SUN_LEN
+#    define SUN_LEN(ptr) ((size_t) (offsetof (sockaddr_un, sun_path) + strlen ((ptr)->sun_path))
+# endif
+# endif
+
+#endif /* HB_IPC_METHOD */
+
+/* wait connection private data. */
+struct SOCKET_WAIT_CONN_PRIVATE{
+  /* the path name wich the connection will be built on. */
+  char path_name[UNIX_PATH_MAX];
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+  /* the domain socket. */
+  int s;
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+  /* the streams pipe */
+  int pipefds[2];
+#endif
+};
+
+/* channel private data. */
+struct SOCKET_CH_PRIVATE{
+  /* the path name wich the connection will be built on. */
+  char path_name[UNIX_PATH_MAX];
+  /* the domain socket. */
+  int s;
+  /* the size of expecting data for below buffered message buf_msg */
+  int remaining_data;
+
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+  /* The address of our peer - used by USE_BINDSTAT_CREDS version of
+   *   socket_verify_auth()
+   */
+  struct sockaddr_un *peer_addr;
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+  uid_t farside_uid;
+  gid_t farside_gid;
+#endif
+
+  /* the buf used to save unfinished message */
+  struct IPC_MESSAGE *buf_msg;
+};
+
+struct IPC_Stats {
+	long	nsent;
+	long	noutqueued;
+	long	send_count;
+	long	nreceived;
+	long	ninqueued;
+	long	recv_count;
+	int	last_recv_errno;
+	int	last_recv_rc;
+	int	last_send_errno;
+	int	last_send_rc;
+};
+
+static struct IPC_Stats SocketIPCStats = {0, 0, 0, 0};
+extern int	debug_level;
+
+/* unix domain socket implementations of IPC functions. */
+
+static int socket_resume_io(struct IPC_CHANNEL *ch);
+
+static struct IPC_MESSAGE* socket_message_new(struct IPC_CHANNEL*ch
+,	int msg_len);
+
+struct IPC_WAIT_CONNECTION *socket_wait_conn_new(GHashTable* ch_attrs);
+
+/* *** FIXME: This is also declared in 'ocf_ipc.c'. */
+struct IPC_CHANNEL* socket_client_channel_new(GHashTable *attrs);
+
+static struct IPC_CHANNEL* socket_server_channel_new(int sockfd);
+
+static struct IPC_CHANNEL * channel_new(int sockfd, int conntype, const char *pathname);
+static int client_channel_new_auth(int sockfd);
+static int verify_creds(struct IPC_AUTH *auth_info, uid_t uid, gid_t gid);
+
+typedef void (*DelProc)(IPC_Message*);
+
+static struct IPC_MESSAGE * ipcmsg_new(struct IPC_CHANNEL* ch,
+  const void* data, int len, void* private, DelProc d);
+
+static pid_t socket_get_farside_pid(int sockfd);
+
+extern int (*ipc_pollfunc_ptr)(struct pollfd *, nfds_t, int);
+
+static int socket_resume_io_read(struct IPC_CHANNEL *ch, int*, gboolean read1anyway);
+
+static struct IPC_OPS socket_ops;
+static gboolean ipc_time_debug_flag = TRUE;
+
+void
+set_ipc_time_debug_flag(gboolean flag)
+{
+	ipc_time_debug_flag = flag;
+}
+
+#ifdef IPC_TIME_DEBUG
+
+extern struct ha_msg* wirefmt2msg(const char* s, size_t length, int flag);
+void cl_log_message (int log_level, const struct ha_msg *m);
+int timediff(longclock_t t1, longclock_t t2);
+void   ha_msg_del(struct ha_msg* msg);
+void	ipc_time_debug(IPC_Channel* ch, IPC_Message* ipcmsg, int whichpos);
+
+#define SET_ENQUEUE_TIME(x,t)	memcpy(&((struct SOCKET_MSG_HEAD*)x->msg_buf)->enqueue_time, &t, sizeof(longclock_t))
+#define SET_SEND_TIME(x,t)	memcpy(&((struct SOCKET_MSG_HEAD*)x->msg_buf)->send_time, &t, sizeof(longclock_t))
+#define SET_RECV_TIME(x,t)	memcpy(&((struct SOCKET_MSG_HEAD*)x->msg_buf)->recv_time, &t, sizeof(longclock_t))
+#define SET_DEQUEUE_TIME(x,t)	memcpy(&((struct SOCKET_MSG_HEAD*)x->msg_buf)->dequeue_time, &t, sizeof(longclock_t))
+
+static
+longclock_t
+get_enqueue_time(IPC_Message *ipcmsg)
+{
+	longclock_t t;
+
+	memcpy(&t,
+	  &(((struct SOCKET_MSG_HEAD *)ipcmsg->msg_buf)->enqueue_time),
+	  sizeof(longclock_t));
+
+	return t;
+}
+
+int
+timediff(longclock_t t1, longclock_t t2)
+{
+	longclock_t	remain;
+
+	remain = sub_longclock(t1, t2);
+
+	return longclockto_ms(remain);
+}
+
+void
+ipc_time_debug(IPC_Channel* ch, IPC_Message* ipcmsg, int whichpos)
+{
+	int msdiff = 0;
+	longclock_t lnow =  time_longclock();
+	char positions[4][16]={
+		"enqueue",
+		"send",
+		"recv",
+		"dequeue"};
+
+	if (ipc_time_debug_flag == FALSE) {
+		return ;
+	}
+
+	if (ipcmsg->msg_body == NULL
+	    || ipcmsg->msg_buf == NULL) {
+		cl_log(LOG_ERR, "msg_body =%p, msg_bu=%p",
+		       ipcmsg->msg_body, ipcmsg->msg_buf);
+		abort();
+		return;
+	}
+
+	switch(whichpos) {
+		case MSGPOS_ENQUEUE:
+			SET_ENQUEUE_TIME(ipcmsg, lnow);
+			break;
+		case MSGPOS_SEND:
+			SET_SEND_TIME(ipcmsg, lnow);
+			goto checktime;
+		case MSGPOS_RECV:
+			SET_RECV_TIME(ipcmsg, lnow);
+			goto checktime;
+		case MSGPOS_DEQUEUE:
+			SET_DEQUEUE_TIME(ipcmsg, lnow);
+
+	checktime:
+			msdiff = timediff(lnow, get_enqueue_time(ipcmsg));
+			if (msdiff > MAXIPCTIME) {
+				struct ha_msg* hamsg = NULL;
+				cl_log(LOG_WARNING,
+				       " message delayed from enqueue to %s %d ms "
+				       "(enqueue-time=%lu, peer pid=%d) ",
+				       positions[whichpos],
+				       msdiff,
+				       longclockto_ms(get_enqueue_time(ipcmsg)),
+				       ch->farside_pid);
+
+				(void)hamsg;
+#if 0
+				hamsg = wirefmt2msg(ipcmsg->msg_body, ipcmsg->msg_len, 0);
+				if (hamsg != NULL) {
+					struct ha_msg *crm_data = NULL;
+					crm_data = cl_get_struct(
+						hamsg, F_CRM_DATA);
+
+					if(crm_data == NULL) {
+						crm_data = cl_get_struct(
+							hamsg, F_CIB_CALLDATA);
+					}
+					if(crm_data != NULL) {
+						cl_msg_remove_value(
+							hamsg, crm_data);
+					}
+
+					cl_log_message(LOG_DEBUG, hamsg);
+					ha_msg_del(hamsg);
+				} else {
+					if (!ipcmsg) {
+						cl_log(LOG_ERR,
+						"IPC msg 0x%lx is unallocated"
+						,	(gulong)ipcmsg);
+						return;
+					}
+					if (!ipcmsg->msg_body) {
+						cl_log(LOG_ERR,
+						"IPC msg body 0x%lx is unallocated"
+						,	(gulong)ipcmsg->msg_body);
+						return;
+					}
+				}
+#endif
+
+			}
+			break;
+		default:
+			cl_log(LOG_ERR, "wrong position value in IPC:%d", whichpos);
+			return;
+	}
+}
+#endif
+
+void dump_ipc_info(const IPC_Channel* chan);
+
+#undef AUDIT_CHANNELS
+
+#ifndef AUDIT_CHANNELS
+#	define	CHANAUDIT(ch)	/*NOTHING */
+#else
+#	define CHANAUDIT(ch)	socket_chan_audit(ch)
+#	define MAXPID	65535
+
+static void
+socket_chan_audit(const struct IPC_CHANNEL* ch)
+{
+	int	badch = FALSE;
+
+  	struct SOCKET_CH_PRIVATE *chp;
+	struct stat		b;
+
+	if ((chp = ch->ch_private) == NULL) {
+		cl_log(LOG_CRIT, "Bad ch_private");
+		badch = TRUE;
+	}
+	if (ch->ops != &socket_ops) {
+		cl_log(LOG_CRIT, "Bad socket_ops");
+		badch = TRUE;
+	}
+	if (ch->ch_status == IPC_DISCONNECT) {
+		return;
+	}
+	if (!IPC_ISRCONN(ch)) {
+		cl_log(LOG_CRIT, "Bad ch_status [%d]", ch->ch_status);
+		badch = TRUE;
+	}
+	if (ch->farside_pid < 0 || ch->farside_pid > MAXPID) {
+		cl_log(LOG_CRIT, "Bad farside_pid");
+		badch = TRUE;
+	}
+	if (fstat(chp->s, &b) < 0) {
+		badch = TRUE;
+	} else if ((b.st_mode & S_IFMT) != S_IFSOCK) {
+		cl_log(LOG_CRIT, "channel @ 0x%lx: not a socket"
+		,	(unsigned long)ch);
+		badch = TRUE;
+	}
+	if (chp->remaining_data < 0) {
+		cl_log(LOG_CRIT, "Negative remaining_data");
+		badch = TRUE;
+	}
+	if (chp->remaining_data < 0 || chp->remaining_data > MAXMSG) {
+		cl_log(LOG_CRIT, "Excessive/bad remaining_data");
+		badch = TRUE;
+	}
+	if (chp->remaining_data && chp->buf_msg == NULL) {
+		cl_log(LOG_CRIT
+		,	"inconsistent remaining_data [%ld]/buf_msg[0x%lx]"
+		,	(long)chp->remaining_data, (unsigned long)chp->buf_msg);
+		badch = TRUE;
+	}
+	if (chp->remaining_data == 0 && chp->buf_msg != NULL) {
+		cl_log(LOG_CRIT
+		,	"inconsistent remaining_data [%ld]/buf_msg[0x%lx] (2)"
+		,	(long)chp->remaining_data, (unsigned long)chp->buf_msg);
+		badch = TRUE;
+	}
+	if (ch->send_queue == NULL || ch->recv_queue == NULL) {
+		cl_log(LOG_CRIT, "bad send/recv queue");
+		badch = TRUE;
+	}
+	if (ch->recv_queue->current_qlen < 0
+	||	ch->recv_queue->current_qlen > ch->recv_queue->max_qlen) {
+		cl_log(LOG_CRIT, "bad recv queue");
+		badch = TRUE;
+	}
+	if (ch->send_queue->current_qlen < 0
+	||	ch->send_queue->current_qlen > ch->send_queue->max_qlen) {
+		cl_log(LOG_CRIT, "bad send_queue");
+		badch = TRUE;
+	}
+	if (badch) {
+		cl_log(LOG_CRIT, "Bad channel @ 0x%lx", (unsigned long)ch);
+		dump_ipc_info(ch);
+		abort();
+	}
+}
+#endif
+
+#ifdef CHEAT_CHECKS
+long	SeqNums[32];
+
+static long
+cheat_get_sequence(IPC_Message* msg)
+{
+	const char header [] = "String-";
+	size_t header_len = sizeof(header)-1;
+	char *	body;
+
+	if (msg == NULL || msg->msg_len < sizeof(header)
+	||	msg->msg_len > sizeof(header) + 10
+	||	strncmp(msg->msg_body, header, header_len) != 0) {
+		return -1L;
+	}
+	body = msg->msg_body;
+	return atol(body+header_len);
+}
+static char SavedReadBody[32];
+static char SavedReceivedBody[32];
+static char SavedQueuedBody[32];
+static char SavedSentBody[32];
+#ifndef MIN
+#	define MIN(a,b)	(a < b ? a : b)
+#endif
+
+static void
+save_body(struct IPC_MESSAGE *msg, char * savearea, size_t length)
+{
+	int mlen = strnlen(msg->msg_body, MIN(length, msg->msg_len));
+	memcpy(savearea, msg->msg_body, mlen);
+	savearea[mlen] = EOS;
+}
+
+static void
+audit_readmsgq_msg(gpointer msg, gpointer user_data)
+{
+	long	cheatseq = cheat_get_sequence(msg);
+
+	if (cheatseq < SeqNums[1] || cheatseq > SeqNums[2]) {
+		cl_log(LOG_ERR
+		,	"Read Q Message %ld not in range [%ld:%ld]"
+		,	cheatseq, SeqNums[1], SeqNums[2]);
+	}
+}
+
+static void
+saveandcheck(struct IPC_CHANNEL * ch, struct IPC_MESSAGE* msg, char * savearea
+,	size_t savesize, long* lastseq, const char * text)
+{
+	long	cheatseq = cheat_get_sequence(msg);
+
+	save_body(msg, savearea, savesize);
+	if (*lastseq != 0 ) {
+		if (cheatseq != *lastseq +1) {
+			int	j;
+			cl_log(LOG_ERR
+			,	"%s packets out of sequence! %ld versus %ld [pid %d]"
+			,	text, cheatseq, *lastseq, (int)getpid());
+			dump_ipc_info(ch);
+			for (j=0; j < 4; ++j) {
+				cl_log(LOG_DEBUG
+				,	"SeqNums[%d] = %ld"
+				,	j, SeqNums[j]);
+			}
+			cl_log(LOG_ERR
+			,	"SocketIPCStats.nsent = %ld"
+			,	SocketIPCStats.nsent);
+			cl_log(LOG_ERR
+			,	"SocketIPCStats.noutqueued = %ld"
+			,	SocketIPCStats.noutqueued);
+			cl_log(LOG_ERR
+			,	"SocketIPCStats.nreceived = %ld"
+			,	SocketIPCStats.nreceived);
+			cl_log(LOG_ERR
+			,	"SocketIPCStats.ninqueued = %ld"
+			,	SocketIPCStats.ninqueued);
+		}
+
+	}
+	g_list_foreach(ch->recv_queue->queue, audit_readmsgq_msg, NULL);
+	if (cheatseq > 0) {
+		*lastseq = cheatseq;
+	}
+}
+
+#	define	CHECKFOO(which, ch, msg, area, text)	{			\
+		saveandcheck(ch,msg,area,sizeof(area),SeqNums+which,text);	\
+	}
+#else
+#	define	CHECKFOO(which, ch, msg, area, text)	/* Nothing */
+#endif
+
+static void
+dump_msg(struct IPC_MESSAGE *msg, const char * label)
+{
+#ifdef CHEAT_CHECKS
+	cl_log(LOG_DEBUG, "%s packet (length %d) [%s] %ld pid %d"
+	,	label,	(int)msg->msg_len, (char*)msg->msg_body
+	,	cheat_get_sequence(msg), (int)getpid());
+#else
+	cl_log(LOG_DEBUG, "%s length %d [%s] pid %d"
+	,	label,	(int)msg->msg_len, (char*)msg->msg_body
+	,	(int)getpid());
+#endif
+}
+
+static void
+dump_msgq_msg(gpointer data, gpointer user_data)
+{
+	dump_msg(data, user_data);
+}
+
+void
+dump_ipc_info(const IPC_Channel* chan)
+{
+	char squeue[] = "Send queue";
+	char rqueue[] = "Receive queue";
+#ifdef CHEAT_CHECKS
+	cl_log(LOG_DEBUG, "Saved Last Body read[%s]", SavedReadBody);
+	cl_log(LOG_DEBUG, "Saved Last Body received[%s]", SavedReceivedBody);
+	cl_log(LOG_DEBUG, "Saved Last Body Queued[%s]", SavedQueuedBody);
+	cl_log(LOG_DEBUG, "Saved Last Body Sent[%s]", SavedSentBody);
+#endif
+	g_list_foreach(chan->send_queue->queue, dump_msgq_msg, squeue);
+	g_list_foreach(chan->recv_queue->queue, dump_msgq_msg, rqueue);
+	CHANAUDIT(chan);
+}
+
+/* destroy socket wait channel */
+static void
+socket_destroy_wait_conn(struct IPC_WAIT_CONNECTION * wait_conn)
+{
+	struct SOCKET_WAIT_CONN_PRIVATE * wc = wait_conn->ch_private;
+
+	if (wc != NULL) {
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+		if (wc->s >= 0) {
+			if (debug_level > 1) {
+				cl_log(LOG_DEBUG
+				,	"%s: closing socket %d"
+				,	__FUNCTION__, wc->s);
+			}
+			close(wc->s);
+			cl_poll_ignore(wc->s);
+			unlink(wc->path_name);
+			wc->s = -1;
+		}
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+		cl_poll_ignore(wc->pipefds[0]);
+		if (wc->pipefds[0] >= 0) {
+			if (debug_level > 1) {
+				cl_log(LOG_DEBUG
+				,	"%s: closing pipe[0] %d"
+				,	__FUNCTION__, wc->pipefds[0]);
+			}
+			wc->pipefds[0] = -1;
+		}
+		if (wc->pipefds[1] >= 0) {
+			if (debug_level > 1) {
+				cl_log(LOG_DEBUG
+				,	"%s: closing pipe[1] %d"
+				,	__FUNCTION__, wc->pipefds[1]);
+			}
+			wc->pipefds[0] = -1;
+		}
+		unlink(wc->path_name);
+#endif
+		g_free(wc);
+	}
+	g_free((void*) wait_conn);
+}
+
+/* return a fd which can be listened on for new connections. */
+static int
+socket_wait_selectfd(struct IPC_WAIT_CONNECTION *wait_conn)
+{
+	struct SOCKET_WAIT_CONN_PRIVATE * wc = wait_conn->ch_private;
+
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+	return (wc == NULL ? -1 : wc->s);
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+	return (wc == NULL ? -1 : wc->pipefds[0]);
+#endif
+}
+
+/* socket accept connection. */
+static struct IPC_CHANNEL*
+socket_accept_connection(struct IPC_WAIT_CONNECTION * wait_conn
+,	struct IPC_AUTH *auth_info)
+{
+	struct IPC_CHANNEL *			ch = NULL;
+	int					s;
+	int					new_sock;
+	struct SOCKET_WAIT_CONN_PRIVATE*	conn_private;
+	struct SOCKET_CH_PRIVATE *		ch_private ;
+	int auth_result = IPC_FAIL;
+	int					saveerrno=errno;
+	gboolean was_error = FALSE;
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+	/* make peer_addr a pointer so it can be used by the
+	 *   USE_BINDSTAT_CREDS implementation of socket_verify_auth()
+	 */
+	struct sockaddr_un *			peer_addr = NULL;
+	socklen_t				sin_size;
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+	struct strrecvfd strrecvfd;
+#endif
+
+	/* get select fd */
+
+	s = wait_conn->ops->get_select_fd(wait_conn);
+	if (s < 0) {
+		cl_log(LOG_ERR, "get_select_fd: invalid fd");
+		return NULL;
+	}
+
+	/* Get client connection. */
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+	peer_addr = g_new(struct sockaddr_un, 1);
+	*peer_addr->sun_path = '\0';
+	sin_size = sizeof(struct sockaddr_un);
+	new_sock = accept(s, (struct sockaddr *)peer_addr, &sin_size);
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+	if (ioctl(s, I_RECVFD, &strrecvfd) == -1) {
+		new_sock = -1;
+	}
+	else {
+		new_sock = strrecvfd.fd;
+	}
+#endif
+	saveerrno=errno;
+	if (new_sock == -1) {
+		if (errno != EAGAIN && errno != EWOULDBLOCK) {
+			cl_perror("socket_accept_connection: accept(sock=%d)"
+			,	s);
+		}
+		was_error = TRUE;
+
+	} else {
+		if ((ch = socket_server_channel_new(new_sock)) == NULL) {
+			cl_log(LOG_ERR
+			,	"socket_accept_connection:"
+			        " Can't create new channel");
+			was_error = TRUE;
+		} else {
+			conn_private=(struct SOCKET_WAIT_CONN_PRIVATE*)
+			(	wait_conn->ch_private);
+			ch_private = (struct SOCKET_CH_PRIVATE *)(ch->ch_private);
+			strncpy(ch_private->path_name,conn_private->path_name
+			,		sizeof(conn_private->path_name));
+
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+			ch_private->peer_addr = peer_addr;
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+			ch_private->farside_uid = strrecvfd.uid;
+			ch_private->farside_gid = strrecvfd.gid;
+#endif
+		}
+	}
+
+	/* Verify the client authorization information. */
+	if(was_error == FALSE) {
+		auth_result = ch->ops->verify_auth(ch, auth_info);
+		if (auth_result == IPC_OK) {
+			ch->ch_status = IPC_CONNECT;
+			ch->farside_pid = socket_get_farside_pid(new_sock);
+			return ch;
+		}
+		saveerrno=errno;
+	}
+
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+	g_free(peer_addr);
+	peer_addr = NULL;
+#endif
+	errno=saveerrno;
+	return NULL;
+}
+
+/*
+ * Called by socket_destroy(). Disconnect the connection
+ * and set ch_status to IPC_DISCONNECT.
+ *
+ * parameters :
+ *     ch (IN) the pointer to the channel.
+ *
+ * return values :
+ *     IPC_OK   the connection is disconnected successfully.
+ *      IPC_FAIL     operation fails.
+*/
+
+static int
+socket_disconnect(struct IPC_CHANNEL* ch)
+{
+	struct SOCKET_CH_PRIVATE* conn_info;
+
+	conn_info = (struct SOCKET_CH_PRIVATE*) ch->ch_private;
+	if (debug_level > 1) {
+		cl_log(LOG_DEBUG
+		,	"%s(sock=%d, ch=0x%lx){"
+		,	__FUNCTION__
+		,	conn_info->s, (unsigned long)ch);
+	}
+#if 0
+	if (ch->ch_status != IPC_DISCONNECT) {
+  		cl_log(LOG_INFO, "forced disconnect for fd %d", conn_info->s);
+	}
+#endif
+	if (ch->ch_status == IPC_CONNECT) {
+		socket_resume_io(ch);
+	}
+
+	if (conn_info->s >= 0) {
+		if (debug_level > 1) {
+			cl_log(LOG_DEBUG
+			,	"%s: closing socket %d"
+			,	__FUNCTION__, conn_info->s);
+		}
+		close(conn_info->s);
+		cl_poll_ignore(conn_info->s);
+		conn_info->s = -1;
+	}
+	ch->ch_status = IPC_DISCONNECT;
+	if (debug_level > 1) {
+		cl_log(LOG_DEBUG, "}/*%s(sock=%d, ch=0x%lx)*/"
+		,	__FUNCTION__, conn_info->s, (unsigned long)ch);
+	}
+	return IPC_OK;
+}
+
+/*
+ * destroy a ipc queue and clean all memory space assigned to this queue.
+ * parameters:
+ *      q  (IN) the pointer to the queue which should be destroied.
+ *
+ *	FIXME:  This function does not free up messages that might
+ *	be in the queue.
+ */
+
+static void
+socket_destroy_queue(struct IPC_QUEUE * q)
+{
+  g_list_free(q->queue);
+
+  g_free((void *) q);
+}
+
+static void
+socket_destroy_channel(struct IPC_CHANNEL * ch)
+{
+	--ch->refcount;
+	if (ch->refcount > 0) {
+		return;
+	}
+	if (ch->ch_status == IPC_CONNECT) {
+		socket_resume_io(ch);
+	}
+	if (debug_level > 1) {
+		cl_log(LOG_DEBUG, "socket_destroy(ch=0x%lx){"
+		,	(unsigned long)ch);
+	}
+	socket_disconnect(ch);
+	socket_destroy_queue(ch->send_queue);
+	socket_destroy_queue(ch->recv_queue);
+
+	if (ch->pool) {
+		ipc_bufpool_unref(ch->pool);
+	}
+
+	if (ch->ch_private != NULL) {
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+		struct SOCKET_CH_PRIVATE *priv = (struct SOCKET_CH_PRIVATE *)
+			ch->ch_private;
+		if(priv->peer_addr != NULL) {
+			if (*priv->peer_addr->sun_path) {
+				unlink(priv->peer_addr->sun_path);
+			}
+			g_free((void*)(priv->peer_addr));
+		}
+#endif
+    		g_free((void*)(ch->ch_private));
+	}
+	memset(ch, 0xff, sizeof(*ch));
+	g_free((void*)ch);
+	if (debug_level > 1) {
+		cl_log(LOG_DEBUG, "}/*socket_destroy(ch=0x%lx)*/"
+		,	(unsigned long)ch);
+	}
+}
+
+static int
+socket_check_disc_pending(struct IPC_CHANNEL* ch)
+{
+	int		rc;
+	struct pollfd	sockpoll;
+
+	if (ch->ch_status == IPC_DISCONNECT) {
+		cl_log(LOG_ERR, "check_disc_pending() already disconnected");
+		return IPC_BROKEN;
+	}
+	if (ch->recv_queue->current_qlen > 0) {
+		return IPC_OK;
+	}
+	sockpoll.fd = ch->ops->get_recv_select_fd(ch);
+	sockpoll.events = POLLIN;
+
+	rc = ipc_pollfunc_ptr(&sockpoll, 1, 0);
+
+ 	if (rc < 0) {
+		cl_log(LOG_INFO
+		,	"socket_check_disc_pending() bad poll call");
+		ch->ch_status = IPC_DISCONNECT;
+ 		return IPC_BROKEN;
+	}
+
+	if (sockpoll.revents & POLLHUP) {
+		if (sockpoll.revents & POLLIN) {
+			ch->ch_status = IPC_DISC_PENDING;
+		} else {
+#if 1
+			cl_log(LOG_INFO, "HUP without input");
+#endif
+			ch->ch_status = IPC_DISCONNECT;
+			return IPC_BROKEN;
+		}
+	}
+	if (sockpoll.revents & POLLIN) {
+		int dummy;
+		socket_resume_io_read(ch, &dummy, FALSE);
+	}
+	return IPC_OK;
+}
+
+static int
+socket_initiate_connection(struct IPC_CHANNEL * ch)
+{
+	struct SOCKET_CH_PRIVATE* conn_info;
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+	struct sockaddr_un peer_addr; /* connector's address information */
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+#endif
+
+	conn_info = (struct SOCKET_CH_PRIVATE*) ch->ch_private;
+
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+	/* Prepare the socket */
+	memset(&peer_addr, 0, sizeof(peer_addr));
+	peer_addr.sun_family = AF_LOCAL;    /* host byte order */
+
+	if (strlen(conn_info->path_name) >= sizeof(peer_addr.sun_path)) {
+		return IPC_FAIL;
+	}
+	strncpy(peer_addr.sun_path, conn_info->path_name, sizeof(peer_addr.sun_path));
+
+	/* Send connection request */
+	if (connect(conn_info->s, (struct sockaddr *)&peer_addr
+	, 	sizeof(struct sockaddr_un)) == -1) {
+		return IPC_FAIL;
+	}
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+
+#endif
+
+	ch->ch_status = IPC_CONNECT;
+	ch->farside_pid = socket_get_farside_pid(conn_info->s);
+	return IPC_OK;
+}
+
+static void
+socket_set_high_flow_callback(IPC_Channel* ch,
+			      flow_callback_t callback,
+			      void* userdata) {
+	ch->high_flow_callback = callback;
+	ch->high_flow_userdata = userdata;
+}
+
+static void
+socket_set_low_flow_callback(IPC_Channel* ch,
+			     flow_callback_t callback,
+			     void* userdata) {
+	ch->low_flow_callback = callback;
+	ch->low_flow_userdata = userdata;
+}
+
+static void
+socket_check_flow_control(struct IPC_CHANNEL* ch,
+			  int orig_qlen,
+			  int curr_qlen)
+{
+	if (!IPC_ISRCONN(ch)) {
+		return;
+	}
+
+	if (curr_qlen >= ch->high_flow_mark
+	    && ch->high_flow_callback) {
+			ch->high_flow_callback(ch, ch->high_flow_userdata);
+	}
+
+	if (curr_qlen <= ch->low_flow_mark
+	    && orig_qlen > ch->low_flow_mark
+	    && ch->low_flow_callback) {
+		ch->low_flow_callback(ch, ch->low_flow_userdata);
+	}
+}
+
+static int
+socket_send(struct IPC_CHANNEL * ch, struct IPC_MESSAGE* msg)
+{
+	int orig_qlen;
+	int diff;
+	struct IPC_MESSAGE* newmsg;
+
+	if (msg->msg_len > MAXMSG) {
+		cl_log(LOG_ERR, "%s: sorry, cannot send messages "
+			"bigger than %d (requested %lu)",
+			__FUNCTION__, MAXMSG, (unsigned long)msg->msg_len);
+		return IPC_FAIL;
+	}
+	if (msg->msg_len < 0) {
+		cl_log(LOG_ERR, "socket_send: "
+		       "invalid message");
+		return IPC_FAIL;
+	}
+
+	if (ch->ch_status != IPC_CONNECT) {
+		return IPC_FAIL;
+	}
+
+	ch->ops->resume_io(ch);
+
+	if (ch->send_queue->maxqlen_cnt &&
+		time(NULL) - ch->send_queue->last_maxqlen_warn >= 60) {
+	    cl_log(LOG_ERR, "%u messages dropped on a non-blocking channel (send queue maximum length %d)",
+		   ch->send_queue->maxqlen_cnt, (int)ch->send_queue->max_qlen);
+	    ch->send_queue->maxqlen_cnt = 0;
+	}
+	if ( !ch->should_send_block &&
+	    ch->send_queue->current_qlen >= ch->send_queue->max_qlen) {
+		if (!ch->send_queue->maxqlen_cnt) {
+			ch->send_queue->last_maxqlen_warn = time(NULL);
+		}
+		ch->send_queue->maxqlen_cnt++;
+
+		if (ch->should_block_fail) {
+			return IPC_FAIL;
+		} else {
+			return IPC_OK;
+		}
+	}
+
+	while (ch->send_queue->current_qlen >= ch->send_queue->max_qlen) {
+		if (ch->ch_status != IPC_CONNECT) {
+		 	cl_log(LOG_WARNING, "socket_send:"
+			" message queue exceeded and IPC not connected");
+			return IPC_FAIL;
+		}
+		cl_shortsleep();
+		ch->ops->resume_io(ch);
+	}
+
+	/* add the message into the send queue */
+	CHECKFOO(0,ch, msg, SavedQueuedBody, "queued message");
+	SocketIPCStats.noutqueued++;
+
+	diff = 0;
+	if (msg->msg_buf ) {
+		diff = (char*)msg->msg_body - (char*)msg->msg_buf;
+	}
+	if ( diff < (int)sizeof(struct SOCKET_MSG_HEAD) ) {
+		/* either we don't have msg->msg_buf set
+		 * or we don't have enough bytes for socket head
+		 * we delete this message and creates
+		 * a new one and delete the old one
+		 */
+
+		newmsg= socket_message_new(ch, msg->msg_len);
+		if (newmsg == NULL) {
+			cl_log(LOG_ERR, "socket_resume_io_write: "
+			       "allocating memory for new ipc msg failed");
+			return IPC_FAIL;
+		}
+
+		memcpy(newmsg->msg_body, msg->msg_body, msg->msg_len);
+
+		if(msg->msg_done) {
+			msg->msg_done(msg);
+		};
+		msg = newmsg;
+	}
+#ifdef IPC_TIME_DEBUG
+	ipc_time_debug(ch,msg, MSGPOS_ENQUEUE);
+#endif
+	ch->send_queue->queue = g_list_append(ch->send_queue->queue,
+					      msg);
+	orig_qlen = ch->send_queue->current_qlen++;
+
+	socket_check_flow_control(ch, orig_qlen, orig_qlen +1 );
+
+	/* resume io */
+	ch->ops->resume_io(ch);
+	return IPC_OK;
+}
+
+static int
+socket_recv(struct IPC_CHANNEL * ch, struct IPC_MESSAGE** message)
+{
+	GList *element;
+
+	int		nbytes;
+	int		result;
+
+	socket_resume_io(ch);
+	result = socket_resume_io_read(ch, &nbytes, TRUE);
+
+	*message = NULL;
+
+	if (ch->recv_queue->current_qlen == 0) {
+		return result != IPC_OK ? result : IPC_FAIL;
+		/*return IPC_OK;*/
+	}
+	element = g_list_first(ch->recv_queue->queue);
+
+	if (element == NULL) {
+		/* Internal accounting error, but correctable */
+		cl_log(LOG_ERR
+		, "recv failure: qlen (%ld) > 0, but no message found."
+		,	(long)ch->recv_queue->current_qlen);
+		ch->recv_queue->current_qlen = 0;
+		return IPC_FAIL;
+	}
+	*message = (struct IPC_MESSAGE *) (element->data);
+#ifdef IPC_TIME_DEBUG
+	ipc_time_debug(ch, *message, MSGPOS_DEQUEUE);
+#endif
+
+	CHECKFOO(1,ch, *message, SavedReadBody, "read message");
+	SocketIPCStats.nreceived++;
+	ch->recv_queue->queue =	g_list_remove(ch->recv_queue->queue
+	,	element->data);
+	ch->recv_queue->current_qlen--;
+	return IPC_OK;
+}
+
+static int
+socket_check_poll(struct IPC_CHANNEL * ch
+,		struct pollfd * sockpoll)
+{
+	if (ch->ch_status == IPC_DISCONNECT) {
+		return IPC_OK;
+	}
+	if (sockpoll->revents & POLLHUP) {
+		/* If input present, or this is an output-only poll... */
+		if (sockpoll->revents & POLLIN
+		|| (sockpoll-> events & POLLIN) == 0 ) {
+			ch->ch_status = IPC_DISC_PENDING;
+			return IPC_OK;
+		}
+#if 1
+		cl_log(LOG_INFO, "socket_check_poll(): HUP without input");
+#endif
+		ch->ch_status = IPC_DISCONNECT;
+		return IPC_BROKEN;
+
+	} else if (sockpoll->revents & (POLLNVAL|POLLERR)) {
+		/* Have we already closed the socket? */
+		if (fcntl(sockpoll->fd, F_GETFL) < 0) {
+			cl_perror("socket_check_poll(pid %d): bad fd [%d]"
+			,	(int) getpid(), sockpoll->fd);
+			ch->ch_status = IPC_DISCONNECT;
+			return IPC_OK;
+		}
+		cl_log(LOG_ERR
+		,	"revents failure: fd %d, flags 0x%x"
+		,	sockpoll->fd, sockpoll->revents);
+		errno = EINVAL;
+		return IPC_FAIL;
+	}
+	return IPC_OK;
+}
+
+static int
+socket_waitfor(struct IPC_CHANNEL * ch
+,	gboolean (*finished)(struct IPC_CHANNEL * ch))
+{
+	struct pollfd sockpoll;
+
+	CHANAUDIT(ch);
+	if (finished(ch)) {
+		return IPC_OK;
+	}
+
+ 	if (ch->ch_status == IPC_DISCONNECT) {
+ 		return IPC_BROKEN;
+	}
+	sockpoll.fd = ch->ops->get_recv_select_fd(ch);
+
+	while (!finished(ch) &&	IPC_ISRCONN(ch)) {
+		int	rc;
+
+		sockpoll.events = POLLIN;
+
+		/* Cannot call resume_io after the call to finished()
+		 * and before the call to poll because we might
+		 * change the state of the thing finished() is
+		 * waiting for.
+		 * This means that the poll call below would be
+		 * not only pointless, but might
+		 * make us hang forever waiting for this
+		 * event which has already happened
+		 */
+		if (ch->send_queue->current_qlen > 0) {
+			sockpoll.events |= POLLOUT;
+		}
+
+		rc = ipc_pollfunc_ptr(&sockpoll, 1, -1);
+
+		if (rc < 0) {
+			return (errno == EINTR ? IPC_INTR : IPC_FAIL);
+		}
+
+		rc = socket_check_poll(ch, &sockpoll);
+		if (sockpoll.revents & POLLIN) {
+			socket_resume_io(ch);
+		}
+		if (rc != IPC_OK) {
+			CHANAUDIT(ch);
+			return rc;
+		}
+	}
+
+	CHANAUDIT(ch);
+	return IPC_OK;
+}
+
+static int
+socket_waitin(struct IPC_CHANNEL * ch)
+{
+	return socket_waitfor(ch, ch->ops->is_message_pending);
+}
+static gboolean
+socket_is_output_flushed(struct IPC_CHANNEL * ch)
+{
+	return ! ch->ops->is_sending_blocked(ch);
+}
+
+static int
+socket_waitout(struct IPC_CHANNEL * ch)
+{
+	int	rc;
+	CHANAUDIT(ch);
+	rc = socket_waitfor(ch, socket_is_output_flushed);
+
+	if (rc != IPC_OK) {
+		cl_log(LOG_ERR
+		,	"socket_waitout failure: rc = %d", rc);
+	} else if (ch->ops->is_sending_blocked(ch)) {
+		cl_log(LOG_ERR, "socket_waitout output still blocked");
+	}
+	CHANAUDIT(ch);
+	return rc;
+}
+
+static gboolean
+socket_is_message_pending(struct IPC_CHANNEL * ch)
+{
+	int nbytes;
+
+	socket_resume_io_read(ch, &nbytes, TRUE);
+	ch->ops->resume_io(ch);
+	if (ch->recv_queue->current_qlen > 0) {
+		return TRUE;
+	}
+
+	return !IPC_ISRCONN(ch);
+}
+
+static gboolean
+socket_is_output_pending(struct IPC_CHANNEL * ch)
+{
+	socket_resume_io(ch);
+	return 	ch->ch_status == IPC_CONNECT
+	&&	 ch->send_queue->current_qlen > 0;
+}
+
+static gboolean
+socket_is_sendq_full(struct IPC_CHANNEL * ch)
+{
+	ch->ops->resume_io(ch);
+	return(ch->send_queue->current_qlen == ch->send_queue->max_qlen);
+}
+
+static gboolean
+socket_is_recvq_full(struct IPC_CHANNEL * ch)
+{
+	ch->ops->resume_io(ch);
+	return(ch->recv_queue->current_qlen == ch->recv_queue->max_qlen);
+}
+
+static int
+socket_get_conntype(struct IPC_CHANNEL* ch)
+{
+	return ch->conntype;
+}
+
+static int
+socket_assert_auth(struct IPC_CHANNEL *ch, GHashTable *auth)
+{
+	cl_log(LOG_ERR
+	, "the assert_auth function for domain socket is not implemented");
+	return IPC_FAIL;
+}
+
+static int
+socket_resume_io_read(struct IPC_CHANNEL *ch, int* nbytes, gboolean read1anyway)
+{
+	struct SOCKET_CH_PRIVATE*	conn_info;
+	int				retcode = IPC_OK;
+	struct pollfd			sockpoll;
+	int				debug_loopcount = 0;
+	int				debug_bytecount = 0;
+	size_t				maxqlen = ch->recv_queue->max_qlen;
+	struct ipc_bufpool*		pool = ch->pool;
+	int				nmsgs = 0;
+	int				spaceneeded;
+	*nbytes = 0;
+
+	CHANAUDIT(ch);
+	conn_info = (struct SOCKET_CH_PRIVATE *) ch->ch_private;
+
+	if (ch->ch_status == IPC_DISCONNECT) {
+		return IPC_BROKEN;
+	}
+
+	if (pool == NULL) {
+		ch->pool = pool = ipc_bufpool_new(0);
+		if (pool == NULL) {
+			cl_log(LOG_ERR, "socket_resume_io_read: "
+			       "memory allocation for ipc pool failed");
+			return IPC_FAIL;
+		}
+	}
+
+	if (ipc_bufpool_full(pool, ch, &spaceneeded)) {
+		struct ipc_bufpool*	newpool;
+
+		newpool = ipc_bufpool_new(spaceneeded);
+		if (newpool == NULL) {
+			cl_log(LOG_ERR, "socket_resume_io_read: "
+			       "memory allocation for a new ipc pool failed");
+			return IPC_FAIL;
+		}
+
+		ipc_bufpool_partial_copy(newpool, pool);
+		ipc_bufpool_unref(pool);
+		ch->pool = pool = newpool;
+	}
+	if (maxqlen <= 0 && read1anyway) {
+		maxqlen = 1;
+	}
+  	if (ch->recv_queue->current_qlen < maxqlen && retcode == IPC_OK) {
+		void *				msg_begin;
+		int				msg_len;
+		int				len;
+#if HB_IPC_METHOD == HB_IPC_STREAM
+		struct strbuf d;
+		int flags, rc;
+#endif
+
+		CHANAUDIT(ch);
+		++debug_loopcount;
+
+		len = ipc_bufpool_spaceleft(pool);
+		msg_begin = pool->currpos;
+
+		CHANAUDIT(ch);
+
+		/* Now try to receive some data */
+
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+		msg_len = recv(conn_info->s, msg_begin, len, MSG_DONTWAIT);
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+		d.maxlen = len;
+		d.len = 0;
+		d.buf = msg_begin;
+		flags = 0;
+		rc = getmsg(conn_info->s, NULL, &d, &flags);
+		msg_len = (rc < 0) ? rc : d.len;
+#endif
+		SocketIPCStats.last_recv_rc = msg_len;
+		SocketIPCStats.last_recv_errno = errno;
+		++SocketIPCStats.recv_count;
+
+		/* Did we get an error? */
+		if (msg_len < 0) {
+			switch (errno) {
+			case EAGAIN:
+				if (ch->ch_status==IPC_DISC_PENDING) {
+					ch->ch_status =IPC_DISCONNECT;
+					retcode = IPC_BROKEN;
+				}
+				break;
+
+			case ECONNREFUSED:
+			case ECONNRESET:
+				ch->ch_status = IPC_DISC_PENDING;
+				retcode= socket_check_disc_pending(ch);
+				break;
+
+			default:
+				cl_perror("socket_resume_io_read"
+					  ": unknown recv error, peerpid=%d",
+					  ch->farside_pid);
+				ch->ch_status = IPC_DISCONNECT;
+				retcode = IPC_FAIL;
+				break;
+			}
+
+		} else if (msg_len == 0) {
+			ch->ch_status = IPC_DISC_PENDING;
+			if(ch->recv_queue->current_qlen <= 0) {
+				ch->ch_status = IPC_DISCONNECT;
+				retcode = IPC_FAIL;
+			}
+		} else {
+			/* We read something! */
+			/* Note that all previous cases break out of the loop */
+			debug_bytecount += msg_len;
+			*nbytes = msg_len;
+			nmsgs = ipc_bufpool_update(pool, ch, msg_len, ch->recv_queue) ;
+
+			if (nmsgs < 0) {
+				/* we didn't like the other side */
+				cl_log(LOG_ERR, "socket_resume_io_read: "
+					   "disconnecting the other side");
+				ch->ch_status = IPC_DISCONNECT;
+				retcode = IPC_FAIL;
+			} else {
+				SocketIPCStats.ninqueued += nmsgs;
+			}
+		}
+	}
+
+	/* Check for errors uncaught by recv() */
+	/* NOTE: It doesn't seem right we have to do this every time */
+	/* FIXME?? */
+
+	memset(&sockpoll,0, sizeof(struct pollfd));
+	if ((retcode == IPC_OK)
+	&&	(sockpoll.fd = conn_info->s) >= 0) {
+		/* Just check for errors, not for data */
+		sockpoll.events = 0;
+		ipc_pollfunc_ptr(&sockpoll, 1, 0);
+		retcode = socket_check_poll(ch, &sockpoll);
+	}
+
+	CHANAUDIT(ch);
+	if (retcode != IPC_OK) {
+		return retcode;
+	}
+
+	return IPC_ISRCONN(ch) ? IPC_OK : IPC_BROKEN;
+}
+
+static int
+socket_resume_io_write(struct IPC_CHANNEL *ch, int* nmsg)
+{
+	int				retcode = IPC_OK;
+	struct SOCKET_CH_PRIVATE*	conn_info;
+
+	CHANAUDIT(ch);
+	*nmsg = 0;
+	conn_info = (struct SOCKET_CH_PRIVATE *) ch->ch_private;
+
+	while (ch->ch_status == IPC_CONNECT
+	&&		retcode == IPC_OK
+	&&		ch->send_queue->current_qlen > 0) {
+
+		GList *				element;
+		struct IPC_MESSAGE *		msg;
+		struct SOCKET_MSG_HEAD		head;
+                struct IPC_MESSAGE* 		oldmsg = NULL;
+		int				sendrc = 0;
+                struct IPC_MESSAGE* 		newmsg;
+		char*				p;
+		unsigned int			bytes_remaining;
+		int				diff;
+
+		CHANAUDIT(ch);
+		element = g_list_first(ch->send_queue->queue);
+		if (element == NULL) {
+			/* OOPS!  - correct consistency problem */
+			ch->send_queue->current_qlen = 0;
+			break;
+		}
+		msg = (struct IPC_MESSAGE *) (element->data);
+
+		diff = 0;
+		if (msg->msg_buf ) {
+			diff = (char*)msg->msg_body - (char*)msg->msg_buf;
+		}
+		if ( diff < (int)sizeof(struct SOCKET_MSG_HEAD) ) {
+			/* either we don't have msg->msg_buf set
+			 * or we don't have enough bytes for socket head
+			 * we delete this message and creates
+			 * a new one and delete the old one
+			 */
+
+			newmsg= socket_message_new(ch, msg->msg_len);
+			if (newmsg == NULL) {
+				cl_log(LOG_ERR, "socket_resume_io_write: "
+					"allocating memory for new ipc msg failed");
+                		return IPC_FAIL;
+			}
+
+                	memcpy(newmsg->msg_body, msg->msg_body, msg->msg_len);
+                	oldmsg = msg;
+			msg = newmsg;
+		}
+
+                head.msg_len = msg->msg_len;
+		head.magic = HEADMAGIC;
+		memcpy(msg->msg_buf, &head, sizeof(struct SOCKET_MSG_HEAD));
+
+		if (ch->bytes_remaining == 0) {
+			/*we start to send a new message*/
+#ifdef IPC_TIME_DEBUG
+			ipc_time_debug(ch, msg, MSGPOS_SEND);
+#endif
+			bytes_remaining = msg->msg_len + ch->msgpad;
+			p = msg->msg_buf;
+		} else {
+			bytes_remaining = ch->bytes_remaining;
+			p = ((char*)msg->msg_buf) + msg->msg_len + ch->msgpad
+				- bytes_remaining;
+
+		}
+
+		sendrc = 0;
+
+                do {
+#if HB_IPC_METHOD == HB_IPC_STREAM
+			struct strbuf d;
+			int msglen, putmsgrc;
+#endif
+
+                        CHANAUDIT(ch);
+
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+			sendrc = send(conn_info->s, p
+			,       bytes_remaining, (MSG_DONTWAIT|MSG_NOSIGNAL));
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+			d.maxlen = 0;
+			d.len = msglen = bytes_remaining;
+			d.buf = p;
+			putmsgrc = putmsg(conn_info->s, NULL, &d, 0);
+			sendrc = putmsgrc == 0 ? msglen : -1;
+#endif
+                        SocketIPCStats.last_send_rc = sendrc;
+                        SocketIPCStats.last_send_errno = errno;
+                        ++SocketIPCStats.send_count;
+
+			if (sendrc <= 0) {
+				break;
+			} else {
+				p = p + sendrc;
+				bytes_remaining -= sendrc;
+			}
+
+                } while(bytes_remaining > 0 );
+
+		ch->bytes_remaining = bytes_remaining;
+
+		if (sendrc < 0) {
+			switch (errno) {
+			case EAGAIN:
+				retcode = IPC_OK;
+				break;
+			case EPIPE:
+				ch->ch_status = IPC_DISC_PENDING;
+				socket_check_disc_pending(ch);
+				retcode = IPC_BROKEN;
+				break;
+			default:
+				cl_perror("socket_resume_io_write"
+					  ": send2 bad errno");
+				ch->ch_status = IPC_DISCONNECT;
+				retcode = IPC_FAIL;
+				break;
+			}
+			break;
+		} else {
+			int orig_qlen;
+
+			CHECKFOO(3,ch, msg, SavedSentBody, "sent message")
+
+			if (oldmsg) {
+		                if (msg->msg_done != NULL) {
+                                	msg->msg_done(msg);
+                        	}
+				msg=oldmsg;
+			}
+
+			if(ch->bytes_remaining ==0) {
+				ch->send_queue->queue = g_list_remove(ch->send_queue->queue,	msg);
+				if (msg->msg_done != NULL) {
+					msg->msg_done(msg);
+				}
+
+				SocketIPCStats.nsent++;
+				orig_qlen = ch->send_queue->current_qlen--;
+				socket_check_flow_control(ch, orig_qlen, orig_qlen -1 );
+				(*nmsg)++;
+			}
+		}
+	}
+	CHANAUDIT(ch);
+	if (retcode != IPC_OK) {
+		return retcode;
+	}
+	return IPC_ISRCONN(ch) ? IPC_OK : IPC_BROKEN;
+}
+
+static int
+socket_resume_io(struct IPC_CHANNEL *ch)
+{
+	int		rc1 = IPC_OK;
+	int		rc2 = IPC_OK;
+	int		nwmsg = 1;
+	int		nbytes_r = 1;
+	gboolean	OKonce = FALSE;
+
+	CHANAUDIT(ch);
+	if (!IPC_ISRCONN(ch)) {
+		return IPC_BROKEN;
+	}
+
+	do {
+		if (nbytes_r > 0) {
+			rc1 = socket_resume_io_read(ch, &nbytes_r, FALSE);
+		}
+		if (nwmsg > 0) {
+			nwmsg = 0;
+			rc2 = socket_resume_io_write(ch, &nwmsg);
+		}
+		if (rc1 == IPC_OK || rc2 == IPC_OK) {
+			OKonce = TRUE;
+		}
+	} while ((nbytes_r > 0  || nwmsg > 0) && IPC_ISRCONN(ch));
+
+	if (IPC_ISRCONN(ch)) {
+		if (rc1 != IPC_OK) {
+			cl_log(LOG_ERR
+			       ,	"socket_resume_io_read() failure");
+		}
+		if (rc2 != IPC_OK && IPC_CONNECT == ch->ch_status) {
+			cl_log(LOG_ERR
+			,	"socket_resume_io_write() failure");
+		}
+	} else {
+		return (OKonce ? IPC_OK : IPC_BROKEN);
+	}
+
+	return (rc1 != IPC_OK ? rc1 : rc2);
+}
+
+static int
+socket_get_recv_fd(struct IPC_CHANNEL *ch)
+{
+	struct SOCKET_CH_PRIVATE* chp = ch ->ch_private;
+
+	return (chp == NULL ? -1 : chp->s);
+}
+
+static int
+socket_get_send_fd(struct IPC_CHANNEL *ch)
+{
+	return socket_get_recv_fd(ch);
+}
+
+static void
+socket_adjust_buf(struct IPC_CHANNEL *ch, int optname, unsigned q_len)
+{
+	const char *direction = optname == SO_SNDBUF ? "snd" : "rcv";
+	int fd = socket_get_send_fd(ch);
+	unsigned byte;
+
+	/* Arbitrary scaling.
+	 * DEFAULT_MAX_QLEN is 64, default socket buf is often 64k to 128k,
+	 * at least on those linux I checked.
+	 * Keep that ratio, and allow for some overhead. */
+	if (q_len == 0)
+		/* client does not want anything,
+		 * reduce system buffers as well */
+		byte = 4096;
+	else if (q_len < 512)
+		byte = (32 + q_len) * 1024;
+	else
+		byte = q_len * 1024;
+
+	if (0 == setsockopt(fd, SOL_SOCKET, optname, &byte, sizeof(byte))) {
+		if (debug_level > 1) {
+			cl_log(LOG_DEBUG, "adjusted %sbuf size to %u",
+					direction, byte);
+		}
+	} else {
+		/* If this fails, you may need to adjust net.core.rmem_max,
+		 * ...wmem_max, or equivalent */
+		cl_log(LOG_WARNING, "adjust %sbuf size to %u failed: %s",
+			direction, byte, strerror(errno));
+	}
+}
+
+static int
+socket_set_send_qlen (struct IPC_CHANNEL* ch, int q_len)
+{
+  /* This seems more like an assertion failure than a normal error */
+  if (ch->send_queue == NULL) {
+    return IPC_FAIL;
+  }
+  socket_adjust_buf(ch, SO_SNDBUF, q_len);
+  ch->send_queue->max_qlen = q_len;
+  return IPC_OK;
+}
+
+static int
+socket_set_recv_qlen (struct IPC_CHANNEL* ch, int q_len)
+{
+  /* This seems more like an assertion failure than a normal error */
+  if (ch->recv_queue == NULL) {
+    return IPC_FAIL;
+  }
+  socket_adjust_buf(ch, SO_RCVBUF, q_len);
+  ch->recv_queue->max_qlen = q_len;
+  return IPC_OK;
+}
+
+static int ipcmsg_count_allocated = 0;
+static int ipcmsg_count_freed = 0;
+void socket_ipcmsg_dump_stats(void);
+void
+socket_ipcmsg_dump_stats(void) {
+	cl_log(LOG_INFO, "ipcsocket ipcmsg allocated=%d, freed=%d, diff=%d",
+	       ipcmsg_count_allocated,
+	       ipcmsg_count_freed,
+	       ipcmsg_count_allocated - ipcmsg_count_freed);
+}
+
+static void
+socket_del_ipcmsg(IPC_Message* m)
+{
+	if (m == NULL) {
+		cl_log(LOG_ERR, "socket_del_ipcmsg:"
+		       "msg is NULL");
+		return;
+	}
+
+	if (m->msg_body) {
+		memset(m->msg_body, 0, m->msg_len);
+	}
+	if (m->msg_buf) {
+		g_free(m->msg_buf);
+	}
+
+	memset(m, 0, sizeof(*m));
+	g_free(m);
+
+	ipcmsg_count_freed ++;
+}
+
+static IPC_Message*
+socket_new_ipcmsg(IPC_Channel* ch, const void* data, int len, void* private)
+{
+	IPC_Message*	hdr;
+
+	if (ch == NULL || len < 0) {
+		cl_log(LOG_ERR, "socket_new_ipcmsg:"
+		       " invalid parameter");
+		return NULL;
+	}
+
+	if (ch->msgpad > MAX_MSGPAD) {
+		cl_log(LOG_ERR, "socket_new_ipcmsg: too many pads "
+		       "something is wrong");
+		return NULL;
+	}
+
+	hdr = ipcmsg_new(ch, data, len, private, socket_del_ipcmsg);
+
+	if (hdr) ipcmsg_count_allocated ++;
+
+	return hdr;
+}
+
+static
+struct IPC_MESSAGE *
+ipcmsg_new(struct IPC_CHANNEL * ch, const void* data, int len, void* private,
+	DelProc delproc)
+{
+	struct IPC_MESSAGE * hdr;
+	char*	copy = NULL;
+	char*	buf;
+	char*	body;
+
+	if ((hdr = g_new(struct IPC_MESSAGE, 1))  == NULL) {
+		return NULL;
+	}
+	memset(hdr, 0, sizeof(*hdr));
+
+	if (len > 0) {
+		if ((copy = (char*)g_malloc(ch->msgpad + len)) == NULL) {
+			g_free(hdr);
+			return NULL;
+		}
+		if (data) {
+			memcpy(copy + ch->msgpad, data, len);
+		}
+		buf = copy;
+		body = copy + ch->msgpad;;
+	} else {
+		len = 0;
+		buf = body = NULL;
+	}
+	hdr->msg_len = len;
+	hdr->msg_buf = buf;
+	hdr->msg_body = body;
+	hdr->msg_ch = ch;
+	hdr->msg_done = delproc;
+	hdr->msg_private = private;
+
+	return hdr;
+}
+
+static int
+socket_get_chan_status(IPC_Channel* ch)
+{
+	socket_resume_io(ch);
+	return ch->ch_status;
+}
+
+/* socket object of the function table */
+static struct IPC_WAIT_OPS socket_wait_ops = {
+  socket_destroy_wait_conn,
+  socket_wait_selectfd,
+  socket_accept_connection,
+};
+
+/*
+ * create a new ipc queue whose length = 0 and inner queue = NULL.
+ * return the pointer to a new ipc queue or NULL is the queue can't be created.
+ */
+
+static struct IPC_QUEUE*
+socket_queue_new(void)
+{
+  struct IPC_QUEUE *temp_queue;
+
+  /* temp queue with length = 0 and inner queue = NULL. */
+  temp_queue =  g_new(struct IPC_QUEUE, 1);
+  temp_queue->current_qlen = 0;
+  temp_queue->max_qlen = DEFAULT_MAX_QLEN;
+  temp_queue->queue = NULL;
+  temp_queue->last_maxqlen_warn = 0;
+  temp_queue->maxqlen_cnt = 0;
+  return temp_queue;
+}
+
+/*
+ * socket_wait_conn_new:
+ * Called by ipc_wait_conn_constructor to get a new socket
+ * waiting connection.
+ * (better explanation of this role might be nice)
+ *
+ * Parameters :
+ *     ch_attrs (IN) the attributes used to create this connection.
+ *
+ * Return :
+ *	the pointer to the new waiting connection or NULL if the connection
+ *	can't be created.
+ *
+ * NOTE :
+ *   for domain socket implementation, the only attribute needed is path name.
+ *	so the user should
+ *   create the hash table like this:
+ *     GHashTable * attrs;
+ *     attrs = g_hash_table_new(g_str_hash, g_str_equal);
+ *     g_hash_table_insert(attrs, PATH_ATTR, path_name);
+ *     here PATH_ATTR is defined as "path".
+ *
+ * NOTE :
+ *   The streams implementation uses "Streams Programming Guide", Solaris 8,
+ *   as its guide (sample code near end of "Configuration" chapter 11).
+ */
+struct IPC_WAIT_CONNECTION *
+socket_wait_conn_new(GHashTable *ch_attrs)
+{
+  struct IPC_WAIT_CONNECTION * temp_ch;
+  char *path_name;
+  char *mode_attr;
+  int s, flags;
+  struct SOCKET_WAIT_CONN_PRIVATE *wait_private;
+  mode_t s_mode;
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+  struct sockaddr_un my_addr;
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+  int pipefds[2];
+#endif
+
+  path_name = (char *) g_hash_table_lookup(ch_attrs, IPC_PATH_ATTR);
+  mode_attr = (char *) g_hash_table_lookup(ch_attrs, IPC_MODE_ATTR);
+
+  if (mode_attr != NULL) {
+    s_mode = (mode_t)strtoul((const char *)mode_attr, NULL, 8);
+  } else {
+    s_mode = 0777;
+  }
+  if (path_name == NULL) {
+    return NULL;
+  }
+
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+  /* prepare the unix domain socket */
+  if ((s = socket(AF_LOCAL, SOCK_STREAM, 0)) == -1) {
+    cl_perror("socket_wait_conn_new: socket() failure");
+    return NULL;
+  }
+
+  if (unlink(path_name) < 0 && errno != ENOENT) {
+	  cl_perror("socket_wait_conn_new: unlink failure(%s)",
+		    path_name);
+  }
+  memset(&my_addr, 0, sizeof(my_addr));
+  my_addr.sun_family = AF_LOCAL;         /* host byte order */
+
+  if (strlen(path_name) >= sizeof(my_addr.sun_path)) {
+    close(s);
+    return NULL;
+  }
+
+  strncpy(my_addr.sun_path, path_name, sizeof(my_addr.sun_path));
+
+  if (bind(s, (struct sockaddr *)&my_addr, sizeof(my_addr)) == -1) {
+    cl_perror("socket_wait_conn_new: trying to create in %s bind:"
+    ,	path_name);
+    close(s);
+    return NULL;
+  }
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+  /* Set up the communication channel the clients will use to us (server) */
+  if (pipe(pipefds) == -1) {
+    cl_perror("pipe() failure");
+    return NULL;
+  }
+
+  /* Let clients have unique connections to us */
+  if (ioctl(pipefds[1], I_PUSH, "connld") == -1) {
+    cl_perror("ioctl(%d, I_PUSH, \"connld\") failure", pipefds[1]);
+    return NULL;
+  }
+
+  if (unlink(path_name) < 0 && errno != ENOENT) {
+	  cl_perror("socket_wait_conn_new: unlink failure(%s)",
+		    path_name);
+  }
+
+  if (mkfifo(path_name, s_mode) == -1) {
+    cl_perror("socket_wait_conn_new: mkfifo(%s, ...) failure", path_name);
+    return NULL;
+  }
+
+  if (fattach(pipefds[1], path_name) == -1) {
+    cl_perror("socket_wait_conn_new: fattach(..., %s) failure", path_name);
+    return NULL;
+  }
+
+  /* the pseudo-socket is the other part of the pipe */
+  s = pipefds[0];
+#endif
+
+  /* Change the permission of the socket */
+  if (chmod(path_name,s_mode) < 0) {
+    cl_perror("socket_wait_conn_new: failure trying to chmod %s"
+    ,	path_name);
+    close(s);
+    return NULL;
+  }
+
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+  /* listen to the socket */
+  if (listen(s, MAX_LISTEN_NUM) == -1) {
+    cl_perror("socket_wait_conn_new: listen(MAX_LISTEN_NUM)");
+    close(s);
+    return NULL;
+  }
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+
+#endif
+
+  flags = fcntl(s, F_GETFL);
+  if (flags == -1) {
+    cl_perror("socket_wait_conn_new: cannot read file descriptor flags");
+    close(s);
+    return NULL;
+  }
+  flags |= O_NONBLOCK;
+  if (fcntl(s, F_SETFL, flags) < 0) {
+    cl_perror("socket_wait_conn_new: cannot set O_NONBLOCK");
+    close(s);
+    return NULL;
+  }
+
+  wait_private =  g_new(struct SOCKET_WAIT_CONN_PRIVATE, 1);
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+  wait_private->s = s;
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+  wait_private->pipefds[0] = pipefds[0];
+  wait_private->pipefds[1] = pipefds[1];
+#endif
+  strncpy(wait_private->path_name, path_name, sizeof(wait_private->path_name));
+  temp_ch = g_new(struct IPC_WAIT_CONNECTION, 1);
+  temp_ch->ch_private = (void *) wait_private;
+  temp_ch->ch_status = IPC_WAIT;
+  temp_ch->ops = (struct IPC_WAIT_OPS *)&socket_wait_ops;
+
+  return temp_ch;
+}
+
+/*
+ * will be called by ipc_channel_constructor to create a new socket channel.
+ * parameters :
+ *      attrs (IN) the hash table of the attributes used to create this channel.
+ *
+ * return:
+ *      the pointer to the new waiting channel or NULL if the channel can't be created.
+*/
+
+struct IPC_CHANNEL *
+socket_client_channel_new(GHashTable *ch_attrs) {
+  char *path_name;
+  int sockfd;
+
+  /*
+   * I don't really understand why the client and the server use different
+   * parameter names...
+   *
+   * It's a really bad idea to store both integers and strings
+   * in the same table.
+   *
+   * Maybe we need an internal function with a different set of parameters?
+   */
+
+  /*
+   * if we want to seperate them. I suggest
+   * <client side>
+   * user call ipc_channel_constructor(ch_type,attrs) to create a new channel.
+   * ipc_channel_constructor() call socket_channel_new(GHashTable*)to
+   * create a new socket channel.
+   * <server side>
+   * wait_conn->accept_connection() will call another function to create a
+   * new channel.  This function will take socketfd as the parameter to
+   * create a socket channel.
+   */
+
+  path_name = (char *) g_hash_table_lookup(ch_attrs, IPC_PATH_ATTR);
+  if (path_name == NULL) {
+	return NULL;
+  }
+
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+    /* prepare the socket */
+    if ((sockfd = socket(AF_LOCAL, SOCK_STREAM, 0)) == -1) {
+      cl_perror("socket_client_channel_new: socket");
+      return NULL;
+    }
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+    sockfd = open(path_name, O_RDWR|O_NONBLOCK);
+    if (sockfd == -1) {
+      cl_perror("socket_client_channel_new: open(%s, ...) failure", path_name);
+      return NULL;
+    }
+#endif
+
+	if (client_channel_new_auth(sockfd) < 0) {
+		close(sockfd);
+		return NULL;
+	}
+  return channel_new(sockfd, IPC_CLIENT, path_name);
+}
+
+static
+int client_channel_new_auth(int sockfd) {
+#ifdef USE_BINDSTAT_CREDS
+  char rand_id[16];
+  char uuid_str_tmp[40];
+  struct sockaddr_un sock_addr;
+
+  /* Prepare the socket */
+  memset(&sock_addr, 0, sizeof(sock_addr));
+  sock_addr.sun_family = AF_UNIX;
+
+  /* make sure socket paths never clash */
+  uuid_generate(rand_id);
+  uuid_unparse(rand_id, uuid_str_tmp);
+
+  snprintf(sock_addr.sun_path, sizeof(sock_addr.sun_path),
+	   "%s/%s", HA_VARLIBHBDIR, uuid_str_tmp);
+
+  unlink(sock_addr.sun_path);
+  if(bind(sockfd, (struct sockaddr*)&sock_addr, SUN_LEN(&sock_addr)) < 0) {
+	  perror("Client bind() failure");
+	  return 0;
+  }
+#endif
+
+  return 0;
+}
+
+static
+struct IPC_CHANNEL *
+socket_server_channel_new(int sockfd) {
+	return channel_new(sockfd, IPC_SERVER, "?");
+}
+
+static
+struct IPC_CHANNEL *
+channel_new(int sockfd, int conntype, const char *path_name) {
+  struct IPC_CHANNEL * temp_ch;
+  struct SOCKET_CH_PRIVATE* conn_info;
+  int flags;
+
+  if (path_name == NULL || strlen(path_name) >= sizeof(conn_info->path_name)) {
+	return NULL;
+  }
+
+  temp_ch = g_new(struct IPC_CHANNEL, 1);
+  if (temp_ch == NULL) {
+	  cl_log(LOG_ERR, "channel_new: allocating memory for channel failed");
+	  return NULL;
+  }
+  memset(temp_ch, 0, sizeof(struct IPC_CHANNEL));
+
+  conn_info = g_new(struct SOCKET_CH_PRIVATE, 1);
+
+  flags = fcntl(sockfd, F_GETFL);
+  if (flags == -1) {
+	  cl_perror("channel_new: cannot read file descriptor flags");
+	  g_free(conn_info); conn_info = NULL;
+	  g_free(temp_ch);
+	  if (conntype == IPC_CLIENT) close(sockfd);
+	  return NULL;
+  }
+  flags |= O_NONBLOCK;
+  if (fcntl(sockfd, F_SETFL, flags) < 0) {
+	  cl_perror("channel_new: cannot set O_NONBLOCK");
+	  g_free(conn_info); conn_info = NULL;
+	  g_free(temp_ch);
+	  if (conntype == IPC_CLIENT) close(sockfd);
+	  return NULL;
+  }
+
+  conn_info->s = sockfd;
+  conn_info->remaining_data = 0;
+  conn_info->buf_msg = NULL;
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+  conn_info->peer_addr = NULL;
+#endif
+  strncpy(conn_info->path_name, path_name, sizeof(conn_info->path_name));
+
+#ifdef DEBUG
+  cl_log(LOG_INFO, "Initializing socket %d to DISCONNECT", sockfd);
+#endif
+  temp_ch->ch_status = IPC_DISCONNECT;
+  temp_ch->ch_private = (void*) conn_info;
+  temp_ch->ops = (struct IPC_OPS *)&socket_ops;
+  temp_ch->msgpad = sizeof(struct SOCKET_MSG_HEAD);
+  temp_ch->bytes_remaining = 0;
+  temp_ch->should_send_block = FALSE;
+  temp_ch->should_block_fail = TRUE;
+  temp_ch->send_queue = socket_queue_new();
+  temp_ch->recv_queue = socket_queue_new();
+  temp_ch->pool = NULL;
+  temp_ch->high_flow_mark = temp_ch->send_queue->max_qlen;
+  temp_ch->low_flow_mark = -1;
+  temp_ch->conntype = conntype;
+  temp_ch->refcount = 0;
+  temp_ch->farside_uid = -1;
+  temp_ch->farside_gid = -1;
+
+  return temp_ch;
+}
+
+/*
+ * Create a new pair of pre-connected IPC channels similar to
+ * the result of pipe(2), or socketpair(2).
+ */
+
+int
+ipc_channel_pair(IPC_Channel* channels[2])
+{
+	int	sockets[2];
+	int	rc;
+	int	j;
+	const char *pname;
+
+#if HB_IPC_METHOD == HB_IPC_SOCKET
+	pname = "[socketpair]";
+
+	if ((rc = socketpair(AF_LOCAL, SOCK_STREAM, 0, sockets)) < 0) {
+		return IPC_FAIL;
+	}
+#elif HB_IPC_METHOD == HB_IPC_STREAM
+	pname = "[pipe]";
+
+	if ((rc = pipe(sockets)) < 0) {
+		return IPC_FAIL;
+	}
+	rc = 0;
+	for (j=0; j < 2; ++j) {
+		if (fcntl(sockets[j], F_SETFL, O_NONBLOCK) < 0) {
+			cl_perror("ipc_channel_pair: cannot set O_NONBLOCK");
+			rc = -1;
+		}
+	}
+	if (rc < 0) {
+		close(sockets[0]);
+		close(sockets[1]);
+		return IPC_FAIL;
+	}
+#endif
+
+	if ((channels[0] = socket_server_channel_new(sockets[0])) == NULL) {
+		close(sockets[0]);
+		close(sockets[1]);
+		return IPC_FAIL;
+	}
+	if ((channels[1] = socket_server_channel_new(sockets[1])) == NULL) {
+		close(sockets[0]);
+		close(sockets[1]);
+		channels[0]->ops->destroy(channels[0]);
+		return IPC_FAIL;
+	}
+	for (j=0; j < 2; ++j) {
+  		struct SOCKET_CH_PRIVATE* p = channels[j]->ch_private;
+		channels[j]->ch_status = IPC_CONNECT;
+		channels[j]->conntype = IPC_PEER;
+		/* Valid, but not terribly meaningful */
+		channels[j]->farside_pid = getpid();
+  		strncpy(p->path_name, pname, sizeof(p->path_name));
+	}
+
+	return IPC_OK;
+}
+
+/* brief free the memory space allocated to msg and destroy msg. */
+
+static void
+socket_free_message(struct IPC_MESSAGE * msg) {
+#if 0
+	memset(msg->msg_body, 0xff, msg->msg_len);
+#endif
+       if (msg->msg_buf) {
+               g_free(msg->msg_buf);
+       } else {
+               g_free(msg->msg_body);
+       }
+#if 0
+	memset(msg, 0xff, sizeof(*msg));
+#endif
+	g_free((void *)msg);
+}
+
+/*
+ * create a new ipc message whose msg_body's length is msg_len.
+ *
+ * parameters :
+ *       msg_len (IN) the length of this message body in this message.
+ *
+ * return :
+ *       the pointer to the new message or NULL if the message can't be created.
+ */
+
+static struct IPC_MESSAGE*
+socket_message_new(struct IPC_CHANNEL *ch, int msg_len)
+{
+	return ipcmsg_new(ch, NULL, msg_len, NULL, socket_free_message);
+}
+
+/***********************************************************************
+ *
+ * IPC authentication schemes...  More machine dependent than
+ * we'd like, but don't know any better way...
+ *
+ ***********************************************************************/
+
+static int
+verify_creds(struct IPC_AUTH *auth_info, uid_t uid, gid_t gid)
+{
+	int ret = IPC_FAIL;
+
+	if (!auth_info || (!auth_info->uid && !auth_info->gid)) {
+		return IPC_OK;
+	}
+	if (	auth_info->uid
+	&&	(g_hash_table_lookup(auth_info->uid
+		,	GUINT_TO_POINTER((guint)uid)) != NULL)) {
+		ret = IPC_OK;
+	} else if (auth_info->gid
+	&&	(g_hash_table_lookup(auth_info->gid
+		,	GUINT_TO_POINTER((guint)gid)) != NULL)) {
+		ret = IPC_OK;
+  	}
+	return ret;
+}
+
+/***********************************************************************
+ * SO_PEERCRED VERSION... (Linux)
+ ***********************************************************************/
+
+#ifdef USE_SO_PEERCRED
+/* verify the authentication information. */
+static int
+socket_verify_auth(struct IPC_CHANNEL* ch, struct IPC_AUTH * auth_info)
+{
+	struct SOCKET_CH_PRIVATE *	conn_info;
+	int				ret = IPC_FAIL;
+	struct ucred			cred;
+	socklen_t			n = sizeof(cred);
+
+	if (ch == NULL || ch->ch_private == NULL) {
+		return IPC_FAIL;
+	}
+	if (auth_info == NULL
+	||	(auth_info->uid == NULL && auth_info->gid == NULL)) {
+		ret = IPC_OK;    /* no restriction for authentication */
+	  }
+
+	/* Get the credential information for our peer */
+	conn_info = (struct SOCKET_CH_PRIVATE *) ch->ch_private;
+	if (getsockopt(conn_info->s, SOL_SOCKET, SO_PEERCRED, &cred, &n) != 0
+	||	(size_t)n != sizeof(cred)) {
+		return ret;
+	}
+
+	ch->farside_uid = cred.uid;
+	ch->farside_gid = cred.gid;
+	if (ret == IPC_OK) {
+		return ret;
+	}
+#if 0
+	cl_log(LOG_DEBUG, "SO_PEERCRED returned [%d, (%ld:%ld)]"
+	,	cred.pid, (long)cred.uid, (long)cred.uid);
+	cl_log(LOG_DEBUG, "Verifying authentication: cred.uid=%d cred.gid=%d"
+	,	cred.uid, cred.gid);
+	cl_log(LOG_DEBUG, "Verifying authentication: uidptr=0x%lx gidptr=0x%lx"
+	,	(unsigned long)auth_info->uid
+	,	(unsigned long)auth_info->gid);
+#endif
+	/* verify the credential information. */
+	return verify_creds(auth_info, cred.uid, cred.gid);
+}
+
+/* get farside pid for our peer process */
+
+static
+pid_t
+socket_get_farside_pid(int sockfd)
+{
+  socklen_t n;
+  struct ucred *cred;
+  pid_t f_pid;
+
+  /* Get the credential information from peer */
+  n = sizeof(struct ucred);
+  cred = g_new(struct ucred, 1);
+  if (getsockopt(sockfd, SOL_SOCKET, SO_PEERCRED, cred, &n) != 0) {
+    g_free(cred);
+    return -1;
+  }
+
+  f_pid = cred->pid;
+  g_free(cred);
+  return f_pid;
+}
+#endif /* SO_PEERCRED version */
+
+#ifdef USE_GETPEEREID
+/*
+ * This is implemented in OpenBSD and FreeBSD.
+ *
+ * It's not a half-bad interface...
+ *
+ * This should probably be our standard way of doing it, and put it
+ * as a replacement library.  That would simplify things...
+ */
+
+static int
+socket_verify_auth(struct IPC_CHANNEL* ch, struct IPC_AUTH * auth_info)
+{
+	struct SOCKET_CH_PRIVATE *conn_info;
+	uid_t	euid;
+	gid_t	egid;
+	int	ret = IPC_FAIL;
+
+	if (auth_info == NULL
+	||	(auth_info->uid == NULL && auth_info->gid == NULL)) {
+		ret = IPC_OK;    /* no restriction for authentication */
+	}
+	conn_info = (struct SOCKET_CH_PRIVATE *) ch->ch_private;
+
+	if (getpeereid(conn_info->s, &euid, &egid) < 0) {
+		cl_perror("getpeereid() failure");
+		return ret;
+	}
+
+	ch->farside_uid = euid;
+	ch->farside_gid = egid;
+
+	/* verify the credential information. */
+	return verify_creds(auth_info, euid, egid);
+}
+
+static
+pid_t
+socket_get_farside_pid(int sock)
+{
+	return -1;
+}
+#endif /* USE_GETPEEREID */
+
+/***********************************************************************
+ * SCM_CREDS VERSION... (*BSD systems)
+ ***********************************************************************/
+#ifdef USE_SCM_CREDS
+/* FIXME!  Need to implement SCM_CREDS mechanism for BSD-based systems
+ * This isn't an emergency, but should be done in the future...
+ * Hint: * Postgresql does both types of authentication...
+ * see src/backend/libpq/auth.c
+ * Not clear its SO_PEERCRED implementation works though ;-)
+ */
+
+/* Done.... Haven't tested yet. */
+static int
+socket_verify_auth(struct IPC_CHANNEL* ch, struct IPC_AUTH * auth_info)
+{
+  struct msghdr msg;
+  /* Credentials structure */
+
+#define	EXTRASPACE	0
+
+#ifdef HAVE_STRUCT_CMSGCRED
+	/* FreeBSD */
+  typedef struct cmsgcred Cred;
+#	define crRuid	cmcred_uid
+#	define crEuid	cmcred_euid
+#	define crRgid	cmcred_gid
+#	define crEgid	cmcred_groups[0]	/* Best guess */
+#	define crpid	cmcred_pid
+#	define crngrp	cmcred_ngroups
+#	define crgrps	cmcred_groups
+
+#elif HAVE_STRUCT_FCRED
+	/* Stevens' book */
+  typedef struct fcred Cred;
+#	define crRuid	fc_uid
+#	define crRgid	fc_rgid
+#	define crEgid	fc_gid
+#	define crngrp	fc_ngroups
+#	define crgrps	fc_groups
+
+#elif HAVE_STRUCT_SOCKCRED
+	/* NetBSD */
+  typedef struct sockcred Cred;
+#	define crRuid	sc_uid
+#	define crEuid	sc_euid
+#	define crRgid	sc_gid
+#	define crEgid	sc_egid
+#	define crngrp	sc_ngroups
+#	define crgrps	sc_groups
+#	undef EXTRASPACE
+#	define EXTRASPACE	SOCKCREDSIZE(ngroups)
+
+#elif HAVE_STRUCT_CRED
+  typedef struct cred Cred;
+#define cruid c_uid
+
+#elif HAVE_STRUCT_UCRED
+ typedef struct ucred Cred;
+
+ /* reuse this define for the moment */
+#  if HAVE_STRUCT_UCRED_DARWIN
+#	define crEuid	cr_uid
+#	define crEgid	cr_groups[0]		/* Best guess */
+#	define crgrps	cr_groups
+#	define crngrp	cr_ngroups
+#  else
+#	define crEuid	c_uid
+#	define crEgid	c_gid
+#  endif
+#else
+#	error "No credential type found!"
+#endif
+
+  struct SOCKET_CH_PRIVATE *conn_info;
+  int ret = IPC_FAIL;
+  char         buf;
+
+  /* Compute size without padding */
+  #define CMSGSIZE	(sizeof(struct cmsghdr)+(sizeof(Cred))+EXTRASPACE)
+
+  union {
+  	char		mem[CMSGSIZE];
+	struct cmsghdr	hdr;
+	Cred		credu;
+  }cmsgmem;
+  Cred	   cred;
+
+  /* Point to start of first structure */
+  struct cmsghdr *cmsg = &cmsgmem.hdr;
+
+  if (auth_info == NULL
+  ||	(auth_info->uid == NULL && auth_info->gid == NULL)) {
+    ret = IPC_OK;    /* no restriction for authentication */
+  }
+  conn_info = (struct SOCKET_CH_PRIVATE *) ch->ch_private;
+
+  memset(&msg, 0, sizeof(msg));
+  msg.msg_iov =  g_new(struct iovec, 1);
+  msg.msg_iovlen = 1;
+  msg.msg_control = (char *) cmsg;
+  msg.msg_controllen = CMSGSIZE;
+  memset(cmsg, 0, sizeof(cmsgmem));
+
+  /*
+   * The one character which is received here is not meaningful; its
+   * purpose is only to make sure that recvmsg() blocks long enough for
+   * the other side to send its credentials.
+   */
+  msg.msg_iov->iov_base = &buf;
+  msg.msg_iov->iov_len = 1;
+
+  if (recvmsg(conn_info->s, &msg, 0) < 0
+      || cmsg->cmsg_len < CMSGSIZE
+      || cmsg->cmsg_type != SCM_CREDS) {
+      cl_perror("can't get credential information from peer");
+      return ret;
+    }
+
+  /* Avoid alignment issues - just copy it! */
+  memcpy(&cred, CMSG_DATA(cmsg), sizeof(cred));
+
+  ch->farside_uid = cred.crEuid;
+  ch->farside_gid = cred.crEgid;
+  if (ret == IPC_OK) {
+      return ret;
+  }
+
+  /* verify the credential information. */
+  return verify_creds(auth_info, cred.crEuid, cred.crEgid);
+}
+
+/*
+ * FIXME!  Need to implement SCM_CREDS mechanism for BSD-based systems
+ * this is similar to the SCM_CREDS mechanism for verify_auth() function.
+ * here we just want to get the pid of the other side from the credential
+ * information.
+ */
+
+static
+pid_t
+socket_get_farside_pid(int sock)
+{
+	/* FIXME! */
+	return -1;
+}
+#endif /* SCM_CREDS version */
+
+/***********************************************************************
+ * Bind/Stat VERSION... (Supported on OSX/Darwin and 4.3+BSD at least...)
+ *
+ * This is for use on systems such as OSX-Darwin where
+ *   none of the other options is available.
+ *
+ * This implementation has been adapted from "Advanced Programming
+ *   in the Unix Environment", Section 15.5.2, by W. Richard Stevens.
+ *
+ */
+#ifdef USE_BINDSTAT_CREDS
+
+static int
+socket_verify_auth(struct IPC_CHANNEL* ch, struct IPC_AUTH * auth_info)
+{
+	int len = 0;
+	int ret = IPC_FAIL;
+	struct stat stat_buf;
+	struct sockaddr_un *peer_addr = NULL;
+	struct SOCKET_CH_PRIVATE *ch_private = NULL;
+
+	if(ch != NULL) {
+		ch_private = (struct SOCKET_CH_PRIVATE *)(ch->ch_private);
+		if(ch_private != NULL) {
+			peer_addr = ch_private->peer_addr;
+		}
+	}
+
+	if(ch == NULL) {
+		cl_log(LOG_ERR, "No channel to authenticate");
+		return IPC_FAIL;
+
+	} else if (auth_info == NULL
+	    ||	(auth_info->uid == NULL && auth_info->gid == NULL)) {
+		ret = IPC_OK;    /* no restriction for authentication */
+
+	}
+
+	if(ch_private == NULL) {
+		cl_log(LOG_ERR, "No channel private data available");
+		return ret;
+
+	} else if(peer_addr == NULL) {
+		cl_log(LOG_ERR, "No peer information available");
+		return ret;
+	}
+
+	len = SUN_LEN(peer_addr);
+
+	if(len < 1) {
+		cl_log(LOG_ERR, "No peer information available");
+		return ret;
+	}
+	peer_addr->sun_path[len] = 0;
+	stat(peer_addr->sun_path, &stat_buf);
+
+	ch->farside_uid = stat_buf.st_uid;
+	ch->farside_gid = stat_buf.st_gid;
+	if (ret == IPC_OK) {
+		return ret;
+	}
+
+	if ((auth_info->uid == NULL || g_hash_table_size(auth_info->uid) == 0)
+	    && auth_info->gid != NULL
+	    && g_hash_table_size(auth_info->gid) != 0) {
+		cl_log(LOG_WARNING,
+		       "GID-Only IPC security is not supported"
+		       " on this platform.");
+		return IPC_BROKEN;
+	}
+
+	/* verify the credential information. */
+	return verify_creds(auth_info, stat_buf.st_uid, stat_buf.st_gid);
+}
+
+static pid_t
+socket_get_farside_pid(int sock)
+{
+	return -1;
+}
+#endif /* Bind/stat version */
+
+/***********************************************************************
+ * USE_STREAM_CREDS VERSION... (e.g. Solaris pre-10)
+ ***********************************************************************/
+#ifdef USE_STREAM_CREDS
+static int
+socket_verify_auth(struct IPC_CHANNEL* ch, struct IPC_AUTH * auth_info)
+{
+	struct SOCKET_CH_PRIVATE *conn_info;
+
+	if (ch == NULL || ch->ch_private == NULL) {
+		return IPC_FAIL;
+	}
+
+	conn_info = (struct SOCKET_CH_PRIVATE *) ch->ch_private;
+
+	ch->farside_uid = conn_info->farside_uid;
+	ch->farside_gid = conn_info->farside_gid;
+
+	/* verify the credential information. */
+	return verify_creds(auth_info,
+		conn_info->farside_uid, conn_info->farside_gid);
+}
+
+static
+pid_t
+socket_get_farside_pid(int sock)
+{
+	return -1;
+}
+#endif
+
+/***********************************************************************
+ * GETPEERUCRED VERSION... (e.g. Solaris 10 upwards)
+ ***********************************************************************/
+
+#ifdef USE_GETPEERUCRED
+/* verify the authentication information. */
+static int
+socket_verify_auth(struct IPC_CHANNEL* ch, struct IPC_AUTH * auth_info)
+{
+	struct SOCKET_CH_PRIVATE *conn_info;
+	ucred_t *ucred = NULL;
+	int rc = IPC_FAIL;
+
+	if (ch == NULL || ch->ch_private == NULL) {
+		return IPC_FAIL;
+	}
+
+	conn_info = (struct SOCKET_CH_PRIVATE *) ch->ch_private;
+
+	if (auth_info == NULL
+	  || (auth_info->uid == NULL && auth_info->gid == NULL)) {
+		rc = IPC_OK;	/* no restriction for authentication */
+	}
+
+	if (getpeerucred(conn_info->s, &ucred) < 0) {
+		cl_perror("getpeereid() failure");
+		return rc;
+	}
+
+	ch->farside_uid = ucred_geteuid(ucred);
+	ch->farside_gid = ucred_getegid(ucred);
+	if (rc == IPC_OK) {
+		return rc;
+	}
+
+	/* verify the credential information. */
+	rc = verify_creds(auth_info,
+		ucred_geteuid(ucred), ucred_getegid(ucred));
+	ucred_free(ucred);
+	return rc;
+}
+
+static
+pid_t
+socket_get_farside_pid(int sockfd)
+{
+	ucred_t *ucred = NULL;
+	pid_t pid;
+
+	if (getpeerucred(sockfd, &ucred) < 0) {
+		cl_perror("getpeereid() failure");
+		return IPC_FAIL;
+	}
+
+	pid = ucred_getpid(ucred);
+
+	ucred_free(ucred);
+
+	return pid;
+}
+#endif
+
+/***********************************************************************
+ * DUMMY VERSION... (other systems...)
+ *
+ * Other options that seem to be out there include
+ * SCM_CREDENTIALS and LOCAL_CREDS
+ * There are some kludgy things you can do with SCM_RIGHTS
+ * to pass an fd which could only be opened by the user id to
+ * validate the user id, but I don't know of a similar kludge which
+ * would work for group ids.  And, even the uid one will fail
+ * if normal users are allowed to give away (chown) files.
+ *
+ * Unfortunately, this set of authentication routines have become
+ * very important to this API and its users (like heartbeat).
+ *
+ ***********************************************************************/
+
+#ifdef USE_DUMMY_CREDS
+static int
+socket_verify_auth(struct IPC_CHANNEL* ch, struct IPC_AUTH * auth_info)
+{
+	return IPC_FAIL;
+}
+
+static
+pid_t
+socket_get_farside_pid(int sock)
+{
+	return -1;
+}
+#endif /* Dummy version */
+
+/* socket object of the function table */
+static struct IPC_OPS socket_ops = {
+	socket_destroy_channel,
+	socket_initiate_connection,
+	socket_verify_auth,
+	socket_assert_auth,
+	socket_send,
+	socket_recv,
+	socket_waitin,
+	socket_waitout,
+	socket_is_message_pending,
+	socket_is_output_pending,
+	socket_resume_io,
+	socket_get_send_fd,
+	socket_get_recv_fd,
+	socket_set_send_qlen,
+	socket_set_recv_qlen,
+	socket_set_high_flow_callback,
+	socket_set_low_flow_callback,
+	socket_new_ipcmsg,
+	socket_get_chan_status,
+	socket_is_sendq_full,
+	socket_is_recvq_full,
+	socket_get_conntype,
+	socket_disconnect,
+};