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

Liu Yuan namei.unix at gmail.com
Sat Mar 21 12:59:01 CET 2015


On Fri, Mar 20, 2015 at 06:39:02PM +0900, Saeki Masaki wrote:
> This change is a preparation patch for add store_driver.
> Put files together to the new folder.
> 
> 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 87ab759..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/əʌo7/!"
> -
> -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/!"

Why NONE_EXIST_PATH is changed after rename? I guess something wrong with your
editor.

Thanks,
Yuan



More information about the sheepdog mailing list