[sheepdog] [PATCH] sheep: handle a case of small buffer in cinfo collection

[Hitoshi Mitake]

Current cinfo_collection_work() and SD_OP_VDI_STATE_SNAPSHOT_CTL
doesn't handle a case of small buffer. This patch solves it.

Cc: Saeki Masaki <saeki.masaki at po.ntts.co.jp>
Signed-off-by: Hitoshi Mitake <mitake.hitoshi at lab.ntt.co.jp>
---
 sheep/group.c      | 56 +++++++++++++++++++++++++++++++++++++++---------------
 sheep/ops.c        | 21 ++++++++++----------
 sheep/sheep_priv.h |  3 ++-
 sheep/vdi.c        | 17 +++++++++++++++--
 4 files changed, 69 insertions(+), 28 deletions(-)

diff --git a/sheep/group.c b/sheep/group.c
index 2b98a9b..e633bc3 100644
--- a/sheep/group.c
+++ b/sheep/group.c
@@ -694,43 +694,69 @@ struct cinfo_collection_work {
 
 static struct cinfo_collection_work *collect_work;
 
-static void cinfo_collection_work(struct work *work)
+static struct vdi_state *do_cinfo_collection_work(uint32_t epoch,
+						  struct sd_node *n,
+						  int *nr_vdi_states)
 {
+	struct vdi_state *vs = NULL;
+	unsigned int rlen = 512;
 	struct sd_req hdr;
 	struct sd_rsp *rsp = (struct sd_rsp *)&hdr;
+	int ret;
+
+	vs = xcalloc(rlen, sizeof(*vs));
+
+retry:
+	sd_init_req(&hdr, SD_OP_VDI_STATE_SNAPSHOT_CTL);
+	hdr.vdi_state_snapshot.get = 1;
+	hdr.vdi_state_snapshot.tgt_epoch = epoch;
+	hdr.data_length = rlen;
+
+	ret = sheep_exec_req(&n->nid, &hdr, (char *)vs);
+	if (ret == SD_RES_SUCCESS) {
+		sd_debug("succeed to obtain snapshot of vdi states");
+		*nr_vdi_states = rsp->data_length / sizeof(*vs);
+		return vs;
+	} else if (ret == SD_RES_BUFFER_SMALL) {
+		sd_debug("buffer is small for obtaining snapshot of vdi states,"
+			 " doubling it (%lu -> %lu)", rlen * sizeof(*vs),
+			 rlen * 2 * sizeof(*vs));
+		rlen *= 2;
+		vs = xrealloc(vs, sizeof(*vs) * rlen);
+		goto retry;
+	}
+
+	sd_err("failed to obtain snapshot of vdi states from node %s",
+	       node_to_str(n));
+	return NULL;
+}
+
+static void cinfo_collection_work(struct work *work)
+{
+	struct sd_req hdr;
 	struct vdi_state *vs = NULL;
-	unsigned int rlen;
 	struct cinfo_collection_work *w =
 		container_of(work, struct cinfo_collection_work, work);
 	struct sd_node *n;
-	int ret;
+	int ret, nr_vdi_states = 0;
 
 	sd_debug("start collection of cinfo...");
 
 	assert(w == collect_work);
 
-	rlen = SD_DATA_OBJ_SIZE; /* FIXME */
-	vs = xzalloc(rlen);
-
 	rb_for_each_entry(n, &w->members->nroot, rb) {
 		if (node_is_local(n))
 			continue;
 
-		sd_init_req(&hdr, SD_OP_VDI_STATE_SNAPSHOT_CTL);
-		hdr.vdi_state_snapshot.get = 1;
-		hdr.vdi_state_snapshot.tgt_epoch = w->epoch;
-		hdr.data_length = rlen;
-
-		ret = sheep_exec_req(&n->nid, &hdr, (char *)vs);
-		if (ret == SD_RES_SUCCESS)
+		vs = do_cinfo_collection_work(w->epoch, n, &nr_vdi_states);
+		if (vs)
 			goto get_succeed;
 	}
 
 	panic("getting a snapshot of vdi state at epoch %d failed", w->epoch);
 
 get_succeed:
-
-	w->nr_vdi_states = rsp->data_length / sizeof(*vs);
+	w->nr_vdi_states = nr_vdi_states;
 	w->result = vs;
 
 	sd_debug("collecting cinfo done, freeing from remote nodes");
diff --git a/sheep/ops.c b/sheep/ops.c
index a617a83..3991060 100644
--- a/sheep/ops.c
+++ b/sheep/ops.c
@@ -1391,21 +1391,22 @@ static int local_vdi_state_snapshot_ctl(const struct sd_req *req,
 {
 	bool get = !!req->vdi_state_snapshot.get;
 	int epoch = req->vdi_state_snapshot.tgt_epoch;
-	int ret;
+	int ret, length = 0;
 
 	sd_info("%s vdi state snapshot at epoch %d",
 		get ? "getting" : "freeing", epoch);
 
 	if (get) {
-		/*
-		 * FIXME: assuming request has enough space for storing
-		 * the snapshot
-		 */
-		ret = get_vdi_state_snapshot(epoch, data);
-		if (0 <= ret)
-			rsp->data_length = ret;
-		else
-			return SD_RES_AGAIN;
+		ret = get_vdi_state_snapshot(epoch, data, req->data_length,
+					     &length);
+		if (ret == SD_RES_SUCCESS)
+			rsp->data_length = length;
+		else {
+			sd_err("failed to get vdi state snapshot: %s",
+			       sd_strerror(ret));
+
+			return ret;
+		}
 	} else
 		free_vdi_state_snapshot(epoch);
 
diff --git a/sheep/sheep_priv.h b/sheep/sheep_priv.h
index 5fc6b90..72817b9 100644
--- a/sheep/sheep_priv.h
+++ b/sheep/sheep_priv.h
@@ -344,7 +344,8 @@ bool vdi_lock(uint32_t vid, const struct node_id *owner, int type);
 bool vdi_unlock(uint32_t vid, const struct node_id *owner, int type);
 void apply_vdi_lock_state(struct vdi_state *vs);
 void take_vdi_state_snapshot(int epoch);
-int get_vdi_state_snapshot(int epoch, void *data);
+int get_vdi_state_snapshot(int epoch, void *data, int data_len_max,
+			   int *data_len_result);
 void free_vdi_state_snapshot(int epoch);
 void log_vdi_op_lock(uint32_t vid, const struct node_id *owner, int type);
 void log_vdi_op_unlock(uint32_t vid, const struct node_id *owner, int type);
diff --git a/sheep/vdi.c b/sheep/vdi.c
index 1c8fb36..cf515e1 100644
--- a/sheep/vdi.c
+++ b/sheep/vdi.c
@@ -1793,9 +1793,22 @@ main_fn void take_vdi_state_snapshot(int epoch)
 	sd_debug("a number of vdi state: %d", snapshot->nr_vs);
 }
 
-main_fn int get_vdi_state_snapshot(int epoch, void *data)
+main_fn int get_vdi_state_snapshot(int epoch, void *data, int data_len_max,
+				   int *data_len_result)
 {
 	struct vdi_state_snapshot *snapshot;
+	int nr_states = 0;
+
+	list_for_each_entry(snapshot, &vdi_state_snapshot_list, list) {
+		if (snapshot->epoch == epoch)
+			nr_states++;
+	}
+
+	if (data_len_max < nr_states) {
+		sd_info("maximum allowed length: %d, a number of states: %d",
+			data_len_max, nr_states);
+		return SD_RES_BUFFER_SMALL;
+	}
 
 	list_for_each_entry(snapshot, &vdi_state_snapshot_list, list) {
 		if (snapshot->epoch == epoch) {
@@ -1807,7 +1820,7 @@ main_fn int get_vdi_state_snapshot(int epoch, void *data)
 
 	sd_info("get request for not prepared vdi state snapshot, epoch: %d",
 		epoch);
-	return -1;
+	return SD_RES_AGAIN;
 }
 
 main_fn void free_vdi_state_snapshot(int epoch)
-- 
1.8.3.2