pohmelfs: call for inclusion

[ take4 updates patch to use correct dentry_path(). Al, if you feel this chunk should be killed, drop me a note and we will go further instead of running over and over the same shit, also setup git repo to pull, rebased against v3.3 commit]

I'm please to announce new and completely rewritten distributed filesystem - POHMELFS

It went a long way from parallel NFS design which lived in drivers/staging/pohmelfs for years effectively without usage case - that design was dead.

New pohmelfs uses elliptics network [1] as its storage backend, which was proved as effective distributed system. Elliptics is used in production in Yandex search company for several years now and clusters range from small (like 6 nodes in 3 datacenters to host 15 billions of small files or hundred of nodes to scale to 1 Pb used for streaming).

We start to cook up 2 small clusters for production pohmelfs testing - one of them is largest in East Europe mirror site, another one is used for internal package storage.

Pohmelfs is just a POSIX frontend to elliptics. It supports hardlinks, symlinks, data checksums, multiple copies and so on. Pohmelfs uses local cache (synced on timely basis) for all operations, and only sync (or close with sync_on_close mount option) or writeback will flush data to remote nodes. There is also background work to flush data to storage like commit in ext3
Directory objects are synced to the storage when they are created on pohmelfs node, file data is being sent from cache later.

All recovery process is handled by elliptics and is performed without pohmelfs clients every noticing that. In particular reads are always directed to the replica with the latest data (determined according to ellptics metadata checked at file open time).

Writes in pohmelfs can be configured to succeed when quorum commits it or when specified in 'successful_write_count=' number of writes return ok. This is useful when you do not care much about number of active replicas, since you know that your data is safe. Or when you create new storage and want to copy to single datacenter to save bandwidth.

But that's enough for advertisement. Here is the code.

1. Elliptics network
http://www.ioremap.net/projects/elliptics
http://www.elliptics.ru (russian support forum - we understand english and will answer pohmelfs questions too)

2. Pohmelfs
http://www.ioremap.net/projects/pohmelfs
http://www.ioremap.net/taxonomy/term/4 (development blog section)

3. POHMELFS git repo (rebased on v3.3)
http://www.ioremap.net/git/pohmelfs.git

Signed-off-by Evgeniy Polyakov <[email protected]>

fs/Kconfig +1 -0

index d621f02..d7b8308 100644
@@ -261,6 +261,7 @@ config NFS_COMMON
 source "net/sunrpc/Kconfig"
 source "fs/ceph/Kconfig"
 source "fs/cifs/Kconfig"
+source "fs/pohmelfs/Kconfig"
 source "fs/ncpfs/Kconfig"
 source "fs/coda/Kconfig"
 source "fs/afs/Kconfig"

fs/Makefile +1 -0

index 93804d4..a2a819f 100644
@@ -124,3 +124,4 @@ obj-$(CONFIG_GFS2_FS)           += gfs2/
 obj-y				+= exofs/ # Multiple modules
 obj-$(CONFIG_CEPH_FS)		+= ceph/
 obj-$(CONFIG_PSTORE)		+= pstore/
+obj-$(CONFIG_POHMELFS)		+= pohmelfs/

fs/pohmelfs/Kconfig new file +11 -0

new file mode 100644
index 0000000..6358362
@@ -0,0 +1,11 @@
+config POHMELFS
+	tristate "POHMELFS distributed filesystem"
+	depends on INET && EXPERIMENTAL
+	select CRYPTO_HASH
+	help
+	  POHMELFS is a POSIX frontend to Elliptics network
+
+	  Elliptics is a key/value storage, which by default implements
+	  distributed hash table structure.
+
+	  More information can be found at http://www.ioremap.net/projects/elliptics

fs/pohmelfs/Makefile new file +7 -0

new file mode 100644
index 0000000..f38002d
@@ -0,0 +1,7 @@
+#
+# Makefile for the linux pohmel filesystem routines.
+#
+
+obj-$(CONFIG_POHMELFS) += pohmelfs.o
+
+pohmelfs-y := dir.o file.o inode.o net.o route.o super.o trans.o symlink.o stat.o pool.o

fs/pohmelfs/Module.symvers new file +0 -0

new file mode 100644
index 0000000..e69de29

fs/pohmelfs/README new file +84 -0

new file mode 100644
index 0000000..2e42d5a
@@ -0,0 +1,84 @@
+Pohmelfs is a POSIX frontend to elliptics distributed network build on top of DHT design
+You may find more about elliptics at http://www.ioremap.net/projects/elliptics
+Or example pohmelfs raid1 configuration at http://www.ioremap.net/node/535
+
+Here I will desribe pohmelfs mount options
+
+server=addr:port:family
+Remote node to connect (family may be 2 for IPv4 and 6 for IPv6)
+You may specify multiple nodes, usually it is ok to put here only subset
+of all remote nodes in cluster, pohmelfs will automatically discover other nodes
+
+fsid=<string>
+Filesystem ID - you may have multiple filesystems in the same elliptics cluster
+This ID may be thought of as container or namespace identity
+By default it is 'pohmelfs' (without quotes)
+
+sync_timeout=<int>
+Timeout in seconds used to synchronize local cache with the storage
+In particular all pending writes will be flushed to storage.
+If you read directory, which previously was read more than 'sync_timeout' seconds,
+it will be reread from storage, otherwise it will be read from local cache.
+The same logic _will_ apply to file content, right now once read, file will not
+be reread again until cache is dropped
+
+groups=<int>:<int>:...
+You may specify group IDs to store data to.
+One may think about group ID as replica ID, i.e. if you specify groups=1:2:3,
+each write will put data into groups with IDs 1, 2 and 3
+Read will fetch data from group 1 first, then 2 and 3
+If your replicas are not in sync, read will fetch elliptics metadata first,
+determine which replica has the most recent data, and will first try to read
+that group
+
+http_compat=<int>
+Specifies whether to use hash of full path name as inode ID (512 bits, sha512 is used)
+Provided number limits number of temporal pages allocated for path traversal, i.e.
+number of parallel pathes hashed
+Having something like 5-10 is ok for common cases
+
+readcsum/noreadcsum
+Specifies whether to turn on or off remote checksumming
+Having read csums for large files may be not a very good idea, since every read
+will force server to check whole file checksum, so for multi-gigabyte files read
+of the single page may take a while (until it is already cached)
+
+successful_write_count=<num>
+If not specified, write will be considered successful only if quorum
+(number of groups above / 2 + 1) of writes succeeded. You may alter this number
+by given option.
+Please note, that if write does not succeed, error may only be detected as returned
+value from sync() or close() syscall. Also, unsuccessful write is rescheduled and
+all its pages are redirtied again to be resent in future.
+
+keepalive_idle=<int>
+Number of seconds to wait before starting to send first TCP keepalive message
+
+keepalive_cnt=<int>
+Number of TCP keepalive messages to send before closing connection
+
+keepalive_interval=<int>
+Number of seconds between TCP keepalive messages
+
+readdir_allocation=<int>
+Number of pages allocated in one kmalloc() call when reading directory content from server
+Please note that higher-order allocations may fail, but low-ordered (like 1 or 2 pages)
+ends up in slow directory read for large directories.
+It may take up to couple of seconds to read directory with several thousands of entries,
+but usually because VFS will call ->lookup() method to every directory entry
+
+sync_on_close
+Forces flushing inode (and its data) to disk when file is closed
+
+connection_pool_size=<int>
+Number of simultaneous connections to every remote node. Connections are selected
+in round-robin fashion, but 1/4 of them (or at least one) are reserved for small-sized requests,
+which usually carry metadata messages like directory listing or file lookup requests.
+Messing them with bulk IO requests is always a bad idea.
+
+read_wait_timeout=<int>/write_wait_timeout=<int>
+Maximum number of milliseconds to wait for appropriate request to complete.
+By default both are equal to 5 seconds, which is not always a good idea especially for huge
+readahead, big cache writeback intervals and/or rather slow disks.
+These timeouts are used not only for IO requests, but also for metadata commands like
+directory listing or object lookup.

fs/pohmelfs/dir.c new file +1124 -0

