[Cluster-devel] cluster/cman daemon/ais.c daemon/cnxman-privat ...

[Cluster-devel] cluster/cman daemon/ais.c daemon/cnxman-privat ...

[pcaulfield]

CVSROOT:	/cvs/cluster
Module name:	cluster
Changes by:	pcaulfield at sourceware.org	2006-08-17 13:22:40

Modified files:
	cman/daemon    : ais.c cnxman-private.h cnxman-socket.h 
	                 commands.c commands.h daemon.c daemon.h 
	cman/lib       : libcman.c libcman.h 
	cman/tests     : libtest.c 

Log message:
	Add a confchg callback to libcman, similar to the openAIS ones.
	this gives a race-free notification of cluster change deltas and will
	probably simplify client code hugely.
	(or it would if most of it hadn't already been written!)

Patches:
http://sourceware.org/cgi-bin/cvsweb.cgi/cluster/cman/daemon/ais.c.diff?cvsroot=cluster&r1=1.38&r2=1.39
http://sourceware.org/cgi-bin/cvsweb.cgi/cluster/cman/daemon/cnxman-private.h.diff?cvsroot=cluster&r1=1.22&r2=1.23
http://sourceware.org/cgi-bin/cvsweb.cgi/cluster/cman/daemon/cnxman-socket.h.diff?cvsroot=cluster&r1=1.15&r2=1.16
http://sourceware.org/cgi-bin/cvsweb.cgi/cluster/cman/daemon/commands.c.diff?cvsroot=cluster&r1=1.47&r2=1.48
http://sourceware.org/cgi-bin/cvsweb.cgi/cluster/cman/daemon/commands.h.diff?cvsroot=cluster&r1=1.12&r2=1.13
http://sourceware.org/cgi-bin/cvsweb.cgi/cluster/cman/daemon/daemon.c.diff?cvsroot=cluster&r1=1.30&r2=1.31
http://sourceware.org/cgi-bin/cvsweb.cgi/cluster/cman/daemon/daemon.h.diff?cvsroot=cluster&r1=1.5&r2=1.6
http://sourceware.org/cgi-bin/cvsweb.cgi/cluster/cman/lib/libcman.c.diff?cvsroot=cluster&r1=1.25&r2=1.26
http://sourceware.org/cgi-bin/cvsweb.cgi/cluster/cman/lib/libcman.h.diff?cvsroot=cluster&r1=1.26&r2=1.27
http://sourceware.org/cgi-bin/cvsweb.cgi/cluster/cman/tests/libtest.c.diff?cvsroot=cluster&r1=1.4&r2=1.5

--- cluster/cman/daemon/ais.c	2006/08/11 12:34:18	1.38
+++ cluster/cman/daemon/ais.c	2006/08/17 13:22:39	1.39
@@ -34,6 +34,7 @@
 #include <openais/service/config.h>
 #include <openais/lcr/lcr_comp.h>
 #include <openais/service/swab.h>
+#include <openais/service/print.h>
 
 #include "cnxman-socket.h"
 #include "commands.h"
@@ -60,6 +61,7 @@
 static char errorstring[512];
 static int startup_pipe;
 static unsigned int debug_mask;
+static int first_trans = 1;
 static struct objdb_iface_ver0 *global_objdb;
 static totempg_groups_handle group_handle;
 static struct totempg_group cman_group[1] = {
@@ -390,6 +392,9 @@
 {
 	int i;
 	static int last_memb_count = 0;
+	static int saved_left_list_entries;
+	static int saved_left_list_size;
+	static unsigned int *saved_left_list = NULL;
 
 	P_AIS("confchg_fn called type = %d, seq=%lld\n", configuration_type, ring_id->seq);
 
@@ -401,10 +406,37 @@
 	for (i=0; i<joined_list_entries; i++)
 		add_ais_node(joined_list[i], incarnation, member_list_entries);
 
+	/* Save the left list for later so we can do a consolidated confchg message */
+	if (configuration_type == TOTEM_CONFIGURATION_TRANSITIONAL) {
+		if (saved_left_list == NULL) {
+			saved_left_list_size = left_list_entries*2;
+			saved_left_list = malloc(sizeof(int) * saved_left_list_size);
+			if (!saved_left_list) {
+				log_printf(LOG_LEVEL_CRIT, "cannot allocate memory for confchg message");
+				exit(3);
+			}
+		}
+		if (saved_left_list_size < left_list_entries) {
+			saved_left_list_size = left_list_entries*2;
+			saved_left_list = realloc(saved_left_list, sizeof(int) * saved_left_list_size);
+			if (!saved_left_list) {
+				log_printf(LOG_LEVEL_CRIT, "cannot reallocate memory for confchg message");
+				exit(3);
+			}
+		}
+		saved_left_list_entries = left_list_entries;
+		memcpy(saved_left_list, left_list, left_list_entries * sizeof(int));
+	}
+
 	if (configuration_type == TOTEM_CONFIGURATION_REGULAR) {
 		P_AIS("last memb_count = %d, current = %d\n", last_memb_count, member_list_entries);
-		send_transition_msg(last_memb_count);
+		send_transition_msg(last_memb_count, first_trans);
 		last_memb_count = member_list_entries;
+		first_trans = 0;
+
+		cman_send_confchg(member_list,  member_list_entries,
+				  saved_left_list, saved_left_list_entries,
+				  joined_list, joined_list_entries);
 	}
 }
 
--- cluster/cman/daemon/cnxman-private.h	2006/06/30 13:00:27	1.22
+++ cluster/cman/daemon/cnxman-private.h	2006/08/17 13:22:39	1.23
@@ -70,7 +70,7 @@
 
 struct cl_transmsg {
 	unsigned char cmd;
-	unsigned char pad;
+	unsigned char first_trans;
 	uint16_t cluster_id;
 	int high_nodeid;
 	int expected_votes;
@@ -129,6 +129,7 @@
 	uint32_t   port;        /* If bound client */
 	enum {SHUTDOWN_REPLY_UNK=0, SHUTDOWN_REPLY_YES, SHUTDOWN_REPLY_NO} shutdown_reply;
 	uint32_t   events;      /* Registered for events */
+	uint32_t   confchg;     /* Registered for confchg */
 	struct list write_msgs; /* Queued messages to go to data clients */
 	struct cl_comms_socket *clsock;
 	struct connection *next;
--- cluster/cman/daemon/cnxman-socket.h	2006/05/15 14:30:16	1.15
+++ cluster/cman/daemon/cnxman-socket.h	2006/08/17 13:22:39	1.16
@@ -50,10 +50,13 @@
 #define CMAN_CMD_UPDATE_FENCE_INFO  0x800000bd
 #define CMAN_CMD_GET_FENCE_INFO     0x000000be
 #define CMAN_CMD_GET_NODEADDRS      0x000000bf
+#define CMAN_CMD_START_CONFCHG      0x000000c0
+#define CMAN_CMD_STOP_CONFCHG       0x000000c1
 
 #define CMAN_CMD_DATA               0x00000100
 #define CMAN_CMD_BIND               0x00000101
 #define CMAN_CMD_EVENT              0x00000102
+#define CMAN_CMD_CONFCHG            0x00000103
 
 #define CMAN_CMDFLAG_PRIV           0x80000000
 #define CMAN_CMDFLAG_REPLY          0x40000000
@@ -150,6 +153,15 @@
 	int arg;
 };
 
+/* confchg message */
+struct sock_confchg_message {
+	struct sock_header header;
+	int member_entries;
+	int left_entries;
+	int joined_entries;
+	unsigned int entries[]; // In above order.
+};
+
 /* Flags */
 #define CMAN_EXTRA_FLAG_2NODE    1
 #define CMAN_EXTRA_FLAG_ERROR    2
--- cluster/cman/daemon/commands.c	2006/08/15 10:27:13	1.47
+++ cluster/cman/daemon/commands.c	2006/08/17 13:22:39	1.48
@@ -1157,6 +1157,16 @@
 		err = 0;
 		break;
 
+	case CMAN_CMD_START_CONFCHG:
+		con->confchg = 1;
+		err = 0;
+		break;
+
+	case CMAN_CMD_STOP_CONFCHG:
+		con->confchg = 0;
+		err = 0;
+		break;
+
 		/* Return the cnxman version number */
 	case CMAN_CMD_GET_VERSION:
 		err = 0;
@@ -1297,6 +1307,31 @@
 	return ret;
 }
 
+void cman_send_confchg(unsigned int *member_list, int member_list_entries,
+		       unsigned int *left_list, int left_list_entries,
+		       unsigned int *joined_list, int joined_list_entries)
+{
+	char buf[sizeof(struct sock_confchg_message) +
+		 (member_list_entries+left_list_entries+joined_list_entries) * sizeof(int)];
+	struct sock_confchg_message *msg = (struct sock_confchg_message *)buf;
+
+	msg->header.magic = CMAN_MAGIC;
+	msg->header.command = CMAN_CMD_CONFCHG;
+	msg->header.length = sizeof(buf);
+	msg->header.flags = 0;
+
+	msg->member_entries = member_list_entries;
+	msg->joined_entries = joined_list_entries;
+	msg->left_entries = left_list_entries;
+
+	memcpy(msg->entries, member_list, sizeof(int)*member_list_entries);
+	memcpy(msg->entries+member_list_entries, left_list, sizeof(int)*left_list_entries);
+	memcpy(msg->entries+member_list_entries+left_list_entries, joined_list, sizeof(int)*joined_list_entries);
+
+	notify_confchg((struct sock_header *)msg);
+}
+
+
 /* Send a port closedown message to all cluster nodes - this tells them that a
  * port listener has gone away */
 static int send_port_close_msg(unsigned char port)
@@ -1428,7 +1463,7 @@
 }
 
 
