master - fenced: add protocol negotiation

[David Teigland <teigland at fedoraproject dot org>]

Gitweb:        http://git.fedorahosted.org/git/cluster.git?p=cluster.git;a=commitdiff;h=3a05914304d81a545c7022642f6f8e5cb3eda5da
Commit:        3a05914304d81a545c7022642f6f8e5cb3eda5da
Parent:        1ffd1effe1a2a4441a3ab151cbed4ad563256b3b
Author:        David Teigland <teigland@redhat.com>
AuthorDate:    Thu Sep 25 14:54:37 2008 -0500
Committer:     David Teigland <teigland@redhat.com>
CommitterDate: Thu Sep 25 14:54:37 2008 -0500

fenced: add protocol negotiation

Same as dlm_controld and gfs_controld, but without the kernel protocol.

Signed-off-by: David Teigland <teigland@redhat.com>
---
 fence/fenced/cpg.c           |  452 ++++++++++++++++++++++++++++++++++++++----
 fence/fenced/fd.h            |   10 +-
 fence/fenced/main.c          |    6 +
 group/dlm_controld/cpg.c     |    2 +
 group/gfs_controld/cpg-new.c |    2 +
 5 files changed, 432 insertions(+), 40 deletions(-)

diff --git a/fence/fenced/cpg.c b/fence/fenced/cpg.c
index b8bf79a..05308a6 100644
--- a/fence/fenced/cpg.c
+++ b/fence/fenced/cpg.c
@@ -1,10 +1,29 @@
 #include "fd.h"
 #include "config.h"
 
