<commit_before>/******************************************************************************
*******************************************************************************
**
**  Copyright (C) 2004-2006 Red Hat, Inc.  All rights reserved.
**
**  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 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 <sys/types.h>
#include <sys/un.h>
#include <inttypes.h>
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <string.h>
#include <errno.h>
#include "cnxman-socket.h"
#include "libcman.h"

/* List of saved messages */
struct saved_message
{
	struct sock_header *msg;
	struct saved_message *next;
};

struct cman_handle
{
	int magic;
	int fd;
	int zero_fd;
	void *privdata;
	int want_reply;
	cman_callback_t event_callback;
	cman_datacallback_t data_callback;
	cman_confchgcallback_t confchg_callback;

	void *reply_buffer;
	int reply_buflen;
	int reply_status;

	struct saved_message *saved_data_msg;
	struct saved_message *saved_event_msg;
	struct saved_message *saved_reply_msg;
};

#define VALIDATE_HANDLE(h) do {if (!(h) || (h)->magic != CMAN_MAGIC) {errno = EINVAL; return -1;}} while (0)

/*
 * Wait for an command/request reply.
 * Data/event messages will be queued.
 *
 */
static int wait_for_reply(struct cman_handle *h, void *msg, int max_len)
{
	int ret;

	h->want_reply = 1;
	h->reply_buffer = msg;
	h->reply_buflen = max_len;

	do
	{
		ret = cman_dispatch(h, CMAN_DISPATCH_BLOCKING | CMAN_DISPATCH_IGNORE_EVENT | CMAN_DISPATCH_IGNORE_DATA);

	} while (h->want_reply == 1 && ret >= 0);

	h->reply_buffer = NULL;
	h->reply_buflen = 0;

	/* Error in local comms */
	if (ret < 0) {
		return -1;
	}
	/* cnxman daemon returns -ve errno values on error */
	if (h->reply_status < 0) {
		errno = -h->reply_status;
		return -1;
	}
	else {
		return h->reply_status;
	}
}


static void copy_node(cman_node_t *unode, struct cl_cluster_node *knode)
{
	unode->cn_nodeid = knode->node_id;
	unode->cn_member = knode->state == NODESTATE_MEMBER?1:0;
	strcpy(unode->cn_name, knode->name);
	unode->cn_incarnation = knode->incarnation;
	unode->cn_jointime = knode->jointime;

	memset(&unode->cn_address, 0, sizeof(unode->cn_address));
	memcpy(&unode->cn_address.cna_address, knode->addr, knode->addrlen);
	unode->cn_address.cna_addrlen = knode->addrlen;
}

/* Add to a list. saved_message *m is the head of the list in the cman_handle */
static void add_to_waitlist(struct saved_message **m, struct sock_header *msg)
{
	struct saved_message *next = *m;
	struct saved_message *last = *m;
	struct saved_message *this;

	this = malloc(sizeof(struct saved_message));
	if (!this)
		return;

	this->msg = malloc(msg->length);
	if (!this->msg)
	{
		free(this);
		return;
	}

	memcpy(this->msg, msg, msg->length);
	this->next = NULL;

	if (!next)
	{
		*m = this;
		return;
	}

	for (; next; next = next->next)
	{
		last = next;
	}
	last->next = this;
}