new file mode 100644
index 0000000..67b61ac
@@ -0,0 +1,1124 @@
+/*
+ * Copyright (C) 2011+ Evgeniy Polyakov <[email protected]>
+ */
+
+#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
+
+#include <linux/fs.h>
+#include <linux/dcache.h>
+#include <linux/quotaops.h>
+
+#include "pohmelfs.h"
+
+#define POHMELFS_LOOKUP_SCRIPT				"pohmelfs_lookup.py"
+#define POHMELFS_UNLINK_SCRIPT				"pohmelfs_unlink.py"
+#define POHMELFS_DATA_UNLINK_SCRIPT			"pohmelfs_data_unlink.py"
+#define POHMELFS_HARDLINK_SCRIPT			"pohmelfs_hardlink.py"
+#define POHMELFS_RENAME_SCRIPT				"pohmelfs_rename.py"
+#define POHMELFS_INODE_INFO_SCRIPT_INSERT		"pohmelfs_inode_info_insert.py"
+#define POHMELFS_READDIR_SCRIPT				"pohmelfs_readdir.py"
+#define POHMELFS_DENTRY_NAME_SCRIPT			"pohmelfs_dentry_name="
+
+static void pohmelfs_init_local(struct pohmelfs_inode *pi, struct inode *dir)
+{
+	struct inode *inode = &pi->vfs_inode;
+
+	inode_init_owner(inode, dir, inode->i_mode);
+	pi->local = 1;
+
+	mark_inode_dirty(inode);
+}
+
+static int pohmelfs_send_dentry_complete(struct pohmelfs_trans *t, struct pohmelfs_state *recv)
+{
+	struct pohmelfs_inode *pi = pohmelfs_inode(t->inode);
+	struct pohmelfs_wait *wait = t->priv;
+	struct dnet_cmd *cmd = &recv->cmd;
+	unsigned long long trans = cmd->trans & ~DNET_TRANS_REPLY;
+
+	if (cmd->flags & DNET_FLAGS_MORE) {
+		if (cmd->status == 0 && cmd->size != sizeof(struct dnet_attr) + 2)
+			cmd->status = -EINVAL;
+
+		pr_debug("%s: %llu, cmd_size: %llu, flags: %x, status: %d\n",
+			 pohmelfs_dump_id(pi->id.id), trans, cmd->size,
+			 cmd->flags, cmd->status);
+
+		if (!cmd->status)
+			wait->condition = 1;
+		else
+			wait->condition = cmd->status;
+		wake_up(&wait->wq);
+	}
+
+	return 0;
+}
+
+static int pohmelfs_send_inode_info_init(struct pohmelfs_trans *t)
+{
+	struct pohmelfs_wait *wait = t->priv;
+
+	pohmelfs_wait_get(wait);
+	return 0;
+}
+
+static void pohmelfs_send_inode_info_destroy(struct pohmelfs_trans *t)
+{
+	struct pohmelfs_wait *wait = t->priv;
+
+	if (!wait->condition)
+		wait->condition = 1;
+	wake_up(&wait->wq);
+	pohmelfs_wait_put(wait);
+}
+
+static int pohmelfs_lookup_complete(struct pohmelfs_trans *t, struct pohmelfs_state *recv)
+{
+	struct pohmelfs_inode *parent = pohmelfs_inode(t->inode);
+	struct pohmelfs_wait *wait = t->priv;
+	struct dnet_cmd *cmd = &recv->cmd;
+	unsigned long long trans = cmd->trans & ~DNET_TRANS_REPLY;
+	int err = cmd->status;
+
+	if (err)
+		goto err_out_exit;
+
+	if (cmd->flags & DNET_FLAGS_MORE) {
+		struct pohmelfs_sb *psb = pohmelfs_sb(t->inode->i_sb);
+		struct pohmelfs_inode_info *info;
+		struct pohmelfs_inode *pi;
+
+		if (cmd->size != sizeof(struct dnet_attr) + sizeof(struct pohmelfs_inode_info)) {
+			err = -ENOENT;
+			goto err_out_exit;
+		}
+
+		pr_debug("%s: %llu, size: %llu, min size: %zu, flags: %x, status: %d\n",
+			 pohmelfs_dump_id(parent->id.id), trans, cmd->size,
+			 sizeof(struct dnet_attr) + sizeof(struct pohmelfs_inode_info),
+			 cmd->flags, cmd->status);
+
+
+		info = t->recv_data + sizeof(struct dnet_attr);
+		pohmelfs_convert_inode_info(info);
+
+		pi = pohmelfs_existing_inode(psb, info);
+		if (IS_ERR(pi)) {
+			err = PTR_ERR(pi);
+
+			if (err != -EEXIST)
+				goto err_out_exit;
+
+			err = 0;
+			pi = pohmelfs_sb_inode_lookup(psb, &info->id);
+			if (!pi) {
+				err = -ENOENT;
+				goto err_out_exit;
+			}
+
+			pohmelfs_fill_inode(&pi->vfs_inode, info);
+		}
+
+		wait->ret = pi;
+	}
+
+err_out_exit:
+	if (err)
+		wait->condition = err;
+	else
+		wait->condition = 1;
+	wake_up(&wait->wq);
+
+	return 0;
+}
+
+int pohmelfs_send_script_request(struct pohmelfs_inode *parent, struct pohmelfs_script_req *req)
+{
+	struct pohmelfs_sb *psb = pohmelfs_sb(parent->vfs_inode.i_sb);
+	struct pohmelfs_wait *wait;
+	struct pohmelfs_io *pio;
+	struct dnet_exec *e;
+	int script_len;
+	long ret;
+	int err;
+
+	/* 2 commas, \n and 0-byte, which is accounted in sizeof(string) */
+	script_len = sizeof(POHMELFS_DENTRY_NAME_SCRIPT) + req->obj_len + 3;
+
+	wait = pohmelfs_wait_alloc(parent);
+	if (!wait) {
+		err = -ENOMEM;
+		goto err_out_exit;
+	}
+
+	pio = kmem_cache_zalloc(pohmelfs_io_cache, GFP_NOIO);
+	if (!pio) {
+		err = -ENOMEM;
+		goto err_out_wait_put;
+	}
+
+	e = kmalloc(sizeof(struct dnet_exec) + req->script_namelen + script_len + req->binary_size, GFP_NOIO);
+	if (!e) {
+		err = -ENOMEM;
+		goto err_out_free_pio;
+	}
+
+	memset(e, 0, sizeof(struct dnet_exec));
+
+	snprintf(e->data, req->script_namelen + script_len, "%s%s'%s'\n", req->script_name, POHMELFS_DENTRY_NAME_SCRIPT, req->obj_name);
+	script_len--; /* do not include last 0-byte in the script */
+
+	memcpy(e->data + req->script_namelen + script_len, req->binary, req->binary_size);
+
+	e->type = DNET_EXEC_PYTHON_SCRIPT_NAME;
+	e->name_size = req->script_namelen;
+	e->script_size = script_len;
+	e->binary_size = req->binary_size;
+	dnet_convert_exec(e);
+
+	pio->pi = parent;
+	pio->id = req->id;
+	pio->group_id = req->group_id;
+	pio->cflags = DNET_FLAGS_NEED_ACK | req->cflags;
+
+	pio->cmd = DNET_CMD_EXEC;
+	pio->size = sizeof(struct dnet_exec) + req->script_namelen + script_len + req->binary_size;
+	pio->data = e;
+	pio->priv = wait;
+	pio->cb.init = pohmelfs_send_inode_info_init;
+	pio->cb.destroy = pohmelfs_send_inode_info_destroy;
+	pio->cb.complete = req->complete;
+
+	if (pio->group_id) {
+		err = pohmelfs_send_buf_single(pio, NULL);
+	} else {
+		err = pohmelfs_send_buf(pio);
+	}
+	if (err)
+		goto err_out_free;
+
+	{
+		int len = 6;
+		char parent_id_str[len*2+1];
+
+		pr_debug("SENT: %.*s: %s: inode->id: %s, ino: %lu, object: %s, binary size: %d, ret: %p, condition: %d\n",
+			 req->script_namelen, req->script_name,
+			 pohmelfs_dump_id(req->id->id),
+			 pohmelfs_dump_id_len_raw(parent->id.id, len,
+						  parent_id_str),
+			 parent->vfs_inode.i_ino, req->obj_name,
+			 req->binary_size, req->ret, req->ret_cond);
+	}
+
+	if (req->sync) {
+		ret = wait_event_interruptible_timeout(wait->wq, wait->condition != 0, msecs_to_jiffies(psb->read_wait_timeout));
+		if (ret <= 0) {
+			err = ret;
+			if (ret == 0)
+				err = -ETIMEDOUT;
+			goto err_out_free;
+		}
+
+		if (wait->condition < 0)
+			err = wait->condition;
+
+		req->ret = wait->ret;
+		req->ret_cond = wait->condition;
+	}
+
+err_out_free:
+	kfree(e);
+err_out_free_pio:
+	kmem_cache_free(pohmelfs_io_cache, pio);
+err_out_wait_put:
+	pohmelfs_wait_put(wait);
+err_out_exit:
+	{
+		int len = 6;
+		char parent_id_str[len*2+1];
+
+		pr_debug("DONE: %.*s: %s: inode->id: %s, ino: %lu, object: %s, binary size: %d, ret: %p, condition: %d, err: %d\n",
+			 req->script_namelen, req->script_name,
+			 pohmelfs_dump_id(req->id->id),
+			 pohmelfs_dump_id_len_raw(parent->id.id, len,
+						  parent_id_str),
+			 parent->vfs_inode.i_ino, req->obj_name,
+			 req->binary_size, req->ret, req->ret_cond, err);
+	}
+	return err;
+}
+
+int pohmelfs_send_dentry(struct pohmelfs_inode *pi, struct dnet_raw_id *id, const char *sname, int len, int sync)
+{
+	struct pohmelfs_script_req req;
+	struct pohmelfs_dentry *pd;
+	int err;
+
+	if (!len) {
+		err = -EINVAL;
+		goto err_out_exit;
+	}
+
+	pd = kmem_cache_alloc(pohmelfs_dentry_cache, GFP_NOIO);
+	if (!pd) {
+		err = -ENOMEM;
+		goto err_out_exit;
+	}
+
+	pd->parent_id = *id;
+	pd->disk.id = pi->id;
+	pd->disk.ino = cpu_to_le64(pi->vfs_inode.i_ino);
+	pd->disk.type = (pi->vfs_inode.i_mode >> 12) & 15;
+	pd->disk.len = len;
+
+	memset(&req, 0, sizeof(struct pohmelfs_script_req));
+
+	req.id = id;
+
+	req.script_name = POHMELFS_INODE_INFO_SCRIPT_INSERT;
+	req.script_namelen = sizeof(POHMELFS_INODE_INFO_SCRIPT_INSERT) - 1; /* not including 0-byte */
+
+	req.obj_name = (char *)sname;
+	req.obj_len = len;
+
+	req.binary = pd;
+	req.binary_size = sizeof(struct pohmelfs_dentry);
+
+	req.group_id = 0;
+	req.id = id;
+
+	req.sync = sync;
+	req.complete = pohmelfs_send_dentry_complete;
+
+	err = pohmelfs_send_script_request(pi, &req);
+	if (err)
+		goto err_out_free;
+
+err_out_free:
+	kmem_cache_free(pohmelfs_dentry_cache, pd);
+err_out_exit:
+	return err;
+}
+
+static int pohmelfs_create(struct inode *dir, struct dentry *dentry, umode_t mode,
+		struct nameidata *nd)
+{
+	struct pohmelfs_sb *psb = pohmelfs_sb(dir->i_sb);
+	struct pohmelfs_inode *pi;
+	int err;
+
+	inode_inc_link_count(dir);
+
+	pi = pohmelfs_new_inode(psb, mode);
+	if (IS_ERR(pi)) {
+		err = PTR_ERR(pi);
+		goto err_out_exit;
+	}
+	pohmelfs_init_local(pi, dir);
+	mark_inode_dirty(dir);
+
+	/*
+	 * calling d_instantiate() implies that
+	 * ->lookup() used d_splice_alias() with NULL inode
+	 *  when it failed to find requested object
+	 */
+	d_instantiate(dentry, &pi->vfs_inode);
+	if (psb->http_compat)
+		pohmelfs_http_compat_id(pi);
+
+	err = pohmelfs_send_dentry(pi, &pohmelfs_inode(dir)->id, dentry->d_name.name, dentry->d_name.len, 1);
+	if (err)
+		goto err_out_exit;
+
+	pr_debug("%s: ino: %lu, parent dir: %lu, object: %s\n",
+		 pohmelfs_dump_id(pi->id.id), pi->vfs_inode.i_ino,
+		 dir->i_ino, dentry->d_name.name);
+
+	return 0;
+
+err_out_exit:
+	inode_dec_link_count(dir);
+	return err;
+}
+
+static struct pohmelfs_inode *pohmelfs_lookup_group(struct inode *dir, struct dentry *dentry, int group_id)
+{
+	struct pohmelfs_inode *parent = pohmelfs_inode(dir);
+	struct pohmelfs_script_req req;
+	struct pohmelfs_inode *pi;
+	int err;
+
+	memset(&req, 0, sizeof(struct pohmelfs_script_req));
+
+	req.script_name = POHMELFS_LOOKUP_SCRIPT;
+	req.script_namelen = sizeof(POHMELFS_LOOKUP_SCRIPT) - 1; /* not including 0-byte */
+
+	req.obj_name = (char *)dentry->d_name.name;
+	req.obj_len = dentry->d_name.len;
+
+	req.binary = &parent->id;
+	req.binary_size = sizeof(struct dnet_raw_id);
+
+	req.id = &parent->id;
+	req.complete = pohmelfs_lookup_complete;
+
+	req.group_id = group_id;
+	req.sync = 1;
+	req.cflags = 0;
+
+	err = pohmelfs_send_script_request(parent, &req);
+	if (err)
+		goto err_out_exit;
+
+	pi = req.ret;
+	if (!pi) {
+		err = -ENOENT;
+		goto err_out_exit;
+	}
+
+	return pi;
+
+err_out_exit:
+	pr_debug("%s: group: %d: parent ino: %lu, name: %s: %d\n",
+		 pohmelfs_dump_id(parent->id.id), group_id,
+		 parent->vfs_inode.i_ino, dentry->d_name.name, err);
+	return ERR_PTR(err);
+}
+
+static struct dentry *pohmelfs_lookup(struct inode *dir, struct dentry *dentry, struct nameidata *nd)
+{
+	struct pohmelfs_sb *psb = pohmelfs_sb(dir->i_sb);
+	struct inode *inode = NULL;
+	struct pohmelfs_inode *pi;
+	int i, err = -ENOENT;
+
+	for (i = 0; i < psb->group_num; ++i) {
+		pi = pohmelfs_lookup_group(dir, dentry, psb->groups[i]);
+		if (IS_ERR(pi)) {
+			err = PTR_ERR(pi);
+			continue;
+		}
+
+		inode = &pi->vfs_inode;
+		err = 0;
+		break;
+	}
+
+	return d_splice_alias(inode, dentry);
+}
+
+static int pohmelfs_mkdir(struct inode *dir, struct dentry *dentry, umode_t mode)
+{
+	struct pohmelfs_sb *psb = pohmelfs_sb(dir->i_sb);
+	struct pohmelfs_inode *pi;
+	int err;
+
+	inode_inc_link_count(dir);
+
+	pi = pohmelfs_new_inode(psb, mode | S_IFDIR);
+	if (IS_ERR(pi)) {
+		err = PTR_ERR(pi);
+		goto err_out_dir;
+	}
+	pohmelfs_init_local(pi, dir);
+	mark_inode_dirty(dir);
+
+	d_instantiate(dentry, &pi->vfs_inode);
+	if (psb->http_compat)
+		pohmelfs_http_compat_id(pi);
+
+	err = pohmelfs_send_dentry(pi, &pohmelfs_inode(dir)->id, dentry->d_name.name, dentry->d_name.len, 1);
+	if (err)
+		goto err_out_dir;
+
+	pr_debug("%s: ino: %lu, parent dir: %lu, object: %s, refcnt: %d\n",
+		 pohmelfs_dump_id(pi->id.id), pi->vfs_inode.i_ino,
+		 dir->i_ino, dentry->d_name.name, dentry->d_count);
+	return 0;
+
+err_out_dir:
+	inode_dec_link_count(dir);
+	return err;
+}
+
+static int pohmelfs_unlink(struct inode *dir, struct dentry *dentry)
+{
+	struct pohmelfs_inode *parent = pohmelfs_inode(dir);
+	struct inode *inode = dentry->d_inode;
+	struct pohmelfs_inode *pi = pohmelfs_inode(inode);
+	struct pohmelfs_script_req req;
+	int err;
+
+	inode->i_ctime = dir->i_ctime;
+	mark_inode_dirty(dir);
+
+	memset(&req, 0, sizeof(struct pohmelfs_script_req));
+
+	req.script_name = POHMELFS_UNLINK_SCRIPT;
+	req.script_namelen = sizeof(POHMELFS_UNLINK_SCRIPT) - 1; /* not including 0-byte */
+
+	req.obj_name = (char *)dentry->d_name.name;
+	req.obj_len = dentry->d_name.len;
+
+	req.binary = &parent->id;
+	req.binary_size = sizeof(struct dnet_raw_id);
+
+	req.group_id = 0;
+	req.id = &parent->id;
+	req.complete = pohmelfs_send_dentry_complete;
+
+	req.sync = 1;
+
+	err = pohmelfs_send_script_request(parent, &req);
+	if (err)
+		return err;
+
+	req.script_name = POHMELFS_DATA_UNLINK_SCRIPT;
+	req.script_namelen = sizeof(POHMELFS_DATA_UNLINK_SCRIPT) - 1; /* not including 0-byte */
+
+	req.binary = &pi->id;
+	req.binary_size = sizeof(struct dnet_raw_id);
+
+	return pohmelfs_send_script_request(parent, &req);
+}
+
+static int pohmelfs_rmdir(struct inode *dir, struct dentry *dentry)
+{
+	return pohmelfs_unlink(dir, dentry);
+}
+
+struct pohmelfs_rename_req {
+	struct dnet_raw_id		old_dir_id;
+
+	struct pohmelfs_dentry		dentry;
+} __attribute__ ((packed));
+
+static int pohmelfs_rename(struct inode *old_dir, struct dentry *old_dentry,
+			struct inode *new_dir, struct dentry *new_dentry)
+{
+	struct pohmelfs_inode *old_parent = pohmelfs_inode(old_dir);
+	struct inode *inode = old_dentry->d_inode;
+	struct inode *new_inode = new_dentry->d_inode;
+	struct pohmelfs_inode *pi = pohmelfs_inode(inode);
+	struct pohmelfs_script_req req;
+	struct pohmelfs_rename_req *r;
+	int size = sizeof(struct pohmelfs_rename_req) + new_dentry->d_name.len;
+	int err;
+
+	pr_debug("%s: rename: %.*s -> %.*s: mtime: %ld\n",
+		 pohmelfs_dump_id(pi->id.id),
+		 old_dentry->d_name.len, old_dentry->d_name.name,
+		 new_dentry->d_name.len, new_dentry->d_name.name,
+		 inode->i_mtime.tv_sec);
+
+	if (pohmelfs_sb(inode->i_sb)->http_compat) {
+		err = -ENOTSUPP;
+		goto err_out_exit;
+	}
+
+	r = kzalloc(size, GFP_NOIO);
+	if (!r) {
+		err = -ENOMEM;
+		goto err_out_exit;
+	}
+
+	r->old_dir_id = pohmelfs_inode(old_dir)->id;
+	r->dentry.parent_id = pohmelfs_inode(new_dir)->id;
+	r->dentry.disk.id = pohmelfs_inode(inode)->id;
+	r->dentry.disk.ino = cpu_to_le64(inode->i_ino);
+	r->dentry.disk.type = (inode->i_mode >> 12) & 15;
+	r->dentry.disk.len = new_dentry->d_name.len;
+
+	memcpy(r->dentry.disk.name, new_dentry->d_name.name, new_dentry->d_name.len);
+
+	memset(&req, 0, sizeof(struct pohmelfs_script_req));
+
+	req.script_name = POHMELFS_RENAME_SCRIPT;
+	req.script_namelen = sizeof(POHMELFS_RENAME_SCRIPT) - 1; /* not including 0-byte */
+
+	req.obj_name = (char *)old_dentry->d_name.name;
+	req.obj_len = old_dentry->d_name.len;
+
+	req.binary = r;
+	req.binary_size = size;
+
+	req.sync = 1;
+	req.group_id = 0;
+	req.id = &old_parent->id;
+	req.complete = pohmelfs_send_dentry_complete;
+
+	if (new_inode) {
+		new_inode->i_ctime = CURRENT_TIME_SEC;
+	}
+	inode->i_ctime = CURRENT_TIME_SEC;
+	mark_inode_dirty(inode);
+	mark_inode_dirty(new_dir);
+
+	err = pohmelfs_send_script_request(old_parent, &req);
+	if (err)
+		goto err_out_free;
+
+err_out_free:
+	kfree(r);
+err_out_exit:
+	return err;
+}
+
+static int pohmelfs_symlink(struct inode *dir, struct dentry *dentry, const char *symname)
+{
+	struct pohmelfs_sb *psb = pohmelfs_sb(dir->i_sb);
+	struct pohmelfs_inode *parent = pohmelfs_inode(dir);
+	struct pohmelfs_inode *pi;
+	struct inode *inode;
+	unsigned len = strlen(symname)+1;
+	int err = 0;
+
+	inode_inc_link_count(dir);
+	pi = pohmelfs_new_inode(psb, S_IFLNK | S_IRWXUGO);
+	if (IS_ERR(pi)) {
+		err = PTR_ERR(pi);
+		goto err_out_exit;
+	}
+	inode = &pi->vfs_inode;
+	pohmelfs_init_local(pi, dir);
+	mark_inode_dirty(dir);
+
+	err = page_symlink(inode, symname, len);
+	if (err)
+		goto err_out_put;
+
+	d_instantiate(dentry, inode);
+	if (psb->http_compat)
+		pohmelfs_http_compat_id(pi);
+
+	err = pohmelfs_send_dentry(pi, &parent->id, dentry->d_name.name, dentry->d_name.len, 1);
+	if (err)
+		goto err_out_exit;
+
+	return 0;
+
+err_out_put:
+	iput(inode);
+err_out_exit:
+	inode_dec_link_count(dir);
+	return err;
+}
+
+static int pohmelfs_link(struct dentry *old_dentry, struct inode *dir, struct dentry *dentry)
+{
+	struct inode *inode = old_dentry->d_inode;
+	struct pohmelfs_inode *parent = pohmelfs_inode(dir);
+	struct pohmelfs_inode *pi = pohmelfs_inode(inode);
+	struct pohmelfs_script_req req;
+	int err;
+
+	if (pohmelfs_sb(inode->i_sb)->http_compat) {
+		err = -ENOTSUPP;
+		goto err_out_exit;
+	}
+
+	dquot_initialize(dir);
+
+	inode->i_ctime = CURRENT_TIME_SEC;
+	inode_inc_link_count(inode);
+	ihold(inode);
+
+	err = pohmelfs_send_dentry(pi, &parent->id, dentry->d_name.name, dentry->d_name.len, 1);
+	if (err) {
+		goto err_out_put;
+	}
+
+	memset(&req, 0, sizeof(struct pohmelfs_script_req));
+
+	req.script_name = POHMELFS_HARDLINK_SCRIPT;
+	req.script_namelen = sizeof(POHMELFS_HARDLINK_SCRIPT) - 1; /* not including 0-byte */
+
+	req.obj_name = (char *)dentry->d_name.name;
+	req.obj_len = dentry->d_name.len;
+
+	req.binary = &pi->id;
+	req.binary_size = sizeof(struct dnet_raw_id);
+
+	req.group_id = 0;
+	req.id = &pi->id;
+	req.complete = pohmelfs_send_dentry_complete;
+
+	req.sync = 1;
+
+	err = pohmelfs_send_script_request(parent, &req);
+	if (err)
+		goto err_out_unlink;
+
+	mark_inode_dirty(dir);
+	mark_inode_dirty(inode);
+	d_instantiate(dentry, inode);
+	return 0;
+
+err_out_unlink:
+	req.binary = &parent->id;
+	req.script_name = POHMELFS_UNLINK_SCRIPT;
+	req.script_namelen = sizeof(POHMELFS_UNLINK_SCRIPT) - 1; /* not including 0-byte */
+	pohmelfs_send_script_request(parent, &req);
+err_out_put:
+	inode_dec_link_count(inode);
+	iput(inode);
+err_out_exit:
+	return err;
+}
+
+static int pohmelfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode, dev_t rdev)
+{
+	struct pohmelfs_sb *psb = pohmelfs_sb(dir->i_sb);
+	struct pohmelfs_inode *pi;
+	struct inode *inode;
+	int err;
+
+	if (!new_valid_dev(rdev))
+		return -EINVAL;
+
+	inode_inc_link_count(dir);
+	dquot_initialize(dir);
+
+	pi = pohmelfs_new_inode(psb, mode);
+	if (IS_ERR(pi)) {
+		err = PTR_ERR(pi);
+		goto err_out_exit;
+	}
+	inode = &pi->vfs_inode;
+	pohmelfs_init_local(pi, dir);
+	mark_inode_dirty(dir);
+
+	init_special_inode(inode, inode->i_mode, rdev);
+	inode->i_op = &pohmelfs_special_inode_operations;
+
+	d_instantiate(dentry, inode);
+	if (psb->http_compat)
+		pohmelfs_http_compat_id(pi);
+
+	err = pohmelfs_send_dentry(pi, &pohmelfs_inode(dir)->id, dentry->d_name.name, dentry->d_name.len, 1);
+	if (err)
+		goto err_out_exit;
+
+	return 0;
+
+err_out_exit:
+	inode_dec_link_count(dir);
+	return err;
+}
+
+const struct inode_operations pohmelfs_dir_inode_operations = {
+	.create		= pohmelfs_create,
+	.lookup 	= pohmelfs_lookup,
+	.mkdir		= pohmelfs_mkdir,
+	.unlink		= pohmelfs_unlink,
+	.rmdir		= pohmelfs_rmdir,
+	.rename		= pohmelfs_rename,
+	.symlink	= pohmelfs_symlink,
+	.link		= pohmelfs_link,
+	.mknod		= pohmelfs_mknod,
+};
+
+static int pohmelfs_readdir_complete(struct pohmelfs_trans *t, struct pohmelfs_state *recv)
+{
+	struct pohmelfs_inode *pi = pohmelfs_inode(t->inode);
+	struct pohmelfs_wait *wait = t->priv;
+	struct dnet_cmd *cmd = &recv->cmd;
+
+	pr_debug("%s: cmd size: %llu, flags: %x\n",
+		 pohmelfs_dump_id(pi->id.id), (unsigned long long)cmd->size,
+		 cmd->flags);
+
+	if (cmd->flags & DNET_FLAGS_MORE) {
+		if (cmd->size > sizeof(struct dnet_attr)) {
+			wait->ret = t->recv_data;
+			wait->condition = cmd->size;
+
+			t->recv_data = NULL;
+			wake_up(&wait->wq);
+		}
+	} else {
+		if (!wait->condition) {
+			wait->condition = cmd->status;
+			if (!wait->condition)
+				wait->condition = 1;
+		}
+	}
+
+	return 0;
+}
+
+static int pohmelfs_dentry_add(struct dentry *parent_dentry, struct pohmelfs_inode *pi, char *name, int len)
+{
+	struct inode *inode = &pi->vfs_inode;
+	struct dentry *dentry, *old;
+	struct qstr str;
+	int err = 0;
+
+	str.name = name;
+	str.len = len;
+	str.hash = full_name_hash(str.name, str.len);
+
+	dentry = d_lookup(parent_dentry, &str);
+	if (dentry) {
+		err = -EEXIST;
+
+		dput(dentry);
+		goto err_out_exit;
+	}
+	/*
+	 * if things are ok, dentry has 2 references -
+	 * one in parent dir, and another its own,
+	 * which we should drop
+	 */
+	dentry = d_alloc(parent_dentry, &str);
+	if (!dentry) {
+		err = -ENOMEM;
+		goto err_out_exit;
+	}
+
+	old = d_splice_alias(inode, dentry);
+	if (unlikely(old)) {
+		dput(dentry);
+		err = -EEXIST;
+	} else {
+		dput(dentry);
+	}
+
+err_out_exit:
+	return err;
+}
+
+static int pohmelfs_update_inode(struct dentry *parent_dentry, struct pohmelfs_inode_info *info, char *name)
+{
+	struct pohmelfs_sb *psb = pohmelfs_sb(parent_dentry->d_inode->i_sb);
+	struct pohmelfs_inode *pi;
+	struct inode *inode;
+	int err = 0;
+
+	pi = pohmelfs_sb_inode_lookup(psb, &info->id);
+	if (pi) {
+		inode = &pi->vfs_inode;
+		pohmelfs_fill_inode(inode, info);
+	} else {
+		pi = pohmelfs_existing_inode(psb, info);
+		if (IS_ERR(pi)) {
+			err = PTR_ERR(pi);
+			goto err_out_exit;
+		}
+		inode = &pi->vfs_inode;
+	}
+
+	mutex_lock(&inode->i_mutex);
+	err = pohmelfs_dentry_add(parent_dentry, pi, name, info->namelen);
+	mutex_unlock(&inode->i_mutex);
+	if (err)
+		iput(inode);
+
+err_out_exit:
+	return err;
+}
+
+struct pohmelfs_fetch_info {
+	struct dentry		*parent;
+	struct kref		refcnt;
+	int			len;
+	char			name[0];
+};
+
+static void pohmelfs_fetch_inode_info_free(struct kref *kref)
+{
+	struct pohmelfs_fetch_info *fi = container_of(kref, struct pohmelfs_fetch_info, refcnt);
+
+	dput(fi->parent);
+	kfree(fi);
+}
+
+static void pohmelfs_fetch_inode_info_destroy(struct pohmelfs_trans *t)
+{
+	struct pohmelfs_fetch_info *fi = t->priv;
+
+	kref_put(&fi->refcnt, pohmelfs_fetch_inode_info_free);
+}
+
+static int pohmelfs_fetch_inode_info_init(struct pohmelfs_trans *t)
+{
+	struct pohmelfs_fetch_info *fi = t->priv;
+
+	kref_get(&fi->refcnt);
+	return 0;
+}
+
+static int pohmelfs_fetch_inode_info_complete(struct pohmelfs_trans *t, struct pohmelfs_state *recv)
+{
+	struct pohmelfs_fetch_info *fi = t->priv;
+	struct dnet_cmd *cmd = &recv->cmd;
+	struct pohmelfs_inode_info *info;
+	int err;
+
+	if (cmd->status)
+		return 0;
+
+	if (cmd->size < sizeof(struct dnet_attr) + sizeof(struct dnet_io_attr) + sizeof(struct pohmelfs_inode_info))
+		return 0;
+
+	info = t->recv_data + sizeof(struct dnet_attr) + sizeof(struct dnet_io_attr);
+	pohmelfs_convert_inode_info(info);
+
+	info->namelen = fi->len;
+	err = pohmelfs_update_inode(fi->parent, info, fi->name);
+
+	pr_debug("%s: fetched: '%.*s': %d\n",
+		 pohmelfs_dump_id(cmd->id.id), fi->len, fi->name, err);
+	return 0;
+}
+
+static int pohmelfs_fetch_inode_info_group(struct dentry *parent, struct pohmelfs_inode *pi,
+		struct pohmelfs_dentry_disk *d, int *groups, int group_num)
+{
+	struct pohmelfs_sb *psb = pohmelfs_sb(pi->vfs_inode.i_sb);
+	struct pohmelfs_io *pio;
+	struct pohmelfs_fetch_info *fi;
+	int err, i;
+
+	pio = kmem_cache_zalloc(pohmelfs_io_cache, GFP_NOIO);
+	if (!pio) {
+		err = -ENOMEM;
+		goto err_out_exit;
+	}
+
+	fi = kmalloc(sizeof(struct pohmelfs_fetch_info) + d->len, GFP_NOIO);
+	if (!fi) {
+		err = -ENOMEM;
+		goto err_out_free;
+	}
+
+	memcpy(fi->name, d->name, d->len);
+	fi->len = d->len;
+	kref_init(&fi->refcnt);
+	fi->parent = dget(parent);
+
+	pio->pi = pi;
+	pio->id = &d->id;
+	pio->cmd = DNET_CMD_READ;
+	pio->cflags = DNET_FLAGS_NEED_ACK | DNET_FLAGS_NOLOCK;
+	if (psb->no_read_csum)
+		pio->ioflags = DNET_IO_FLAGS_NOCSUM;
+	pio->type = POHMELFS_INODE_COLUMN;
+	pio->cb.complete = pohmelfs_fetch_inode_info_complete;
+	pio->cb.init = pohmelfs_fetch_inode_info_init;
+	pio->cb.destroy = pohmelfs_fetch_inode_info_destroy;
+	pio->priv = fi;
+
+	err = -ENOENT;
+	for (i = 0; i < group_num; ++i) {
+		pio->group_id = groups[i];
+		err = pohmelfs_send_io_group(pio, groups[i]);
+		if (!err)
+			break;
+	}
+
+	kref_put(&fi->refcnt, pohmelfs_fetch_inode_info_free);
+err_out_free:
+	kmem_cache_free(pohmelfs_io_cache, pio);
+err_out_exit:
+	return err;
+}
+
+static int pohmelfs_fetch_inode_info(struct dentry *parent, struct pohmelfs_inode *pi, struct pohmelfs_dentry_disk *d)
+{
+	struct pohmelfs_sb *psb = pohmelfs_sb(pi->vfs_inode.i_sb);
+	if (pi->groups)
+		return pohmelfs_fetch_inode_info_group(parent, pi, d, pi->groups, pi->group_num);
+	else
+		return pohmelfs_fetch_inode_info_group(parent, pi, d, psb->groups, psb->group_num);
+}
+
+static int pohmelfs_readdir_process(void *data, int size, struct file *filp, void *dirent, filldir_t filldir)
+{
+	struct dentry *dentry = filp->f_path.dentry, *child;
+	struct inode *dir = dentry->d_inode;
+	void *orig_data = data;
+	int orig_size = size;
+	struct qstr str;
+	int err = 0;
+
+	while (size > 0) {
+		struct pohmelfs_dentry_disk *d = data;
+
+		if (size < sizeof(struct pohmelfs_dentry_disk)) {
+			err = -EINVAL;
+			goto err_out_exit;
+		}
+
+		if (size < d->len) {
+			err = -EINVAL;
+			goto err_out_exit;
+		}
+
+		str.name = d->name;
+		str.len = d->len;
+		str.hash = full_name_hash(str.name, str.len);
+
+		child = d_lookup(dentry, &str);
+		pr_debug("%s: child: %.*s/%.*s: %p\n",
+			 pohmelfs_dump_id(d->id.id),
+			 dentry->d_name.len, dentry->d_name.name,
+			 d->len, d->name,
+			 child);
+		if (!child) {
+			pohmelfs_fetch_inode_info(dentry, pohmelfs_inode(dir), d);
+		} else {
+			dput(child);
+		}
+
+		size -= sizeof(struct pohmelfs_dentry_disk) + d->len;
+		data += sizeof(struct pohmelfs_dentry_disk) + d->len;
+	}
+
+	data = orig_data;
+	size = orig_size;
+	while (size > 0) {
+		struct pohmelfs_dentry_disk *d = data;
+
+		err = filldir(dirent, d->name, d->len, filp->f_pos, le64_to_cpu(d->ino), d->type);
+		if (err)
+			return 0;
+
+		filp->f_pos += 1;
+		size -= sizeof(struct pohmelfs_dentry_disk) + d->len;
+		data += sizeof(struct pohmelfs_dentry_disk) + d->len;
+	}
+
+err_out_exit:
+	return err;
+}
+
+struct pohmelfs_readdir {
+	struct dnet_raw_id			id;
+	int					max_size;
+	int					fpos;
+};
+
+static void *pohmelfs_readdir_group(int group_id, struct file *filp, int *sizep)
+{
+	struct dentry *dentry = filp->f_path.dentry;
+	struct inode *dir = dentry->d_inode;
+	struct pohmelfs_sb *psb = pohmelfs_sb(dir->i_sb);
+	struct pohmelfs_inode *parent = pohmelfs_inode(dir);
+	struct pohmelfs_readdir rd;
+	struct pohmelfs_script_req req;
+	void *data;
+	int size;
+	int err;
+
+	memset(&req, 0, sizeof(struct pohmelfs_script_req));
+
+	req.script_name = POHMELFS_READDIR_SCRIPT;
+	req.script_namelen = sizeof(POHMELFS_READDIR_SCRIPT) - 1; /* not including 0-byte */
+
+	req.obj_name = (char *)dentry->d_name.name;
+	req.obj_len = dentry->d_name.len;
+
+	rd.id = parent->id;
+	rd.max_size = psb->readdir_allocation * PAGE_SIZE - sizeof(struct dnet_attr); /* cmd->size should fit one page */
+	rd.fpos = filp->f_pos - 2; /* account for . and .. */
+
+	req.binary = &rd;
+	req.binary_size = sizeof(struct pohmelfs_readdir);
+
+	req.id = &parent->id;
+	req.complete = pohmelfs_readdir_complete;
+	req.cflags = 0;
+
+	req.group_id = group_id;
+	req.sync = 1;
+
+	err = pohmelfs_send_script_request(parent, &req);
+	if (err < 0)
+		goto err_out_exit;
+
+	data = req.ret;
+	size = req.ret_cond;
+	if (!data || !size) {
+		err = -ENOENT;
+		goto err_out_exit;
+	}
+
+	*sizep = size;
+	return data;
+
+err_out_exit:
+	return ERR_PTR(err);
+}
+
+static int pohmelfs_dir_open(struct inode *dir, struct file *filp)
+{
+#if 0
+	struct pohmelfs_inode *pi = pohmelfs_inode(dir);
+
+	if (!pohmelfs_need_resync(pi))
+		return dcache_dir_open(dir, filp);
+#endif
+	filp->f_pos = 0;
+	return 0;
+}
+
+static int pohmelfs_dir_close(struct inode *inode, struct file *filp)
+{
+	if (filp->private_data)
+		return dcache_dir_close(inode, filp);
+	return 0;
+}
+
+static int pohmelfs_readdir(struct file *filp, void *dirent, filldir_t filldir)
+{
+	struct dentry *dentry = filp->f_path.dentry;
+	struct inode *dir = dentry->d_inode;
+	struct pohmelfs_inode *pi = pohmelfs_inode(dir);
+	struct pohmelfs_sb *psb = pohmelfs_sb(dir->i_sb);
+	int i, err = -ENOENT;
+
+	if (filp->private_data) {
+		return dcache_readdir(filp, dirent, filldir);
+	}
+
+	if (filp->f_pos == 0) {
+		err = filldir(dirent, ".", 1, filp->f_pos, dir->i_ino, DT_DIR);
+		if (err)
+			return err;
+		filp->f_pos++;
+	}
+
+	if (filp->f_pos == 1) {
+		err = filldir(dirent, "..", 2, filp->f_pos, parent_ino(dentry), DT_DIR);
+		if (err)
+			return err;
+		filp->f_pos++;
+	}
+
+	for (i = 0; i < psb->group_num; ++i) {
+		int size;
+		void *data;
+
+		data = pohmelfs_readdir_group(psb->groups[i], filp, &size);
+		if (IS_ERR(data)) {
+			err = PTR_ERR(data);
+			continue;
+		}
+
+		pi->update = get_seconds();
+		err = pohmelfs_readdir_process(data + sizeof(struct dnet_attr), size - sizeof(struct dnet_attr), filp, dirent, filldir);
+		kfree(data);
+
+		break;
+	}
+
+	return err;
+}
+
+const struct file_operations pohmelfs_dir_fops = {
+	.open		= pohmelfs_dir_open,
+	.release	= pohmelfs_dir_close,
+	.read		= generic_read_dir,
+	.readdir	= pohmelfs_readdir,
+};

