Use column 3 to store inode information instead of saving it in directory. This fixes...

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

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

#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_inode_dirty(struct pohmelfs_inode *parent, struct pohmelfs_inode *pi)
{
        struct inode *inode = &pi->vfs_inode;
        struct inode *dir = &parent->vfs_inode;

        pi->parent_id = parent->id;
        inode_init_owner(inode, dir, inode->i_mode);

        inode->i_mtime = inode->i_ctime = CURRENT_TIME;
        dir->i_mtime = CURRENT_TIME;

        mark_inode_dirty(inode);
        mark_inode_dirty(dir);
}

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("pohmelfs: %s: pohmelfs_send_dentry_complete: %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;

        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("pohmelfs: %s: pohmelfs_lookup_complete: %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);
                }

                pi->parent_id = parent->id;
                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;
        if (req->complete == pohmelfs_lookup_complete)
                pio->cflags |= DNET_FLAGS_NOLOCK;

        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;

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

        {
                int len = 6;
                char parent_id_str[len*2+1];

                pr_debug("pohmelfs: %.*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);
        }

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:
        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 = le64_to_cpu(pi->vfs_inode.i_ino);
        pd->disk.type = (pi->vfs_inode.i_mode >> 12) & 15;
        pd->disk.len = len;

        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, int mode,
                struct nameidata *nd)
{
        struct pohmelfs_sb *psb = pohmelfs_sb(dir->i_sb);
        struct pohmelfs_inode *parent = pohmelfs_inode(dir);
        struct pohmelfs_inode *pi;
        int err;

        pi = pohmelfs_new_inode(psb, mode);
        if (IS_ERR(pi)) {
                err = PTR_ERR(pi);
                goto err_out_exit;
        }

        inode_inc_link_count(dir);
        pohmelfs_inode_dirty(parent, pi);

        err = pohmelfs_send_dentry(pi, &pi->parent_id, dentry->d_name.name, dentry->d_name.len, 0);
        if (err)
                goto err_out_put;

        pr_debug("pohmelfs: create: %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);

        /*
         * 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);

        return 0;

err_out_put:
        inode_dec_link_count(dir);
        iput(&pi->vfs_inode);
err_out_exit:
        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;

        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;

        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("pohmelfs: pohmelfs_lookup_group: %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;
        }

        if (err && (err != -ENOENT) && (err != -EOPNOTSUPP))
                return ERR_PTR(err);

        return d_splice_alias(inode, dentry);
}

static int pohmelfs_mkdir(struct inode *dir, struct dentry *dentry, int mode)
{
        struct pohmelfs_sb *psb = pohmelfs_sb(dir->i_sb);
        struct pohmelfs_inode *parent = pohmelfs_inode(dir);
        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_inode_dirty(parent, pi);

        d_instantiate(dentry, &pi->vfs_inode);
        pr_debug("pohmelfs: mkdir: %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 pohmelfs_inode *pi = pohmelfs_inode(dentry->d_inode);
        struct pohmelfs_script_req req;
        int err;

        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 = 0;

        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 dnet_raw_id              new_dir_id;
        int                             new_len;
        char                            new_name[0];
} __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 pohmelfs_script_req req;
        struct pohmelfs_rename_req *r;
        int size = sizeof(struct pohmelfs_rename_req) + new_dentry->d_name.len;
        int err;

        r = kmalloc(size, GFP_NOIO);
        if (!r) {
                err = -ENOMEM;
                goto err_out_exit;
        }

        r->old_dir_id = pohmelfs_inode(old_dir)->id;
        r->new_dir_id = pohmelfs_inode(new_dir)->id;
        r->new_len = cpu_to_le32(new_dentry->d_name.len);
        memcpy(r->new_name, new_dentry->d_name.name, new_dentry->d_name.len);

        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 = 0;
        req.group_id = 0;
        req.id = &old_parent->id;
        req.complete = pohmelfs_send_dentry_complete;

        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 *pi;
        struct inode *inode;
        unsigned len = strlen(symname)+1;
        int err = 0;

        pi = pohmelfs_new_inode(psb, S_IFLNK | S_IRWXUGO);
        if (IS_ERR(pi)) {
                err = PTR_ERR(pi);
                goto err_out_exit;
        }

        pohmelfs_inode_dirty(pohmelfs_inode(dir), pi);
        inode = &pi->vfs_inode;

        err = page_symlink(inode, symname, len);
        if (err)
                goto err_out_put;

        d_instantiate(dentry, inode);

        return 0;

err_out_put:
        iput(inode);
err_out_exit:
        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;

        dquot_initialize(dir);

        inode->i_ctime = CURRENT_TIME_SEC;
        inode_inc_link_count(inode);
        ihold(inode);

        pohmelfs_inode_dirty(parent, pi);

        err = pohmelfs_send_dentry(pi, &pi->parent_id, dentry->d_name.name, dentry->d_name.len, 1);
        if (err) {
                goto err_out_put;
        }

        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 = 0;

        err = pohmelfs_send_script_request(parent, &req);
        if (err)
                goto err_out_unlink;

        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);
        return err;
}

static int pohmelfs_mknod(struct inode *dir, struct dentry *dentry, int 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;

        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;

        init_special_inode(inode, inode->i_mode, rdev);
        inode->i_op = &pohmelfs_special_inode_operations;

        pohmelfs_inode_dirty(pohmelfs_inode(dir), pi);

        d_instantiate(dentry, inode);
        return 0;

err_out_exit:
        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,
};

struct pohmelfs_readdir_header {
        char                            magic[8];
        unsigned short                  version;
        unsigned short                  chunk_size;
        unsigned int                    chunk_num;
} __attribute__((packed));

static void pohmelfs_convert_readdir_header(struct pohmelfs_readdir_header *h)
{
        h->version = dnet_bswap16(h->version);
        h->chunk_size = dnet_bswap16(h->chunk_size);
        h->chunk_num = dnet_bswap32(h->chunk_num);
}

struct pohmelfs_readdir_chunk_header {
        unsigned short                  length;
        unsigned short                  num;
        unsigned short                  key_size;
        unsigned short                  payload_size;
} __attribute__((packed));

static void pohmelfs_convert_readdir_chunk_header(struct pohmelfs_readdir_chunk_header *h)
{
        h->length = dnet_bswap16(h->length);
        h->num = dnet_bswap16(h->num);
        h->key_size = dnet_bswap16(h->key_size);
        h->payload_size = dnet_bswap16(h->payload_size);
}

/* Chunk size = maximum file name length + sizeof header + sizeof pohmelfs_inode_info
 * It allows to store whole file entry on 1 chunk
 */
#define POHMELFS_CHUNK_SIZE     (NAME_MAX + 1 + sizeof(struct pohmelfs_inode_info) + sizeof(struct pohmelfs_readdir_chunk_header))

enum pohmelfs_readdir_warm_states {
        POHMELFS_READDIR_WANT_HEADER = 1,
        POHMELFS_READDIR_WANT_RECV_CHUNK,
};

struct pohmelfs_readdir_warm_priv {
        struct pohmelfs_wait            *wait;

        struct kref                     refcnt;

        struct pohmelfs_readdir_header  header;

        int                             state;
        int                             read_total; /* number of inode offsets read or processed total (in all chunks summed) */
        int                             read_in_inode; /* offset of name+pohmelfs_inode_info read in below buffer */

        char                            chunk[POHMELFS_CHUNK_SIZE];
};

static void pohmelfs_readdir_warm_free(struct kref *kref)
{
        struct pohmelfs_readdir_warm_priv *priv = container_of(kref, struct pohmelfs_readdir_warm_priv, refcnt);

        if (priv->wait)
                pohmelfs_wait_put(priv->wait);
        kfree(priv);
}

static void pohmelfs_readdir_warm_destroy(struct pohmelfs_trans *t)
{
        struct pohmelfs_readdir_warm_priv *priv = t->priv;
        struct pohmelfs_wait *wait = priv->wait;

        wake_up(&wait->wq);
        kref_put(&priv->refcnt, pohmelfs_readdir_warm_free);
}

static int pohmelfs_readdir_warm_complete(struct pohmelfs_trans *t, struct pohmelfs_state *recv)
{
        struct pohmelfs_inode *pi = pohmelfs_inode(t->inode);
        struct pohmelfs_readdir_warm_priv *priv = t->priv;
        struct pohmelfs_wait *wait = priv->wait;
        struct dnet_cmd *cmd = &recv->cmd;

        if (t->recv_data) {
                kfree(t->recv_data);
                t->recv_data = NULL;
        }

        pr_debug("pohmelfs: %s: comlete cmd size: %llu, recv offset: %llu, flags: %x\n",
                        pohmelfs_dump_id(pi->id.id), (unsigned long long)cmd->size, t->recv_offset, cmd->flags);

        if (!(cmd->flags & DNET_FLAGS_MORE)) {
                wait->condition = cmd->status;
                if (!wait->condition)
                        wait->condition = 1;
        }

        return 0;
}

static int pohmelfs_dentry_add(struct pohmelfs_inode *parent, struct pohmelfs_inode *pi, char *name, int len)
{
        struct inode *inode = &pi->vfs_inode;
        struct inode *dir = &parent->vfs_inode;
        struct dentry *dentry, *parent_dentry, *old;
        struct qstr str;
        int err;

        str.name = name;
        str.len = len;
        str.hash = full_name_hash(str.name, str.len);

        /* we do not need to hold dir->i_mutex here, don't we? :) */
        parent_dentry = d_find_alias(dir);
        if (!parent_dentry) {
                err = -ENOENT;
                goto err_out_exit;
        }

        dentry = d_lookup(parent_dentry, &str);
        if (dentry) {
                err = -EEXIST;

                dput(dentry);
                goto err_out_put_parent;
        }
        /*
         * 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_put_parent;
        }

        old = d_splice_alias(inode, dentry);
        if (unlikely(old)) {
                dput(dentry);
                dentry = old;
        } else {
                dput(dentry);
        }

        dput(parent_dentry);
        return 0;

err_out_put_parent:
        dput(parent_dentry);
err_out_exit:
        return err;
}

static int pohmelfs_update_inode(struct pohmelfs_inode *parent, struct pohmelfs_inode_info *info, char *name)
{
        struct pohmelfs_sb *psb = pohmelfs_sb(parent->vfs_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;

                pi->parent_id = parent->id;
        }

        err = pohmelfs_dentry_add(parent, pi, name, info->namelen);
        if (err)
                iput(inode);

err_out_exit:
        return err;
}

static int pohmelfs_readdir_warm_scratch(struct pohmelfs_trans *t, struct pohmelfs_state *recv)
{
        struct pohmelfs_inode *pi = pohmelfs_inode(t->inode);
        struct pohmelfs_readdir_warm_priv *priv = t->priv;
        struct dnet_cmd *cmd = &recv->cmd;
        int err = 0;

        pr_debug("pohmelfs: %s: cmd size: %llu, recv offset: %llu\n",
                        pohmelfs_dump_id(pi->id.id), (unsigned long long)cmd->size, t->recv_offset);
        while (t->recv_offset != cmd->size) {
                long rest = cmd->size - t->recv_offset;

                if (rest > POHMELFS_CHUNK_SIZE)
                        rest = POHMELFS_CHUNK_SIZE;

                err = pohmelfs_recv(t, recv, priv->chunk, rest);
                if (err < 0)
                        break;

                err = 0;
        }
        pr_debug("pohmelfs: %s: done cmd size: %llu, recv offset: %llu\n",
                        pohmelfs_dump_id(pi->id.id), (unsigned long long)cmd->size, t->recv_offset);

        return err;
}

static int pohmelfs_readdir_warm_recv_reply(struct pohmelfs_trans *t, struct pohmelfs_state *recv)
{
        struct pohmelfs_inode *pi = pohmelfs_inode(t->inode);
        struct pohmelfs_readdir_warm_priv *priv = t->priv;
        struct dnet_cmd *cmd = &recv->cmd;
        int attr_size = sizeof(struct dnet_attr) + sizeof(struct dnet_io_attr);
        long old_recv_offset;
        void *data;
        int size;
        int err = 0;

        pr_debug("pohmelfs: %s: recv reply: cmd size: %llu, recv offset: %llu\n",
                        pohmelfs_dump_id(pi->id.id), (unsigned long long)cmd->size, t->recv_offset);

        if (t->recv_offset < attr_size) {
                data = &t->cmd.attr;

                data += t->recv_offset;
                size = attr_size - t->recv_offset;

                err = pohmelfs_recv(t, recv, data, size);
                if (err < 0)
                        goto err_out_exit;

                if (t->recv_offset == attr_size) {
                        dnet_convert_attr(&t->cmd.attr);
                        dnet_convert_io_attr(&t->cmd.p.io);

                        pr_debug("pohmelfs: %d:%s: cmd size: %llu, io size: %llu\n",
                                cmd->id.group_id, pohmelfs_dump_id(cmd->id.id),
                                (unsigned long long)cmd->size, (unsigned long long)t->cmd.p.io.size);

                        priv->state = POHMELFS_READDIR_WANT_HEADER;
                }
        }

        if (priv->state == POHMELFS_READDIR_WANT_HEADER) {
                int header_size_to_read = sizeof(struct pohmelfs_readdir_header) - (t->recv_offset - attr_size);

                data = &priv->header;
                data += sizeof(struct pohmelfs_readdir_header) - header_size_to_read;

                err = pohmelfs_recv(t, recv, data, header_size_to_read);
                if (err < 0)
                        goto err_out_exit;

                pohmelfs_convert_readdir_header(&priv->header);
                priv->read_total = 0;
                priv->read_in_inode = 0;
                priv->state = POHMELFS_READDIR_WANT_RECV_CHUNK;

                pr_debug("pohmelfs: %d:%s: header: header size: %d, version: %hd, chunk_size: %hd, chunk_num: %d\n",
                        cmd->id.group_id, pohmelfs_dump_id(cmd->id.id), header_size_to_read,
                        priv->header.version, priv->header.chunk_size, priv->header.chunk_num);

                if (priv->header.chunk_size > POHMELFS_CHUNK_SIZE) {
                        err = -E2BIG;
                        goto err_out_exit;
                }
        }

get_new_chunk:
        if (priv->read_total == priv->header.chunk_num) {
                err = pohmelfs_readdir_warm_scratch(t, recv);
                goto err_out_exit;
        }

        if (priv->state == POHMELFS_READDIR_WANT_RECV_CHUNK) {
                data = priv->chunk + priv->read_in_inode;
                size = POHMELFS_CHUNK_SIZE - priv->read_in_inode;

                old_recv_offset = t->recv_offset;

                err = pohmelfs_recv(t, recv, data, size);
                if (err < 0)
                        goto err_out_exit;

                priv->read_in_inode += t->recv_offset - old_recv_offset;

                if (priv->read_in_inode == POHMELFS_CHUNK_SIZE) {
                        struct pohmelfs_readdir_chunk_header *chunk_header;
                        struct pohmelfs_inode_info *info;
                        char *filename;

                        priv->read_in_inode = 0;
                        priv->read_total++;

                        chunk_header = (struct pohmelfs_readdir_chunk_header *)priv->chunk;
                        pohmelfs_convert_readdir_chunk_header(chunk_header);

                        /*
                         * Here we assume that record always fits in 1 chunk.
                         * In future this code should be changed to read several chunks
                         * and concatenate it to build continous buffer for
                         * file name and pohmelfs_inode_info structure
                         */

                        info = (struct pohmelfs_inode_info *)(priv->chunk + sizeof(struct pohmelfs_readdir_chunk_header) + chunk_header->key_size);
                        pohmelfs_convert_inode_info(info);

                        filename = (char *)(priv->chunk + sizeof(struct pohmelfs_readdir_chunk_header));

                        err = pohmelfs_update_inode(priv->wait->pi, info, filename);
                        pr_debug("pohmelfs: %d:%s: inode: %llu, namelen: %d, name: %.*s: %d\n",
                                cmd->id.group_id, pohmelfs_dump_id(info->id.id), (unsigned long long)info->ino,
                                info->namelen, info->namelen, filename, err);
                } else {
                        err = -EAGAIN;
                        goto err_out_exit;
                }

                if ((priv->read_total < priv->header.chunk_num) && (t->recv_offset < cmd->size))
                        goto get_new_chunk;
        }

        return 0;

err_out_exit:
        return err;
}

static int pohmelfs_readdir_warm_init(struct pohmelfs_trans *t)
{
        struct pohmelfs_readdir_warm_priv *priv = t->priv;

        kref_get(&priv->refcnt);
        return 0;
}

static int pohmelfs_warm_dir_group(struct inode *dir, int group_id)
{
        struct pohmelfs_sb *psb = pohmelfs_sb(dir->i_sb);
        struct pohmelfs_inode *parent = pohmelfs_inode(dir);
        struct pohmelfs_io *io;
        struct pohmelfs_readdir_warm_priv *priv;
        struct pohmelfs_wait *wait;
        long ret;
        int err;

        io = kmem_cache_zalloc(pohmelfs_io_cache, GFP_NOIO);
        if (!io) {
                err = -ENOMEM;
                goto err_out_exit;
        }

        priv = kzalloc(sizeof(struct pohmelfs_readdir_warm_priv), GFP_NOIO);
        if (!priv) {
                err = -ENOMEM;
                goto err_out_free;
        }

        kref_init(&priv->refcnt);

        wait = pohmelfs_wait_alloc(parent);
        if (!wait) {
                err = -ENOMEM;
                goto err_out_put;
        }

        priv->wait = wait;

        io->pi = parent;
        io->id = &parent->id;
        io->cflags = DNET_FLAGS_NEED_ACK | DNET_FLAGS_NOLOCK;
        io->cmd = DNET_CMD_READ;
        io->cb.recv_reply = pohmelfs_readdir_warm_recv_reply;
        io->cb.complete = pohmelfs_readdir_warm_complete;
        io->cb.destroy = pohmelfs_readdir_warm_destroy;
        io->cb.init = pohmelfs_readdir_warm_init;
        io->priv = priv;

        err = pohmelfs_send_io_group(io, group_id);
        if (err)
                goto err_out_put;

        /* destruction callback will drop reference */
        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_put;
        }

        if (wait->condition < 0) {
                err = wait->condition;
                goto err_out_put;
        }

        /* drop the reference we grabbed at creation time */
        kref_put(&priv->refcnt, pohmelfs_readdir_warm_free);
        kmem_cache_free(pohmelfs_io_cache, io);
        return 0;

err_out_put:
        kref_put(&priv->refcnt, pohmelfs_readdir_warm_free);
err_out_free:
        kmem_cache_free(pohmelfs_io_cache, io);
err_out_exit:
        return err;
}

