/*
   Copyright (c) 2008-2012 Red Hat, Inc. <http://www.redhat.com>
   This file is part of GlusterFS.

   This file is licensed to you under your choice of the GNU Lesser
   General Public License, version 3 or any later version (LGPLv3 or
   later), or the GNU General Public License, version 2 (GPLv2), in all
   cases as published by the Free Software Foundation.
*/
#include <fnmatch.h>

#include "quota.h"
#include "common-utils.h"
#include "defaults.h"

int32_t
quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
                   char *name, uuid_t par);

struct volume_options options[];


static int32_t
__quota_init_inode_ctx (inode_t *inode, int64_t hard_lim, int64_t soft_lim,
                        xlator_t *this, dict_t *dict, struct iatt *buf,
                        quota_inode_ctx_t **context)
{
        int32_t            ret  = -1;
        int64_t           *size = 0;
        quota_inode_ctx_t *ctx  = NULL;

        if (inode == NULL) {
                goto out;
        }

        QUOTA_ALLOC_OR_GOTO (ctx, quota_inode_ctx_t, out);

        ctx->hard_lim = hard_lim;
        ctx->soft_lim = soft_lim;

        if (buf)
                ctx->buf = *buf;

        LOCK_INIT(&ctx->lock);

        if (context != NULL) {
                *context = ctx;
        }

        INIT_LIST_HEAD (&ctx->parents);

        if (dict != NULL) {
                ret = dict_get_bin (dict, QUOTA_SIZE_KEY, (void **) &size);
                if (ret == 0) {
                        ctx->size = ntoh64 (*size);
                        gettimeofday (&ctx->tv, NULL);
                }
        }

        ret = __inode_ctx_put (inode, this, (uint64_t )(long)ctx);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot set quota context in inode (gfid:%s)",
                        uuid_utoa (inode->gfid));
        }
out:
        return ret;
}

int
quota_loc_fill (loc_t *loc, inode_t *inode, inode_t *parent, char *path)
{
        int ret = -1;

        if (!loc) {
                return ret;
        }

        if (inode) {
                loc->inode = inode_ref (inode);
        }

        if (parent) {
                loc->parent = inode_ref (parent);
        }

        loc->path = gf_strdup (path);
        if (!loc->path) {
                goto loc_wipe;
        }

        loc->name = strrchr (loc->path, '/');
        if (loc->name) {
                loc->name++;
        } else {
                goto loc_wipe;
        }

        ret = 0;

loc_wipe:
        if (ret < 0) {
                loc_wipe (loc);
        }

        return ret;
}


int
quota_inode_loc_fill (inode_t *inode, loc_t *loc)
{
        char            *resolvedpath = NULL;
        inode_t         *parent       = NULL;
        int              ret          = -1;
        xlator_t        *this         = NULL;

        if ((!inode) || (!loc)) {
                return ret;
        }

        this = THIS;

        if ((inode) && __is_root_gfid (inode->gfid)) {
                loc->parent = NULL;
                goto ignore_parent;
        }

        parent = inode_parent (inode, 0, NULL);
        if (!parent) {
                gf_log (this->name, GF_LOG_DEBUG,
                        "cannot find parent for inode (gfid:%s)",
                        uuid_utoa (inode->gfid));
                goto err;
        }

ignore_parent:
        ret = inode_path (inode, NULL, &resolvedpath);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_DEBUG,
                        "cannot construct path for inode (gfid:%s)",
                        uuid_utoa (inode->gfid));
                goto err;
        }

        ret = quota_loc_fill (loc, inode, parent, resolvedpath);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_WARNING, "cannot fill loc");
                goto err;
        }

err:
        if (parent) {
                inode_unref (parent);
        }

        GF_FREE (resolvedpath);

        return ret;
}


int32_t
quota_local_cleanup (xlator_t *this, quota_local_t *local)
{
        if (local == NULL) {
                goto out;
        }

        loc_wipe (&local->loc);
        loc_wipe (&local->newloc);
        loc_wipe (&local->oldloc);
        loc_wipe (&local->validate_loc);

        inode_unref (local->inode);
        LOCK_DESTROY (&local->lock);

        mem_put (local);
out:
        return 0;
}


static inline quota_local_t *
quota_local_new ()
{
        quota_local_t *local = NULL;
        local = mem_get0 (THIS->local_pool);
        if (local)
                LOCK_INIT (&local->lock);
        return local;
}


quota_dentry_t *
__quota_dentry_new (quota_inode_ctx_t *ctx, char *name, uuid_t par)
{
        quota_dentry_t    *dentry = NULL;
        GF_UNUSED int32_t  ret    = 0;

        QUOTA_ALLOC_OR_GOTO (dentry, quota_dentry_t, err);

        INIT_LIST_HEAD (&dentry->next);

        dentry->name = gf_strdup (name);
        if (dentry->name == NULL) {
                GF_FREE (dentry);
                goto err;
        }

        uuid_copy (dentry->par, par);

        list_add_tail (&dentry->next, &ctx->parents);
err:
        return dentry;
}


void
__quota_dentry_free (quota_dentry_t *dentry)
{
        if (dentry == NULL) {
                goto out;
        }

        list_del_init (&dentry->next);

        GF_FREE (dentry->name);
        GF_FREE (dentry);
out:
        return;
}

inline void
quota_resume_fop_if_validation_done (quota_local_t *local)
{
        call_stub_t *stub       = NULL;
        int          link_count = -1;

        if (local == NULL)
                goto out;

        LOCK (&local->lock);
        {
                link_count = local->link_count;
                if (link_count == 0) {
                        stub = local->stub;
                        local->stub = NULL;
                }
        }
        UNLOCK (&local->lock);

        if (stub != NULL) {
                call_resume (stub);
        }
out:
        return;
}

inline void
quota_handle_validate_error (quota_local_t *local, int32_t op_ret,
                             int32_t op_errno)
{
        if (local == NULL)
                goto out;

        LOCK (&local->lock);
        {
                if (op_ret < 0) {
                        local->op_ret = op_ret;
                        local->op_errno = op_errno;
                }

                /* we abort checking limits on this path to root */
                local->link_count--;
        }
        UNLOCK (&local->lock);

        quota_resume_fop_if_validation_done (local);
out:
        return;
}

int32_t
quota_validate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                    int32_t op_ret, int32_t op_errno, inode_t *inode,
                    struct iatt *buf, dict_t *xdata, struct iatt *postparent)
{
        quota_local_t     *local      = NULL;
        int32_t            ret        = 0;
        quota_inode_ctx_t *ctx        = NULL;
        int64_t           *size       = 0;
        uint64_t           value      = 0;

        local = frame->local;

        if (op_ret < 0) {
                goto unwind;
        }

        GF_ASSERT (local);
        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO_WITH_ERROR ("quota", this, unwind, op_errno,
                                        EINVAL);
        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, xdata, unwind, op_errno,
                                        EINVAL);

        ret = inode_ctx_get (local->validate_loc.inode, this, &value);

        ctx = (quota_inode_ctx_t *)(unsigned long)value;
        if ((ret == -1) || (ctx == NULL)) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context is not present in inode (gfid:%s)",
                        uuid_utoa (local->validate_loc.inode->gfid));
                op_errno = EINVAL;
                goto unwind;
        }

        ret = dict_get_bin (xdata, QUOTA_SIZE_KEY, (void **) &size);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_WARNING,
                        "size key not present in dict");
                op_errno = EINVAL;
                goto unwind;
        }

        local->just_validated = 1; /* so that we don't go into infinite
                                    * loop of validation and checking
                                    * limit when timeout is zero.
                                    */
        LOCK (&ctx->lock);
        {
                ctx->size = ntoh64 (*size);
                gettimeofday (&ctx->tv, NULL);
        }
        UNLOCK (&ctx->lock);

        quota_check_limit (frame, local->validate_loc.inode, this, NULL, NULL);
        return 0;

unwind:
        quota_handle_validate_error (local, op_ret, op_errno);
        return 0;
}


static inline uint64_t
quota_time_elapsed (struct timeval *now, struct timeval *then)
{
        return (now->tv_sec - then->tv_sec);
}


int32_t
quota_timeout (struct timeval *tv, int32_t timeout)
{
        struct timeval now       = {0,};
        int32_t        timed_out = 0;

        gettimeofday (&now, NULL);

        if (quota_time_elapsed (&now, tv) >= timeout) {
                timed_out = 1;
        }

        return timed_out;
}

int32_t
quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
                   char *name, uuid_t par)
{
        int32_t            ret                 = -1, op_errno = EINVAL;
        inode_t           *_inode              = NULL, *parent = NULL;
        quota_inode_ctx_t *ctx                 = NULL;
        quota_priv_t      *priv                = NULL;
        quota_local_t     *local               = NULL;
        char               need_validate       = 0;
        gf_boolean_t       soft_limit_exceeded = 0, hard_limit_exceeded = 0;
        int64_t            delta               = 0, wouldbe_size = 0;
        uint64_t           value               = 0;
        char               just_validated      = 0;
        uuid_t             trav_uuid           = {0,};
        dict_t            *xdata               = NULL;
        uint32_t           timeout             = 0;

        GF_VALIDATE_OR_GOTO ("quota", this, err);
        GF_VALIDATE_OR_GOTO (this->name, frame, err);
        GF_VALIDATE_OR_GOTO (this->name, inode, err);

        local = frame->local;
        GF_VALIDATE_OR_GOTO (this->name, local, err);

        delta = local->delta;

        GF_VALIDATE_OR_GOTO (this->name, local->stub, err);
        /* Allow all the trusted clients
         * Don't block the gluster internal processes like rebalance, gsyncd,
         * self heal etc from the disk quotas.
         *
         * Method: Allow all the clients with PID negative. This is by the
         * assumption that any kernel assigned pid doesn't have the negative
         * number.
         */
        if (0 > frame->root->pid) {
                ret = 0;
                return ret;
        }

        priv = this->private;

        inode_ctx_get (inode, this, &value);
        ctx = (quota_inode_ctx_t *)(unsigned long)value;

        _inode = inode_ref (inode);

        LOCK (&local->lock);
        {
                just_validated = local->just_validated;
                local->just_validated = 0;
        }
        UNLOCK (&local->lock);

        if ( par != NULL ) {
                uuid_copy (trav_uuid, par);
        }

        do {
                if (ctx != NULL) {
                        wouldbe_size = ctx->size + delta;

                        LOCK (&ctx->lock);
                        {
                                timeout = priv->soft_timeout;

                                if ((ctx->soft_lim >= 0)
                                    && (wouldbe_size > ctx->soft_lim)) {
                                        timeout = priv->hard_timeout;
                                }

                                if (!just_validated
                                    && quota_timeout (&ctx->tv, timeout)) {
                                        need_validate = 1;
                                } else if (wouldbe_size >= ctx->hard_lim) {
                                        hard_limit_exceeded = 1;
                                } else if (wouldbe_size >= ctx->soft_lim) {
                                        soft_limit_exceeded = 1;
                                }
                        }
                        UNLOCK (&ctx->lock);

                        if (need_validate) {
                                goto validate;
                        }

                        if (hard_limit_exceeded) {
                                op_errno = EDQUOT;
                                goto err;
                        }

                        if (soft_limit_exceeded) {
                                gf_log (this->name, GF_LOG_WARNING,
                                        "size (%"PRId64") of directory (%s) "
                                        "has exceeded soft-limit (%"PRId64")",
                                        wouldbe_size, uuid_utoa (_inode->gfid),
                                        ctx->soft_lim);
                        }
                }

                if (__is_root_gfid (_inode->gfid)) {
                        LOCK (&local->lock);
                        {
                                --local->link_count;
                        }
                        UNLOCK (&local->lock);

                        break;
                }

                parent = inode_parent (_inode, trav_uuid, name);

                if (name != NULL) {
                        name = NULL;
                        uuid_clear (trav_uuid);
                }

                if (parent == NULL) {
                        gf_log (this->name, GF_LOG_DEBUG,
                                "cannot find parent for inode (gfid:%s), hence "
                                "aborting enforcing quota-limits and continuing"
                                " with the fop", uuid_utoa (_inode->gfid));
                }

                inode_unref (_inode);
                _inode = parent;
                just_validated = 0;

                if (_inode == NULL) {
                        break;
                }

                value = 0;
                inode_ctx_get (_inode, this, &value);
                ctx = (quota_inode_ctx_t *)(unsigned long)value;
        } while (1);

        if (_inode != NULL) {
                inode_unref (_inode);
        }

        quota_resume_fop_if_validation_done (local);
        return 0;

validate:
        LOCK (&local->lock);
        {
                loc_wipe (&local->validate_loc);

                ret = quota_inode_loc_fill (_inode, &local->validate_loc);
                if (ret < 0) {
                        gf_log (this->name, GF_LOG_WARNING,
                                "cannot fill loc for inode (gfid:%s), hence "
                                "aborting quota-checks and continuing with fop",
                                uuid_utoa (_inode->gfid));
                }
        }
        UNLOCK (&local->lock);

        if (ret < 0) {
                op_errno = ENOMEM;
                goto err;
        }

        xdata = dict_new ();
        if (xdata == NULL) {
                op_errno = ENOMEM;
                goto err;
        }

        ret = dict_set_int8 (xdata, QUOTA_SIZE_KEY, 1);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_WARNING, "dict set failed");
                op_errno = ENOMEM;
                goto err;
        }

        ret = dict_set_str (xdata, "volume-uuid", priv->volume_uuid);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_WARNING, "dict set failed");
                op_errno = ENOMEM;
                goto err;
        }

        ret = quota_enforcer_lookup (frame, this, &local->validate_loc, xdata,
                                     quota_validate_cbk);

        if (ret < 0) {
                op_errno = ENOTCONN;
                goto err;
        }

        inode_unref (_inode);

        return 0;