fs/pohmelfs/file.c new file +483 -0

new file mode 100644
index 0000000..46b6812
@@ -0,0 +1,483 @@
+/*
+ * Copyright (C) 2011+ Evgeniy Polyakov <[email protected]>
+ */
+
+#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
+
+#include <linux/fs.h>
+
+#include "pohmelfs.h"
+
+#define POHMELFS_READ_LATEST_GROUPS_SCRIPT		"pohmelfs_read_latest_groups.py"
+
+static int pohmelfs_write_init(struct pohmelfs_trans *t)
+{
+	struct pohmelfs_wait *wait = t->priv;
+
+	pohmelfs_wait_get(wait);
+	return 0;
+}
+
+static void pohmelfs_write_destroy(struct pohmelfs_trans *t)
+{
+	struct pohmelfs_wait *wait = t->priv;
+
+	wake_up(&wait->wq);
+	pohmelfs_wait_put(wait);
+}
+
+static int pohmelfs_write_complete(struct pohmelfs_trans *t, struct pohmelfs_state *recv)
+{
+	struct pohmelfs_wait *wait = t->priv;
+	struct pohmelfs_inode *pi = pohmelfs_inode(t->inode);
+	struct dnet_cmd *cmd = &recv->cmd;
+	unsigned long long trans = cmd->trans & ~DNET_TRANS_REPLY;
+
+	pr_debug("%s: %llu, flags: %x, status: %d\n",
+		 pohmelfs_dump_id(pi->id.id), trans, cmd->flags, cmd->status);
+
+	if (cmd->flags & DNET_FLAGS_MORE)
+		return 0;
+
+	wait->condition = cmd->status;
+	if (!wait->condition)
+		wait->condition = 1;
+
+	return 0;
+}
+
+static int pohmelfs_send_write_metadata(struct pohmelfs_inode *pi, struct pohmelfs_io *pio, struct pohmelfs_wait *wait)
+{
+	struct pohmelfs_sb *psb = pohmelfs_sb(pi->vfs_inode.i_sb);
+	struct timespec ts = CURRENT_TIME;
+	struct dnet_meta_update *mu;
+	struct dnet_meta *m;
+	int err, size;
+	void *data;
+
+	size = sizeof(struct dnet_meta) * 4 +
+			sizeof(struct dnet_meta_check_status) +
+			sizeof(struct dnet_meta_update) +
+			psb->fsid_len +
+			psb->group_num * sizeof(int);
+
+	data = kzalloc(size, GFP_NOIO);
+	if (!data) {
+		err = -ENOMEM;
+		goto err_out_exit;
+	}
+
+	m = data;
+	m->type = DNET_META_GROUPS;
+	m->size = psb->group_num * sizeof(int);
+	memcpy(m->data, psb->groups, m->size);
+	dnet_convert_meta(m);
+
+	m = (struct dnet_meta *)(m->data + le32_to_cpu(m->size));
+	m->type = DNET_META_NAMESPACE;
+	m->size = psb->fsid_len;
+	memcpy(m->data, psb->fsid, psb->fsid_len);
+	dnet_convert_meta(m);
+
+	m = (struct dnet_meta *)(m->data + le32_to_cpu(m->size));
+	m->type = DNET_META_UPDATE;
+	m->size = sizeof(struct dnet_meta_update);
+	mu = (struct dnet_meta_update *)m->data;
+	mu->tm.tsec = ts.tv_sec;
+	mu->tm.tnsec = ts.tv_nsec;
+	dnet_convert_meta_update(mu);
+	dnet_convert_meta(m);
+
+	m = (struct dnet_meta *)(m->data + le32_to_cpu(m->size));
+	m->type = DNET_META_CHECK_STATUS;
+	m->size = sizeof(struct dnet_meta_check_status);
+	/* do not fill, it will be updated on server */
+	dnet_convert_meta(m);
+
+	pio->pi = pi;
+	pio->id = &pi->id;
+	pio->cmd = DNET_CMD_WRITE;
+	pio->ioflags = DNET_IO_FLAGS_OVERWRITE | DNET_IO_FLAGS_META;
+	pio->cflags = DNET_FLAGS_NEED_ACK;
+	pio->type = 1;
+	pio->cb.init = pohmelfs_write_init;
+	pio->cb.destroy = pohmelfs_write_destroy;
+	pio->cb.complete = pohmelfs_write_complete;
+	pio->priv = wait;
+	pio->data = data;
+	pio->size = size;
+
+	err = pohmelfs_send_io(pio);
+	if (err)
+		goto err_out_free;
+
+err_out_free:
+	kfree(data);
+err_out_exit:
+	return err;
+}
+
+static int pohmelfs_write_command_complete(struct pohmelfs_trans *t, struct pohmelfs_state *recv)
+{
+	struct dnet_cmd *cmd = &recv->cmd;
+	struct pohmelfs_write_ctl *ctl = t->wctl;
+
+	if (cmd->flags & DNET_FLAGS_MORE)
+		return 0;
+
+	if (cmd->status == 0)
+		atomic_inc(&ctl->good_writes);
+	else {
+		struct inode *inode = t->inode;
+		struct pohmelfs_inode *pi = pohmelfs_inode(inode);
+		unsigned long long size = le64_to_cpu(t->cmd.p.io.size);
+		unsigned long long offset = le64_to_cpu(t->cmd.p.io.offset);
+
+		pr_debug("%s: write failed: ino: %lu, isize: %llu, offset: %llu, size: %llu: %d\n",
+			 pohmelfs_dump_id(pi->id.id), inode->i_ino,
+			 inode->i_size, offset, size, cmd->status);
+	}
+
+	return 0;
+}
+
+static int pohmelfs_write_command_init(struct pohmelfs_trans *t)
+{
+	struct pohmelfs_write_ctl *ctl = t->wctl;
+
+	kref_get(&ctl->refcnt);
+	return 0;
+}
+
+static void pohmelfs_write_command_destroy(struct pohmelfs_trans *t)
+{
+	struct pohmelfs_write_ctl *ctl = t->wctl;
+
+	kref_put(&ctl->refcnt, pohmelfs_write_ctl_release);
+}
+
+int pohmelfs_write_command(struct pohmelfs_inode *pi, struct pohmelfs_write_ctl *ctl, loff_t offset, size_t len)
+{
+	int err;
+	struct inode *inode = &pi->vfs_inode;
+	struct pohmelfs_io *pio;
+	uint64_t prepare_size = i_size_read(&pi->vfs_inode);
+
+	pio = kmem_cache_zalloc(pohmelfs_io_cache, GFP_NOIO);
+	if (!pio) {
+		err = -ENOMEM;
+		goto err_out_exit;
+	}
+
+	pio->pi = pi;
+	pio->id = &pi->id;
+	pio->cmd = DNET_CMD_WRITE;
+	pio->offset = offset;
+	pio->size = len;
+	pio->cflags = DNET_FLAGS_NEED_ACK;
+
+	/*
+	 * We always set prepare bit, since elliptics/eblob reuses existing (previously prepared/reserved) area
+	 * But it also allows to 'miss' prepare message (for example if we sent prepare bit when node was offline)
+	 */
+	pio->ioflags = DNET_IO_FLAGS_OVERWRITE | DNET_IO_FLAGS_PLAIN_WRITE | DNET_IO_FLAGS_PREPARE;
+
+	pio->num = prepare_size;
+
+	/* commit when whole inode is written */
+	if (offset + len == prepare_size) {
+		pio->ioflags |= DNET_IO_FLAGS_COMMIT;
+	}
+
+	pio->wctl = ctl;
+	pio->priv = ctl;
+	pio->cb.complete = pohmelfs_write_command_complete;
+	pio->cb.init = pohmelfs_write_command_init;
+	pio->cb.destroy = pohmelfs_write_command_destroy;
+
+	pr_debug("%s: ino: %lu, offset: %llu, len: %zu, total size: %llu\n",
+		 pohmelfs_dump_id(pi->id.id), inode->i_ino,
+		 (unsigned long long)offset, len, inode->i_size);
+
+	err = pohmelfs_send_io(pio);
+	if (err)
+		goto err_out_free;
+
+err_out_free:
+	kmem_cache_free(pohmelfs_io_cache, pio);
+err_out_exit:
+	return err;
+}
+
+int pohmelfs_metadata_inode(struct pohmelfs_inode *pi, int sync)
+{
+	struct inode *inode = &pi->vfs_inode;
+	struct pohmelfs_sb *psb = pohmelfs_sb(inode->i_sb);
+	struct pohmelfs_io *pio;
+	struct pohmelfs_wait *wait;
+	long ret;
+	int err;
+
+	wait = pohmelfs_wait_alloc(pi);
+	if (!wait) {
+		err = -ENOMEM;
+		goto err_out_exit;
+	}
+
+	pio = kmem_cache_zalloc(pohmelfs_io_cache, GFP_NOIO);
+	if (!pio) {
+		err = -ENOMEM;
+		goto err_out_put;
+	}
+
+	err = pohmelfs_send_write_metadata(pi, pio, wait);
+	if (err)
+		goto err_out_free;
+
+	if (sync) {
+		ret = wait_event_interruptible_timeout(wait->wq,
+				wait->condition != 0 && atomic_read(&wait->refcnt.refcount) <= 2,
+				msecs_to_jiffies(psb->write_wait_timeout));
+		if (ret <= 0) {
+			err = ret;
+			if (ret == 0)
+				err = -ETIMEDOUT;
+			goto err_out_free;
+		}
+
+		if (wait->condition < 0) {
+			err = wait->condition;
+			goto err_out_free;
+		}
+	}
+
+err_out_free:
+	kmem_cache_free(pohmelfs_io_cache, pio);
+err_out_put:
+	pohmelfs_wait_put(wait);
+err_out_exit:
+	return err;
+}
+
+static long pohmelfs_fallocate(struct file *file, int mode, loff_t offset, loff_t len)
+{
+	struct inode *inode = file->f_path.dentry->d_inode;
+	struct pohmelfs_inode *pi = pohmelfs_inode(inode);
+	struct pohmelfs_io *pio;
+	int err;
+
+	if (offset + len < i_size_read(inode)) {
+		err = 0;
+		goto err_out_exit;
+	}
+
+	pio = kmem_cache_zalloc(pohmelfs_io_cache, GFP_NOIO);
+	if (!pio) {
+		err = -ENOMEM;
+		goto err_out_exit;
+	}
+
+	pio->pi = pi;
+	pio->id = &pi->id;
+	pio->cmd = DNET_CMD_WRITE;
+	pio->cflags = DNET_FLAGS_NEED_ACK;
+	pio->ioflags = DNET_IO_FLAGS_PREPARE;
+	pio->num = i_size_read(inode);
+
+	pr_debug("%s: ino: %lu, offset: %llu, len: %llu, total size: %llu\n",
+		 pohmelfs_dump_id(pi->id.id), inode->i_ino,
+		 (unsigned long long)offset, (unsigned long long)len,
+		 inode->i_size);
+
+	err = pohmelfs_send_io(pio);
+	if (err)
+		goto err_out_free;
+
+err_out_free:
+	kmem_cache_free(pohmelfs_io_cache, pio);
+err_out_exit:
+	return err;
+}
+
+struct pohmelfs_latest_ctl {
+	struct dnet_id			id;
+	uint64_t			offset;
+	uint64_t			size;
+};
+
+static int pohmelfs_read_latest_complete(struct pohmelfs_trans *t, struct pohmelfs_state *recv)
+{
+	struct pohmelfs_inode *pi = pohmelfs_inode(t->inode);
+	struct pohmelfs_wait *wait = t->priv;
+	struct dnet_cmd *cmd = &recv->cmd;
+	int err = cmd->status;
+
+	if (cmd->status)
+		goto err_out_exit;
+
+	if (cmd->flags & DNET_FLAGS_MORE) {
+		pr_debug("%s: group: %d, attr size: %lld\n",
+			 pohmelfs_dump_id(cmd->id.id), cmd->id.group_id,
+			 cmd->size - sizeof(struct dnet_attr));
+		if (cmd->size < sizeof(struct dnet_attr) + 4) {
+			err = -ENOENT;
+			goto err_out_exit;
+		}
+
+		mutex_lock(&pi->lock);
+		if (!pi->groups) {
+			pi->groups = kmalloc(cmd->size - sizeof(struct dnet_attr), GFP_NOIO);
+			if (!pi->groups) {
+				err = -ENOMEM;
+				mutex_unlock(&pi->lock);
+				goto err_out_exit;
+			}
+
+			pi->group_num = (cmd->size - sizeof(struct dnet_attr)) / sizeof(int);
+			memcpy(pi->groups, t->recv_data + sizeof(struct dnet_attr), pi->group_num * sizeof(int));
+
+			pr_debug("%s: group: %d, received: %d groups\n",
+				 pohmelfs_dump_id(cmd->id.id), cmd->id.group_id,
+				 pi->group_num);
+		}
+		mutex_unlock(&pi->lock);
+	}
+
+err_out_exit:
+	if (err)
+		wait->condition = err;
+	else
+		wait->condition = 1;
+	return 0;
+}
+
+static int pohmelfs_read_latest_group(struct pohmelfs_inode *pi, struct pohmelfs_latest_ctl *r, int group_id)
+{
+	struct pohmelfs_script_req req;
+
+	memset(&req, 0, sizeof(struct pohmelfs_script_req));
+
+	req.script_name = POHMELFS_READ_LATEST_GROUPS_SCRIPT;
+	req.script_namelen = sizeof(POHMELFS_READ_LATEST_GROUPS_SCRIPT) - 1;
+
+	req.obj_name = "noname";
+	req.obj_len = 5;
+
+	req.binary = r;
+	req.binary_size = sizeof(struct pohmelfs_latest_ctl);
+
+	req.id = &pi->id;
+	req.group_id = group_id;
+	req.sync = 1;
+	req.cflags = 0;
+	req.complete = pohmelfs_read_latest_complete;
+
+	return pohmelfs_send_script_request(pi, &req);
+}
+
+static int pohmelfs_read_latest(struct pohmelfs_inode *pi)
+{
+	struct pohmelfs_latest_ctl *r;
+	struct pohmelfs_sb *psb = pohmelfs_sb(pi->vfs_inode.i_sb);
+	int i, err = -ENOENT;
+
+	r = kzalloc(sizeof(struct pohmelfs_latest_ctl), GFP_NOIO);
+	if (!r) {
+		err = -ENOMEM;
+		goto err_out_exit;
+	}
+
+	dnet_setup_id(&r->id, 0, pi->id.id);
+
+	for (i = 0; i < psb->group_num; ++i) {
+		r->id.group_id = psb->groups[i];
+
+		err = pohmelfs_read_latest_group(pi, r, psb->groups[i]);
+		if (err)
+			continue;
+
+		break;
+	}
+
+	kfree(r);
+
+	pr_debug("%s: %d groups\n", pohmelfs_dump_id(pi->id.id), pi->group_num);
+
+err_out_exit:
+	return err;
+}
+
+static int pohmelfs_file_open(struct inode *inode, struct file *filp)
+{
+	struct pohmelfs_inode *pi = pohmelfs_inode(inode);
+
+	if (!pi->group_num && !pi->local)
+		pohmelfs_read_latest(pi);
+
+	if (pohmelfs_need_resync(pi))
+		invalidate_mapping_pages(&inode->i_data, 0, -1);
+
+	return generic_file_open(inode, filp);
+}
+
+/*
+ * We want fsync() to work on POHMELFS.
+ */
+static int pohmelfs_fsync(struct file *filp, loff_t start, loff_t end, int datasync)
+{
+	struct inode *inode = filp->f_mapping->host;
+	int err = filemap_write_and_wait_range(inode->i_mapping, start, end);
+	if (!err) {
+		mutex_lock(&inode->i_mutex);
+		err = sync_inode_metadata(inode, 1);
+		mutex_unlock(&inode->i_mutex);
+	}
+	pr_debug("%s: start: %lld, end: %lld, nrpages: %ld, dirty: %d: %d\n",
+		 pohmelfs_dump_id(pohmelfs_inode(inode)->id.id),
+		 (unsigned long long)start, (unsigned long long)end,
+		 inode->i_mapping->nrpages,
+		 mapping_cap_writeback_dirty(inode->i_mapping), err);
+	return err;
+}
+
+static int pohmelfs_flush(struct file *filp, fl_owner_t id)
+{
+	struct inode *inode = filp->f_mapping->host;
+	struct pohmelfs_sb *psb = pohmelfs_sb(inode->i_sb);
+	int err = 0;
+
+	if (psb->sync_on_close)
+		err = pohmelfs_fsync(filp, 0, ~0ULL, 1);
+
+	if (!err && test_bit(AS_EIO, &inode->i_mapping->flags))
+		err = -EIO;
+
+	pr_debug("%s: %d\n",
+		 pohmelfs_dump_id(pohmelfs_inode(inode)->id.id), err);
+	return err;
+}
+
+const struct file_operations pohmelfs_file_ops = {
+	.open		= pohmelfs_file_open,
+
+	.llseek		= generic_file_llseek,
+
+	.read		= do_sync_read,
+	.aio_read	= generic_file_aio_read,
+
+	.mmap		= generic_file_mmap,
+
+	.splice_read	= generic_file_splice_read,
+	.splice_write	= generic_file_splice_write,
+
+	.write		= do_sync_write,
+	.aio_write	= generic_file_aio_write,
+
+	.fallocate	= pohmelfs_fallocate,
+
+	.fsync		= pohmelfs_fsync,
+	.flush		= pohmelfs_flush,
+};
+
+const struct inode_operations pohmelfs_file_inode_operations = {
+};

