[sheepdog] [PATCH v2] sheep: update node in cluster driver

[Liu Yuan]

When we have update node info in sheep, we also have to update the corresponding
information in the driver, since node join/leave will send the sd nodes in the
corresponding handlers to sheep that will overload the nodes info in sheep.

Signed-off-by: Liu Yuan <namei.unix at gmail.com>
---
 v2:
 - correct shepherd_update_node

 sheep/cluster.h           |    3 +++
 sheep/cluster/corosync.c  |   27 +++++++++++++++++++++++++--
 sheep/cluster/local.c     |   22 ++++++++++++++++++++++
 sheep/cluster/shepherd.c  |    8 ++++++++
 sheep/cluster/zookeeper.c |   19 +++++++++++++++++++
 sheep/group.c             |    1 +
 6 files changed, 78 insertions(+), 2 deletions(-)

diff --git a/sheep/cluster.h b/sheep/cluster.h
index 0f6164c..a19a10b 100644
--- a/sheep/cluster.h
+++ b/sheep/cluster.h
@@ -101,6 +101,9 @@ struct cluster_driver {
 	 */
 	void (*unblock)(void *msg, size_t msg_len);
 
+	/* Update the specific node in the driver's private copy of nodes */
+	void (*update_node)(struct sd_node *);
+
 	struct list_head list;
 };
 
diff --git a/sheep/cluster/corosync.c b/sheep/cluster/corosync.c
index 8262a55..767c826 100644
--- a/sheep/cluster/corosync.c
+++ b/sheep/cluster/corosync.c
@@ -106,6 +106,18 @@ static inline int find_cpg_node(struct cpg_node *nodes, size_t nr_nodes,
 	return -1;
 }
 
+static inline int find_sd_node(struct cpg_node *nodes, size_t nr_nodes,
+			       struct sd_node *key)
+{
+	int i;
+
+	for (i = 0; i < nr_nodes; i++)
+		if (node_eq(&nodes[i].ent, key))
+			return i;
+
+	return -1;
+}
+
 static inline void add_cpg_node(struct cpg_node *nodes, size_t nr_nodes,
 				struct cpg_node *added)
 {
@@ -324,8 +336,8 @@ static bool __corosync_dispatch_one(struct corosync_event *cevent)
 		case CJ_RES_FAIL:
 			build_node_list(cpg_nodes, nr_cpg_nodes, entries);
 			sd_join_handler(&cevent->sender.ent, entries,
-						       nr_cpg_nodes, cevent->result,
-						       cevent->msg);
+					nr_cpg_nodes, cevent->result,
+					cevent->msg);
 			break;
 		}
 		break;
@@ -814,6 +826,16 @@ again:
 	return 0;
 }
 
+static void corosync_update_node(struct sd_node *node)
+{
+	int idx;
+
+	idx = find_sd_node(cpg_nodes, nr_cpg_nodes, node);
+	if (idx < 0)
+		return;
+	cpg_nodes[idx].ent = *node;
+}
+
 static struct cluster_driver cdrv_corosync = {
 	.name		= "corosync",
 
@@ -824,6 +846,7 @@ static struct cluster_driver cdrv_corosync = {
 	.notify		= corosync_notify,
 	.block		= corosync_block,
 	.unblock	= corosync_unblock,
+	.update_node	= corosync_update_node,
 };
 
 cdrv_register(cdrv_corosync);
diff --git a/sheep/cluster/local.c b/sheep/cluster/local.c
index 4624429..2d298d8 100644
--- a/sheep/cluster/local.c
+++ b/sheep/cluster/local.c
@@ -62,6 +62,7 @@ enum local_event_type {
 	EVENT_GATEWAY,
 	EVENT_BLOCK,
 	EVENT_NOTIFY,
+	EVENT_UPDATE_NODE,
 };
 
 struct local_event {
@@ -286,6 +287,10 @@ static void add_event(enum local_event_type type, struct local_node *lnode,
 	case EVENT_NOTIFY:
 	case EVENT_BLOCK:
 		break;
+	case EVENT_UPDATE_NODE:
+		n = find_lnode(lnode, ev.nr_lnodes, ev.lnodes);
+		n->node = lnode->node;
+		break;
 	case EVENT_JOIN_RESPONSE:
 		abort();
 	}
@@ -493,6 +498,8 @@ static bool local_process_event(void)
 	case EVENT_NOTIFY:
 		sd_notify_handler(&ev->sender.node, ev->buf, ev->buf_len);
 		break;
+	case EVENT_UPDATE_NODE:
+		break;
 	}
 out:
 	shm_queue_remove(ev);
@@ -569,6 +576,20 @@ static int local_init(const char *option)
 	return 0;
 }
 
