pohmelfs: Use current logging styles.

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

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

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

#define UNHASHED_OBSCURE_STRING_SIZE            sizeof(" (deleted)")

/*
 * 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, void *data, int len)
{
        struct path path;
        struct dentry *d;
        char *ptr;
        int err = 0, strlen, reduce = 0;

        d = d_find_alias(&pi->vfs_inode);
        if (!d) {
                err = -ENOENT;
                goto err_out_exit;
        }

        spin_lock(&current->fs->lock);
        path.mnt = mntget(current->fs->root.mnt);
        spin_unlock(&current->fs->lock);

        path.dentry = d;

        if (!IS_ROOT(d) && d_unhashed(d))
                reduce = 1;

        ptr = d_path(&path, data, len);
        if (IS_ERR(ptr)) {
                err = PTR_ERR(ptr);
                goto err_out_put;
        }

        if (reduce && len >= UNHASHED_OBSCURE_STRING_SIZE) {
                char *end = data + len - UNHASHED_OBSCURE_STRING_SIZE;
                *end = '\0';
        }

        strlen = len - (ptr - (char *)data);
        memmove(data, ptr, strlen);
        ptr = data;

        err = strlen - 1; /* no 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, ptr, strlen);

err_out_put:
        dput(d);
        mntput(path.mnt);
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);
}