[sheepdog] [PATCH] object cache: use read/write lock

[Liu Yuan]

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

This allow concurrent read access to the same object, which is crucial to cloned
VMs which shared one base in the same node.

Signed-off-by: Liu Yuan <tailai.ly at taobao.com>
---
 include/util.h       |   11 -----------
 sheep/object_cache.c |   52 ++++++++++++++++++++++++++++++++++++++++----------
 2 files changed, 42 insertions(+), 21 deletions(-)

diff --git a/include/util.h b/include/util.h
index 0747ba4..eafd9ca 100644
--- a/include/util.h
+++ b/include/util.h
@@ -60,17 +60,6 @@ static inline void *zalloc(size_t size)
 	return calloc(1, size);
 }
 
-static inline int xlockf(int fd, int cmd, off_t offset, off_t len)
-{
-	if (lseek(fd, offset, SEEK_SET) < 0) {
-		eprintf("%m\n");
-		return -1;
-	}
-
-	return lockf(fd, cmd, len);
-}
-
-
 typedef void (*try_to_free_t)(size_t);
 extern try_to_free_t set_try_to_free_routine(try_to_free_t);
 
diff --git a/sheep/object_cache.c b/sheep/object_cache.c
index ce30c17..305f931 100644
--- a/sheep/object_cache.c
+++ b/sheep/object_cache.c
@@ -264,6 +264,7 @@ static int write_cache_object(uint32_t vid, uint32_t idx, void *buf, size_t coun
 	size_t size;
 	int fd, flags = def_open_flags, ret = SD_RES_SUCCESS;
 	struct strbuf p;
+	struct flock fl;
 
 	strbuf_init(&p, PATH_MAX);
 	strbuf_addstr(&p, cache_dir);
@@ -273,24 +274,38 @@ static int write_cache_object(uint32_t vid, uint32_t idx, void *buf, size_t coun
 		flags |= O_DIRECT;
 
 	fd = open(p.buf, flags, def_fmode);
-	if (xlockf(fd, F_LOCK, offset, count) < 0) {
-		ret = SD_RES_EIO;
+	if (fd < 0) {
 		eprintf("%m\n");
+		ret = SD_RES_EIO;
 		goto out;
 	}
+
+	fl.l_type = F_WRLCK;
+	fl.l_whence = SEEK_SET;
+	fl.l_start = offset;
+	fl.l_len = count;
+	fl.l_pid = getpid();
+	if (fcntl(fd, F_SETLKW, &fl) < 0) {
+		eprintf("%m\n");
+		ret = SD_RES_EIO;
+		goto out_close;
+	}
 	size = xpwrite(fd, buf, count, offset);
-	if (xlockf(fd, F_ULOCK, offset, count) < 0) {
+
+	fl.l_type = F_UNLCK;
+	if (fcntl(fd, F_SETLK, &fl) < 0) {
 		ret = SD_RES_EIO;
 		eprintf("%m\n");
-		goto out;
+		goto out_close;
 	}
 	if (size != count) {
 		eprintf("size %zu, count:%zu, offset %zu %m\n",
 			size, count, offset);
 		ret = SD_RES_EIO;
 	}
-out:
+out_close:
 	close(fd);
+out:
 	strbuf_release(&p);
 	return ret;
 }
@@ -300,6 +315,7 @@ static int read_cache_object(uint32_t vid, uint32_t idx, void *buf, size_t count
 	size_t size;
 	int fd, flags = def_open_flags, ret = SD_RES_SUCCESS;
 	struct strbuf p;
+	struct flock fl;
 
 	strbuf_init(&p, PATH_MAX);
 	strbuf_addstr(&p, cache_dir);
@@ -309,25 +325,41 @@ static int read_cache_object(uint32_t vid, uint32_t idx, void *buf, size_t count
 		flags |= O_DIRECT;
 
 	fd = open(p.buf, flags, def_fmode);
-	if (xlockf(fd, F_LOCK, offset, count) < 0) {
-		ret = SD_RES_EIO;
+	if (fd < 0) {
 		eprintf("%m\n");
+		ret = SD_RES_EIO;
 		goto out;
 	}
 
+	fl.l_type = F_RDLCK;
+	fl.l_whence = SEEK_SET;
+	fl.l_start = offset;
+	fl.l_len = count;
+	fl.l_pid = getpid();
+	if (fcntl(fd, F_SETLKW, &fl) < 0) {
+		eprintf("%m\n");
+		ret = SD_RES_EIO;
+		goto out_close;
+	}
+
 	size = xpread(fd, buf, count, offset);
-	if (xlockf(fd, F_ULOCK, offset, count) < 0) {
+
+	fl.l_type = F_UNLCK;
+	if (fcntl(fd, F_SETLK, &fl) < 0) {
 		ret = SD_RES_EIO;
 		eprintf("%m\n");
-		goto out;
+		goto out_close;
 	}
+
 	if (size != count) {
 		eprintf("size %zu, count:%zu, offset %zu %m\n",
 			size, count, offset);
 		ret = SD_RES_EIO;
 	}
-out:
+
+out_close:
 	close(fd);
+out:
 	strbuf_release(&p);
 	return ret;
 }
-- 
1.7.10.2