static int process_cman_message(struct cman_handle *h, int flags, struct sock_header *msg)
{
	/* Data for us */
	if ((msg->command & CMAN_CMDMASK_CMD) == CMAN_CMD_DATA)
	{
		struct sock_data_header *dmsg = (struct sock_data_header *)msg;
		char *buf = (char *)msg;

		if (flags & CMAN_DISPATCH_IGNORE_DATA)
		{
			add_to_waitlist(&h->saved_data_msg, msg);
		}
		else
		{
			if (h->data_callback)
				h->data_callback(h, h->privdata,
						 buf+sizeof(*dmsg), msg->length-sizeof(*dmsg),
						 dmsg->port, dmsg->nodeid);
		}
		return 0;
	}

	/* Got a reply to a previous information request */
	if ((msg->command & CMAN_CMDFLAG_REPLY) && h->want_reply)
	{
		char *replybuf = (char *)msg;
		int replylen = msg->length - sizeof(struct sock_reply_header);
		struct sock_reply_header *reply = (struct sock_reply_header *)msg;

		if (flags & CMAN_DISPATCH_IGNORE_REPLY)
		{
			add_to_waitlist(&h->saved_reply_msg, msg);
			return 0;
		}

		replybuf += sizeof(struct sock_reply_header);
		if (replylen <= h->reply_buflen)
		{
			memcpy(h->reply_buffer, replybuf, replylen);
		}
		h->want_reply = 0;
		h->reply_status = reply->status;

		return 1;
	}

	/* OOB event */
	if (msg->command == CMAN_CMD_EVENT || msg->command == CMAN_CMD_CONFCHG)
	{
		if (flags & CMAN_DISPATCH_IGNORE_EVENT)
		{
			add_to_waitlist(&h->saved_event_msg, msg);
		}
		else
		{
			if (msg->command == CMAN_CMD_EVENT && h->event_callback) {
				struct sock_event_message *emsg = (struct sock_event_message *)msg;
				h->event_callback(h, h->privdata, emsg->reason, emsg->arg);
			}

			if (msg->command == CMAN_CMD_CONFCHG && h->confchg_callback)
			{
				struct sock_confchg_message *cmsg = (struct sock_confchg_message *)msg;

				h->confchg_callback(h, h->privdata,
						    cmsg->entries,cmsg->member_entries, 
						    &cmsg->entries[cmsg->member_entries], cmsg->left_entries, 
						    &cmsg->entries[cmsg->member_entries+cmsg->left_entries], cmsg->joined_entries);
			}
		}
	}

	return 0;
}

static int loopy_writev(int fd, struct iovec *iovptr, size_t iovlen)
{
	size_t byte_cnt=0;
	int len;

	while (iovlen > 0)
	{
		len = writev(fd, iovptr, iovlen);
		if (len <= 0)
			return len;

		byte_cnt += len;
		while (len >= iovptr->iov_len)
		{
			len -= iovptr->iov_len;
			iovptr++;
			iovlen--;
		}

		if (iovlen <=0 )
			break;

		iovptr->iov_base += len;
		iovptr->iov_len -= len;
	}
	return byte_cnt;
}


static int send_message(struct cman_handle *h, int msgtype, void *inbuf, int inlen)
{
	struct sock_header header;
	size_t len;
	struct iovec iov[2];
	size_t iovlen = 1;

	header.magic = CMAN_MAGIC;
	header.version = CMAN_VERSION;
	header.command = msgtype;
	header.flags = 0;
	header.length = sizeof(header) + inlen;

	iov[0].iov_len = sizeof(header);
	iov[0].iov_base = &header;
	if (inbuf)
	{
		iov[1].iov_len = inlen;
		iov[1].iov_base = inbuf;
		iovlen++;
	}

	len = loopy_writev(h->fd, iov, iovlen);
	if (len < 0)
		return len;
	return 0;
}

/* Does something similar to the ioctl calls */
static int info_call(struct cman_handle *h, int msgtype, void *inbuf, int inlen, void *outbuf, int outlen)
{
	if (send_message(h, msgtype, inbuf, inlen))
		return -1;

	return wait_for_reply(h, outbuf, outlen);
}