fs/pohmelfs/inode.c new file +1084 -0

new file mode 100644
index 0000000..3f7c623
@@ -0,0 +1,1084 @@
+/*
+ * Copyright (C) 2011+ Evgeniy Polyakov <[email protected]>
+ */
+
+#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
+
+#include <linux/buffer_head.h>
+#include <linux/cred.h>
+#include <linux/fiemap.h>
+#include <linux/fs.h>
+#include <linux/fs_struct.h>
+#include <linux/mpage.h>
+#include <linux/mount.h>
+#include <linux/mm.h>
+#include <linux/namei.h>
+#include <linux/pagevec.h>
+#include <linux/pagemap.h>
+#include <linux/random.h>
+#include <linux/scatterlist.h>
+#include <linux/slab.h>
+#include <linux/time.h>
+#include <linux/writeback.h>
+
+#include "pohmelfs.h"
+
+char *pohmelfs_dump_id_len_raw(const unsigned char *id, unsigned int len, char *dst)
+{
+	unsigned int i;
+
+	if (len > SHA512_DIGEST_SIZE)
+		len = SHA512_DIGEST_SIZE;
+
+	for (i=0; i<len; ++i)
+		sprintf(&dst[2*i], "%02x", id[i]);
+	return dst;
+}
+
+#define pohmelfs_dump_len 6
+typedef struct {
+	char			id_str[pohmelfs_dump_len * 2 + 1];
+} pohmelfs_dump_t;
+static DEFINE_PER_CPU(pohmelfs_dump_t, pohmelfs_dump_per_cpu);
+
+char *pohmelfs_dump_id(const unsigned char *id)
+{
+	pohmelfs_dump_t *ptr;
+
+	ptr = &get_cpu_var(pohmelfs_dump_per_cpu);
+	pohmelfs_dump_id_len_raw(id, pohmelfs_dump_len, ptr->id_str);
+	put_cpu_var(ptr);
+
+	return ptr->id_str;
+}
+
+#define dnet_raw_id_scratch 6
+typedef struct {
+	unsigned long 			rand;
+	struct timespec			ts;
+} dnet_raw_id_scratch_t;
+static DEFINE_PER_CPU(dnet_raw_id_scratch_t, dnet_raw_id_scratch_per_cpu);
+
+static int pohmelfs_gen_id(struct pohmelfs_sb *psb, struct dnet_raw_id *id)
+{
+	dnet_raw_id_scratch_t *sc;
+	int err;
+	long rand;
+
+	get_random_bytes(&rand, sizeof(sc->rand));
+
+	sc = &get_cpu_var(dnet_raw_id_scratch_per_cpu);
+	sc->rand ^= rand;
+	sc->ts = CURRENT_TIME;
+
+	err = pohmelfs_hash(psb, sc, sizeof(dnet_raw_id_scratch_t), id);
+	put_cpu_var(sc);
+
+	return err;
+}
+
+/*
+ * Create path from root for given inode.
+ * Path is formed as set of stuctures, containing name of the object
+ * and its inode data (mode, permissions and so on).
+ */
+static int pohmelfs_construct_path_string(struct pohmelfs_inode *pi, char *data, int len)
+{
+	struct dentry *d;
+	char *ptr;
+	int err;
+
+	d = d_find_alias(&pi->vfs_inode);
+	if (!d) {
+		err = -ENOENT;
+		goto err_out_exit;
+	}
+
+	ptr = dentry_path_raw(d, data, len);
+	if (IS_ERR(ptr)) {
+		err = PTR_ERR(ptr);
+		goto err_out_put;
+	}
+
+	err = ptr - data - 1; /* not including 0-byte */
+
+	pr_debug("dname: '%s', len: %u, maxlen: %u, name: '%s', strlen: %d\n",
+		 d->d_name.name, d->d_name.len, len, data, err);
+
+err_out_put:
+	dput(d);
+err_out_exit:
+	return err;
+}
+
+int pohmelfs_http_compat_id(struct pohmelfs_inode *pi)
+{
+	struct pohmelfs_sb *psb = pohmelfs_sb(pi->vfs_inode.i_sb);
+	struct timespec ts = CURRENT_TIME;
+	int idx = ts.tv_nsec % psb->http_compat;
+	struct pohmelfs_path *p = &psb->path[idx];
+	int err;
+
+	mutex_lock(&p->lock);
+	err = pohmelfs_construct_path_string(pi, p->data, PAGE_SIZE);
+	if (err > 0) {
+		pohmelfs_hash(psb, p->data, err, &pi->id);
+	}
+	mutex_unlock(&p->lock);
+
+	return err;
+}
+
+static int pohmelfs_sb_inode_insert(struct pohmelfs_sb *psb, struct pohmelfs_inode *pi)
+{
+	struct rb_node **n = &psb->inode_root.rb_node, *parent = NULL;
+	struct pohmelfs_inode *tmp;
+	int cmp, err = 0;
+
+	spin_lock(&psb->inode_lock);
+	while (*n) {
+		parent = *n;
+
+		tmp = rb_entry(parent, struct pohmelfs_inode, node);
+
+		cmp = dnet_id_cmp_str(tmp->id.id, pi->id.id);
+		if (cmp < 0)
+			n = &parent->rb_left;
+		else if (cmp > 0)
+			n = &parent->rb_right;
+		else {
+			err = -EEXIST;
+			goto err_out_unlock;
+		}
+	}
+
+	rb_link_node(&pi->node, parent, n);
+	rb_insert_color(&pi->node, &psb->inode_root);
+
+err_out_unlock:
+	spin_unlock(&psb->inode_lock);
+
+	return err;
+}
+
+struct pohmelfs_inode *pohmelfs_sb_inode_lookup(struct pohmelfs_sb *psb, struct dnet_raw_id *id)
+{
+	struct rb_node *n = psb->inode_root.rb_node;
+	struct pohmelfs_inode *pi, *found = NULL;
+	int cmp;
+
+	spin_lock(&psb->inode_lock);
+	while (n) {
+		pi = rb_entry(n, struct pohmelfs_inode, node);
+
+		cmp = dnet_id_cmp_str(pi->id.id, id->id);
+		if (cmp < 0) {
+			n = n->rb_left;
+		} else if (cmp > 0)
+			n = n->rb_right;
+		else {
+			found = pi;
+			break;
+		}
+	}
+	if (found) {
+		if (!igrab(&found->vfs_inode))
+			found = NULL;
+	}
+	spin_unlock(&psb->inode_lock);
+
+	return found;
+}
+
+struct inode *pohmelfs_alloc_inode(struct super_block *sb)
+{
+	struct pohmelfs_inode *pi;
+
+	pi = kmem_cache_zalloc(pohmelfs_inode_cache, GFP_NOIO);
+	if (!pi)
+		goto err_out_exit;
+
+	inode_init_once(&pi->vfs_inode);
+
+	rb_init_node(&pi->node);
+	mutex_init(&pi->lock);
+
+	return &pi->vfs_inode;
+
+err_out_exit:
+	return NULL;
+}
+
+void pohmelfs_destroy_inode(struct inode *inode)
+{
+	struct pohmelfs_inode *pi = pohmelfs_inode(inode);
+
+	pr_debug("%s: ino: %ld, dirty: %lx\n",
+		 pohmelfs_dump_id(pi->id.id), inode->i_ino,
+		 inode->i_state & I_DIRTY);
+
+	kfree(pi->groups);
+	kmem_cache_free(pohmelfs_inode_cache, pi);
+}
+
+int pohmelfs_hash(struct pohmelfs_sb *psb, const void *data, const size_t size, struct dnet_raw_id *id)
+{
+	struct scatterlist sg;
+	struct hash_desc desc;
+
+	sg_init_table(&sg, 1);
+	sg_set_buf(&sg, data, size);
+
+	desc.tfm = psb->hash;
+	desc.flags = 0;
+
+	return crypto_hash_digest(&desc, &sg, size, id->id);
+}
+
+struct pohmelfs_readpages_priv {
+	struct pohmelfs_wait		wait;
+	struct kref			refcnt;
+	int				page_num, page_index;
+	struct page			*pages[0];
+};
+
+static void pohmelfs_readpages_free(struct kref *kref)
+{
+	struct pohmelfs_readpages_priv *rp = container_of(kref, struct pohmelfs_readpages_priv, refcnt);
+	struct pohmelfs_inode *pi = rp->wait.pi;
+	int i;
+
+	pr_debug("%s: read: %ld/%ld, wait: %d\n",
+		 pohmelfs_dump_id(pi->id.id), atomic_long_read(&rp->wait.count),
+		 rp->page_num * PAGE_CACHE_SIZE, rp->wait.condition);
+
+	for (i = 0; i < rp->page_num; ++i) {
+		struct page *page = rp->pages[i];
+
+		flush_dcache_page(page);
+		SetPageUptodate(page);
+		unlock_page(page);
+		page_cache_release(page);
+	}
+
+	iput(&rp->wait.pi->vfs_inode);
+	kfree(rp);
+}
+
+static void pohmelfs_readpages_destroy(struct pohmelfs_trans *t)
+{
+	struct pohmelfs_readpages_priv *rp = t->priv;
+	struct pohmelfs_wait *wait = &rp->wait;
+
+	if (!wait->condition)
+		wait->condition = 1;
+
+	wake_up(&wait->wq);
+	kref_put(&rp->refcnt, pohmelfs_readpages_free);
+}
+
+static int pohmelfs_readpages_complete(struct pohmelfs_trans *t, struct pohmelfs_state *recv)
+{
+	struct pohmelfs_readpages_priv *rp = t->priv;
+	struct pohmelfs_wait *wait = &rp->wait;
+	struct dnet_cmd *cmd = &recv->cmd;
+
+	if (!(cmd->flags & DNET_FLAGS_MORE)) {
+		if (!wait->condition) {
+			wait->condition = cmd->status;
+			if (!wait->condition)
+				wait->condition = 1;
+			wake_up(&rp->wait.wq);
+		}
+	}
+
+	pr_debug("%d:%s: read: %ld, wait: %d\n",
+		 cmd->id.group_id, pohmelfs_dump_id(wait->pi->id.id),
+		 atomic_long_read(&wait->count), wait->condition);
+
+	return 0;
+}
+
+static int pohmelfs_readpages_init(struct pohmelfs_trans *t)
+{
+	struct pohmelfs_readpages_priv *rp = t->priv;
+
+	kref_get(&rp->refcnt);
+	return 0;
+}
+
+static int pohmelfs_readpages_recv_reply(struct pohmelfs_trans *t, struct pohmelfs_state *recv)
+{
+	struct pohmelfs_readpages_priv *rp = t->priv;
+	struct pohmelfs_wait *wait = &rp->wait;
+	struct pohmelfs_inode *pi = wait->pi;
+	unsigned int asize = sizeof(struct dnet_attr) + sizeof(struct dnet_io_attr);
+	void *data = &t->cmd.attr; /* overwrite send buffer used for attr/ioattr */
+	struct dnet_cmd *cmd = &recv->cmd;
+	struct page *page;
+	pgoff_t offset;
+	int err, size;
+
+	if (t->io_offset < asize) {
+		size = asize - t->io_offset;
+		data += t->io_offset;
+		err = pohmelfs_recv(t, recv, data, size);
+		if (err < 0)
+			goto err_out_exit;
+
+		dnet_convert_io_attr(&t->cmd.p.io);
+	}
+
+	while (t->io_offset != cmd->size) {
+		offset = (t->io_offset - asize) & (PAGE_CACHE_SIZE - 1);
+		size = PAGE_CACHE_SIZE - offset;
+		page = rp->pages[rp->page_index];
+
+		if (size > cmd->size - t->io_offset)
+			size = cmd->size - t->io_offset;
+
+		data = kmap(page);
+		err = pohmelfs_recv(t, recv, data + offset, size);
+		kunmap(page);
+
+		if (err > 0 && ((err + offset == PAGE_CACHE_SIZE) || (t->io_offset == cmd->size)))  {
+			rp->page_index++;
+		}
+
+		if (err < 0)
+			goto err_out_exit;
+
+		atomic_long_add(err, &wait->count);
+	}
+
+	err = 0;
+
+err_out_exit:
+	if ((err < 0) && (err != -ENOENT) && (err != -EAGAIN))
+		pr_err("%d:%s: offset: %lld, data size: %llu, err: %d\n",
+		       cmd->id.group_id, pohmelfs_dump_id(pi->id.id),
+		       t->io_offset - asize + t->cmd.p.io.offset,
+		       (unsigned long long)cmd->size - asize, err);
+
+	return err;
+}
+
+static int pohmelfs_readpages_group(struct pohmelfs_inode *pi, struct pohmelfs_readpages_priv *rp, int group_id)
+{
+	struct pohmelfs_sb *psb = pohmelfs_sb(pi->vfs_inode.i_sb);
+	struct pohmelfs_wait *wait = &rp->wait;
+	struct pohmelfs_io *io;
+	long ret;
+	int err;
+
+	io = kmem_cache_zalloc(pohmelfs_io_cache, GFP_NOIO);
+	if (!io) {
+		err = -ENOMEM;
+		goto err_out_exit;
+	}
+
+	io->pi = pi;
+	io->id = &pi->id;
+	io->cmd = DNET_CMD_READ;
+	/*
+	 * We send read command with lock, so its will be picked by the same threads as process
+	 * bulk write commands leaving nonblocking threads free for metadata commands like
+	 * directory reading, lookup and so on
+	 */
+	//io->cflags = DNET_FLAGS_NEED_ACK | DNET_FLAGS_NOLOCK;
+	io->cflags = DNET_FLAGS_NEED_ACK;
+	io->offset = page_offset(rp->pages[0]);
+	io->size = rp->page_num * PAGE_CACHE_SIZE;
+	if (psb->no_read_csum)
+		io->ioflags = DNET_IO_FLAGS_NOCSUM;
+	io->cb.init = pohmelfs_readpages_init;
+	io->cb.complete = pohmelfs_readpages_complete;
+	io->cb.destroy = pohmelfs_readpages_destroy;
+	io->cb.recv_reply = pohmelfs_readpages_recv_reply;
+	io->priv = rp;
+
+	err = pohmelfs_send_io_group(io, group_id);
+	if (err)
+		goto err_out_free;
+
+	ret = wait_event_interruptible_timeout(wait->wq, wait->condition != 0, msecs_to_jiffies(psb->read_wait_timeout));
+	if (ret <= 0) {
+		err = ret;
+		if (ret == 0)
+			err = -ETIMEDOUT;
+		goto err_out_free;
+	}
+
+	if (wait->condition < 0) {
+		err = wait->condition;
+		goto err_out_free;
+	}
+
+	err = atomic_long_read(&wait->count);
+
+err_out_free:
+	kmem_cache_free(pohmelfs_io_cache, io);
+err_out_exit:
+	return err;
+}
+
+static int pohmelfs_readpages_groups(struct pohmelfs_inode *pi, struct pohmelfs_readpages_priv *rp,
+		int *groups, int group_num)
+{
+	int err = -ENOENT;
+	int i;
+
+	for (i = 0; i < group_num; ++i) {
+		err = pohmelfs_readpages_group(pi, rp, groups[i]);
+		if (err < 0)
+			continue;
+
+		break;
+	}
+
+	pi->update = get_seconds();
+	return err;
+}
+
+static struct pohmelfs_readpages_priv *pohmelfs_readpages_alloc(struct pohmelfs_inode *pi, int page_num)
+{
+	struct pohmelfs_readpages_priv *rp;
+	int err;
+
+	rp = kzalloc(sizeof(struct pohmelfs_readpages_priv) + page_num * sizeof(struct page *), GFP_NOIO);
+	if (!rp) {
+		err = -ENOMEM;
+		goto err_out_exit;
+	}
+
+	err = pohmelfs_wait_init(&rp->wait, pi);
+	if (err)
+		goto err_out_free;
+
+	rp->page_num = page_num;
+	kref_init(&rp->refcnt);
+	return rp;
+
+err_out_free:
+	kfree(rp);
+err_out_exit:
+	return ERR_PTR(err);
+}
+
+static int pohmelfs_readpages_send(struct pohmelfs_inode *pi, struct pohmelfs_readpages_priv *rp)
+{
+	struct pohmelfs_sb *psb = pohmelfs_sb(pi->vfs_inode.i_sb);
+	int err;
+
+	if (pi->group_num) {
+		err = pohmelfs_readpages_groups(pi, rp, pi->groups, pi->group_num);
+	} else {
+		err = pohmelfs_readpages_groups(pi, rp, psb->groups, psb->group_num);
+	}
+
+	return err;
+}
+
+static int pohmelfs_readpages_send_list(struct address_space *mapping, struct list_head *page_list, int num)
+{
+	struct inode *inode = mapping->host;
+	struct pohmelfs_inode *pi = pohmelfs_inode(inode);
+	int err = 0, i;
+	struct pohmelfs_readpages_priv *rp;
+	struct page *tmp, *page;
+
+	if (list_empty(page_list))
+		goto err_out_exit;
+
+	rp = pohmelfs_readpages_alloc(pi, num);
+	if (IS_ERR(rp)) {
+		err = PTR_ERR(rp);
+		goto err_out_exit;
+	}
+
+	i = 0;
+	list_for_each_entry_safe(page, tmp, page_list, lru) {
+		list_del(&page->lru);
+
+		if (add_to_page_cache_lru(page, mapping, page->index, GFP_KERNEL)) {
+			/* Failed - free current page, optionally send already grabbed and free others */
+			page_cache_release(page);
+			break;
+		}
+
+		rp->pages[i] = page;
+		i++;
+	}
+
+	if (i > 0) {
+		rp->page_num = i;
+		err = pohmelfs_readpages_send(pi, rp);
+
+		pr_debug("%s: ino: %lu, offset: %lu, pages: %u/%u: %d\n",
+			 pohmelfs_dump_id(pi->id.id), inode->i_ino,
+			 (long)page_offset(rp->pages[0]),
+			 rp->page_num, num, err);
+	}
+
+	kref_put(&rp->refcnt, pohmelfs_readpages_free);
+
+	/* Cleanup pages which were not added into page cache */
+	list_for_each_entry_safe(page, tmp, page_list, lru) {
+		list_del(&page->lru);
+		page_cache_release(page);
+	}
+
+err_out_exit:
+	return err;
+}
+
+static int pohmelfs_readpages(struct file *filp, struct address_space *mapping,
+			struct list_head *page_list, unsigned nr_pages)
+{
+	struct page *tmp, *page;
+	pgoff_t idx;
+	LIST_HEAD(head);
+	int err = 0, i = 0;
+
+	while (!list_empty(page_list)) {
+		page = list_entry(page_list->prev, struct page, lru);
+		idx = page->index;
+		i = 0;
+
+		INIT_LIST_HEAD(&head);
+
+		list_for_each_entry_safe_reverse(page, tmp, page_list, lru) {
+			if (idx != page->index) {
+				struct pohmelfs_inode *pi = pohmelfs_inode(mapping->host);
+				pr_debug("%s: index mismatch: want: %ld, page-index: %ld, total: %d\n",
+					 pohmelfs_dump_id(pi->id.id),
+					 (long)idx, (long)page->index,
+					 nr_pages);
+				break;
+			}
+
+			list_move_tail(&page->lru, &head);
+			i++;
+			idx++;
+		}
+
+		err = pohmelfs_readpages_send_list(mapping, &head, i);
+	}
+	if (err >= 0)
+		err = 0;
+
+	return err;
+}
+
+static int pohmelfs_readpage(struct file *file, struct page *page)
+{
+	struct inode *inode = page->mapping->host;
+	struct pohmelfs_inode *pi = pohmelfs_inode(inode);
+	struct pohmelfs_readpages_priv *rp;
+	int err;
+
+	if (inode->i_size <= page->index << PAGE_CACHE_SHIFT) {
+		SetPageUptodate(page);
+		unlock_page(page);
+		return 0;
+	}
+
+	rp = pohmelfs_readpages_alloc(pi, 1);
+	if (IS_ERR(rp)) {
+		err = PTR_ERR(rp);
+		goto err_out_exit;
+	}
+
+	rp->pages[0] = page;
+	page_cache_get(page);
+
+	err = pohmelfs_readpages_send(pi, rp);
+	if (err >= 0)
+		err = 0;
+
+	kref_put(&rp->refcnt, pohmelfs_readpages_free);
+err_out_exit:
+	if (err < 0)
+		pr_err("%s: %s: ino: %lu, offset: %lu, uptodate: %d, err: %d\n",
+		       __func__, pohmelfs_dump_id(pi->id.id), inode->i_ino,
+		       (long)page_offset(page), PageUptodate(page), err);
+
+	return err;
+}
+
+void pohmelfs_write_ctl_release(struct kref *kref)
+{
+	struct pohmelfs_write_ctl *ctl = container_of(kref, struct pohmelfs_write_ctl, refcnt);
+	struct address_space *mapping = ctl->pvec.pages[0]->mapping;
+	struct inode *inode = mapping->host;
+	struct pohmelfs_sb *psb = pohmelfs_sb(inode->i_sb);
+	int bad_write = atomic_read(&ctl->good_writes) < psb->group_num / 2 + 1;
+	struct page *page;
+	unsigned int i;
+
+	if (psb->successful_write_count && (atomic_read(&ctl->good_writes) >= psb->successful_write_count))
+		bad_write = 0;
+
+	if (bad_write) {
+		struct pohmelfs_inode *pi = pohmelfs_inode(inode);
+		unsigned long long offset = page_offset(ctl->pvec.pages[0]);
+
+		pr_debug("%s: bad write: ino: %lu, isize: %llu, offset: %llu: writes: %d/%d\n",
+			 pohmelfs_dump_id(pi->id.id),
+			 inode->i_ino, inode->i_size, offset,
+			 atomic_read(&ctl->good_writes), psb->group_num);
+		mapping_set_error(mapping, -EIO);
+	}
+
+	for (i = 0; i < pagevec_count(&ctl->pvec); ++i) {
+		page = ctl->pvec.pages[i];
+
+		if (PageLocked(page)) {
+			end_page_writeback(page);
+
+			if (bad_write) {
+				SetPageError(page);
+				ClearPageUptodate(page);
+				/*
+				 * Do not reschedule failed write page again
+				 * This may explode systems with large caches
+				 * when there is no connection to elliptics cluster
+				 */
+				//set_page_dirty(page);
+			}
+			unlock_page(page);
+		}
+	}
+
+	pagevec_release(&ctl->pvec);
+	kmem_cache_free(pohmelfs_write_cache, ctl);
+}
+
+static int pohmelfs_writepages_chunk(struct pohmelfs_inode *pi, struct pohmelfs_write_ctl *ctl,
+		struct writeback_control *wbc, struct address_space *mapping)
+{
+	struct inode *inode = &pi->vfs_inode;
+	uint64_t offset, size;
+	unsigned i;
+	int err = 0, good = 0;
+
+	offset = page_offset(ctl->pvec.pages[0]);
+
+	size = 0;
+	/* we will lookup them again when doing actual send */
+	for (i = 0; i< pagevec_count(&ctl->pvec); ++i) {
+		struct page *page = ctl->pvec.pages[i];
+
+		lock_page(page);
+#if 1
+		if (unlikely(page->mapping != mapping)) {
+continue_unlock:
+			unlock_page(page);
+			continue;
+		}
+
+		if (wbc->sync_mode != WB_SYNC_NONE)
+			wait_on_page_writeback(page);
+		if (PageWriteback(page)) {
+			unlock_page(page);
+			break;
+		}
+
+		if (!PageDirty(page))
+			goto continue_unlock;
+
+		if (!clear_page_dirty_for_io(page))
+			goto continue_unlock;
+#else
+		clear_page_dirty_for_io(page);
+#endif
+
+		set_page_writeback(page);
+
+		good++;
+		size += PAGE_CACHE_SIZE;
+		wbc->nr_to_write--;
+	}
+
+	if (good != 0) {
+		size = pagevec_count(&ctl->pvec) * PAGE_CACHE_SIZE;
+		if (offset + size > inode->i_size)
+			size = inode->i_size - offset;
+
+		err = pohmelfs_write_command(pi, ctl, offset, size);
+		if (err)
+			goto err_out_exit;
+	}
+
+err_out_exit:
+	kref_put(&ctl->refcnt, pohmelfs_write_ctl_release);
+	return err;
+}
+
+static int pohmelfs_writepages_send(struct address_space *mapping, struct writeback_control *wbc, struct pagevec *pvec, int start, int end)
+{
+	struct inode *inode = mapping->host;
+	struct pohmelfs_inode *pi = pohmelfs_inode(inode);
+	struct pohmelfs_write_ctl *ctl;
+	int err, i;
+
+	ctl = kmem_cache_zalloc(pohmelfs_write_cache, GFP_NOIO);
+	if (!ctl) {
+		err = -ENOMEM;
+		goto err_out_exit;
+	}
+
+	kref_init(&ctl->refcnt);
+	atomic_set(&ctl->good_writes, 0);
+
+	for (i = start; i < end; ++i)
+		pagevec_add(&ctl->pvec, pvec->pages[i]);
+
+	err = pohmelfs_writepages_chunk(pi, ctl, wbc, mapping);
+	if (err)
+		goto err_out_exit;
+
+err_out_exit:
+	return err;
+}
+
+static int pohmelfs_writepages(struct address_space *mapping, struct writeback_control *wbc)
+{
+	struct inode *inode = mapping->host;
+	struct pohmelfs_inode *pi = pohmelfs_inode(inode);
+	pgoff_t index, start, end /* inclusive */, idx;
+	int done = 0;
+	int range_whole = 0;
+	int should_loop = 1;
+	int nr_pages, err = 0, i, start_idx;
+	struct pagevec pvec;
+	int written = 0;
+
+	index = wbc->range_start >> PAGE_CACHE_SHIFT;
+	end = wbc->range_end >> PAGE_CACHE_SHIFT;
+
+	pr_debug("%s: ino: %ld, nr: %ld, index: %llu, end: %llu, total_size: %lu, sync: %d\n",
+		 pohmelfs_dump_id(pohmelfs_inode(inode)->id.id), inode->i_ino,
+		 wbc->nr_to_write, wbc->range_start, wbc->range_end,
+		 (unsigned long)inode->i_size, wbc->sync_mode);
+
+	if (wbc->range_cyclic) {
+		start = mapping->writeback_index; /* Start from prev offset */
+		end = -1;
+	} else {
+		start = wbc->range_start >> PAGE_CACHE_SHIFT;
+		end = wbc->range_end >> PAGE_CACHE_SHIFT;
+		if (wbc->range_start == 0 && wbc->range_end == LLONG_MAX)
+			range_whole = 1;
+		should_loop = 0;
+	}
+	index = start;
+
+retry:
+	while (!done && index <= end) {
+		nr_pages = pagevec_lookup_tag(&pvec, mapping, &index, PAGECACHE_TAG_DIRTY,
+			      min(end - index, (pgoff_t)PAGEVEC_SIZE-1) + 1);
+		if (!nr_pages) {
+			err = 0;
+			break;
+		}
+
+		idx = pvec.pages[0]->index;
+		for (start_idx = 0, i = 0; i< nr_pages; ++i) {
+			struct page *page = pvec.pages[i];
+
+			/* non-contiguous pages detected */
+			if (idx != page->index) {
+				err = pohmelfs_writepages_send(mapping, wbc, &pvec, start_idx, i);
+				if (err)
+					goto err_out_exit;
+				start_idx = i;
+			}
+
+			idx++;
+		}
+
+		err = pohmelfs_writepages_send(mapping, wbc, &pvec, start_idx, nr_pages);
+		if (err)
+			goto err_out_exit;
+
+		if (wbc->nr_to_write <= 0)
+			done = 1;
+
+		written += nr_pages;
+	}
+
+	if (should_loop && !done) {
+		/* more to do; loop back to beginning of file */
+		should_loop = 0;
+		index = 0;
+		goto retry;
+	}
+
+	if (wbc->range_cyclic || (range_whole && wbc->nr_to_write > 0))
+		mapping->writeback_index = index;
+
+	if (written) {
+		err = pohmelfs_metadata_inode(pi, wbc->sync_mode != WB_SYNC_NONE);
+		if (err)
+			goto err_out_exit;
+	}
+
+
+	if (test_and_clear_bit(AS_EIO, &mapping->flags))
+		err = -EIO;
+err_out_exit:
+	pr_debug("%s: metadata write complete: %d\n",
+		 pohmelfs_dump_id(pi->id.id), err);
+	return err;
+}
+
+static const struct address_space_operations pohmelfs_aops = {
+	.write_begin		= simple_write_begin,
+	.write_end		= simple_write_end,
+	.writepages		= pohmelfs_writepages,
+	.readpage		= pohmelfs_readpage,
+	.readpages		= pohmelfs_readpages,
+	.set_page_dirty 	= __set_page_dirty_nobuffers,
+};
+
+void pohmelfs_convert_inode_info(struct pohmelfs_inode_info *info)
+{
+	info->ino = cpu_to_le64(info->ino);
+	info->mode = cpu_to_le64(info->mode);
+	info->nlink = cpu_to_le64(info->nlink);
+	info->uid = cpu_to_le32(info->uid);
+	info->gid = cpu_to_le32(info->gid);
+	info->namelen = cpu_to_le32(info->namelen);
+	info->blocks = cpu_to_le64(info->blocks);
+	info->rdev = cpu_to_le64(info->rdev);
+	info->size = cpu_to_le64(info->size);
+	info->version = cpu_to_le64(info->version);
+	info->blocksize = cpu_to_le64(info->blocksize);
+	info->flags = cpu_to_le64(info->flags);
+
+	dnet_convert_time(&info->ctime);
+	dnet_convert_time(&info->mtime);
+	dnet_convert_time(&info->atime);
+}
+
+void pohmelfs_fill_inode_info(struct inode *inode, struct pohmelfs_inode_info *info)
+{
+	struct pohmelfs_inode *pi = pohmelfs_inode(inode);
+
+	memcpy(info->id.id, pi->id.id, DNET_ID_SIZE);
+
+	info->ino = inode->i_ino;
+	info->mode = inode->i_mode;
+	info->nlink = inode->i_nlink;
+	info->uid = inode->i_uid;
+	info->gid = inode->i_gid;
+	info->blocks = inode->i_blocks;
+	info->rdev = inode->i_rdev;
+	info->size = inode->i_size;
+	info->version = inode->i_version;
+	info->blocksize = 1 << inode->i_blkbits;
+
+	info->ctime.tsec = inode->i_ctime.tv_sec;
+	info->ctime.tnsec = inode->i_ctime.tv_nsec;
+
+	info->mtime.tsec = inode->i_mtime.tv_sec;
+	info->mtime.tnsec = inode->i_mtime.tv_nsec;
+
+	info->atime.tsec = inode->i_atime.tv_sec;
+	info->atime.tnsec = inode->i_atime.tv_nsec;
+
+	info->flags = 0;
+}
+
+void pohmelfs_fill_inode(struct inode *inode, struct pohmelfs_inode_info *info)
+{
+	pr_debug("%s: ino: %lu inode is regular: %d, dir: %d, link: %d, mode: %o, "
+		 "namelen: %u, size: %llu, state: %lx, mtime: %llu.%llu/%lu.%lu\n",
+		 pohmelfs_dump_id(info->id.id), inode->i_ino,
+		 S_ISREG(inode->i_mode), S_ISDIR(inode->i_mode),
+		 S_ISLNK(inode->i_mode), inode->i_mode, info->namelen,
+		 inode->i_size, inode->i_state,
+		 (unsigned long long)info->mtime.tsec,
+		 (unsigned long long)info->mtime.tnsec,
+		 inode->i_mtime.tv_sec, inode->i_mtime.tv_nsec);
+
+	if (info->mtime.tsec < inode->i_mtime.tv_sec)
+		return;
+	if ((info->mtime.tsec == inode->i_mtime.tv_sec) &&
+			(info->mtime.tnsec < inode->i_mtime.tv_nsec))
+		return;
+
+	pohmelfs_inode(inode)->id = info->id;
+
+	inode->i_mode = info->mode;
+	set_nlink(inode, info->nlink);
+	inode->i_uid = info->uid;
+	inode->i_gid = info->gid;
+	inode->i_blocks = info->blocks;
+	inode->i_rdev = info->rdev;
+	inode->i_size = info->size;
+	inode->i_version = info->version;
+	inode->i_blkbits = ffs(info->blocksize);
+
+	inode->i_mtime = pohmelfs_date(&info->mtime);
+	inode->i_atime = pohmelfs_date(&info->atime);
+	inode->i_ctime = pohmelfs_date(&info->ctime);
+}
+
+static void pohmelfs_inode_info_current(struct pohmelfs_sb *psb, struct pohmelfs_inode_info *info)
+{
+	struct timespec ts = CURRENT_TIME;
+	struct dnet_time dtime;
+
+	info->nlink = S_ISDIR(info->mode) ? 2 : 1;
+	info->uid = current_fsuid();
+	info->gid = current_fsgid();
+	info->size = 0;
+	info->blocksize = PAGE_SIZE;
+	info->blocks = 0;
+	info->rdev = 0;
+	info->version = 0;
+
+	dtime.tsec = ts.tv_sec;
+	dtime.tnsec = ts.tv_nsec;
+
+	info->ctime = dtime;
+	info->mtime = dtime;
+	info->atime = dtime;
+
+	pohmelfs_gen_id(psb, &info->id);
+}
+
+const struct inode_operations pohmelfs_special_inode_operations = {
+	.setattr			= simple_setattr,
+};
+
+struct pohmelfs_inode *pohmelfs_existing_inode(struct pohmelfs_sb *psb, struct pohmelfs_inode_info *info)
+{
+	struct pohmelfs_inode *pi;
+	struct inode *inode;
+	int err;
+
+	inode = iget_locked(psb->sb, atomic_long_inc_return(&psb->ino));
+	if (!inode) {
+		err = -ENOMEM;
+		goto err_out_exit;
+	}
+
+	pi = pohmelfs_inode(inode);
+
+	if (inode->i_state & I_NEW) {
+		pohmelfs_fill_inode(inode, info);
+		/*
+		 * i_mapping is a pointer to i_data during inode initialization.
+		 */
+		inode->i_data.a_ops = &pohmelfs_aops;
+
+		if (S_ISREG(inode->i_mode)) {
+			inode->i_fop = &pohmelfs_file_ops;
+			inode->i_op = &pohmelfs_file_inode_operations;
+		} else if (S_ISDIR(inode->i_mode)) {
+			inode->i_fop = &pohmelfs_dir_fops;
+			inode->i_op = &pohmelfs_dir_inode_operations;
+		} else if (S_ISLNK(inode->i_mode)) {
+			inode->i_op = &pohmelfs_symlink_inode_operations;
+			inode->i_mapping->a_ops = &pohmelfs_aops;
+		} else {
+			inode->i_op = &pohmelfs_special_inode_operations;
+		}
+
+		err = pohmelfs_sb_inode_insert(psb, pi);
+		if (err)
+			goto err_out_put;
+
+		unlock_new_inode(inode);
+	}
+
+	return pi;
+
+err_out_put:
+	unlock_new_inode(inode);
+	iput(inode);
+err_out_exit:
+	return ERR_PTR(err);
+}
+
+struct pohmelfs_inode *pohmelfs_new_inode(struct pohmelfs_sb *psb, int mode)
+{
+	struct pohmelfs_inode *pi;
+	struct pohmelfs_inode_info *info;
+	int err;
+
+	info = kmem_cache_zalloc(pohmelfs_inode_info_cache, GFP_NOIO);
+	if (!info) {
+		err = -ENOMEM;
+		goto err_out_exit;
+	}
+
+	info->mode = mode;
+
+	pohmelfs_inode_info_current(psb, info);
+
+	pi = pohmelfs_existing_inode(psb, info);
+	if (IS_ERR(pi)) {
+		err = PTR_ERR(pi);
+		goto err_out_free;
+	}
+
+	kmem_cache_free(pohmelfs_inode_info_cache, info);
+	return pi;
+
+err_out_free:
+	kmem_cache_free(pohmelfs_inode_info_cache, info);
+err_out_exit:
+	return ERR_PTR(err);
+}
+
+int pohmelfs_wait_init(struct pohmelfs_wait *wait, struct pohmelfs_inode *pi)
+{
+	if (!igrab(&pi->vfs_inode))
+		return -EINVAL;
+
+	wait->pi = pi;
+
+	atomic_long_set(&wait->count, 0);
+	init_waitqueue_head(&wait->wq);
+	kref_init(&wait->refcnt);
+
+	return 0;
+}
+
+struct pohmelfs_wait *pohmelfs_wait_alloc(struct pohmelfs_inode *pi)
+{
+	struct pohmelfs_wait *wait;
+
+	wait = kmem_cache_zalloc(pohmelfs_wait_cache, GFP_NOIO);
+	if (!wait) {
+		goto err_out_exit;
+	}
+
+	if (pohmelfs_wait_init(wait, pi))
+		goto err_out_free;
+
+	return wait;
+
+err_out_free:
+	kmem_cache_free(pohmelfs_wait_cache, wait);
+err_out_exit:
+	return NULL;
+}
+
+static void pohmelfs_wait_free(struct kref *kref)
+{
+	struct pohmelfs_wait *wait = container_of(kref, struct pohmelfs_wait, refcnt);
+	struct inode *inode = &wait->pi->vfs_inode;
+
+	iput(inode);
+	kmem_cache_free(pohmelfs_wait_cache, wait);
+}
+
+void pohmelfs_wait_put(struct pohmelfs_wait *wait)
+{
+	kref_put(&wait->refcnt, pohmelfs_wait_free);
+}

