[sheepdog] [PATCH v8 11/19] sheep, dog: migrate deletion process from sheep to dog

[Hitoshi Mitake]

From: Hitoshi Mitake <mitake.hitoshi at lab.ntt.co.jp>

Deletion process also requires modification of generational reference
status so it must not be executed with snapshot creation or COW
request at the same time.

This patch migrates deletion process from sheep to dog. The new
deletion process in dog is simply implemented as COW: zero filling
sd_inode->data_vdi_id[] and write it to sheep cluster. The GC code can
decrement reference by this inode update.

Cc: MORITA Kazutaka <morita.kazutaka at lab.ntt.co.jp>
Cc: Valerio Pachera <sirio81 at gmail.com>
Cc: Alessandro Bolgia <alessandro at extensys.it>
Signed-off-by: Hitoshi Mitake <mitake.hitoshi at lab.ntt.co.jp>
---
 dog/vdi.c   | 66 ++++++++++++++++++++++++++++++++++++++++++++++++++++++-------
 sheep/vdi.c | 18 +++++------------
 2 files changed, 64 insertions(+), 20 deletions(-)

diff --git a/dog/vdi.c b/dog/vdi.c
index 16630ac..6ef0fc1 100644
--- a/dog/vdi.c
+++ b/dog/vdi.c
@@ -697,18 +697,23 @@ static int vdi_resize(int argc, char **argv)
 	return EXIT_SUCCESS;
 }
 
+#define NR_BATCHED_DISCARD 128	/* TODO: the value should be optional */
+
 static int do_vdi_delete(const char *vdiname, int snap_id, const char *snap_tag)
 {
-	int ret;
+	int ret, nr_objs;
 	struct sd_req hdr;
 	struct sd_rsp *rsp = (struct sd_rsp *)&hdr;
 	char data[SD_MAX_VDI_LEN + SD_MAX_VDI_TAG_LEN];
 	uint32_t vid;
+	struct sd_inode *inode = xzalloc(sizeof(*inode));
+	int i = 0;
 
 	ret = find_vdi_name(vdiname, snap_id, snap_tag, &vid, 0);
 	if (ret < 0) {
 		sd_err("Failed to open VDI %s", vdiname);
-		return EXIT_FAILURE;
+		ret = EXIT_FAILURE;
+		goto out;
 	}
 
 	sd_init_req(&hdr, SD_OP_DELETE_CACHE);
@@ -717,7 +722,50 @@ static int do_vdi_delete(const char *vdiname, int snap_id, const char *snap_tag)
 	ret = send_light_req(&sd_nid, &hdr);
 	if (ret) {
 		sd_err("failed to execute request");
-		return EXIT_FAILURE;
+		ret = EXIT_FAILURE;
+		goto out;
+	}
+
+	ret = dog_read_object(vid_to_vdi_oid(vid), inode, sizeof(*inode),
+			      0, false);
+	if (ret) {
+		sd_err("failed to read inode object: %"PRIx64,
+		       vid_to_vdi_oid(vid));
+		ret = EXIT_FAILURE;
+		goto out;
+	}
+
+	nr_objs = count_data_objs(inode);
+	while (i < nr_objs) {
+		int start_idx, nr_filled_idx;
+
+		while (i < nr_objs && !inode->data_vdi_id[i])
+			i++;
+		start_idx = i;
+
+		nr_filled_idx = 0;
+		while (i < nr_objs && nr_filled_idx < NR_BATCHED_DISCARD) {
+			if (inode->data_vdi_id[i]) {
+				inode->data_vdi_id[i] = 0;
+				nr_filled_idx++;
+			}
+
+			i++;
+		}
+
+		ret = dog_write_object(vid_to_vdi_oid(vid), 0,
+				       &inode->data_vdi_id[start_idx],
+				       (i - start_idx) * sizeof(uint32_t),
+				       offsetof(struct sd_inode,
+						data_vdi_id[start_idx]),
+				       0, inode->nr_copies, inode->copy_policy,
+				       false, true);
+		if (ret) {
+			sd_err("failed to update inode for discarding objects:"
+			       " %"PRIx64, vid_to_vdi_oid(vid));
+			ret = EXIT_FAILURE;
+			goto out;
+		}
 	}
 
 	sd_init_req(&hdr, SD_OP_DEL_VDI);
@@ -730,18 +778,22 @@ static int do_vdi_delete(const char *vdiname, int snap_id, const char *snap_tag)
 		pstrcpy(data + SD_MAX_VDI_LEN, SD_MAX_VDI_TAG_LEN, snap_tag);
 
 	ret = dog_exec_req(&sd_nid, &hdr, data);
-	if (ret < 0)
-		return EXIT_SYSFAIL;
+	if (ret < 0) {
+		ret = EXIT_SYSFAIL;
+		goto out;
+	}
 
 	if (rsp->result != SD_RES_SUCCESS) {
 		sd_err("Failed to delete %s: %s", vdiname,
 		       sd_strerror(rsp->result));
 		if (rsp->result == SD_RES_NO_VDI)
-			return EXIT_MISSING;
+			ret = EXIT_MISSING;
 		else
-			return EXIT_FAILURE;
+			ret = EXIT_FAILURE;
 	}
 
+out:
+	free(inode);
 	return EXIT_SUCCESS;
 }
 
diff --git a/sheep/vdi.c b/sheep/vdi.c
index 4d5d635..460d694 100644
--- a/sheep/vdi.c
+++ b/sheep/vdi.c
@@ -849,21 +849,13 @@ static void delete_vdi_work(struct work *work)
 	if (inode->store_policy == 0) {
 		nr_objs = count_data_objs(inode);
 		for (nr_deleted = 0, i = 0; i < nr_objs; i++) {
-			uint64_t oid;
 			uint32_t vid = sd_inode_get_vid(inode, i);
 
-			if (!vid)
-				continue;
-
-			oid = vid_to_data_oid(vid, i);
-			ret = sd_dec_object_refcnt(oid,
-						inode->gref[i].generation,
-						inode->gref[i].count);
-			if (ret != SD_RES_SUCCESS)
-				sd_err("discard ref %" PRIx64 " fail, %d",
-				       oid, ret);
-
-			nr_deleted++;
+			if (vid) {
+				sd_err("vid: %"PRIx32" still has objects", vid);
+				dw->succeed = false;
+				goto out;
+			}
 		}
 	} else {
 		/*
-- 
1.9.1