pohmelfs: Use current logging styles.

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

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

#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt

#include <linux/module.h>
#include <linux/string.h>
#include <linux/fs.h>
#include <linux/slab.h>
#include <linux/inet.h>
#include <linux/init.h>
#include <linux/in.h>
#include <linux/in6.h>
#include <linux/blkdev.h>
#include <linux/parser.h>
#include <linux/random.h>
#include <linux/buffer_head.h>
#include <linux/exportfs.h>
#include <linux/vfs.h>
#include <linux/seq_file.h>
#include <linux/mount.h>
#include <linux/quotaops.h>
#include <asm/uaccess.h>

#include "pohmelfs.h"

#define POHMELFS_MAGIC_NUM      0x504f482e

struct kmem_cache *pohmelfs_inode_cache;
struct kmem_cache *pohmelfs_trans_cache;
struct kmem_cache *pohmelfs_inode_info_cache;
struct kmem_cache *pohmelfs_route_cache;
struct kmem_cache *pohmelfs_wait_cache;
struct kmem_cache *pohmelfs_io_cache;
struct kmem_cache *pohmelfs_inode_info_binary_package_cache;
struct kmem_cache *pohmelfs_write_cache;
struct kmem_cache *pohmelfs_dentry_cache;

static atomic_t psb_bdi_num = ATOMIC_INIT(0);

static void pohmelfs_http_compat_cleanup(struct pohmelfs_sb *psb)
{
        struct pohmelfs_path *p;
        int i;

        for (i = 0; i < psb->http_compat; ++i) {
                p = &psb->path[i];

                mutex_destroy(&p->lock);
                kfree(p->data);
        }
}

static int pohmelfs_http_compat_init(struct pohmelfs_sb *psb)
{
        int i, err;
        struct pohmelfs_path *path, *p;

        path = kmalloc(psb->http_compat * sizeof(struct pohmelfs_path), GFP_KERNEL);
        if (!path) {
                err = -ENOMEM;
                goto err_out_exit;
        }

        for (i = 0; i < psb->http_compat; ++i) {
                p = &path[i];

                mutex_init(&p->lock);

                p->data = kmalloc(PAGE_SIZE, GFP_KERNEL);
                if (!p->data) {
                        err = -ENOMEM;
                        goto err_out_free;
                }
        }

        psb->path = path;
        return 0;

err_out_free:
        while (--i >= 0) {
                p = &path[i];

                mutex_destroy(&p->lock);
                kfree(p->data);
        }

        kfree(path);
err_out_exit:
        psb->http_compat = 0;
        return err;
}

static void pohmelfs_cleanup_psb(struct pohmelfs_sb *psb)
{
        struct pohmelfs_addr *a, *tmp;

        psb->need_exit = 1;
        cancel_delayed_work(&psb->sync_work);
        destroy_workqueue(psb->wq);

        pohmelfs_pool_clean(psb->conn, psb->conn_num);

        list_for_each_entry_safe(a, tmp, &psb->addr_list, addr_entry) {
                list_del(&a->addr_entry);
                kfree(a);
        }

        crypto_free_hash(psb->hash);

        pohmelfs_http_compat_cleanup(psb);

        kfree(psb->groups);
        kfree(psb->fsid);
}

static void pohmelfs_put_super(struct super_block *sb)
{
        struct pohmelfs_sb *psb = pohmelfs_sb(sb);

        pohmelfs_cleanup_psb(psb);
        bdi_destroy(&psb->bdi);
}

struct pohmelfs_size {
        int                     group_id;
        uint64_t                bsize;          /* Block size */
        uint64_t                frsize;         /* Fragment size */
        uint64_t                blocks;         /* Filesystem size in frsize units */
        uint64_t                bfree;          /* # free blocks */
        uint64_t                bavail;         /* # free blocks for non-root */
};