err:
        quota_handle_validate_error (local, -1, op_errno);

        inode_unref (_inode);
        return 0;
}

static int32_t
quota_inode_ctx_get (inode_t *inode, int64_t hard_lim, int64_t soft_lim,
                     xlator_t *this, dict_t *dict, struct iatt *buf,
                     quota_inode_ctx_t **ctx, char create_if_absent)
{
        int32_t  ret = 0;
        uint64_t ctx_int;

        LOCK (&inode->lock);
        {
                ret = __inode_ctx_get (inode, this, &ctx_int);

                if ((ret == 0) && (ctx != NULL)) {
                        *ctx = (quota_inode_ctx_t *) (unsigned long)ctx_int;
                } else if (create_if_absent) {
                        ret = __quota_init_inode_ctx (inode, hard_lim, soft_lim,
                                                      this, dict, buf, ctx);
                }
        }
        UNLOCK (&inode->lock);

        return ret;
}


int32_t
quota_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                  int32_t op_ret, int32_t op_errno, inode_t *inode,
                  struct iatt *buf, dict_t *dict, struct iatt *postparent)
{
        int32_t            ret        = -1;
        char               found      = 0;
        quota_inode_ctx_t *ctx        = NULL;
        quota_dentry_t    *dentry     = NULL;
        uint64_t           value      = 0;
        int64_t           *size       = NULL;
        quota_limit_t     *limit      = NULL;
        int64_t            hard_lim   = -1, soft_lim_percent = -1, *ptr = NULL;
        quota_local_t     *local      = NULL;

        if (op_ret < 0)
                goto unwind;

        if (dict != NULL) {
                ret = dict_get_bin (dict, QUOTA_LIMIT_KEY, (void **) &ptr);
                limit = ptr;

                if (limit) {
                        hard_lim = ntoh64 (limit->hard_lim);
                        soft_lim_percent = ntoh64 (limit->soft_lim_percent);
                }
        }

        local = frame->local;

        inode_ctx_get (inode, this, &value);
        ctx = (quota_inode_ctx_t *)(unsigned long)value;

        if ((((ctx == NULL) || (ctx->hard_lim == hard_lim))
             && (hard_lim < 0) && !((IA_ISREG (buf->ia_type))
                                    || (IA_ISLNK (buf->ia_type))))) {
                goto unwind;
        }

        ret = quota_inode_ctx_get (inode, hard_lim, soft_lim,
                                   this, dict, buf, &ctx, 1);
        if ((ret == -1) || (ctx == NULL)) {
                gf_log (this->name, GF_LOG_WARNING, "cannot create quota "
                        "context in inode(gfid:%s)",
                        uuid_utoa (inode->gfid));
                op_ret = -1;
                op_errno = ENOMEM;
                goto unwind;
        }

        LOCK (&ctx->lock);
        {
                ctx->hard_lim = hard_lim;
                ctx->soft_lim = (hard_lim > 0) ?
                        (soft_lim_percent * hard_lim)/100 : -1;

                ctx->buf = *buf;

                if (!(IA_ISREG (buf->ia_type) || IA_ISLNK (buf->ia_type))) {
                        goto unlock;
                }

                if (local->loc.name == NULL)
                        goto unlock;

                list_for_each_entry (dentry, &ctx->parents, next) {
                        if ((strcmp (dentry->name, local->loc.name) == 0) &&
                            (uuid_compare (local->loc.parent->gfid,
                                           dentry->par) == 0)) {
                                found = 1;
                                break;
                        }
                }

                if (!found) {
                        dentry = __quota_dentry_new (ctx,
                                                     (char *)local->loc.name,
                                                     local->loc.parent->gfid);
                        if (dentry == NULL) {
                                /*
                                  gf_log (this->name, GF_LOG_WARNING,
                                        "cannot create a new dentry (par:%"
                                        PRId64", name:%s) for inode(ino:%"
                                        PRId64", gfid:%s)",
                                        uuid_utoa (local->loc.inode->gfid));
                                */
                                op_ret = -1;
                                op_errno = ENOMEM;
                                goto unlock;
                        }
                }
        }
unlock:
        UNLOCK (&ctx->lock);

unwind:
        QUOTA_STACK_UNWIND (lookup, frame, op_ret, op_errno, inode, buf,
                            dict, postparent);
        return 0;
}


int32_t
quota_lookup (call_frame_t *frame, xlator_t *this, loc_t *loc,
              dict_t *xattr_req)
{
        quota_priv_t  *priv  = NULL;
        int32_t        ret   = -1;
        quota_local_t *local = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        local = quota_local_new ();
        if (local == NULL) {
                goto err;
        }

        frame->local = local;
        loc_copy (&local->loc, loc);

        if (xattr_req == NULL) {
                xattr_req = dict_new ();
                if (xattr_req == NULL)
                        goto err;
        }

        ret = dict_set_int8 (xattr_req, QUOTA_LIMIT_KEY, 1);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_WARNING,
                        "dict set of key for hard-limit failed");
                goto err;
        }

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_lookup_cbk: default_lookup_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->lookup, loc,
                    xattr_req);

        ret = 0;

err:
        if (ret < 0) {
                QUOTA_STACK_UNWIND (lookup, frame, -1, ENOMEM,
                                    NULL, NULL, NULL, NULL);
        }

        return 0;
}


void
quota_update_size (xlator_t *this, inode_t *inode, char *name, uuid_t par,
                   int64_t delta)
{
        inode_t           *_inode    = NULL;
        inode_t           *parent    = NULL;
        uint64_t           value     = 0;
        quota_inode_ctx_t *ctx       = NULL;
        uuid_t             trav_uuid = {0,};

        GF_VALIDATE_OR_GOTO ("quota", this, out);
        GF_VALIDATE_OR_GOTO (this->name, inode, out);

        inode_ctx_get (inode, this, &value);
        ctx = (quota_inode_ctx_t *)(unsigned long)value;

        _inode = inode_ref (inode);

        if ( par != NULL ) {
                uuid_copy (trav_uuid, par);
        }

        do {
                if ((ctx != NULL) && (ctx->hard_lim >= 0)) {
                        LOCK (&ctx->lock);
                        {
                                ctx->size += delta;
                                if (ctx->size < 0)
                                        ctx->size = 0;
                        }
                        UNLOCK (&ctx->lock);
                }

                if (__is_root_gfid (_inode->gfid)) {
                        break;
                }

                parent = inode_parent (_inode, trav_uuid, name);
                if (parent == NULL) {
                        gf_log (this->name, GF_LOG_DEBUG,
                                "cannot find parent for inode (gfid:%s), hence "
                                "aborting size updation of parents",
                                uuid_utoa (_inode->gfid));
                }

                if (name != NULL) {
                        name = NULL;
                        uuid_clear (trav_uuid);
                }

                inode_unref (_inode);
                _inode = parent;

                if (_inode == NULL) {
                        break;
                }

                value = 0;
                ctx = NULL;
                inode_ctx_get (_inode, this, &value);
                ctx = (quota_inode_ctx_t *)(unsigned long)value;
        } while (1);

out:
        return;
}


int32_t
quota_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                  int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
                  struct iatt *postbuf, dict_t *xdata)
{
        int32_t                  ret            = 0;
        uint64_t                 ctx_int        = 0;
        quota_inode_ctx_t       *ctx            = NULL;
        quota_local_t           *local          = NULL;
        quota_dentry_t          *dentry         = NULL;
        int64_t                  delta          = 0;

        local = frame->local;

        if ((op_ret < 0) || (local == NULL)) {
                goto out;
        }

        ret = inode_ctx_get (local->loc.inode, this, &ctx_int);
        if (ret) {
                gf_log (this->name, GF_LOG_WARNING,
                        "%s: failed to get the context", local->loc.path);
                goto out;
        }

        ctx = (quota_inode_ctx_t *)(unsigned long) ctx_int;

        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context not set in %s (gfid:%s)",
                        local->loc.path, uuid_utoa (local->loc.inode->gfid));
                goto out;
        }

        LOCK (&ctx->lock);
        {
                ctx->buf = *postbuf;
        }
        UNLOCK (&ctx->lock);

        list_for_each_entry (dentry, &ctx->parents, next) {
                delta = (postbuf->ia_blocks - prebuf->ia_blocks) * 512;
                quota_update_size (this, local->loc.inode, dentry->name,
                                   dentry->par, delta);
        }

out:
        QUOTA_STACK_UNWIND (writev, frame, op_ret, op_errno, prebuf, postbuf,
                            xdata);

        return 0;
}