+/* FIXME: we have to call nr of nodes times to update nodes information */
+static void local_update_node(struct sd_node *node)
+{
+	struct local_node n = {
+		.node = *node,
+	};
+
+	shm_queue_lock();
+
+	add_event(EVENT_UPDATE_NODE, &n, NULL, 0);
+
+	shm_queue_unlock();
+}
+
 static struct cluster_driver cdrv_local = {
 	.name		= "local",
 
@@ -579,6 +600,7 @@ static struct cluster_driver cdrv_local = {
 	.notify		= local_notify,
 	.block		= local_block,
 	.unblock	= local_unblock,
+	.update_node    = local_update_node,
 };
 
 cdrv_register(cdrv_local);
diff --git a/sheep/cluster/shepherd.c b/sheep/cluster/shepherd.c
index 2bc93d2..7f65a4b 100644
--- a/sheep/cluster/shepherd.c
+++ b/sheep/cluster/shepherd.c
@@ -685,6 +685,13 @@ static void shepherd_unblock(void *msg, size_t msg_len)
 	do_shepherd_notify(true, msg, msg_len);
 }
 
+static void shepherd_update_node(struct sd_node *node)
+{
+	for (int i = 0; i < nr_nodes; i++)
+		if (node_eq(node, &nodes[i]))
+			nodes[i] = *node;
+}
+
 static struct cluster_driver cdrv_shepherd = {
 	.name		= "shepherd",
 
@@ -694,6 +701,7 @@ static struct cluster_driver cdrv_shepherd = {
 	.notify		= shepherd_notify,
 	.block		= shepherd_block,
 	.unblock	= shepherd_unblock,
+	.update_node	= shepherd_update_node,
 };
 
 cdrv_register(cdrv_shepherd);
diff --git a/sheep/cluster/zookeeper.c b/sheep/cluster/zookeeper.c
index 476b863..2be3e4b 100644
--- a/sheep/cluster/zookeeper.c
+++ b/sheep/cluster/zookeeper.c
@@ -870,6 +870,24 @@ static int zk_init(const char *option)
 	return 0;
 }
 
+static void zk_update_node(struct sd_node *node)
+{
+	struct zk_node n = {
+		.node = *node,
+	};
+	struct zk_node *t;
+
+	sd_dprintf("%s", node_to_str(&n.node));
+
+	pthread_rwlock_rdlock(&zk_tree_lock);
+	t = zk_tree_search_nolock(&n.node.nid);
+	if (t) {
+		t->node = n.node;
+		build_node_list();
+	}
+	pthread_rwlock_unlock(&zk_tree_lock);
+}
+
 static struct cluster_driver cdrv_zookeeper = {
 	.name       = "zookeeper",
 
@@ -879,6 +897,7 @@ static struct cluster_driver cdrv_zookeeper = {
 	.notify     = zk_notify,
 	.block      = zk_block,
 	.unblock    = zk_unblock,
+	.update_node = zk_update_node,
 };
 
 cdrv_register(cdrv_zookeeper);
diff --git a/sheep/group.c b/sheep/group.c
index b9399ee..f74ef10 100644
--- a/sheep/group.c
+++ b/sheep/group.c
@@ -1242,6 +1242,7 @@ void update_node_size(struct sd_node *node)
 	int idx = get_node_idx(cur_vinfo, node);
 	assert(idx != -1);
 	cur_vinfo->nodes[idx].space = node->space;
+	sys->cdrv->update_node(node);
 }
 
 void kick_node_recover(void)
-- 
1.7.9.5