: Cluster-aware log module - enabling cluster-aware mirrors

@@ -0,0 +1,269 @@
+/*
+ * Copyright (C) 2006 Red Hat, Inc.
+ *
+ * This file is released under the LGPL.
+ */
+
+#include <linux/kernel.h>
+#include <linux/module.h>
+#include <net/sock.h>
+#include <linux/workqueue.h>
+#include <linux/connector.h>
+#include "dm.h"
+#include "dm-clog-tfr.h"
+
+#include <asm/div64.h> /* Unnecessary */
+
+#define SHORT_UUID(x) (strlen(x) > 8) ? ((x) + (strlen(x) - 8)) : (x)
+
+static uint32_t seq = 0;
+
+/*
+ * Pre-allocated space for speed
+ */
+#define DM_CLOG_PREALLOCED_SIZE 512
+static struct cn_msg *prealloced_cn_msg = NULL;
+static struct clog_tfr *prealloced_clog_tfr = NULL;
+
+static struct cb_id cn_clog_id = { 0x4, 0x1 };
+static DEFINE_MUTEX(_lock);
+
+struct receiving_pkg {
+	struct list_head list;
+	struct completion complete;
+
+	unsigned long long start_time;
+	uint32_t seq;
+
+	int error;
+	int *data_size;
+	char *data;
+};
+
+static spinlock_t receiving_list_lock = SPIN_LOCK_UNLOCKED;
+static struct list_head receiving_list;
+
+static int dm_clog_sendto_server(struct clog_tfr *tfr)
+{
+	int r;
+	int size;
+	struct cn_msg *msg = prealloced_cn_msg;
+
+	if (tfr != prealloced_clog_tfr) {
+		size = sizeof(struct cn_msg) +
+			sizeof(struct clog_tfr) + tfr->data_size;
+		msg = kmalloc(size, GFP_NOIO);
+		if (!msg)
+			return -ENOMEM;
+		memcpy((msg + 1), tfr,
+		       sizeof(struct clog_tfr) + tfr->data_size);
+	}
+
+	memset(msg, 0, sizeof(struct cn_msg));
+
+	msg->id.idx = cn_clog_id.idx;
+	msg->id.val = cn_clog_id.val;
+	msg->ack = 0;
+	msg->seq = tfr->seq;
+	msg->len = sizeof(struct clog_tfr) + tfr->data_size;
+
+	r = cn_netlink_send(msg, 0, gfp_any());
+
+	if (msg != prealloced_cn_msg)
+		kfree(msg);
+
+	return r;
+}
+
+/*
+ * fill_pkg
+ * @msg
+ * @tfr
+ *
+ * Parameters can be either msg or tfr, but not both.  This
+ * function fills in the reply for a waiting request.  If just
+ * msg is given, then the reply is simply an ACK from userspace
+ * that the request was received.
+ *
+ * Returns: 0 on success, -ENOENT on failure
+ */
+static int fill_pkg(struct cn_msg *msg, struct clog_tfr *tfr)
+{
+	uint32_t rtn_seq = (msg) ? msg->seq : (tfr) ? tfr->seq : 0;
+	struct receiving_pkg *pkg;
+
+	list_for_each_entry(pkg, &receiving_list, list) {
+		if (rtn_seq != pkg->seq)
+			continue;
+
+		if (msg) {
+			pkg->error = -msg->ack;
+		} else if (tfr->data_size > *(pkg->data_size)) {
+			DMERR("Insufficient space to receive package [%s]",
+			      RQ_TYPE(tfr->request_type));
+			*(pkg->data_size) = 0;
+			pkg->error = -ENOSPC;
+		} else {
+			pkg->error = tfr->error;
+			memcpy(pkg->data, tfr->data, tfr->data_size);
+			*(pkg->data_size) = tfr->data_size;
+		}
+		complete(&pkg->complete);
+		return 0;
+	}
+
+	return -ENOENT;
+}
+
+/*
+ * cn_clog_callback
+ * @data
+ *
+ * This is the connector callback that delivers data
+ * that was sent from userspace.
+ */
+static void cn_clog_callback(void *data)
+{
+	struct cn_msg *msg = (struct cn_msg *)data;
+	struct clog_tfr *tfr = (struct clog_tfr *)(msg + 1);
+
+	spin_lock(&receiving_list_lock);
+	if (msg->len == 0)
+		fill_pkg(msg, NULL);
+	else if (msg->len < sizeof(*tfr))
+		DMERR("Incomplete message received: [%u]", msg->seq);
+	else
+		fill_pkg(NULL, tfr);
+	spin_unlock(&receiving_list_lock);
+}
+
+/*
+ * dm_clog_consult_server
+ * @uuid: log's uuid (must be DM_UUID_LEN in size)
+ * @request_type:
+ * @data: data to tx to the server
+ * @data_size: size of data in bytes
+ * @rdata: place to put return data from server
+ * @rdata_size: value-result (amount of space given/amount of space used)
+ *
+ * Only one process at a time can communicate with the server.
+ * rdata_size is undefined on failure.
+ *
+ * Returns: 0 on success, -EXXX on failure
+ */
+int dm_clog_consult_server(const char *uuid, int request_type,
+			   char *data, int data_size,
+			   char *rdata, int *rdata_size)
+{
+	int r = 0;
+	int dummy = 0;
+	int overhead_size = sizeof(struct clog_tfr *) + sizeof(struct cn_msg);
+	struct clog_tfr *tfr = prealloced_clog_tfr;
+	struct receiving_pkg pkg;
+
+	if (data_size > (DM_CLOG_PREALLOCED_SIZE - overhead_size)) {
+		DMINFO("Size of tfr exceeds preallocated size");
+		/* FIXME: is kmalloc sufficient if we need this much space? */
+		tfr = kzalloc(data_size + sizeof(*tfr), GFP_NOIO);
+	}
+
+	if (!tfr)
+		return -ENOMEM;
+
+	if (!rdata_size)
+		rdata_size = &dummy;
+resend:
+	/*
+	 * We serialize the sending of requests so we can
+	 * use the preallocated space.
+	 */
+	mutex_lock(&_lock);
+
+	memset(tfr, 0, DM_CLOG_PREALLOCED_SIZE - overhead_size);
+	memcpy(tfr->uuid, uuid, DM_UUID_LEN);
+	tfr->seq = seq++;
+	tfr->request_type = request_type;
+	tfr->data_size = data_size;
+	if (data && data_size)
+		memcpy(tfr->data, data, data_size);
+
+	memset(&pkg, 0, sizeof(pkg));
+	init_completion(&pkg.complete);
+	pkg.seq = tfr->seq;
+	pkg.data_size = rdata_size;
+	pkg.data = rdata;
+	spin_lock(&receiving_list_lock);
+	list_add(&(pkg.list), &receiving_list);
+	spin_unlock(&receiving_list_lock);
+	pkg.start_time = jiffies;
+
+	r = dm_clog_sendto_server(tfr);
+
+	mutex_unlock(&_lock);
+
+	if (r) {
+		DMERR("Unable to send cluster log request [%s] to server: %d",
+		      RQ_TYPE(request_type), r);
+		spin_lock(&receiving_list_lock);
+		list_del_init(&(pkg.list));
+		spin_unlock(&receiving_list_lock);
+
+		goto out;
+	}
+
+	r = wait_for_completion_timeout(&(pkg.complete), 15 * HZ);
+	spin_lock(&receiving_list_lock);
+	list_del_init(&(pkg.list));
+	spin_unlock(&receiving_list_lock);
+	if (!r) {
+		DMWARN("[%s] Request timed out: [%s/%u] - retrying",
+		       SHORT_UUID(uuid), RQ_TYPE(request_type), pkg.seq);
+		goto resend;
+	}
+
+	/* FIXME: Pull time checking code - it's just for profiling */
+	pkg.start_time = (jiffies - pkg.start_time);
+	do_div(pkg.start_time, HZ);
+	if (pkg.start_time > 0)
+		DMWARN("Excessive delay in request processing, %llu sec: [%s/%u]",
+		       pkg.start_time, RQ_TYPE(request_type), pkg.seq);
+
+	r = pkg.error;
+	if (r == -EAGAIN)
+		goto resend;
+
+out:
+	if (tfr != (struct clog_tfr *)prealloced_clog_tfr)
+		kfree(tfr);
+
+	return r;
+}
+
+int dm_clog_tfr_init(void)
+{
+	int r;
+	void *prealloced;
+
+	INIT_LIST_HEAD(&receiving_list);
+
+	prealloced = kmalloc(DM_CLOG_PREALLOCED_SIZE, GFP_KERNEL);
+	if (!prealloced)
+		return -ENOMEM;
+
+	prealloced_cn_msg = prealloced;
+	prealloced_clog_tfr = prealloced + sizeof(struct cn_msg);
+
+	r = cn_add_callback(&cn_clog_id, "clulog", cn_clog_callback);
+	if (r) {
+		cn_del_callback(&cn_clog_id);
+		return r;
+	}
+
+	return 0;
+}
+
+void dm_clog_tfr_exit(void)
+{
+	cn_del_callback(&cn_clog_id);
+	kfree(prealloced_cn_msg);
+}

75 lines not on this page · raw message

778 lines not on this page · raw message