int32_t
quota_writev_helper (call_frame_t *frame, xlator_t *this, fd_t *fd,
                     struct iovec *vector, int32_t count, off_t off,
                     uint32_t flags, struct iobref *iobref, dict_t *xdata)
{
        quota_local_t *local    = NULL;
        int32_t        op_errno = EINVAL;
        quota_priv_t  *priv     = NULL;

        priv = this->private;

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto unwind;
        }

        if (local->op_ret == -1) {
                op_errno = local->op_errno;
                goto unwind;
        }

        STACK_WIND (frame,
                    priv->is_quota_on? quota_writev_cbk: default_writev_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->writev, fd,
                    vector, count, off, flags, iobref, xdata);
        return 0;

unwind:
        QUOTA_STACK_UNWIND (writev, frame, -1, op_errno, NULL, NULL, NULL);
        return 0;
}


int32_t
quota_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
              struct iovec *vector, int32_t count, off_t off,
              uint32_t flags, struct iobref *iobref, dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
        int32_t            ret     = -1, op_errno = EINVAL;
        int32_t            parents = 0;
        uint64_t           size    = 0;
        quota_local_t     *local   = NULL;
        quota_inode_ctx_t *ctx     = NULL;
        quota_dentry_t    *dentry  = NULL;
        call_stub_t       *stub    = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO ("quota", this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, fd, unwind);

        local = quota_local_new ();
        if (local == NULL) {
                goto unwind;
        }

        frame->local = local;
        local->loc.inode = inode_ref (fd->inode);

        ret = quota_inode_ctx_get (fd->inode, -1, -1, this, NULL, NULL, &ctx,
                                   0);
        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context not set in inode (gfid:%s)",
                        uuid_utoa (fd->inode->gfid));
                goto unwind;
        }

        stub = fop_writev_stub (frame, quota_writev_helper, fd, vector, count,
                                off, flags, iobref, xdata);
        if (stub == NULL) {
                op_errno = ENOMEM;
                goto unwind;
        }

        priv = this->private;
        GF_VALIDATE_OR_GOTO (this->name, priv, unwind);

        size = iov_length (vector, count);
        LOCK (&ctx->lock);
        {
                list_for_each_entry (dentry, &ctx->parents, next) {
                        parents++;
                }
        }
        UNLOCK (&ctx->lock);

        local->delta = size;
        local->link_count = parents;
        local->stub = stub;

        list_for_each_entry (dentry, &ctx->parents, next) {
                quota_check_limit (frame, fd->inode, this, dentry->name,
                                   dentry->par);
        }

        return 0;

unwind:
        QUOTA_STACK_UNWIND (writev, frame, -1, op_errno, NULL, NULL, NULL);
        return 0;

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_writev_cbk: default_writev_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->writev, fd,
                    vector, count, off, flags, iobref, xdata);
        return 0;
}


int32_t
quota_mkdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 int32_t op_ret, int32_t op_errno, inode_t *inode,
                 struct iatt *buf, struct iatt *preparent,
                 struct iatt *postparent, dict_t *xdata)
{
        QUOTA_STACK_UNWIND (mkdir, frame, op_ret, op_errno, inode,
                            buf, preparent, postparent, xdata);
        return 0;
}


int32_t
quota_mkdir_helper (call_frame_t *frame, xlator_t *this, loc_t *loc,
                    mode_t mode, mode_t umask, dict_t *xdata)
{
        quota_local_t *local    = NULL;
        int32_t        op_errno = EINVAL;
        quota_priv_t  *priv     = NULL;

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto unwind;
        }

        priv = this->private;
        op_errno = local->op_errno;

        if (local->op_ret == -1) {
                goto unwind;
        }

        STACK_WIND (frame,
                    priv->is_quota_on? quota_mkdir_cbk: default_mkdir_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->mkdir, loc,
                    mode, umask, xdata);

        return 0;

unwind:
        QUOTA_STACK_UNWIND (mkdir, frame, -1, op_errno, NULL, NULL,
                            NULL, NULL, NULL);
        return 0;
}


int32_t
quota_mkdir (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
             mode_t umask, dict_t *xdata)
{
        quota_priv_t  *priv  = NULL;
        int32_t        ret   = 0, op_errno = 0;
        quota_local_t *local = NULL;
        call_stub_t   *stub  = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        local = quota_local_new ();
        if (local == NULL) {
                op_errno = ENOMEM;
                goto err;
        }

        frame->local = local;

        ret = loc_copy (&local->loc, loc);
        if (ret) {
                op_errno = ENOMEM;
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
                goto err;
        }

        stub = fop_mkdir_stub (frame, quota_mkdir_helper, loc, mode, umask,
                               xdata);
        if (stub == NULL) {
                op_errno = ENOMEM;
                goto err;
        }

        local->stub = stub;
        local->delta = 0;
        local->link_count = 1;

        quota_check_limit (frame, loc->parent, this, NULL, NULL);
        return 0;

err:
        QUOTA_STACK_UNWIND (mkdir, frame, -1, op_errno, NULL, NULL, NULL,
                            NULL, NULL);

        return 0;

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_mkdir_cbk: default_mkdir_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->mkdir, loc,
                    mode, umask, xdata);

        return 0;
}


int32_t
quota_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                  int32_t op_ret, int32_t op_errno, fd_t *fd, inode_t *inode,
                  struct iatt *buf, struct iatt *preparent,
                  struct iatt *postparent, dict_t *xdata)
{
        int32_t            ret    = -1;
        quota_local_t     *local  = NULL;
        quota_inode_ctx_t *ctx    = NULL;
        quota_dentry_t    *dentry = NULL;

        local = frame->local;
        if (op_ret < 0) {
                goto unwind;
        }

        ret = quota_inode_ctx_get (inode, -1, -1, this, NULL, buf, &ctx, 1);
        if ((ret == -1) || (ctx == NULL)) {
                gf_log (this->name, GF_LOG_WARNING, "cannot create quota "
                        "context in inode(gfid:%s)",
                        uuid_utoa (inode->gfid));
                op_ret = -1;
                op_errno = ENOMEM;
                goto unwind;
        }

        LOCK (&ctx->lock);
        {
                ctx->buf = *buf;

                dentry = __quota_dentry_new (ctx, (char *)local->loc.name,
                                             local->loc.parent->gfid);
                if (dentry == NULL) {
                        gf_log (this->name, GF_LOG_WARNING,
                                "cannot create a new dentry (name:%s) for "
                                "inode(gfid:%s)", local->loc.name,
                                uuid_utoa (local->loc.inode->gfid));
                        op_ret = -1;
                        op_errno = ENOMEM;
                        goto unlock;
                }
        }
unlock:
        UNLOCK (&ctx->lock);

unwind:
        QUOTA_STACK_UNWIND (create, frame, op_ret, op_errno, fd, inode, buf,
                            preparent, postparent, xdata);
        return 0;
}


int32_t
quota_create_helper (call_frame_t *frame, xlator_t *this, loc_t *loc,
                     int32_t flags, mode_t mode, mode_t umask, fd_t *fd,
                     dict_t *xdata)
{
        quota_local_t *local    = NULL;
        int32_t        op_errno = EINVAL;
        quota_priv_t  *priv     = NULL;

        local = frame->local;

        priv = this->private;

        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto unwind;
        }

        if (local->op_ret == -1) {
                op_errno = local->op_errno;
                goto unwind;
        }


        STACK_WIND (frame,
                    priv->is_quota_on? quota_create_cbk: default_create_cbk,
                    FIRST_CHILD (this), FIRST_CHILD (this)->fops->create, loc,
                    flags, mode, umask, fd, xdata);
        return 0;

unwind:
        QUOTA_STACK_UNWIND (create, frame, -1, op_errno, NULL, NULL,
                            NULL, NULL, NULL, NULL);
        return 0;
}


int32_t
quota_create (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
              mode_t mode, mode_t umask, fd_t *fd, dict_t *xdata)
{
        quota_priv_t  *priv     = NULL;
        int32_t        ret      = -1;
        quota_local_t *local    = NULL;
        int32_t        op_errno = 0;
        call_stub_t   *stub     = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        local = quota_local_new ();
        if (local == NULL) {
                op_errno = ENOMEM;
                goto err;
        }

        frame->local = local;

        ret = loc_copy (&local->loc, loc);
        if (ret) {
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
                op_errno = ENOMEM;
                goto err;
        }

        stub = fop_create_stub (frame, quota_create_helper, loc, flags, mode,
                                umask, fd, xdata);
        if (stub == NULL) {
                goto err;
        }

        local->link_count = 1;
        local->stub = stub;
        local->delta = 0;

        quota_check_limit (frame, loc->parent, this, NULL, NULL);
        return 0;
err:
        QUOTA_STACK_UNWIND (create, frame, -1, op_errno, NULL, NULL, NULL,
                            NULL, NULL, NULL);

        return 0;

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_create_cbk: default_create_cbk,
                    FIRST_CHILD (this), FIRST_CHILD (this)->fops->create, loc,
                    flags, mode, umask, fd, xdata);
        return 0;
}


int32_t
quota_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                  int32_t op_ret, int32_t op_errno, struct iatt *preparent,
                  struct iatt *postparent, dict_t *xdata)
{
        quota_local_t     *local = NULL;
        quota_inode_ctx_t *ctx   = NULL;
        uint64_t           value = 0;
        quota_dentry_t    *dentry = NULL;
        quota_dentry_t    *old_dentry = NULL;

        if (op_ret < 0) {
                goto out;
        }

        local = (quota_local_t *) frame->local;

        inode_ctx_get (local->loc.inode, this, &value);
        ctx = (quota_inode_ctx_t *)(unsigned long)value;

        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context not set in inode (gfid:%s)",
                        uuid_utoa (local->loc.inode->gfid));
                goto out;
        }

        quota_update_size (this, local->loc.inode, (char *)local->loc.name,
                           local->loc.parent->gfid,
                           (-(ctx->buf.ia_blocks * 512)));

        LOCK (&ctx->lock);
        {
                list_for_each_entry (dentry, &ctx->parents, next) {
                        if ((strcmp (dentry->name, local->loc.name) == 0) &&
                            (uuid_compare (local->loc.parent->gfid,
                                           dentry->par) == 0)) {
                                old_dentry = dentry;
                                break;
                        }
                }
                if (old_dentry)
                        __quota_dentry_free (old_dentry);
        }
        UNLOCK (&ctx->lock);

out:
        QUOTA_STACK_UNWIND (unlink, frame, op_ret, op_errno, preparent,
                            postparent, xdata);
        return 0;
}


int32_t
quota_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
              dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
        int32_t        ret = 0;
        quota_local_t *local = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        local = quota_local_new ();
        if (local == NULL) {
                goto err;
        }

        frame->local = local;

        ret = loc_copy (&local->loc, loc);
        if (ret) {
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
                goto err;
        }

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_unlink_cbk: default_unlink_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->unlink, loc,
                    xflag, xdata);

        ret = 0;

err:
        if (ret == -1) {
                QUOTA_STACK_UNWIND (unlink, frame, -1, 0, NULL, NULL, NULL);
        }

        return 0;
}