-void send_transition_msg(int last_memb_count)
+void send_transition_msg(int last_memb_count, int first_trans)
 {
 	char buf[sizeof(struct cl_transmsg)+1024];
 	struct cl_transmsg *msg = (struct cl_transmsg *)buf;
@@ -1438,6 +1473,7 @@
 
 	P_MEMB("sending TRANSITION message. cluster_name = %s\n", cluster_name);
 	msg->cmd = CLUSTER_MSG_TRANSITION;
+	msg->first_trans = first_trans;
 	msg->high_nodeid = get_highest_nodeid();
 	msg->expected_votes = us->expected_votes;
 	msg->cluster_id = cluster_id;
@@ -1601,6 +1637,9 @@
 	node = find_node_by_nodeid(nodeid);
 	assert(node);
 
+	if (node->flags & NODE_FLAGS_GOTTRANSITION) {
+
+	}
 	node->flags = msg->flags;
 	if (node->fence_agent && msg->fence_agent[0] && strcmp(node->fence_agent, msg->fence_agent))
 	{
--- cluster/cman/daemon/commands.h	2006/05/22 09:08:48	1.12
+++ cluster/cman/daemon/commands.h	2006/08/17 13:22:39	1.13
@@ -24,12 +24,17 @@
 extern void commands_init(void);
 extern int process_command(struct connection *con, int cmd, char *cmdbuf,
 			   char **retbuf, int *retlen, int retsize, int offset);
-extern void send_transition_msg(int last_memb_count);
+extern void send_transition_msg(int last_memb_count, int first_trans);
 
 extern void add_ais_node(int nodeid, uint64_t incarnation, int total_members);
 extern void del_ais_node(int nodeid);
 extern void add_ccs_node(char *name, int nodeid, int votes, int expected_votes);
 extern void override_expected(int expected);
+extern void cman_send_confchg(unsigned int *member_list, int member_list_entries,
+			      unsigned int *left_list, int left_list_entries,
+			      unsigned int *joined_list, int joined_list_entries);
+
+
 
 /* Startup stuff called from cmanccs: */
 extern int cman_set_nodename(char *name);
--- cluster/cman/daemon/daemon.c	2006/08/11 12:34:18	1.30
+++ cluster/cman/daemon/daemon.c	2006/08/17 13:22:39	1.31
@@ -412,6 +412,16 @@
 	}
 }
 