static cman_handle_t open_socket(const char *name, int namelen, void *privdata)
{
	struct cman_handle *h;
	struct sockaddr_un sockaddr;

	h = malloc(sizeof(struct cman_handle));
	if (!h)
		return NULL;

	h->magic = CMAN_MAGIC;
	h->privdata = privdata;
	h->event_callback = NULL;
	h->data_callback = NULL;
	h->confchg_callback = NULL;
	h->want_reply = 0;
	h->saved_data_msg = NULL;
	h->saved_event_msg = NULL;
	h->saved_reply_msg = NULL;

	h->fd = socket(PF_UNIX, SOCK_STREAM, 0);
	if (h->fd == -1)
	{
		int saved_errno = errno;
		free(h);
		errno = saved_errno;
		return NULL;
	}

	fcntl(h->fd, F_SETFD, 1); /* Set close-on-exec */
	memset(&sockaddr, 0, sizeof(sockaddr));
	memcpy(sockaddr.sun_path, name, namelen);
	sockaddr.sun_family = AF_UNIX;

	if (connect(h->fd, (struct sockaddr *) &sockaddr, sizeof(sockaddr)) < 0)
	{
		int saved_errno = errno;
		close(h->fd);
		free(h);
		errno = saved_errno;
		return NULL;
	}

	/* Get a handle on /dev/zero too. This is always active so we
	   can return it from cman_get_fd() if we have cached messages */
	h->zero_fd = open("/dev/zero", O_RDONLY);
	if (h->zero_fd < 0)
	{
		int saved_errno = errno;
		close(h->fd);
		free(h);
		h = NULL;
		errno = saved_errno;
	}

	return (cman_handle_t)h;
}

cman_handle_t cman_admin_init(void *privdata)
{
	return open_socket(ADMIN_SOCKNAME, sizeof(ADMIN_SOCKNAME), privdata);
}

cman_handle_t cman_init(void *privdata)
{
	return open_socket(CLIENT_SOCKNAME, sizeof(CLIENT_SOCKNAME), privdata);
}

int cman_finish(cman_handle_t handle)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	h->magic = 0;
	close(h->fd);
	close(h->zero_fd);
	free(h);

	return 0;
}

int cman_setprivdata(cman_handle_t *handle, void *privdata)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	h->privdata = privdata;
	return 0;
}

int cman_getprivdata(cman_handle_t *handle, void **privdata)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	*privdata = h->privdata;

	return 0;
}


int cman_start_notification(cman_handle_t handle, cman_callback_t callback)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	if (!callback)
	{
		errno = EINVAL;
		return -1;
	}
	if (info_call(h, CMAN_CMD_NOTIFY, NULL, 0, NULL, 0))
		return -1;
	h->event_callback = callback;

	return 0;
}

int cman_stop_notification(cman_handle_t handle)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	if (info_call(h, CMAN_CMD_REMOVENOTIFY, NULL, 0, NULL, 0))
		return -1;
	h->event_callback = NULL;

	return 0;
}

int cman_start_confchg(cman_handle_t handle, cman_confchgcallback_t callback)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	if (!callback)
	{
		errno = EINVAL;
		return -1;
	}
	if (info_call(h, CMAN_CMD_START_CONFCHG, NULL, 0, NULL, 0))
		return -1;
	h->confchg_callback = callback;

	return 0;
}

int cman_stop_confchg(cman_handle_t handle)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	if (info_call(h, CMAN_CMD_STOP_CONFCHG, NULL, 0, NULL, 0))
		return -1;
	h->confchg_callback = NULL;

	return 0;
}


int cman_get_fd(cman_handle_t handle)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	/* If we have saved messages then return an FD to /dev/zero which
	   will always be readable */
	if (h->saved_data_msg || h->saved_event_msg || h->saved_reply_msg)
		return h->zero_fd;
	else
		return h->fd;
}