int32_t
quota_link_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                int32_t op_ret, int32_t op_errno, inode_t *inode,
                struct iatt *buf, struct iatt *preparent,
                struct iatt *postparent, dict_t *xdata)
{
        int32_t               ret          = -1;
        quota_local_t        *local        = NULL;
        quota_inode_ctx_t    *ctx          = NULL;
        quota_dentry_t       *dentry       = NULL;
        char                  found        = 0;

        if (op_ret < 0) {
                goto out;
        }

        local = (quota_local_t *) frame->local;

        quota_update_size (this, local->loc.parent, NULL, NULL,
                           (buf->ia_blocks * 512));

        ret = quota_inode_ctx_get (inode, -1, -1, this, NULL, NULL, &ctx, 0);
        if ((ret == -1) || (ctx == NULL)) {
                gf_log (this->name, GF_LOG_WARNING, "cannot find quota "
                        "context in %s (gfid:%s)", local->loc.path,
                        uuid_utoa (inode->gfid));
                op_ret = -1;
                op_errno = EINVAL;
                goto out;
        }

        LOCK (&ctx->lock);
        {
                list_for_each_entry (dentry, &ctx->parents, next) {
                        if ((strcmp (dentry->name, local->loc.name) == 0) &&
                            (uuid_compare (local->loc.parent->gfid,
                                           dentry->par) == 0)) {
                                found = 1;
                                gf_log (this->name, GF_LOG_WARNING,
                                        "new entry being linked (name:%s) for "
                                        "inode (gfid:%s) is already present "
                                        "in inode-dentry-list", dentry->name,
                                        uuid_utoa (local->loc.inode->gfid));
                                break;
                        }
                }

                if (!found) {
                        dentry = __quota_dentry_new (ctx,
                                                     (char *)local->loc.name,
                                                     local->loc.parent->gfid);
                        if (dentry == NULL) {
                                gf_log (this->name, GF_LOG_WARNING,
                                        "cannot create a new dentry (name:%s) "
                                        "for inode(gfid:%s)", local->loc.name,
                                        uuid_utoa (local->loc.inode->gfid));
                                op_ret = -1;
                                op_errno = ENOMEM;
                                goto unlock;
                        }
                }

                ctx->buf = *buf;
        }
unlock:
        UNLOCK (&ctx->lock);

out:
        QUOTA_STACK_UNWIND (link, frame, op_ret, op_errno, inode, buf,
                            preparent, postparent, xdata);

        return 0;
}


int32_t
quota_link_helper (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
                   loc_t *newloc, dict_t *xdata)
{
        quota_local_t *local    = NULL;
        int32_t        op_errno = EINVAL;
        quota_priv_t  *priv     = NULL;

        priv = this->private;

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto unwind;
        }

        op_errno = local->op_errno;

        if (local->op_ret == -1) {
                goto unwind;
        }

        STACK_WIND (frame, priv->is_quota_on? quota_link_cbk: default_link_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->link, oldloc,
                    newloc, xdata);
        return 0;

unwind:
        QUOTA_STACK_UNWIND (link, frame, -1, op_errno, NULL, NULL,
                            NULL, NULL, NULL);
        return 0;
}


int32_t
quota_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
            dict_t *xdata)
{
        quota_priv_t      *priv  = NULL;
        int32_t            ret   = -1, op_errno = ENOMEM;
        quota_local_t     *local = NULL;
        quota_inode_ctx_t *ctx   = NULL;
        call_stub_t       *stub  = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);


        local = quota_local_new ();
        if (local == NULL) {
                goto err;
        }

        frame->local = (void *) local;

        ret = loc_copy (&local->loc, newloc);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
                goto err;
        }

        ret = quota_inode_ctx_get (oldloc->inode, -1, -1, this, NULL, NULL,
                                   &ctx, 0);
        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context not set in inode (gfid:%s)",
                        oldloc->inode ? uuid_utoa (oldloc->inode->gfid) : "0");
                op_errno = EINVAL;
                goto err;
        }

        stub = fop_link_stub (frame, quota_link_helper, oldloc, newloc, xdata);
        if (stub == NULL) {
                goto err;
        }

        local->link_count = 1;
        local->stub = stub;
        local->delta = ctx->buf.ia_blocks * 512;

        ret = quota_inode_ctx_get (oldloc->inode, -1, -1, this, NULL, NULL,
                                   &ctx, 0);
        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context not set in inode (gfid:%s)",
                        oldloc->inode ? uuid_utoa (oldloc->inode->gfid) : "0");
                op_errno = EINVAL;
                goto err;
        }

        quota_check_limit (frame, newloc->parent, this, NULL, NULL);
        return 0;

err:
        if (ret < 0) {
                QUOTA_STACK_UNWIND (link, frame, -1, op_errno, NULL, NULL,
                                    NULL, NULL, NULL);
        }

        return 0;

wind:
        STACK_WIND (frame, priv->is_quota_on? quota_link_cbk: default_link_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->link, oldloc,
                    newloc, xdata);
        return 0;
}


int32_t
quota_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                  int32_t op_ret, int32_t op_errno, struct iatt *buf,
                  struct iatt *preoldparent, struct iatt *postoldparent,
                  struct iatt *prenewparent, struct iatt *postnewparent,
                  dict_t *xdata)
{
        int32_t               ret              = -1;
        int64_t               size             = 0;
        quota_local_t        *local            = NULL;
        quota_inode_ctx_t    *ctx              = NULL;
        quota_dentry_t       *old_dentry       = NULL, *dentry = NULL;
        char                  new_dentry_found = 0;

        if (op_ret < 0) {
                goto out;
        }

        local = frame->local;
        if (local == NULL) {
                op_ret = -1;
                op_errno = EINVAL;
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto out;
        }

        if (IA_ISREG (local->oldloc.inode->ia_type)
            || IA_ISLNK (local->oldloc.inode->ia_type)) {
                size = buf->ia_blocks * 512;
        }

        if (local->oldloc.parent != local->newloc.parent) {
                quota_update_size (this, local->oldloc.parent, NULL, NULL,
                                   (-size));
                quota_update_size (this, local->newloc.parent, NULL, NULL,
                                   size);
        }

        if (!(IA_ISREG (local->oldloc.inode->ia_type)
              || IA_ISLNK (local->oldloc.inode->ia_type))) {
                goto out;
        }

        ret = quota_inode_ctx_get (local->oldloc.inode, -1, -1, this, NULL,
                                   NULL, &ctx, 0);
        if ((ret == -1) || (ctx == NULL)) {
                gf_log (this->name, GF_LOG_WARNING, "quota context not"
                        "set in inode(gfid:%s)",
                        uuid_utoa (local->oldloc.inode->gfid));
                op_ret = -1;
                op_errno = EINVAL;
                goto out;
        }

        LOCK (&ctx->lock);
        {
                /* decision of whether to create a context in newloc->inode
                 * depends on fuse_rename_cbk's choice of inode it retains
                 * after rename. currently it just associates oldloc->inode
                 * with new parent and name. If this changes, following code
                 * should be changed to set a new context in newloc->inode.
                 */
                list_for_each_entry (dentry, &ctx->parents, next) {
                        if ((strcmp (dentry->name, local->oldloc.name) == 0) &&
                            (uuid_compare (local->oldloc.parent->gfid,
                                           dentry->par) == 0)) {
                                old_dentry = dentry;
                        } else if ((strcmp (dentry->name,
                                            local->newloc.name) == 0) &&
                                   (uuid_compare (local->oldloc.parent->gfid,
                                                  dentry->par) == 0)) {
                                new_dentry_found = 1;
                                gf_log (this->name, GF_LOG_WARNING,
                                        "new entry being linked (name:%s) for "
                                        "inode (gfid:%s) is already present "
                                        "in inode-dentry-list", dentry->name,
                                        uuid_utoa (local->newloc.inode->gfid));
                                break;
                        }
                }

                if (old_dentry != NULL) {
                        __quota_dentry_free (old_dentry);
                } else {
                        gf_log (this->name, GF_LOG_WARNING,
                                "dentry corresponding to the path just renamed "
                                "(name:%s) is not present", local->oldloc.name);
                }

                if (!new_dentry_found) {
                        dentry = __quota_dentry_new (ctx,
                                                     (char *)local->newloc.name,
                                                     local->newloc.parent->gfid);
                        if (dentry == NULL) {
                                gf_log (this->name, GF_LOG_WARNING,
                                        "cannot create a new dentry (name:%s) "
                                        "for inode(gfid:%s)",
                                        local->newloc.name,
                                        uuid_utoa (local->newloc.inode->gfid));
                                op_ret = -1;
                                op_errno = ENOMEM;
                                goto unlock;
                        }
                }

                ctx->buf = *buf;
        }
unlock:
        UNLOCK (&ctx->lock);

out:
        QUOTA_STACK_UNWIND (rename, frame, op_ret, op_errno, buf, preoldparent,
                            postoldparent, prenewparent, postnewparent, xdata);

        return 0;
}


int32_t
quota_rename_helper (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
                     loc_t *newloc, dict_t *xdata)
{
        quota_local_t *local    = NULL;
        int32_t        op_errno = EINVAL;
        quota_priv_t  *priv     = NULL;

        priv = this->private;

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto unwind;
        }

        op_errno = local->op_errno;

        if (local->op_ret == -1) {
                goto unwind;
        }

        STACK_WIND (frame,
                    priv->is_quota_on? quota_rename_cbk: default_rename_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->rename, oldloc,
                    newloc, xdata);

        return 0;

unwind:
        QUOTA_STACK_UNWIND (rename, frame, -1, op_errno, NULL, NULL,
                            NULL, NULL, NULL, NULL);
        return 0;
}


int32_t
quota_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
              loc_t *newloc, dict_t *xdata)
{
        quota_priv_t      *priv  = NULL;
        int32_t            ret   = -1, op_errno = ENOMEM;
        quota_local_t     *local = NULL;
        quota_inode_ctx_t *ctx   = NULL;
        call_stub_t       *stub  = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        local = quota_local_new ();
        if (local == NULL) {
                goto err;
        }

        frame->local = local;

        ret = loc_copy (&local->oldloc, oldloc);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
                goto err;
        }

        ret = loc_copy (&local->newloc, newloc);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
                goto err;
        }

        stub = fop_rename_stub (frame, quota_rename_helper, oldloc, newloc,
                                xdata);
        if (stub == NULL) {
                goto err;
        }

        local->link_count = 1;
        local->stub = stub;

        if (IA_ISREG (oldloc->inode->ia_type)
            || IA_ISLNK (oldloc->inode->ia_type)) {
                ret = quota_inode_ctx_get (oldloc->inode, -1, -1, this, NULL,
                                           NULL, &ctx, 0);
                if (ctx == NULL) {
                        gf_log (this->name, GF_LOG_WARNING,
                                "quota context not set in inode (gfid:%s)",
                                oldloc->inode ? uuid_utoa (oldloc->inode->gfid)
                                : "0");
                        op_errno = EINVAL;
                        goto err;
                }
                local->delta = ctx->buf.ia_blocks * 512;
        } else {
                local->delta = 0;
        }

        quota_check_limit (frame, newloc->parent, this, NULL, NULL);
        return 0;

