There is no problem letting new nodes join a cluster in WAIT_FOR_JOIN state as
long as they aren't counted towards the quorum of nodes requires to restart
the cluster.  For this repurpose the CJ_RES_JOIN_LATER return value that is
currently equivalent to CJ_RES_FAIL to indicate we have a node that could
join, but wasn't part of the cluster before shutdown.  A new nr_delayed_nodes
counter and list of delayed nodes is also added to the join message to make
sure all nodes know about this kind of delayed node.

Signed-off-by: Christoph Hellwig <hch@lst.de>

---
 sheep/cluster/corosync.c  |    2 +-
 sheep/cluster/zookeeper.c |    8 +++-
 sheep/group.c             |  109 ++++++++++++++++++++++++++++++++-------------
 sheep/sheep_priv.h        |    6 +++
 4 files changed, 93 insertions(+), 32 deletions(-)

diff --git a/sheep/cluster/corosync.c b/sheep/cluster/corosync.c
index f477b51..4436981 100644
--- a/sheep/cluster/corosync.c
+++ b/sheep/cluster/corosync.c
@@ -285,11 +285,11 @@ static int __corosync_dispatch_one(struct corosync_event *cevent)
 		switch (cevent->result) {
 		case CJ_RES_SUCCESS:
 		case CJ_RES_MASTER_TRANSFER:
+		case CJ_RES_JOIN_LATER:
 			add_cpg_node(cpg_nodes, nr_cpg_nodes, &cevent->sender);
 			nr_cpg_nodes++;
 			/* fall through */
 		case CJ_RES_FAIL:
-		case CJ_RES_JOIN_LATER:
 			build_node_list(cpg_nodes, nr_cpg_nodes, entries);
 			sd_join_handler(&cevent->sender.ent, entries,
 						       nr_cpg_nodes, cevent->result,
diff --git a/sheep/cluster/zookeeper.c b/sheep/cluster/zookeeper.c
index 273b34d..57be370 100644
--- a/sheep/cluster/zookeeper.c
+++ b/sheep/cluster/zookeeper.c
@@ -732,7 +732,10 @@ static void zk_handler(int listen_fd, int events, void *data)
 		dprintf("one sheep joined[down], nr_nodes:%ld, sender:%s, joined:%d\n",
 				nr_zk_nodes, node_to_str(&ev.sender.node), ev.sender.joined);
 
-		if (ev.join_result == CJ_RES_SUCCESS) {
+		switch (ev.join_result) {
+		case CJ_RES_SUCCESS:
+		case CJ_RES_JOIN_LATER:
+		/* what about CJ_RES_MASTER_TRANSFER ? */
 			sprintf(path, MEMBER_ZNODE "/%s", node_to_str(&ev.sender.node));
 			if (node_eq(&ev.sender.node, &this_node.node)) {
 				dprintf("create path:%s\n", path);
@@ -744,6 +747,9 @@ static void zk_handler(int listen_fd, int events, void *data)
 				rc = zk_exists(zhandle, path, 1, NULL);
 				dprintf("watch path:%s, exists:%d\n", path, (rc == ZOK));
 			}
+			break;
+		default:
+			break;
 		}
 
 		build_node_list(zk_node_btroot);
diff --git a/sheep/group.c b/sheep/group.c
index 6fda1ce..34f807d 100644
--- a/sheep/group.c
+++ b/sheep/group.c
@@ -36,6 +36,7 @@ struct join_message {
 	uint8_t nr_copies;
 	uint16_t nr_nodes;
 	uint16_t nr_failed_nodes;
+	uint16_t nr_delayed_nodes;
 	uint16_t cluster_flags;
 	uint32_t cluster_status;
 	uint32_t epoch;
@@ -420,6 +421,26 @@ static struct sd_node *find_entry_epoch(struct sd_node *entry,
 }
 
 /*
+ * Add a node to the list of nodes that weren't part of the cluster before
+ * it shut down, and thus do not count toward the nodes required to allow
+ * an automated restart.  These nodes will become part of the cluster by
+ * the time it does get restarted.
+ */
+static bool add_delayed_node(int epoch, struct sd_node *node)
+{
+	struct node *n;
+
+	if (find_entry_list(node, &sys->delayed_nodes))
+		return false;
+	assert(!find_entry_epoch(node, epoch));
+
+	n = xmalloc(sizeof(*n));
+	n->ent = *node;
+	list_add_tail(&n->list, &sys->delayed_nodes);
+	return true;
+}
+
+/*
  * For a node that failed to join check if was part of the original
  * epoch, and if so add it to the list of node expected to be present
  * but failing to join.
@@ -439,28 +460,41 @@ static bool add_failed_node(int epoch, struct sd_node *node)
 	return true;
 }
 
-static void add_failed_nodes(int epoch, struct join_message *jm)
+/*
+ * Add the failed and delayed nodes in a join message to the local
+ * lists of such nodes.
+ */
+static void update_exceptional_node_list(int epoch, struct join_message *jm)
 {
 	int i;
 
 	for (i = 0; i < jm->nr_failed_nodes; i++)
 		add_failed_node(epoch, &jm->nodes[i]);
+	for ( ; i < jm->nr_failed_nodes + jm->nr_delayed_nodes; i++)
+		add_delayed_node(epoch, &jm->nodes[i]);
 }
 
-static void format_failed_node_list(struct join_message *jm)
+/*
+ * Format the lists of failed or delayed nodes into the join message.
+ */
+static void format_exceptional_node_list(struct join_message *jm)
 {
 	struct node *n;
 
 	list_for_each_entry(n, &sys->failed_nodes, list)
 		jm->nodes[jm->nr_failed_nodes++] = n->ent;
+	list_for_each_entry(n, &sys->delayed_nodes, list)
+		jm->nodes[jm->nr_failed_nodes + jm->nr_delayed_nodes++] = n->ent;
 }
 
-static void clear_failed_node_list(void)
+static void clear_exceptional_node_lists(void)
 {
 	struct node *n, *t;
 
 	list_for_each_entry_safe(n, t, &sys->failed_nodes, list)
 		list_del(&n->list);
+	list_for_each_entry_safe(n, t, &sys->delayed_nodes, list)
+		list_del(&n->list);
 }
 
 static int cluster_sanity_check(uint32_t epoch, uint64_t ctime)
@@ -479,21 +513,24 @@ static int cluster_sanity_check(uint32_t epoch, uint64_t ctime)
 		eprintf("joining node epoch too large: %"
 			PRIu32 " vs %" PRIu32 "\n",
 			epoch, local_epoch);
-		return CJ_RES_JOIN_LATER;
+		return CJ_RES_FAIL;
 	}
 
 	return CJ_RES_SUCCESS;
 }
 
-static int cluster_wait_for_join_check(struct sd_node *entries,
-		int nr_entries, uint32_t epoch, uint64_t ctime,
-		uint32_t *status, uint8_t *inc_epoch)
+static int cluster_wait_for_join_check(struct sd_node *joined,
+		struct sd_node *entries, int nr_entries, uint32_t epoch,
+		uint64_t ctime, uint32_t *status, uint8_t *inc_epoch)
 {
 	struct sd_node local_entries[SD_MAX_NODES];
-	int nr, nr_local_entries, nr_failed_entries;
+	int nr, nr_local_entries, nr_failed_entries, nr_delayed_nodes;
 	uint32_t local_epoch = get_latest_epoch();
 	int ret;
 
+	if (nr_entries == 0)
+		return CJ_RES_JOIN_LATER;
+
 	ret = cluster_sanity_check(epoch, ctime);
 	if (ret != CJ_RES_SUCCESS)  {
 		if (epoch > sys->epoch) {
@@ -504,15 +541,22 @@ static int cluster_wait_for_join_check(struct sd_node *entries,
 		return ret;
 	}
 
+	nr_local_entries = epoch_log_read(epoch, local_entries,
+					  sizeof(local_entries));
+	if (nr_local_entries == -1)
+		return CJ_RES_FAIL;
+
 	if (epoch < local_epoch) {
 		eprintf("joining node epoch too small: %"
 			PRIu32 " vs %" PRIu32 "\n",
 			epoch, local_epoch);
+
+		if (bsearch(joined, local_entries, nr_local_entries,
+			    sizeof(struct sd_node), node_cmp))
+			return CJ_RES_FAIL;
 		return CJ_RES_JOIN_LATER;
 	}
 
-	nr_local_entries = epoch_log_read(epoch, local_entries,
-					  sizeof(local_entries));
 	if (nr_entries != nr_local_entries) {
 		eprintf("epoch log entries do not match: %d vs %d\n",
 			nr_entries, nr_local_entries);
@@ -531,12 +575,14 @@ static int cluster_wait_for_join_check(struct sd_node *entries,
 	else
 		nr = current_vnode_info->nr_nodes + 1;
 
+	nr_delayed_nodes = get_nodes_nr_from(&sys->delayed_nodes);
+
 	/*
 	 * If we have all members from the last epoch log in the in-memory
-	 * node list we can set the cluster live now without incrementing
-	 * the epoch.
+	 * node list, and no new nodes joining we can set the cluster live
+	 * now without incrementing the epoch.
 	 */
-	if (nr == nr_local_entries) {
+	if (nr == nr_local_entries && !nr_delayed_nodes) {
 		*status = SD_STATUS_OK;
 		return CJ_RES_SUCCESS;
 	}
@@ -546,7 +592,7 @@ static int cluster_wait_for_join_check(struct sd_node *entries,
 	 * update the epoch before setting the cluster live.
 	 */
 	nr_failed_entries = get_nodes_nr_from(&sys->failed_nodes);
-	if (nr_local_entries == nr + nr_failed_entries) {
+	if (nr_local_entries == nr + nr_failed_entries - nr_delayed_nodes) {
 		if (inc_epoch)
 			*inc_epoch = 1;
 		*status = SD_STATUS_OK;
@@ -600,8 +646,9 @@ static int get_cluster_status(struct sd_node *joined, struct sd_node *entries,
 		return cluster_running_check(nr_entries, epoch, ctime,
 					     inc_epoch);
 	case SD_STATUS_WAIT_FOR_JOIN:
-		return cluster_wait_for_join_check(entries, nr_entries, epoch,
-						   ctime, status, inc_epoch);
+		return cluster_wait_for_join_check(joined, entries, nr_entries,
+						   epoch, ctime, status,
+						   inc_epoch);
 	default:
 		eprintf("invalid system status: 0x%x\n", sys->status);
 		abort();
@@ -726,7 +773,7 @@ static void finish_join(struct join_message *msg, struct sd_node *joined,
 	log_last_epoch(msg, joined, nodes, nr_nodes);
 
 	if (msg->cluster_status != SD_STATUS_OK)
-		add_failed_nodes(get_latest_epoch(), msg);
+		update_exceptional_node_list(get_latest_epoch(), msg);
 
 	if (!sd_store && strlen((char *)msg->store)) {
 		sd_store = find_store_driver((char *)msg->store);
@@ -793,7 +840,7 @@ static void update_cluster_info(struct join_message *msg,
 	sys_stat_set(msg->cluster_status);
 
 	if (sys_can_recover() && msg->inc_epoch) {
-		clear_failed_node_list();
+		clear_exceptional_node_lists();
 		start_recovery(current_vnode_info, old_vnode_info);
 	}
 
@@ -895,8 +942,9 @@ enum cluster_join_result sd_check_join_cb(struct sd_node *joining, void *opaque)
 	if (sd_store)
 		strcpy((char *)jm->store, sd_store->name);
 
-	if (ret == CJ_RES_SUCCESS && jm->cluster_status != SD_STATUS_OK)
-		format_failed_node_list(jm);
+	if (jm->cluster_status != SD_STATUS_OK &&
+	    (ret == CJ_RES_SUCCESS || CJ_RES_JOIN_LATER))
+		format_exceptional_node_list(jm);
 	return ret;
 }
 
@@ -938,23 +986,22 @@ void sd_join_handler(struct sd_node *joined, struct sd_node *members,
 		void *opaque)
 {
 	int i;
-	int nr, nr_local, nr_failed;
+	int nr, nr_local, nr_failed, nr_delayed_nodes;
 	struct join_message *jm = opaque;
 	uint32_t le = get_latest_epoch();
 
 	if (node_eq(joined, &sys->this_node)) {
 		if (result == CJ_RES_FAIL) {
-			eprintf("Fail to join. The joining node has an invalid epoch.\n");
-			sys->cdrv->leave();
-			exit(1);
-		} else if (result == CJ_RES_JOIN_LATER) {
-			eprintf("Fail to join. The joining node should be added after the cluster start working.\n");
+			eprintf("Failed to join, exiting.\n");
 			sys->cdrv->leave();
 			exit(1);
 		}
 	}
 
 	switch (result) {
+	case CJ_RES_JOIN_LATER:
+		add_delayed_node(le, joined);
+		/*FALLGHRU*/
 	case CJ_RES_SUCCESS:
 		dprintf("join %s\n", node_to_str(joined));
 		for (i = 0; i < nr_members; i++)
@@ -970,7 +1017,6 @@ void sd_join_handler(struct sd_node *joined, struct sd_node *members,
 			vprintf(SDOG_DEBUG, "join Sheepdog cluster\n");
 		break;
 	case CJ_RES_FAIL:
-	case CJ_RES_JOIN_LATER:
 		if (!sys_stat_wait_join())
 			break;
 
@@ -980,15 +1026,16 @@ void sd_join_handler(struct sd_node *joined, struct sd_node *members,
 		nr_local = get_nodes_nr_epoch(sys->epoch);
 		nr = nr_members;
 		nr_failed = get_nodes_nr_from(&sys->failed_nodes);
+		nr_delayed_nodes = get_nodes_nr_from(&sys->delayed_nodes);
 
 		dprintf("%d == %d + %d\n", nr_local, nr, nr_failed);
-		if (nr_local == nr + nr_failed) {
+		if (nr_local == nr + nr_failed - nr_delayed_nodes) {
 			sys_stat_set(SD_STATUS_OK);
 			log_current_epoch();
 		}
 		break;
 	case CJ_RES_MASTER_TRANSFER:
-		add_failed_nodes(le, jm);
+		update_exceptional_node_list(le, jm);
 
 		/* Sheep needs this to identify itself as master.
 		 * Now mastership transfer is done.
@@ -1004,9 +1051,10 @@ void sd_join_handler(struct sd_node *joined, struct sd_node *members,
 		nr_local = get_nodes_nr_epoch(sys->epoch);
 		nr = nr_members;
 		nr_failed = get_nodes_nr_from(&sys->failed_nodes);
+		nr_delayed_nodes = get_nodes_nr_from(&sys->delayed_nodes);
 
 		dprintf("%d == %d + %d\n", nr_local, nr, nr_failed);
-		if (nr_local == nr + nr_failed) {
+		if (nr_local == nr + nr_failed - nr_delayed_nodes) {
 			sys_stat_set(SD_STATUS_OK);
 			log_current_epoch();
 		}
@@ -1092,6 +1140,7 @@ int create_cluster(int port, int64_t zone, int nr_vnodes,
 		sys_stat_set(SD_STATUS_WAIT_FOR_JOIN);
 	INIT_LIST_HEAD(&sys->pending_list);
 	INIT_LIST_HEAD(&sys->failed_nodes);
+	INIT_LIST_HEAD(&sys->delayed_nodes);
 
 	INIT_LIST_HEAD(&sys->consistent_obj_list);
 	INIT_LIST_HEAD(&sys->blocking_conn_list);
diff --git a/sheep/sheep_priv.h b/sheep/sheep_priv.h
index 4d49a2e..4c760e0 100644
--- a/sheep/sheep_priv.h
+++ b/sheep/sheep_priv.h
@@ -112,6 +112,12 @@ struct cluster_info {
 	 */
 	struct list_head failed_nodes;
 
+	/*
+	 * List of nodes that weren't part of the last epoch, but joined
+	 * before restarting the cluster.
+	 */
+	struct list_head delayed_nodes;
+
 	struct sd_node nodes[SD_MAX_NODES];
 	int nr_nodes;
 
-- 
1.7.10