+void notify_confchg(struct sock_header *message)
+{
+	struct connection *thiscon;
+
+	list_iterate_items(thiscon, &client_list) {
+		if (thiscon->confchg)
+			send_reply_message(thiscon, message);
+	}
+}
+
 void wake_daemon(void)
 {
 	P_DAEMON("Wake daemon called\n");
--- cluster/cman/daemon/daemon.h	2006/03/15 15:15:20	1.5
+++ cluster/cman/daemon/daemon.h	2006/08/17 13:22:39	1.6
@@ -18,7 +18,7 @@
 extern void cman_set_realtime(void);
 extern int cman_init(void);
 extern int cman_finish(void);
-
+extern void notify_confchg(struct sock_header *message);
 
 extern volatile sig_atomic_t quit_threads;
 extern int num_connections;
--- cluster/cman/lib/libcman.c	2006/07/11 08:13:18	1.25
+++ cluster/cman/lib/libcman.c	2006/08/17 13:22:39	1.26
@@ -49,6 +49,7 @@
 	int want_reply;
 	cman_callback_t event_callback;
 	cman_datacallback_t data_callback;
+	cman_confchgcallback_t confchg_callback;
 
 	void *reply_buffer;
 	int reply_buflen;
@@ -192,18 +193,28 @@
 	}
 
 	/* OOB event */