err:
        QUOTA_STACK_UNWIND (rename, frame, -1, op_errno, NULL,
                            NULL, NULL, NULL, NULL, NULL);
        return 0;

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_rename_cbk: default_rename_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->rename, oldloc,
                    newloc, xdata);

        return 0;
}


int32_t
quota_symlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                   int32_t op_ret, int32_t op_errno, inode_t *inode,
                   struct iatt *buf, struct iatt *preparent,
                   struct iatt *postparent, dict_t *xdata)
{
        int64_t            size   = 0;
        quota_local_t     *local  = NULL;
        quota_inode_ctx_t *ctx    = NULL;
        quota_dentry_t    *dentry = NULL;

        if (op_ret < 0) {
                goto out;
        }

        local = frame->local;
        size = buf->ia_blocks * 512;

        quota_update_size (this, local->loc.parent, NULL, NULL, size);

        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                             &ctx, 1);
        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context not set in inode (gfid:%s)",
                        uuid_utoa (local->loc.inode->gfid));
                goto out;
        }

        LOCK (&ctx->lock);
        {
                ctx->buf = *buf;

                dentry = __quota_dentry_new (ctx, (char *)local->loc.name,
                                             local->loc.parent->gfid);
                if (dentry == NULL) {
                        gf_log (this->name, GF_LOG_WARNING,
                                "cannot create a new dentry (name:%s) for "
                                "inode(gfid:%s)", local->loc.name,
                                uuid_utoa (local->loc.inode->gfid));
                        op_ret = -1;
                        op_errno = ENOMEM;
                }
        }
        UNLOCK (&ctx->lock);

out:
        QUOTA_STACK_UNWIND (symlink, frame, op_ret, op_errno, inode, buf,
                            preparent, postparent, xdata);

        return 0;
}


int
quota_symlink_helper (call_frame_t *frame, xlator_t *this, const char *linkpath,
                      loc_t *loc, mode_t umask, dict_t *xdata)
{
        quota_local_t *local    = NULL;
        int32_t        op_errno = EINVAL;
        quota_priv_t  *priv     = NULL;

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto unwind;
        }

        priv = this->private;

        if (local->op_ret == -1) {
                op_errno = local->op_errno;
                goto unwind;
        }

        STACK_WIND (frame,
                    priv->is_quota_on? quota_symlink_cbk: default_symlink_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->symlink,
                    linkpath, loc, umask, xdata);
        return 0;

unwind:
        QUOTA_STACK_UNWIND (symlink, frame, -1, op_errno, NULL, NULL,
                            NULL, NULL, NULL);
        return 0;
}


int
quota_symlink (call_frame_t *frame, xlator_t *this, const char *linkpath,
               loc_t *loc, mode_t umask, dict_t *xdata)
{
        quota_priv_t  *priv     = NULL;
        int32_t        ret      = -1;
        int32_t        op_errno = ENOMEM;
        quota_local_t *local    = NULL;
        call_stub_t   *stub     = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        local = quota_local_new ();
        if (local == NULL) {
                goto err;
        }

        frame->local = local;

        ret = loc_copy (&local->loc, loc);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
                goto err;
        }

        stub = fop_symlink_stub (frame, quota_symlink_helper, linkpath, loc,
                                 umask, xdata);
        if (stub == NULL) {
                goto err;
        }

        local->stub = stub;
        local->delta = strlen (linkpath);
        local->link_count = 1;

        quota_check_limit (frame, loc->parent, this, NULL, NULL);
        return 0;

err:
        QUOTA_STACK_UNWIND (symlink, frame, -1, op_errno, NULL, NULL, NULL,
                            NULL, NULL);

        return 0;

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_symlink_cbk: default_symlink_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->symlink,
                    linkpath, loc, umask, xdata);
        return 0;
}


int32_t
quota_truncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                    int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
                    struct iatt *postbuf, dict_t *xdata)
{
        quota_local_t     *local = NULL;
        quota_inode_ctx_t *ctx   = NULL;
        int64_t            delta = 0;

        if (op_ret < 0) {
                goto out;
        }

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto out;
        }

        delta = (postbuf->ia_blocks - prebuf->ia_blocks) * 512;

        quota_update_size (this, local->loc.inode, NULL, NULL, delta);

        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                             &ctx, 0);
        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context not set in inode (gfid:%s)",
                        uuid_utoa (local->loc.inode->gfid));
                goto out;
        }

        LOCK (&ctx->lock);
        {
                ctx->buf = *postbuf;
        }
        UNLOCK (&ctx->lock);

out:
        QUOTA_STACK_UNWIND (truncate, frame, op_ret, op_errno, prebuf,
                            postbuf, xdata);
        return 0;
}


int32_t
quota_truncate (call_frame_t *frame, xlator_t *this, loc_t *loc, off_t offset,
                dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
        int32_t          ret   = -1;
        quota_local_t   *local = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);


        local = quota_local_new ();
        if (local == NULL) {
                goto err;
        }

        frame->local = local;

        ret =  loc_copy (&local->loc, loc);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
                goto err;
        }

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_truncate_cbk: default_truncate_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->truncate, loc,
                    offset, xdata);

        return 0;
err:
        QUOTA_STACK_UNWIND (truncate, frame, -1, ENOMEM, NULL, NULL, NULL);

        return 0;
}


int32_t
quota_ftruncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                     int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
                     struct iatt *postbuf, dict_t *xdata)
{
        quota_local_t     *local = NULL;
        quota_inode_ctx_t *ctx   = NULL;
        int64_t            delta = 0;

        if (op_ret < 0) {
                goto out;
        }

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto out;
        }

        delta = (postbuf->ia_blocks - prebuf->ia_blocks) * 512;

        quota_update_size (this, local->loc.inode, NULL, NULL, delta);

        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                             &ctx, 0);
        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context not set in inode (gfid:%s)",
                        uuid_utoa (local->loc.inode->gfid));
                goto out;
        }

        LOCK (&ctx->lock);
        {
                ctx->buf = *postbuf;
        }
        UNLOCK (&ctx->lock);

out:
        QUOTA_STACK_UNWIND (ftruncate, frame, op_ret, op_errno, prebuf,
                            postbuf, xdata);
        return 0;
}


int32_t
quota_ftruncate (call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
                 dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
        quota_local_t   *local = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        local = quota_local_new ();
        if (local == NULL)
                goto err;

        frame->local = local;

        local->loc.inode = inode_ref (fd->inode);

wind:
        STACK_WIND (frame, priv->is_quota_on?
                    quota_ftruncate_cbk: default_ftruncate_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->ftruncate, fd,
                    offset, xdata);

        return 0;
err:
        QUOTA_STACK_UNWIND (ftruncate, frame, -1, ENOMEM, NULL, NULL, NULL);

        return 0;
}


int32_t
quota_send_dir_limit_to_cli (call_frame_t *frame, xlator_t *this,
                             inode_t *inode, const char *name)
{
        int32_t            ret               = 0;
        char               dir_limit [1024]  = {0, };
        dict_t            *dict              = NULL;
        quota_inode_ctx_t *ctx               = NULL;
        uint64_t           value             = 0;
        quota_priv_t      *priv              = NULL;

        priv = this->private;
        if (!priv->is_quota_on) {
                snprintf (dir_limit, 1024, "Quota is disabled please turn on");
                goto dict_set;
        }

        ret = inode_ctx_get (inode, this, &value);
        if (ret < 0)
                goto out;

        ctx = (quota_inode_ctx_t *)(unsigned long)value;
        snprintf (dir_limit, 1024, "%"PRId64",%"PRId64, ctx->size,
                  ctx->hard_lim);

dict_set:
        dict = dict_new ();
        if (dict == NULL) {
                ret = -1;
                goto out;
        }

        ret = dict_set_str (dict, (char *) name, dir_limit);
        if (ret < 0)
                goto out;

        gf_log (this->name, GF_LOG_DEBUG, "str = %s", dir_limit);

        QUOTA_STACK_UNWIND (getxattr, frame, 0, 0, dict, NULL);

        ret = 0;

out:
        return ret;
}


int32_t
quota_fgetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
                 const char *name, dict_t *xdata)
{
        int32_t ret     = 0;

        if (name && strcasecmp (name, "trusted.limit.list") == 0) {
                ret = quota_send_dir_limit_to_cli (frame, this, fd->inode,
                                                   name);
                if (ret == 0) {
                        return 0;
                }
        }

        STACK_WIND (frame, default_fgetxattr_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->fgetxattr, fd, name, xdata);
        return 0;
}


int32_t
quota_getxattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
                const char *name, dict_t *xdata)
{
        int32_t ret     = 0;

        if ((name != NULL) && strcasecmp (name, "trusted.limit.list") == 0) {
                ret = quota_send_dir_limit_to_cli (frame, this, loc->inode,
                                                   name);
                if (ret == 0)
                        return 0;
        }

        STACK_WIND (frame, default_getxattr_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->getxattr, loc, name, xdata);
        return 0;
}


int32_t
quota_stat_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                int32_t op_ret, int32_t op_errno, struct iatt *buf,
                dict_t *xdata)
{
        quota_local_t     *local = NULL;
        quota_inode_ctx_t *ctx   = NULL;

        if (op_ret < 0) {
                goto out;
        }

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto out;
        }

        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                             &ctx, 0);
        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_DEBUG,
                        "quota context not set in inode (gfid:%s)",
                        uuid_utoa (local->loc.inode->gfid));
                goto out;
        }

        LOCK (&ctx->lock);
        {
                if (buf)
                        ctx->buf = *buf;
        }
        UNLOCK (&ctx->lock);

out:
        QUOTA_STACK_UNWIND (stat, frame, op_ret, op_errno, buf, xdata);
        return 0;
}


int32_t
quota_stat (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
        quota_local_t *local = NULL;
        int32_t        ret   = -1;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);


        local = quota_local_new ();
        if (local == NULL) {
                goto unwind;
        }

        frame->local = local;
        ret = loc_copy (&local->loc, loc);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
                goto unwind;
        }

wind:
        STACK_WIND (frame, priv->is_quota_on? quota_stat_cbk: default_stat_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->stat, loc,
                    xdata);
        return 0;

unwind:
        QUOTA_STACK_UNWIND (stat, frame, -1, ENOMEM, NULL, NULL);
        return 0;
}


int32_t
quota_fstat_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 int32_t op_ret, int32_t op_errno, struct iatt *buf,
                 dict_t *xdata)
{
        quota_local_t     *local = NULL;
        quota_inode_ctx_t *ctx   = NULL;

        if (op_ret < 0) {
                goto out;
        }

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto out;
        }

        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                             &ctx, 0);
        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context not set in inode (gfid:%s)",
                        uuid_utoa (local->loc.inode->gfid));
                goto out;
        }

        LOCK (&ctx->lock);
        {
                if (buf)
                        ctx->buf = *buf;
        }
        UNLOCK (&ctx->lock);

out:
        QUOTA_STACK_UNWIND (fstat, frame, op_ret, op_errno, buf, xdata);
        return 0;
}


