[sheepdog] [PATCH 1/2] sheep: rename files for store_driver

Saeki Masaki saeki.masaki at po.ntts.co.jp
Tue Mar 17 10:00:53 CET 2015


This change is a preparation patch for add store_driver.
Put files together to the new folder.

and fix little style problem in md.c

Signed-off-by: Masaki Saeki <saeki.masaki at po.ntts.co.jp>
---
 sheep/Makefile.am         |    6 +-
 sheep/md.c                |  878 ---------------------------------------------
 sheep/plain_store.c       |  759 ---------------------------------------
 sheep/store.c             |  511 --------------------------
 sheep/store/common.c      |  511 ++++++++++++++++++++++++++
 sheep/store/md.c          |  878 +++++++++++++++++++++++++++++++++++++++++++++
 sheep/store/plain_store.c |  759 +++++++++++++++++++++++++++++++++++++++
 7 files changed, 2152 insertions(+), 2150 deletions(-)
 delete mode 100644 sheep/md.c
 delete mode 100644 sheep/plain_store.c
 delete mode 100644 sheep/store.c
 create mode 100644 sheep/store/common.c
 create mode 100644 sheep/store/md.c
 create mode 100644 sheep/store/plain_store.c

diff --git a/sheep/Makefile.am b/sheep/Makefile.am
index 7a08838..3ddd761 100644
--- a/sheep/Makefile.am
+++ b/sheep/Makefile.am
@@ -24,10 +24,12 @@ AM_CPPFLAGS		= -I$(top_builddir)/include -I$(top_srcdir)/include \
 
 sbin_PROGRAMS		= sheep
 
-sheep_SOURCES		= sheep.c group.c request.c gateway.c store.c vdi.c \
+sheep_SOURCES		= sheep.c group.c request.c gateway.c vdi.c \
 			  journal.c ops.c recovery.c cluster/local.c \
 			  object_cache.c object_list_cache.c \
-			  plain_store.c config.c migrate.c md.c
+			  store/common.c store/md.c \
+			  store/plain_store.c \
+			  config.c migrate.c
 
 if BUILD_HTTP
 sheep_SOURCES		+= http/http.c http/kv.c http/s3.c http/swift.c \
