Update the node and vnodes lists as well as the epoch information on the
master node before replying to the slaves so that we can avoid a race
window which gives different sheep the same starting epoch.

This detailed order order of corosync messages:

slave0:cfgchange(join) ->
                             master:send-response(0),

slave1:cfgchange(join) ->
                             master:send-response(1)
			     master:recv-response(0) -> inc_epoch
			     master:recv-response(1) -> inc_epoch

will cause two responses to contain the same epoch, which gives one slave
the wrong starting epoch, and thus causes epoch mismatches in the cluster.

It can be fairly easily reproduced by starting a number of sheep very
quickly on a formatted cluster.  The actual reproduces is part of a bigger
software project, but I and my coworkers hope to contribute a simpler
reproducer as part of a test suite soon.

Implementing the fix requires passing the authoritative node list from the
cluster drivers to sd_check_join_cb, similar to how we do it for other
callbacks from the cluster drivers.  The callers in corosync and zookeeper
don't actually have this list as they didn't add the joining node yet, so
this adds additional complications.  The corosync version of this has been
heavily tested, but the zookeeper variant is entirely untested so far.

Signed-off-by: Christoph Hellwig <hch@lst.de>
---
 sheep/cluster.h           |    4 ++--
 sheep/cluster/corosync.c  |   11 +++++++++--
 sheep/cluster/local.c     |    3 ++-
 sheep/cluster/zookeeper.c |   14 +++++++++++++-
 sheep/group.c             |   28 +++++++++++++++++++---------
 5 files changed, 45 insertions(+), 15 deletions(-)