-static unsigned int protocol_active[3] = {1, 0, 0};
-static cpg_handle_t cpg_handle_daemon;
-static struct cpg_address daemon_member_list[MAX_NODES];
-static int daemon_member_list_entries;
+struct protocol_version {
+	uint16_t major;
+	uint16_t minor;
+	uint16_t patch;
+	uint16_t flags;
+};
+
+struct protocol {
+	union {
+		struct protocol_version dm_ver;
+		uint16_t                daemon_max[4];
+	};
+	union {
+		struct protocol_version dr_ver;
+		uint16_t                daemon_run[4];
+	};
+};
+
+struct node_daemon {
+	struct list_head list;
+	int nodeid;
+	struct protocol proto;
+};
 
 struct member {
 	struct list_head list;
@@ -46,6 +65,13 @@ struct id_info {
 	uint64_t fence_external_time;
 };
 
+static cpg_handle_t cpg_handle_daemon;
+static int daemon_cpg_fd;
+static struct protocol our_protocol;
+static struct list_head daemon_nodes;
+static struct cpg_address daemon_member[MAX_NODES];
+static int daemon_member_count;
+
 static void fd_info_in(struct fd_info *fi)
 {
 	fi->fd_info_size  = le32_to_cpu(fi->fd_info_size);
@@ -136,9 +162,9 @@ static void fd_send_message(struct fd *fd, char *buf, int len)
 	struct fd_header *hd = (struct fd_header *) buf;
 	int type = hd->type;
 
-	hd->version[0]  = cpu_to_le16(protocol_active[0]);
-	hd->version[1]  = cpu_to_le16(protocol_active[1]);
-	hd->version[2]  = cpu_to_le16(protocol_active[2]);
+	hd->version[0]  = cpu_to_le16(our_protocol.daemon_run[0]);
+	hd->version[1]  = cpu_to_le16(our_protocol.daemon_run[1]);
+	hd->version[2]  = cpu_to_le16(our_protocol.daemon_run[2]);
 	hd->type	= cpu_to_le16(hd->type);
 	hd->nodeid      = cpu_to_le32(our_nodeid);
 	hd->to_nodeid   = cpu_to_le32(hd->to_nodeid);
@@ -1215,6 +1241,19 @@ static void confchg_cb(cpg_handle_t handle, struct cpg_name *group_name,
 		add_victims_init(fd, cg);
 }
 
+static void fd_header_in(struct fd_header *hd)
+{
+	hd->version[0]  = le16_to_cpu(hd->version[0]);
+	hd->version[1]  = le16_to_cpu(hd->version[1]);
+	hd->version[2]  = le16_to_cpu(hd->version[2]);
+	hd->type        = le16_to_cpu(hd->type);
+	hd->nodeid      = le32_to_cpu(hd->nodeid);
+	hd->to_nodeid   = le32_to_cpu(hd->to_nodeid);
+	hd->global_id   = le32_to_cpu(hd->global_id);
+	hd->flags       = le32_to_cpu(hd->flags);
+	hd->msgdata     = le32_to_cpu(hd->msgdata);
+}
+
 static void deliver_cb(cpg_handle_t handle, struct cpg_name *group_name,
 		       uint32_t nodeid, uint32_t pid, void *data, int len)
 {
@@ -1227,23 +1266,21 @@ static void deliver_cb(cpg_handle_t handle, struct cpg_name *group_name,
 		return;
 	}
 
-	hd = (struct fd_header *)data;
+	if (len < sizeof(*hd)) {
+		log_error("deliver_cb short message %d", len);
+		return;
+	}
 
-	hd->version[0]  = le16_to_cpu(hd->version[0]);
-	hd->version[1]  = le16_to_cpu(hd->version[1]);
-	hd->version[2]  = le16_to_cpu(hd->version[2]);
-	hd->type        = le16_to_cpu(hd->type);
-	hd->nodeid      = le32_to_cpu(hd->nodeid);
-	hd->to_nodeid   = le32_to_cpu(hd->to_nodeid);
-	hd->global_id   = le32_to_cpu(hd->global_id);
-	hd->flags       = le32_to_cpu(hd->flags);
-	hd->msgdata     = le32_to_cpu(hd->msgdata);
+	hd = (struct fd_header *)data;
+	fd_header_in(hd);
 
-	if (hd->version[0] != protocol_active[0]) {
+	if (hd->version[0] != our_protocol.daemon_run[0] ||
+	    hd->version[1] != our_protocol.daemon_run[1]) {
 		log_error("reject message from %d version %u.%u.%u vs %u.%u.%u",
 			  nodeid, hd->version[0], hd->version[1],
-			  hd->version[2], protocol_active[0],
-			  protocol_active[1], protocol_active[2]);
+			  hd->version[2], our_protocol.daemon_run[0],
+			  our_protocol.daemon_run[1],
+			  our_protocol.daemon_run[2]);
 		return;
 	}
 
@@ -1372,6 +1409,320 @@ int fd_leave(struct fd *fd)
 	return 0;
 }
 
+static struct node_daemon *get_node_daemon(int nodeid)
+{
+	struct node_daemon *node;
+
+	list_for_each_entry(node, &daemon_nodes, list) {
+		if (node->nodeid == nodeid)
+			return node;
+	}
+	return NULL;
+}
+
+static void add_node_daemon(int nodeid)
+{
+	struct node_daemon *node;
+
+	if (get_node_daemon(nodeid))
+		return;
+
+	node = malloc(sizeof(struct node_daemon));
+	if (!node) {
+		log_error("add_node_daemon no mem");
+		return;
+	}
+	memset(node, 0, sizeof(struct node_daemon));
+	node->nodeid = nodeid;
+	list_add_tail(&node->list, &daemon_nodes);
+}
+
+static void pv_in(struct protocol_version *pv)
+{
+	pv->major = le16_to_cpu(pv->major);
+	pv->minor = le16_to_cpu(pv->minor);
+	pv->patch = le16_to_cpu(pv->patch);
+	pv->flags = le16_to_cpu(pv->flags);
+}
+
+static void pv_out(struct protocol_version *pv)
+{
+	pv->major = cpu_to_le16(pv->major);
+	pv->minor = cpu_to_le16(pv->minor);
+	pv->patch = cpu_to_le16(pv->patch);
+	pv->flags = cpu_to_le16(pv->flags);
+}
+
+static void protocol_in(struct protocol *proto)
+{
+	pv_in(&proto->dm_ver);
+	pv_in(&proto->dr_ver);
+}
+
+static void protocol_out(struct protocol *proto)
+{
+	pv_out(&proto->dm_ver);
+	pv_out(&proto->dr_ver);
+}
+
+/* go through member list saved in last confchg, see if we have received a
+   proto message from each */
+
+static int all_protocol_messages(void)
+{
+	struct node_daemon *node;
+	int i;
+
+	if (!daemon_member_count)
+		return 0;
+
+	for (i = 0; i < daemon_member_count; i++) {
+		node = get_node_daemon(daemon_member[i].nodeid);
+		if (!node) {
+			log_error("all_protocol_messages no node %d",
+				  daemon_member[i].nodeid);
+			return 0;
+		}
+
+		if (!node->proto.daemon_max[0])
+			return 0;
+	}
+	return 1;
+}
+
+static int pick_min_protocol(struct protocol *proto)
+{
+	uint16_t mind[4];
+	struct node_daemon *node;
+	int i;
+
+	memset(&mind, 0, sizeof(mind));
+
+	/* first choose the minimum major */
+
+	for (i = 0; i < daemon_member_count; i++) {
+		node = get_node_daemon(daemon_member[i].nodeid);
+		if (!node) {
+			log_error("pick_min_protocol no node %d",
+				  daemon_member[i].nodeid);
+			return -1;
+		}
+
+		if (!mind[0] || node->proto.daemon_max[0] < mind[0])
+			mind[0] = node->proto.daemon_max[0];
+	}
+
+	if (!mind[0]) {
+		log_error("pick_min_protocol zero major number");
+		return -1;
+	}
+
+	/* second pick the minimum minor with the chosen major */
+
+	for (i = 0; i < daemon_member_count; i++) {
+		node = get_node_daemon(daemon_member[i].nodeid);
+		if (!node)
+			continue;
+
+		if (mind[0] == node->proto.daemon_max[0]) {
+			if (!mind[1] || node->proto.daemon_max[1] < mind[1])
+				mind[1] = node->proto.daemon_max[1];
+		}
+	}
+
+	if (!mind[1]) {
+		log_error("pick_min_protocol zero minor number");
+		return -1;
+	}
+
+	/* third pick the minimum patch with the chosen major.minor */
+
+	for (i = 0; i < daemon_member_count; i++) {
+		node = get_node_daemon(daemon_member[i].nodeid);
+		if (!node)
+			continue;
+
+		if (mind[0] == node->proto.daemon_max[0] &&
+		    mind[1] == node->proto.daemon_max[1]) {
+			if (!mind[2] || node->proto.daemon_max[2] < mind[2])
+				mind[2] = node->proto.daemon_max[2];
+		}
+	}
+
+	if (!mind[2]) {
+		log_error("pick_min_protocol zero patch number");
+		return -1;
+	}
+
+	memcpy(&proto->daemon_run, &mind, sizeof(mind));
+	return 0;
+}
+
+static void receive_protocol(struct fd_header *hd, int len)
+{
+	struct protocol *p;
+	struct node_daemon *node;
+
+	p = (struct protocol *)((char *)hd + sizeof(struct fd_header));
+	protocol_in(p);
+
+	if (len < sizeof(struct fd_header) + sizeof(struct protocol)) {
+		log_error("receive_protocol invalid len %d from %d",
+			  len, hd->nodeid);
+		return;
+	}
+
+	/* zero is an invalid version value */
+
+	if (!p->daemon_max[0] || !p->daemon_max[1] || !p->daemon_max[2]) {
+		log_error("receive_protocol invalid max value from %d "
+			  "daemon %u.%u.%u", hd->nodeid,
+			  p->daemon_max[0], p->daemon_max[1], p->daemon_max[2]);
+		return;
+	}
+
+	/* the run values will be zero until a version is set, after
+	   which none of the run values can be zero */
+
+	if (p->daemon_run[0] && (!p->daemon_run[1] || !p->daemon_run[2])) {
+		log_error("receive_protocol invalid run value from %d "
+			  "daemon %u.%u.%u", hd->nodeid,
+			  p->daemon_run[0], p->daemon_run[1], p->daemon_run[2]);
+		return;
+	}
+
+	/* if we have zero run values, and this msg has non-zero run values,
+	   then adopt them as ours; otherwise save this proto message */
+
+	if (our_protocol.daemon_run[0])
+		return;
+
+	if (p->daemon_run[0]) {
+		memcpy(&our_protocol.daemon_run, &p->daemon_run,
+		       sizeof(struct protocol_version));
+		log_debug("run protocol from nodeid %d", hd->nodeid);
+		return;
+	}
+
+	/* save this node's proto so we can tell when we've got all, and
+	   use it to select a minimum protocol from all */
+
+	node = get_node_daemon(hd->nodeid);
+	if (!node) {
+		log_error("receive_protocol no node %d", hd->nodeid);
+		return;
+	}
+	memcpy(&node->proto, p, sizeof(struct protocol));
+}
+
+static void send_protocol(struct protocol *proto)
+{
+	struct fd_header *hd;
+	struct protocol *pr;
+	char *buf;
+	int len;
+
+	len = sizeof(struct fd_header) + sizeof(struct protocol);
+	buf = malloc(len);
+	if (!buf) {
+		log_error("send_protocol no mem %d", len);
+		return;
+	}
+	memset(buf, 0, len);
+
+	hd = (struct fd_header *)buf;
+	pr = (struct protocol *)(buf + sizeof(*hd));
+
+	hd->type = cpu_to_le16(FD_MSG_PROTOCOL);
+	hd->nodeid = cpu_to_le32(our_nodeid);
+
+	memcpy(pr, proto, sizeof(struct protocol));
+	protocol_out(pr);
+
+	_send_message(cpg_handle_daemon, buf, len, FD_MSG_PROTOCOL);
+}
+
+int set_protocol(void)
+{
+	struct protocol proto;
+	struct pollfd pollfd;
+	int sent_proposal = 0;
+	int rv;
+
+	memset(&pollfd, 0, sizeof(pollfd));
+	pollfd.fd = daemon_cpg_fd;
+	pollfd.events = POLLIN;
+
+	while (1) {
+		if (our_protocol.daemon_run[0])
+			break;
+
+		if (!sent_proposal && all_protocol_messages()) {
+			/* propose a protocol; look through info from all
+			   nodes and pick the min and propose that */
+
+			sent_proposal = 1;
+
+			/* copy our max values */
+			memcpy(&proto, &our_protocol, sizeof(struct protocol));
+
+			rv = pick_min_protocol(&proto);
+			if (rv < 0)
+				return rv;
+
+			log_debug("set_protocol member_count %d propose "
+				  "daemon %u.%u.%u", daemon_member_count,
+				  proto.daemon_run[0], proto.daemon_run[1],
+				  proto.daemon_run[2]);
+
+			send_protocol(&proto);
+		}
+
+		/* only process messages/events from daemon cpg until protocol
+		   is established */
+
+		rv = poll(&pollfd, 1, -1);
+		if (rv == -1 && errno == EINTR) {
+			if (daemon_quit)
+				return -1;
+			continue;
+		}
+		if (rv < 0) {
+			log_error("set_protocol poll errno %d", errno);
+			return -1;
+		}
+
+		if (pollfd.revents & POLLIN)
+			process_cpg(0);
+		if (pollfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
+			log_error("set_protocol poll revents %u",
+				  pollfd.revents);
+			return -1;
+		}
+	}
+
+	if (our_protocol.daemon_run[0] != our_protocol.daemon_max[0] ||
+	    our_protocol.daemon_run[1] > our_protocol.daemon_max[1]) {
+		log_error("incompatible daemon protocol run %u.%u.%u max %u.%u.%u",
+			our_protocol.daemon_run[0],
+			our_protocol.daemon_run[1],
+			our_protocol.daemon_run[2],
+			our_protocol.daemon_max[0],
+			our_protocol.daemon_max[1],
+			our_protocol.daemon_max[2]);
+		return -1;
+	}
+
+	log_debug("daemon run %u.%u.%u max %u.%u.%u",
+		  our_protocol.daemon_run[0],
+		  our_protocol.daemon_run[1],
+		  our_protocol.daemon_run[2],
+		  our_protocol.daemon_max[0],
+		  our_protocol.daemon_max[1],
+		  our_protocol.daemon_max[2]);
+	return 0;
+}
+
 /* process_cpg(), setup_cpg(), close_cpg() are for the "daemon" cpg which
    tracks the presence of other daemons; it's not the fenced domain cpg.
    Joining this cpg tells others that we don't have uncontrolled dlm/gfs
@@ -1381,6 +1732,23 @@ int fd_leave(struct fd *fd)
 static void deliver_cb_daemon(cpg_handle_t handle, struct cpg_name *group_name,
 		uint32_t nodeid, uint32_t pid, void *data, int len)
 {
+	struct fd_header *hd;
+
+	if (len < sizeof(*hd)) {
+		log_error("deliver_cb short message %d", len);
+		return;
+	}
+
+	hd = (struct fd_header *)data;
+	fd_header_in(hd);
+
+	switch (hd->type) {
+	case FD_MSG_PROTOCOL:
+		receive_protocol(hd, len);
+		break;
+	default:
+		log_error("deliver_cb_daemon unknown msg type %d", hd->type);
+	}
 }
 
 static void confchg_cb_daemon(cpg_handle_t handle, struct cpg_name *group_name,
@@ -1388,10 +1756,18 @@ static void confchg_cb_daemon(cpg_handle_t handle, struct cpg_name *group_name,
 		struct cpg_address *left_list, int left_list_entries,
 		struct cpg_address *joined_list, int joined_list_entries)
 {
-	memset(&daemon_member_list, 0, sizeof(daemon_member_list));
-	memcpy(&daemon_member_list, member_list,
-	       member_list_entries * sizeof(struct cpg_address));
-	daemon_member_list_entries = member_list_entries;
+	int i;
+
+	if (joined_list_entries)
+		send_protocol(&our_protocol);
+
+	memset(&daemon_member, 0, sizeof(daemon_member));
+	daemon_member_count = member_list_entries;
+
+	for (i = 0; i < member_list_entries; i++) {
+		daemon_member[i] = member_list[i];
+		add_node_daemon(member_list[i].nodeid);
+	}
 }
 
 static cpg_callbacks_t cpg_callbacks_daemon = {
@@ -1414,8 +1790,8 @@ int in_daemon_member_list(int nodeid)
 
 	cpg_dispatch(cpg_handle_daemon, CPG_DISPATCH_ALL);
 
-	for (i = 0; i < daemon_member_list_entries; i++) {
-		if (daemon_member_list[i].nodeid == nodeid)
+	for (i = 0; i < daemon_member_count; i++) {
+		if (daemon_member[i].nodeid == nodeid)
 			return 1;
 	}
 	return 0;
@@ -1424,26 +1800,30 @@ int in_daemon_member_list(int nodeid)
 int setup_cpg(void)
 {
 	cpg_error_t error;
-	cpg_handle_t h;
 	struct cpg_name name;
-	int i = 0, f;
+	int i = 0;
 
-	error = cpg_initialize(&h, &cpg_callbacks_daemon);
+	INIT_LIST_HEAD(&daemon_nodes);
+
+	memset(&our_protocol, 0, sizeof(our_protocol));
+	our_protocol.daemon_max[0] = 1;
+	our_protocol.daemon_max[1] = 1;
+	our_protocol.daemon_max[2] = 1;
+
+	error = cpg_initialize(&cpg_handle_daemon, &cpg_callbacks_daemon);
 	if (error != CPG_OK) {
 		log_error("daemon cpg_initialize error %d", error);
 		goto fail;
 	}
 
-	cpg_fd_get(h, &f);
-
-	cpg_handle_daemon = h;
+	cpg_fd_get(cpg_handle_daemon, &daemon_cpg_fd);
 
 	memset(&name, 0, sizeof(name));
 	sprintf(name.value, "fenced:daemon");
 	name.length = strlen(name.value) + 1;
 
  retry:
-	error = cpg_join(h, &name);
+	error = cpg_join(cpg_handle_daemon, &name);
 	if (error == CPG_ERR_TRY_AGAIN) {
 		sleep(1);
 		if (!(++i % 10))
@@ -1455,11 +1835,11 @@ int setup_cpg(void)
 		goto fail;
 	}
 
-	log_debug("setup_cpg %d", f);
-	return f;
+	log_debug("setup_cpg %d", daemon_cpg_fd);
+	return daemon_cpg_fd;
 
  fail:
-	cpg_finalize(h);
+	cpg_finalize(cpg_handle_daemon);
 	return -1;
 }
 
diff --git a/fence/fenced/fd.h b/fence/fenced/fd.h
index a649366..0ebd721 100644
--- a/fence/fenced/fd.h
+++ b/fence/fenced/fd.h
@@ -95,10 +95,11 @@ do { \
 	log_printf(lvl, fmt, ##args); \
 } while (0)
 
-#define FD_MSG_START		1
-#define FD_MSG_VICTIM_DONE	2
-#define FD_MSG_COMPLETE		3
-#define FD_MSG_EXTERNAL		4
+#define FD_MSG_PROTOCOL		1
+#define FD_MSG_START		2
+#define FD_MSG_VICTIM_DONE	3
+#define FD_MSG_COMPLETE		4
+#define FD_MSG_EXTERNAL		5
 
 #define FD_MFLG_JOINING		1  /* accompanies start, we are joining */
 #define FD_MFLG_COMPLETE	2  /* accompanies start, we have complete info */
@@ -207,6 +208,7 @@ int read_ccs(struct fd *fd);
 void process_cpg(int ci);
 int setup_cpg(void);
 void close_cpg(void);
+int set_protocol(void);
 void free_cg(struct change *cg);
 void node_history_fence(struct fd *fd, int victim, int master, int how,
 			uint64_t mastertime);
diff --git a/fence/fenced/main.c b/fence/fenced/main.c
index 31115a2..909720b 100644
--- a/fence/fenced/main.c
+++ b/fence/fenced/main.c
@@ -773,6 +773,12 @@ static void loop(void)
 	}
 	log_debug("group_mode %d compat %d", group_mode, cfgd_groupd_compat);
 
+	if (group_mode == GROUP_LIBCPG) {
+		rv = set_protocol();
+		if (rv < 0)
+			goto out;
+	}
+
 	for (;;) {
 		rv = poll(pollfd, client_maxi + 1, -1);
 		if (rv == -1 && errno == EINTR) {
diff --git a/group/dlm_controld/cpg.c b/group/dlm_controld/cpg.c
index e585878..3f8457e 100644
--- a/group/dlm_controld/cpg.c
+++ b/group/dlm_controld/cpg.c
@@ -130,6 +130,8 @@ static void ids_in(struct ls_info *li, struct id_info *ids)
 char *msg_name(int type)
 {
 	switch (type) {
+	case DLM_MSG_PROTOCOL:
+		return "protocol";
 	case DLM_MSG_START:
 		return "start";
 	case DLM_MSG_PLOCK:
diff --git a/group/gfs_controld/cpg-new.c b/group/gfs_controld/cpg-new.c
index d7bc4ed..1fc06f6 100644
--- a/group/gfs_controld/cpg-new.c
+++ b/group/gfs_controld/cpg-new.c
@@ -276,6 +276,8 @@ static void leave_mountgroup(struct mountgroup *mg, int mnterr);
 static char *msg_name(int type)
 {
 	switch (type) {
+	case GFS_MSG_PROTOCOL:
+		return "protocol";
 	case GFS_MSG_START:
 		return "start";
 	case GFS_MSG_MOUNT_DONE: