path: root/agents/virt/server/cpg.c

#include "config.h"

#include <stdio.h>
#include <sys/types.h>
#include <stdint.h>
#include <malloc.h>
#include <signal.h>
#include <unistd.h>
#include <sys/select.h>
#include <string.h>
#include <errno.h>
#include <time.h>
#include <sys/uio.h>
#include <list.h>
#include <pthread.h>

#include <corosync/cpg.h>

#include "debug.h"
#include "virt.h"
#include "cpg.h"

#define NODE_ID_NONE ((uint32_t) -1)

struct msg_queue_node {
	list_head();
	uint32_t seqno;
#define STATE_CLEAR 0
#define STATE_MESSAGE 1
	uint32_t state;
	void *msg;
	size_t msglen;
};

struct wire_msg {
#define TYPE_REQUEST 0
#define TYPE_REPLY 1
#define TYPE_STORE_VM 2
	uint32_t type;
	uint32_t seqno;
	uint32_t target;
	uint32_t pad;
	char data[0];
};

static uint32_t seqnum = 0;
static struct msg_queue_node *pending = NULL;
static cpg_handle_t cpg_handle;
static struct cpg_name gname;

static pthread_mutex_t cpg_mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t cpg_cond = PTHREAD_COND_INITIALIZER;
static pthread_t cpg_thread = 0;

static pthread_mutex_t cpg_ids_mutex = PTHREAD_MUTEX_INITIALIZER;
static uint32_t my_node_id = NODE_ID_NONE;
static uint32_t high_id_from_callback = NODE_ID_NONE;

static request_callback_fn req_callback_fn;
static request_callback_fn store_callback_fn;
static confchange_callback_fn conf_leave_fn;
static confchange_callback_fn conf_join_fn;


int
cpg_get_ids(uint32_t *my_id, uint32_t *high_id)
{
	if (!my_id && !high_id)
		return -1;

	pthread_mutex_lock(&cpg_ids_mutex);
	if (my_id)
		*my_id = my_node_id;

	if (high_id)
		*high_id = high_id_from_callback;
	pthread_mutex_unlock(&cpg_ids_mutex);

	return 0;
}

static void
cpg_deliver_func(cpg_handle_t h,
		 const struct cpg_name *group_name,
		 uint32_t nodeid,
		 uint32_t pid,
		 void *msg,
		 size_t msglen)
{
	struct msg_queue_node *n;
	struct wire_msg *m = msg;
	int x, found;

	pthread_mutex_lock(&cpg_mutex);
	if (m->type == TYPE_REPLY) {
		/* Reply to a request we sent */
		found = 0;

		list_for(&pending, n, x) {
			if (m->seqno != n->seqno)
				continue;
			if (m->target != my_node_id)
				continue;
			found = 1;
			break;
		}

		if (!found)
			goto out_unlock;

		/* Copy our message in to a buffer */
		n->msglen = msglen - sizeof(*m);
		if (!n->msglen) {
			/* XXX do what? */
		}
		n->msg = malloc(n->msglen);
		if (!n->msg) {
			goto out_unlock;
		}
		n->state = STATE_MESSAGE;
		memcpy(n->msg, (char *)msg + sizeof(*m), n->msglen);

		list_remove(&pending, n);
		list_insert(&pending, n);

		dbg_printf(2, "Seqnum %d replied; removing from list\n", n->seqno);

		pthread_cond_broadcast(&cpg_cond);
		goto out_unlock;
	}
	pthread_mutex_unlock(&cpg_mutex);

	if (m->type == TYPE_REQUEST) {
		req_callback_fn(&m->data, msglen - sizeof(*m),
				 nodeid, m->seqno);
	}
	if (m->type == TYPE_STORE_VM) {
		store_callback_fn(&m->data, msglen - sizeof(*m),
				 nodeid, m->seqno);
	}

	return;

out_unlock:
	pthread_mutex_unlock(&cpg_mutex);
}


static void
cpg_config_change(cpg_handle_t h,
		  const struct cpg_name *group_name,
		  const struct cpg_address *members, size_t memberlen,
		  const struct cpg_address *left, size_t leftlen,
		  const struct cpg_address *join, size_t joinlen)
{
	int x;
	int high;

	pthread_mutex_lock(&cpg_ids_mutex);
	high = my_node_id;

	for (x = 0; x < memberlen; x++) {
		if (members[x].nodeid > high)
			high = members[x].nodeid;
	}

	high_id_from_callback = high;
	pthread_mutex_unlock(&cpg_ids_mutex);

	if (joinlen > 0)
		conf_join_fn(join, joinlen);

	if (leftlen > 0)
		conf_leave_fn(left, leftlen);
}


static cpg_callbacks_t my_callbacks = {
	.cpg_deliver_fn = cpg_deliver_func,
	.cpg_confchg_fn = cpg_config_change
};