Index: sheepdog/sheep/cluster.h
===================================================================
--- sheepdog.orig/sheep/cluster.h	2012-05-14 21:16:32.551883383 +0200
+++ sheepdog/sheep/cluster.h	2012-05-14 21:19:18.707881795 +0200
@@ -189,7 +189,7 @@ void sd_join_handler(struct sd_node *joi
 void sd_leave_handler(struct sd_node *left, struct sd_node *members,
 		size_t nr_members);
 void sd_notify_handler(struct sd_node *sender, void *msg, size_t msg_len);
-enum cluster_join_result sd_check_join_cb(struct sd_node *joining,
-		void *opaque);
+enum cluster_join_result sd_check_join_cb(struct sd_node *members,
+		size_t nr_members, struct sd_node *joining, void *opaque);
 
 #endif
Index: sheepdog/sheep/cluster/corosync.c
===================================================================
--- sheepdog.orig/sheep/cluster/corosync.c	2012-05-14 21:16:32.551883383 +0200
+++ sheepdog/sheep/cluster/corosync.c	2012-05-14 21:21:06.611880766 +0200
@@ -295,8 +295,17 @@ static int __corosync_dispatch_one(struc
 				/* check_join() must be called only once */
 				return 0;
 
-			res = sd_check_join_cb(&cevent->sender.ent,
-						     cevent->msg);
+			/*
+			 * We didn't add the newly joining node to cpg_nodes
+			 * yet, but sd_check_join_cb expects to pass the
+			 * authoritative list of nodes, so we add it to entries
+			 * here.
+			 */
+			build_node_list(cpg_nodes, nr_cpg_nodes, entries);
+			entries[nr_cpg_nodes] = cevent->sender.ent;
+			res = sd_check_join_cb(entries, nr_cpg_nodes + 1,
+					       &cevent->sender.ent,
+					       cevent->msg);
 			if (res == CJ_RES_MASTER_TRANSFER)
 				nr_cpg_nodes = 0;
 
Index: sheepdog/sheep/cluster/local.c
===================================================================
--- sheepdog.orig/sheep/cluster/local.c	2012-05-14 21:16:32.551883383 +0200
+++ sheepdog/sheep/cluster/local.c	2012-05-14 21:19:18.707881795 +0200
@@ -404,7 +404,8 @@ static int local_dispatch(void)
 	case EVENT_JOIN:
 		if (ev->blocked) {
 			if (node_cmp(&ev->nodes[0], &this_node) == 0) {
-				res = sd_check_join_cb(&ev->sender, ev->buf);
+				res = sd_check_join_cb(ev->nodes, ev->nr_nodes,
+							&ev->sender, ev->buf);
 				ev->join_result = res;
 				ev->blocked = 0;
 				msync(ev, sizeof(*ev), MS_SYNC);
Index: sheepdog/sheep/cluster/zookeeper.c
===================================================================
--- sheepdog.orig/sheep/cluster/zookeeper.c	2012-05-14 21:16:32.551883383 +0200
+++ sheepdog/sheep/cluster/zookeeper.c	2012-05-14 21:20:27.855881137 +0200
@@ -777,7 +777,21 @@ static int zk_dispatch(void)
 			dprintf("one sheep joined[up], nr_nodes:%ld, sender:%s, joined:%d\n",
 					nr_zk_nodes, node_to_str(&ev.sender.node), ev.sender.joined);
 			if (is_master(zhandle, &this_node)) {
-				res = sd_check_join_cb(&ev.sender.node, ev.buf);
+				/*
+				 * We didn't add the newly joining node to
+				 * sd_nodes yet, but sd_check_join_cb expects
+				 * to pass the authoritative list of nodes,
+				 * so we add it here.
+				 */
+				struct sd_node join_nodes[SD_MAX_NODES];
+
+				memcpy(join_nodes, sd_nodes,
+					SD_MAX_NODES * sizeof(struct sd_node));
+				join_nodes[nr_sd_nodes] = ev.sender.node;
+
+				res = sd_check_join_cb(join_nodes,
+						       nr_sd_nodes + 1,
+						       &ev.sender.node, ev.buf);
 				ev.join_result = res;
 				ev.blocked = 0;
 				ev.sender.joined = 1;
Index: sheepdog/sheep/group.c
===================================================================
--- sheepdog.orig/sheep/group.c	2012-05-14 21:19:06.355881912 +0200
+++ sheepdog/sheep/group.c	2012-05-14 21:19:36.787881624 +0200
@@ -50,6 +50,7 @@ struct join_message {
 	uint64_t ctime;
 	uint32_t result;
 	uint8_t inc_epoch; /* set non-zero when we increment epoch of all nodes */
+	struct sd_node master;
 	uint8_t store[STORE_LEN];
 	union {
 		struct sd_node nodes[0];
@@ -587,12 +588,6 @@ static void update_cluster_info(struct j
 	eprintf("status = %d, epoch = %d, %x, %d\n", msg->cluster_status,
 		msg->epoch, msg->result, sys->join_finished);
 
-	if (sys_stat_join_failed())
-		return;
-
-	if (!sys->join_finished)
-		finish_join(msg, joined, nodes, nr_nodes);
-
 	update_node_info(nodes, nr_nodes);
 
 	if (msg->cluster_status == SD_STATUS_OK ||
@@ -609,7 +604,6 @@ static void update_cluster_info(struct j
 			set_cluster_ctime(msg->ctime);
 		}
 	}
-	sys_stat_set(msg->cluster_status);
 }
 
 static void __sd_notify(struct event_struct *cevent)
@@ -743,7 +737,8 @@ static void __sd_leave(struct event_stru
 	}
 }
 
-enum cluster_join_result sd_check_join_cb(struct sd_node *joining, void *opaque)
+enum cluster_join_result sd_check_join_cb(struct sd_node *nodes,
+		size_t nr_nodes, struct sd_node *joining, void *opaque)
 {
 	struct join_message *jm = opaque;
 
@@ -807,10 +802,13 @@ enum cluster_join_result sd_check_join_c
 		eprintf("transfer mastership (%d, %d)\n", jm->epoch, sys->epoch);
 		return CJ_RES_MASTER_TRANSFER;
 	}
+
 	jm->epoch = sys->epoch;
+	jm->master = sys->this_node;
 
 	switch (jm->result) {
 	case SD_RES_SUCCESS:
+		update_cluster_info(jm, joining, nodes, nr_nodes);
 		return CJ_RES_SUCCESS;
 	case SD_RES_OLD_NODE_VER:
 	case SD_RES_NEW_NODE_VER:
@@ -856,8 +854,20 @@ static void __sd_join_done(struct event_
 	print_node_list(sys->nodes, sys->nr_nodes);
 
 	if (!sys_stat_join_failed()) {
-		update_cluster_info(jm, &w->joined, w->member_list,
+		if (!sys->join_finished) {
+			finish_join(jm, &w->joined, w->member_list,
 				    w->member_list_entries);
+		}
+
+		/*
+		 * The master calls update_cluster_info from sd_check_join_cb.
+		 */
+		if (node_cmp(&jm->master, &sys->this_node) != 0) {
+			update_cluster_info(jm, &w->joined, w->member_list,
+					    w->member_list_entries);
+		}
+
+		sys_stat_set(jm->cluster_status);
 	}
 
 	if (sys_can_recover() && jm->inc_epoch) {