int32_t
quota_fstat (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
        quota_local_t *local = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        local = quota_local_new ();
        if (local == NULL) {
                goto unwind;
        }

        frame->local = local;

        local->loc.inode = inode_ref (fd->inode);

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_fstat_cbk: default_fstat_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->fstat, fd,
                    xdata);
        return 0;

unwind:
        QUOTA_STACK_UNWIND (fstat, frame, -1, ENOMEM, NULL, NULL);
        return 0;
}


int32_t
quota_readlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                    int32_t op_ret, int32_t op_errno, const char *path,
                    struct iatt *buf, dict_t *xdata)
{
        quota_local_t     *local = NULL;
        quota_inode_ctx_t *ctx   = NULL;

        if (op_ret < 0) {
                goto out;
        }

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto out;
        }

        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                             &ctx, 0);
        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context not set in inode (gfid:%s)",
                        uuid_utoa (local->loc.inode->gfid));
                goto out;
        }

        LOCK (&ctx->lock);
        {
                ctx->buf = *buf;
        }
        UNLOCK (&ctx->lock);

out:
        QUOTA_STACK_UNWIND (readlink, frame, op_ret, op_errno, path, buf,
                            xdata);
        return 0;
}


int32_t
quota_readlink (call_frame_t *frame, xlator_t *this, loc_t *loc, size_t size,
                dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
        quota_local_t *local = NULL;
        int32_t        ret   = -1;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        local = quota_local_new ();
        if (local == NULL) {
                goto unwind;
        }

        frame->local = local;

        ret = loc_copy (&local->loc, loc);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
                goto unwind;
        }

wind:
        STACK_WIND (frame, priv->is_quota_on?
                    quota_readlink_cbk: default_readlink_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->readlink, loc,
                    size, xdata);
        return 0;

unwind:
        QUOTA_STACK_UNWIND (readlink, frame, -1, ENOMEM, NULL, NULL, NULL);
        return 0;
}


int32_t
quota_readv_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 int32_t op_ret, int32_t op_errno, struct iovec *vector,
                 int32_t count, struct iatt *buf, struct iobref *iobref,
                 dict_t *xdata)
{
        quota_local_t     *local = NULL;
        quota_inode_ctx_t *ctx   = NULL;

        if (op_ret < 0) {
                goto out;
        }

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto out;
        }

        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                             &ctx, 0);
        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context not set in inode (gfid:%s)",
                        uuid_utoa (local->loc.inode->gfid));
                goto out;
        }

        LOCK (&ctx->lock);
        {
                ctx->buf = *buf;
        }
        UNLOCK (&ctx->lock);

out:
        QUOTA_STACK_UNWIND (readv, frame, op_ret, op_errno, vector, count,
                            buf, iobref, xdata);
        return 0;
}


int32_t
quota_readv (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
             off_t offset, uint32_t flags, dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
        quota_local_t *local = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        local = quota_local_new ();
        if (local == NULL) {
                goto unwind;
        }

        frame->local = local;

        local->loc.inode = inode_ref (fd->inode);

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_readv_cbk: default_readv_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->readv, fd,
                    size, offset, flags, xdata);
        return 0;

unwind:
        QUOTA_STACK_UNWIND (readv, frame, -1, ENOMEM, NULL, -1, NULL, NULL,
                            NULL);
        return 0;
}


int32_t
quota_fsync_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
                 struct iatt *postbuf, dict_t *xdata)
{
        quota_local_t     *local = NULL;
        quota_inode_ctx_t *ctx   = NULL;

        if (op_ret < 0) {
                goto out;
        }

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto out;
        }

        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                             &ctx, 0);
        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context not set in inode (gfid:%s)",
                        uuid_utoa (local->loc.inode->gfid));
                goto out;
        }

        LOCK (&ctx->lock);
        {
                ctx->buf = *postbuf;
        }
        UNLOCK (&ctx->lock);

out:
        QUOTA_STACK_UNWIND (fsync, frame, op_ret, op_errno, prebuf, postbuf,
                            xdata);
        return 0;
}


int32_t
quota_fsync (call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t flags,
             dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
        quota_local_t *local = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        local = quota_local_new ();
        if (local == NULL) {
                goto unwind;
        }

        local->loc.inode = inode_ref (fd->inode);

        frame->local = local;

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_fsync_cbk: default_fsync_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->fsync, fd,
                    flags, xdata);
        return 0;

unwind:
        QUOTA_STACK_UNWIND (fsync, frame, -1, ENOMEM, NULL, NULL, NULL);
        return 0;

}


int32_t
quota_setattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                   int32_t op_ret, int32_t op_errno, struct iatt *statpre,
                   struct iatt *statpost, dict_t *xdata)
{
        quota_local_t     *local = NULL;
        quota_inode_ctx_t *ctx   = NULL;

        if (op_ret < 0) {
                goto out;
        }

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto out;
        }

        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                             &ctx, 0);
        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_DEBUG,
                        "quota context not set in inode (gfid:%s)",
                        uuid_utoa (local->loc.inode->gfid));
                goto out;
        }

        LOCK (&ctx->lock);
        {
                if (statpost)
                        ctx->buf = *statpost;
        }
        UNLOCK (&ctx->lock);

out:
        QUOTA_STACK_UNWIND (setattr, frame, op_ret, op_errno, statpre,
                            statpost, xdata);
        return 0;
}


int32_t
quota_setattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
               struct iatt *stbuf, int32_t valid, dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
        quota_local_t *local = NULL;
        int32_t        ret   = -1;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        local = quota_local_new ();
        if (local == NULL) {
                goto unwind;
        }

        frame->local = local;

        ret = loc_copy (&local->loc, loc);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
                goto unwind;
        }

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_setattr_cbk: default_setattr_cbk,
                    FIRST_CHILD (this), FIRST_CHILD (this)->fops->setattr, loc,
                    stbuf, valid, xdata);
        return 0;

unwind:
        QUOTA_STACK_UNWIND (setattr, frame, -1, ENOMEM, NULL, NULL, NULL);
        return 0;
}


int32_t
quota_fsetattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                    int32_t op_ret, int32_t op_errno, struct iatt *statpre,
                    struct iatt *statpost, dict_t *xdata)
{
        quota_local_t     *local = NULL;
        quota_inode_ctx_t *ctx   = NULL;

        if (op_ret < 0) {
                goto out;
        }

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto out;
        }

        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                             &ctx, 0);
        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context not set in inode (gfid:%s)",
                        uuid_utoa (local->loc.inode->gfid));
                goto out;
        }

        LOCK (&ctx->lock);
        {
                ctx->buf = *statpost;
        }
        UNLOCK (&ctx->lock);

out:
        QUOTA_STACK_UNWIND (fsetattr, frame, op_ret, op_errno, statpre,
                            statpost, xdata);
        return 0;
}


int32_t
quota_fsetattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
                struct iatt *stbuf, int32_t valid, dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
        quota_local_t *local = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);


        local = quota_local_new ();
        if (local == NULL) {
                goto unwind;
        }

        frame->local = local;

        local->loc.inode = inode_ref (fd->inode);

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_fsetattr_cbk: default_fsetattr_cbk,
                    FIRST_CHILD (this), FIRST_CHILD (this)->fops->fsetattr, fd,
                    stbuf, valid, xdata);
        return 0;

unwind:
        QUOTA_STACK_UNWIND (fsetattr, frame, -1, ENOMEM, NULL, NULL, NULL);
        return 0;
}


int32_t
quota_mknod_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 int32_t op_ret, int32_t op_errno, inode_t *inode,
                 struct iatt *buf, struct iatt *preparent,
                 struct iatt *postparent, dict_t *xdata)
{
        int32_t            ret    = -1;
        quota_local_t     *local  = NULL;
        quota_inode_ctx_t *ctx    = NULL;
        quota_dentry_t    *dentry = NULL;

        local = frame->local;
        if (op_ret < 0) {
                goto unwind;
        }

        ret = quota_inode_ctx_get (inode, -1, -1, this, NULL, buf, &ctx, 1);
        if ((ret == -1) || (ctx == NULL)) {
                gf_log (this->name, GF_LOG_WARNING, "cannot create quota "
                        "context in inode (gfid:%s)", uuid_utoa (inode->gfid));
                op_ret = -1;
                op_errno = ENOMEM;
                goto unwind;
        }

        LOCK (&ctx->lock);
        {
                ctx->buf = *buf;

                dentry = __quota_dentry_new (ctx, (char *)local->loc.name,
                                             local->loc.parent->gfid);
                if (dentry == NULL) {
                        gf_log (this->name, GF_LOG_WARNING,
                                "cannot create a new dentry (name:%s) for "
                                "inode(gfid:%s)", local->loc.name,
                                uuid_utoa (local->loc.inode->gfid));
                        op_ret = -1;
                        op_errno = ENOMEM;
                        goto unlock;
                }
        }
unlock:
        UNLOCK (&ctx->lock);

unwind:
        QUOTA_STACK_UNWIND (mknod, frame, op_ret, op_errno, inode,
                            buf, preparent, postparent, xdata);
        return 0;
}


int
quota_mknod_helper (call_frame_t *frame, xlator_t *this, loc_t *loc,
                    mode_t mode, dev_t rdev, mode_t umask, dict_t *xdata)
{
        quota_local_t *local    = NULL;
        int32_t        op_errno = EINVAL;
        quota_priv_t  *priv     = NULL;

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto unwind;
        }

        priv = this->private;

        if (local->op_ret == -1) {
                op_errno = local->op_errno;
                goto unwind;
        }

        STACK_WIND (frame,
                    priv->is_quota_on? quota_mknod_cbk: default_mknod_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->mknod, loc,
                    mode, rdev, umask, xdata);

        return 0;

unwind:
        QUOTA_STACK_UNWIND (mknod, frame, -1, op_errno, NULL, NULL,
                            NULL, NULL, NULL);
        return 0;
}


int
quota_mknod (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
             dev_t rdev, mode_t umask, dict_t *xdata)
{
        quota_priv_t  *priv     = NULL;
        int32_t        ret      = -1;
        quota_local_t *local    = NULL;
        call_stub_t   *stub     = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        local = quota_local_new ();
        if (local == NULL) {
                goto err;
        }

        frame->local = local;

        ret = loc_copy (&local->loc, loc);
        if (ret) {
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
                goto err;
        }

        stub = fop_mknod_stub (frame, quota_mknod_helper, loc, mode, rdev,
                               umask, xdata);
        if (stub == NULL) {
                goto err;
        }

        local->link_count = 1;
        local->stub = stub;
        local->delta = 0;

        quota_check_limit (frame, loc->parent, this, NULL, NULL);
        return 0;
err:
        QUOTA_STACK_UNWIND (mknod, frame, -1, ENOMEM, NULL, NULL, NULL, NULL,
                            NULL);

        return 0;

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_mknod_cbk: default_mknod_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->mknod, loc,
                    mode, rdev, umask, xdata);

        return 0;

}


int
quota_setxattr_cbk (call_frame_t *frame, void *cookie,
                    xlator_t *this, int op_ret, int op_errno, dict_t *xdata)
{
        QUOTA_STACK_UNWIND (setxattr, frame, op_ret, op_errno, xdata);
        return 0;
}

