This will allow us to keep the previous node list around for recovery, and thus
avoid reading the node list when starting recovery.

I've decided to make struct vnode_info public as part of this even if I'm not
entirely happy about it, but adding accessors for the node array seemed like
an even uglier alternative.

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

diff --git a/sheep/group.c b/sheep/group.c
index bb0f56d..143883d 100644
--- a/sheep/group.c
+++ b/sheep/group.c
@@ -31,13 +31,6 @@ struct node {
 	struct list_head list;
 };
 
-struct vnode_info {
-	struct sd_vnode entries[SD_MAX_VNODES];
-	int nr_vnodes;
-	int nr_zones;
-	int refcnt;
-};
-
 struct join_message {
 	uint8_t proto_ver;
 	uint8_t nr_copies;
@@ -121,15 +114,16 @@ static int get_zones_nr_from(struct sd_node *nodes, int nr_nodes)
 
 bool have_enough_zones(void)
 {
-	int nr_zones = get_zones_nr_from(sys->nodes, sys->nr_nodes);
-
-	dprintf("flags %d, nr_zones %d, copies %d\n", sys->flags, nr_zones,
-		sys->nr_copies);
-
 	if (sys_flag_nohalt())
 		return true;
 
-	if (nr_zones >= sys->nr_copies)
+	if (!current_vnode_info)
+		return false;
+
+	dprintf("flags %d, nr_zones %d, copies %d\n",
+		sys->flags, current_vnode_info->nr_zones, sys->nr_copies);
+
+	if (current_vnode_info->nr_zones >= sys->nr_copies)
 		return true;
 	return false;
 }
@@ -190,10 +184,10 @@ void put_vnode_info(struct vnode_info *vnode_info)
 struct sd_vnode *oid_to_vnode(struct vnode_info *vnode_info, uint64_t oid,
 		int copy_idx)
 {
-	int idx = obj_to_sheep(vnode_info->entries, vnode_info->nr_vnodes,
+	int idx = obj_to_sheep(vnode_info->vnodes, vnode_info->nr_vnodes,
 			oid, copy_idx);
 
-	return &vnode_info->entries[idx];
+	return &vnode_info->vnodes[idx];
 }
 
 void oid_to_vnodes(struct vnode_info *vnode_info, uint64_t oid, int nr_copies,
@@ -201,12 +195,12 @@ void oid_to_vnodes(struct vnode_info *vnode_info, uint64_t oid, int nr_copies,
 {
 	int idx_buf[SD_MAX_COPIES], i, n;
 
-	obj_to_sheeps(vnode_info->entries, vnode_info->nr_vnodes,
+	obj_to_sheeps(vnode_info->vnodes, vnode_info->nr_vnodes,
 			oid, nr_copies, idx_buf);
 
 	for (i = 0; i < nr_copies; i++) {
 		n = idx_buf[i];
-		vnodes[i] = &vnode_info->entries[n];
+		vnodes[i] = &vnode_info->vnodes[n];
 	}
 }
 
@@ -214,9 +208,16 @@ struct vnode_info *alloc_vnode_info(struct sd_node *nodes, size_t nr_nodes)
 {
 	struct vnode_info *vnode_info;
 
+	print_node_list(nodes, nr_nodes);
+
 	vnode_info = xzalloc(sizeof(*vnode_info));
+
+	vnode_info->nr_nodes = nr_nodes;
+	memcpy(vnode_info->nodes, nodes, sizeof(*nodes) * nr_nodes);
+	qsort(vnode_info->nodes, nr_nodes, sizeof(*nodes), node_cmp);
+
 	vnode_info->nr_vnodes = nodes_to_vnodes(nodes, nr_nodes,
-						vnode_info->entries);
+						vnode_info->vnodes);
 	vnode_info->nr_zones = get_zones_nr_from(nodes, nr_nodes);
 	uatomic_set(&vnode_info->refcnt, 1);
 	return vnode_info;
@@ -452,7 +453,11 @@ static int get_cluster_status(struct sd_node *from,
 			ret = SD_RES_NOT_FORMATTED;
 		break;
 	case SD_STATUS_WAIT_FOR_JOIN:
-		nr = sys->nr_nodes + 1;
+		if (!current_vnode_info)
+			nr = 1;
+		else
+			nr = current_vnode_info->nr_nodes + 1;
+
 		nr_local_entries = epoch_log_read_nr(epoch, (char *)local_entries,
 						  sizeof(local_entries));
 
@@ -560,16 +565,12 @@ static void get_vdi_bitmap_done(struct work *work)
 	free(w);
 }
 
-static void update_node_info(struct sd_node *nodes, size_t nr_nodes)
+int log_current_epoch(void)
 {
-	print_node_list(nodes, nr_nodes);
-
-	sys->nr_nodes = nr_nodes;
-	memcpy(sys->nodes, nodes, sizeof(*sys->nodes) * sys->nr_nodes);
-	qsort(sys->nodes, sys->nr_nodes, sizeof(*sys->nodes), node_cmp);
-
-	put_vnode_info(current_vnode_info);
-	current_vnode_info = alloc_vnode_info(sys->nodes, sys->nr_nodes);
+	if (!current_vnode_info)
+		return update_epoch_log(sys->epoch, NULL, 0);
+	return update_epoch_log(sys->epoch, current_vnode_info->nodes,
+				current_vnode_info->nr_nodes);
 }
 
 static void log_last_epoch(struct join_message *msg, struct sd_node *joined,
@@ -585,8 +586,9 @@ static void log_last_epoch(struct join_message *msg, struct sd_node *joined,
 			if (!node_eq(nodes + i, joined))
 				old_nodes[count++] = nodes[i];
 		}
-		update_node_info(old_nodes, count);
-		update_epoch_log(sys->epoch, sys->nodes, sys->nr_nodes);
+		put_vnode_info(current_vnode_info);
+		current_vnode_info = alloc_vnode_info(old_nodes, count);
+		log_current_epoch();
 	}
 }
 
@@ -658,13 +660,14 @@ static void update_cluster_info(struct join_message *msg,
 	if (!sys->join_finished)
 		finish_join(msg, joined, nodes, nr_nodes);
 
-	update_node_info(nodes, nr_nodes);
+	put_vnode_info(current_vnode_info);
+	current_vnode_info = alloc_vnode_info(nodes, nr_nodes);
 
 	if (msg->cluster_status == SD_STATUS_OK ||
 	    msg->cluster_status == SD_STATUS_HALT) {
 		if (msg->inc_epoch) {
 			uatomic_inc(&sys->epoch);
-			update_epoch_log(sys->epoch, sys->nodes, sys->nr_nodes);
+			log_current_epoch();
 		}
 		/* Fresh node */
 		if (!sys_stat_ok() && !sys_stat_halt()) {
@@ -988,7 +991,7 @@ void sd_join_handler(struct sd_node *joined, struct sd_node *members,
 		dprintf("%d == %d + %d\n", nr_local, nr, nr_leave);
 		if (nr_local == nr + nr_leave) {
 			sys_stat_set(SD_STATUS_OK);
-			update_epoch_log(sys->epoch, sys->nodes, sys->nr_nodes);
+			log_current_epoch();
 		}
 		break;
 	case CJ_RES_MASTER_TRANSFER:
@@ -1013,9 +1016,10 @@ void sd_join_handler(struct sd_node *joined, struct sd_node *members,
 		 */
 		if (!sys->join_finished) {
 			sys->join_finished = 1;
-			assert(sys->nr_nodes == 0);
-			update_node_info(&sys->this_node, 1);
 			sys->epoch = get_latest_epoch();
+
+			put_vnode_info(current_vnode_info);
+			current_vnode_info = alloc_vnode_info(&sys->this_node, 1);
 		}
 
 		nr_local = get_nodes_nr_epoch(sys->epoch);
@@ -1025,7 +1029,7 @@ void sd_join_handler(struct sd_node *joined, struct sd_node *members,
 		dprintf("%d == %d + %d\n", nr_local, nr, nr_leave);
 		if (nr_local == nr + nr_leave) {
 			sys_stat_set(SD_STATUS_OK);
-			update_epoch_log(sys->epoch, sys->nodes, sys->nr_nodes);
+			log_current_epoch();
 		}
 
 		if (node_eq(joined, &sys->this_node))
@@ -1047,11 +1051,12 @@ void sd_leave_handler(struct sd_node *left, struct sd_node *members,
 	if (sys_stat_shutdown())
 		return;
 
-	update_node_info(members, nr_members);
+	put_vnode_info(current_vnode_info);
+	current_vnode_info = alloc_vnode_info(members, nr_members);
 
 	if (sys_can_recover()) {
 		uatomic_inc(&sys->epoch);
-		update_epoch_log(sys->epoch, sys->nodes, sys->nr_nodes);
+		log_current_epoch();
 		start_recovery(sys->epoch);
 	}
 
diff --git a/sheep/ops.c b/sheep/ops.c
index 29be94f..1be9a27 100644
--- a/sheep/ops.c
+++ b/sheep/ops.c
@@ -245,7 +245,7 @@ static int cluster_make_fs(const struct sd_req *req, struct sd_rsp *rsp,
 	sys->epoch = 1;
 	sys->recovered_epoch = 1;
 
-	ret = update_epoch_log(sys->epoch, sys->nodes, sys->nr_nodes);
+	ret = log_current_epoch();
 	if (ret)
 		return SD_RES_EIO;
 
@@ -333,15 +333,19 @@ static int local_get_node_list(const struct sd_req *req, struct sd_rsp *rsp,
 			       void *data)
 {
 	struct sd_node_rsp *node_rsp = (struct sd_node_rsp *)rsp;
+	struct vnode_info *vnode_info;
 	int nr_nodes;
 
-	nr_nodes = sys->nr_nodes;
-	memcpy(data, sys->nodes, sizeof(*sys->nodes) * nr_nodes);
+	vnode_info = get_vnode_info();
+	nr_nodes = vnode_info->nr_nodes;
+
+	memcpy(data, vnode_info->nodes, sizeof(struct sd_node) * nr_nodes);
 	node_rsp->data_length = nr_nodes * sizeof(struct sd_node);
 	node_rsp->nr_nodes = nr_nodes;
 	node_rsp->local_idx = get_node_idx(&sys->this_node, data, nr_nodes);
 	node_rsp->master_idx = -1;
 
+	put_vnode_info(vnode_info);
 	return SD_RES_SUCCESS;
 }
 
@@ -471,7 +475,7 @@ static int cluster_manual_recover(const struct sd_req *req, struct sd_rsp *rsp,
 	sys->flags = f;
 
 	sys->epoch++; /* some nodes are left, so we get a new epoch */
-	ret = update_epoch_log(sys->epoch, sys->nodes, sys->nr_nodes);
+	ret = log_current_epoch();
 	if (ret) {
 		ret = SD_RES_EIO;
 		sys->epoch--;
diff --git a/sheep/recovery.c b/sheep/recovery.c
index 9b71d9b..69432ee 100644
--- a/sheep/recovery.c
+++ b/sheep/recovery.c
@@ -37,11 +37,6 @@ struct recovery_work {
 	int count;
 	uint64_t *oids;
 
-	int old_nr_nodes;
-	struct sd_node old_nodes[SD_MAX_NODES];
-	int cur_nr_nodes;
-	struct sd_node cur_nodes[SD_MAX_NODES];
-
 	struct vnode_info *old_vnodes;
 	struct vnode_info *cur_vnodes;
 };
@@ -212,8 +207,8 @@ again:
 	for (i = 0; i < nr_copies; i++) {
 		struct sd_vnode *tgt_vnode = oid_to_vnode(old, oid, i);
 
-		if (is_invalid_vnode(tgt_vnode, rw->cur_nodes,
-				     rw->cur_nr_nodes))
+		if (is_invalid_vnode(tgt_vnode, rw->cur_vnodes->nodes,
+				     rw->cur_vnodes->nr_nodes))
 			continue;
 		ret = recover_object_from_replica(oid, tgt_vnode,
 						  epoch, tgt_epoch);
@@ -568,10 +563,8 @@ static int screen_obj_list(struct recovery_work *rw,  uint64_t *list, int list_n
 
 static int newly_joined(struct sd_node *node, struct recovery_work *rw)
 {
-	struct sd_node *old = rw->old_nodes;
-	int old_nr = rw->old_nr_nodes;
-
-	if (bsearch(node, old, old_nr, sizeof(struct sd_node), node_cmp))
+	if (bsearch(node, rw->old_vnodes->nodes, rw->old_vnodes->nr_nodes,
+		    sizeof(struct sd_node), node_cmp))
 		return 0;
 	return 1;
 }
@@ -582,8 +575,8 @@ static int fill_obj_list(struct recovery_work *rw)
 	uint8_t *buf = NULL;
 	size_t buf_size = SD_DATA_OBJ_SIZE; /* FIXME */
 	int retry_cnt;
-	struct sd_node *cur = rw->cur_nodes;
-	int cur_nr = rw->cur_nr_nodes;
+	struct sd_node *cur = rw->cur_vnodes->nodes;
+	int cur_nr = rw->cur_vnodes->nr_nodes;
 	int start = random() % cur_nr;
 	int end = cur_nr;
 
@@ -641,29 +634,23 @@ again:
 /* setup node list and virtual node list */
 static int init_rw(struct recovery_work *rw)
 {
+	struct sd_node nodes[SD_MAX_NODES];
+	int nr_nodes;
 	uint32_t epoch = rw->epoch;
 
-	rw->cur_nr_nodes = epoch_log_read_nr(epoch, (char *)rw->cur_nodes,
-					     sizeof(rw->cur_nodes));
-	qsort(rw->cur_nodes, rw->cur_nr_nodes, sizeof(struct sd_node),
-		  node_cmp);
-
-	if (rw->cur_nr_nodes <= 0) {
+	nr_nodes = epoch_log_read_nr(epoch, (char *)nodes, sizeof(nodes));
+	if (nr_nodes <= 0) {
 		eprintf("failed to read epoch log for epoch %"PRIu32"\n", epoch);
 		return -1;
 	}
+	rw->cur_vnodes = alloc_vnode_info(nodes, nr_nodes);
 
-	rw->old_nr_nodes = epoch_log_read_nr(epoch - 1, (char *)rw->old_nodes,
-					     sizeof(rw->old_nodes));
-	if (rw->old_nr_nodes <= 0) {
+	nr_nodes = epoch_log_read_nr(epoch - 1, (char *)nodes, sizeof(nodes));
+	if (nr_nodes <= 0) {
 		eprintf("failed to read epoch log for epoch %"PRIu32"\n", epoch - 1);
 		return -1;
 	}
-	qsort(rw->old_nodes, rw->old_nr_nodes, sizeof(struct sd_node),
-		  node_cmp);
-
-	rw->old_vnodes = alloc_vnode_info(rw->old_nodes, rw->old_nr_nodes);
-	rw->cur_vnodes = alloc_vnode_info(rw->cur_nodes, rw->cur_nr_nodes);
+	rw->old_vnodes = alloc_vnode_info(nodes, nr_nodes);
 	return 0;
 }
 
@@ -676,8 +663,7 @@ static void do_recovery_work(struct work *work)
 	if (!sys->nr_copies)
 		return;
 
-	if (rw->cur_nr_nodes == 0)
-		init_rw(rw);
+	init_rw(rw);
 
 	if (fill_obj_list(rw) < 0) {
 		eprintf("fatal recovery error\n");
diff --git a/sheep/sheep_priv.h b/sheep/sheep_priv.h
index 9a23f57..e5f8f44 100644
--- a/sheep/sheep_priv.h
+++ b/sheep/sheep_priv.h
@@ -50,8 +50,16 @@ struct client_info {
 	int refcnt;
 };
 
-struct request;
-struct vnode_info;
+struct vnode_info {
+	struct sd_vnode vnodes[SD_MAX_VNODES];
+	int nr_vnodes;
+
+	struct sd_node nodes[SD_MAX_NODES];
+	int nr_nodes;
+
+	int nr_zones;
+	int refcnt;
+};
 
 struct request {
 	struct sd_req rq;
@@ -105,9 +113,6 @@ struct cluster_info {
 	 */
 	struct list_head leave_list;
 
-	struct sd_node nodes[SD_MAX_NODES];
-	int nr_nodes;
-
 	/* this array contains a list of ordered virtual nodes */
 	struct sd_vnode vnodes[SD_MAX_VNODES];
 	int nr_vnodes;
@@ -258,6 +263,7 @@ int read_epoch(uint32_t *epoch, uint64_t *ctime,
 	       struct sd_node *entries, int *nr_entries);
 
 int update_epoch_log(uint32_t epoch, struct sd_node *nodes, size_t nr_nodes);
+int log_current_epoch(void);
 
 int set_cluster_copies(uint8_t copies);
 int get_cluster_copies(uint8_t *copies);