static int pohmelfs_warm_dir(struct inode *dir)
{
        struct pohmelfs_sb *psb = pohmelfs_sb(dir->i_sb);
        int i, err = -ENOENT;

        for (i = 0; i < psb->group_num; ++i) {
                err = pohmelfs_warm_dir_group(dir, psb->groups[i]);
                if (err)
                        continue;

                return 0;
        }

        return err;
}

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("pohmelfs: %s: readdir comlete: cmd size: %llu, recv offset: %llu, flags: %x\n",
                        pohmelfs_dump_id(pi->id.id), (unsigned long long)cmd->size, t->recv_offset, 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;
                }
        } else {
                if (!wait->condition) {
                        wait->condition = cmd->status;
                        if (!wait->condition)
                                wait->condition = 1;
                }
        }

        return 0;
}

static int pohmelfs_readdir_process(void *data, int size, struct file *filp, void *dirent, filldir_t filldir)
{
        int err = 0;

        while (size > 0) {
                struct pohmelfs_dentry_disk *d = data;

                if (size < sizeof(struct pohmelfs_dentry_disk)) {
                        err = -EINVAL;
                        break;
                }

                if (size < d->len) {
                        err = -EINVAL;
                        break;
                }

                err = filldir(dirent, d->name, d->len, filp->f_pos, le64_to_cpu(d->ino), d->type);
                if (err)
                        break;

                filp->f_pos++;
                size -= sizeof(struct pohmelfs_dentry_disk) + d->len;
                data += sizeof(struct pohmelfs_dentry_disk) + d->len;
        }

        return err;
}

