[Sheepdog] [PATCH v5 4/8] sheep: teach sheep to use object cache

[Liu Yuan]

From: Liu Yuan <tailai.ly at taobao.com>

We only intrude IO code for gateway requests. Object IO path from
recovery logic is intact.

Signed-off-by: Liu Yuan <tailai.ly at taobao.com>
---
 include/sheepdog_proto.h |    1 +
 sheep/object_cache.c     |   11 ++++-----
 sheep/store.c            |   54 +++++++++++++++++++++++++++++++++++++++++----
 3 files changed, 55 insertions(+), 11 deletions(-)

diff --git a/include/sheepdog_proto.h b/include/sheepdog_proto.h
index 2d0d5ec..84f12f1 100644
--- a/include/sheepdog_proto.h
+++ b/include/sheepdog_proto.h
@@ -32,6 +32,7 @@
 
 #define SD_FLAG_CMD_WRITE    0x01
 #define SD_FLAG_CMD_COW      0x02
+#define SD_FLAG_CMD_CACHE    0x04
 
 #define SD_RES_SUCCESS       0x00 /* Success */
 #define SD_RES_UNKNOWN       0x01 /* Unknown error */
diff --git a/sheep/object_cache.c b/sheep/object_cache.c
index 929e28d..789a3ef 100644
--- a/sheep/object_cache.c
+++ b/sheep/object_cache.c
@@ -43,7 +43,7 @@ static inline int hash(uint64_t vid)
 	return hash_64(vid, HASH_BITS);
 }
 
-static struct object_cache_entry *dirty_rb_insert(struct rb_root *root,
+static struct object_cache_entry *dirty_tree_insert(struct rb_root *root,
 		struct object_cache_entry *new)
 {
 	struct rb_node **p = &root->rb_node;
@@ -68,7 +68,7 @@ static struct object_cache_entry *dirty_rb_insert(struct rb_root *root,
 }
 
 __attribute__ ((unused))
-static struct object_cache_entry *dirty_rb_search(struct rb_root *root,
+static struct object_cache_entry *dirty_tree_search(struct rb_root *root,
 		struct object_cache_entry *entry)
 {
 	struct rb_node *n = root->rb_node;
@@ -142,7 +142,6 @@ struct object_cache *find_object_cache(uint32_t vid)
 	return lookup_object_cache(vid, 1);
 }
 
-/* The caller is responsible to release fd */
 int object_cache_lookup(struct object_cache *oc, uint32_t idx)
 {
 	struct strbuf buf;
@@ -224,13 +223,13 @@ out:
 	return ret;
 }
 
-static void add_to_dirty_rb_and_list(struct object_cache *oc, uint32_t idx)
+static void add_to_dirty_tree_and_list(struct object_cache *oc, uint32_t idx)
 {
 	struct object_cache_entry *entry = xzalloc(sizeof(*entry));
 
 	entry->idx = idx;
 	pthread_mutex_lock(&oc->lock);
-	if (!dirty_rb_insert(&oc->dirty_rb, entry))
+	if (!dirty_tree_insert(&oc->dirty_rb, entry))
 		list_add(&entry->list, &oc->dirty_list);
 	else
 		free(entry);
@@ -248,7 +247,7 @@ int object_cache_rw(struct object_cache *oc, uint32_t idx, struct request *req)
 		ret = write_cache_object(oc->vid, idx, req->data, hdr->data_length, hdr->offset);
 		if (ret != SD_RES_SUCCESS)
 			goto out;
-		add_to_dirty_rb_and_list(oc, idx);
+		add_to_dirty_tree_and_list(oc, idx);
 	} else {
 		ret = read_cache_object(oc->vid, idx, req->data, hdr->data_length, hdr->offset);
 		if (ret != SD_RES_SUCCESS)
diff --git a/sheep/store.c b/sheep/store.c
index 82bbff1..d3127d6 100644
--- a/sheep/store.c
+++ b/sheep/store.c
@@ -758,6 +758,44 @@ out:
 	return ret;
 }
 
+static int handle_gateway_request(struct request *req)
+{
+	struct sd_obj_req *hdr = (struct sd_obj_req *)&req->rq;
+	uint64_t oid = hdr->oid;
+	uint32_t vid = oid_to_vid(oid);
+	uint32_t idx = data_oid_to_idx(oid);
+	struct object_cache *cache;
+	int ret;
+
+	if (is_vdi_obj(oid))
+		idx |= 1 << CACHE_VDI_SHIFT;
+
+	cache = find_object_cache(vid);
+	cache->oid = oid;
+	if (object_cache_lookup(cache, idx) < 0) {
+		ret = object_cache_pull(cache, idx);
+		if (ret != SD_RES_SUCCESS)
+			return ret;
+	}
+	return object_cache_rw(cache, idx, req);
+}
+
+static int bypass_object_cache(struct sd_obj_req *hdr)
+{
+	uint64_t oid = hdr->oid;
+
+	if (!(hdr->flags & SD_FLAG_CMD_CACHE))
+		return 1;
+
+	/* For create, we skip the cache because of consistency check.
+	 * For vmstate && vdi_attr object, we don't do caching
+	 */
+	if (hdr->opcode == SD_OP_CREATE_AND_WRITE_OBJ || is_vmstate_obj(oid)
+			|| is_vdi_attr_obj(oid))
+		return 1;
+	return 0;
+}
+
 void do_io_request(struct work *work)
 {
 	struct request *req = container_of(work, struct request, work);
@@ -782,11 +820,13 @@ void do_io_request(struct work *work)
 			if (ret != SD_RES_SUCCESS)
 				goto out;
 		}
-
-		if (hdr->flags & SD_FLAG_CMD_WRITE)
-			ret = forward_write_obj_req(req);
-		else
-			ret = forward_read_obj_req(req);
+		if (bypass_object_cache(hdr)) {
+			if (hdr->flags & SD_FLAG_CMD_WRITE)
+				ret = forward_write_obj_req(req);
+			else
+				ret = forward_read_obj_req(req);
+		} else
+			ret = handle_gateway_request(req);
 	}
 out:
 	if (ret != SD_RES_SUCCESS)
@@ -2009,6 +2049,10 @@ int init_store(const char *d)
 			return ret;
 	} else
 		dprintf("no store found\n");
+
+	ret = object_cache_init(d);
+	if (ret)
+		return 1;
 	return ret;
 }
 
-- 
1.7.8.2