int cman_dispatch(cman_handle_t handle, int flags)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	int len;
	int offset;
	int recv_flags = 0;
	char buf[PIPE_BUF];
	VALIDATE_HANDLE(h);

	if (!(flags & CMAN_DISPATCH_BLOCKING))
		recv_flags |= MSG_DONTWAIT;

	do
	{
		int res;
		char *bufptr = buf;
		struct sock_header *header = (struct sock_header *)buf;

		/* First, drain any waiting queues */
		if (h->saved_reply_msg && !(flags & CMAN_DISPATCH_IGNORE_REPLY))
		{
			struct saved_message *smsg = h->saved_reply_msg;

			res = process_cman_message(h, flags, smsg->msg);
			h->saved_reply_msg = smsg->next;
			len = smsg->msg->length;
			free(smsg);
			if (res)
				break;
			else
				continue;
		}
		if (h->saved_data_msg && !(flags & CMAN_DISPATCH_IGNORE_DATA))
		{
			struct saved_message *smsg = h->saved_data_msg;

			res = process_cman_message(h, flags, smsg->msg);
			h->saved_data_msg = smsg->next;
			len = smsg->msg->length;
			free(smsg);
			if (res)
				break;
			else
				continue;
		}
		if (h->saved_event_msg && !(flags & CMAN_DISPATCH_IGNORE_EVENT))
		{
			struct saved_message *smsg = h->saved_event_msg;

			res = process_cman_message(h, flags, smsg->msg);
			h->saved_event_msg = smsg->next;
			len = smsg->msg->length;
			free(smsg);
			if (res)
				break;
			else
				continue;
		}

		/* Now look for new messages */
		len = recv(h->fd, buf, sizeof(struct sock_header), recv_flags);

		if (len == 0) {
			errno = EHOSTDOWN;
			return -1;
		}

		if (len < 0 &&
		    (errno == EINTR || errno == EAGAIN))
			return 0;

		if (len < 0)
			return -1;

		offset = len;

		/* It's too big! */
		if (header->length > sizeof(buf))
		{
			bufptr = malloc(header->length);<commit_msg>Fix bug where cman_dispatch(CMAN_DISPATCH_ONE) could dispatch several messages.<commit_after>/******************************************************************************
*******************************************************************************
**
**  Copyright (C) 2004-2006 Red Hat, Inc.  All rights reserved.
**
**  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 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 <sys/types.h>
#include <sys/un.h>
#include <inttypes.h>
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <string.h>
#include <errno.h>
#include "cnxman-socket.h"
#include "libcman.h"

/* List of saved messages */
struct saved_message
{
	struct sock_header *msg;
	struct saved_message *next;
};

struct cman_handle
{
	int magic;
	int fd;
	int zero_fd;
	void *privdata;
	int want_reply;
	cman_callback_t event_callback;
	cman_datacallback_t data_callback;
	cman_confchgcallback_t confchg_callback;

	void *reply_buffer;
	int reply_buflen;
	int reply_status;

	struct saved_message *saved_data_msg;
	struct saved_message *saved_event_msg;
	struct saved_message *saved_reply_msg;
};

#define VALIDATE_HANDLE(h) do {if (!(h) || (h)->magic != CMAN_MAGIC) {errno = EINVAL; return -1;}} while (0)

/*
 * Wait for an command/request reply.
 * Data/event messages will be queued.
 *
 */
static int wait_for_reply(struct cman_handle *h, void *msg, int max_len)
{
	int ret;

	h->want_reply = 1;
	h->reply_buffer = msg;
	h->reply_buflen = max_len;

	do
	{
		ret = cman_dispatch(h, CMAN_DISPATCH_BLOCKING | CMAN_DISPATCH_IGNORE_EVENT | CMAN_DISPATCH_IGNORE_DATA);

	} while (h->want_reply == 1 && ret >= 0);

	h->reply_buffer = NULL;
	h->reply_buflen = 0;

	/* Error in local comms */
	if (ret < 0) {
		return -1;
	}
	/* cnxman daemon returns -ve errno values on error */
	if (h->reply_status < 0) {
		errno = -h->reply_status;
		return -1;
	}
	else {
		return h->reply_status;
	}
}


static void copy_node(cman_node_t *unode, struct cl_cluster_node *knode)
{
	unode->cn_nodeid = knode->node_id;
	unode->cn_member = knode->state == NODESTATE_MEMBER?1:0;
	strcpy(unode->cn_name, knode->name);
	unode->cn_incarnation = knode->incarnation;
	unode->cn_jointime = knode->jointime;

	memset(&unode->cn_address, 0, sizeof(unode->cn_address));
	memcpy(&unode->cn_address.cna_address, knode->addr, knode->addrlen);
	unode->cn_address.cna_addrlen = knode->addrlen;
}

/* Add to a list. saved_message *m is the head of the list in the cman_handle */
static void add_to_waitlist(struct saved_message **m, struct sock_header *msg)
{
	struct saved_message *next = *m;
	struct saved_message *last = *m;
	struct saved_message *this;

	this = malloc(sizeof(struct saved_message));
	if (!this)
		return;

	this->msg = malloc(msg->length);
	if (!this->msg)
	{
		free(this);
		return;
	}

	memcpy(this->msg, msg, msg->length);
	this->next = NULL;

	if (!next)
	{
		*m = this;
		return;
	}

	for (; next; next = next->next)
	{
		last = next;
	}
	last->next = this;
}

static int process_cman_message(struct cman_handle *h, int flags, struct sock_header *msg)
{
	/* Data for us */
	if ((msg->command & CMAN_CMDMASK_CMD) == CMAN_CMD_DATA)
	{
		struct sock_data_header *dmsg = (struct sock_data_header *)msg;
		char *buf = (char *)msg;

		if (flags & CMAN_DISPATCH_IGNORE_DATA)
		{
			add_to_waitlist(&h->saved_data_msg, msg);
		}
		else
		{
			if (h->data_callback)
				h->data_callback(h, h->privdata,
						 buf+sizeof(*dmsg), msg->length-sizeof(*dmsg),
						 dmsg->port, dmsg->nodeid);
		}
		return 0;
	}

	/* Got a reply to a previous information request */
	if ((msg->command & CMAN_CMDFLAG_REPLY) && h->want_reply)
	{
		char *replybuf = (char *)msg;
		int replylen = msg->length - sizeof(struct sock_reply_header);
		struct sock_reply_header *reply = (struct sock_reply_header *)msg;

		if (flags & CMAN_DISPATCH_IGNORE_REPLY)
		{
			add_to_waitlist(&h->saved_reply_msg, msg);
			return 0;
		}

		replybuf += sizeof(struct sock_reply_header);
		if (replylen <= h->reply_buflen)
		{
			memcpy(h->reply_buffer, replybuf, replylen);
		}
		h->want_reply = 0;
		h->reply_status = reply->status;

		return 1;
	}

	/* OOB event */
	if (msg->command == CMAN_CMD_EVENT || msg->command == CMAN_CMD_CONFCHG)
	{
		if (flags & CMAN_DISPATCH_IGNORE_EVENT)
		{
			add_to_waitlist(&h->saved_event_msg, msg);
		}
		else
		{
			if (msg->command == CMAN_CMD_EVENT && h->event_callback) {
				struct sock_event_message *emsg = (struct sock_event_message *)msg;
				h->event_callback(h, h->privdata, emsg->reason, emsg->arg);
			}

			if (msg->command == CMAN_CMD_CONFCHG && h->confchg_callback)
			{
				struct sock_confchg_message *cmsg = (struct sock_confchg_message *)msg;

				h->confchg_callback(h, h->privdata,
						    cmsg->entries,cmsg->member_entries, 
						    &cmsg->entries[cmsg->member_entries], cmsg->left_entries, 
						    &cmsg->entries[cmsg->member_entries+cmsg->left_entries], cmsg->joined_entries);
			}
		}
	}

	return 0;
}

static int loopy_writev(int fd, struct iovec *iovptr, size_t iovlen)
{
	size_t byte_cnt=0;
	int len;

	while (iovlen > 0)
	{
		len = writev(fd, iovptr, iovlen);
		if (len <= 0)
			return len;

		byte_cnt += len;
		while (len >= iovptr->iov_len)
		{
			len -= iovptr->iov_len;
			iovptr++;
			iovlen--;
		}

		if (iovlen <=0 )
			break;

		iovptr->iov_base += len;
		iovptr->iov_len -= len;
	}
	return byte_cnt;
}


static int send_message(struct cman_handle *h, int msgtype, void *inbuf, int inlen)
{
	struct sock_header header;
	size_t len;
	struct iovec iov[2];
	size_t iovlen = 1;

	header.magic = CMAN_MAGIC;
	header.version = CMAN_VERSION;
	header.command = msgtype;
	header.flags = 0;
	header.length = sizeof(header) + inlen;

	iov[0].iov_len = sizeof(header);
	iov[0].iov_base = &header;
	if (inbuf)
	{
		iov[1].iov_len = inlen;
		iov[1].iov_base = inbuf;
		iovlen++;
	}

	len = loopy_writev(h->fd, iov, iovlen);
	if (len < 0)
		return len;
	return 0;
}

/* Does something similar to the ioctl calls */
static int info_call(struct cman_handle *h, int msgtype, void *inbuf, int inlen, void *outbuf, int outlen)
{
	if (send_message(h, msgtype, inbuf, inlen))
		return -1;

	return wait_for_reply(h, outbuf, outlen);
}

static cman_handle_t open_socket(const char *name, int namelen, void *privdata)
{
	struct cman_handle *h;
	struct sockaddr_un sockaddr;

	h = malloc(sizeof(struct cman_handle));
	if (!h)
		return NULL;

	h->magic = CMAN_MAGIC;
	h->privdata = privdata;
	h->event_callback = NULL;
	h->data_callback = NULL;
	h->confchg_callback = NULL;
	h->want_reply = 0;
	h->saved_data_msg = NULL;
	h->saved_event_msg = NULL;
	h->saved_reply_msg = NULL;

	h->fd = socket(PF_UNIX, SOCK_STREAM, 0);
	if (h->fd == -1)
	{
		int saved_errno = errno;
		free(h);
		errno = saved_errno;
		return NULL;
	}

	fcntl(h->fd, F_SETFD, 1); /* Set close-on-exec */
	memset(&sockaddr, 0, sizeof(sockaddr));
	memcpy(sockaddr.sun_path, name, namelen);
	sockaddr.sun_family = AF_UNIX;

	if (connect(h->fd, (struct sockaddr *) &sockaddr, sizeof(sockaddr)) < 0)
	{
		int saved_errno = errno;
		close(h->fd);
		free(h);
		errno = saved_errno;
		return NULL;
	}

	/* Get a handle on /dev/zero too. This is always active so we
	   can return it from cman_get_fd() if we have cached messages */
	h->zero_fd = open("/dev/zero", O_RDONLY);
	if (h->zero_fd < 0)
	{
		int saved_errno = errno;
		close(h->fd);
		free(h);
		h = NULL;
		errno = saved_errno;
	}

	return (cman_handle_t)h;
}

cman_handle_t cman_admin_init(void *privdata)
{
	return open_socket(ADMIN_SOCKNAME, sizeof(ADMIN_SOCKNAME), privdata);
}

cman_handle_t cman_init(void *privdata)
{
	return open_socket(CLIENT_SOCKNAME, sizeof(CLIENT_SOCKNAME), privdata);
}

int cman_finish(cman_handle_t handle)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	h->magic = 0;
	close(h->fd);
	close(h->zero_fd);
	free(h);

	return 0;
}

int cman_setprivdata(cman_handle_t *handle, void *privdata)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	h->privdata = privdata;
	return 0;
}