static int pohmelfs_statfs(struct dentry *dentry, struct kstatfs *buf)
{
        struct super_block *sb = dentry->d_sb;
        struct pohmelfs_sb *psb = pohmelfs_sb(sb);
        struct pohmelfs_connection *c;
        struct pohmelfs_state *st;
        struct pohmelfs_size *sz;
        uint64_t min_size = ~0ULL;
        int pos = -1;
        int err, i;

        sz = kzalloc(psb->group_num * sizeof(struct pohmelfs_size), GFP_KERNEL);
        if (!sz) {
                err = -ENOMEM;
                goto err_out_exit;
        }

        for (i = 0; i < psb->group_num; ++i) {
                sz[i].group_id = psb->groups[i];
        }

        memset(buf, 0, sizeof(struct kstatfs));

        buf->f_type = POHMELFS_MAGIC_NUM; /* 'POH.' */
        buf->f_namelen = 4096;
        buf->f_files = 0;
        buf->f_bfree = buf->f_bavail = buf->f_blocks = 0;

        mutex_lock(&psb->conn_lock);
        c = &psb->conn[0];

        spin_lock(&c->state_lock);
        list_for_each_entry(st, &c->state_list, state_entry) {
                for (i = 0; i < psb->group_num; ++i) {
                        if (sz[i].group_id == st->group_id) {
                                sz[i].bsize = sb->s_blocksize;
                                sz[i].frsize = st->frsize;
                                sz[i].blocks += (st->blocks * st->frsize) >> PAGE_SHIFT;
                                sz[i].bfree += (st->bfree * st->bsize) >> PAGE_SHIFT;
                                sz[i].bavail += (st->bavail * st->bsize) >> PAGE_SHIFT;
                                break;
                        }
                }


        }
        spin_unlock(&c->state_lock);
        mutex_unlock(&psb->conn_lock);

        for (i = 0; i < psb->group_num; ++i) {
                /* skip empty groups */
                if (sz[i].blocks && (sz[i].bavail < min_size)) {
                        min_size = sz[i].bavail;
                        pos = i;
                }
        }

        if (pos == -1) {
                buf->f_bfree = buf->f_bavail = buf->f_blocks = ~0ULL >> PAGE_SHIFT;
        } else {
                buf->f_bsize = sz[pos].bsize;
                buf->f_frsize = sz[pos].frsize;
                buf->f_blocks = sz[pos].blocks;
                buf->f_bavail = sz[pos].bfree;
                buf->f_bfree = sz[pos].bavail;
        }

        kfree(sz);
        err = 0;

err_out_exit:
        return err;
}

