[sheepdog] [PATCH] sheep: rework get_vdis_from to reduce network overhead

[Ruoyu]

Current get_vdis_from function prepare a buffer large enough (4M)
to get vdi state. It is not elegant and transfers too much between
sheep nodes.

Signed-off-by: Ruoyu <liangry at ucweb.com>
---
 sheep/group.c      | 18 ++++++++++++++----
 sheep/ops.c        |  4 +---
 sheep/sheep_priv.h |  3 ++-
 sheep/vdi.c        | 35 ++++++++++++++++++++++++-----------
 4 files changed, 41 insertions(+), 19 deletions(-)

diff --git a/sheep/group.c b/sheep/group.c
index adfd798..6b1ec5d 100644
--- a/sheep/group.c
+++ b/sheep/group.c
@@ -471,21 +471,31 @@ static int get_vdis_from(struct sd_node *node)
 	struct sd_req hdr;
 	struct sd_rsp *rsp = (struct sd_rsp *)&hdr;
 	struct vdi_state *vs = NULL;
-	int i, ret = SD_RES_SUCCESS;
+	int i, ret;
 	unsigned int rlen;
 	int count;
 
 	if (node_is_local(node))
-		goto out;
+		return SD_RES_SUCCESS;
 
-	rlen = SD_DATA_OBJ_SIZE; /* FIXME */
+#define DEFAULT_VDI_STATE_COUNT 512
+	rlen = DEFAULT_VDI_STATE_COUNT * sizeof(struct vdi_state);
 	vs = xzalloc(rlen);
+retry:
 	sd_init_req(&hdr, SD_OP_GET_VDI_COPIES);
 	hdr.data_length = rlen;
 	hdr.epoch = sys_epoch();
 	ret = sheep_exec_req(&node->nid, &hdr, (char *)vs);
-	if (ret != SD_RES_SUCCESS)
+	switch (ret) {
+	case SD_RES_SUCCESS:
+		break;
+	case SD_RES_BUFFER_SMALL:
+		rlen *= 2;
+		vs = xrealloc(vs, rlen);
+		goto retry;
+	default:
 		goto out;
+	}
 
 	count = rsp->data_length / sizeof(*vs);
 	for (i = 0; i < count; i++) {
diff --git a/sheep/ops.c b/sheep/ops.c
index fb26077..d76a0bb 100644
--- a/sheep/ops.c
+++ b/sheep/ops.c
@@ -412,9 +412,7 @@ static int local_read_vdis(const struct sd_req *req, struct sd_rsp *rsp,
 static int local_get_vdi_copies(const struct sd_req *req, struct sd_rsp *rsp,
 			   void *data)
 {
-	rsp->data_length = fill_vdi_state_list(data);
-
-	return SD_RES_SUCCESS;
+	return fill_vdi_state_list(req, rsp, data);
 }
 
 static int local_stat_sheep(struct request *req)
diff --git a/sheep/sheep_priv.h b/sheep/sheep_priv.h
index 02f75ad..3f25bec 100644
--- a/sheep/sheep_priv.h
+++ b/sheep/sheep_priv.h
@@ -327,7 +327,8 @@ int init_base_path(const char *dir);
 int init_disk_space(const char *d);
 int lock_base_dir(const char *d);
 
-int fill_vdi_state_list(void *data);
+int fill_vdi_state_list(const struct sd_req *hdr,
+		struct sd_rsp *rsp, void *data);
 bool oid_is_readonly(uint64_t oid);
 int get_vdi_copy_number(uint32_t vid);
 int get_vdi_copy_policy(uint32_t vid);
diff --git a/sheep/vdi.c b/sheep/vdi.c
index b58b538..747c810 100644
--- a/sheep/vdi.c
+++ b/sheep/vdi.c
@@ -172,25 +172,38 @@ int add_vdi_state(uint32_t vid, int nr_copies, bool snapshot, uint8_t cp)
 	return SD_RES_SUCCESS;
 }
 
-int fill_vdi_state_list(void *data)
+int fill_vdi_state_list(const struct sd_req *hdr,
+			struct sd_rsp *rsp, void *data)
 {
-	int nr = 0;
-	struct vdi_state *vs = data;
+#define DEFAULT_VDI_STATE_COUNT 512
+	int last = 0, end = DEFAULT_VDI_STATE_COUNT;
 	struct vdi_state_entry *entry;
+	struct vdi_state *vs = xzalloc(end * sizeof(struct vdi_state));
 
 	sd_read_lock(&vdi_state_lock);
 	rb_for_each_entry(entry, &vdi_state_root, node) {
-		memset(vs, 0, sizeof(*vs));
-		vs->vid = entry->vid;
-		vs->nr_copies = entry->nr_copies;
-		vs->snapshot = entry->snapshot;
-		vs->copy_policy = entry->copy_policy;
-		vs++;
-		nr++;
+		if (last >= end) {
+			end *= 2;
+			vs = xrealloc(vs, end * sizeof(struct vdi_state));
+		}
+
+		vs[last].vid = entry->vid;
+		vs[last].nr_copies = entry->nr_copies;
+		vs[last].snapshot = entry->snapshot;
+		vs[last].copy_policy = entry->copy_policy;
+		last++;
 	}
 	sd_rw_unlock(&vdi_state_lock);
 
-	return nr * sizeof(*vs);
+	if (hdr->data_length < last * sizeof(struct vdi_state)) {
+		free(vs);
+		return SD_RES_BUFFER_SMALL;
+	}
+
+	rsp->data_length = last * sizeof(struct vdi_state);
+	memcpy(data, vs, rsp->data_length);
+	free(vs);
+	return SD_RES_SUCCESS;
 }
 
 static inline bool vdi_is_deleted(struct sd_inode *inode)
-- 
1.8.3.2