int cman_getprivdata(cman_handle_t *handle, void **privdata)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	*privdata = h->privdata;

	return 0;
}


int cman_start_notification(cman_handle_t handle, cman_callback_t callback)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	if (!callback)
	{
		errno = EINVAL;
		return -1;
	}
	if (info_call(h, CMAN_CMD_NOTIFY, NULL, 0, NULL, 0))
		return -1;
	h->event_callback = callback;

	return 0;
}

int cman_stop_notification(cman_handle_t handle)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	if (info_call(h, CMAN_CMD_REMOVENOTIFY, NULL, 0, NULL, 0))
		return -1;
	h->event_callback = NULL;

	return 0;
}

int cman_start_confchg(cman_handle_t handle, cman_confchgcallback_t callback)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	if (!callback)
	{
		errno = EINVAL;
		return -1;
	}
	if (info_call(h, CMAN_CMD_START_CONFCHG, NULL, 0, NULL, 0))
		return -1;
	h->confchg_callback = callback;

	return 0;
}

int cman_stop_confchg(cman_handle_t handle)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	if (info_call(h, CMAN_CMD_STOP_CONFCHG, NULL, 0, NULL, 0))
		return -1;
	h->confchg_callback = NULL;

	return 0;
}


int cman_get_fd(cman_handle_t handle)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	VALIDATE_HANDLE(h);

	/* If we have saved messages then return an FD to /dev/zero which
	   will always be readable */
	if (h->saved_data_msg || h->saved_event_msg || h->saved_reply_msg)
		return h->zero_fd;
	else
		return h->fd;
}