int
quota_setxattr (call_frame_t *frame, xlator_t *this,
                loc_t *loc, dict_t *dict, int flags, dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
        int                      op_errno       = EINVAL;
        int                      op_ret         = -1;
        int64_t                 *size           = 0;
        uint64_t                 value          = 0;
        quota_inode_ctx_t       *ctx            = NULL;
        int                      ret            = -1;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        VALIDATE_OR_GOTO (frame, err);
        VALIDATE_OR_GOTO (this, err);
        VALIDATE_OR_GOTO (loc, err);

        if (0 <= frame->root->pid)
                GF_IF_INTERNAL_XATTR_GOTO ("trusted.glusterfs.quota*", dict,
                                            op_errno, err);

        ret = dict_get_bin (dict, QUOTA_UPDATE_USAGE_KEY, (void **) &size);
        if (0 == ret) {

                inode_ctx_get (loc->inode, this, &value);
                ctx = (quota_inode_ctx_t *)(unsigned long) value;
                if (NULL == ctx) {
                        gf_log (this->name, GF_LOG_ERROR, "Couldn't get the "
                                "context for %s. Usage may cross the limit.",
                                loc->path);
                        op_ret = -1;
                        goto err;
                }

                LOCK (&ctx->lock);
                {
                        ctx->size = ntoh64 (*size);
                        if (ctx->size < 0)
                                ctx->size = 0;
                }
                UNLOCK (&ctx->lock);

                QUOTA_STACK_UNWIND (setxattr, frame, ret, 0, xdata);
                return 0;
        }

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_setxattr_cbk: default_setxattr_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->setxattr, loc,
                    dict, flags, xdata);
        return 0;
err:
        QUOTA_STACK_UNWIND (setxattr, frame, op_ret, op_errno, NULL);
        return 0;
}

int
quota_fsetxattr_cbk (call_frame_t *frame, void *cookie,
                     xlator_t *this, int op_ret, int op_errno, dict_t *xdata)
{
        QUOTA_STACK_UNWIND (fsetxattr, frame, op_ret, op_errno, xdata);
        return 0;
}

int
quota_fsetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
                 dict_t *dict, int flags, dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
        int32_t         op_ret   = -1;
        int32_t         op_errno = EINVAL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        VALIDATE_OR_GOTO (frame, err);
        VALIDATE_OR_GOTO (this, err);
        VALIDATE_OR_GOTO (fd, err);

        if (0 <= frame->root->pid)
                GF_IF_INTERNAL_XATTR_GOTO ("trusted.glusterfs.quota*", dict,
                                           op_errno, err);

wind:
        STACK_WIND (frame, priv->is_quota_on?
                    quota_fsetxattr_cbk: default_fsetxattr_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->fsetxattr, fd,
                    dict, flags, xdata);
        return 0;
 err:
        QUOTA_STACK_UNWIND (fsetxattr, frame, op_ret, op_errno, NULL);
        return 0;
}


int
quota_removexattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                       int32_t op_ret, int32_t op_errno, dict_t *xdata)
{
        QUOTA_STACK_UNWIND (removexattr, frame, op_ret, op_errno, xdata);
        return 0;
}

int
quota_removexattr (call_frame_t *frame, xlator_t *this,
                   loc_t *loc, const char *name, dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
        int32_t         op_errno = EINVAL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        VALIDATE_OR_GOTO (this, err);

        GF_IF_NATIVE_XATTR_GOTO ("trusted.quota*",
                                 name, op_errno, err);

        VALIDATE_OR_GOTO (frame, err);
        VALIDATE_OR_GOTO (loc, err);

wind:
        STACK_WIND (frame, priv->is_quota_on?
                    quota_removexattr_cbk: default_removexattr_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->removexattr,
                    loc, name, xdata);
        return 0;
err:
        QUOTA_STACK_UNWIND (removexattr, frame, -1,  op_errno, NULL);
        return 0;
}


int
quota_fremovexattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                        int32_t op_ret, int32_t op_errno, dict_t *xdata)
{
        QUOTA_STACK_UNWIND (fremovexattr, frame, op_ret, op_errno, xdata);
        return 0;
}

int
quota_fremovexattr (call_frame_t *frame, xlator_t *this,
                    fd_t *fd, const char *name, dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
        int32_t         op_ret   = -1;
        int32_t         op_errno = EINVAL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        VALIDATE_OR_GOTO (frame, err);
        VALIDATE_OR_GOTO (this, err);
        VALIDATE_OR_GOTO (fd, err);

        GF_IF_NATIVE_XATTR_GOTO ("trusted.quota*",
                                 name, op_errno, err);

wind:
        STACK_WIND (frame, priv->is_quota_on?
                    quota_fremovexattr_cbk: default_fremovexattr_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->fremovexattr,
                    fd, name, xdata);
        return 0;
 err:
        QUOTA_STACK_UNWIND (fremovexattr, frame, op_ret, op_errno, NULL);
        return 0;
}


int32_t
quota_statfs_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                  int32_t op_ret, int32_t op_errno, struct statvfs *buf,
                  dict_t *xdata)
{
	inode_t           *root_inode = NULL;

	root_inode = cookie;

        /* This fop will fail mostly in case of client disconnect's,
         * which is already logged. Hence, not logging here */
        if (op_ret == -1)
                goto unwind;
	/*
	 * We should never get here unless quota_statfs (below) sent us a
	 * cookie, and it would only do so if the value was non-NULL.  This
	 * check is therefore just routine defensive coding.
	 */
	if (!root_inode) {
		gf_log(this->name,GF_LOG_WARNING,
		       "null inode, cannot adjust for quota");
		goto unwind;
	}
	if (!root_inode->table || (root_inode != root_inode->table->root)) {
		gf_log(this->name,GF_LOG_WARNING,
		       "non-root inode, cannot adjust for quota");
		goto unwind;
	}
/*
        inode_ctx_get (root_inode, this, &value);
	if (!value) {
		goto unwind;
	}
        ctx = (quota_inode_ctx_t *)(unsigned long)value;
*/

unwind:
	if (root_inode) {
		inode_unref(root_inode);
	}
        STACK_UNWIND_STRICT (statfs, frame, op_ret, op_errno, buf, xdata);
        return 0;
}


int32_t
quota_statfs (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
{
        quota_priv_t       *priv        = NULL;
	inode_t *root_inode = NULL;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

	if (priv->consider_statfs && loc->inode) {
		root_inode = loc->inode->table->root;
		inode_ref(root_inode);
		STACK_WIND_COOKIE (frame, quota_statfs_cbk, root_inode,
				   FIRST_CHILD(this),
				   FIRST_CHILD(this)->fops->statfs, loc, xdata);
	}
	else {
		/*
		 * We have to make sure that we never get to quota_statfs_cbk
		 * with a cookie that points to something other than an inode,
		 * which is exactly what would happen with STACK_UNWIND using
		 * that as a callback.  Therefore, use default_statfs_cbk in
		 * this case instead.
                 *
                 * Also if the option deem-statfs is not set to "on" don't
                 * bother calculating quota limit on / in statfs_cbk.
		 */
                if (priv->consider_statfs)
                        gf_log(this->name,GF_LOG_WARNING,
                               "missing inode, cannot adjust for quota");
wind:
		STACK_WIND (frame, default_statfs_cbk, FIRST_CHILD(this),
			    FIRST_CHILD(this)->fops->statfs, loc, xdata);
	}
        return 0;
}


int
quota_readdirp_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                    int op_ret, int op_errno, gf_dirent_t *entries,
                    dict_t *xdata)
{
        gf_dirent_t *entry = NULL;

        if (op_ret <= 0)
                goto unwind;

        list_for_each_entry (entry, &entries->list, list) {
                /* TODO: fill things */
        }

unwind:
        STACK_UNWIND_STRICT (readdirp, frame, op_ret, op_errno, entries, xdata);

        return 0;
}
int
quota_readdirp (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
                off_t offset, dict_t *dict)
{
        quota_priv_t       *priv        = NULL;
        int ret = 0;

        priv = this->private;

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        if (dict) {
                ret = dict_set_uint64 (dict, QUOTA_SIZE_KEY, 0);
                if (ret < 0) {
                        goto err;
                }
        }

wind:
        STACK_WIND (frame,
                    priv->is_quota_on? quota_readdirp_cbk: default_readdirp_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->readdirp, fd,
                    size, offset, dict);
        return 0;
err:
        STACK_UNWIND_STRICT (readdirp, frame, -1, EINVAL, NULL, NULL);
        return 0;
}

int32_t
quota_fallocate_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
                    int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
                    struct iatt *postbuf, dict_t *xdata)
{
        int32_t                  ret            = 0;
        uint64_t                 ctx_int        = 0;
        quota_inode_ctx_t       *ctx            = NULL;
        quota_local_t           *local          = NULL;
        quota_dentry_t          *dentry         = NULL;
        int64_t                  delta          = 0;

        local = frame->local;

        if ((op_ret < 0) || (local == NULL)) {
                goto out;
        }

        ret = inode_ctx_get (local->loc.inode, this, &ctx_int);
        if (ret) {
                gf_log (this->name, GF_LOG_WARNING,
                        "%s: failed to get the context", local->loc.path);
                goto out;
        }

        ctx = (quota_inode_ctx_t *)(unsigned long) ctx_int;

        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context not set in %s (gfid:%s)",
                        local->loc.path, uuid_utoa (local->loc.inode->gfid));
                goto out;
        }

        LOCK (&ctx->lock);
        {
                ctx->buf = *postbuf;
        }
        UNLOCK (&ctx->lock);

        list_for_each_entry (dentry, &ctx->parents, next) {
                delta = (postbuf->ia_blocks - prebuf->ia_blocks) * 512;
                quota_update_size (this, local->loc.inode,
                                   dentry->name, dentry->par, delta);
        }

out:
        QUOTA_STACK_UNWIND (fallocate, frame, op_ret, op_errno, prebuf, postbuf,
                            xdata);

        return 0;
}


int32_t
quota_fallocate_helper (call_frame_t *frame, xlator_t *this, fd_t *fd,
                        int32_t mode, off_t offset, size_t len, dict_t *xdata)
{
        quota_local_t *local    = NULL;
        int32_t        op_errno = EINVAL;
        quota_priv_t  *priv     = NULL;

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                goto unwind;
        }

        priv = this->private;

        if (local->op_ret == -1) {
                op_errno = local->op_errno;
                goto unwind;
        }

        STACK_WIND (frame, priv->is_quota_on?
                    quota_fallocate_cbk: default_fallocate_cbk,
                    FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->fallocate, fd, mode, offset, len,
		    xdata);
        return 0;

unwind:
        QUOTA_STACK_UNWIND (fallocate, frame, -1, op_errno, NULL, NULL, NULL);
        return 0;
}


