Use dentry_path() to create full path to inode object

[pohmelfs.git] / fs / pohmelfs / dir.c

/*
 * Copyright (C) 2011+ Evgeniy Polyakov <zbr@ioremap.net>
 */

#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,
};