int
cpg_send_req(void *data, size_t len, uint32_t *seqno)
{
	struct iovec iov;
	struct msg_queue_node *n;
	struct wire_msg *m;
	size_t msgsz = sizeof(*m) + len;
	int ret;

	n = malloc(sizeof(*n));
	if (!n)
		return -1;

	m = malloc(msgsz);
	if (!m) {
		free(n);
		return -1;
	}

	/* only incremented on send */
	n->state = STATE_CLEAR;
	n->msg = NULL;
	n->msglen = 0;

	pthread_mutex_lock(&cpg_mutex);
	list_insert(&pending, n);
	n->seqno = ++seqnum;
	m->seqno = seqnum;
	*seqno = seqnum;
	pthread_mutex_unlock(&cpg_mutex);

	m->type = TYPE_REQUEST;		/* XXX swab? */
	m->target = NODE_ID_NONE;
	memcpy(&m->data, data, len);

	iov.iov_base = m;
	iov.iov_len = msgsz;
	ret = cpg_mcast_joined(cpg_handle, CPG_TYPE_AGREED, &iov, 1);

	free(m);
	if (ret == CS_OK)
		return 0;
	return -1;
}


int
cpg_send_vm_state(virt_state_t *vs)
{
	struct iovec iov;
	struct msg_queue_node *n;
	struct wire_msg *m;
	size_t msgsz = sizeof(*m) + sizeof(*vs);
	int ret;

	n = calloc(1, (sizeof(*n)));
	if (!n)
		return -1;

	m = calloc(1, msgsz);
	if (!m) {
		free(n);
		return -1;
	}

	n->state = STATE_MESSAGE;
	n->msg = NULL;
	n->msglen = 0;

	pthread_mutex_lock(&cpg_mutex);
	list_insert(&pending, n);
	pthread_mutex_unlock(&cpg_mutex);

	m->type = TYPE_STORE_VM;
	m->target = NODE_ID_NONE;

	memcpy(&m->data, vs, sizeof(*vs));

	iov.iov_base = m;
	iov.iov_len = msgsz;
	ret = cpg_mcast_joined(cpg_handle, CPG_TYPE_AGREED, &iov, 1);

	free(m);
	if (ret == CS_OK)
		return 0;

	return -1;
}


int
cpg_send_reply(void *data, size_t len, uint32_t nodeid, uint32_t seqno)
{
	struct iovec iov;
	struct wire_msg *m;
	size_t msgsz = sizeof(*m) + len;
	int ret;

	m = malloc(msgsz);
	if (!m)
		return -1;

	/* only incremented on send */
	m->seqno = seqno;
	m->type = TYPE_REPLY;		/* XXX swab? */
	m->target = nodeid;
	memcpy(&m->data, data, len);

	iov.iov_base = m;
	iov.iov_len = msgsz;
	ret = cpg_mcast_joined(cpg_handle, CPG_TYPE_AGREED, &iov, 1);

	free(m);
	if (ret == CS_OK)
		return 0;

	return -1;
}


int
cpg_wait_reply(void **data, size_t *len, uint32_t seqno)
{
	struct msg_queue_node *n;
	int x, found = 0;

	while (!found) {
		found = 0;
		pthread_mutex_lock(&cpg_mutex);
		pthread_cond_wait(&cpg_cond, &cpg_mutex);

		list_for(&pending, n, x) {
			if (n->seqno != seqno)
				continue;
			if (n->state != STATE_MESSAGE)
				continue;
			found = 1;
			goto out;
		}
		pthread_mutex_unlock(&cpg_mutex);
	}

out:
	list_remove(&pending, n);
	pthread_mutex_unlock(&cpg_mutex);

	*data = n->msg;
	*len = n->msglen;
	free(n);

	return 0;
}


static void *
cpg_dispatch_thread(void *arg)
{
	cpg_dispatch(cpg_handle, CS_DISPATCH_BLOCKING);

	return NULL;
}


int
cpg_start(	const char *name,
			request_callback_fn req_cb_fn,
			request_callback_fn store_cb_fn,
			confchange_callback_fn join_fn,
			confchange_callback_fn leave_fn)
{
	cpg_handle_t h;
	int ret;

	errno = EINVAL;

	if (!name)
		return -1;

	ret = snprintf(gname.value, sizeof(gname.value), "%s", name);
	if (ret <= 0)
		return -1;

	if (ret >= sizeof(gname.value)) {
		errno = ENAMETOOLONG;
		return -1;
	}
	gname.length = ret;

	memset(&h, 0, sizeof(h));
	if (cpg_initialize(&h, &my_callbacks) != CS_OK) {
		perror("cpg_initialize");
		return -1;
	}

	if (cpg_join(h, &gname) != CS_OK) {
		perror("cpg_join");
		return -1;
	}

	cpg_local_get(h, &my_node_id);
	dbg_printf(2, "My CPG nodeid is %d\n", my_node_id);

	pthread_mutex_lock(&cpg_mutex);
	pthread_create(&cpg_thread, NULL, cpg_dispatch_thread, NULL);

	memcpy(&cpg_handle, &h, sizeof(h));

	req_callback_fn = req_cb_fn;
	store_callback_fn = store_cb_fn;
	conf_join_fn = join_fn;
	conf_leave_fn = leave_fn;

	pthread_mutex_unlock(&cpg_mutex);

	return 0;
}


int
cpg_stop(void)
{
	pthread_cancel(cpg_thread);
	pthread_join(cpg_thread, NULL);
	cpg_leave(cpg_handle, &gname);
	cpg_finalize(cpg_handle);

	return 0;
}