-	if (msg->command == CMAN_CMD_EVENT)
+	if (msg->command == CMAN_CMD_EVENT || msg->command == CMAN_CMD_CONFCHG)
 	{
-		struct sock_event_message *emsg = (struct sock_event_message *)msg;
-
 		if (flags & CMAN_DISPATCH_IGNORE_EVENT)
 		{
 			add_to_waitlist(&h->saved_event_msg, msg);
 		}
 		else
 		{
-			if (h->event_callback)
+			if (msg->command == CMAN_CMD_EVENT && h->event_callback) {
+				struct sock_event_message *emsg = (struct sock_event_message *)msg;
 				h->event_callback(h, h->private, 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->private,
+						    cmsg->entries,cmsg->member_entries, 
+						    &cmsg->entries[cmsg->member_entries], cmsg->left_entries, 
+						    &cmsg->entries[cmsg->member_entries+cmsg->left_entries], cmsg->joined_entries);
+			}
 		}
 	}
 
@@ -404,6 +415,35 @@
 	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)
 {
--- cluster/cman/lib/libcman.h	2006/07/11 08:13:18	1.26
+++ cluster/cman/lib/libcman.h	2006/08/17 13:22:39	1.27
@@ -181,6 +181,11 @@
 				    char *buf, int len, uint8_t port, int nodeid);
 
 
+typedef void (*cman_confchgcallback_t)(cman_handle_t handle, void *private,
+				       unsigned int *member_list, int member_list_entries,
+				       unsigned int *left_list, int left_list_entries,
+				       unsigned int *joined_list, int joined_list_entries);
+
 /*
  * cman_init        returns the handle you need to pass to the other API calls.
  * cman_admin_init  opens admin socket for privileged operations.
@@ -203,6 +208,15 @@
 int cman_start_notification(cman_handle_t handle, cman_callback_t callback);
 int cman_stop_notification(cman_handle_t handle);
 
+/*
+ * Start/stop AIS-style confchg callbacks. These are less racy than the
+ * old cman callbacks in that the caller will get one for each AIS
+ * confchg message and it will contain all of the nodes that joined &
+ * left in that transition.
+ */
+int cman_start_confchg(cman_handle_t handle, cman_confchgcallback_t callback);
+int cman_stop_confchg(cman_handle_t handle);
+
 /* Call this if you get a TRY_SHUTDOWN event to signal whether you
  * will let cman shutdown or not.
  */
--- cluster/cman/tests/libtest.c	2006/08/11 10:53:48	1.4
+++ cluster/cman/tests/libtest.c	2006/08/17 13:22:39	1.5
@@ -10,6 +10,24 @@
 	printf("callback called reason = %d, arg=%d\n", reason, arg);
 }
 
+static void confchg_callback(cman_handle_t handle, void *private,
+			     unsigned int *member_list, int member_list_entries,
+			     unsigned int *left_list, int left_list_entries,
+			     unsigned int *joined_list, int joined_list_entries)
+{
+	int i;
+	printf("Confchg callback\n");
+	printf("member_list: %d entries:\n", member_list_entries);
+	for (i=0; i<member_list_entries; i++)
+		printf("  %d\n", member_list[i]);
+	printf("left_list: %d entries:\n", left_list_entries);
+	for (i=0; i<left_list_entries; i++)
+		printf("  %d\n", left_list[i]);
+	printf("joined_list: %d entries:\n", joined_list_entries);
+	for (i=0; i<joined_list_entries; i++)
+		printf("  %d\n", joined_list[i]);
+}
+
 static void print_node(cman_node_t *node)
 {
 	printf("  node id     %d\n", node->cn_nodeid);
@@ -87,10 +105,17 @@
 		perror("get_node failed");
 	}
 
-	if (!cman_start_notification(h, cman_callback))
+	if (cman_start_notification(h, cman_callback))
 	{
 		perror("start_notification");
 	}
+
+	if (cman_start_confchg(h, confchg_callback))
+	{
+		perror("start_confchg");
+	}
+
+
 	while (1) {
 	  int ret = cman_dispatch(h, CMAN_DISPATCH_BLOCKING | CMAN_DISPATCH_ALL);
 	  if (ret == -1) {