diff --git a/sheep/md.c b/sheep/md.c
deleted file mode 100644
index c00d7a5..0000000
--- a/sheep/md.c
+++ /dev/null
@@ -1,878 +0,0 @@
-/*
- * Copyright (C) 2013 Taobao Inc.
- *
- * Liu Yuan <namei.unix at gmail.com>
- *
- * This program is free software; you can redistribute it and/or
- * modify it under the terms of the GNU General Public License version
- * 2 as published by the Free Software Foundation.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#include "sheep_priv.h"
-
-#define MD_VDISK_SIZE ((uint64_t)1*1024*1024*1024) /* 1G */
-
-#define NONE_EXIST_PATH "/all/disks/are/broken/,ps/ノ厂経7/!"
-
-struct md md = {
-	.vroot = RB_ROOT,
-	.root = RB_ROOT,
-	.lock = SD_RW_LOCK_INITIALIZER,
-};
-
-static inline uint32_t nr_online_disks(void)
-{
-	uint32_t nr;
-
-	sd_read_lock(&md.lock);
-	nr = md.nr_disks;
-	sd_rw_unlock(&md.lock);
-
-	return nr;
-}
-
-static inline int vdisk_number(const struct disk *disk)
-{
-	return DIV_ROUND_UP(disk->space, MD_VDISK_SIZE);
-}
-
-static int disk_cmp(const struct disk *d1, const struct disk *d2)
-{
-	return strcmp(d1->path, d2->path);
-}
-
-static int vdisk_cmp(const struct vdisk *d1, const struct vdisk *d2)
-{
-	return intcmp(d1->hash, d2->hash);
-}
-
-static struct vdisk *vdisk_insert(struct vdisk *new)
-{
-	return rb_insert(&md.vroot, new, rb, vdisk_cmp);
-}
-
-/* If v1_hash < hval <= v2_hash, then oid is resident in v2 */
-static struct vdisk *hval_to_vdisk(uint64_t hval)
-{
-	struct vdisk dummy = { .hash = hval };
-
-	return rb_nsearch(&md.vroot, &dummy, rb, vdisk_cmp);
-}
-
-static struct vdisk *oid_to_vdisk(uint64_t oid)
-{
-	return hval_to_vdisk(sd_hash_oid(oid));
-}
-
-static void create_vdisks(const struct disk *disk)
-{
-	uint64_t hval = sd_hash(disk->path, strlen(disk->path));
-	const struct sd_node *n = &sys->this_node;
-	uint64_t node_hval;
-	int nr;
-
-	if (is_cluster_diskmode(&sys->cinfo)) {
-		node_hval = sd_hash(&n->nid, offsetof(typeof(n->nid), io_addr));
-		hval = fnv_64a_64(node_hval, hval);
-		nr = DIV_ROUND_UP(disk->space, WEIGHT_MIN);
-		if (0 == n->nid.port)
-			return;
-	} else
-		nr = vdisk_number(disk);
-
-	for (int i = 0; i < nr; i++) {
-		struct vdisk *v = xmalloc(sizeof(*v));
-
-		hval = sd_hash_next(hval);
-		v->hash = hval;
-		v->disk = disk;
-		if (unlikely(vdisk_insert(v)))
-			panic("vdisk hash collison");
-	}
-}
-
-static inline void vdisk_free(struct vdisk *v)
-{
-	rb_erase(&v->rb, &md.vroot);
-	free(v);
-}
-
-static void remove_vdisks(const struct disk *disk)
-{
-	uint64_t hval = sd_hash(disk->path, strlen(disk->path));
-	const struct sd_node *n = &sys->this_node;
-	uint64_t node_hval;
-	int nr;
-
-	if (is_cluster_diskmode(&sys->cinfo)) {
-		node_hval = sd_hash(&n->nid, offsetof(typeof(n->nid), io_addr));
-		hval = fnv_64a_64(node_hval, hval);
-		nr = DIV_ROUND_UP(disk->space, WEIGHT_MIN);
-	} else
-		nr = vdisk_number(disk);
-
-	for (int i = 0; i < nr; i++) {
-		struct vdisk *v;
-
-		hval = sd_hash_next(hval);
-		v = hval_to_vdisk(hval);
-		assert(v->hash == hval);
-
-		vdisk_free(v);
-	}
-}
-
-static inline void trim_last_slash(char *path)
-{
-	assert(path[0]);
-	while (path[strlen(path) - 1] == '/')
-		path[strlen(path) - 1] = '\0';
-}
-
-static struct disk *path_to_disk(const char *path)
-{
-	struct disk key = {};
-
-	pstrcpy(key.path, sizeof(key.path), path);
-	trim_last_slash(key.path);
-
-	return rb_search(&md.root, &key, rb, disk_cmp);
-}
-
-size_t get_store_objsize(uint64_t oid)
-{
-	if (is_erasure_oid(oid)) {
-		uint8_t policy = get_vdi_copy_policy(oid_to_vid(oid));
-		int d;
-		ec_policy_to_dp(policy, &d, NULL);
-		return get_vdi_object_size(oid_to_vid(oid)) / d;
-	}
-	return get_objsize(oid, get_vdi_object_size(oid_to_vid(oid)));
-}
-
-static int get_total_object_size(uint64_t oid, const char *wd, uint32_t epoch,
-				 uint8_t ec_index, struct vnode_info *vinfo,
-				 void *total)
-{
-	uint64_t *t = total;
-	struct stat s;
-	char path[PATH_MAX];
-
-	snprintf(path, PATH_MAX, "%s/%016" PRIx64, wd, oid);
-	if (stat(path, &s) == 0)
-		*t += s.st_blocks * SECTOR_SIZE;
-	else
-		*t += get_store_objsize(oid);
-
-	return SD_RES_SUCCESS;
-}
-
-static int64_t find_string_integer(const char *str, const char *delimiter)
-{
-	char *pos = strstr(str, delimiter), *p;
-	int64_t ret;
-
-	ret = strtoll(pos + 1, &p, 10);
-	if (ret == LLONG_MAX || p == pos + 1) {
-		sd_err("%s strtoul failed, delimiter %s, %m", str, delimiter);
-		return -1;
-	}
-
-	return ret;
-}
-
-/* If cleanup is true, temporary objects will be removed */
-static int for_each_object_in_path(const char *path,
-				   int (*func)(uint64_t, const char *, uint32_t,
-					       uint8_t, struct vnode_info *,
-					       void *),
-				   bool cleanup, struct vnode_info *vinfo,
-				   void *arg)
-{
-	DIR *dir;
-	struct dirent *d;
-	uint64_t oid;
-	int ret = SD_RES_SUCCESS;
-	char file_name[PATH_MAX];
-
-	dir = opendir(path);
-	if (unlikely(!dir)) {
-		sd_err("failed to open %s, %m", path);
-		return SD_RES_EIO;
-	}
-
-	while ((d = readdir(dir))) {
-		uint32_t epoch = 0;
-		uint8_t ec_index = SD_MAX_COPIES;
-
-		/* skip ".", ".." and ".stale" */
-		if (unlikely(!strncmp(d->d_name, ".", 1)))
-			continue;
-
-		sd_debug("%s, %s", path, d->d_name);
-		oid = strtoull(d->d_name, NULL, 16);
-		if (oid == 0 || oid == ULLONG_MAX)
-			continue;
-
-		/* don't call callback against temporary objects */
-		if (is_tmp_dentry(d->d_name)) {
-			if (cleanup) {
-				snprintf(file_name, sizeof(file_name),
-						"%s/%s", path, d->d_name);
-				sd_debug("remove tmp object %s", file_name);
-				if (unlink(file_name) < 0)
-					sd_err("failed to unlink %s: %m",
-							file_name);
-			}
-			continue;
-		}
-
-		if (is_stale_dentry(d->d_name)) {
-			epoch = find_string_integer(d->d_name, ".");
-			if (epoch < 0)
-				continue;
-		}
-
-		if (is_ec_dentry(d->d_name)) {
-			ec_index = find_string_integer(d->d_name, "_");
-			if (ec_index < 0)
-				continue;
-		}
-
-		ret = func(oid, path, epoch, ec_index, vinfo, arg);
-		if (ret != SD_RES_SUCCESS)
-			break;
-	}
-	closedir(dir);
-	return ret;
-}
-
-static uint64_t get_path_free_size(const char *path, uint64_t *used)
-{
-	struct statvfs fs;
-	uint64_t size;
-
-	if (statvfs(path, &fs) < 0) {
-		sd_err("get disk %s space failed %m", path);
-		return 0;
-	}
-	size = (int64_t)fs.f_frsize * fs.f_bavail;
-
-	if (!used)
-		goto out;
-	if (for_each_object_in_path(path, get_total_object_size, false,
-				    NULL, used)
-	    != SD_RES_SUCCESS)
-		return 0;
-out:
-	return size;
-}
-
-/*
- * If path is broken during initialization or not support xattr return 0. We can
- * safely use 0 to represent failure case  because 0 space path can be
- * considered as broken path.
- */
-static uint64_t init_path_space(const char *path, bool purge)
-{
-	uint64_t size;
-	char stale[PATH_MAX];
-
-	if (!is_xattr_enabled(path)) {
-		sd_warn("multi-disk support need xattr feature for path: %s",
-			path);
-		goto broken_path;
-	}
-
-	if (purge && purge_directory(path) < 0)
-		sd_err("failed to purge %s", path);
-
-	snprintf(stale, PATH_MAX, "%s/.stale", path);
-	if (xmkdir(stale, sd_def_dmode) < 0) {
-		sd_err("can't mkdir for %s, %m", stale);
-		goto broken_path;
-	}
-
-#define MDNAME	"user.md.size"
-#define MDSIZE	sizeof(uint64_t)
-	if (getxattr(path, MDNAME, &size, MDSIZE) < 0) {
-		if (errno == ENODATA) {
-			goto create;
-		} else {
-			sd_err("%s, %m", path);
-			goto broken_path;
-		}
-	}
-
-	return size;
-create:
-	size = get_path_free_size(path, NULL);
-	if (!size)
-		goto broken_path;
-	if (setxattr(path, MDNAME, &size, MDSIZE, 0) < 0) {
-		sd_err("%s, %m", path);
-		goto broken_path;
-	}
-	return size;
-broken_path:
-	return 0;
-}
-
-/* We don't need lock at init stage */
-bool md_add_disk(const char *path, bool purge)
-{
-	struct disk *new;
-
-	if (path_to_disk(path)) {
-		sd_err("duplicate path %s", path);
-		return false;
-	}
-
-	if (xmkdir(path, sd_def_dmode) < 0) {
-		sd_err("can't mkdir for %s, %m", path);
-		return false;
-	}
-
-	new = xmalloc(sizeof(*new));
-	pstrcpy(new->path, PATH_MAX, path);
-	trim_last_slash(new->path);
-	new->space = init_path_space(new->path, purge);
-	if (!new->space) {
-		free(new);
-		return false;
-	}
-
-	create_vdisks(new);
-	rb_insert(&md.root, new, rb, disk_cmp);
-	md.space += new->space;
-	md.nr_disks++;
-
-	sd_info("%s, vdisk nr %d, total disk %d", new->path, vdisk_number(new),
-		md.nr_disks);
-	return true;
-}
-
-static inline void md_remove_disk(struct disk *disk)
-{
-	sd_info("%s from multi-disk array", disk->path);
-	rb_erase(&disk->rb, &md.root);
-	md.nr_disks--;
-	remove_vdisks(disk);
-	free(disk);
-}
-
-uint64_t md_init_space(void)
-{
-	return md.space;
-}
-
-static const char *md_get_object_dir_nolock(uint64_t oid)
-{
-	const struct vdisk *vd;
-
-	if (unlikely(md.nr_disks == 0))
-		return NONE_EXIST_PATH; /* To generate EIO */
-
-	vd = oid_to_vdisk(oid);
-	return vd->disk->path;
-}
-
-const char *md_get_object_dir(uint64_t oid)
-{
-	const char *p;
-
-	sd_read_lock(&md.lock);
-	p = md_get_object_dir_nolock(oid);
-	sd_rw_unlock(&md.lock);
-
-	return p;
-}
-
-struct process_path_arg {
-	const char *path;
-	struct vnode_info *vinfo;
-	int (*func)(uint64_t oid, const char *, uint32_t, uint8_t,
-		    struct vnode_info *, void *arg);
-	bool cleanup;
-	void *opaque;
-	int result;
-};
-
-static void *thread_process_path(void *arg)
-{
-	int ret;
-	struct process_path_arg *parg = (struct process_path_arg *)arg;
-
-	ret = for_each_object_in_path(parg->path, parg->func, parg->cleanup,
-				      parg->vinfo, parg->opaque);
-	if (ret != SD_RES_SUCCESS)
-		parg->result = ret;
-
-	return arg;
-}
-
-main_fn int for_each_object_in_wd(int (*func)(uint64_t oid, const char *path,
-				      uint32_t epoch, uint8_t ec_index,
-				      struct vnode_info *vinfo, void *arg),
-				  bool cleanup, void *arg)
-{
-	int ret = SD_RES_SUCCESS;
-	const struct disk *disk;
-	struct process_path_arg *thread_args, *path_arg;
-	struct vnode_info *vinfo;
-	void *ret_arg;
-	sd_thread_t *thread_array;
-	int nr_thread = 0, idx = 0;
-
-	sd_read_lock(&md.lock);
-
-	rb_for_each_entry(disk, &md.root, rb) {
-		nr_thread++;
-	}
-
-	thread_args = xmalloc(nr_thread * sizeof(struct process_path_arg));
-	thread_array = xmalloc(nr_thread * sizeof(sd_thread_t));
-
-	vinfo = get_vnode_info();
-
-	rb_for_each_entry(disk, &md.root, rb) {
-		thread_args[idx].path = disk->path;
-		thread_args[idx].vinfo = vinfo;
-		thread_args[idx].func = func;
-		thread_args[idx].cleanup = cleanup;
-		thread_args[idx].opaque = arg;
-		thread_args[idx].result = SD_RES_SUCCESS;
-		ret = sd_thread_create_with_idx("foreach wd",
-						thread_array + idx,
-						thread_process_path,
-						(void *)(thread_args + idx));
-		if (ret) {
-			/*
-			 * If we can't create enough threads to process
-			 * files, the data-consistent will be broken if
-			 * we continued.
-			 */
-			panic("Failed to create thread for path %s",
-			      disk->path);
-		}
-		idx++;
-	}
-
-	sd_debug("Create %d threads for all path", nr_thread);
-	/* wait for all threads to exit */
-	for (idx = 0; idx < nr_thread; idx++) {
-		ret = sd_thread_join(thread_array[idx], &ret_arg);
-		if (ret)
-			sd_err("Failed to join thread");
-		if (ret_arg) {
-			path_arg = (struct process_path_arg *)ret_arg;
-			if (path_arg->result != SD_RES_SUCCESS)
-				sd_err("%s, %s", path_arg->path,
-				       sd_strerror(path_arg->result));
-		}
-	}
-
-	put_vnode_info(vinfo);
-	sd_rw_unlock(&md.lock);
-
-	free(thread_args);
-	free(thread_array);
-	return ret;
-}
-
-int for_each_object_in_stale(int (*func)(uint64_t oid, const char *path,
-					 uint32_t epoch, uint8_t,
-					 struct vnode_info *, void *arg),
-			     void *arg)
-{
-	int ret = SD_RES_SUCCESS;
-	char path[PATH_MAX];
-	const struct disk *disk;
-
-	sd_read_lock(&md.lock);
-	rb_for_each_entry(disk, &md.root, rb) {
-		snprintf(path, sizeof(path), "%s/.stale", disk->path);
-		ret = for_each_object_in_path(path, func, false, NULL, arg);
-		if (ret != SD_RES_SUCCESS)
-			break;
-	}
-	sd_rw_unlock(&md.lock);
-	return ret;
-}
-
-
-int for_each_obj_path(int (*func)(const char *path))
-{
-	int ret = SD_RES_SUCCESS;
-	const struct disk *disk;
-
-	sd_read_lock(&md.lock);
-	rb_for_each_entry(disk, &md.root, rb) {
-		ret = func(disk->path);
-		if (ret != SD_RES_SUCCESS)
-			break;
-	}
-	sd_rw_unlock(&md.lock);
-	return ret;
-}
-
-struct md_work {
-	struct work work;
-	char path[PATH_MAX];
-};
-
-static inline void kick_recover(void)
-{
-	struct vnode_info *vinfo = get_vnode_info();
-
-	if (is_cluster_diskmode(&sys->cinfo))
-		sys->cdrv->update_node(&sys->this_node);
-	else {
-		start_recovery(vinfo, vinfo, false);
-		put_vnode_info(vinfo);
-	}
-}
-
-static void md_do_recover(struct work *work)
-{
-	struct md_work *mw = container_of(work, struct md_work, work);
-	struct disk *disk;
-	int nr = 0;
-
-	sd_write_lock(&md.lock);
-	disk = path_to_disk(mw->path);
-	if (!disk)
-		/* Just ignore the duplicate EIO of the same path */
-		goto out;
-	md_remove_disk(disk);
-	nr = md.nr_disks;
-out:
-	sd_rw_unlock(&md.lock);
-
-	if (disk) {
-		if (nr > 0) {
-			update_node_disks();
-			kick_recover();
-		} else {
-			leave_cluster();
-		}
-	}
-
-	free(mw);
-}
-
-int md_handle_eio(const char *fault_path)
-{
-	struct md_work *mw;
-
-	if (nr_online_disks() == 0)
-		return SD_RES_EIO;
-
-	mw = xzalloc(sizeof(*mw));
-	mw->work.done = md_do_recover;
-	pstrcpy(mw->path, PATH_MAX, fault_path);
-	queue_work(sys->md_wqueue, &mw->work);
-
-	/* Fool the requester to retry */
-	return SD_RES_NETWORK_ERROR;
-}
-
-static inline bool md_access(const char *path)
-{
-	if (access(path, R_OK | W_OK) < 0) {
-		if (unlikely(errno != ENOENT))
-			sd_err("failed to check %s, %m", path);
-		return false;
-	}
-
-	return true;
-}
-
-static int get_old_new_path(uint64_t oid, uint32_t epoch, uint8_t ec_index,
-			    const char *path, char *old, char *new)
-{
-	if (!epoch) {
-		if (!is_erasure_oid(oid)) {
-			snprintf(old, PATH_MAX, "%s/%016" PRIx64, path, oid);
-			snprintf(new, PATH_MAX, "%s/%016" PRIx64,
-				 md_get_object_dir_nolock(oid), oid);
-		} else {
-			snprintf(old, PATH_MAX, "%s/%016" PRIx64"_%d", path,
-				 oid, ec_index);
-			snprintf(new, PATH_MAX, "%s/%016" PRIx64"_%d",
-				 md_get_object_dir_nolock(oid), oid, ec_index);
-		}
-	} else {
-		if (!is_erasure_oid(oid)) {
-			snprintf(old, PATH_MAX,
-				 "%s/.stale/%016"PRIx64".%"PRIu32, path,
-				 oid, epoch);
-			snprintf(new, PATH_MAX,
-				 "%s/.stale/%016"PRIx64".%"PRIu32,
-				 md_get_object_dir_nolock(oid), oid, epoch);
-		} else {
-			snprintf(old, PATH_MAX,
-				 "%s/.stale/%016"PRIx64"_%d.%"PRIu32, path,
-				 oid, ec_index, epoch);
-			snprintf(new, PATH_MAX,
-				 "%s/.stale/%016"PRIx64"_%d.%"PRIu32,
-				 md_get_object_dir_nolock(oid),
-				 oid, ec_index ,epoch);
-		}
-	}
-
-	if (!md_access(old))
-		return -1;
-
-	return 0;
-}
-
-static int md_move_object(uint64_t oid, const char *old, const char *new)
-{
-	struct strbuf buf = STRBUF_INIT;
-	int fd, ret = -1;
-	size_t sz = get_store_objsize(oid);
-
-	fd = open(old, O_RDONLY);
-	if (fd < 0) {
-		sd_err("failed to open %s", old);
-		goto out;
-	}
-
-	ret = strbuf_read(&buf, fd, sz);
-	if (ret != sz) {
-		sd_err("failed to read %s, size %zu, %d, %m", old, sz, ret);
-		ret = -1;
-		goto out_close;
-	}
-
-	if (atomic_create_and_write(new, buf.buf, buf.len, false) < 0) {
-		if (errno != EEXIST) {
-			sd_err("failed to create %s", new);
-			ret = -1;
-			goto out_close;
-		}
-	}
-	unlink(old);
-	ret = 0;
-out_close:
-	close(fd);
-out:
-	strbuf_release(&buf);
-	return ret;
-}
-
-static int md_check_and_move(uint64_t oid, uint32_t epoch, uint8_t ec_index,
-			     const char *path)
-{
-	char old[PATH_MAX], new[PATH_MAX];
-
-	if (get_old_new_path(oid, epoch, ec_index, path, old, new) < 0)
-		return SD_RES_EIO;
-	/*
-	 * Recovery thread and main thread might try to recover the same object.
-	 * Either one succeeds, the other will fail and proceed and end up
-	 * trying to move the object to where it is already in place, in this
-	 * case we simply return.
-	 */
-	if (!strcmp(old, new))
-		return SD_RES_SUCCESS;
-
-	/* We can't use rename(2) across device */
-	if (md_move_object(oid, old, new) < 0) {
-		sd_err("move old %s to new %s failed", old, new);
-		return SD_RES_EIO;
-	}
-
-	sd_debug("from %s to %s", old, new);
-	return SD_RES_SUCCESS;
-}
-
-static int scan_wd(uint64_t oid, uint32_t epoch, uint8_t ec_index)
-{
-	int ret = SD_RES_EIO;
-	const struct disk *disk;
-
-	sd_read_lock(&md.lock);
-	rb_for_each_entry(disk, &md.root, rb) {
-		ret = md_check_and_move(oid, epoch, ec_index, disk->path);
-		if (ret == SD_RES_SUCCESS)
-			break;
-	}
-	sd_rw_unlock(&md.lock);
-	return ret;
-}
-
-bool md_exist(uint64_t oid, uint8_t ec_index, char *path)
-{
-	if (md_access(path))
-		return true;
-	/*
-	 * We have to iterate the WD because we don't have epoch-like history
-	 * track to locate the objects for multiple disk failure. Simply do
-	 * hard iteration simplify the code a lot.
-	 */
-	if (scan_wd(oid, 0, ec_index) == SD_RES_SUCCESS)
-		return true;
-
-	return false;
-}
-
-int md_get_stale_path(uint64_t oid, uint32_t epoch, uint8_t ec_index,
-		      char *path)
-{
-	if (unlikely(!epoch))
-		panic("invalid 0 epoch");
-
-	if (is_erasure_oid(oid)) {
-		if (unlikely(ec_index >= SD_MAX_COPIES))
-			panic("invalid ec index %d", ec_index);
-
-		snprintf(path, PATH_MAX, "%s/.stale/%016"PRIx64"_%d.%"PRIu32,
-			 md_get_object_dir(oid), oid, ec_index, epoch);
-	} else
-		snprintf(path, PATH_MAX, "%s/.stale/%016"PRIx64".%"PRIu32,
-			 md_get_object_dir(oid), oid, epoch);
-
-	if (md_access(path))
-		return SD_RES_SUCCESS;
-
-	if (scan_wd(oid, epoch, ec_index) == SD_RES_SUCCESS)
-		return SD_RES_SUCCESS;
-
-	return SD_RES_NO_OBJ;
-}
-
-uint32_t md_get_info(struct sd_md_info *info)
-{
-	uint32_t ret = sizeof(*info);
-	const struct disk *disk;
-	int i = 0;
-
-	memset(info, 0, ret);
-	sd_read_lock(&md.lock);
-	rb_for_each_entry(disk, &md.root, rb) {
-		info->disk[i].idx = i;
-		pstrcpy(info->disk[i].path, PATH_MAX, disk->path);
-		/* FIXME: better handling failure case. */
-		info->disk[i].free = get_path_free_size(info->disk[i].path,
-							&info->disk[i].used);
-		i++;
-	}
-	info->nr = md.nr_disks;
-	sd_rw_unlock(&md.lock);
-	return ret;
-}
-
-static inline void md_del_disk(const char *path)
-{
-	struct disk *disk = path_to_disk(path);
-
-	if (!disk) {
-		sd_err("invalid path %s", path);
-		return;
-	}
-	md_remove_disk(disk);
-}
-
-#ifdef HAVE_DISKVNODES
-void update_node_disks(void)
-{
-	const struct disk *disk;
-	int i = 0;
-	bool rb_empty = false;
-
-	if (!sys)
-		return;
-
-	memset(sys->this_node.disks, 0, sizeof(struct disk_info) * DISK_MAX);
-	sd_read_lock(&md.lock);
-	rb_for_each_entry(disk, &md.root, rb) {
-		sys->this_node.disks[i].disk_id =
-			sd_hash(disk->path, strlen(disk->path));
-		sys->this_node.disks[i].disk_space = disk->space;
-		i++;
-	}
-	sd_rw_unlock(&md.lock);
-
-	if (RB_EMPTY_ROOT(&md.vroot))
-		rb_empty = true;
-	sd_write_lock(&md.lock);
-	rb_for_each_entry(disk, &md.root, rb) {
-		if (!rb_empty)
-			remove_vdisks(disk);
-		create_vdisks(disk);
-	}
-	sd_rw_unlock(&md.lock);
-}
-#else
-void update_node_disks(void)
-{
-}
-#endif
-
-static int do_plug_unplug(char *disks, bool plug)
-{
-	const char *path;
-	int old_nr, ret = SD_RES_UNKNOWN;
-
-	sd_write_lock(&md.lock);
-	old_nr = md.nr_disks;
-	path = strtok(disks, ",");
-	do {
-		if (plug) {
-			if (!md_add_disk(path, true))
-				sd_err("failed to add %s", path);
-		} else {
-			md_del_disk(path);
-		}
-	} while ((path = strtok(NULL, ",")));
-
-	/* If no disks change, bail out */
-	if (old_nr == md.nr_disks)
-		goto out;
-
-	ret = SD_RES_SUCCESS;
-out:
-	sd_rw_unlock(&md.lock);
-
-	if (ret == SD_RES_SUCCESS) {
-		update_node_disks();
-		kick_recover();
-	}
-
-	return ret;
-}
-
-int md_plug_disks(char *disks)
-{
-	return do_plug_unplug(disks, true);
-}
-
-int md_unplug_disks(char *disks)
-{
-	return do_plug_unplug(disks, false);
-}
-
-uint64_t md_get_size(uint64_t *used)
-{
-	uint64_t fsize = 0;
-	const struct disk *disk;
-
-	*used = 0;
-	sd_read_lock(&md.lock);
-	rb_for_each_entry(disk, &md.root, rb) {
-		fsize += get_path_free_size(disk->path, used);
-	}
-	sd_rw_unlock(&md.lock);
-
-	return fsize + *used;
-}
-
-uint32_t md_nr_disks(void)
-{
-	return nr_online_disks();
-}
diff --git a/sheep/plain_store.c b/sheep/plain_store.c
deleted file mode 100644
index 4c19832..0000000
--- a/sheep/plain_store.c
+++ /dev/null
@@ -1,759 +0,0 @@
-/*
- * Copyright (C) 2012 Nippon Telegraph and Telephone Corporation.
- *
- * This program is free software; you can redistribute it and/or
- * modify it under the terms of the GNU General Public License version
- * 2 as published by the Free Software Foundation.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#include <libgen.h>
-#include <linux/falloc.h>
-
-#include "sheep_priv.h"
-
-#ifndef FALLOC_FL_PUNCH_HOLE
-#define FALLOC_FL_PUNCH_HOLE 0x02
-#endif
-
-#define sector_algined(x) ({ ((x) & (SECTOR_SIZE - 1)) == 0; })
-
-static inline bool iocb_is_aligned(const struct siocb *iocb)
-{
-	return  sector_algined(iocb->offset) && sector_algined(iocb->length);
-}
-
-static int prepare_iocb(uint64_t oid, const struct siocb *iocb, bool create)
-{
-	int syncflag = create ? O_SYNC : O_DSYNC;
-	int flags = syncflag | O_RDWR;
-
-	if (uatomic_is_true(&sys->use_journal) || sys->nosync == true)
-		flags &= ~syncflag;
-
-	if (sys->backend_dio && iocb_is_aligned(iocb)) {
-		if (!is_aligned_to_pagesize(iocb->buf))
-			panic("Memory isn't aligned to pagesize %p", iocb->buf);
-		flags |= O_DIRECT;
-	}
-
-	if (create)
-		flags |= O_CREAT | O_EXCL;
-
-	return flags;
-}
-
-static int get_store_path(uint64_t oid, uint8_t ec_index, char *path)
-{
-	if (is_erasure_oid(oid)) {
-		if (unlikely(ec_index >= SD_MAX_COPIES))
-			panic("invalid ec_index %d", ec_index);
-		return snprintf(path, PATH_MAX, "%s/%016"PRIx64"_%d",
-				md_get_object_dir(oid), oid, ec_index);
-	}
-
-	return snprintf(path, PATH_MAX, "%s/%016" PRIx64,
-			md_get_object_dir(oid), oid);
-}
-
-static int get_store_tmp_path(uint64_t oid, uint8_t ec_index, char *path)
-{
-	if (is_erasure_oid(oid)) {
-		if (unlikely(ec_index >= SD_MAX_COPIES))
-			panic("invalid ec_index %d", ec_index);
-		return snprintf(path, PATH_MAX, "%s/%016"PRIx64"_%d.tmp",
-				md_get_object_dir(oid), oid, ec_index);
-	}
-
-	return snprintf(path, PATH_MAX, "%s/%016" PRIx64".tmp",
-			md_get_object_dir(oid), oid);
-}
-
-static int get_store_stale_path(uint64_t oid, uint32_t epoch, uint8_t ec_index,
-				char *path)
-{
-	return md_get_stale_path(oid, epoch, ec_index, path);
-}
-
-/*
- * Check if oid is in this nodes (if oid is in the wrong place, it will be moved
- * to the correct one after this call in a MD setup.
- */
-bool default_exist(uint64_t oid, uint8_t ec_index)
-{
-	char path[PATH_MAX];
-
-	get_store_path(oid, ec_index, path);
-
-	return md_exist(oid, ec_index, path);
-}
-
-static int err_to_sderr(const char *path, uint64_t oid, int err)
-{
-	struct stat s;
-	char p[PATH_MAX], *dir;
-
-	/* Use a temporary buffer since dirname() may modify its argument. */
-	pstrcpy(p, sizeof(p), path);
-	dir = dirname(p);
-
-	sd_debug("%s", path);
-	switch (err) {
-	case ENOENT:
-		if (stat(dir, &s) < 0) {
-			sd_err("%s corrupted", dir);
-			return md_handle_eio(dir);
-		}
-		sd_debug("object %016" PRIx64 " not found locally", oid);
-		return SD_RES_NO_OBJ;
-	case ENOSPC:
-		/* TODO: stop automatic recovery */
-		sd_err("diskfull, oid=%"PRIx64, oid);
-		return SD_RES_NO_SPACE;
-	case EMFILE:
-	case ENFILE:
-	case EINTR:
-	case EAGAIN:
-	case EEXIST:
-		sd_err("%m, oid=%"PRIx64, oid);
-		/* make gateway try again */
-		return SD_RES_NETWORK_ERROR;
-	default:
-		sd_err("oid=%"PRIx64", %m", oid);
-		return md_handle_eio(dir);
-	}
-}
-
-static int discard(int fd, uint64_t start, uint32_t end)
-{
-	int ret = xfallocate(fd, FALLOC_FL_KEEP_SIZE | FALLOC_FL_PUNCH_HOLE,
-			     start, end - start);
-	if (ret < 0) {
-		if (errno == ENOSYS || errno == EOPNOTSUPP)
-			sd_info("FALLOC_FL_PUNCH_HOLE is not supported "
-				"on this filesystem");
-		else
-			sd_err("failed to discard object, %m");
-	}
-
-	return ret;
-}
-
-/* Trim zero blocks of the beginning and end of the object. */
-static int default_trim(int fd, uint64_t oid, const struct siocb *iocb,
-			uint64_t *poffset, uint32_t *plen)
-{
-	trim_zero_blocks(iocb->buf, poffset, plen);
-
-	if (iocb->offset < *poffset) {
-		sd_debug("discard between %d, %ld, %" PRIx64, iocb->offset,
-			 *poffset, oid);
-
-		if (discard(fd, iocb->offset, *poffset) < 0)
-			return -1;
-	}
-
-	if (*poffset + *plen < iocb->offset + iocb->length) {
-		uint64_t end = iocb->offset + iocb->length;
-		uint32_t object_size = get_vdi_object_size(oid_to_vid(oid));
-		if (end == get_objsize(oid, object_size))
-			/* This is necessary to punch the last block */
-			end = round_up(end, BLOCK_SIZE);
-		sd_debug("discard between %ld, %ld, %" PRIx64, *poffset + *plen,
-			 end, oid);
-
-		if (discard(fd, *poffset + *plen, end) < 0)
-			return -1;
-	}
-
-	return 0;
-}
-
-int default_write(uint64_t oid, const struct siocb *iocb)
-{
-	int flags = prepare_iocb(oid, iocb, false), fd,
-	    ret = SD_RES_SUCCESS;
-	char path[PATH_MAX];
-	ssize_t size;
-	uint32_t len = iocb->length;
-	uint64_t offset = iocb->offset;
-	static bool trim_is_supported = true;
-
-	if (iocb->epoch < sys_epoch()) {
-		sd_debug("%"PRIu32" sys %"PRIu32, iocb->epoch, sys_epoch());
-		return SD_RES_OLD_NODE_VER;
-	}
-
-	if (uatomic_is_true(&sys->use_journal) &&
-	    unlikely(journal_write_store(oid, iocb->buf, iocb->length,
-					 iocb->offset, false))
-	    != SD_RES_SUCCESS) {
-		sd_err("turn off journaling");
-		uatomic_set_false(&sys->use_journal);
-		flags |= O_DSYNC;
-		sync();
-	}
-
-	get_store_path(oid, iocb->ec_index, path);
-
-	/*
-	 * Make sure oid is in the right place because oid might be misplaced
-	 * in a wrong place, due to 'shutdown/restart with less/more disks' or
-	 * any bugs. We need call err_to_sderr() to return EIO if disk is broken
-	 */
-	if (!default_exist(oid, iocb->ec_index))
-		return err_to_sderr(path, oid, ENOENT);
-
-	fd = open(path, flags, sd_def_fmode);
-	if (unlikely(fd < 0))
-		return err_to_sderr(path, oid, errno);
-
-	if (trim_is_supported && is_sparse_object(oid)) {
-		if (default_trim(fd, oid, iocb, &offset, &len) < 0) {
-			trim_is_supported = false;
-			offset = iocb->offset;
-			len = iocb->length;
-		}
-	}
-
-	size = xpwrite(fd, iocb->buf, len, offset);
-	if (unlikely(size != len)) {
-		sd_err("failed to write object %"PRIx64", path=%s, offset=%"
-		       PRId32", size=%"PRId32", result=%zd, %m", oid, path,
-		       iocb->offset, iocb->length, size);
-		ret = err_to_sderr(path, oid, errno);
-		goto out;
-	}
-out:
-	close(fd);
-	return ret;
-}
-
-static int make_stale_dir(const char *path)
-{
-	char p[PATH_MAX];
-
-	snprintf(p, PATH_MAX, "%s/.stale", path);
-	if (xmkdir(p, sd_def_dmode) < 0) {
-		sd_err("%s failed, %m", p);
-		return SD_RES_EIO;
-	}
-	return SD_RES_SUCCESS;
-}
-
-static int purge_dir(const char *path)
-{
-	if (purge_directory(path) < 0)
-		return SD_RES_EIO;
-
-	return SD_RES_SUCCESS;
-}
-
-static int purge_stale_dir(const char *path)
-{
-	char p[PATH_MAX];
-
-	snprintf(p, PATH_MAX, "%s/.stale", path);
-
-	if (purge_directory_async(p) < 0)
-		return SD_RES_EIO;
-
-	return SD_RES_SUCCESS;
-}
-
-int default_cleanup(void)
-{
-	int ret;
-
-	ret = for_each_obj_path(purge_stale_dir);
-	if (ret != SD_RES_SUCCESS)
-		return ret;
-
-	return SD_RES_SUCCESS;
-}
-
-static int init_vdi_state(uint64_t oid, const char *wd, uint32_t epoch)
-{
-	int ret;
-	struct sd_inode *inode = xzalloc(SD_INODE_HEADER_SIZE);
-	struct siocb iocb = {
-		.epoch = epoch,
-		.buf = inode,
-		.length = SD_INODE_HEADER_SIZE,
-	};
-
-	ret = default_read(oid, &iocb);
-	if (ret != SD_RES_SUCCESS) {
-		sd_err("failed to read inode header %" PRIx64 " %" PRId32
-		       "wat %s", oid, epoch, wd);
-		goto out;
-	}
-	add_vdi_state_unordered(oid_to_vid(oid), inode->nr_copies,
-		      vdi_is_snapshot(inode), inode->copy_policy,
-		      inode->block_size_shift, inode->parent_vdi_id);
-
-	if (inode->name[0] == '\0')
-		atomic_set_bit(oid_to_vid(oid), sys->vdi_deleted);
-
-	atomic_set_bit(oid_to_vid(oid), sys->vdi_inuse);
-
-	ret = SD_RES_SUCCESS;
-out:
-	free(inode);
-	return ret;
-}
-
-static int init_objlist_and_vdi_bitmap(uint64_t oid, const char *wd,
-				       uint32_t epoch, uint8_t ec_index,
-				       struct vnode_info *vinfo,
-				       void *arg)
-{
-	int ret;
-	objlist_cache_insert(oid);
-
-	if (is_vdi_obj(oid)) {
-		sd_debug("found the VDI object %" PRIx64" epoch %"PRIu32
-			 " at %s", oid, epoch, wd);
-		ret = init_vdi_state(oid, wd, epoch);
-		if (ret != SD_RES_SUCCESS)
-			return ret;
-	}
-	return SD_RES_SUCCESS;
-}
-
-int default_init(void)
-{
-	int ret;
-
-	sd_debug("use plain store driver");
-	ret = for_each_obj_path(make_stale_dir);
-	if (ret != SD_RES_SUCCESS)
-		return ret;
-
-	for_each_object_in_stale(init_objlist_and_vdi_bitmap, NULL);
-
-	return for_each_object_in_wd(init_objlist_and_vdi_bitmap, true, NULL);
-}
-
-static int default_read_from_path(uint64_t oid, const char *path,
-				  const struct siocb *iocb)
-{
-	int flags = prepare_iocb(oid, iocb, false), fd,
-	    ret = SD_RES_SUCCESS;
-	ssize_t size;
-
-	/*
-	 * Make sure oid is in the right place because oid might be misplaced
-	 * in a wrong place, due to 'shutdown/restart with less disks' or any
-	 * bugs. We need call err_to_sderr() to return EIO if disk is broken.
-	 *
-	 * For stale path, get_store_stale_path already does default_exist job.
-	 */
-	if (!is_stale_path(path) && !default_exist(oid, iocb->ec_index))
-		return err_to_sderr(path, oid, ENOENT);
-
-	fd = open(path, flags);
-	if (fd < 0)
-		return err_to_sderr(path, oid, errno);
-
-	size = xpread(fd, iocb->buf, iocb->length, iocb->offset);
-	if (size < 0) {
-		sd_err("failed to read object %"PRIx64", path=%s, offset=%"
-		       PRId32", size=%"PRId32", result=%zd, %m", oid, path,
-		       iocb->offset, iocb->length, size);
-		ret = err_to_sderr(path, oid, errno);
-	}
-	close(fd);
-	return ret;
-}
-
-int default_read(uint64_t oid, const struct siocb *iocb)
-{
-	int ret;
-	char path[PATH_MAX];
-
-	get_store_path(oid, iocb->ec_index, path);
-	ret = default_read_from_path(oid, path, iocb);
-
-	/*
-	 * If the request is against the older epoch, try to read from
-	 * the stale directory
-	 */
-	if (ret == SD_RES_NO_OBJ && iocb->epoch > 0 &&
-	    iocb->epoch < sys_epoch()) {
-		get_store_stale_path(oid, iocb->epoch, iocb->ec_index, path);
-		ret = default_read_from_path(oid, path, iocb);
-	}
-
-	return ret;
-}
-
-int default_create_and_write(uint64_t oid, const struct siocb *iocb)
-{
-	char path[PATH_MAX], tmp_path[PATH_MAX], *dir;
-	int flags = prepare_iocb(oid, iocb, true);
-	int ret, fd;
-	uint32_t len = iocb->length;
-	uint32_t object_size = 0;
-	size_t obj_size;
-	uint64_t offset = iocb->offset;
-
-	sd_debug("%"PRIx64, oid);
-	get_store_path(oid, iocb->ec_index, path);
-	get_store_tmp_path(oid, iocb->ec_index, tmp_path);
-
-	if (uatomic_is_true(&sys->use_journal) &&
-	    journal_write_store(oid, iocb->buf, iocb->length,
-				iocb->offset, true)
-	    != SD_RES_SUCCESS) {
-		sd_err("turn off journaling");
-		uatomic_set_false(&sys->use_journal);
-		flags |= O_SYNC;
-		sync();
-	}
-
-	fd = open(tmp_path, flags, sd_def_fmode);
-	if (fd < 0) {
-		if (errno == EEXIST) {
-			/*
-			 * This happens if node membership changes during object
-			 * creation; while gateway retries a CREATE request,
-			 * recovery process could also recover the object at the
-			 * same time.  They should try to write the same date,
-			 * so it is okay to simply return success here.
-			 */
-			sd_debug("%s exists", tmp_path);
-			return SD_RES_SUCCESS;
-		}
-
-		sd_err("failed to open %s: %m", tmp_path);
-		return err_to_sderr(path, oid, errno);
-	}
-
-	obj_size = get_store_objsize(oid);
-
-	trim_zero_blocks(iocb->buf, &offset, &len);
-
-	object_size = get_vdi_object_size(oid_to_vid(oid));
-
-	if (offset != 0 || len != get_objsize(oid, object_size)) {
-		if (is_sparse_object(oid))
-			ret = xftruncate(fd, obj_size);
-		else
-			ret = prealloc(fd, obj_size);
-		if (ret < 0) {
-			ret = err_to_sderr(path, oid, errno);
-			goto out;
-		}
-	}
-
-	ret = xpwrite(fd, iocb->buf, len, offset);
-	if (ret != len) {
-		sd_err("failed to write object. %m");
-		ret = err_to_sderr(path, oid, errno);
-		goto out;
-	}
-
-	ret = rename(tmp_path, path);
-	if (ret < 0) {
-		sd_err("failed to rename %s to %s: %m", tmp_path, path);
-		ret = err_to_sderr(path, oid, errno);
-		goto out;
-	}
-
-	close(fd);
-
-	if (uatomic_is_true(&sys->use_journal) || sys->nosync == true) {
-		objlist_cache_insert(oid);
-		return SD_RES_SUCCESS;
-	}
-
-	pstrcpy(tmp_path, sizeof(tmp_path), path);
-	dir = dirname(tmp_path);
-	fd = open(dir, O_DIRECTORY | O_RDONLY);
-	if (fd < 0) {
-		sd_err("failed to open directory %s: %m", dir);
-		return err_to_sderr(path, oid, errno);
-	}
-
-	if (fsync(fd) != 0) {
-		sd_err("failed to write directory %s: %m", dir);
-		ret = err_to_sderr(path, oid, errno);
-		close(fd);
-		if (unlink(path) != 0)
-			sd_err("failed to unlink %s: %m", path);
-		return ret;
-	}
-	close(fd);
-	objlist_cache_insert(oid);
-	return SD_RES_SUCCESS;
-
-out:
-	if (unlink(tmp_path) != 0)
-		sd_err("failed to unlink %s: %m", tmp_path);
-	close(fd);
-	return ret;
-}
-
-int default_link(uint64_t oid, uint32_t tgt_epoch)
-{
-	char path[PATH_MAX], stale_path[PATH_MAX];
-
-	sd_debug("try link %"PRIx64" from snapshot with epoch %d", oid,
-		 tgt_epoch);
-
-	snprintf(path, PATH_MAX, "%s/%016"PRIx64, md_get_object_dir(oid), oid);
-	get_store_stale_path(oid, tgt_epoch, 0, stale_path);
-
-	if (link(stale_path, path) < 0) {
-		/*
-		 * Recovery thread and main thread might try to recover the
-		 * same object and we might get EEXIST in such case.
-		 */
-		if (errno == EEXIST)
-			goto out;
-
-		sd_debug("failed to link from %s to %s, %m", stale_path, path);
-		return err_to_sderr(path, oid, errno);
-	}
-out:
-	return SD_RES_SUCCESS;
-}
-
-/*
- * For replicated object, if any of the replica belongs to this node, we
- * consider it not stale.
- *
- * For erasure coded object, since every copy is unique and if it migrates to
- * other node(index gets changed even it has some other copy belongs to it)
- * because of hash ring changes, we consider it stale.
- */
-static bool oid_stale(uint64_t oid, int ec_index, struct vnode_info *vinfo)
-{
-	uint32_t i, nr_copies;
-	const struct sd_vnode *v;
-	bool ret = true;
-	const struct sd_vnode *obj_vnodes[SD_MAX_COPIES];
-
-	nr_copies = get_obj_copy_number(oid, vinfo->nr_zones);
-	oid_to_vnodes(oid, &vinfo->vroot, nr_copies, obj_vnodes);
-	for (i = 0; i < nr_copies; i++) {
-		v = obj_vnodes[i];
-		if (vnode_is_local(v)) {
-			if (ec_index < SD_MAX_COPIES) {
-				if (i == ec_index)
-					ret = false;
-			} else {
-				ret = false;
-			}
-			break;
-		}
-	}
-
-	return ret;
-}
-
-static int move_object_to_stale_dir(uint64_t oid, const char *wd,
-				    uint32_t epoch, uint8_t ec_index,
-				    struct vnode_info *vinfo, void *arg)
-{
-	char path[PATH_MAX], stale_path[PATH_MAX];
-	uint32_t tgt_epoch = *(uint32_t *)arg;
-
-	/* ec_index from md.c is reliable so we can directly use it */
-	if (ec_index < SD_MAX_COPIES) {
-		snprintf(path, PATH_MAX, "%s/%016"PRIx64"_%d",
-			 md_get_object_dir(oid), oid, ec_index);
-		snprintf(stale_path, PATH_MAX,
-			 "%s/.stale/%016"PRIx64"_%d.%"PRIu32,
-			 md_get_object_dir(oid), oid, ec_index, tgt_epoch);
-	} else {
-		snprintf(path, PATH_MAX, "%s/%016" PRIx64,
-			 md_get_object_dir(oid), oid);
-		snprintf(stale_path, PATH_MAX, "%s/.stale/%016"PRIx64".%"PRIu32,
-			 md_get_object_dir(oid), oid, tgt_epoch);
-	}
-
-	if (unlikely(rename(path, stale_path)) < 0) {
-		sd_err("failed to move stale object %" PRIX64 " to %s, %m", oid,
-		       path);
-		return SD_RES_EIO;
-	}
-
-	sd_debug("moved object %"PRIx64, oid);
-	return SD_RES_SUCCESS;
-}
-
-static int check_stale_objects(uint64_t oid, const char *wd, uint32_t epoch,
-			       uint8_t ec_index, struct vnode_info *vinfo,
-			       void *arg)
-{
-	if (oid_stale(oid, ec_index, vinfo))
-		return move_object_to_stale_dir(oid, wd, 0, ec_index,
-						NULL, arg);
-
-	return SD_RES_SUCCESS;
-}
-
-int default_update_epoch(uint32_t epoch)
-{
-	assert(epoch);
-	return for_each_object_in_wd(check_stale_objects, false, &epoch);
-}
-
-int default_format(void)
-{
-	unsigned ret;
-
-	sd_debug("try get a clean store");
-	ret = for_each_obj_path(purge_dir);
-	if (ret != SD_RES_SUCCESS)
-		return ret;
-
-	if (sys->enable_object_cache)
-		object_cache_format();
-
-	return SD_RES_SUCCESS;
-}
-
-int default_remove_object(uint64_t oid, uint8_t ec_index)
-{
-	char path[PATH_MAX];
-
-	if (uatomic_is_true(&sys->use_journal))
-		journal_remove_object(oid);
-
-	get_store_path(oid, ec_index, path);
-
-	if (unlink(path) < 0) {
-		if (errno == ENOENT)
-			return SD_RES_NO_OBJ;
-
-		sd_err("failed, %s, %m", path);
-		return SD_RES_EIO;
-	}
-
-	return SD_RES_SUCCESS;
-}
-
-#define SHA1NAME "user.obj.sha1"
-
-static int get_object_sha1(const char *path, uint8_t *sha1)
-{
-	if (getxattr(path, SHA1NAME, sha1, SHA1_DIGEST_SIZE)
-	    != SHA1_DIGEST_SIZE) {
-		if (errno == ENODATA)
-			sd_debug("sha1 is not cached yet, %s", path);
-		else
-			sd_err("fail to get xattr, %s", path);
-		return -1;
-	}
-
-	return 0;
-}
-
-static int set_object_sha1(const char *path, const uint8_t *sha1)
-{
-	int ret;
-
-	ret = setxattr(path, SHA1NAME, sha1, SHA1_DIGEST_SIZE, 0);
-	if (ret < 0)
-		sd_err("fail to set sha1, %s", path);
-
-	return ret;
-}
-
-static int get_object_path(uint64_t oid, uint32_t epoch, char *path,
-			   size_t size)
-{
-	if (default_exist(oid, 0)) {
-		snprintf(path, PATH_MAX, "%s/%016"PRIx64,
-			 md_get_object_dir(oid), oid);
-	} else {
-		get_store_stale_path(oid, epoch, 0, path);
-		if (access(path, F_OK) < 0) {
-			if (errno == ENOENT)
-				return SD_RES_NO_OBJ;
-			return SD_RES_EIO;
-		}
-
-	}
-
-	return SD_RES_SUCCESS;
-}
-
-int default_get_hash(uint64_t oid, uint32_t epoch, uint8_t *sha1)
-{
-	int ret;
-	void *buf;
-	struct siocb iocb = {};
-	uint32_t length;
-	bool is_readonly_obj = oid_is_readonly(oid);
-	char path[PATH_MAX];
-
-	ret = get_object_path(oid, epoch, path, sizeof(path));
-	if (ret != SD_RES_SUCCESS)
-		return ret;
-
-	if (is_readonly_obj) {
-		if (get_object_sha1(path, sha1) == 0) {
-			sd_debug("use cached sha1 digest %s",
-				 sha1_to_hex(sha1));
-			return SD_RES_SUCCESS;
-		}
-	}
-
-	length = get_store_objsize(oid);
-	buf = valloc(length);
-	if (buf == NULL)
-		return SD_RES_NO_MEM;
-
-	iocb.epoch = epoch;
-	iocb.buf = buf;
-	iocb.length = length;
-
-	ret = default_read_from_path(oid, path, &iocb);
-	if (ret != SD_RES_SUCCESS) {
-		free(buf);
-		return ret;
-	}
-
-	get_buffer_sha1(buf, length, sha1);
-	free(buf);
-
-	sd_debug("the message digest of %"PRIx64" at epoch %d is %s", oid,
-		 epoch, sha1_to_hex(sha1));
-
-	if (is_readonly_obj)
-		set_object_sha1(path, sha1);
-
-	return ret;
-}
-
-int default_purge_obj(void)
-{
-	uint32_t tgt_epoch = get_latest_epoch();
-
-	return for_each_object_in_wd(move_object_to_stale_dir, true,
-				     &tgt_epoch);
-}
-
-static struct store_driver plain_store = {
-	.name = "plain",
-	.init = default_init,
-	.exist = default_exist,
-	.create_and_write = default_create_and_write,
-	.write = default_write,
-	.read = default_read,
-	.link = default_link,
-	.update_epoch = default_update_epoch,
-	.cleanup = default_cleanup,
-	.format = default_format,
-	.remove_object = default_remove_object,
-	.get_hash = default_get_hash,
-	.purge_obj = default_purge_obj,
-};
-
-add_store_driver(plain_store);
diff --git a/sheep/store.c b/sheep/store.c
deleted file mode 100644
index 8843fb8..0000000
--- a/sheep/store.c
+++ /dev/null
@@ -1,511 +0,0 @@
-/*
- * Copyright (C) 2009-2011 Nippon Telegraph and Telephone Corporation.
- *
- * This program is free software; you can redistribute it and/or
- * modify it under the terms of the GNU General Public License version
- * 2 as published by the Free Software Foundation.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#include "sheep_priv.h"
-
-char *obj_path;
-char *epoch_path;
-
-struct store_driver *sd_store;
-LIST_HEAD(store_drivers);
-
-int update_epoch_log(uint32_t epoch, struct sd_node *nodes, size_t nr_nodes)
-{
-	int ret, len, nodes_len;
-	time_t t;
-	char path[PATH_MAX], *buf;
-
-	sd_debug("update epoch: %d, %zu", epoch, nr_nodes);
-
-	/* Piggyback the epoch creation time for 'dog cluster info' */
-	time(&t);
-	nodes_len = nr_nodes * sizeof(struct sd_node);
-	len = nodes_len + sizeof(time_t);
-	buf = xmalloc(len);
-	memcpy(buf, nodes, nodes_len);
-	memcpy(buf + nodes_len, &t, sizeof(time_t));
-
-	/*
-	 * rb field is unused in epoch file, zero-filling it
-	 * is good for epoch file recovery because it is unified
-	 */
-	for (int i = 0; i < nr_nodes; i++)
-		memset(buf + i * sizeof(struct sd_node)
-				+ offsetof(struct sd_node, rb),
-				0, sizeof(struct rb_node));
-
-	snprintf(path, sizeof(path), "%s%08u", epoch_path, epoch);
-
-	ret = atomic_create_and_write(path, buf, len, true);
-
-	free(buf);
-	return ret;
-}
-
-static int do_epoch_log_read(uint32_t epoch, struct sd_node *nodes, int len,
-			     int *nr_nodes, time_t *timestamp)
-{
-	int fd, ret, buf_len;
-	char path[PATH_MAX];
-	struct stat epoch_stat;
-
-	snprintf(path, sizeof(path), "%s%08u", epoch_path, epoch);
-	fd = open(path, O_RDONLY);
-	if (fd < 0) {
-		sd_debug("failed to open epoch %"PRIu32" log, %m", epoch);
-		goto err;
-	}
-
-	memset(&epoch_stat, 0, sizeof(epoch_stat));
-	ret = fstat(fd, &epoch_stat);
-	if (ret < 0) {
-		sd_err("failed to stat epoch %"PRIu32" log, %m", epoch);
-		goto err;
-	}
-
-	buf_len = epoch_stat.st_size - sizeof(*timestamp);
-	if (buf_len < 0) {
-		sd_err("invalid epoch %"PRIu32" log", epoch);
-		goto err;
-	}
-	if (len < buf_len) {
-		close(fd);
-		return SD_RES_BUFFER_SMALL;
-	}
-
-	ret = xread(fd, nodes, buf_len);
-	if (ret < 0) {
-		sd_err("failed to read epoch %"PRIu32" log, %m", epoch);
-		goto err;
-	}
-
-	/* Broken epoch, just ignore */
-	if (ret % sizeof(struct sd_node) != 0) {
-		sd_err("invalid epoch %"PRIu32" log", epoch);
-		goto err;
-	}
-
-	*nr_nodes = ret / sizeof(struct sd_node);
-
-	if (timestamp) {
-		ret = xread(fd, timestamp, sizeof(*timestamp));
-		if (ret != sizeof(*timestamp)) {
-			sd_err("invalid epoch %"PRIu32" log", epoch);
-			goto err;
-		}
-	}
-
-	close(fd);
-	return SD_RES_SUCCESS;
-err:
-	if (fd >= 0)
-		close(fd);
-	return SD_RES_NO_TAG;
-}
-
-int epoch_log_read(uint32_t epoch, struct sd_node *nodes,
-				int len, int *nr_nodes)
-{
-	return do_epoch_log_read(epoch, nodes, len, nr_nodes, NULL);
-}
-
-int epoch_log_read_with_timestamp(uint32_t epoch, struct sd_node *nodes,
-				int len, int *nr_nodes, time_t *timestamp)
-{
-	return do_epoch_log_read(epoch, nodes, len, nr_nodes, timestamp);
-}
-
-uint32_t get_latest_epoch(void)
-{
-	DIR *dir;
-	struct dirent *d;
-	uint32_t e, epoch = 0;
-	char *p;
-
-	dir = opendir(epoch_path);
-	if (!dir)
-		panic("failed to get the latest epoch: %m");
-
-	while ((d = readdir(dir))) {
-		e = strtol(d->d_name, &p, 10);
-		if (d->d_name == p)
-			continue;
-
-		if (strlen(d->d_name) != 8)
-			continue;
-
-		if (e > epoch)
-			epoch = e;
-	}
-	closedir(dir);
-
-	return epoch;
-}
-
-int lock_base_dir(const char *d)
-{
-#define LOCK_PATH "/lock"
-	char *lock_path;
-	int ret = 0;
-	int fd, len = strlen(d) + strlen(LOCK_PATH) + 1;
-
-	lock_path = xzalloc(len);
-	snprintf(lock_path, len, "%s" LOCK_PATH, d);
-
-	fd = open(lock_path, O_WRONLY|O_CREAT, sd_def_fmode);
-	if (fd < 0) {
-		sd_err("failed to open lock file %s (%m)", lock_path);
-		ret = -1;
-		goto out;
-	}
-
-	if (lockf(fd, F_TLOCK, 1) < 0) {
-		if (errno == EACCES || errno == EAGAIN)
-			sd_err("another sheep daemon is using %s", d);
-		else
-			sd_err("unable to get base dir lock (%m)");
-		ret = -1;
-		goto out;
-	}
-
-out:
-	free(lock_path);
-	return ret;
-}
-
-int init_base_path(const char *d)
-{
-	if (xmkdir(d, sd_def_dmode) < 0) {
-		sd_err("cannot create the directory %s (%m)", d);
-		return -1;
-	}
-
-	return 0;
-}
-
-static inline int check_path_len(const char *path)
-{
-	int len = strlen(path);
-	if (len > PATH_MAX) {
-		sd_err("insanely long object directory %s", path);
-		return -1;
-	}
-
-	return 0;
-}
-
-static int is_meta_store(const char *path)
-{
-	char conf[PATH_MAX];
-	char epoch[PATH_MAX];
-
-	snprintf(conf, PATH_MAX, "%s/config", path);
-	snprintf(epoch, PATH_MAX, "%s/epoch", path);
-	if (!access(conf, R_OK) && !access(epoch, R_OK))
-		return true;
-
-	return false;
-}
-
-static int init_obj_path(const char *base_path, char *argp)
-{
-	char *p;
-	int len;
-
-	if (check_path_len(base_path) < 0)
-		return -1;
-
-#define OBJ_PATH "/obj"
-	len = strlen(base_path) + strlen(OBJ_PATH) + 1;
-	obj_path = xzalloc(len);
-	snprintf(obj_path, len, "%s" OBJ_PATH, base_path);
-
-	/* Eat up the first component */
-	strtok(argp, ",");
-	p = strtok(NULL, ",");
-	if (!p) {
-		/*
-		 * If We have only one path, meta-store and object-store share
-		 * it. This is helpful to upgrade old sheep cluster to
-		 * the MD-enabled.
-		 */
-		md_add_disk(obj_path, false);
-	} else {
-		do {
-			if (is_meta_store(p)) {
-				sd_err("%s is meta-store, abort", p);
-				return -1;
-			}
-			md_add_disk(p, false);
-		} while ((p = strtok(NULL, ",")));
-	}
-
-	if (md_nr_disks() <= 0) {
-		sd_err("There isn't any available disk!");
-		return -1;
-	}
-
-	return xmkdir(obj_path, sd_def_dmode);
-}
-
-static int init_epoch_path(const char *base_path)
-{
-#define EPOCH_PATH "/epoch/"
-	int len = strlen(base_path) + strlen(EPOCH_PATH) + 1;
-	epoch_path = xzalloc(len);
-	snprintf(epoch_path, len, "%s" EPOCH_PATH, base_path);
-
-	return xmkdir(epoch_path, sd_def_dmode);
-}
-
-/*
- * If the node is gateway, this function only finds the store driver.
- * Otherwise, this function initializes the backend store
- */
-int init_store_driver(bool is_gateway)
-{
-	char driver_name[STORE_LEN], *p;
-
-	pstrcpy(driver_name, sizeof(driver_name), (char *)sys->cinfo.store);
-
-	p = memchr(driver_name, '\0', STORE_LEN);
-	if (!p) {
-		/*
-		 * If the driver name is not NUL terminated we are in deep
-		 * trouble, let's get out here.
-		 */
-		sd_debug("store name not NUL terminated");
-		return SD_RES_NO_STORE;
-	}
-
-	/*
-	 * The store file might not exist in case this is a new sheep that
-	 * never joined a cluster before.
-	 */
-	if (p == driver_name)
-		return 0;
-
-	sd_store = find_store_driver(driver_name);
-	if (!sd_store) {
-		sd_debug("store %s not found", driver_name);
-		return SD_RES_NO_STORE;
-	}
-
-	if (is_gateway)
-		return SD_RES_SUCCESS;
-
-	return sd_store->init();
-}
-
-int init_disk_space(const char *base_path)
-{
-	int ret = SD_RES_SUCCESS;
-	uint64_t space_size = 0, mds;
-	struct statvfs fs;
-
-	if (sys->gateway_only)
-		goto out;
-
-	/* We need to init md even we don't need to update sapce */
-	mds = md_init_space();
-
-	/* If it is restarted */
-	ret = get_node_space(&space_size);
-	if (space_size != 0) {
-		sys->disk_space = space_size;
-		goto out;
-	}
-
-	/* User has specified the space at startup */
-	if (sys->disk_space) {
-		ret = set_node_space(sys->disk_space);
-		goto out;
-	}
-
-	if (mds) {
-		sys->disk_space = mds;
-	} else {
-		ret = statvfs(base_path, &fs);
-		if (ret < 0) {
-			sd_debug("get disk space failed %m");
-			ret = SD_RES_EIO;
-			goto out;
-		}
-		sys->disk_space = (uint64_t)fs.f_frsize * fs.f_bavail;
-	}
-
-	ret = set_node_space(sys->disk_space);
-out:
-	sd_debug("disk free space is %" PRIu64, sys->disk_space);
-	return ret;
-}
-
-/* Initialize all the global pathnames used internally */
-int init_global_pathnames(const char *d, char *argp)
-{
-	int ret;
-
-	ret = init_obj_path(d, argp);
-	if (ret)
-		return ret;
-
-	ret = init_epoch_path(d);
-	if (ret)
-		return ret;
-
-	init_config_path(d);
-
-	return 0;
-}
-
-/* Write data to both local object cache (if enabled) and backends */
-int sd_write_object(uint64_t oid, char *data, unsigned int datalen,
-		    uint64_t offset, bool create)
-{
-	struct sd_req hdr;
-	int ret;
-
-	if (sys->enable_object_cache && object_is_cached(oid)) {
-		ret = object_cache_write(oid, data, datalen, offset,
-					 create);
-		if (ret == SD_RES_NO_CACHE)
-			goto forward_write;
-
-		if (ret != 0) {
-			sd_err("write cache failed %" PRIx64 " %" PRIx32, oid,
-			       ret);
-			return ret;
-		}
-	}
-
-forward_write:
-	if (create)
-		sd_init_req(&hdr, SD_OP_CREATE_AND_WRITE_OBJ);
-	else
-		sd_init_req(&hdr, SD_OP_WRITE_OBJ);
-	hdr.flags = SD_FLAG_CMD_WRITE;
-	hdr.data_length = datalen;
-
-	hdr.obj.oid = oid;
-	hdr.obj.offset = offset;
-
-	ret = exec_local_req(&hdr, data);
-	if (ret != SD_RES_SUCCESS)
-		sd_err("failed to write object %" PRIx64 ", %s", oid,
-			   sd_strerror(ret));
-
-	return ret;
-}
-
-int read_backend_object(uint64_t oid, char *data, unsigned int datalen,
-			uint64_t offset)
-{
-	struct sd_req hdr;
-	int ret;
-
-	sd_init_req(&hdr, SD_OP_READ_OBJ);
-	hdr.data_length = datalen;
-	hdr.obj.oid = oid;
-	hdr.obj.offset = offset;
-
-	ret = exec_local_req(&hdr, data);
-	if (ret != SD_RES_SUCCESS)
-		sd_err("failed to read object %" PRIx64 ", %s", oid,
-		       sd_strerror(ret));
-	return ret;
-}
-
-/*
- * Read data firstly from local object cache(if enabled), if fail,
- * try read backends
- */
-int sd_read_object(uint64_t oid, char *data, unsigned int datalen,
-		   uint64_t offset)
-{
-	int ret;
-
-	if (sys->enable_object_cache && object_is_cached(oid)) {
-		ret = object_cache_read(oid, data, datalen, offset);
-		if (ret != SD_RES_SUCCESS) {
-			sd_err("try forward read %" PRIx64 " %s", oid,
-			       sd_strerror(ret));
-			goto forward_read;
-		}
-		return ret;
-	}
-
-forward_read:
-	return read_backend_object(oid, data, datalen, offset);
-}
-
-int sd_remove_object(uint64_t oid)
-{
-	struct sd_req hdr;
-	int ret;
-
-	if (sys->enable_object_cache && object_is_cached(oid)) {
-		ret = object_cache_remove(oid);
-		if (ret != SD_RES_SUCCESS)
-			return ret;
-	}
-
-	sd_init_req(&hdr, SD_OP_REMOVE_OBJ);
-	hdr.obj.oid = oid;
-
-	ret = exec_local_req(&hdr, NULL);
-	if (ret != SD_RES_SUCCESS)
-		sd_err("failed to remove object %" PRIx64 ", %s", oid,
-		       sd_strerror(ret));
-
-	return ret;
-}
-
-int sd_discard_object(uint64_t oid)
-{
-	int ret;
-	struct sd_req hdr;
-
-	sd_init_req(&hdr, SD_OP_DISCARD_OBJ);
-	hdr.obj.oid = oid;
-
-	ret = exec_local_req(&hdr, NULL);
-	if (ret != SD_RES_SUCCESS)
-		sd_err("Failed to discard data obj %"PRIu64" %s", oid,
-		       sd_strerror(ret));
-
-	return ret;
-}
-
-int sd_dec_object_refcnt(uint64_t data_oid, uint32_t generation,
-			 uint32_t refcnt)
-{
-	struct sd_req hdr;
-	int ret;
-	uint64_t ledger_oid = data_oid_to_ledger_oid(data_oid);
-
-	sd_debug("%"PRIx64", %" PRId32 ", %" PRId32,
-		 data_oid, generation, refcnt);
-
-	if (generation == 0 && refcnt == 0)
-		return sd_remove_object(data_oid);
-
-	sd_init_req(&hdr, SD_OP_DECREF_OBJ);
-	hdr.ref.oid = ledger_oid;
-	hdr.ref.generation = generation;
-	hdr.ref.count = refcnt;
-
-	ret = exec_local_req(&hdr, NULL);
-	if (ret != SD_RES_SUCCESS)
-		sd_err("failed to decrement reference %" PRIx64 ", %s",
-		       ledger_oid, sd_strerror(ret));
-
-	return ret;
-}
diff --git a/sheep/store/common.c b/sheep/store/common.c
new file mode 100644
index 0000000..8843fb8
--- /dev/null
+++ b/sheep/store/common.c
@@ -0,0 +1,511 @@
+/*
+ * Copyright (C) 2009-2011 Nippon Telegraph and Telephone Corporation.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License version
+ * 2 as published by the Free Software Foundation.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "sheep_priv.h"
+
+char *obj_path;
+char *epoch_path;
+
+struct store_driver *sd_store;
+LIST_HEAD(store_drivers);
+
+int update_epoch_log(uint32_t epoch, struct sd_node *nodes, size_t nr_nodes)
+{
+	int ret, len, nodes_len;
+	time_t t;
+	char path[PATH_MAX], *buf;
+
+	sd_debug("update epoch: %d, %zu", epoch, nr_nodes);
+
+	/* Piggyback the epoch creation time for 'dog cluster info' */
+	time(&t);
+	nodes_len = nr_nodes * sizeof(struct sd_node);
+	len = nodes_len + sizeof(time_t);
+	buf = xmalloc(len);
+	memcpy(buf, nodes, nodes_len);
+	memcpy(buf + nodes_len, &t, sizeof(time_t));
+
+	/*
+	 * rb field is unused in epoch file, zero-filling it
+	 * is good for epoch file recovery because it is unified
+	 */
+	for (int i = 0; i < nr_nodes; i++)
+		memset(buf + i * sizeof(struct sd_node)
+				+ offsetof(struct sd_node, rb),
+				0, sizeof(struct rb_node));
+
+	snprintf(path, sizeof(path), "%s%08u", epoch_path, epoch);
+
+	ret = atomic_create_and_write(path, buf, len, true);
+
+	free(buf);
+	return ret;
+}
+
+static int do_epoch_log_read(uint32_t epoch, struct sd_node *nodes, int len,
+			     int *nr_nodes, time_t *timestamp)
+{
+	int fd, ret, buf_len;
+	char path[PATH_MAX];
+	struct stat epoch_stat;
+
+	snprintf(path, sizeof(path), "%s%08u", epoch_path, epoch);
+	fd = open(path, O_RDONLY);
+	if (fd < 0) {
+		sd_debug("failed to open epoch %"PRIu32" log, %m", epoch);
+		goto err;
+	}
+
+	memset(&epoch_stat, 0, sizeof(epoch_stat));
+	ret = fstat(fd, &epoch_stat);
+	if (ret < 0) {
+		sd_err("failed to stat epoch %"PRIu32" log, %m", epoch);
+		goto err;
+	}
+
+	buf_len = epoch_stat.st_size - sizeof(*timestamp);
+	if (buf_len < 0) {
+		sd_err("invalid epoch %"PRIu32" log", epoch);
+		goto err;
+	}
+	if (len < buf_len) {
+		close(fd);
+		return SD_RES_BUFFER_SMALL;
+	}
+
+	ret = xread(fd, nodes, buf_len);
+	if (ret < 0) {
+		sd_err("failed to read epoch %"PRIu32" log, %m", epoch);
+		goto err;
+	}
+
+	/* Broken epoch, just ignore */
+	if (ret % sizeof(struct sd_node) != 0) {
+		sd_err("invalid epoch %"PRIu32" log", epoch);
+		goto err;
+	}
+
+	*nr_nodes = ret / sizeof(struct sd_node);
+
+	if (timestamp) {
+		ret = xread(fd, timestamp, sizeof(*timestamp));
+		if (ret != sizeof(*timestamp)) {
+			sd_err("invalid epoch %"PRIu32" log", epoch);
+			goto err;
+		}
+	}
+
+	close(fd);
+	return SD_RES_SUCCESS;
+err:
+	if (fd >= 0)
+		close(fd);
+	return SD_RES_NO_TAG;
+}
+
+int epoch_log_read(uint32_t epoch, struct sd_node *nodes,
+				int len, int *nr_nodes)
+{
+	return do_epoch_log_read(epoch, nodes, len, nr_nodes, NULL);
+}
+
+int epoch_log_read_with_timestamp(uint32_t epoch, struct sd_node *nodes,
+				int len, int *nr_nodes, time_t *timestamp)
+{
+	return do_epoch_log_read(epoch, nodes, len, nr_nodes, timestamp);
+}
+
+uint32_t get_latest_epoch(void)
+{
+	DIR *dir;
+	struct dirent *d;
+	uint32_t e, epoch = 0;
+	char *p;
+
+	dir = opendir(epoch_path);
+	if (!dir)
+		panic("failed to get the latest epoch: %m");
+
+	while ((d = readdir(dir))) {
+		e = strtol(d->d_name, &p, 10);
+		if (d->d_name == p)
+			continue;
+
+		if (strlen(d->d_name) != 8)
+			continue;
+
+		if (e > epoch)
+			epoch = e;
+	}
+	closedir(dir);
+
+	return epoch;
+}
+
+int lock_base_dir(const char *d)
+{
+#define LOCK_PATH "/lock"
+	char *lock_path;
+	int ret = 0;
+	int fd, len = strlen(d) + strlen(LOCK_PATH) + 1;
+
+	lock_path = xzalloc(len);
+	snprintf(lock_path, len, "%s" LOCK_PATH, d);
+
+	fd = open(lock_path, O_WRONLY|O_CREAT, sd_def_fmode);
+	if (fd < 0) {
+		sd_err("failed to open lock file %s (%m)", lock_path);
+		ret = -1;
+		goto out;
+	}
+
+	if (lockf(fd, F_TLOCK, 1) < 0) {
+		if (errno == EACCES || errno == EAGAIN)
+			sd_err("another sheep daemon is using %s", d);
+		else
+			sd_err("unable to get base dir lock (%m)");
+		ret = -1;
+		goto out;
+	}
+
+out:
+	free(lock_path);
+	return ret;
+}
+
+int init_base_path(const char *d)
+{
+	if (xmkdir(d, sd_def_dmode) < 0) {
+		sd_err("cannot create the directory %s (%m)", d);
+		return -1;
+	}
+
+	return 0;
+}
+
+static inline int check_path_len(const char *path)
+{
+	int len = strlen(path);
+	if (len > PATH_MAX) {
+		sd_err("insanely long object directory %s", path);
+		return -1;
+	}
+
+	return 0;
+}
+
+static int is_meta_store(const char *path)
+{
+	char conf[PATH_MAX];
+	char epoch[PATH_MAX];
+
+	snprintf(conf, PATH_MAX, "%s/config", path);
+	snprintf(epoch, PATH_MAX, "%s/epoch", path);
+	if (!access(conf, R_OK) && !access(epoch, R_OK))
+		return true;
+
+	return false;
+}
+
+static int init_obj_path(const char *base_path, char *argp)
+{
+	char *p;
+	int len;
+
+	if (check_path_len(base_path) < 0)
+		return -1;
+
+#define OBJ_PATH "/obj"
+	len = strlen(base_path) + strlen(OBJ_PATH) + 1;
+	obj_path = xzalloc(len);
+	snprintf(obj_path, len, "%s" OBJ_PATH, base_path);
+
+	/* Eat up the first component */
+	strtok(argp, ",");
+	p = strtok(NULL, ",");
+	if (!p) {
+		/*
+		 * If We have only one path, meta-store and object-store share
+		 * it. This is helpful to upgrade old sheep cluster to
+		 * the MD-enabled.
+		 */
+		md_add_disk(obj_path, false);
+	} else {
+		do {
+			if (is_meta_store(p)) {
+				sd_err("%s is meta-store, abort", p);
+				return -1;
+			}
+			md_add_disk(p, false);
+		} while ((p = strtok(NULL, ",")));
+	}
+
+	if (md_nr_disks() <= 0) {
+		sd_err("There isn't any available disk!");
+		return -1;
+	}
+
+	return xmkdir(obj_path, sd_def_dmode);
+}
+
+static int init_epoch_path(const char *base_path)
+{
+#define EPOCH_PATH "/epoch/"
+	int len = strlen(base_path) + strlen(EPOCH_PATH) + 1;
+	epoch_path = xzalloc(len);
+	snprintf(epoch_path, len, "%s" EPOCH_PATH, base_path);
+
+	return xmkdir(epoch_path, sd_def_dmode);
+}
+
+/*
+ * If the node is gateway, this function only finds the store driver.
+ * Otherwise, this function initializes the backend store
+ */
+int init_store_driver(bool is_gateway)
+{
+	char driver_name[STORE_LEN], *p;
+
+	pstrcpy(driver_name, sizeof(driver_name), (char *)sys->cinfo.store);
+
+	p = memchr(driver_name, '\0', STORE_LEN);
+	if (!p) {
+		/*
+		 * If the driver name is not NUL terminated we are in deep
+		 * trouble, let's get out here.
+		 */
+		sd_debug("store name not NUL terminated");
+		return SD_RES_NO_STORE;
+	}
+
+	/*
+	 * The store file might not exist in case this is a new sheep that
+	 * never joined a cluster before.
+	 */
+	if (p == driver_name)
+		return 0;
+
+	sd_store = find_store_driver(driver_name);
+	if (!sd_store) {
+		sd_debug("store %s not found", driver_name);
+		return SD_RES_NO_STORE;
+	}
+
+	if (is_gateway)
+		return SD_RES_SUCCESS;
+
+	return sd_store->init();
+}
+
+int init_disk_space(const char *base_path)
+{
+	int ret = SD_RES_SUCCESS;
+	uint64_t space_size = 0, mds;
+	struct statvfs fs;
+
+	if (sys->gateway_only)
+		goto out;
+
+	/* We need to init md even we don't need to update sapce */
+	mds = md_init_space();
+
+	/* If it is restarted */
+	ret = get_node_space(&space_size);
+	if (space_size != 0) {
+		sys->disk_space = space_size;
+		goto out;
+	}
+
+	/* User has specified the space at startup */
+	if (sys->disk_space) {
+		ret = set_node_space(sys->disk_space);
+		goto out;
+	}
+
+	if (mds) {
+		sys->disk_space = mds;
+	} else {
+		ret = statvfs(base_path, &fs);
+		if (ret < 0) {
+			sd_debug("get disk space failed %m");
+			ret = SD_RES_EIO;
+			goto out;
+		}
+		sys->disk_space = (uint64_t)fs.f_frsize * fs.f_bavail;
+	}
+
+	ret = set_node_space(sys->disk_space);
+out:
+	sd_debug("disk free space is %" PRIu64, sys->disk_space);
+	return ret;
+}
+
+/* Initialize all the global pathnames used internally */
+int init_global_pathnames(const char *d, char *argp)
+{
+	int ret;
+
+	ret = init_obj_path(d, argp);
+	if (ret)
+		return ret;
+
+	ret = init_epoch_path(d);
+	if (ret)
+		return ret;
+
+	init_config_path(d);
+
+	return 0;
+}
+
+/* Write data to both local object cache (if enabled) and backends */
+int sd_write_object(uint64_t oid, char *data, unsigned int datalen,
+		    uint64_t offset, bool create)
+{
+	struct sd_req hdr;
+	int ret;
+
+	if (sys->enable_object_cache && object_is_cached(oid)) {
+		ret = object_cache_write(oid, data, datalen, offset,
+					 create);
+		if (ret == SD_RES_NO_CACHE)
+			goto forward_write;
+
+		if (ret != 0) {
+			sd_err("write cache failed %" PRIx64 " %" PRIx32, oid,
+			       ret);
+			return ret;
+		}
+	}
+
+forward_write:
+	if (create)
+		sd_init_req(&hdr, SD_OP_CREATE_AND_WRITE_OBJ);
+	else
+		sd_init_req(&hdr, SD_OP_WRITE_OBJ);
+	hdr.flags = SD_FLAG_CMD_WRITE;
+	hdr.data_length = datalen;
+
+	hdr.obj.oid = oid;
+	hdr.obj.offset = offset;
+
+	ret = exec_local_req(&hdr, data);
+	if (ret != SD_RES_SUCCESS)
+		sd_err("failed to write object %" PRIx64 ", %s", oid,
+			   sd_strerror(ret));
+
+	return ret;
+}
+
+int read_backend_object(uint64_t oid, char *data, unsigned int datalen,
+			uint64_t offset)
+{
+	struct sd_req hdr;
+	int ret;
+
+	sd_init_req(&hdr, SD_OP_READ_OBJ);
+	hdr.data_length = datalen;
+	hdr.obj.oid = oid;
+	hdr.obj.offset = offset;
+
+	ret = exec_local_req(&hdr, data);
+	if (ret != SD_RES_SUCCESS)
+		sd_err("failed to read object %" PRIx64 ", %s", oid,
+		       sd_strerror(ret));
+	return ret;
+}
+
+/*
+ * Read data firstly from local object cache(if enabled), if fail,
+ * try read backends
+ */
+int sd_read_object(uint64_t oid, char *data, unsigned int datalen,
+		   uint64_t offset)
+{
+	int ret;
+
+	if (sys->enable_object_cache && object_is_cached(oid)) {
+		ret = object_cache_read(oid, data, datalen, offset);
+		if (ret != SD_RES_SUCCESS) {
+			sd_err("try forward read %" PRIx64 " %s", oid,
+			       sd_strerror(ret));
+			goto forward_read;
+		}
+		return ret;
+	}
+
+forward_read:
+	return read_backend_object(oid, data, datalen, offset);
+}
+
+int sd_remove_object(uint64_t oid)
+{
+	struct sd_req hdr;
+	int ret;
+
+	if (sys->enable_object_cache && object_is_cached(oid)) {
+		ret = object_cache_remove(oid);
+		if (ret != SD_RES_SUCCESS)
+			return ret;
+	}
+
+	sd_init_req(&hdr, SD_OP_REMOVE_OBJ);
+	hdr.obj.oid = oid;
+
+	ret = exec_local_req(&hdr, NULL);
+	if (ret != SD_RES_SUCCESS)
+		sd_err("failed to remove object %" PRIx64 ", %s", oid,
+		       sd_strerror(ret));
+
+	return ret;
+}
+
+int sd_discard_object(uint64_t oid)
+{
+	int ret;
+	struct sd_req hdr;
+
+	sd_init_req(&hdr, SD_OP_DISCARD_OBJ);
+	hdr.obj.oid = oid;
+
+	ret = exec_local_req(&hdr, NULL);
+	if (ret != SD_RES_SUCCESS)
+		sd_err("Failed to discard data obj %"PRIu64" %s", oid,
+		       sd_strerror(ret));
+
+	return ret;
+}
+
+int sd_dec_object_refcnt(uint64_t data_oid, uint32_t generation,
+			 uint32_t refcnt)
+{
+	struct sd_req hdr;
+	int ret;
+	uint64_t ledger_oid = data_oid_to_ledger_oid(data_oid);
+
+	sd_debug("%"PRIx64", %" PRId32 ", %" PRId32,
+		 data_oid, generation, refcnt);
+
+	if (generation == 0 && refcnt == 0)
+		return sd_remove_object(data_oid);
+
+	sd_init_req(&hdr, SD_OP_DECREF_OBJ);
+	hdr.ref.oid = ledger_oid;
+	hdr.ref.generation = generation;
+	hdr.ref.count = refcnt;
+
+	ret = exec_local_req(&hdr, NULL);
+	if (ret != SD_RES_SUCCESS)
+		sd_err("failed to decrement reference %" PRIx64 ", %s",
+		       ledger_oid, sd_strerror(ret));
+
+	return ret;
+}
diff --git a/sheep/store/md.c b/sheep/store/md.c
new file mode 100644
index 0000000..87ab759
--- /dev/null
+++ b/sheep/store/md.c
@@ -0,0 +1,878 @@
+/*
+ * Copyright (C) 2013 Taobao Inc.
+ *
+ * Liu Yuan <namei.unix at gmail.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License version
+ * 2 as published by the Free Software Foundation.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "sheep_priv.h"
+
+#define MD_VDISK_SIZE ((uint64_t)1*1024*1024*1024) /* 1G */
+
+#define NONE_EXIST_PATH "/all/disks/are/broken/,ps/ノ厂経7/!"
+
+struct md md = {
+	.vroot = RB_ROOT,
+	.root = RB_ROOT,
+	.lock = SD_RW_LOCK_INITIALIZER,
+};
+
+static inline uint32_t nr_online_disks(void)
+{
+	uint32_t nr;
+
+	sd_read_lock(&md.lock);
+	nr = md.nr_disks;
+	sd_rw_unlock(&md.lock);
+
+	return nr;
+}
+
+static inline int vdisk_number(const struct disk *disk)
+{
+	return DIV_ROUND_UP(disk->space, MD_VDISK_SIZE);
+}
+
+static int disk_cmp(const struct disk *d1, const struct disk *d2)
+{
+	return strcmp(d1->path, d2->path);
+}
+
+static int vdisk_cmp(const struct vdisk *d1, const struct vdisk *d2)
+{
+	return intcmp(d1->hash, d2->hash);
+}
+
+static struct vdisk *vdisk_insert(struct vdisk *new)
+{
+	return rb_insert(&md.vroot, new, rb, vdisk_cmp);
+}
+
+/* If v1_hash < hval <= v2_hash, then oid is resident in v2 */
+static struct vdisk *hval_to_vdisk(uint64_t hval)
+{
+	struct vdisk dummy = { .hash = hval };
+
+	return rb_nsearch(&md.vroot, &dummy, rb, vdisk_cmp);
+}
+
+static struct vdisk *oid_to_vdisk(uint64_t oid)
+{
+	return hval_to_vdisk(sd_hash_oid(oid));
+}
+
+static void create_vdisks(const struct disk *disk)
+{
+	uint64_t hval = sd_hash(disk->path, strlen(disk->path));
+	const struct sd_node *n = &sys->this_node;
+	uint64_t node_hval;
+	int nr;
+
+	if (is_cluster_diskmode(&sys->cinfo)) {
+		node_hval = sd_hash(&n->nid, offsetof(typeof(n->nid), io_addr));
+		hval = fnv_64a_64(node_hval, hval);
+		nr = DIV_ROUND_UP(disk->space, WEIGHT_MIN);
+		if (0 == n->nid.port)
+			return;
+	} else
+		nr = vdisk_number(disk);
+
+	for (int i = 0; i < nr; i++) {
+		struct vdisk *v = xmalloc(sizeof(*v));
+
+		hval = sd_hash_next(hval);
+		v->hash = hval;
+		v->disk = disk;
+		if (unlikely(vdisk_insert(v)))
+			panic("vdisk hash collison");
+	}
+}
+
+static inline void vdisk_free(struct vdisk *v)
+{
+	rb_erase(&v->rb, &md.vroot);
+	free(v);
+}
+
+static void remove_vdisks(const struct disk *disk)
+{
+	uint64_t hval = sd_hash(disk->path, strlen(disk->path));
+	const struct sd_node *n = &sys->this_node;
+	uint64_t node_hval;
+	int nr;
+
+	if (is_cluster_diskmode(&sys->cinfo)) {
+		node_hval = sd_hash(&n->nid, offsetof(typeof(n->nid), io_addr));
+		hval = fnv_64a_64(node_hval, hval);
+		nr = DIV_ROUND_UP(disk->space, WEIGHT_MIN);
+	} else
+		nr = vdisk_number(disk);
+
+	for (int i = 0; i < nr; i++) {
+		struct vdisk *v;
+
+		hval = sd_hash_next(hval);
+		v = hval_to_vdisk(hval);
+		assert(v->hash == hval);
+
+		vdisk_free(v);
+	}
+}
+
+static inline void trim_last_slash(char *path)
+{
+	assert(path[0]);
+	while (path[strlen(path) - 1] == '/')
+		path[strlen(path) - 1] = '\0';
+}
+
+static struct disk *path_to_disk(const char *path)
+{
+	struct disk key = {};
+
+	pstrcpy(key.path, sizeof(key.path), path);
+	trim_last_slash(key.path);
+
+	return rb_search(&md.root, &key, rb, disk_cmp);
+}
+
+size_t get_store_objsize(uint64_t oid)
+{
+	if (is_erasure_oid(oid)) {
+		uint8_t policy = get_vdi_copy_policy(oid_to_vid(oid));
+		int d;
+		ec_policy_to_dp(policy, &d, NULL);
+		return get_vdi_object_size(oid_to_vid(oid)) / d;
+	}
+	return get_objsize(oid, get_vdi_object_size(oid_to_vid(oid)));
+}
+
+static int get_total_object_size(uint64_t oid, const char *wd, uint32_t epoch,
+				 uint8_t ec_index, struct vnode_info *vinfo,
+				 void *total)
+{
+	uint64_t *t = total;
+	struct stat s;
+	char path[PATH_MAX];
+
+	snprintf(path, PATH_MAX, "%s/%016" PRIx64, wd, oid);
+	if (stat(path, &s) == 0)
+		*t += s.st_blocks * SECTOR_SIZE;
+	else
+		*t += get_store_objsize(oid);
+
+	return SD_RES_SUCCESS;
+}
+
+static int64_t find_string_integer(const char *str, const char *delimiter)
+{
+	char *pos = strstr(str, delimiter), *p;
+	int64_t ret;
+
+	ret = strtoll(pos + 1, &p, 10);
+	if (ret == LLONG_MAX || p == pos + 1) {
+		sd_err("%s strtoul failed, delimiter %s, %m", str, delimiter);
+		return -1;
+	}
+
+	return ret;
+}
+
+/* If cleanup is true, temporary objects will be removed */
+static int for_each_object_in_path(const char *path,
+				   int (*func)(uint64_t, const char *, uint32_t,
+					       uint8_t, struct vnode_info *,
+					       void *),
+				   bool cleanup, struct vnode_info *vinfo,
+				   void *arg)
+{
+	DIR *dir;
+	struct dirent *d;
+	uint64_t oid;
+	int ret = SD_RES_SUCCESS;
+	char file_name[PATH_MAX];
+
+	dir = opendir(path);
+	if (unlikely(!dir)) {
+		sd_err("failed to open %s, %m", path);
+		return SD_RES_EIO;
+	}
+
+	while ((d = readdir(dir))) {
+		uint32_t epoch = 0;
+		uint8_t ec_index = SD_MAX_COPIES;
+
+		/* skip ".", ".." and ".stale" */
+		if (unlikely(!strncmp(d->d_name, ".", 1)))
+			continue;
+
+		sd_debug("%s, %s", path, d->d_name);
+		oid = strtoull(d->d_name, NULL, 16);
+		if (oid == 0 || oid == ULLONG_MAX)
+			continue;
+
+		/* don't call callback against temporary objects */
+		if (is_tmp_dentry(d->d_name)) {
+			if (cleanup) {
+				snprintf(file_name, sizeof(file_name),
+						"%s/%s", path, d->d_name);
+				sd_debug("remove tmp object %s", file_name);
+				if (unlink(file_name) < 0)
+					sd_err("failed to unlink %s: %m",
+							file_name);
+			}
+			continue;
+		}
+
+		if (is_stale_dentry(d->d_name)) {
+			epoch = find_string_integer(d->d_name, ".");
+			if (epoch < 0)
+				continue;
+		}
+
+		if (is_ec_dentry(d->d_name)) {
+			ec_index = find_string_integer(d->d_name, "_");
+			if (ec_index < 0)
+				continue;
+		}
+
+		ret = func(oid, path, epoch, ec_index, vinfo, arg);
+		if (ret != SD_RES_SUCCESS)
+			break;
+	}
+	closedir(dir);
+	return ret;
+}
+
+static uint64_t get_path_free_size(const char *path, uint64_t *used)
+{
+	struct statvfs fs;
+	uint64_t size;
+
+	if (statvfs(path, &fs) < 0) {
+		sd_err("get disk %s space failed %m", path);
+		return 0;
+	}
+	size = (int64_t)fs.f_frsize * fs.f_bavail;
+
+	if (!used)
+		goto out;
+	if (for_each_object_in_path(path, get_total_object_size, false,
+				    NULL, used)
+	    != SD_RES_SUCCESS)
+		return 0;
+out:
+	return size;
+}
+
+/*
+ * If path is broken during initialization or not support xattr return 0. We can
+ * safely use 0 to represent failure case  because 0 space path can be
+ * considered as broken path.
+ */
+static uint64_t init_path_space(const char *path, bool purge)
+{
+	uint64_t size;
+	char stale[PATH_MAX];
+
+	if (!is_xattr_enabled(path)) {
+		sd_warn("multi-disk support need xattr feature for path: %s",
+			path);
+		goto broken_path;
+	}
+
+	if (purge && purge_directory(path) < 0)
+		sd_err("failed to purge %s", path);
+
+	snprintf(stale, PATH_MAX, "%s/.stale", path);
+	if (xmkdir(stale, sd_def_dmode) < 0) {
+		sd_err("can't mkdir for %s, %m", stale);
+		goto broken_path;
+	}
+
+#define MDNAME	"user.md.size"
+#define MDSIZE	sizeof(uint64_t)
+	if (getxattr(path, MDNAME, &size, MDSIZE) < 0) {
+		if (errno == ENODATA) {
+			goto create;
+		} else {
+			sd_err("%s, %m", path);
+			goto broken_path;
+		}
+	}
+
+	return size;
+create:
+	size = get_path_free_size(path, NULL);
+	if (!size)
+		goto broken_path;
+	if (setxattr(path, MDNAME, &size, MDSIZE, 0) < 0) {
+		sd_err("%s, %m", path);
+		goto broken_path;
+	}
+	return size;
+broken_path:
+	return 0;
+}
+
+/* We don't need lock at init stage */
+bool md_add_disk(const char *path, bool purge)
+{
+	struct disk *new;
+
+	if (path_to_disk(path)) {
+		sd_err("duplicate path %s", path);
+		return false;
+	}
+
+	if (xmkdir(path, sd_def_dmode) < 0) {
+		sd_err("can't mkdir for %s, %m", path);
+		return false;
+	}
+
+	new = xmalloc(sizeof(*new));
+	pstrcpy(new->path, PATH_MAX, path);
+	trim_last_slash(new->path);
+	new->space = init_path_space(new->path, purge);
+	if (!new->space) {
+		free(new);
+		return false;
+	}
+
+	create_vdisks(new);
+	rb_insert(&md.root, new, rb, disk_cmp);
+	md.space += new->space;
+	md.nr_disks++;
+
+	sd_info("%s, vdisk nr %d, total disk %d", new->path, vdisk_number(new),
+		md.nr_disks);
+	return true;
+}
+
+static inline void md_remove_disk(struct disk *disk)
+{
+	sd_info("%s from multi-disk array", disk->path);
+	rb_erase(&disk->rb, &md.root);
+	md.nr_disks--;
+	remove_vdisks(disk);
+	free(disk);
+}
+
+uint64_t md_init_space(void)
+{
+	return md.space;
+}
+
+static const char *md_get_object_dir_nolock(uint64_t oid)
+{
+	const struct vdisk *vd;
+
+	if (unlikely(md.nr_disks == 0))
+		return NONE_EXIST_PATH; /* To generate EIO */
+
+	vd = oid_to_vdisk(oid);
+	return vd->disk->path;
+}
+
+const char *md_get_object_dir(uint64_t oid)
+{
+	const char *p;
+
+	sd_read_lock(&md.lock);
+	p = md_get_object_dir_nolock(oid);
+	sd_rw_unlock(&md.lock);
+
+	return p;
+}
+
+struct process_path_arg {
+	const char *path;
+	struct vnode_info *vinfo;
+	int (*func)(uint64_t oid, const char *, uint32_t, uint8_t,
+		    struct vnode_info *, void *arg);
+	bool cleanup;
+	void *opaque;
+	int result;
+};
+
+static void *thread_process_path(void *arg)
+{
+	int ret;
+	struct process_path_arg *parg = (struct process_path_arg *)arg;
+
+	ret = for_each_object_in_path(parg->path, parg->func, parg->cleanup,
+				      parg->vinfo, parg->opaque);
+	if (ret != SD_RES_SUCCESS)
+		parg->result = ret;
+
+	return arg;
+}
+
+main_fn int for_each_object_in_wd(int (*func)(uint64_t oid, const char *path,
+				      uint32_t epoch, uint8_t ec_index,
+				      struct vnode_info *vinfo, void *arg),
+				  bool cleanup, void *arg)
+{
+	int ret = SD_RES_SUCCESS;
+	const struct disk *disk;
+	struct process_path_arg *thread_args, *path_arg;
+	struct vnode_info *vinfo;
+	void *ret_arg;
+	sd_thread_t *thread_array;
+	int nr_thread = 0, idx = 0;
+
+	sd_read_lock(&md.lock);
+
+	rb_for_each_entry(disk, &md.root, rb) {
+		nr_thread++;
+	}
+
+	thread_args = xmalloc(nr_thread * sizeof(struct process_path_arg));
+	thread_array = xmalloc(nr_thread * sizeof(sd_thread_t));
+
+	vinfo = get_vnode_info();
+
+	rb_for_each_entry(disk, &md.root, rb) {
+		thread_args[idx].path = disk->path;
+		thread_args[idx].vinfo = vinfo;
+		thread_args[idx].func = func;
+		thread_args[idx].cleanup = cleanup;
+		thread_args[idx].opaque = arg;
+		thread_args[idx].result = SD_RES_SUCCESS;
+		ret = sd_thread_create_with_idx("foreach wd",
+						thread_array + idx,
+						thread_process_path,
+						(void *)(thread_args + idx));
+		if (ret) {
+			/*
+			 * If we can't create enough threads to process
+			 * files, the data-consistent will be broken if
+			 * we continued.
+			 */
+			panic("Failed to create thread for path %s",
+			      disk->path);
+		}
+		idx++;
+	}
+
+	sd_debug("Create %d threads for all path", nr_thread);
+	/* wait for all threads to exit */
+	for (idx = 0; idx < nr_thread; idx++) {
+		ret = sd_thread_join(thread_array[idx], &ret_arg);
+		if (ret)
+			sd_err("Failed to join thread");
+		if (ret_arg) {
+			path_arg = (struct process_path_arg *)ret_arg;
+			if (path_arg->result != SD_RES_SUCCESS)
+				sd_err("%s, %s", path_arg->path,
+				       sd_strerror(path_arg->result));
+		}
+	}
+
+	put_vnode_info(vinfo);
+	sd_rw_unlock(&md.lock);
+
+	free(thread_args);
+	free(thread_array);
+	return ret;
+}
+
+int for_each_object_in_stale(int (*func)(uint64_t oid, const char *path,
+					 uint32_t epoch, uint8_t,
+					 struct vnode_info *, void *arg),
+			     void *arg)
+{
+	int ret = SD_RES_SUCCESS;
+	char path[PATH_MAX];
+	const struct disk *disk;
+
+	sd_read_lock(&md.lock);
+	rb_for_each_entry(disk, &md.root, rb) {
+		snprintf(path, sizeof(path), "%s/.stale", disk->path);
+		ret = for_each_object_in_path(path, func, false, NULL, arg);
+		if (ret != SD_RES_SUCCESS)
+			break;
+	}
+	sd_rw_unlock(&md.lock);
+	return ret;
+}
+
+
+int for_each_obj_path(int (*func)(const char *path))
+{
+	int ret = SD_RES_SUCCESS;
+	const struct disk *disk;
+
+	sd_read_lock(&md.lock);
+	rb_for_each_entry(disk, &md.root, rb) {
+		ret = func(disk->path);
+		if (ret != SD_RES_SUCCESS)
+			break;
+	}
+	sd_rw_unlock(&md.lock);
+	return ret;
+}
+
+struct md_work {
+	struct work work;
+	char path[PATH_MAX];
+};
+
+static inline void kick_recover(void)
+{
+	struct vnode_info *vinfo = get_vnode_info();
+
+	if (is_cluster_diskmode(&sys->cinfo))
+		sys->cdrv->update_node(&sys->this_node);
+	else {
+		start_recovery(vinfo, vinfo, false);
+		put_vnode_info(vinfo);
+	}
+}
+
+static void md_do_recover(struct work *work)
+{
+	struct md_work *mw = container_of(work, struct md_work, work);
+	struct disk *disk;
+	int nr = 0;
+
+	sd_write_lock(&md.lock);
+	disk = path_to_disk(mw->path);
+	if (!disk)
+		/* Just ignore the duplicate EIO of the same path */
+		goto out;
+	md_remove_disk(disk);
+	nr = md.nr_disks;
+out:
+	sd_rw_unlock(&md.lock);
+
+	if (disk) {
+		if (nr > 0) {
+			update_node_disks();
+			kick_recover();
+		} else {
+			leave_cluster();
+		}
+	}
+
+	free(mw);
+}
+
+int md_handle_eio(const char *fault_path)
+{
+	struct md_work *mw;
+
+	if (nr_online_disks() == 0)
+		return SD_RES_EIO;
+
+	mw = xzalloc(sizeof(*mw));
+	mw->work.done = md_do_recover;
+	pstrcpy(mw->path, PATH_MAX, fault_path);
+	queue_work(sys->md_wqueue, &mw->work);
+
+	/* Fool the requester to retry */
+	return SD_RES_NETWORK_ERROR;
+}
+
+static inline bool md_access(const char *path)
+{
+	if (access(path, R_OK | W_OK) < 0) {
+		if (unlikely(errno != ENOENT))
+			sd_err("failed to check %s, %m", path);
+		return false;
+	}
+
+	return true;
+}
+
+static int get_old_new_path(uint64_t oid, uint32_t epoch, uint8_t ec_index,
+			    const char *path, char *old, char *new)
+{
+	if (!epoch) {
+		if (!is_erasure_oid(oid)) {
+			snprintf(old, PATH_MAX, "%s/%016" PRIx64, path, oid);
+			snprintf(new, PATH_MAX, "%s/%016" PRIx64,
+				 md_get_object_dir_nolock(oid), oid);
+		} else {
+			snprintf(old, PATH_MAX, "%s/%016" PRIx64"_%d", path,
+				 oid, ec_index);
+			snprintf(new, PATH_MAX, "%s/%016" PRIx64"_%d",
+				 md_get_object_dir_nolock(oid), oid, ec_index);
+		}
+	} else {
+		if (!is_erasure_oid(oid)) {
+			snprintf(old, PATH_MAX,
+				 "%s/.stale/%016"PRIx64".%"PRIu32, path,
+				 oid, epoch);
+			snprintf(new, PATH_MAX,
+				 "%s/.stale/%016"PRIx64".%"PRIu32,
+				 md_get_object_dir_nolock(oid), oid, epoch);
+		} else {
+			snprintf(old, PATH_MAX,
+				 "%s/.stale/%016"PRIx64"_%d.%"PRIu32, path,
+				 oid, ec_index, epoch);
+			snprintf(new, PATH_MAX,
+				 "%s/.stale/%016"PRIx64"_%d.%"PRIu32,
+				 md_get_object_dir_nolock(oid),
+				 oid, ec_index, epoch);
+		}
+	}
+
+	if (!md_access(old))
+		return -1;
+
+	return 0;
+}
+
+static int md_move_object(uint64_t oid, const char *old, const char *new)
+{
+	struct strbuf buf = STRBUF_INIT;
+	int fd, ret = -1;
+	size_t sz = get_store_objsize(oid);
+
+	fd = open(old, O_RDONLY);
+	if (fd < 0) {
+		sd_err("failed to open %s", old);
+		goto out;
+	}
+
+	ret = strbuf_read(&buf, fd, sz);
+	if (ret != sz) {
+		sd_err("failed to read %s, size %zu, %d, %m", old, sz, ret);
+		ret = -1;
+		goto out_close;
+	}
+
+	if (atomic_create_and_write(new, buf.buf, buf.len, false) < 0) {
+		if (errno != EEXIST) {
+			sd_err("failed to create %s", new);
+			ret = -1;
+			goto out_close;
+		}
+	}
+	unlink(old);
+	ret = 0;
+out_close:
+	close(fd);
+out:
+	strbuf_release(&buf);
+	return ret;
+}
+
+static int md_check_and_move(uint64_t oid, uint32_t epoch, uint8_t ec_index,
+			     const char *path)
+{
+	char old[PATH_MAX], new[PATH_MAX];
+
+	if (get_old_new_path(oid, epoch, ec_index, path, old, new) < 0)
+		return SD_RES_EIO;
+	/*
+	 * Recovery thread and main thread might try to recover the same object.
+	 * Either one succeeds, the other will fail and proceed and end up
+	 * trying to move the object to where it is already in place, in this
+	 * case we simply return.
+	 */
+	if (!strcmp(old, new))
+		return SD_RES_SUCCESS;
+
+	/* We can't use rename(2) across device */
+	if (md_move_object(oid, old, new) < 0) {
+		sd_err("move old %s to new %s failed", old, new);
+		return SD_RES_EIO;
+	}
+
+	sd_debug("from %s to %s", old, new);
+	return SD_RES_SUCCESS;
+}
+
+static int scan_wd(uint64_t oid, uint32_t epoch, uint8_t ec_index)
+{
+	int ret = SD_RES_EIO;
+	const struct disk *disk;
+
+	sd_read_lock(&md.lock);
+	rb_for_each_entry(disk, &md.root, rb) {
+		ret = md_check_and_move(oid, epoch, ec_index, disk->path);
+		if (ret == SD_RES_SUCCESS)
+			break;
+	}
+	sd_rw_unlock(&md.lock);
+	return ret;
+}
+
+bool md_exist(uint64_t oid, uint8_t ec_index, char *path)
+{
+	if (md_access(path))
+		return true;
+	/*
+	 * We have to iterate the WD because we don't have epoch-like history
+	 * track to locate the objects for multiple disk failure. Simply do
+	 * hard iteration simplify the code a lot.
+	 */
+	if (scan_wd(oid, 0, ec_index) == SD_RES_SUCCESS)
+		return true;
+
+	return false;
+}
+
+int md_get_stale_path(uint64_t oid, uint32_t epoch, uint8_t ec_index,
+		      char *path)
+{
+	if (unlikely(!epoch))
+		panic("invalid 0 epoch");
+
+	if (is_erasure_oid(oid)) {
+		if (unlikely(ec_index >= SD_MAX_COPIES))
+			panic("invalid ec index %d", ec_index);
+
+		snprintf(path, PATH_MAX, "%s/.stale/%016"PRIx64"_%d.%"PRIu32,
+			 md_get_object_dir(oid), oid, ec_index, epoch);
+	} else
+		snprintf(path, PATH_MAX, "%s/.stale/%016"PRIx64".%"PRIu32,
+			 md_get_object_dir(oid), oid, epoch);
+
+	if (md_access(path))
+		return SD_RES_SUCCESS;
+
+	if (scan_wd(oid, epoch, ec_index) == SD_RES_SUCCESS)
+		return SD_RES_SUCCESS;
+
+	return SD_RES_NO_OBJ;
+}
+
+uint32_t md_get_info(struct sd_md_info *info)
+{
+	uint32_t ret = sizeof(*info);
+	const struct disk *disk;
+	int i = 0;
+
+	memset(info, 0, ret);
+	sd_read_lock(&md.lock);
+	rb_for_each_entry(disk, &md.root, rb) {
+		info->disk[i].idx = i;
+		pstrcpy(info->disk[i].path, PATH_MAX, disk->path);
+		/* FIXME: better handling failure case. */
+		info->disk[i].free = get_path_free_size(info->disk[i].path,
+							&info->disk[i].used);
+		i++;
+	}
+	info->nr = md.nr_disks;
+	sd_rw_unlock(&md.lock);
+	return ret;
+}
+
+static inline void md_del_disk(const char *path)
+{
+	struct disk *disk = path_to_disk(path);
+
+	if (!disk) {
+		sd_err("invalid path %s", path);
+		return;
+	}
+	md_remove_disk(disk);
+}
+
+#ifdef HAVE_DISKVNODES
+void update_node_disks(void)
+{
+	const struct disk *disk;
+	int i = 0;
+	bool rb_empty = false;
+
+	if (!sys)
+		return;
+
+	memset(sys->this_node.disks, 0, sizeof(struct disk_info) * DISK_MAX);
+	sd_read_lock(&md.lock);
+	rb_for_each_entry(disk, &md.root, rb) {
+		sys->this_node.disks[i].disk_id =
+			sd_hash(disk->path, strlen(disk->path));
+		sys->this_node.disks[i].disk_space = disk->space;
+		i++;
+	}
+	sd_rw_unlock(&md.lock);
+
+	if (RB_EMPTY_ROOT(&md.vroot))
+		rb_empty = true;
+	sd_write_lock(&md.lock);
+	rb_for_each_entry(disk, &md.root, rb) {
+		if (!rb_empty)
+			remove_vdisks(disk);
+		create_vdisks(disk);
+	}
+	sd_rw_unlock(&md.lock);
+}
+#else
+void update_node_disks(void)
+{
+}
+#endif
+
+static int do_plug_unplug(char *disks, bool plug)
+{
+	const char *path;
+	int old_nr, ret = SD_RES_UNKNOWN;
+
+	sd_write_lock(&md.lock);
+	old_nr = md.nr_disks;
+	path = strtok(disks, ",");
+	do {
+		if (plug) {
+			if (!md_add_disk(path, true))
+				sd_err("failed to add %s", path);
+		} else {
+			md_del_disk(path);
+		}
+	} while ((path = strtok(NULL, ",")));
+
+	/* If no disks change, bail out */
+	if (old_nr == md.nr_disks)
+		goto out;
+
+	ret = SD_RES_SUCCESS;
+out:
+	sd_rw_unlock(&md.lock);
+
+	if (ret == SD_RES_SUCCESS) {
+		update_node_disks();
+		kick_recover();
+	}
+
+	return ret;
+}
+
+int md_plug_disks(char *disks)
+{
+	return do_plug_unplug(disks, true);
+}
+
+int md_unplug_disks(char *disks)
+{
+	return do_plug_unplug(disks, false);
+}
+
+uint64_t md_get_size(uint64_t *used)
+{
+	uint64_t fsize = 0;
+	const struct disk *disk;
+
+	*used = 0;
+	sd_read_lock(&md.lock);
+	rb_for_each_entry(disk, &md.root, rb) {
+		fsize += get_path_free_size(disk->path, used);
+	}
+	sd_rw_unlock(&md.lock);
+
+	return fsize + *used;
+}
+
+uint32_t md_nr_disks(void)
+{
+	return nr_online_disks();
+}
diff --git a/sheep/store/plain_store.c b/sheep/store/plain_store.c
new file mode 100644
index 0000000..4c19832
--- /dev/null
+++ b/sheep/store/plain_store.c
@@ -0,0 +1,759 @@
+/*
+ * Copyright (C) 2012 Nippon Telegraph and Telephone Corporation.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License version
+ * 2 as published by the Free Software Foundation.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <libgen.h>
+#include <linux/falloc.h>
+
+#include "sheep_priv.h"
+
+#ifndef FALLOC_FL_PUNCH_HOLE
+#define FALLOC_FL_PUNCH_HOLE 0x02
+#endif
+
+#define sector_algined(x) ({ ((x) & (SECTOR_SIZE - 1)) == 0; })
+
+static inline bool iocb_is_aligned(const struct siocb *iocb)
+{
+	return  sector_algined(iocb->offset) && sector_algined(iocb->length);
+}
+
+static int prepare_iocb(uint64_t oid, const struct siocb *iocb, bool create)
+{
+	int syncflag = create ? O_SYNC : O_DSYNC;
+	int flags = syncflag | O_RDWR;
+
+	if (uatomic_is_true(&sys->use_journal) || sys->nosync == true)
+		flags &= ~syncflag;
+
+	if (sys->backend_dio && iocb_is_aligned(iocb)) {
+		if (!is_aligned_to_pagesize(iocb->buf))
+			panic("Memory isn't aligned to pagesize %p", iocb->buf);
+		flags |= O_DIRECT;
+	}
+
+	if (create)
+		flags |= O_CREAT | O_EXCL;
+
+	return flags;
+}
+
+static int get_store_path(uint64_t oid, uint8_t ec_index, char *path)
+{
+	if (is_erasure_oid(oid)) {
+		if (unlikely(ec_index >= SD_MAX_COPIES))
+			panic("invalid ec_index %d", ec_index);
+		return snprintf(path, PATH_MAX, "%s/%016"PRIx64"_%d",
+				md_get_object_dir(oid), oid, ec_index);
+	}
+
+	return snprintf(path, PATH_MAX, "%s/%016" PRIx64,
+			md_get_object_dir(oid), oid);
+}
+
+static int get_store_tmp_path(uint64_t oid, uint8_t ec_index, char *path)
+{
+	if (is_erasure_oid(oid)) {
+		if (unlikely(ec_index >= SD_MAX_COPIES))
+			panic("invalid ec_index %d", ec_index);
+		return snprintf(path, PATH_MAX, "%s/%016"PRIx64"_%d.tmp",
+				md_get_object_dir(oid), oid, ec_index);
+	}
+
+	return snprintf(path, PATH_MAX, "%s/%016" PRIx64".tmp",
+			md_get_object_dir(oid), oid);
+}
+
+static int get_store_stale_path(uint64_t oid, uint32_t epoch, uint8_t ec_index,
+				char *path)
+{
+	return md_get_stale_path(oid, epoch, ec_index, path);
+}
+
+/*
+ * Check if oid is in this nodes (if oid is in the wrong place, it will be moved
+ * to the correct one after this call in a MD setup.
+ */
+bool default_exist(uint64_t oid, uint8_t ec_index)
+{
+	char path[PATH_MAX];
+
+	get_store_path(oid, ec_index, path);
+
+	return md_exist(oid, ec_index, path);
+}
+
+static int err_to_sderr(const char *path, uint64_t oid, int err)
+{
+	struct stat s;
+	char p[PATH_MAX], *dir;
+
+	/* Use a temporary buffer since dirname() may modify its argument. */
+	pstrcpy(p, sizeof(p), path);
+	dir = dirname(p);
+
+	sd_debug("%s", path);
+	switch (err) {
+	case ENOENT:
+		if (stat(dir, &s) < 0) {
+			sd_err("%s corrupted", dir);
+			return md_handle_eio(dir);
+		}
+		sd_debug("object %016" PRIx64 " not found locally", oid);
+		return SD_RES_NO_OBJ;
+	case ENOSPC:
+		/* TODO: stop automatic recovery */
+		sd_err("diskfull, oid=%"PRIx64, oid);
+		return SD_RES_NO_SPACE;
+	case EMFILE:
+	case ENFILE:
+	case EINTR:
+	case EAGAIN:
+	case EEXIST:
+		sd_err("%m, oid=%"PRIx64, oid);
+		/* make gateway try again */
+		return SD_RES_NETWORK_ERROR;
+	default:
+		sd_err("oid=%"PRIx64", %m", oid);
+		return md_handle_eio(dir);
+	}
+}
+
+static int discard(int fd, uint64_t start, uint32_t end)
+{
+	int ret = xfallocate(fd, FALLOC_FL_KEEP_SIZE | FALLOC_FL_PUNCH_HOLE,
+			     start, end - start);
+	if (ret < 0) {
+		if (errno == ENOSYS || errno == EOPNOTSUPP)
+			sd_info("FALLOC_FL_PUNCH_HOLE is not supported "
+				"on this filesystem");
+		else
+			sd_err("failed to discard object, %m");
+	}
+
+	return ret;
+}
+
+/* Trim zero blocks of the beginning and end of the object. */
+static int default_trim(int fd, uint64_t oid, const struct siocb *iocb,
+			uint64_t *poffset, uint32_t *plen)
+{
+	trim_zero_blocks(iocb->buf, poffset, plen);
+
+	if (iocb->offset < *poffset) {
+		sd_debug("discard between %d, %ld, %" PRIx64, iocb->offset,
+			 *poffset, oid);
+
+		if (discard(fd, iocb->offset, *poffset) < 0)
+			return -1;
+	}
+
+	if (*poffset + *plen < iocb->offset + iocb->length) {
+		uint64_t end = iocb->offset + iocb->length;
+		uint32_t object_size = get_vdi_object_size(oid_to_vid(oid));
+		if (end == get_objsize(oid, object_size))
+			/* This is necessary to punch the last block */
+			end = round_up(end, BLOCK_SIZE);
+		sd_debug("discard between %ld, %ld, %" PRIx64, *poffset + *plen,
+			 end, oid);
+
+		if (discard(fd, *poffset + *plen, end) < 0)
+			return -1;
+	}
+
+	return 0;
+}
+
+int default_write(uint64_t oid, const struct siocb *iocb)
+{
+	int flags = prepare_iocb(oid, iocb, false), fd,
+	    ret = SD_RES_SUCCESS;
+	char path[PATH_MAX];
+	ssize_t size;
+	uint32_t len = iocb->length;
+	uint64_t offset = iocb->offset;
+	static bool trim_is_supported = true;
+
+	if (iocb->epoch < sys_epoch()) {
+		sd_debug("%"PRIu32" sys %"PRIu32, iocb->epoch, sys_epoch());
+		return SD_RES_OLD_NODE_VER;
+	}
+
+	if (uatomic_is_true(&sys->use_journal) &&
+	    unlikely(journal_write_store(oid, iocb->buf, iocb->length,
+					 iocb->offset, false))
+	    != SD_RES_SUCCESS) {
+		sd_err("turn off journaling");
+		uatomic_set_false(&sys->use_journal);
+		flags |= O_DSYNC;
+		sync();
+	}
+
+	get_store_path(oid, iocb->ec_index, path);
+
+	/*
+	 * Make sure oid is in the right place because oid might be misplaced
+	 * in a wrong place, due to 'shutdown/restart with less/more disks' or
+	 * any bugs. We need call err_to_sderr() to return EIO if disk is broken
+	 */
+	if (!default_exist(oid, iocb->ec_index))
+		return err_to_sderr(path, oid, ENOENT);
+
+	fd = open(path, flags, sd_def_fmode);
+	if (unlikely(fd < 0))
+		return err_to_sderr(path, oid, errno);
+
+	if (trim_is_supported && is_sparse_object(oid)) {
+		if (default_trim(fd, oid, iocb, &offset, &len) < 0) {
+			trim_is_supported = false;
+			offset = iocb->offset;
+			len = iocb->length;
+		}
+	}
+
+	size = xpwrite(fd, iocb->buf, len, offset);
+	if (unlikely(size != len)) {
+		sd_err("failed to write object %"PRIx64", path=%s, offset=%"
+		       PRId32", size=%"PRId32", result=%zd, %m", oid, path,
+		       iocb->offset, iocb->length, size);
+		ret = err_to_sderr(path, oid, errno);
+		goto out;
+	}
+out:
+	close(fd);
+	return ret;
+}
+
+static int make_stale_dir(const char *path)
+{
+	char p[PATH_MAX];
+
+	snprintf(p, PATH_MAX, "%s/.stale", path);
+	if (xmkdir(p, sd_def_dmode) < 0) {
+		sd_err("%s failed, %m", p);
+		return SD_RES_EIO;
+	}
+	return SD_RES_SUCCESS;
+}
+
+static int purge_dir(const char *path)
+{
+	if (purge_directory(path) < 0)
+		return SD_RES_EIO;
+
+	return SD_RES_SUCCESS;
+}
+
+static int purge_stale_dir(const char *path)
+{
+	char p[PATH_MAX];
+
+	snprintf(p, PATH_MAX, "%s/.stale", path);
+
+	if (purge_directory_async(p) < 0)
+		return SD_RES_EIO;
+
+	return SD_RES_SUCCESS;
+}
+
+int default_cleanup(void)
+{
+	int ret;
+
+	ret = for_each_obj_path(purge_stale_dir);
+	if (ret != SD_RES_SUCCESS)
+		return ret;
+
+	return SD_RES_SUCCESS;
+}
+
+static int init_vdi_state(uint64_t oid, const char *wd, uint32_t epoch)
+{
+	int ret;
+	struct sd_inode *inode = xzalloc(SD_INODE_HEADER_SIZE);
+	struct siocb iocb = {
+		.epoch = epoch,
+		.buf = inode,
+		.length = SD_INODE_HEADER_SIZE,
+	};
+
+	ret = default_read(oid, &iocb);
+	if (ret != SD_RES_SUCCESS) {
+		sd_err("failed to read inode header %" PRIx64 " %" PRId32
+		       "wat %s", oid, epoch, wd);
+		goto out;
+	}
+	add_vdi_state_unordered(oid_to_vid(oid), inode->nr_copies,
+		      vdi_is_snapshot(inode), inode->copy_policy,
+		      inode->block_size_shift, inode->parent_vdi_id);
+
+	if (inode->name[0] == '\0')
+		atomic_set_bit(oid_to_vid(oid), sys->vdi_deleted);
+
+	atomic_set_bit(oid_to_vid(oid), sys->vdi_inuse);
+
+	ret = SD_RES_SUCCESS;
+out:
+	free(inode);
+	return ret;
+}
+
+static int init_objlist_and_vdi_bitmap(uint64_t oid, const char *wd,
+				       uint32_t epoch, uint8_t ec_index,
+				       struct vnode_info *vinfo,
+				       void *arg)
+{
+	int ret;
+	objlist_cache_insert(oid);
+
+	if (is_vdi_obj(oid)) {
+		sd_debug("found the VDI object %" PRIx64" epoch %"PRIu32
+			 " at %s", oid, epoch, wd);
+		ret = init_vdi_state(oid, wd, epoch);
+		if (ret != SD_RES_SUCCESS)
+			return ret;
+	}
+	return SD_RES_SUCCESS;
+}
+
+int default_init(void)
+{
+	int ret;
+
+	sd_debug("use plain store driver");
+	ret = for_each_obj_path(make_stale_dir);
+	if (ret != SD_RES_SUCCESS)
+		return ret;
+
+	for_each_object_in_stale(init_objlist_and_vdi_bitmap, NULL);
+
+	return for_each_object_in_wd(init_objlist_and_vdi_bitmap, true, NULL);
+}
+
+static int default_read_from_path(uint64_t oid, const char *path,
+				  const struct siocb *iocb)
+{
+	int flags = prepare_iocb(oid, iocb, false), fd,
+	    ret = SD_RES_SUCCESS;
+	ssize_t size;
+
+	/*
+	 * Make sure oid is in the right place because oid might be misplaced
+	 * in a wrong place, due to 'shutdown/restart with less disks' or any
+	 * bugs. We need call err_to_sderr() to return EIO if disk is broken.
+	 *
+	 * For stale path, get_store_stale_path already does default_exist job.
+	 */
+	if (!is_stale_path(path) && !default_exist(oid, iocb->ec_index))
+		return err_to_sderr(path, oid, ENOENT);
+
+	fd = open(path, flags);
+	if (fd < 0)
+		return err_to_sderr(path, oid, errno);
+
+	size = xpread(fd, iocb->buf, iocb->length, iocb->offset);
+	if (size < 0) {
+		sd_err("failed to read object %"PRIx64", path=%s, offset=%"
+		       PRId32", size=%"PRId32", result=%zd, %m", oid, path,
+		       iocb->offset, iocb->length, size);
+		ret = err_to_sderr(path, oid, errno);
+	}
+	close(fd);
+	return ret;
+}
+
+int default_read(uint64_t oid, const struct siocb *iocb)
+{
+	int ret;
+	char path[PATH_MAX];
+
+	get_store_path(oid, iocb->ec_index, path);
+	ret = default_read_from_path(oid, path, iocb);
+
+	/*
+	 * If the request is against the older epoch, try to read from
+	 * the stale directory
+	 */
+	if (ret == SD_RES_NO_OBJ && iocb->epoch > 0 &&
+	    iocb->epoch < sys_epoch()) {
+		get_store_stale_path(oid, iocb->epoch, iocb->ec_index, path);
+		ret = default_read_from_path(oid, path, iocb);
+	}
+
+	return ret;
+}
+
+int default_create_and_write(uint64_t oid, const struct siocb *iocb)
+{
+	char path[PATH_MAX], tmp_path[PATH_MAX], *dir;
+	int flags = prepare_iocb(oid, iocb, true);
+	int ret, fd;
+	uint32_t len = iocb->length;
+	uint32_t object_size = 0;
+	size_t obj_size;
+	uint64_t offset = iocb->offset;
+
+	sd_debug("%"PRIx64, oid);
+	get_store_path(oid, iocb->ec_index, path);
+	get_store_tmp_path(oid, iocb->ec_index, tmp_path);
+
+	if (uatomic_is_true(&sys->use_journal) &&
+	    journal_write_store(oid, iocb->buf, iocb->length,
+				iocb->offset, true)
+	    != SD_RES_SUCCESS) {
+		sd_err("turn off journaling");
+		uatomic_set_false(&sys->use_journal);
+		flags |= O_SYNC;
+		sync();
+	}
+
+	fd = open(tmp_path, flags, sd_def_fmode);
+	if (fd < 0) {
+		if (errno == EEXIST) {
+			/*
+			 * This happens if node membership changes during object
+			 * creation; while gateway retries a CREATE request,
+			 * recovery process could also recover the object at the
+			 * same time.  They should try to write the same date,
+			 * so it is okay to simply return success here.
+			 */
+			sd_debug("%s exists", tmp_path);
+			return SD_RES_SUCCESS;
+		}
+
+		sd_err("failed to open %s: %m", tmp_path);
+		return err_to_sderr(path, oid, errno);
+	}
+
+	obj_size = get_store_objsize(oid);
+
+	trim_zero_blocks(iocb->buf, &offset, &len);
+
+	object_size = get_vdi_object_size(oid_to_vid(oid));
+
+	if (offset != 0 || len != get_objsize(oid, object_size)) {
+		if (is_sparse_object(oid))
+			ret = xftruncate(fd, obj_size);
+		else
+			ret = prealloc(fd, obj_size);
+		if (ret < 0) {
+			ret = err_to_sderr(path, oid, errno);
+			goto out;
+		}
+	}
+
+	ret = xpwrite(fd, iocb->buf, len, offset);
+	if (ret != len) {
+		sd_err("failed to write object. %m");
+		ret = err_to_sderr(path, oid, errno);
+		goto out;
+	}
+
+	ret = rename(tmp_path, path);
+	if (ret < 0) {
+		sd_err("failed to rename %s to %s: %m", tmp_path, path);
+		ret = err_to_sderr(path, oid, errno);
+		goto out;
+	}
+
+	close(fd);
+
+	if (uatomic_is_true(&sys->use_journal) || sys->nosync == true) {
+		objlist_cache_insert(oid);
+		return SD_RES_SUCCESS;
+	}
+
+	pstrcpy(tmp_path, sizeof(tmp_path), path);
+	dir = dirname(tmp_path);
+	fd = open(dir, O_DIRECTORY | O_RDONLY);
+	if (fd < 0) {
+		sd_err("failed to open directory %s: %m", dir);
+		return err_to_sderr(path, oid, errno);
+	}
+
+	if (fsync(fd) != 0) {
+		sd_err("failed to write directory %s: %m", dir);
+		ret = err_to_sderr(path, oid, errno);
+		close(fd);
+		if (unlink(path) != 0)
+			sd_err("failed to unlink %s: %m", path);
+		return ret;
+	}
+	close(fd);
+	objlist_cache_insert(oid);
+	return SD_RES_SUCCESS;
+
+out:
+	if (unlink(tmp_path) != 0)
+		sd_err("failed to unlink %s: %m", tmp_path);
+	close(fd);
+	return ret;
+}
+
+int default_link(uint64_t oid, uint32_t tgt_epoch)
+{
+	char path[PATH_MAX], stale_path[PATH_MAX];
+
+	sd_debug("try link %"PRIx64" from snapshot with epoch %d", oid,
+		 tgt_epoch);
+
+	snprintf(path, PATH_MAX, "%s/%016"PRIx64, md_get_object_dir(oid), oid);
+	get_store_stale_path(oid, tgt_epoch, 0, stale_path);
+
+	if (link(stale_path, path) < 0) {
+		/*
+		 * Recovery thread and main thread might try to recover the
+		 * same object and we might get EEXIST in such case.
+		 */
+		if (errno == EEXIST)
+			goto out;
+
+		sd_debug("failed to link from %s to %s, %m", stale_path, path);
+		return err_to_sderr(path, oid, errno);
+	}
+out:
+	return SD_RES_SUCCESS;
+}
+
+/*
+ * For replicated object, if any of the replica belongs to this node, we
+ * consider it not stale.
+ *
+ * For erasure coded object, since every copy is unique and if it migrates to
+ * other node(index gets changed even it has some other copy belongs to it)
+ * because of hash ring changes, we consider it stale.
+ */
+static bool oid_stale(uint64_t oid, int ec_index, struct vnode_info *vinfo)
+{
+	uint32_t i, nr_copies;
+	const struct sd_vnode *v;
+	bool ret = true;
+	const struct sd_vnode *obj_vnodes[SD_MAX_COPIES];
+
+	nr_copies = get_obj_copy_number(oid, vinfo->nr_zones);
+	oid_to_vnodes(oid, &vinfo->vroot, nr_copies, obj_vnodes);
+	for (i = 0; i < nr_copies; i++) {
+		v = obj_vnodes[i];
+		if (vnode_is_local(v)) {
+			if (ec_index < SD_MAX_COPIES) {
+				if (i == ec_index)
+					ret = false;
+			} else {
+				ret = false;
+			}
+			break;
+		}
+	}
+
+	return ret;
+}
+
+static int move_object_to_stale_dir(uint64_t oid, const char *wd,
+				    uint32_t epoch, uint8_t ec_index,
+				    struct vnode_info *vinfo, void *arg)
+{
+	char path[PATH_MAX], stale_path[PATH_MAX];
+	uint32_t tgt_epoch = *(uint32_t *)arg;
+
+	/* ec_index from md.c is reliable so we can directly use it */
+	if (ec_index < SD_MAX_COPIES) {
+		snprintf(path, PATH_MAX, "%s/%016"PRIx64"_%d",
+			 md_get_object_dir(oid), oid, ec_index);
+		snprintf(stale_path, PATH_MAX,
+			 "%s/.stale/%016"PRIx64"_%d.%"PRIu32,
+			 md_get_object_dir(oid), oid, ec_index, tgt_epoch);
+	} else {
+		snprintf(path, PATH_MAX, "%s/%016" PRIx64,
+			 md_get_object_dir(oid), oid);
+		snprintf(stale_path, PATH_MAX, "%s/.stale/%016"PRIx64".%"PRIu32,
+			 md_get_object_dir(oid), oid, tgt_epoch);
+	}
+
+	if (unlikely(rename(path, stale_path)) < 0) {
+		sd_err("failed to move stale object %" PRIX64 " to %s, %m", oid,
+		       path);
+		return SD_RES_EIO;
+	}
+
+	sd_debug("moved object %"PRIx64, oid);
+	return SD_RES_SUCCESS;
+}
+
+static int check_stale_objects(uint64_t oid, const char *wd, uint32_t epoch,
+			       uint8_t ec_index, struct vnode_info *vinfo,
+			       void *arg)
+{
+	if (oid_stale(oid, ec_index, vinfo))
+		return move_object_to_stale_dir(oid, wd, 0, ec_index,
+						NULL, arg);
+
+	return SD_RES_SUCCESS;
+}
+
+int default_update_epoch(uint32_t epoch)
+{
+	assert(epoch);
+	return for_each_object_in_wd(check_stale_objects, false, &epoch);
+}
+
+int default_format(void)
+{
+	unsigned ret;
+
+	sd_debug("try get a clean store");
+	ret = for_each_obj_path(purge_dir);
+	if (ret != SD_RES_SUCCESS)
+		return ret;
+
+	if (sys->enable_object_cache)
+		object_cache_format();
+
+	return SD_RES_SUCCESS;
+}
+
+int default_remove_object(uint64_t oid, uint8_t ec_index)
+{
+	char path[PATH_MAX];
+
+	if (uatomic_is_true(&sys->use_journal))
+		journal_remove_object(oid);
+
+	get_store_path(oid, ec_index, path);
+
+	if (unlink(path) < 0) {
+		if (errno == ENOENT)
+			return SD_RES_NO_OBJ;
+
+		sd_err("failed, %s, %m", path);
+		return SD_RES_EIO;
+	}
+
+	return SD_RES_SUCCESS;
+}
+
+#define SHA1NAME "user.obj.sha1"
+
+static int get_object_sha1(const char *path, uint8_t *sha1)
+{
+	if (getxattr(path, SHA1NAME, sha1, SHA1_DIGEST_SIZE)
+	    != SHA1_DIGEST_SIZE) {
+		if (errno == ENODATA)
+			sd_debug("sha1 is not cached yet, %s", path);
+		else
+			sd_err("fail to get xattr, %s", path);
+		return -1;
+	}
+
+	return 0;
+}
+
+static int set_object_sha1(const char *path, const uint8_t *sha1)
+{
+	int ret;
+
+	ret = setxattr(path, SHA1NAME, sha1, SHA1_DIGEST_SIZE, 0);
+	if (ret < 0)
+		sd_err("fail to set sha1, %s", path);
+
+	return ret;
+}
+
+static int get_object_path(uint64_t oid, uint32_t epoch, char *path,
+			   size_t size)
+{
+	if (default_exist(oid, 0)) {
+		snprintf(path, PATH_MAX, "%s/%016"PRIx64,
+			 md_get_object_dir(oid), oid);
+	} else {
+		get_store_stale_path(oid, epoch, 0, path);
+		if (access(path, F_OK) < 0) {
+			if (errno == ENOENT)
+				return SD_RES_NO_OBJ;
+			return SD_RES_EIO;
+		}
+
+	}
+
+	return SD_RES_SUCCESS;
+}
+
+int default_get_hash(uint64_t oid, uint32_t epoch, uint8_t *sha1)
+{
+	int ret;
+	void *buf;
+	struct siocb iocb = {};
+	uint32_t length;
+	bool is_readonly_obj = oid_is_readonly(oid);
+	char path[PATH_MAX];
+
+	ret = get_object_path(oid, epoch, path, sizeof(path));
+	if (ret != SD_RES_SUCCESS)
+		return ret;
+
+	if (is_readonly_obj) {
+		if (get_object_sha1(path, sha1) == 0) {
+			sd_debug("use cached sha1 digest %s",
+				 sha1_to_hex(sha1));
+			return SD_RES_SUCCESS;
+		}
+	}
+
+	length = get_store_objsize(oid);
+	buf = valloc(length);
+	if (buf == NULL)
+		return SD_RES_NO_MEM;
+
+	iocb.epoch = epoch;
+	iocb.buf = buf;
+	iocb.length = length;
+
+	ret = default_read_from_path(oid, path, &iocb);
+	if (ret != SD_RES_SUCCESS) {
+		free(buf);
+		return ret;
+	}
+
+	get_buffer_sha1(buf, length, sha1);
+	free(buf);
+
+	sd_debug("the message digest of %"PRIx64" at epoch %d is %s", oid,
+		 epoch, sha1_to_hex(sha1));
+
+	if (is_readonly_obj)
+		set_object_sha1(path, sha1);
+
+	return ret;
+}
+
+int default_purge_obj(void)
+{
+	uint32_t tgt_epoch = get_latest_epoch();
+
+	return for_each_object_in_wd(move_object_to_stale_dir, true,
+				     &tgt_epoch);
+}
+
+static struct store_driver plain_store = {
+	.name = "plain",
+	.init = default_init,
+	.exist = default_exist,
+	.create_and_write = default_create_and_write,
+	.write = default_write,
+	.read = default_read,
+	.link = default_link,
+	.update_epoch = default_update_epoch,
+	.cleanup = default_cleanup,
+	.format = default_format,
+	.remove_object = default_remove_object,
+	.get_hash = default_get_hash,
+	.purge_obj = default_purge_obj,
+};
+
+add_store_driver(plain_store);
-- 
1.7.1






More information about the sheepdog mailing list