int32_t
quota_fallocate(call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t mode,
		off_t offset, size_t len, dict_t *xdata)
{
        int32_t            ret     = -1, op_errno = EINVAL;
        int32_t            parents = 0;
        quota_local_t     *local   = NULL;
        quota_inode_ctx_t *ctx     = NULL;
        quota_priv_t      *priv    = NULL;
        quota_dentry_t    *dentry  = NULL;
        call_stub_t       *stub    = NULL;

        priv = this->private;
        GF_VALIDATE_OR_GOTO (this->name, priv, unwind);

        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO ("quota", this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, fd, unwind);

        local = quota_local_new ();
        if (local == NULL) {
                goto unwind;
        }

        frame->local = local;
        local->loc.inode = inode_ref (fd->inode);

        ret = quota_inode_ctx_get (fd->inode, -1, -1, this, NULL, NULL, &ctx,
                                   0);
        if (ctx == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "quota context not set in inode (gfid:%s)",
                        uuid_utoa (fd->inode->gfid));
                goto unwind;
        }

        stub = fop_fallocate_stub(frame, quota_fallocate_helper, fd, mode,
                                  offset, len, xdata);
        if (stub == NULL) {
                op_errno = ENOMEM;
                goto unwind;
        }

        priv = this->private;
        GF_VALIDATE_OR_GOTO (this->name, priv, unwind);

        LOCK (&ctx->lock);
        {
                list_for_each_entry (dentry, &ctx->parents, next) {
                        parents++;
                }
        }
        UNLOCK (&ctx->lock);

	/*
	 * Note that by using len as the delta we're assuming the range from
	 * offset to offset+len has not already been allocated. This can result
	 * in ENOSPC errors attempting to allocate an already allocated range.
	 */
        local->delta = len;
        local->stub = stub;
        local->link_count = parents;

        list_for_each_entry (dentry, &ctx->parents, next) {
                quota_check_limit (frame, fd->inode, this, dentry->name,
                                   dentry->par);
        }
        return 0;

unwind:
        QUOTA_STACK_UNWIND (fallocate, frame, -1, op_errno, NULL, NULL, NULL);
        return 0;

wind:
        STACK_WIND (frame, priv->is_quota_on?
                    quota_fallocate_cbk: default_fallocate_cbk,
                    FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->fallocate, fd, mode, offset, len,
		    xdata);
        return 0;
}

/* Logs if
*  i.   Usage crossed soft limit
*  ii.  Usage above soft limit and alert-time timed out
*/
void
quota_log_usage (xlator_t *this, quota_inode_ctx_t *ctx, inode_t *inode,
                 int64_t delta)
{
        struct timeval           cur_time       = {0,};
        char                    *usage_str      = NULL;
        char                    *path           = NULL;
        int64_t                  cur_size       = 0;
        quota_priv_t            *priv           = NULL;

        priv = this->private;
        cur_size = ctx->size + delta;
        gettimeofday (&cur_time, NULL);

        /* Usage crossed/reached soft limit? */
        if (DID_REACH_LIMIT (ctx->soft_lim, ctx->size, cur_size)) {
                ctx->prev_log = cur_time;
                usage_str = gf_uint64_2human_readable (ctx->soft_lim);
                inode_path (inode, NULL, &path);

                gf_log (this->name, GF_LOG_ALERT, "Usage crossed soft limit: "
                        "%s for %s", usage_str,
                        path? path: uuid_utoa (inode->gfid));
        }
        /* Usage crossed/reached hard limit? */
        else if (DID_REACH_LIMIT (ctx->hard_lim, ctx->size, cur_size)) {
                ctx->prev_log = cur_time;
                usage_str = gf_uint64_2human_readable (ctx->hard_lim);
                inode_path (inode, NULL, &path);

                gf_log (this->name, GF_LOG_ALERT, "Usage reached hard limit: "
                        "%s for %s", usage_str,
                        path? path: uuid_utoa (inode->gfid));
        }
        /* Usage is above soft limit and 'alert-time' timed out */
        else if (cur_size > ctx->soft_lim &&
                       quota_timeout (&ctx->prev_log, priv->log_timeout)) {
                ctx->prev_log = cur_time;
                usage_str = gf_uint64_2human_readable (cur_size);
                inode_path (inode, NULL, &path);

                gf_log (this->name, GF_LOG_ALERT, "Usage %s %s limit for %s ",
                        usage_str, (cur_size < ctx->hard_lim)?
                        "is above soft": "has reached hard",
                        path? path: uuid_utoa (inode->gfid));
        }
        if (usage_str)
                GF_FREE (usage_str);
}

int32_t
mem_acct_init (xlator_t *this)
{
        int     ret = -1;

        if (!this)
                return ret;

        ret = xlator_mem_acct_init (this, gf_quota_mt_end + 1);

        if (ret != 0) {
                gf_log (this->name, GF_LOG_WARNING, "Memory accounting"
                        "init failed");
                return ret;
        }

        return ret;
}


int32_t
quota_forget (xlator_t *this, inode_t *inode)
{
        quota_priv_t         *priv    = NULL;
        int32_t               ret     = 0;
        uint64_t              ctx_int = 0;
        quota_inode_ctx_t    *ctx     = NULL;
        quota_dentry_t       *dentry  = NULL, *tmp;

        priv = this->private;
        if (!priv->is_quota_on)
                return 0;

        ret = inode_ctx_del (inode, this, &ctx_int);

        if (ret < 0) {
                return 0;
        }

        ctx = (quota_inode_ctx_t *) (long)ctx_int;

        LOCK (&ctx->lock);
        {
                list_for_each_entry_safe (dentry, tmp, &ctx->parents, next) {
                        __quota_dentry_free (dentry);
                }
        }
        UNLOCK (&ctx->lock);

        LOCK_DESTROY (&ctx->lock);

        GF_FREE (ctx);

        return 0;
}

int32_t
init (xlator_t *this)
{
        int32_t       ret       = -1;
        quota_priv_t *priv      = NULL;

        if ((this->children == NULL)
            || this->children->next) {
                gf_log (this->name, GF_LOG_ERROR,
                        "FATAL: quota (%s) not configured with "
                        "exactly one child", this->name);
                return -1;
        }

        if (this->parents == NULL) {
                gf_log (this->name, GF_LOG_WARNING,
                        "dangling volume. check volfile");
        }

        QUOTA_ALLOC_OR_GOTO (priv, quota_priv_t, err);

        LOCK_INIT (&priv->lock);

        this->private = priv;

        GF_OPTION_INIT ("deem-statfs", priv->consider_statfs, bool, err);
        GF_OPTION_INIT ("server-quota", priv->is_quota_on, bool, err);
        GF_OPTION_INIT ("default-soft-limit", priv->default_soft_lim, percent,
                        err);
        GF_OPTION_INIT ("soft-timeout", priv->soft_timeout, uint32, err);
        GF_OPTION_INIT ("hard-timeout", priv->hard_timeout, uint32, err);
        GF_OPTION_INIT ("volume-uuid", priv->volume_uuid, str, err);

        this->local_pool = mem_pool_new (quota_local_t, 64);
        if (!this->local_pool) {
                ret = -1;
                gf_log (this->name, GF_LOG_ERROR,
                        "failed to create local_t's memory pool");
                goto err;
        }

        priv->rpc_clnt = quota_enforcer_init (this, this->options);
        if (priv->rpc_clnt == NULL) {
                ret = -1;
                gf_log (this->name, GF_LOG_WARNING,
                        "rpc init failed");
        }

        ret = 0;
err:
        return ret;
}

int
reconfigure (xlator_t *this, dict_t *options)
{
        int32_t           ret   = -1;
        quota_priv_t     *priv  = NULL;

        priv = this->private;

        GF_OPTION_RECONF ("deem-statfs", priv->consider_statfs, options, bool,
                          out);
        GF_OPTION_RECONF ("server-quota", priv->is_quota_on, options, bool,
                          out);
        GF_OPTION_RECONF ("default-soft-limit", priv->default_soft_lim,
                          options, percent, out);


        ret = 0;
out:
        return ret;
}


void
fini (xlator_t *this)
{
        return;
}


struct xlator_fops fops = {
        .statfs       = quota_statfs,
        .lookup       = quota_lookup,
        .writev       = quota_writev,
        .create       = quota_create,
        .mkdir        = quota_mkdir,
        .truncate     = quota_truncate,
        .ftruncate    = quota_ftruncate,
        .unlink       = quota_unlink,
        .symlink      = quota_symlink,
        .link         = quota_link,
        .rename       = quota_rename,
        .getxattr     = quota_getxattr,
        .fgetxattr    = quota_fgetxattr,
        .stat         = quota_stat,
        .fstat        = quota_fstat,
        .readlink     = quota_readlink,
        .readv        = quota_readv,
        .fsync        = quota_fsync,
        .setattr      = quota_setattr,
        .fsetattr     = quota_fsetattr,
        .mknod        = quota_mknod,
        .setxattr     = quota_setxattr,
        .fsetxattr    = quota_fsetxattr,
        .removexattr  = quota_removexattr,
        .fremovexattr = quota_fremovexattr,
        .readdirp     = quota_readdirp,
	.fallocate    = quota_fallocate,
};

struct xlator_cbks cbks = {
        .forget = quota_forget
};

struct volume_options options[] = {
        {.key = {"limit-set"}},
        {.key = {"deem-statfs"},
         .type = GF_OPTION_TYPE_BOOL,
         .default_value = "off",
         .description = "If set to on, it takes quota limits into"
                        "consideration while estimating fs size. (df command)"
                        " (Default is off)."
        },
        {.key = {"server-quota"},
         .type = GF_OPTION_TYPE_BOOL,
         .default_value = "off",
         .description = "Skip the quota if xlator if the feature is not "
                        "turned on. This is not a user exposed option."
        },
        {.key = {"default-soft-limit"},
         .type = GF_OPTION_TYPE_PERCENT,
         .default_value = "90%",
         .min = 0,
         .max = LONG_MAX,
        },
        {.key = {"soft-timeout"},
         .type = GF_OPTION_TYPE_INT,
         .min = 0,
         .max = 60,
         .default_value = "0",
         .description = "quota caches the directory sizes on client. "
                        "soft-timeout indicates the timeout for the validity of"
                        " cache before soft-limit has been crossed."
        },
        {.key = {"hard-timeout"},
         .type = GF_OPTION_TYPE_INT,
         .min = 0,
         .max = 60,
         .default_value = "0",
         .description = "quota caches the directory sizes on client. "
                        "hard-timeout indicates the timeout for the validity of"
                        " cache after soft-limit has been crossed."
        },
        { .key   = {"username"},
          .type  = GF_OPTION_TYPE_ANY,
        },
        { .key   = {"password"},
          .type  = GF_OPTION_TYPE_ANY,
        },
        { .key   = {"transport-type"},
          .value = {"tcp", "socket", "ib-verbs", "unix", "ib-sdp",
                    "tcp/client", "ib-verbs/client", "rdma"},
          .type  = GF_OPTION_TYPE_STR,
        },
        { .key   = {"remote-host"},
          .type  = GF_OPTION_TYPE_INTERNET_ADDRESS,
        },
        { .key   = {"remote-port"},
          .type  = GF_OPTION_TYPE_INT,
        },
        { .key  = {"volume-uuid"},
          .type = GF_OPTION_TYPE_STR,
          .description = "uuid of the volume this brick is part of."
        },
        {.key = {NULL}}
};