#if 0
static int pohmelfs_show_options(struct seq_file *seq, struct vfsmount *vfs)
{
        struct pohmelfs_sb *psb = pohmelfs_sb(vfs->mnt_sb);
#else
static int pohmelfs_show_options(struct seq_file *seq, struct dentry *dentry)
{
        struct pohmelfs_sb *psb = pohmelfs_sb(dentry->d_inode->i_sb);
#endif
        struct pohmelfs_addr *a;

        mutex_lock(&psb->conn_lock);
        list_for_each_entry(a, &psb->addr_list, addr_entry) {
                struct sockaddr *sa = (struct sockaddr *)&a->sa;
                if (sa->sa_family == AF_INET) {
                        struct sockaddr_in *sin = (struct sockaddr_in *)sa;
                        seq_printf(seq, ",server=%pI4:%d:2", &sin->sin_addr.s_addr, ntohs(sin->sin_port));
                } else if (sa->sa_family == AF_INET6) {
                        struct sockaddr_in6 *sin = (struct sockaddr_in6 *)sa;
                        seq_printf(seq, ",server=%pI6:%d:6", &sin->sin6_addr.s6_addr, ntohs(sin->sin6_port));
                }
        }
        mutex_unlock(&psb->conn_lock);

        if (psb->no_read_csum)
                seq_printf(seq, ",noreadcsum");
        seq_printf(seq, ",sync_timeout=%ld", psb->sync_timeout);
        if (psb->fsid)
                seq_printf(seq, ",fsid=%s", psb->fsid);
        if (psb->successful_write_count)
                seq_printf(seq, ",successful_write_count=%d", psb->successful_write_count);
        seq_printf(seq, ",keepalive_cnt=%d", psb->keepalive_cnt);
        seq_printf(seq, ",keepalive_interval=%d", psb->keepalive_interval);
        seq_printf(seq, ",keepalive_idle=%d", psb->keepalive_idle);
        seq_printf(seq, ",readdir_allocation=%d", psb->readdir_allocation);
        if (psb->http_compat)
                seq_printf(seq, ",http_compat=%d", psb->http_compat);
        if (psb->sync_on_close)
                seq_printf(seq, ",sync_on_close");
        seq_printf(seq, ",connection_pool_size=%d", psb->conn_num);
        seq_printf(seq, ",read_wait_timeout=%ld", psb->read_wait_timeout);
        seq_printf(seq, ",write_wait_timeout=%ld", psb->write_wait_timeout);
        return 0;
}

/*
 * This is tricky function - inode cache can be shrunk and inode is about to be dropped,
 * since its last reference is dropped. But then icache can __iget() on this inode and
 * later iput() it, which will again call ->drop_inode() callback.
 *
 * So, ->drop_inode() can be called multiple times for single inode without its reintialization
 * And we better to be ready for this
 */
static int pohmelfs_drop_inode(struct inode *inode)
{
        struct pohmelfs_inode *pi = pohmelfs_inode(inode);
        struct pohmelfs_sb *psb = pohmelfs_sb(inode->i_sb);

        pr_debug("%s: %ld, mapping: %p\n",
                 pohmelfs_dump_id(pi->id.id), inode->i_ino, inode->i_mapping);

        spin_lock(&psb->inode_lock);
        if (rb_parent(&pi->node) != &pi->node)
                rb_erase(&pi->node, &psb->inode_root);
        rb_init_node(&pi->node);
        spin_unlock(&psb->inode_lock);

        return generic_drop_inode(inode);
}

static int pohmelfs_write_inode_complete(struct pohmelfs_trans *t, struct pohmelfs_state *recv)
{
        struct dnet_cmd *cmd = &recv->cmd;
        struct pohmelfs_inode_info_binary_package *bin = t->priv;
        struct pohmelfs_wait *wait = &bin->wait;

        if (cmd->flags & DNET_FLAGS_MORE)
                return 0;

        wait->condition = cmd->status;
        if (!wait->condition)
                wait->condition = 1;
        wake_up(&wait->wq);

        return 0;
}

static int pohmelfs_write_inode_init(struct pohmelfs_trans *t)
{
        struct pohmelfs_inode_info_binary_package *bin = t->priv;

        kref_get(&bin->wait.refcnt);
        return 0;
}

static void pohmelfs_write_inode_release(struct kref *kref)
{
        struct pohmelfs_wait *wait = container_of(kref, struct pohmelfs_wait, refcnt);
        struct pohmelfs_inode_info_binary_package *bin = container_of(wait, struct pohmelfs_inode_info_binary_package, wait);

        iput(&bin->wait.pi->vfs_inode);
        kmem_cache_free(pohmelfs_inode_info_binary_package_cache, bin);
}

static void pohmelfs_write_inode_destroy(struct pohmelfs_trans *t)
{
        struct pohmelfs_inode_info_binary_package *bin = t->priv;

        /*
         * We own this pointer - it points to &bin->info
         * Zero it here to prevent pohmelfs_trans_release() from freeing it
         */
        t->data = NULL;
        
        kref_put(&bin->wait.refcnt, pohmelfs_write_inode_release);
}

static int pohmelfs_write_inode(struct inode *inode, struct writeback_control *wbc)
{
        struct pohmelfs_inode *pi = pohmelfs_inode(inode);
        struct pohmelfs_inode_info_binary_package *bin;
        struct pohmelfs_sb *psb = pohmelfs_sb(inode->i_sb);
        struct pohmelfs_io *pio;
        int sync = 0;
        long ret;
        int err;

        if (wbc)
                sync = wbc->sync_mode == WB_SYNC_ALL;

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

        bin = kmem_cache_zalloc(pohmelfs_inode_info_binary_package_cache, GFP_NOIO);
        if (!bin) {
                err = -ENOMEM;
                goto err_out_free_pio;
        }

        pohmelfs_fill_inode_info(inode, &bin->info);
        err = pohmelfs_wait_init(&bin->wait, pi);
        if (err)
                goto err_out_put_bin;

        pio->pi = pi;
        pio->id = &pi->id;
        pio->cmd = DNET_CMD_WRITE;
        pio->offset = 0;
        pio->size = sizeof(struct pohmelfs_inode_info);
        pio->cflags = DNET_FLAGS_NEED_ACK;
        pio->priv = bin;
        pio->type = POHMELFS_INODE_COLUMN;
        pio->ioflags = DNET_IO_FLAGS_OVERWRITE;

        pio->data = &bin->info;
        pio->alloc_flags = POHMELFS_IO_OWN;

        pio->cb.complete = pohmelfs_write_inode_complete;
        pio->cb.init = pohmelfs_write_inode_init;
        pio->cb.destroy = pohmelfs_write_inode_destroy;

        err = pohmelfs_send_io(pio);
        if (err)
                goto err_out_put_bin;

        if (sync) {
                struct pohmelfs_wait *wait = &bin->wait;

                ret = wait_event_interruptible_timeout(wait->wq,
                                wait->condition != 0 && atomic_read(&wait->refcnt.refcount) <= 2,
                                msecs_to_jiffies(psb->write_wait_timeout));
                if (ret <= 0) {
                        err = ret;
                        if (ret == 0)
                                err = -ETIMEDOUT;
                        goto err_out_put_bin;
                }

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

err_out_put_bin:
        kref_put(&bin->wait.refcnt, pohmelfs_write_inode_release);
err_out_free_pio:
        kmem_cache_free(pohmelfs_io_cache, pio);
err_out_exit:
        return err;
}

static int pohmelfs_parse_options(struct pohmelfs_sb *psb, char *data);

static int pohmelfs_remount_fs(struct super_block *sb, int *flags, char *data)
{
        struct pohmelfs_sb *psb = pohmelfs_sb(sb);

        return pohmelfs_parse_options(psb, data);
}

static const struct super_operations pohmelfs_sb_ops = {
        .alloc_inode    = pohmelfs_alloc_inode,
        .destroy_inode  = pohmelfs_destroy_inode,
        .drop_inode     = pohmelfs_drop_inode,
        .write_inode    = pohmelfs_write_inode,
        .put_super      = pohmelfs_put_super,
        .show_options   = pohmelfs_show_options,
        .statfs         = pohmelfs_statfs,
        .remount_fs     = pohmelfs_remount_fs,
};

static void pohmelfs_sync(struct work_struct *work)
{
        struct pohmelfs_sb *psb = container_of(to_delayed_work(work), struct pohmelfs_sb, sync_work);
        struct super_block *sb = psb->sb;
        long timeout = msecs_to_jiffies(psb->sync_timeout * 1000);

        if (down_read_trylock(&sb->s_umount)) {
                sync_filesystem(sb);
                up_read(&sb->s_umount);

                pohmelfs_stat(psb, 0);
        } else {
                timeout = 0;
        }

        if (!psb->need_exit)
                queue_delayed_work(psb->wq, &psb->sync_work, timeout);
}

static int pohmelfs_init_psb(struct pohmelfs_sb *psb, struct super_block *sb)
{
        char name[16];
        int err;

        psb->inode_root = RB_ROOT;
        spin_lock_init(&psb->inode_lock);

        atomic_long_set(&psb->ino, 0);
        atomic_long_set(&psb->trans, 0);

        sb->s_fs_info = psb;
        sb->s_op = &pohmelfs_sb_ops;
        sb->s_magic = POHMELFS_MAGIC_NUM;
        sb->s_maxbytes = MAX_LFS_FILESIZE;
        sb->s_blocksize = PAGE_SIZE;
        sb->s_bdi = &psb->bdi;
        sb->s_time_gran = 0;

        psb->read_wait_timeout = 5000;
        psb->write_wait_timeout = 5000;

        psb->sync_timeout = 300;

        psb->keepalive_cnt = 5;
        psb->keepalive_interval = 10;
        psb->keepalive_idle = 30;

        psb->readdir_allocation = 4;
        psb->reconnect_timeout = msecs_to_jiffies(30000);

        psb->conn_num = 5;

        psb->sb = sb;

        psb->hash = crypto_alloc_hash("sha512", 0, CRYPTO_ALG_ASYNC);
        if (IS_ERR(psb->hash)) {
                err = PTR_ERR(psb->hash);
                goto err_out_exit;
        }

        snprintf(name, sizeof(name), "pohmelfs-sync-%d", psb->bdi_num);
        psb->wq = alloc_workqueue(name, WQ_NON_REENTRANT | WQ_UNBOUND | WQ_FREEZABLE | WQ_MEM_RECLAIM, 0);
        if (!psb->wq) {
                err = -ENOMEM;
                goto err_out_crypto_free;
        }

        mutex_init(&psb->conn_lock);
        INIT_LIST_HEAD(&psb->addr_list);

        INIT_DELAYED_WORK(&psb->sync_work, pohmelfs_sync);

        return 0;

err_out_crypto_free:
        crypto_free_hash(psb->hash);
err_out_exit:
        psb->sb = NULL;
        sb->s_fs_info = NULL;
        return err;
}

static int pohmelfs_parse_addr(char *addr, struct sockaddr_storage *a, int *addrlen)
{
        int family, port;
        char *ptr;
        int err = -EINVAL;

        ptr = strrchr(addr, ':');
        if (!ptr)
                goto err_out_print_wrong_param;
        *ptr++ = 0;
        if (!ptr)
                goto err_out_print_wrong_param;

        family = simple_strtol(ptr, NULL, 10);

        ptr = strrchr(addr, ':');
        if (!ptr)
                goto err_out_print_wrong_param;
        *ptr++ = 0;
        if (!ptr)
                goto err_out_print_wrong_param;

        port = simple_strtol(ptr, NULL, 10);

        if (family == AF_INET) {
                struct sockaddr_in *sin = (struct sockaddr_in *)a;

                sin->sin_family = family;
                sin->sin_port = htons(port);

                err = in4_pton(addr, strlen(addr), (u8 *)&sin->sin_addr, ':', NULL);
                *addrlen = sizeof(struct sockaddr_in);
        } else if (family == AF_INET6) {
                struct sockaddr_in6 *sin = (struct sockaddr_in6 *)a;

                sin->sin6_family = family;
                sin->sin6_port = htons(port);
                err = in6_pton(addr, strlen(addr), (u8 *)&sin->sin6_addr, ':', NULL);
                *addrlen = sizeof(struct sockaddr_in6);
        } else {
                err = -ENOTSUPP;
        }

        if (err == 1)
                err = 0;
        else if (!err)
                err = -EINVAL;

        if (err)
                goto err_out_print_wrong_param;

        return 0;

err_out_print_wrong_param:
        pr_err("%s: wrong addr: '%s', should be 'addr:port:family': %d\n",
               __func__, addr, err);
        return err;
}

static int pohmelfs_option(char *option, char *data, int *lenp, int have_data)
{
        int len;
        char *ptr;

        if (!strncmp(option, data, strlen(option))) {
                len = strlen(option);
                ptr = data + len;

                if (have_data && (!ptr || !*ptr))
                        return 0;

                *lenp = len;
                return 1;
        }

        return 0;
}

static int pohmelfs_set_groups(struct pohmelfs_sb *psb, char *value, int len)
{
        int i, num = 0, start = 0, pos = 0;
        char *ptr = value;

        for (i = 0; i < len; ++i) {
                if (value[i] == ':')
                        start = 0;
                else if (!start) {
                        start = 1;
                        num++;
                }
        }

        if (!num) {
                return -ENOENT;
        }

        /*
         * We do not allow to mess with different group sets for already built filesystem
         * But to prevent remount from failing, we just pretend that things went the right way
         */
        if (psb->groups)
                return 0;

        psb->groups = kzalloc(sizeof(int) * num, GFP_KERNEL);
        if (!psb->groups)
                return -ENOMEM;
        psb->group_num = num;

        start = 0;
        for (i = 0; i < len; ++i) {
                if (value[i] == ':') {
                        value[i] = '\0';
                        if (start) {
                                psb->groups[pos] = simple_strtol(ptr, NULL, 10);
                                pos++;
                                start = 0;
                        }
                } else if (!start) {
                        ptr = &value[i];
                        start = 1;
                }
        }

        if (start) {
                psb->groups[pos] = simple_strtol(ptr, NULL, 10);
                pos++;
        }

        return 0;
}

static int pohmelfs_parse_option(struct pohmelfs_sb *psb, char *data)
{
        int len;
        int err = 0;

        pr_debug("option: %s\n", data);

        if (pohmelfs_option("server=", data, &len, 1)) {
                struct pohmelfs_addr *a, *tmp;
                char *addr_str = data + len;

                a = kzalloc(sizeof(struct pohmelfs_addr), GFP_KERNEL);
                if (!a) {
                        err = -ENOMEM;
                        goto err_out_exit;
                }

                err = pohmelfs_parse_addr(addr_str, &a->sa, &a->addrlen);
                if (err)
                        goto err_out_exit;

                mutex_lock(&psb->conn_lock);
                list_for_each_entry(tmp, &psb->addr_list, addr_entry) {
                        if (tmp->addrlen != a->addrlen)
                                continue;

                        if (!memcmp(&tmp->sa, &a->sa, a->addrlen)) {
                                err = -EEXIST;
                                break;
                        }
                }

                if (!err)
                        list_add_tail(&a->addr_entry, &psb->addr_list);
                else
                        kfree(a);
                mutex_unlock(&psb->conn_lock);
                err = 0;
        } else if (pohmelfs_option("fsid=", data, &len, 1)) {
                data += len;
                len = strlen(data);

                psb->fsid = kmalloc(len + 1, GFP_KERNEL);
                if (!psb->fsid) {
                        err = -ENOMEM;
                        goto err_out_exit;
                }

                snprintf(psb->fsid, len + 1, "%s", data);
                psb->fsid_len = len;
        } else if (pohmelfs_option("sync_timeout=", data, &len, 1)) {
                psb->sync_timeout = simple_strtol(data + len, NULL, 10);
        } else if (pohmelfs_option("http_compat=", data, &len, 1)) {
                psb->http_compat = simple_strtol(data + len, NULL, 10);
                err = pohmelfs_http_compat_init(psb);
        } else if (pohmelfs_option("groups=", data, &len, 1)) {
                data += len;
                len = strlen(data);

                err = pohmelfs_set_groups(psb, data, len);
        } else if (pohmelfs_option("noatime", data, &len, 0)) {
                psb->sb->s_flags |= FS_NOATIME_FL;
        } else if (pohmelfs_option("relatime", data, &len, 0)) {
                psb->sb->s_flags |= MS_RELATIME;
        } else if (pohmelfs_option("noreadcsum", data, &len, 0)) {
                psb->no_read_csum = 1;
        } else if (pohmelfs_option("readcsum", data, &len, 0)) {
                psb->no_read_csum = 0;
        } else if (pohmelfs_option("successful_write_count=", data, &len, 1)) {
                psb->successful_write_count = simple_strtol(data + len, NULL, 10);
        } else if (pohmelfs_option("keepalive_cnt=", data, &len, 1)) {
                psb->keepalive_cnt = simple_strtol(data + len, NULL, 10);
        } else if (pohmelfs_option("keepalive_idle=", data, &len, 1)) {
                psb->keepalive_idle = simple_strtol(data + len, NULL, 10);
        } else if (pohmelfs_option("keepalive_interval=", data, &len, 1)) {
                psb->keepalive_interval = simple_strtol(data + len, NULL, 10);
        } else if (pohmelfs_option("readdir_allocation=", data, &len, 1)) {
                psb->readdir_allocation = simple_strtol(data + len, NULL, 10);
        } else if (pohmelfs_option("sync_on_close", data, &len, 0)) {
                psb->sync_on_close = 1;
        } else if (pohmelfs_option("connection_pool_size=", data, &len, 1)) {
                psb->conn_num = simple_strtol(data + len, NULL, 10);
                if (psb->conn_num < 2)
                        psb->conn_num = 2;
        } else if (pohmelfs_option("read_wait_timeout=", data, &len, 1)) {
                psb->read_wait_timeout = simple_strtol(data + len, NULL, 10);
        } else if (pohmelfs_option("write_wait_timeout=", data, &len, 1)) {
                psb->write_wait_timeout = simple_strtol(data + len, NULL, 10);
        } else {
                err = -ENOTSUPP;
        }

err_out_exit:
        return err;
}

static int pohmelfs_parse_options(struct pohmelfs_sb *psb, char *data)
{
        int err = -ENOENT;
        char *ptr, *start;

        ptr = start = data;

        while (ptr && *ptr) {
                if (*ptr == ',') {
                        *ptr = '\0';
                        err = pohmelfs_parse_option(psb, start);
                        if (err)
                                goto err_out_exit;
                        ptr++;
                        if (ptr && *ptr)
                                start = ptr;

                        continue;
                }

                ptr++;
        }

        if (start != ptr) {
                err = pohmelfs_parse_option(psb, start);
                if (err)
                        goto err_out_exit;
        }

err_out_exit:
        return err;
}

static int pohmelfs_fill_super(struct super_block *sb, void *data, int silent)
{
        struct pohmelfs_sb *psb;
        int err;

        psb = kzalloc(sizeof(struct pohmelfs_sb), GFP_KERNEL);
        if (!psb) {
                err = -ENOMEM;
                goto err_out_exit;
        }

        psb->bdi_num = atomic_inc_return(&psb_bdi_num);

        err = bdi_init(&psb->bdi);
        if (err)
                goto err_out_free_psb;

        psb->bdi.ra_pages = default_backing_dev_info.ra_pages;

        err = bdi_register(&psb->bdi, NULL, "pfs-%d", psb->bdi_num);
        if (err) {
                bdi_destroy(&psb->bdi);
                goto err_out_free_psb;
        }

        err = pohmelfs_init_psb(psb, sb);
        if (err)
                goto err_out_free_bdi;

        psb->root = pohmelfs_new_inode(psb, 0755|S_IFDIR);
        if (IS_ERR(psb->root)) {
                err = PTR_ERR(psb->root);
                goto err_out_cleanup_psb;
        }

        err = pohmelfs_parse_options(psb, data);
        if (err)
                goto err_out_put_root;

        if (!psb->group_num || list_empty(&psb->addr_list)) {
                err = -EINVAL;
                pr_err("you have to specify number of groups and add remote node address (at least one)\n");
                goto err_out_put_root;
        }

        if (!psb->fsid_len) {
                char str[] = "pohmelfs";
                err = pohmelfs_hash(psb, str, 8, &psb->root->id);
        } else {
                err = pohmelfs_hash(psb, psb->fsid, psb->fsid_len, &psb->root->id);
        }
        if (err)
                goto err_out_put_root;

        err = psb->conn_num;
        psb->conn_num = 0;
        err = pohmelfs_pool_resize(psb, err);
        if (err)
                goto err_out_put_root;

        sb->s_root = d_make_root(&psb->root->vfs_inode);
        if (!sb->s_root) {
                err = -ENOMEM;
                goto err_out_cleanup_psb;
        }

        queue_delayed_work(psb->wq, &psb->sync_work, msecs_to_jiffies(psb->sync_timeout * 1000));
        pohmelfs_stat(psb, 0);

        return 0;

err_out_put_root:
        iput(&psb->root->vfs_inode);
err_out_cleanup_psb:
        pohmelfs_cleanup_psb(psb);
err_out_free_bdi:
        bdi_destroy(&psb->bdi);
err_out_free_psb:
        kfree(psb);
err_out_exit:
        pr_err("%s: error: %d\n", __func__, err);
        return err;
}

static struct dentry *pohmelfs_mount(struct file_system_type *fs_type,
                       int flags, const char *dev_name, void *data)
{
        return mount_nodev(fs_type, flags, data, pohmelfs_fill_super);
}

static void pohmelfs_kill_sb(struct super_block *sb)
{
        sync_inodes_sb(sb);
        kill_anon_super(sb);
}

static struct file_system_type pohmelfs_type = {
        .owner          = THIS_MODULE,
        .name           = "pohmelfs",
        .mount          = pohmelfs_mount,
        .kill_sb        = pohmelfs_kill_sb,
};

static void pohmelfs_cleanup_cache(void)
{
        kmem_cache_destroy(pohmelfs_trans_cache);
        kmem_cache_destroy(pohmelfs_inode_cache);
        kmem_cache_destroy(pohmelfs_inode_info_cache);
        kmem_cache_destroy(pohmelfs_route_cache);
        kmem_cache_destroy(pohmelfs_wait_cache);
        kmem_cache_destroy(pohmelfs_io_cache);
        kmem_cache_destroy(pohmelfs_inode_info_binary_package_cache);
        kfree(pohmelfs_scratch_buf);
        kmem_cache_destroy(pohmelfs_write_cache);
        kmem_cache_destroy(pohmelfs_dentry_cache);
}

static int pohmelfs_init_cache(void)
{
        int err = -ENOMEM;

        pohmelfs_inode_cache = KMEM_CACHE(pohmelfs_inode, SLAB_RECLAIM_ACCOUNT|SLAB_MEM_SPREAD);
        if (!pohmelfs_inode_cache)
                goto err_out_exit;

        pohmelfs_trans_cache = KMEM_CACHE(pohmelfs_trans, SLAB_RECLAIM_ACCOUNT|SLAB_MEM_SPREAD);
        if (!pohmelfs_trans_cache)
                goto err_out_destroy_inode_cache;

        pohmelfs_inode_info_cache = KMEM_CACHE(pohmelfs_inode_info, SLAB_RECLAIM_ACCOUNT|SLAB_MEM_SPREAD);
        if (!pohmelfs_inode_info_cache)
                goto err_out_destroy_trans_cache;

        pohmelfs_route_cache = KMEM_CACHE(pohmelfs_route, SLAB_RECLAIM_ACCOUNT|SLAB_MEM_SPREAD);
        if (!pohmelfs_route_cache)
                goto err_out_destroy_inode_info_cache;

        pohmelfs_wait_cache = KMEM_CACHE(pohmelfs_wait, SLAB_RECLAIM_ACCOUNT|SLAB_MEM_SPREAD);
        if (!pohmelfs_wait_cache)
                goto err_out_destroy_inode_info_cache;

        pohmelfs_io_cache = KMEM_CACHE(pohmelfs_io, SLAB_RECLAIM_ACCOUNT|SLAB_MEM_SPREAD);
        if (!pohmelfs_io_cache)
                goto err_out_destroy_wait_cache;

        pohmelfs_scratch_buf = kmalloc(pohmelfs_scratch_buf_size, GFP_KERNEL);
        if (!pohmelfs_scratch_buf) {
                err = -ENOMEM;
                goto err_out_destroy_io_cache;
        }

        pohmelfs_inode_info_binary_package_cache = KMEM_CACHE(pohmelfs_inode_info_binary_package, SLAB_RECLAIM_ACCOUNT|SLAB_MEM_SPREAD);
        if (!pohmelfs_inode_info_binary_package_cache)
                goto err_out_free_scratch;

        pohmelfs_write_cache = KMEM_CACHE(pohmelfs_write_ctl, SLAB_RECLAIM_ACCOUNT|SLAB_MEM_SPREAD);
        if (!pohmelfs_write_cache)
                goto err_out_destroy_inode_info_binary_package_cache;

        pohmelfs_dentry_cache = KMEM_CACHE(pohmelfs_dentry, SLAB_RECLAIM_ACCOUNT|SLAB_MEM_SPREAD);
        if (!pohmelfs_dentry_cache)
                goto err_out_destroy_write_cache;

        return 0;

err_out_destroy_write_cache:
        kmem_cache_destroy(pohmelfs_write_cache);
err_out_destroy_inode_info_binary_package_cache:
        kmem_cache_destroy(pohmelfs_inode_info_binary_package_cache);
err_out_free_scratch:
        kfree(pohmelfs_scratch_buf);
err_out_destroy_io_cache:
        kmem_cache_destroy(pohmelfs_io_cache);
err_out_destroy_wait_cache:
        kmem_cache_destroy(pohmelfs_wait_cache);
err_out_destroy_inode_info_cache:
        kmem_cache_destroy(pohmelfs_inode_info_cache);
err_out_destroy_trans_cache:
        kmem_cache_destroy(pohmelfs_trans_cache);
err_out_destroy_inode_cache:
        kmem_cache_destroy(pohmelfs_inode_cache);
err_out_exit:
        return err;
}

static int __init pohmelfs_init(void)
{
        int err;

        err = pohmelfs_init_cache();
        if (err)
                goto err_out_exit;

        err = register_filesystem(&pohmelfs_type);
        if (err)
                goto err_out_cleanup_cache;

        return 0;

err_out_cleanup_cache:
        pohmelfs_cleanup_cache();
err_out_exit:
        return err;
}

static void __exit pohmelfs_exit(void)
{
        unregister_filesystem(&pohmelfs_type);
        pohmelfs_cleanup_cache();
}

module_init(pohmelfs_init)
module_exit(pohmelfs_exit)

MODULE_AUTHOR("Evgeniy Polyakov <zbr@ioremap.net>");
MODULE_DESCRIPTION("POHMELFS");
MODULE_LICENSE("GPL");