fs/pohmelfs/net.c new file +703 -0

new file mode 100644
index 0000000..7907939
@@ -0,0 +1,703 @@
+/*
+ * Copyright (C) 2011+ Evgeniy Polyakov <[email protected]>
+ */
+
+#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
+
+#include <linux/in.h>
+#include <linux/in6.h>
+#include <linux/net.h>
+
+#include <net/sock.h>
+#include <net/tcp.h>
+
+#include "pohmelfs.h"
+
+void *pohmelfs_scratch_buf;
+int pohmelfs_scratch_buf_size = 4096;
+
+void pohmelfs_print_addr(struct sockaddr_storage *addr, const char *fmt, ...)
+{
+	struct sockaddr *sa = (struct sockaddr *)addr;
+	struct va_format vaf;
+	va_list args;
+
+	va_start(args, fmt);
+	vaf.fmt = fmt;
+	vaf.va = &args;
+
+	if (sa->sa_family == AF_INET) {
+		struct sockaddr_in *sin = (struct sockaddr_in *)addr;
+		pr_info("%pI4:%d: %pV",
+			&sin->sin_addr.s_addr, ntohs(sin->sin_port), &vaf);
+	} else if (sa->sa_family == AF_INET6) {
+		struct sockaddr_in6 *sin = (struct sockaddr_in6 *)addr;
+		pr_info("%pI6:%d: %pV",
+			&sin->sin6_addr, ntohs(sin->sin6_port), &vaf);
+	}
+
+	va_end(args);
+}
+
+/*
+ * Basic network sending/receiving functions.
+ * Blocked mode is used.
+ */
+int pohmelfs_data_recv(struct pohmelfs_state *st, void *buf, u64 size, unsigned int flags)
+{
+	struct msghdr msg;
+	struct kvec iov;
+	int err;
+
+	BUG_ON(!size);
+
+	iov.iov_base = buf;
+	iov.iov_len = size;
+
+	msg.msg_iov = (struct iovec *)&iov;
+	msg.msg_iovlen = 1;
+	msg.msg_name = NULL;
+	msg.msg_namelen = 0;
+	msg.msg_control = NULL;
+	msg.msg_controllen = 0;
+	msg.msg_flags = flags;
+
+	err = kernel_recvmsg(st->sock, &msg, &iov, 1, iov.iov_len, msg.msg_flags);
+	if (err < 0)
+		goto err_out_exit;
+
+err_out_exit:
+	return err;
+}
+
+int pohmelfs_recv(struct pohmelfs_trans *t, struct pohmelfs_state *recv, void *data, int size)
+{
+	int err;
+
+	err = pohmelfs_data_recv(recv, data, size, MSG_DONTWAIT);
+	if (err < 0)
+		return err;
+
+	t->io_offset += err;
+	return err;
+}
+
+static int pohmelfs_data_send(struct pohmelfs_trans *t)
+{
+	struct msghdr msg;
+	struct iovec io;
+	int err;
+
+	msg.msg_name = NULL;
+	msg.msg_namelen = 0;
+	msg.msg_control = NULL;
+	msg.msg_controllen = 0;
+	msg.msg_flags = MSG_DONTWAIT;
+
+	msg.msg_iov = &io;
+	msg.msg_iovlen = 1;
+
+
+	if (t->io_offset < t->header_size) {
+		io.iov_base = (void *)(&t->cmd) + t->io_offset;
+		io.iov_len = t->header_size - t->io_offset;
+
+		err = kernel_sendmsg(t->st->sock, &msg, (struct kvec *)msg.msg_iov, 1, io.iov_len);
+		if (err < 0) {
+			if (err == 0)
+				err = -ECONNRESET;
+			goto err_out_exit;
+		}
+
+		t->io_offset += err;
+	}
+
+	if ((t->io_offset >= t->header_size) && t->data) {
+		size_t sent_size = t->io_offset - t->header_size;
+		io.iov_base = t->data + sent_size;
+		io.iov_len = t->data_size - sent_size;
+
+		err = kernel_sendmsg(t->st->sock, &msg, (struct kvec *)msg.msg_iov, 1, io.iov_len);
+		if (err < 0) {
+			if (err == 0)
+				err = -ECONNRESET;
+			goto err_out_exit;
+		}
+
+		t->io_offset += err;
+	}
+
+
+	err = 0;
+
+err_out_exit:
+	return err;
+}
+
+static int pohmelfs_page_send(struct pohmelfs_trans *t)
+{
+	struct pohmelfs_write_ctl *ctl = t->wctl;
+	struct msghdr msg;
+	struct iovec io;
+	unsigned i;
+	int err = -EINVAL;
+
+	if (t->io_offset < t->header_size) {
+		io.iov_base = (void *)(&t->cmd) + t->io_offset;
+		io.iov_len = t->header_size - t->io_offset;
+
+		msg.msg_name = NULL;
+		msg.msg_namelen = 0;
+		msg.msg_control = NULL;
+		msg.msg_controllen = 0;
+		msg.msg_flags = MSG_DONTWAIT;
+
+		msg.msg_iov = &io;
+		msg.msg_iovlen = 1;
+
+		err = kernel_sendmsg(t->st->sock, &msg, (struct kvec *)msg.msg_iov, 1, io.iov_len);
+		if (err < 0) {
+			if (err == 0)
+				err = -ECONNRESET;
+			goto err_out_exit;
+		}
+
+		t->io_offset += err;
+	}
+
+	if (t->io_offset >= t->header_size) {
+		size_t skip_offset = 0;
+		size_t size = le64_to_cpu(t->cmd.cmd.size) + sizeof(struct dnet_cmd) - t->io_offset;
+		size_t current_io_offset = t->io_offset - t->header_size;
+
+		for (i = 0; i < pagevec_count(&ctl->pvec); ++i) {
+			struct page *page = ctl->pvec.pages[i];
+			size_t sz = PAGE_CACHE_SIZE;
+
+			if (sz > size)
+				sz = size;
+
+			if (current_io_offset > skip_offset + sz) {
+				skip_offset += sz;
+				continue;
+			}
+
+			sz -= current_io_offset - skip_offset;
+
+			err = kernel_sendpage(t->st->sock, page, current_io_offset - skip_offset, sz, MSG_DONTWAIT);
+
+			pr_debug("%s: %d/%d: total-size: %llu, io-offset: %llu, rest-size: %zd, current-io: %zd, skip-offset: %zd, sz: %zu: %d\n",
+				 pohmelfs_dump_id(pohmelfs_inode(t->inode)->id.id),
+				 i, pagevec_count(&ctl->pvec),
+				 (unsigned long long)le64_to_cpu(t->cmd.cmd.size) + sizeof(struct dnet_cmd),
+				 t->io_offset, size, current_io_offset,
+				 skip_offset, sz, err);
+
+			if (err <= 0) {
[… more lines in the raw message]

Lines of this file are not shown. View the raw message.

fs/pohmelfs/packet.h new file +752 -0

new file mode 100644
index 0000000..f432987

Lines of this file are not shown. View the raw message.

fs/pohmelfs/pohmelfs.h new file +503 -0

new file mode 100644
index 0000000..3b30a59

Lines of this file are not shown. View the raw message.

fs/pohmelfs/pool.c new file +159 -0

new file mode 100644
index 0000000..c4572c7

Lines of this file are not shown. View the raw message.

fs/pohmelfs/route.c new file +369 -0

new file mode 100644
index 0000000..c6a4755

Lines of this file are not shown. View the raw message.

fs/pohmelfs/stat.c new file +141 -0

new file mode 100644
index 0000000..1001760

Lines of this file are not shown. View the raw message.

fs/pohmelfs/super.c new file +982 -0

new file mode 100644
index 0000000..c719937

Lines of this file are not shown. View the raw message.

fs/pohmelfs/symlink.c new file +13 -0

new file mode 100644
index 0000000..80a9d87

Lines of this file are not shown. View the raw message.

fs/pohmelfs/trans.c new file +432 -0

new file mode 100644
index 0000000..8fb9bf1

Lines of this file are not shown. View the raw message.

The rest of this diff is not shown. View the raw message.

Signature · 2 lines

-- 
	Evgeniy Polyakov