int cman_dispatch(cman_handle_t handle, int flags)
{
	struct cman_handle *h = (struct cman_handle *)handle;
	int len;
	int offset;
	int recv_flags = 0;
	char buf[PIPE_BUF];
	VALIDATE_HANDLE(h);

	if (!(flags & CMAN_DISPATCH_BLOCKING))
		recv_flags |= MSG_DONTWAIT;

	do
	{
		int res;
		char *bufptr = buf;
		struct sock_header *header = (struct sock_header *)buf;

		/* First, drain any waiting queues */
		if (h->saved_reply_msg && !(flags & CMAN_DISPATCH_IGNORE_REPLY))
		{
			struct saved_message *smsg = h->saved_reply_msg;

			res = process_cman_message(h, flags, smsg->msg);
			h->saved_reply_msg = smsg->next;
			len = smsg->msg->length;
			free(smsg);
			if (res || (flags & CMAN_DISPATCH_ONE))
				break;
			else
				continue;
		}
		if (h->saved_data_msg && !(flags & CMAN_DISPATCH_IGNORE_DATA))
		{
			struct saved_message *smsg = h->saved_data_msg;

			res = process_cman_message(h, flags, smsg->msg);
			h->saved_data_msg = smsg->next;
			len = smsg->msg->length;
			free(smsg);
			if (res || (flags & CMAN_DISPATCH_ONE))
				break;
			else
				continue;
		}
		if (h->saved_event_msg && !(flags & CMAN_DISPATCH_IGNORE_EVENT))
		{
			struct saved_message *smsg = h->saved_event_msg;

			res = process_cman_message(h, flags, smsg->msg);
			h->saved_event_msg = smsg->next;
			len = smsg->msg->length;
			free(smsg);
			if (res || (flags & CMAN_DISPATCH_ONE))
				break;
			else
				continue;
		}

		/* Now look for new messages */
		len = recv(h->fd, buf, sizeof(struct sock_header), recv_flags);

		if (len == 0) {
			errno = EHOSTDOWN;
			return -1;
		}

		if (len < 0 &&
		    (errno == EINTR || errno == EAGAIN))
			return 0;

		if (len < 0)
			return -1;

		offset = len;

		/* It's too big! */
		if (header->length > sizeof(buf))
		{
			bufptr = malloc(header->length);
{'fraction_non_alphanumeric': 0.15639946040959818, 'fraction_numerical': 0.007439807055553285, 'mean_word_length': 4.076571902884416, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 299, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7442097', 'n_tokens_mistral': 10111, 'n_tokens_neox': 9340, 'n_words': 2246}