struct pohmelfs_readdir {
        struct dnet_raw_id                      id;
        int                                     max_size;
        int                                     fpos;
};

static int pohmelfs_readdir_group(int group_id, struct file *filp, void *dirent, filldir_t filldir)
{
        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;

        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;

        req.binary = &rd;
        req.binary_size = sizeof(struct pohmelfs_readdir);

        req.id = &parent->id;
        req.complete = pohmelfs_readdir_complete;

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

        err = pohmelfs_readdir_process(data + sizeof(struct dnet_attr), size - sizeof(struct dnet_attr), filp, dirent, filldir);

        kfree(data);

err_out_exit:
        return err;
}

static int pohmelfs_dir_open(struct inode *dir, struct file *filp)
{
        struct pohmelfs_sb *psb = pohmelfs_sb(dir->i_sb);
        struct pohmelfs_inode *pi = pohmelfs_inode(dir);

        if (get_seconds() < pi->update + psb->sync_timeout)
                return dcache_dir_open(dir, filp);

        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);
        }

        for (i = 0; i < psb->group_num; ++i) {
                err = pohmelfs_readdir_group(psb->groups[i], filp, dirent, filldir);
                if (err)
                        continue;

                pi->update = get_seconds();
                return 0;
        }

        return err;
}

const struct file_operations pohmelfs_dir_fops = {
        .open           = pohmelfs_dir_open,
        .release        = pohmelfs_dir_close,
        .read           = generic_read_dir,
        .readdir        = pohmelfs_readdir,
};