[sheepdog] [PATCH 2/6] object cache: fix add_to_object_cache()

[Liu Yuan]

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

add_to_object_cache() and add_to_dirty_tree_and_list() should be operating
atomically, otherwise following race will happen:

	thread 1		thread 2
create A on object cache
add_to_object_cache(A)
				do_reclaim_object(A)
				remove_cache_object(A)
add_to_dirty_tree_and_list(A) <--- panic!

Signed-off-by: Liu Yuan <tailai.ly at taobao.com>
---
 sheep/object_cache.c |   28 ++++++++++++----------------
 1 file changed, 12 insertions(+), 16 deletions(-)

diff --git a/sheep/object_cache.c b/sheep/object_cache.c
index 1ebe896..e986bed 100644
--- a/sheep/object_cache.c
+++ b/sheep/object_cache.c
@@ -495,7 +495,7 @@ dirty_tree_and_list_insert(struct object_cache *oc, uint32_t idx,
 
 	entry = object_tree_search(&oc->object_tree, idx);
 	if (!entry)
-		return NULL;
+		panic("Can not find object entry %" PRIx32 "\n", idx);
 
 	entry->bmap |= bmap;
 	if (create)
@@ -566,8 +566,6 @@ add_to_dirty_tree_and_list(struct object_cache *oc, uint32_t idx,
 {
 	struct object_cache_entry *entry;
 	entry = dirty_tree_and_list_insert(oc, idx, bmap, create);
-	if (!entry)
-		panic("Can not find object entry %" PRIx32 "\n", idx);
 
 	if (cache_in_reclaim(0))
 		return;
@@ -606,7 +604,8 @@ void object_cache_try_to_reclaim(void)
 	queue_work(sys->reclaim_wqueue, work);
 }
 
-static void add_to_object_cache(struct object_cache *oc, uint32_t idx)
+static void add_to_object_cache(struct object_cache *oc, uint32_t idx,
+				int create)
 {
 	struct object_cache_entry *entry, *old;
 	uint32_t data_length;
@@ -633,6 +632,10 @@ static void add_to_object_cache(struct object_cache *oc, uint32_t idx)
 		free(entry);
 		entry = old;
 	}
+	if (create) {
+		uint64_t all = UINT64_MAX;
+		add_to_dirty_tree_and_list(oc, idx, all, create);
+	}
 	pthread_rwlock_unlock(&oc->lock);
 
 	object_cache_try_to_reclaim();
@@ -673,15 +676,8 @@ static int object_cache_lookup(struct object_cache *oc, uint32_t idx,
 	ret = prealloc(fd, data_length);
 	if (ret != SD_RES_SUCCESS)
 		ret = SD_RES_EIO;
-	else {
-		uint64_t bmap = UINT64_MAX;
-
-		add_to_object_cache(oc, idx);
-
-		pthread_rwlock_wrlock(&oc->lock);
-		add_to_dirty_tree_and_list(oc, idx, bmap, 1);
-		pthread_rwlock_unlock(&oc->lock);
-	}
+	else
+		add_to_object_cache(oc, idx, 1);
 	close(fd);
 out:
 	strbuf_release(&buf);
@@ -769,7 +765,7 @@ static int object_cache_pull(struct object_cache *oc, uint32_t idx)
 		dprintf("oid %"PRIx64" pulled successfully\n", oid);
 		ret = create_cache_object(oc, idx, buf, data_length);
 		if (ret == SD_RES_SUCCESS)
-			add_to_object_cache(oc, idx);
+			add_to_object_cache(oc, idx, 0);
 	}
 	free(buf);
 out:
@@ -798,7 +794,7 @@ static int object_cache_push(struct object_cache *oc)
 		del_from_dirty_tree_and_list(entry, &oc->dirty_tree);
 	}
 push_failed:
-	pthread_rwlock_wrlock(&oc->lock);
+	pthread_rwlock_unlock(&oc->lock);
 	return ret;
 }
 
@@ -1139,7 +1135,7 @@ static int load_existing_cache_object(struct object_cache *cache)
 		if (idx == ULLONG_MAX)
 			continue;
 
-		add_to_object_cache(cache, idx);
+		add_to_object_cache(cache, idx, 0);
 		dprintf("load cache %06" PRIx32 "/%08" PRIx32 "\n",
 			cache->vid, idx);
 	}
-- 
1.7.10.2