features/quota: Improvements to quota

Old implementation
* Client side implementation of quota
    - Not secure
    - Increased traffic in updating the ctx

New Implementation
* 2 stages of quota implementation is done: Soft and hard quota
    Upon reaching soft quota limit on the directory it logs/alerts in the quota
    daemon log (ie DEFAULT_LOG_DIR/quotad.log) and no more writes allowed after
    hard quota limit. After reaching the soft-limit the daemon alerts the
    user/admin repeatively for every 'alert-time', which is configurable.

* Quota is moved to server-side.

    There will be 2 quota xlators
    i. Quota Server
        It takes care of the enforcing the quota and maintains the context
        specific to the brick. Since this doesn't have the complete picture of
        the cluster, cluster wide usage is updated from the quota daemon. This
        updated context is saved and used for the enforcement.

        It updates its context by searching the QUOTA_UPDATE_KEY from the dict
        in the setxattr call, and is updated from nowhere else.

        The quota is always loaded in the server graph and is by passed if the
        feature is not enabled.

        Options specific to quota-server:
        server-quota    - Specifies whether the features is on/off. It is used
                          to by pass the quota if turned off.
        deem-statfs     - If set to on, it takes quota limits into
                          consideration while estimating fs size. (df command)

    ii. Quota Daemon
        This is the new xlator introduced with this patch. Its the
        *gluster client* process with no mount point, started upon enabling
        quota or restarting the volume. This is a single process for all the
        volumes in the cluster. Its volfile stored in
        GLUSTERD_DEFAULT_WORKI_DIR/quotad/quotad.vol.

        It queries for the sizes on all the bricks, aggregates the size and
        sends back the updated size, periodically. The timeout between
        successive updation is configurable and typically/by default more for
        below-soft-quota usage and less for above-soft-quota usage. It
        maintains the timeout inside the limit structure based on the usage;
        below soft limit and above soft limit.

        There will be thread running per volume which iterates through the list
        and decides whether the size to be queried in the current iteration
        based on its timeout. It takes the next iteration time taking the least
        of the timeouts in the list of entries.

        Maintains a separate inode table for each volume in the quotad. In the
        first iteration it builds the table for quota-dirs (dirs on which limit
        is set) and its components.

        Options specific to quotad:
        hard-timeout       - Timeout for updation of usage to the quota-server
                             when the usage is crosses the soft-limit.
        soft-timeout       - Timeout for the updation of usage to the
                             quota-server when the usage is below soft-limit.
        alert-time         - Frequency of logging after the usage reached
                             soft limit.
   Options common to both:
   default-soft-limit - This is used when individual paths are not
                        configured with soft-limit and default value of
                        this option is 90% of the hard-limit.
   limit-set          - String containing all the limits.

   Thus in the current implementation we'll have 2 quota xlators: one in server
   graph and one in trusted client (quota daemon) of which the sole
   purpose will be to aggregate the quota size xattrs from all the bricks and
   send the same to server quota xlator.

* Changes in glusterd and CLI

   A single volfile is created for all the volumes, similar to nfs volfile.
   All files related to quota client (volfile, pid etc) are stored in
   GLUSTERD_DEFAULT_WORK_DIR/quotad/.

   The new pattern of the quota limit stores in
   limit-set = <single-dir-limit>[,<single-dir-limit>]
   single-dir-limit = <abs-path>:<hard-limit>[:<soft-limit-in-percent>]

   It also introduces new options:
   volume quota <VOLNAME> {enable|disable|list [<path> ...]|remove <path>| default-soft-limit <percent>} |
   volume quota <VOLNAME> {limit-usage <path> <size> |soft-limit <path> <percent>} |
   volume quota <VOLNAME> {alert-time|soft-timeout|hard-timeout} {<time>}

Credit:
Raghavendra Bhat        <rabhat@redhat.com>
Varun Shastry           <vshastry@redhat.com>
Shishir Gowda           <sgowda@redhat.com>
Kruthika Dhananjay      <kdhananj@redhat.com>
Brian Foster            <bfoster@redhat.com>
Krishnan Parthasarathi  <kparthas@redhat.com>

Change-Id: I16ec5be0c2faaf42b14034b9ccaf17796adef082
BUG: 969461
Signed-off-by: Varun Shastry <vshastry@redhat.com>

7 files changed, 1771 insertions, 893 deletions

diff --git a/xlators/features/marker/src/marker-quota.c b/xlators/features/marker/src/marker-quota.c
index 6f9af6e1..d1e59509 100644
--- a/xlators/features/marker/src/marker-quota.c
+++ b/xlators/features/marker/src/marker-quota.c
@@ -1234,6 +1234,7 @@ mq_get_parent_inode_local (xlator_t *this, quota_local_t *local)
 {
         int32_t            ret = -1;
         quota_inode_ctx_t *ctx = NULL;
+        inode_contribution_t *contribution = NULL;
 
         GF_VALIDATE_OR_GOTO ("marker", this, out);
         GF_VALIDATE_OR_GOTO ("marker", local, out);
@@ -1263,7 +1264,7 @@ mq_get_parent_inode_local (xlator_t *this, quota_local_t *local)
         ret = mq_inode_ctx_get (local->loc.inode, this, &ctx);
         if (ret < 0) {
                 gf_log_callingfn (this->name, GF_LOG_WARNING,
-                        "inode ctx get failed");
+                                  "inode ctx get failed");
                 goto out;
         }
 
@@ -1277,7 +1278,32 @@ mq_get_parent_inode_local (xlator_t *this, quota_local_t *local)
                 goto out;
         }
 
-        local->contri = (inode_contribution_t *) ctx->contribution_head.next;
+        /* Earlier we used to get the next entry in the list maintained
+           by the context. In a good situation it works. i.e the next
+           parent in the directory hierarchy for this path is obtained.
+
+           But consider the below situation:
+           mount-point: /mnt/point
+           quota enabled directories within mount point: /a, /b, /c
+
+           Now when some file (file1) in the directory /c is written some data,
+           then to update the directories, marker has to get the contribution
+           object for the parent inode, i.e /c.
+           Beefore, it was being done by
+           local->contri = (inode_contribution_t *) ctx->contribution_head.next;
+           It works in the normal situations. But suppose /c is moved to /b.
+           Now /b's contribution object is added to the end of the list of
+           parents that the file file1 within /b/c is maintaining. Now if
+           the file /b/c/file1 is copied to /b/c/new, to update the parent in
+           the order c, b and / we cannot go to the next element in the list,
+           as in this case the next contribution object would be / and /b's
+           contribution will be at the end of the list. So get the proper
+           parent's contribution, by searching the entire list.
+        */
+
+        contribution = mq_get_contribution_node (local->loc.parent, ctx);
+        GF_ASSERT (contribution != NULL);
+        local->contri = contribution;
 
         ret = 0;
 out:
@@ -1521,9 +1547,10 @@ mq_update_parent_size (call_frame_t *frame,
         }
         UNLOCK (&local->contri->lock);
 
-        gf_log (this->name, GF_LOG_DEBUG, "%s %"PRId64 "%"PRId64,
-                local->loc.path, local->ctx->size,
-                local->contri->contribution);
+        gf_log_callingfn (this->name, GF_LOG_DEBUG, "path: %s size: %"PRId64
+                          " contribution:%"PRId64,
+                          local->loc.path, local->ctx->size,
+                          local->contri->contribution);
 
         if (dict == NULL) {
                 op_errno = EINVAL;
@@ -1970,9 +1997,39 @@ mq_initiate_quota_txn (xlator_t *this, loc_t *loc)
                 goto out;
         }
 
+        /* Create the contribution node if its absent. Is it right to
+           assume that if the contribution node is not there, then
+           create one and proceed instead of returning?
+           Reason for this assumption is for hard links. Suppose
+           hard link for a file f1 present in a directory d1 is
+           created in the directory d2 (as f2). Now, since d2's
+           contribution is not there in f1's inode ctx, d2's
+           contribution xattr wont be created and will create problems
+           for quota operations.
+        */
         contribution = mq_get_contribution_node (loc->parent, ctx);
-        if (contribution == NULL)
-                goto out;
+        if (!contribution) {
+                if (strcmp (loc->path, "/"))
+                        gf_log_callingfn (this->name, GF_LOG_TRACE,
+                                          "contribution node for the "
+                                          "path (%s) with parent (%s) "
+                                          "not found", loc->path,
+                                          loc->parent?
+                                          uuid_utoa (loc->parent->gfid):
+                                          NULL);
+                contribution = mq_add_new_contribution_node (this, ctx, loc);
+                if (!contribution) {
+                        if(strcmp (loc->path, "/"))
+                                gf_log_callingfn (this->name, GF_LOG_WARNING,
+                                                  "could not allocate "
+                                                  " contribution node for (%s) "
+                                                  "parent: (%s)", loc->path,
+                                                  loc->parent?
+                                                  uuid_utoa (loc->parent->gfid):
+                                                  NULL);
+                        goto out;
+                }
+        }
 
         /* To improve performance, donot start another transaction
          * if one is already in progress for same inode
@@ -2122,8 +2179,11 @@ mq_inspect_file_xattr (xlator_t *this,
         }
 
         contribution = mq_add_new_contribution_node (this, ctx, loc);
-        if (contribution == NULL)
+        if (contribution == NULL) {
+                gf_log_callingfn (this->name, GF_LOG_WARNING, "cannot allocate "
+                                  "contribution node (path:%s)", loc->path);
                 goto out;
+        }
 
         LOCK (&ctx->lock);
         {
@@ -2235,6 +2295,12 @@ _mq_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this,
         int32_t        ret                = 0;
         char           contri_key [512]   = {0, };
         quota_local_t *local              = NULL;
+        inode_t       *inode              = NULL;
+        dentry_t      *tmp                = NULL;
+        gf_boolean_t  last_dentry         = _gf_true;
+        loc_t         loc                 = {0, };
+        dentry_t      *other_dentry       = NULL;
+        gf_boolean_t  remove              = _gf_false;
 
         local = (quota_local_t *) frame->local;
 
@@ -2245,17 +2311,84 @@ _mq_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this,
 
         frame->local = NULL;
 
-        if (local->hl_count > 1) {
-                GET_CONTRI_KEY (contri_key, local->contri->gfid, ret);
+        GET_CONTRI_KEY (contri_key, local->contri->gfid, ret);
 
-                STACK_WIND (frame, mq_removexattr_cbk, FIRST_CHILD(this),
-                            FIRST_CHILD(this)->fops->removexattr,
-                            &local->loc, contri_key, NULL);
-                ret = 0;
-        } else {
-                mq_removexattr_cbk (frame, NULL, this, 0, 0, NULL);
+        if (!local->loc.inode)
+                inode = inode_grep (local->loc.parent->table, local->loc.parent,
+                                    local->loc.name);
+        else
+                inode = inode_ref (local->loc.inode);
+
+        /* Suppose there are 2 directories dir1 and dir2. Quota limit is set on
+           both the directories. There is a file (f1) in dir1. A hark link is
+           created for that file inside the directory dir2 (say f2). Now one
+           more xattr is set in the inode as a new hard link is created in a
+           separate directory.
+           i.e trusted.glusterfs.quota.<gfid of dir2>.contri=<contribution>
+
+           Now when the hardlink f2 is removed, then the new xattr added (i.e
+           the xattr indicating its contribution to ITS parent directory) should
+           be removed (IFF there is not another hardlink for that file in the
+           same directory).
+
+           To do that upon getting unlink first check whether any other hard
+           links for the same inode exists in the same directory. If so do not
+           do anything and proceed for quota transaction.
+           Otherwise, if the removed entry was the only link for that inode
+           within that directory, then get another dentry for the inode
+           (by traversing the list of dentries for the inode) and using the
+           the dentry's parent and name, send removexattr so that the xattr
+           is removed.
+
+           If it is not done, then if the volume is restarted or the brick
+           process is restarted, then wrong quota usage will be shown for the
+           directory dir2.
+        */
+        if (inode) {
+                tmp = NULL;
+                list_for_each_entry (tmp, &inode->dentry_list, inode_list) {
+                        if (local->loc.parent == tmp->parent) {
+                                if (strcmp (local->loc.name, local->loc.name)) {
+                                        last_dentry = _gf_false;
+                                        break;
+                                }
+                        }
+                }
+                remove = last_dentry;
         }
 
+        if (remove) {
+                if (!other_dentry)    {
+                        list_for_each_entry (tmp, &inode->dentry_list,
+                                             inode_list) {
+                                if (local->loc.parent != tmp->parent) {
+                                        other_dentry = tmp;
+                                        break;
+                                }
+                        }
+                }
+
+                if (!other_dentry)
+                        mq_removexattr_cbk (frame, NULL, this, 0, 0, NULL);
+                else {
+                        loc.parent = inode_ref (other_dentry->parent);
+                        loc.name = gf_strdup (other_dentry->name);
+                        uuid_copy (loc.pargfid , other_dentry->parent->gfid);
+                        loc.inode = inode_ref (inode);
+                        uuid_copy (loc.gfid, inode->gfid);
+                        inode_path (other_dentry->parent, other_dentry->name,
+                                    (char **)&loc.path);
+
+                        STACK_WIND (frame, mq_removexattr_cbk,
+                                    FIRST_CHILD(this),
+                                    FIRST_CHILD(this)->fops->removexattr,
+                                    &loc, contri_key, NULL);
+                }
+        } else
+                mq_removexattr_cbk (frame, NULL, this, 0, 0, NULL);
+
+        ret = 0;
+
         if (strcmp (local->parent_loc.path, "/") != 0) {
                 ret = mq_get_parent_inode_local (this, local);
                 if (ret < 0)
@@ -2266,6 +2399,8 @@ _mq_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this,
 out:
         mq_local_unref (this, local);
 
+        loc_wipe (&loc);
+        inode_unref (inode);
         return 0;
 }
 
@@ -2392,8 +2527,11 @@ mq_reduce_parent_size (xlator_t *this, loc_t *loc, int64_t contri)
                 goto out;
 
         contribution = mq_get_contribution_node (loc->parent, ctx);
-        if (contribution == NULL)
+        if (contribution == NULL) {
+                gf_log_callingfn (this->name, GF_LOG_WARNING, "contribution for"
+                                  " the node %s is NULL", loc->path);
                 goto out;
+        }
 
         local = mq_local_new ();
         if (local == NULL) {
@@ -2412,6 +2550,8 @@ mq_reduce_parent_size (xlator_t *this, loc_t *loc, int64_t contri)
         }
 
         if (local->size == 0) {
+                gf_log_callingfn (this->name, GF_LOG_INFO, "local->size is 0 "
+                                  "path: (%s)", loc->path);
                 ret = 0;
                 goto out;
         }
@@ -2424,8 +2564,12 @@ mq_reduce_parent_size (xlator_t *this, loc_t *loc, int64_t contri)
         local->contri = contribution;
 
         ret = mq_inode_loc_fill (NULL, loc->parent, &local->parent_loc);
-        if (ret < 0)
+        if (ret < 0) {
+                gf_log_callingfn (this->name, GF_LOG_INFO, "building parent loc"
+                                  " failed. (gfid: %s)",
+                                  uuid_utoa (loc->parent->gfid));
                 goto out;
+        }
 
         frame = create_frame (this, this->ctx->pool);
         if (!frame) {
diff --git a/xlators/features/marker/src/marker.c b/xlators/features/marker/src/marker.c
index 3325d8af..cd798b3f 100644
--- a/xlators/features/marker/src/marker.c
+++ b/xlators/features/marker/src/marker.c
@@ -802,7 +802,7 @@ marker_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         priv = this->private;
 
-        if ((priv->feature_enabled & GF_QUOTA) && (local->ia_nlink == 1))
+        if (priv->feature_enabled & GF_QUOTA)
                 mq_reduce_parent_size (this, &local->loc, -1);
 
         if (priv->feature_enabled & GF_XTIME)
diff --git a/xlators/features/quota/src/Makefile.am b/xlators/features/quota/src/Makefile.am
index 9546f427..605c198e 100644
--- a/xlators/features/quota/src/Makefile.am
+++ b/xlators/features/quota/src/Makefile.am
@@ -1,11 +1,15 @@
-xlator_LTLIBRARIES = quota.la
+xlator_LTLIBRARIES = quota.la quotad.la
 xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/features
 
 quota_la_LDFLAGS = -module -avoid-version
+quotad_la_LDFLAGS = -module -avoid-version
 
 quota_la_SOURCES = quota.c
 quota_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la
 
+quotad_la_SOURCES = quotad.c
+quotad_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la
+
 noinst_HEADERS = quota-mem-types.h quota.h
 
 AM_CPPFLAGS = $(GF_CPPFLAGS) -I$(top_srcdir)/libglusterfs/src \
@@ -13,5 +17,5 @@ AM_CPPFLAGS = $(GF_CPPFLAGS) -I$(top_srcdir)/libglusterfs/src \
 
 AM_CFLAGS = -Wall $(GF_CFLAGS)
 
-CLEANFILES = 
+CLEANFILES =
 
diff --git a/xlators/features/quota/src/quota-mem-types.h b/xlators/features/quota/src/quota-mem-types.h
index 3082865d..4831476d 100644
--- a/xlators/features/quota/src/quota-mem-types.h
+++ b/xlators/features/quota/src/quota-mem-types.h
@@ -21,6 +21,8 @@ enum gf_quota_mem_types_ {
         gf_quota_mt_int32_t,
         gf_quota_mt_limits_t,
         gf_quota_mt_quota_dentry_t,
+        gf_quota_mt_quota_limits_level_t,
+        gf_quota_mt_qd_vols_conf_t,
         gf_quota_mt_end
 };
 #endif
diff --git a/xlators/features/quota/src/quota.c b/xlators/features/quota/src/quota.c
index c527e7ca..744748fd 100644
--- a/xlators/features/quota/src/quota.c
+++ b/xlators/features/quota/src/quota.c
@@ -18,6 +18,54 @@ 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)
 {
@@ -182,128 +230,16 @@ out:
         return;
 }
 
-
-int32_t
-quota_validate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
-                    int32_t op_ret, int32_t op_errno, dict_t *dict,
-                    dict_t *xdata)
-{
-        quota_local_t     *local          = NULL;
-        uint32_t           validate_count = 0, link_count = 0;
-        int32_t            ret            = 0;
-        quota_inode_ctx_t *ctx            = NULL;
-        int64_t           *size           = 0;
-        uint64_t           value          = 0;
-        call_stub_t       *stub           = NULL;
-
-        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, dict, 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 (dict, 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:
-        LOCK (&local->lock);
-        {
-                local->op_ret = -1;
-                local->op_errno = op_errno;
-
-                validate_count = --local->validate_count;
-                link_count = local->link_count;
-
-                if ((validate_count == 0) && (link_count == 0)) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
-        }
-        UNLOCK (&local->lock);
-
-        if (stub != NULL) {
-                call_resume (stub);
-        }
-
-        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;
         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, need_unwind = 0;
+        char                  need_unwind = 0;
         int64_t               delta          = 0;
-        call_stub_t          *stub           = NULL;
-        int32_t               validate_count = 0, link_count = 0;
         uint64_t              value          = 0;
-        char                  just_validated = 0;
         uuid_t                trav_uuid      = {0,};
 
         GF_VALIDATE_OR_GOTO ("quota", this, out);
@@ -315,25 +251,11 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
 
         delta = local->delta;
 
-        GF_VALIDATE_OR_GOTO (this->name, local->stub, out);
-
-        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;
-
-                if (just_validated) {
-                        local->validate_count--;
-                }
-        }
-        UNLOCK (&local->lock);
 
         if ( par != NULL ) {
                 uuid_copy (trav_uuid, par);
@@ -343,13 +265,9 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
                 if (ctx != NULL) {
                         LOCK (&ctx->lock);
                         {
-                                if (ctx->limit >= 0) {
-                                        if (!just_validated
-                                            && quota_timeout (&ctx->tv,
-                                                              priv->timeout)) {
-                                                need_validate = 1;
-                                        } else if ((ctx->size + delta)
-                                                   >= ctx->limit) {
+                                if (ctx->hard_lim >= 0) {
+                                        if ((ctx->size + delta)
+                                                   >= ctx->hard_lim) {
                                                 local->op_ret = -1;
                                                 local->op_errno = EDQUOT;
                                                 need_unwind = 1;
@@ -358,10 +276,6 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
                         }
                         UNLOCK (&ctx->lock);
 
-                        if (need_validate) {
-                                goto validate;
-                        }
-
                         if (need_unwind) {
                                 break;
                         }
@@ -387,7 +301,6 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
 
                 inode_unref (_inode);
                 _inode = parent;
-                just_validated = 0;
 
                 if (_inode == NULL) {
                         break;
@@ -398,62 +311,10 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
                 ctx = (quota_inode_ctx_t *)(unsigned long)value;
         } while (1);
 
-        ret = 0;
-
-        if (_inode != NULL) {
-                inode_unref (_inode);
-        }
-
-        LOCK (&local->lock);
-        {
-                validate_count = local->validate_count;
-                link_count = local->link_count;
-                if ((validate_count == 0) && (link_count == 0)) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
-        }
-        UNLOCK (&local->lock);
-
-        if (stub != NULL) {
-                call_resume (stub);
-        }
+        inode_unref (_inode);
 
 out:
-        return ret;
-
-validate:
-        LOCK (&local->lock);
-        {
-                loc_wipe (&local->validate_loc);
-
-                if (just_validated) {
-                        local->validate_count--;
-                }
-
-                local->validate_count++;
-                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));
-                        local->validate_count--;
-                }
-        }
-        UNLOCK (&local->lock);
-
-        if (ret < 0) {
-                goto loc_fill_failed;
-        }
-
-        STACK_WIND (frame, quota_validate_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->getxattr, &local->validate_loc,
-                    QUOTA_SIZE_KEY, NULL);
-
-loc_fill_failed:
-        inode_unref (_inode);
-        return 0;
+        return local->op_ret;
 }
 
 
@@ -482,7 +343,7 @@ quota_get_limit_value (inode_t *inode, xlator_t *this, int64_t *n)
 
         list_for_each_entry (limit_node, &priv->limit_head, limit_list) {
                 if (strcmp (limit_node->path, path) == 0) {
-                        *n = limit_node->value;
+                        *n = limit_node->hard_lim;
                         break;
                 }
         }
@@ -495,55 +356,9 @@ out:
 
 
 static int32_t
-__quota_init_inode_ctx (inode_t *inode, int64_t limit, 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->limit = limit;
-        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;
-}
-
-
-static int32_t
-quota_inode_ctx_get (inode_t *inode, int64_t limit, xlator_t *this,
-                     dict_t *dict, struct iatt *buf, quota_inode_ctx_t **ctx,
-                     char create_if_absent)
+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;
@@ -555,8 +370,8 @@ quota_inode_ctx_get (inode_t *inode, int64_t limit, xlator_t *this,
                 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, limit, this, dict,
-                                                      buf, ctx);
+                        ret = __quota_init_inode_ctx (inode, hard_lim, soft_lim,
+                                                      this, dict, buf, ctx);
                 }
         }
         UNLOCK (&inode->lock);
@@ -575,10 +390,10 @@ quota_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         quota_local_t     *local      = NULL;
         quota_inode_ctx_t *ctx        = NULL;
         quota_dentry_t    *dentry     = NULL;
-        int64_t           *size       = 0;
         uint64_t           value      = 0;
         limits_t          *limit_node = NULL;
         quota_priv_t      *priv       = NULL;
+        int64_t           *size       = NULL;
 
         local = frame->local;
 
@@ -588,8 +403,8 @@ quota_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         ctx = (quota_inode_ctx_t *)(unsigned long)value;
 
         if ((op_ret < 0) || (local == NULL)
-            || (((ctx == NULL) || (ctx->limit == local->limit))
-                && (local->limit < 0) && !((IA_ISREG (buf->ia_type))
+            || (((ctx == NULL) || (ctx->hard_lim == local->hard_lim))
+                && (local->hard_lim < 0) && !((IA_ISREG (buf->ia_type))
                                            || (IA_ISLNK (buf->ia_type))))) {
                 goto unwind;
         }
@@ -606,8 +421,8 @@ quota_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         }
         UNLOCK (&priv->lock);
 
-        ret = quota_inode_ctx_get (local->loc.inode, local->limit, this, dict,
-                                   buf, &ctx, 1);
+        ret = quota_inode_ctx_get (local->loc.inode, local->hard_lim,
+                                   local->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)",
@@ -619,22 +434,18 @@ quota_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         LOCK (&ctx->lock);
         {
-
-                if (dict != NULL) {
-                        ret = dict_get_bin (dict, QUOTA_SIZE_KEY,
-                                            (void **) &size);
-                        if (ret == 0) {
-                                ctx->size = ntoh64 (*size);
-                                gettimeofday (&ctx->tv, NULL);
-                        }
-                }
-
-                if (local->limit != ctx->limit) {
-                        ctx->limit = local->limit;
-                }
+                ctx->hard_lim = local->hard_lim;
+                ctx->soft_lim = local->soft_lim;
 
                 ctx->buf = *buf;
 
+                /* will be useful for quotad to determine whether quota xlator's
+                   context is maintaining the correct size.
+                */
+                size = GF_CALLOC (1, sizeof (*size), gf_quota_mt_int64_t);
+                *size = hton64 (ctx->size);
+                ret = dict_set_bin (dict, "trusted.limit.set", size, 8);
+
                 if (!(IA_ISREG (buf->ia_type) || IA_ISLNK (buf->ia_type))) {
                         goto unlock;
                 }
@@ -683,18 +494,23 @@ 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;
-        int64_t             limit       = -1;
         limits_t           *limit_node  = NULL;
-        gf_boolean_t        dict_newed  = _gf_false;
-        quota_priv_t       *priv        = NULL;
         quota_local_t      *local       = NULL;
+        int64_t             hard_lim    = -1;
+        int64_t             soft_lim    = -1;
 
         priv = this->private;
 
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
+
         list_for_each_entry (limit_node, &priv->limit_head, limit_list) {
                 if (strcmp (limit_node->path, loc->path) == 0) {
-                        limit = limit_node->value;
+                        hard_lim = limit_node->hard_lim;
+                        soft_lim = limit_node->soft_lim;
+                        break;
                 }
         }
 
@@ -710,25 +526,18 @@ quota_lookup (call_frame_t *frame, xlator_t *this, loc_t *loc,
 
         frame->local = local;
 
-        local->limit = limit;
+        local->hard_lim = hard_lim;
+        local->soft_lim = soft_lim;
 
-        if (limit < 0) {
+        if (hard_lim < 0) {
                 goto wind;
         }
 
-        if (xattr_req == NULL) {
-                xattr_req  = dict_new ();
-                dict_newed = _gf_true;
-        }
-
-        ret = dict_set_uint64 (xattr_req, QUOTA_SIZE_KEY, 0);
-        if (ret < 0) {
-                goto err;
-        }
-
 wind:
-        STACK_WIND (frame, quota_lookup_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->lookup, loc, xattr_req);
+        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;
 
@@ -738,10 +547,6 @@ err:
                                     NULL, NULL, NULL, NULL);
         }
 
-        if (dict_newed == _gf_true) {
-                dict_unref (xattr_req);
-        }
-
         return 0;
 }
 
@@ -769,10 +574,12 @@ quota_update_size (xlator_t *this, inode_t *inode, char *name, uuid_t par,
         }
 
         do {
-                if ((ctx != NULL) && (ctx->limit >= 0)) {
+                if ((ctx != NULL) && (ctx->hard_lim >= 0)) {
                         LOCK (&ctx->lock);
                         {
                                 ctx->size += delta;
+                                if (ctx->size < 0)
+                                        ctx->size = 0;
                         }
                         UNLOCK (&ctx->lock);
                 }
@@ -801,6 +608,8 @@ quota_update_size (xlator_t *this, inode_t *inode, char *name, uuid_t par,
                         break;
                 }
 
+                value = 0;
+                ctx = NULL;
                 inode_ctx_get (_inode, this, &value);
                 ctx = (quota_inode_ctx_t *)(unsigned long)value;
         } while (1);
@@ -852,8 +661,8 @@ quota_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         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);
+                quota_update_size (this, local->loc.inode, dentry->name,
+                                   dentry->par, delta);
         }
 
 out:
@@ -865,49 +674,22 @@ out:
 
 
 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;
-
-        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, quota_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_priv_t      *priv    = NULL;
-        call_stub_t       *stub    = NULL;
         quota_dentry_t    *dentry  = 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);
@@ -920,7 +702,8 @@ quota_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
         frame->local = local;
         local->loc.inode = inode_ref (fd->inode);
 
-        ret = quota_inode_ctx_get (fd->inode, -1, this, NULL, NULL, &ctx, 0);
+        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)",
@@ -928,13 +711,6 @@ quota_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
                 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);
 
@@ -948,32 +724,22 @@ quota_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
         UNLOCK (&ctx->lock);
 
         local->delta = size;
-        local->stub = stub;
         local->link_count = parents;
 
         list_for_each_entry (dentry, &ctx->parents, next) {
                 ret = quota_check_limit (frame, fd->inode, this, dentry->name,
                                          dentry->par);
                 if (ret == -1) {
-                        break;
-                }
-        }
-
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                local->link_count = 0;
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
+                        op_errno = EDQUOT;
+                        goto unwind;
                 }
         }
-        UNLOCK (&local->lock);
 
-        if (stub != NULL) {
-                call_resume (stub);
-        }
+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;
 
@@ -996,42 +762,16 @@ quota_mkdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
 
 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;
-
-        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, quota_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) {
@@ -1050,34 +790,19 @@ quota_mkdir (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
                 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;
 
-        quota_check_limit (frame, loc->parent, this, NULL, NULL);
-
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
-
-                local->link_count = 0;
+        ret = quota_check_limit (frame, loc->parent, this, NULL, NULL);
+        if (ret == -1) {
+                op_errno = EDQUOT;
+                goto err;
         }
-        UNLOCK (&local->lock);
 
-        if (stub != NULL) {
-                call_resume (stub);
-        }
+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;
 err:
@@ -1104,7 +829,7 @@ quota_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto unwind;
         }
 
-        ret = quota_inode_ctx_get (inode, -1, this, NULL, buf, &ctx, 1);
+        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)",
@@ -1141,46 +866,21 @@ unwind:
 
 
 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;
-
-        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, quota_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;
-        call_stub_t       *stub           = NULL;
+        int32_t            op_errno       = 0;
+
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
 
         local = quota_local_new ();
         if (local == NULL) {
+                op_errno = ENOMEM;
                 goto err;
         }
 
@@ -1189,41 +889,28 @@ quota_create (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
         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);
-
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                local->link_count = 0;
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
+        ret = quota_check_limit (frame, loc->parent, this, NULL, NULL);
+        if (ret == -1) {
+                op_errno = EDQUOT;
+                goto err;
         }
-        UNLOCK (&local->lock);
 
-        if (stub != NULL) {
-                call_resume (stub);
-        }
+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;
 err:
-        QUOTA_STACK_UNWIND (create, frame, -1, ENOMEM, NULL, NULL, NULL, NULL,
-                            NULL, NULL);
+        QUOTA_STACK_UNWIND (create, frame, -1, op_errno, NULL, NULL, NULL,
+                            NULL, NULL, NULL);
 
         return 0;
 }
@@ -1237,6 +924,8 @@ quota_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         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;
@@ -1258,6 +947,21 @@ quota_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                            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);
@@ -1269,9 +973,14 @@ 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;
@@ -1285,8 +994,11 @@ quota_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
                 goto err;
         }
 
-        STACK_WIND (frame, quota_unlink_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->unlink, loc, xflag, xdata);
+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;
 
@@ -1320,7 +1032,7 @@ quota_link_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         quota_update_size (this, local->loc.parent, NULL, NULL,
                            (buf->ia_blocks * 512));
 
-        ret = quota_inode_ctx_get (inode, -1, this, NULL, NULL, &ctx, 0);
+        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,
@@ -1375,44 +1087,19 @@ out:
 
 
 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;
-
-        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, quota_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;
-        call_stub_t       *stub  = NULL;
         quota_inode_ctx_t *ctx = NULL;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
+
         local = quota_local_new ();
         if (local == NULL) {
                 goto err;
@@ -1426,16 +1113,8 @@ quota_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
                 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;
-
-        ret = quota_inode_ctx_get (oldloc->inode, -1, this, NULL, NULL, &ctx,
-                                   0);
+        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)",
@@ -1446,24 +1125,17 @@ quota_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
 
         local->delta = ctx->buf.ia_blocks * 512;
 
-        quota_check_limit (frame, newloc->parent, this, NULL, NULL);
-
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
-
-                local->link_count = 0;
+        ret = quota_check_limit (frame, newloc->parent, this, NULL, NULL);
+        if (ret == -1) {
+                op_errno = EDQUOT;
+                goto err;
         }
-        UNLOCK (&local->lock);
 
-        if (stub != NULL) {
-                call_resume (stub);
-        }
+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);
 
         ret = 0;
 err:
@@ -1484,11 +1156,11 @@ quota_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                   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;
-        int64_t               size             = 0;
 
         if (op_ret < 0) {
                 goto out;
@@ -1508,8 +1180,10 @@ quota_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         }
 
         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);
+                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)
@@ -1517,8 +1191,8 @@ quota_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        ret = quota_inode_ctx_get (local->oldloc.inode, -1, this, NULL, NULL,
-                                   &ctx, 0);
+        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)",
@@ -1570,7 +1244,8 @@ quota_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                         if (dentry == NULL) {
                                 gf_log (this->name, GF_LOG_WARNING,
                                         "cannot create a new dentry (name:%s) "
-                                        "for inode(gfid:%s)", local->newloc.name,
+                                        "for inode(gfid:%s)",
+                                        local->newloc.name,
                                         uuid_utoa (local->newloc.inode->gfid));
                                 op_ret = -1;
                                 op_errno = ENOMEM;
@@ -1592,44 +1267,18 @@ out:
 
 
 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;
-
-        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, quota_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;
-        call_stub_t       *stub           = NULL;
         quota_inode_ctx_t *ctx            = NULL;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
         local = quota_local_new ();
         if (local == NULL) {
                 goto err;
@@ -1649,19 +1298,10 @@ quota_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
                 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, this, NULL, NULL,
-                                           &ctx, 0);
+                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)",
@@ -1675,24 +1315,17 @@ quota_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
                 local->delta = 0;
         }
 
-        quota_check_limit (frame, newloc->parent, this, NULL, NULL);
-
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
-
-                local->link_count = 0;
+        ret = quota_check_limit (frame, newloc->parent, this, NULL, NULL);
+        if (ret == -1) {
+                op_errno = EDQUOT;
+                goto err;
         }
-        UNLOCK (&local->lock);
 
-        if (stub != NULL) {
-                call_resume (stub);
-        }
+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);
 
         ret = 0;
 err:
@@ -1725,7 +1358,7 @@ quota_symlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         quota_update_size (this, local->loc.parent, NULL, NULL, size);
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 1);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -1760,43 +1393,18 @@ out:
 
 
 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;
-
-        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, quota_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) {
@@ -1811,36 +1419,19 @@ quota_symlink (call_frame_t *frame, xlator_t *this, const char *linkpath,
                 goto err;
         }
 
-        local->link_count = 1;
-
-        stub = fop_symlink_stub (frame, quota_symlink_helper, linkpath, loc,
-                                 umask, xdata);
-        if (stub == NULL) {
-                goto err;
-        }
-
-        local->stub = stub;
         local->delta = strlen (linkpath);
 
-        quota_check_limit (frame, loc->parent, this, NULL, NULL);
-
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
-
-                local->link_count = 0;
-        }
-        UNLOCK (&local->lock);
-
-        if (stub != NULL) {
-                call_resume (stub);
+        ret = quota_check_limit (frame, loc->parent, this, NULL, NULL);
+        if (ret == -1) {
+                op_errno = EDQUOT;
+                goto err;
         }
 
+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;
 
 err:
@@ -1857,8 +1448,8 @@ quota_truncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                     struct iatt *postbuf, dict_t *xdata)
 {
         quota_local_t     *local = NULL;
-        int64_t            delta = 0;
         quota_inode_ctx_t *ctx   = NULL;
+        int64_t            delta = 0;
 
         if (op_ret < 0) {
                 goto out;
@@ -1874,7 +1465,7 @@ quota_truncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         quota_update_size (this, local->loc.inode, NULL, NULL, delta);
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -1900,9 +1491,15 @@ 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;
@@ -1916,8 +1513,11 @@ quota_truncate (call_frame_t *frame, xlator_t *this, loc_t *loc, off_t offset,
                 goto err;
         }
 
-        STACK_WIND (frame, quota_truncate_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->truncate, loc, offset, xdata);
+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:
@@ -1933,8 +1533,8 @@ quota_ftruncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                      struct iatt *postbuf, dict_t *xdata)
 {
         quota_local_t     *local = NULL;
-        int64_t            delta = 0;
         quota_inode_ctx_t *ctx   = NULL;
+        int64_t            delta = 0;
 
         if (op_ret < 0) {
                 goto out;
@@ -1950,7 +1550,7 @@ quota_ftruncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         quota_update_size (this, local->loc.inode, NULL, NULL, delta);
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -1976,8 +1576,13 @@ 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;
@@ -1986,8 +1591,11 @@ quota_ftruncate (call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
 
         local->loc.inode = inode_ref (fd->inode);
 
-        STACK_WIND (frame, quota_ftruncate_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->ftruncate, fd, offset, xdata);
+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:
@@ -2006,14 +1614,23 @@ quota_send_dir_limit_to_cli (call_frame_t *frame, xlator_t *this,
         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->limit);
+        snprintf (dir_limit, 1024, "%"PRId64",%"PRId64, ctx->size,
+                  ctx->hard_lim);
 
+dict_set:
         dict = dict_new ();
         if (dict == NULL) {
                 ret = -1;
@@ -2024,7 +1641,7 @@ quota_send_dir_limit_to_cli (call_frame_t *frame, xlator_t *this,
         if (ret < 0)
                 goto out;
 
-        gf_log (this->name, GF_LOG_INFO, "str = %s", dir_limit);
+        gf_log (this->name, GF_LOG_DEBUG, "str = %s", dir_limit);
 
         QUOTA_STACK_UNWIND (getxattr, frame, 0, 0, dict, NULL);
 
@@ -2076,7 +1693,8 @@ quota_getxattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
 
 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)
+                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;
@@ -2091,7 +1709,7 @@ quota_stat_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_DEBUG,
@@ -2116,9 +1734,15 @@ out:
 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;
@@ -2131,8 +1755,10 @@ quota_stat (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
                 goto unwind;
         }
 
-        STACK_WIND (frame, quota_stat_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->stat, loc, xdata);
+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:
@@ -2159,7 +1785,7 @@ quota_fstat_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -2184,8 +1810,13 @@ out:
 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;
@@ -2195,8 +1826,11 @@ quota_fstat (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
 
         local->loc.inode = inode_ref (fd->inode);
 
-        STACK_WIND (frame, quota_fstat_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->fstat, fd, xdata);
+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:
@@ -2223,7 +1857,7 @@ quota_readlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -2239,7 +1873,8 @@ quota_readlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         UNLOCK (&ctx->lock);
 
 out:
-        QUOTA_STACK_UNWIND (readlink, frame, op_ret, op_errno, path, buf, xdata);
+        QUOTA_STACK_UNWIND (readlink, frame, op_ret, op_errno, path, buf,
+                            xdata);
         return 0;
 }
 
@@ -2248,9 +1883,14 @@ 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;
@@ -2264,8 +1904,11 @@ quota_readlink (call_frame_t *frame, xlator_t *this, loc_t *loc, size_t size,
                 goto unwind;
         }
 
-        STACK_WIND (frame, quota_readlink_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->readlink, loc, size, xdata);
+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:
@@ -2293,7 +1936,7 @@ quota_readv_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -2319,8 +1962,13 @@ 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;
@@ -2330,13 +1978,16 @@ quota_readv (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
 
         local->loc.inode = inode_ref (fd->inode);
 
-        STACK_WIND (frame, quota_readv_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->readv, fd, size, offset, flags,
-                    xdata);
+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);
+        QUOTA_STACK_UNWIND (readv, frame, -1, ENOMEM, NULL, -1, NULL, NULL,
+                            NULL);
         return 0;
 }
 
@@ -2359,7 +2010,7 @@ quota_fsync_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -2385,8 +2036,13 @@ 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;
@@ -2396,8 +2052,11 @@ quota_fsync (call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t flags,
 
         frame->local = local;
 
-        STACK_WIND (frame, quota_fsync_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->fsync, fd, flags, xdata);
+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:
@@ -2425,7 +2084,7 @@ quota_setattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_DEBUG,
@@ -2452,9 +2111,14 @@ 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;
@@ -2468,8 +2132,11 @@ quota_setattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
                 goto unwind;
         }
 
-        STACK_WIND (frame, quota_setattr_cbk, FIRST_CHILD (this),
-                    FIRST_CHILD (this)->fops->setattr, loc, stbuf, valid, xdata);
+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:
@@ -2496,7 +2163,7 @@ quota_fsetattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -2522,8 +2189,14 @@ 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;
@@ -2533,8 +2206,11 @@ quota_fsetattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
 
         local->loc.inode = inode_ref (fd->inode);
 
-        STACK_WIND (frame, quota_fsetattr_cbk, FIRST_CHILD (this),
-                    FIRST_CHILD (this)->fops->fsetattr, fd, stbuf, valid, xdata);
+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:
@@ -2559,7 +2235,7 @@ quota_mknod_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto unwind;
         }
 
-        ret = quota_inode_ctx_get (inode, -1, this, NULL, buf, &ctx, 1);
+        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));
@@ -2595,43 +2271,17 @@ unwind:
 
 
 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;
-
-        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, quota_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;
+        int32_t            op_errno       = 0;
+
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -2646,37 +2296,24 @@ quota_mknod (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
                 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);
-
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                local->link_count = 0;
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
+        ret = quota_check_limit (frame, loc->parent, this, NULL, NULL);
+        if (ret == -1) {
+                op_errno = EDQUOT;
+                goto err;
         }
-        UNLOCK (&local->lock);
 
-        if (stub != NULL) {
-                call_resume (stub);
-        }
+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;
 err:
-        QUOTA_STACK_UNWIND (mknod, frame, -1, ENOMEM, NULL, NULL, NULL, NULL,
+        QUOTA_STACK_UNWIND (mknod, frame, -1, op_errno, NULL, NULL, NULL, NULL,
                             NULL);
 
         return 0;
@@ -2694,8 +2331,17 @@ int
 quota_setxattr (call_frame_t *frame, xlator_t *this,
                 loc_t *loc, dict_t *dict, int flags, dict_t *xdata)
 {
-        int             op_errno = EINVAL;
-        int             op_ret   = -1;
+        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);
@@ -2704,10 +2350,36 @@ quota_setxattr (call_frame_t *frame, xlator_t *this,
         GF_IF_INTERNAL_XATTR_GOTO ("trusted.glusterfs.quota*", dict,
                                    op_errno, err);
 
-        STACK_WIND (frame, quota_setxattr_cbk,
-                    FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->setxattr,
-                    loc, dict, flags, xdata);
+        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);
@@ -2726,9 +2398,14 @@ 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);
@@ -2736,10 +2413,11 @@ quota_fsetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
         GF_IF_INTERNAL_XATTR_GOTO ("trusted.glusterfs.quota*", dict,
                                    op_errno, err);
 
-        STACK_WIND (frame, quota_fsetxattr_cbk,
-                    FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->fsetxattr,
-                    fd, dict, flags, xdata);
+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);
@@ -2759,8 +2437,13 @@ 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*",
@@ -2769,9 +2452,10 @@ quota_removexattr (call_frame_t *frame, xlator_t *this,
         VALIDATE_OR_GOTO (frame, err);
         VALIDATE_OR_GOTO (loc, err);
 
-        STACK_WIND (frame, quota_removexattr_cbk,
-                    FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->removexattr,
+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:
@@ -2792,9 +2476,14 @@ 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);
@@ -2802,9 +2491,10 @@ quota_fremovexattr (call_frame_t *frame, xlator_t *this,
         GF_IF_NATIVE_XATTR_GOTO ("trusted.quota*",
                                  name, op_errno, err);
 
-        STACK_WIND (frame, quota_fremovexattr_cbk,
-                    FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->fremovexattr,
+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:
@@ -2860,7 +2550,7 @@ quota_statfs_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         list_for_each_entry (limit_node, &priv->limit_head, limit_list) {
 		/* Notice that this only works for volume-level quota. */
                 if (strcmp (limit_node->path, "/") == 0) {
-                        blocks = limit_node->value / buf->f_bsize;
+                        blocks = limit_node->hard_lim / buf->f_bsize;
                         if (usage > blocks) {
                                 break;
                         }
@@ -2895,11 +2585,13 @@ unwind:
 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;
-        quota_priv_t    *priv = 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);
@@ -2921,6 +2613,7 @@ quota_statfs (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
                 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);
 	}
@@ -2951,8 +2644,13 @@ 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) {
@@ -2960,9 +2658,11 @@ quota_readdirp (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
                 }
         }
 
-        STACK_WIND (frame, quota_readdirp_cbk,
-                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->readdirp,
-                    fd, size, offset, dict);
+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);
@@ -3023,34 +2723,6 @@ out:
 }
 
 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;
-
-        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, quota_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)
 {
@@ -3059,9 +2731,13 @@ quota_fallocate(call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t mode,
         quota_local_t     *local   = NULL;
         quota_inode_ctx_t *ctx     = NULL;
         quota_priv_t      *priv    = NULL;
-        call_stub_t       *stub    = NULL;
         quota_dentry_t    *dentry  = 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);
@@ -3074,7 +2750,7 @@ quota_fallocate(call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t mode,
         frame->local = local;
         local->loc.inode = inode_ref (fd->inode);
 
-        ret = quota_inode_ctx_get (fd->inode, -1, this, NULL, NULL, &ctx, 0);
+        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)",
@@ -3082,15 +2758,6 @@ quota_fallocate(call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t mode,
                 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);
         {
@@ -3106,33 +2773,22 @@ quota_fallocate(call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t mode,
 	 * 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) {
                 ret = quota_check_limit (frame, fd->inode, this, dentry->name,
                                          dentry->par);
                 if (ret == -1) {
-                        break;
+                        op_errno = EDQUOT;
+                        goto unwind;
                 }
         }
 
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                local->link_count = 0;
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
-        }
-        UNLOCK (&local->lock);
-
-        if (stub != NULL) {
-                call_resume (stub);
-        }
-
+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;
 
 unwind:
@@ -3164,11 +2820,16 @@ mem_acct_init (xlator_t *this)
 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) {
@@ -3203,68 +2864,93 @@ quota_parse_limits (quota_priv_t *priv, xlator_t *this, dict_t *xl_options,
         char         *path      = NULL, *saveptr = NULL;
         uint64_t      value     = 0;
         limits_t     *quota_lim = NULL, *old = NULL;
-        char         *last_colon= NULL;
+        double        soft_l    = 0;
+        char         *limit_dir = NULL;
+        char         *saveptr_dir       = NULL;
+        char         *path_str  = NULL;
 
         ret = dict_get_str (xl_options, "limit-set", &str);
 
-        if (str) {
-                path = strtok_r (str, ",", &saveptr);
+        if (ret) {
+                gf_log (this->name, GF_LOG_INFO, "could not get the limits");
+                /* limit may not be set at all on the volume yet */
+                ret = 0;
+                goto err;
+        }
 
-                while (path) {
-                        last_colon = strrchr (path, ':');
-                        *last_colon = '\0';
-                        str_val = last_colon + 1;
+        path_str = gf_strdup (str);
+        if (!path_str)
+                goto err;
 
-                        ret = gf_string2bytesize (str_val, &value);
-                        if (ret != 0)
-                                goto err;
 
-                        QUOTA_ALLOC_OR_GOTO (quota_lim, limits_t, err);
+        limit_dir = strtok_r (path_str, ",", &saveptr);
 
-                        quota_lim->path = path;
+        while (limit_dir) {
+                QUOTA_ALLOC_OR_GOTO (quota_lim, limits_t, err);
+                saveptr_dir = NULL;
 
-                        quota_lim->value = value;
+                path = strtok_r (limit_dir, ":", &saveptr_dir);
 
-                        gf_log (this->name, GF_LOG_INFO, "%s:%"PRId64,
-                                quota_lim->path, quota_lim->value);
-
-                        if (old_list != NULL) {
-                                list_for_each_entry (old, old_list,
-                                                     limit_list) {
-                                        if (strcmp (old->path, quota_lim->path)
-                                            == 0) {
-                                                uuid_copy (quota_lim->gfid,
-                                                           old->gfid);
-                                                break;
-                                        }
-                                }
-                        }
+                str_val = strtok_r (NULL, ":", &saveptr_dir);
 
-                        LOCK (&priv->lock);
-                        {
-                                list_add_tail (&quota_lim->limit_list,
-                                               &priv->limit_head);
+                ret = gf_string2bytesize (str_val, &value);
+                if (ret != 0)
+                        goto err;
+
+                quota_lim->hard_lim = value;
+
+                str_val = strtok_r (NULL, ",", &saveptr_dir);
+
+                soft_l = priv->default_soft_lim;
+                if (str_val) {
+                        ret = gf_string2percent (str_val, &soft_l);
+                        if (ret)
+                                gf_log (this->name, GF_LOG_WARNING,
+                                        "Failed to convert str to "
+                                        "percent. Using default soft "
+                                        "limit");
+                }
+
+                quota_lim->soft_lim = soft_l;
+
+                quota_lim->path = gf_strdup (path);
+
+                gf_log (this->name, GF_LOG_INFO, "%s:%"PRId64,
+                        quota_lim->path, quota_lim->hard_lim);
+
+                if (old_list != NULL) {
+                        list_for_each_entry (old, old_list,
+                                             limit_list) {
+                                if (strcmp (old->path, quota_lim->path) == 0) {
+                                        uuid_copy (quota_lim->gfid,
+                                                   old->gfid);
+                                        break;
+                                }
                         }
-                        UNLOCK (&priv->lock);
+                }
 
-                        path = strtok_r (NULL, ",", &saveptr);
+                LOCK (&priv->lock);
+                {
+                        list_add_tail (&quota_lim->limit_list,
+                                       &priv->limit_head);
                 }
-        } else {
-                gf_log (this->name, GF_LOG_INFO,
-                        "no \"limit-set\" option provided");
+                UNLOCK (&priv->lock);
+
+                limit_dir = strtok_r (NULL, ",", &saveptr);
         }
 
         LOCK (&priv->lock);
         {
                 list_for_each_entry (quota_lim, &priv->limit_head, limit_list) {
                         gf_log (this->name, GF_LOG_INFO, "%s:%"PRId64,
-                                quota_lim->path, quota_lim->value);
+                                quota_lim->path, quota_lim->hard_lim);
                 }
         }
         UNLOCK (&priv->lock);
 
         ret = 0;
 err:
+        GF_FREE (path_str);
         return ret;
 }
 
@@ -3302,8 +2988,10 @@ init (xlator_t *this)
                 goto err;
         }
 
-        GF_OPTION_INIT ("timeout", priv->timeout, int64, err);
         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);
 
         this->local_pool = mem_pool_new (quota_local_t, 64);
         if (!this->local_pool) {
@@ -3329,8 +3017,8 @@ __quota_reconfigure_inode_ctx (xlator_t *this, inode_t *inode, limits_t *limit)
         GF_VALIDATE_OR_GOTO (this->name, inode, out);
         GF_VALIDATE_OR_GOTO (this->name, limit, out);
 
-        ret = quota_inode_ctx_get (inode, limit->value, this, NULL, NULL, &ctx,
-                                   1);
+        ret = quota_inode_ctx_get (inode, limit->hard_lim, limit->soft_lim,
+                                   this, NULL, NULL, &ctx, 1);
         if ((ret == -1) || (ctx == NULL)) {
                 gf_log (this->name, GF_LOG_WARNING, "cannot create quota "
                         "context in inode(gfid:%s)",
@@ -3340,7 +3028,8 @@ __quota_reconfigure_inode_ctx (xlator_t *this, inode_t *inode, limits_t *limit)
 
         LOCK (&ctx->lock);
         {
-                ctx->limit = limit->value;
+                ctx->hard_lim = limit->hard_lim;
+                ctx->soft_lim = limit->soft_lim;
         }
         UNLOCK (&ctx->lock);
 
@@ -3385,6 +3074,13 @@ reconfigure (xlator_t *this, dict_t *options)
 
         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);
+
         INIT_LIST_HEAD (&head);
 
         LOCK (&priv->lock);
@@ -3421,7 +3117,7 @@ reconfigure (xlator_t *this, dict_t *options)
                         }
 
                         if (!found) {
-                                limit->value = -1;
+                                limit->hard_lim = -1;
                                 __quota_reconfigure (this, top->itable, limit);
                         }
 
@@ -3431,9 +3127,7 @@ reconfigure (xlator_t *this, dict_t *options)
         }
         UNLOCK (&priv->lock);
 
-        GF_OPTION_RECONF ("timeout", priv->timeout, options, int64, out);
-        GF_OPTION_RECONF ("deem-statfs", priv->consider_statfs, options, bool,
-                          out);
+
 
         ret = 0;
 out:
@@ -3484,6 +3178,25 @@ struct xlator_cbks cbks = {
 
 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 = {"timeout"},
          .type = GF_OPTION_TYPE_SIZET,
          .min = 0,
@@ -3492,12 +3205,5 @@ struct volume_options options[] = {
          .description = "quota caches the directory sizes on client. Timeout "
                         "indicates the timeout for the cache to be revalidated."
         },
-        {.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 = {NULL}}
 };
diff --git a/xlators/features/quota/src/quota.h b/xlators/features/quota/src/quota.h
index 84ecbb30..de9f6f16 100644
--- a/xlators/features/quota/src/quota.h
+++ b/xlators/features/quota/src/quota.h
@@ -25,6 +25,14 @@
 #define CONTRIBUTION            "contri"
 #define VAL_LENGTH              8
 #define READDIR_BUF             4096
+#define QUOTA_UPDATE_USAGE_KEY  "quota-update-usage"
+
+#define WIND_IF_QUOTAOFF(is_quota_on, label)     \
+        if (!is_quota_on)                       \
+                goto label;
+
+#define DID_REACH_LIMIT(lim, prev_size, cur_size)               \
+        ((cur_size) >= (lim) && (prev_size) < (lim))
 
 #define QUOTA_SAFE_INCREMENT(lock, var)         \
         do {                                    \
@@ -46,7 +54,7 @@
                                  gf_quota_mt_##type);   \
                 if (!var) {                             \
                         gf_log ("", GF_LOG_ERROR,       \
-                                "out of memory :(");    \
+                                "out of memory");    \
                         ret = -1;                       \
                         goto label;                     \
                 }                                       \
@@ -96,6 +104,8 @@
                         goto label;                             \
         } while (0)
 
+
+
 struct quota_dentry {
         char            *name;
         uuid_t           par;
@@ -105,7 +115,8 @@ typedef struct quota_dentry quota_dentry_t;
 
 struct quota_inode_ctx {
         int64_t          size;
-        int64_t          limit;
+        int64_t          hard_lim;
+        int64_t          soft_lim;
         struct iatt      buf;
         struct list_head parents;
         struct timeval   tv;
@@ -125,27 +136,52 @@ struct quota_local {
         int32_t      op_ret;
         int32_t      op_errno;
         int64_t      size;
-        int64_t      limit;
+        int64_t      hard_lim;
+        int64_t      soft_lim;
         char         just_validated;
         inode_t     *inode;
         call_stub_t *stub;
 };
 typedef struct quota_local quota_local_t;
 
+
+struct qd_vols_conf {
+        char                    *name;
+        inode_table_t           *itable;
+        uint32_t                 log_timeout;
+        gf_boolean_t             threads_status;
+        double                   default_soft_lim;
+        gf_lock_t                lock;
+        loc_t                    root_loc;
+	uint32_t		soft_timeout;
+	uint32_t		hard_timeout;
+	struct list_head	limit_head;
+	call_frame_t		*frame;
+};
+typedef struct qd_vols_conf qd_vols_conf_t;
+
+
 struct quota_priv {
-        int64_t           timeout;
-        gf_boolean_t      consider_statfs;
-        struct list_head  limit_head;
-        gf_lock_t         lock;
+        int64_t                 timeout;
+        double                  default_soft_lim;
+        gf_boolean_t            is_quota_on;
+        gf_boolean_t            consider_statfs;
+        struct list_head        limit_head;
+        qd_vols_conf_t        **qd_vols_conf;
+        gf_lock_t               lock;
 };
 typedef struct quota_priv quota_priv_t;
 
+
 struct limits {
         struct list_head  limit_list;
         char             *path;
-        int64_t           value;
         uuid_t            gfid;
+        int64_t           prev_size;
+        struct timeval    prev_log_tv;
+        int64_t           hard_lim;
+        int64_t           soft_lim;
+	struct timeval	  expire;
+	uint32_t	  timeout;
 };
 typedef struct limits     limits_t;
-
-uint64_t cn = 1;
diff --git a/xlators/features/quota/src/quotad.c b/xlators/features/quota/src/quotad.c
new file mode 100644
index 00000000..a5e51add
--- /dev/null
+++ b/xlators/features/quota/src/quotad.c
@@ -0,0 +1,986 @@
+/*
+   Copyright (c) 2013 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"
+#include "syncop.h"
+#include "libgen.h"
+#include "timer.h"
+
+
+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, uint64_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;
+}
+
+/* Returns itable->root, also creates itable if not present */
+inode_t *
+qd_build_root_inode (xlator_t *this, qd_vols_conf_t *this_vol)
+{
+        if (!this_vol->itable) {
+                this_vol->itable = inode_table_new (0, this);
+                if (!this_vol->itable)
+                        return NULL;
+        }
+
+        return inode_ref (this_vol->itable->root);
+}
+
+int
+qd_resolve_root (xlator_t *subvol, loc_t *root_loc,
+                 struct iatt *iatt, dict_t *dict_req, dict_t **dict_rsp)
+{
+        int     ret             = -1;
+
+        ret = syncop_lookup (subvol, root_loc, dict_req, iatt, dict_rsp, NULL);
+        if (-1 == ret) {
+                gf_log (subvol->name, GF_LOG_ERROR, "Received %s for lookup "
+                         "on / (vol:%s)", strerror (errno), subvol->name);
+        }
+        return ret;
+}
+
+int
+qd_build_root_loc (xlator_t *this, xlator_t *subvol, inode_t *inode, loc_t *loc)
+{
+
+        loc->path = gf_strdup ("/");
+        loc->inode = inode;
+        uuid_clear (loc->gfid);
+        loc->gfid[15] = 1;
+
+        return qd_resolve_root (subvol, loc, NULL, NULL, NULL);
+}
+
+
+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 (0 != ret) {
+                gf_log (this->name, GF_LOG_WARNING, "Memory accounting "
+                        "init failed");
+                return ret;
+        }
+
+        return ret;
+}
+
+/**
+ * Takes the limit string, parse it, fill limits_t struct and insert into
+ * above-soft list.
+ *
+ * Format for limit string
+ * <limit-string> = <limit-on-single-dir>[,<limit-on-single-dir>]*
+ * <limit-on-single-dir> =
+ *                    <abs-path-from-volume-root>:<hard-limit>[:<soft-limit-%>]
+ */
+int
+qd_parse_limits (quota_priv_t *priv, xlator_t *this, char *limit_str,
+                    struct list_head *old_list, qd_vols_conf_t *this_vol)
+{
+        int32_t       ret       = -1;
+        char         *str_val   = NULL;
+        char         *path      = NULL, *saveptr = NULL;
+        uint64_t      value     = 0;
+        limits_t     *quota_lim = NULL;
+        char         *str       = NULL;
+        double        soft_l    = -1;
+        char         *limit_on_dir      = NULL;
+        char         *saveptr_dir       = NULL;
+
+        str = gf_strdup (limit_str);
+
+        if (str) {
+                limit_on_dir = strtok_r (str, ",", &saveptr);
+
+                while (limit_on_dir) {
+                        QUOTA_ALLOC_OR_GOTO (quota_lim, limits_t, err);
+                        gettimeofday (&quota_lim->prev_log_tv, NULL);
+
+                        saveptr_dir = NULL;
+
+                        path = strtok_r (limit_on_dir, ":", &saveptr_dir);
+
+                        str_val = strtok_r (NULL, ":", &saveptr_dir);
+
+                        ret = gf_string2bytesize (str_val, &value);
+                        if (0 != ret)
+                                goto err;
+
+                        quota_lim->hard_lim = value;
+
+                        str_val = strtok_r (NULL, ",", &saveptr_dir);
+
+                        soft_l = this_vol->default_soft_lim;
+                        if (str_val) {
+                                ret = gf_string2percent (str_val, &soft_l);
+                                if (ret)
+                                        gf_log (this->name, GF_LOG_WARNING,
+                                                "Failed to convert str to "
+                                                "percent. Using default soft "
+                                                "limit");
+                        }
+
+                        quota_lim->soft_lim = (int64_t) quota_lim->hard_lim *
+                                                                (soft_l / 100);
+
+                        quota_lim->path = gf_strdup (path);
+                        if (!quota_lim->path) {
+                                gf_log (this->name, GF_LOG_ERROR, "ENOMEM "
+                                        "Received copying the path");
+                                goto err;
+                        }
+
+                        quota_lim->prev_size = -1;
+
+                        gf_log (this->name, GF_LOG_INFO, "%s:%"PRId64":%"PRId32,
+                                quota_lim->path, quota_lim->hard_lim,
+                                (int)quota_lim->soft_lim);
+
+                        /* This is used in the reconfigure path, so not used
+                         * by quotad as of now.
+                        if (NULL != old_list) {
+                                list_for_each_entry (old, old_list,
+                                                     limit_list) {
+                                        if (0 ==
+                                            strcmp (old->path, quota_lim->path)) {
+                                                uuid_copy (quota_lim->gfid,
+                                                           old->gfid);
+                                                break;
+                                        }
+                                }
+                        } */
+
+                        LOCK (&this_vol->lock);
+                        {
+                                list_add_tail (&quota_lim->limit_list,
+                                               &this_vol->limit_head);
+                        }
+                        UNLOCK (&this_vol->lock);
+
+                        limit_on_dir = strtok_r (NULL, ",", &saveptr);
+                }
+        } else {
+                gf_log (this->name, GF_LOG_INFO,
+                        "no \"limit-set\" option provided");
+        }
+
+        ret = 0;
+err:
+        GF_FREE (str);
+        return ret;
+}
+
+
+xlator_t *
+qd_get_subvol (xlator_t *this, qd_vols_conf_t *this_vol)
+{
+        xlator_list_t   *subvol        = NULL;
+
+        for (subvol = this->children; subvol; subvol = subvol->next)
+                if (0 == strcmp (subvol->xlator->name, this_vol->name))
+                        return subvol->xlator;
+
+        return NULL;
+}
+
+/* Logs if
+ *  i.   Usage crossed soft limit
+ *  ii.  Usage above soft limit and alert-time timed out
+ */
+void
+qd_log_usage (xlator_t *this, qd_vols_conf_t *conf, limits_t *entry,
+              int64_t cur_size)
+{
+        struct timeval           cur_time       = {0,};
+        char                    *usage_str      = NULL;
+
+        gettimeofday (&cur_time, NULL);
+
+        /* Usage crossed/reached soft limit? */
+        if (DID_REACH_LIMIT (entry->soft_lim, entry->prev_size, cur_size)) {
+                entry->prev_log_tv = cur_time;
+                usage_str = gf_uint64_2human_readable (entry->soft_lim);
+                gf_log (this->name, GF_LOG_ALERT, "Usage crossed soft limit: "
+                        "%s for %s (vol:%s)", usage_str, entry->path,
+                        conf->name);
+        }
+        /* Usage crossed/reached hard limit? */
+        else if (DID_REACH_LIMIT (entry->hard_lim, entry->prev_size,
+                                                                    cur_size)) {
+                entry->prev_log_tv = cur_time;
+                usage_str = gf_uint64_2human_readable (entry->hard_lim);
+                gf_log (this->name, GF_LOG_ALERT, "Usage reached hard limit: "
+                        "%s for %s (vol:%s)", usage_str, entry->path,
+                        conf->name);
+        }
+        /* Usage is above soft limit and 'alert-time' timed out */
+        else if (cur_size > entry->soft_lim &&
+                       quota_timeout (&entry->prev_log_tv, conf->log_timeout)) {
+                entry->prev_log_tv = cur_time;
+                usage_str = gf_uint64_2human_readable (cur_size);
+                gf_log (this->name, GF_LOG_ALERT, "Usage %s %s limit for %s "
+                        "(vol:%s)", usage_str, (cur_size < entry->hard_lim)?
+                        "is above soft": "has reached hard", entry->path,
+                        conf->name);
+        }
+        if (usage_str)
+                GF_FREE (usage_str);
+}
+
+int
+qd_build_child_loc (loc_t *child, loc_t *parent, char *name)
+{
+        if (!child) {
+                goto err;
+        }
+
+        if (strcmp (parent->path, "/") == 0)
+                gf_asprintf ((char **)&child->path, "/%s", name);
+        else
+                gf_asprintf ((char **)&child->path, "%s/%s", parent->path,
+                             name);
+
+        if (!child->path) {
+                goto err;
+        }
+
+        child->name = strrchr (child->path, '/');
+        if (child->name)
+                child->name++;
+
+        child->parent = inode_ref (parent->inode);
+        if (!child->inode)
+                child->inode = inode_new (parent->inode->table);
+
+        if (!child->inode) {
+                goto err;
+        }
+        if (!uuid_is_null(parent->gfid))
+                uuid_copy (child->pargfid, parent->gfid);
+        else
+                uuid_copy (child->pargfid, parent->inode->gfid);
+
+        return 0;
+err:
+        loc_wipe (child);
+        return -1;
+}
+
+#define QUOTA_IDLE_TIMEOUT 30
+int
+qd_check_enforce (qd_vols_conf_t *conf, dict_t *dict, limits_t *entry,
+                  loc_t *entry_loc, xlator_t *subvol)
+{
+        xlator_t        *this = THIS;
+        int64_t         *size = NULL;
+        int              ret  = -1;
+        int64_t          cur_size       = 0;
+        int64_t          tmp_size       = 0;
+        dict_t          *setxattr_dict  = NULL;
+        int64_t         *quota_xlator_size = NULL;
+
+        ret = dict_get_bin (dict, QUOTA_SIZE_KEY, (void **)&size);
+        if (ret) {
+                gf_log (this->name, GF_LOG_WARNING, "Couldn't get size"
+                        " from the dict (%s)", entry->path);
+                goto out;
+        }
+
+        cur_size = ntoh64 (*size);
+
+        qd_log_usage (this, conf, entry, cur_size);
+
+        /* if size hasn't changed, skip the update and bump the timeout */
+        if (entry->prev_size == cur_size) {
+                /* Because of dht we might be getting the correct aggregated
+                   size. But the quota xlator in the server graph might not
+                   be having the right value. So get the size that quota xlator
+                   is maintaining in its context. If it is different than the
+                   total aggregated size, then update the quota xlator with
+                   the correct size through setxattr.
+                */
+                ret = dict_get_bin (dict, "trusted.limit.set",
+                                    (void **)&quota_xlator_size);
+                if (ret)
+                        gf_log (this->name, GF_LOG_WARNING, "Couldn't get size"
+                                " from the dict (%s)", entry->path);
+                else
+                        tmp_size = ntoh64 (*quota_xlator_size);
+
+                if (entry->prev_size == tmp_size) {
+                        entry->timeout = min((entry->timeout << 1),
+                                             QUOTA_IDLE_TIMEOUT);
+                        goto out;
+                }
+	}
+
+        QUOTA_ALLOC_OR_GOTO (size, int64_t, out);
+        *size = hton64 (cur_size);
+
+        setxattr_dict = dict_new();
+        ret = dict_set_bin (setxattr_dict, QUOTA_UPDATE_USAGE_KEY, size,
+                           sizeof (int64_t));
+        if (-1 == ret) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "Couldn't set dict");
+                goto out;
+        }
+
+
+        /* There is a possibility that, after a setxattr is done,
+           a rename might happen and the resolve will fail again.
+        */
+        ret = syncop_setxattr (subvol, entry_loc, setxattr_dict, 0);
+        if (ret) {
+		if (errno == ESTALE || errno == ENOENT)
+			inode_forget (entry_loc->inode, 0);
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Received ERROR:%s in updating quota value %s "
+                        " (vol:%s). Quota enforcement may not be"
+                        " accurate", strerror (errno), entry->path,
+                         subvol->name);
+                goto out;
+        }
+
+	LOCK(&conf->lock);
+	/* set the timeout based on usage */
+	if (cur_size < entry->soft_lim)
+		entry->timeout = conf->soft_timeout;
+	else
+		entry->timeout = conf->hard_timeout;
+
+	entry->prev_size = cur_size;
+	UNLOCK(&conf->lock);
+
+out:
+        if (setxattr_dict)
+                dict_unref (setxattr_dict);
+
+        return ret;
+}
+
+int
+qd_resolve_handle_error (loc_t *comp_loc, inode_t *inode,
+                         int op_ret, int op_errno)
+{
+
+        if (op_ret) {
+                switch (op_errno) {
+                case ESTALE:
+                case ENOENT:
+			if (comp_loc->inode)
+				inode_forget (comp_loc->inode, 0);
+			break;
+                default:
+                        gf_log ("", GF_LOG_ERROR, "lookup on %s returned %s",
+                                comp_loc->path, strerror(op_errno));
+                }
+        }
+
+        return 0;
+}
+
+int
+qd_resolve_entry (qd_vols_conf_t *conf, xlator_t *subvol, limits_t *entry,
+                   dict_t *dict_req, loc_t *loc, dict_t **dict_rsp,
+                   int force)
+{
+        char            *component      = NULL;
+        char            *next_comp      = NULL;
+        char            *saveptr        = NULL;
+        char            *path           = NULL;
+        int              ret            = -1;
+        struct iatt      iatt           = {0,};
+        inode_t         *inode          = NULL;
+        dict_t          *tmp_dict       = NULL;
+        loc_t            comp_loc       = {0,};
+        loc_t            par_loc        = {0,};
+        int              need_lookup    = 0;
+
+        path = gf_strdup (entry->path);
+
+        loc_copy (&par_loc, &conf->root_loc);
+        for (component = strtok_r (path, "/", &saveptr);
+	     component; component = next_comp) {
+
+		next_comp = strtok_r (NULL, "/", &saveptr);
+
+                inode = inode_grep (conf->itable,
+                                    par_loc.inode,
+                                    component);
+
+                /* No need to forget and unref the inode if revalidate flag
+                   is set (i.e @force). Reason:
+                   Just because we want to revalidate, does not mean
+                   the inode is not valid. For invalidating the inode,
+                   (i.e forgetting it) let lookup be sent, and the decision
+                   should be made depending upon lookup's reply.
+                */
+
+                /* if inode is found, then skip lookup unless last component */
+                if (inode) {
+                        comp_loc.inode = inode;
+                        need_lookup = 0;
+                } else {
+                        need_lookup = 1;
+                }
+
+                qd_build_child_loc (&comp_loc, &par_loc, component);
+                /* Get the xattrs in lookup for the last component */
+                if (!next_comp) {
+                        tmp_dict = dict_req;
+                        need_lookup = 1;
+                }
+
+                if (need_lookup || force) {
+                        ret = syncop_lookup (subvol, &comp_loc, tmp_dict, &iatt,
+                                             dict_rsp, NULL);
+                        if (ret) {
+                                /* invalidate inode got from inode_grep if
+                                 * ESTALE/ENOENT */
+                                qd_resolve_handle_error (&comp_loc, inode,
+                                                         ret, errno);
+                                goto out;
+                        }
+
+                        if (!IA_ISDIR (iatt.ia_type)) {
+			        gf_log (subvol->name, GF_LOG_ERROR,
+                                        "%s is not a directory",
+                                        comp_loc.path);
+			        goto out;
+		        }
+
+                        inode = inode_link (comp_loc.inode, par_loc.inode,
+                                            component, &iatt);
+                        comp_loc.inode = inode;
+                        inode_lookup (comp_loc.inode);
+                        inode_unref (inode);
+                }
+                inode = NULL;
+                loc_wipe (&par_loc);
+                loc_copy (&par_loc, &comp_loc);
+                loc_wipe (&comp_loc);
+	}
+        ret = 0;
+        loc_copy (loc, &par_loc);
+
+out:
+        GF_FREE(path);
+        loc_wipe (&par_loc);
+        if (ret)
+                loc_wipe (&comp_loc);
+        return ret;
+}
+
+int
+qd_handle_entry (qd_vols_conf_t *conf, xlator_t *subvol, limits_t *entry,
+                 dict_t *dict_req, int revalidate)
+{
+        dict_t          *dict_rsp  = NULL;
+        loc_t            entry_loc = {0,};
+        int              ret       = -1;
+
+        if (!strcmp (entry->path, "/")) {
+                ret = qd_resolve_root (subvol, &conf->root_loc, NULL, dict_req,
+                                       &dict_rsp);
+                if (ret) {
+                        gf_log (subvol->name, GF_LOG_ERROR, "lookup on / "
+                                "(%s)", strerror(errno));
+                        goto err;
+                }
+                loc_copy (&entry_loc, &conf->root_loc);
+
+        } else {
+                ret = qd_resolve_entry (conf, subvol, entry, dict_req,
+                                        &entry_loc, &dict_rsp, revalidate);
+                /* if resolve failed, force resolve from "/" once */
+                if (ret) {
+                        if (errno == ENOENT)
+                                entry->timeout = QUOTA_IDLE_TIMEOUT;
+			else
+				gf_log (subvol->name, GF_LOG_ERROR,
+					"Quota check on %s failed ERR:%s",
+					entry->path, strerror (errno));
+                        if (!revalidate) {
+                                ret = qd_handle_entry (conf, subvol, entry,
+                                                       dict_req, 1);
+                                if (ret)
+                                        goto err;
+                        }
+                        goto err;
+                }
+        }
+        ret = qd_check_enforce (conf, dict_rsp, entry, &entry_loc, subvol);
+        if (ret)
+                gf_log (subvol->name, GF_LOG_ERROR,
+                        "Failed to enforce quota on %s", entry_loc.path);
+err:
+        loc_wipe (&entry_loc);
+        if (dict_rsp)
+                dict_unref (dict_rsp);
+        return 0;
+}
+
+int
+qd_iterator (qd_vols_conf_t  *conf, xlator_t *subvol)
+{
+        limits_t        *entry          = NULL;
+        limits_t        *next           = NULL;
+        int32_t          ret;
+        dict_t          *dict_req       = NULL;
+	struct timeval	now;
+	struct timeval	next_expire = {0,};
+
+        GF_VALIDATE_OR_GOTO ("qd-iterator", conf, out);
+        GF_VALIDATE_OR_GOTO ("qd-iterator", subvol, out);
+
+        dict_req = dict_new ();
+        GF_VALIDATE_OR_GOTO ("qd-iterator", dict_req, out);
+        ret = dict_set_uint64 (dict_req, QUOTA_SIZE_KEY, 0);
+        if (ret) {
+                gf_log ("qd-iterator", GF_LOG_ERROR, "dict set failed for "
+                        "QUOTA SIZE key");
+                goto out;
+        }
+
+	gettimeofday(&now, NULL);
+        list_for_each_entry_safe (entry, next, &conf->limit_head, limit_list) {
+		/* just use the same start time across the entire iteration */
+		if (timercmp(&now, &entry->expire, <))
+			goto check_next_expire;
+
+                ret = qd_handle_entry (conf, subvol, entry, dict_req, 0);
+                if (ret) {
+                        gf_log ("qd-iterator", GF_LOG_ERROR, "Failed to check "
+                                "quota limit on %s", entry->path);
+                }
+
+		gettimeofday(&entry->expire, NULL);
+		entry->expire.tv_sec += entry->timeout;
+		gf_log("qd-iterator", GF_LOG_TRACE,
+			"volume %s path %s usage %lu (hard %lu soft %lu) expire in %ds",
+			conf->name, entry->path, entry->prev_size,
+			entry->hard_lim, entry->soft_lim, entry->timeout);
+
+check_next_expire:
+		if (!next_expire.tv_sec ||
+		    timercmp(&entry->expire, &next_expire, <))
+			next_expire = entry->expire;
+        }
+
+	ret = 0;
+	/* return a hint as to when we'll next have something to do */
+	gettimeofday(&now, NULL);
+	if (timercmp(&now, &next_expire, <)) {
+		timersub(&next_expire, &now, &next_expire);
+		ret = next_expire.tv_sec;
+                if (next_expire.tv_usec)
+                        ret++;
+	}
+out:
+        if (dict_req)
+                dict_unref (dict_req);
+
+        return ret;
+}
+
+int
+qd_trigger_periodically (void *args)
+{
+        int                      ret            = -1;
+        xlator_t                *this           = NULL;
+        xlator_t                *subvol         = NULL;
+        inode_t                 *root_inode     = NULL;
+        qd_vols_conf_t          *conf           = NULL;
+
+        this = THIS;
+        conf = args;
+
+        subvol = qd_get_subvol (this, conf);
+        if (!subvol) {
+                gf_log (this->name, GF_LOG_ERROR, "No subvol found");
+                return -1;
+        }
+
+        if (!conf->root_loc.path) {
+                root_inode = qd_build_root_inode (this, conf);
+                if (!root_inode) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "New itable creation failed");
+                        return -1;
+                }
+
+                ret = qd_build_root_loc (this, subvol, root_inode,
+                        &conf->root_loc);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Failed to build root_loc for %s", conf->name);
+                        return -1;
+                }
+        }
+
+	if (list_empty(&conf->limit_head))
+		return QUOTA_IDLE_TIMEOUT;
+
+        ret = qd_iterator (conf, subvol);
+        if (ret < 0)
+                gf_log (this->name, GF_LOG_WARNING,
+                        "Couldn't update the usage, frequent "
+                        "log may lead usage to cross beyond "
+                        "limit");
+
+        return ret;
+}
+
+static void create_iter_task(void *);
+
+int
+qd_trigger_periodically_cbk(int ret, call_frame_t *frame, void *args)
+{
+	qd_vols_conf_t	*conf = args;
+	struct timeval delta = { 0, };
+
+        if (ret < 0)
+                gf_log ("quotad", GF_LOG_ERROR,
+                        "Synctask stopped unexpectedly, trying to restart");
+        else
+		delta.tv_sec = ret;
+
+	conf->frame = frame;
+	gf_timer_call_after(THIS->ctx, delta, create_iter_task, conf);
+
+        return ret;
+}
+
+static void
+create_iter_task(void *data)
+{
+	qd_vols_conf_t *conf = data;
+	int ret;
+
+	ret = synctask_new(THIS->ctx->env, qd_trigger_periodically,
+			   qd_trigger_periodically_cbk, conf->frame,
+			   conf);
+	if (ret < 0)
+		gf_log("quotad", GF_LOG_ERROR,
+		       "Synctask creation failed for %s", conf->name);
+}
+
+int
+qd_start_threads (xlator_t *this, int subvol_idx)
+{
+        quota_priv_t            *priv           = NULL;
+        int                      ret            = 0;
+        qd_vols_conf_t          *this_vol       = NULL;
+
+        priv = this->private;
+
+        this_vol = priv->qd_vols_conf[subvol_idx];
+
+        if (list_empty (&this_vol->limit_head)) {
+                gf_log (this->name, GF_LOG_DEBUG, "No limit is set on "
+                        "volume %s", this_vol->name);
+                /* Dafault ret is 0 */
+                goto err;
+        }
+
+        ret = synctask_new (this->ctx->env,
+                            qd_trigger_periodically,
+                            qd_trigger_periodically_cbk,
+                            NULL, (void *) this_vol);
+        if (-1 == ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Synctask creation "
+                        "failed for %s", this_vol->name);
+                goto err;
+        }
+err:
+        return ret;
+}
+
+int
+qd_reconfigure (xlator_t *this, dict_t *options)
+{
+        /* As of now quotad is restarted upon alteration of volfile */
+        return 0;
+}
+
+void
+qd_fini (xlator_t *this)
+{
+        return;
+}
+
+int32_t
+qd_init (xlator_t *this)
+{
+        int32_t          ret            = -1;
+        quota_priv_t    *priv           = NULL;
+        int              i              = 0;
+        char            *option_str     = NULL;
+        xlator_list_t   *subvol         = NULL;
+        char            *limits         = NULL;
+        int              subvol_cnt     = 0;
+        qd_vols_conf_t  *this_vol       = NULL;
+
+        if (NULL == this->children) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "FATAL: quota (%s) not configured for min of 1 child",
+                        this->name);
+                ret = -1;
+                goto err;
+        }
+
+        QUOTA_ALLOC_OR_GOTO (priv, quota_priv_t, err);
+
+        LOCK_INIT (&priv->lock);
+
+        this->private = priv;
+
+        for (subvol_cnt = 0, subvol = this->children;
+             subvol;
+             subvol_cnt++, subvol = subvol->next);
+
+        priv->qd_vols_conf = GF_CALLOC (sizeof (qd_vols_conf_t *), subvol_cnt,
+                                        gf_quota_mt_qd_vols_conf_t);
+        if (!priv->qd_vols_conf) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to allocate memory");
+                goto err;
+        }
+
+        for (i = 0, subvol = this->children;
+             subvol;
+             subvol = subvol->next, i++) {
+
+                QUOTA_ALLOC_OR_GOTO (priv->qd_vols_conf[i],
+                                     qd_vols_conf_t, err);
+
+                this_vol = priv->qd_vols_conf[i];
+
+                LOCK_INIT (&this_vol->lock);
+                INIT_LIST_HEAD (&this_vol->limit_head);
+
+                this_vol->name = subvol->xlator->name;
+
+                ret = gf_asprintf (&option_str, "%s.default-soft-limit",
+                                   this_vol->name);
+                if (0 > ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "gf_asprintf failed");
+                        goto err;
+                }
+                GF_OPTION_INIT (option_str, this_vol->default_soft_lim, percent,
+                                err);
+                GF_FREE (option_str);
+
+                ret = gf_asprintf (&option_str, "%s.alert-time",
+                                   this_vol->name);
+                if (0 > ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "gf_asprintf failed");
+                        goto err;
+                }
+                GF_OPTION_INIT (option_str, this_vol->log_timeout, time, err);
+                GF_FREE (option_str);
+
+                ret = gf_asprintf (&option_str, "%s.limit-set", this_vol->name);
+                if (0 > ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "gf_asprintf failed");
+                        goto err;
+                }
+                ret = dict_get_str (this->options, option_str, &limits);
+                if (ret < 0) {
+                        gf_log (this->name, GF_LOG_ERROR, "dict get failed or "
+                                "no limits set");
+                        continue;
+                }
+
+                ret = qd_parse_limits (priv, this, limits, NULL, this_vol);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Couldn't parse limits for %s", this_vol->name);
+                        goto err;
+                }
+                GF_FREE (option_str);
+
+                ret = gf_asprintf (&option_str, "%s.soft-timeout",
+                                   this_vol->name);
+                if (0 > ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "gf_asprintf failed");
+                        goto err;
+                }
+                GF_OPTION_INIT (option_str, this_vol->soft_timeout,
+                                time, err);
+                GF_FREE (option_str);
+
+                ret = gf_asprintf (&option_str, "%s.hard-timeout",
+                                   this_vol->name);
+                if (0 > ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "gf_asprintf failed");
+                        goto err;
+                }
+                GF_OPTION_INIT (option_str, this_vol->hard_timeout,
+                                time, err);
+                GF_FREE (option_str);
+        }
+
+        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;
+        }
+
+        ret = 0;
+err:
+        /* Free all allocated variables */
+        if (ret) {
+                /* can reach here from GF_OPTION_INIT, so cleaning opt_str */
+                GF_FREE (option_str);
+
+                for (i = 0; i < subvol_cnt; i++)
+                        GF_FREE (priv->qd_vols_conf[i]);
+                GF_FREE (priv->qd_vols_conf);
+
+                GF_FREE (priv);
+        }
+        return ret;
+}
+
+int
+qd_notify (xlator_t *this, int event, void *data, ...)
+{
+        xlator_list_t   *subvol                 = NULL;
+        xlator_t        *subvol_rec             = NULL;
+        quota_priv_t    *priv                   = NULL;
+        int              i                      = 0;
+        int              ret                    = 0;
+
+        subvol_rec = data;
+        priv = this->private;
+
+        for (i=0, subvol = this->children; subvol; i++, subvol = subvol->next) {
+                if (! strcmp (priv->qd_vols_conf[i]->name, subvol_rec->name))
+                        break;
+        }
+        if (!subvol) {
+                default_notify (this, event, data);
+                goto out;
+        }
+
+        switch (event) {
+        case GF_EVENT_CHILD_UP:
+        {
+                /* handle spurious CHILD_UP and DOWN events */
+                if (!priv->qd_vols_conf[i]->threads_status) {
+                        ret = qd_start_threads (this, i);
+                        if (-1 == ret) {
+                                gf_log (this->name, GF_LOG_ERROR, "Couldn't "
+                                        "start the threads for volumes");
+                                goto out;
+                        }
+
+                        priv->qd_vols_conf[i]->threads_status = _gf_true;
+                }
+                break;
+        }
+        case GF_EVENT_CHILD_DOWN:
+        {
+                gf_log (this->name, GF_LOG_ERROR, "vol %s is down.",
+                        priv->qd_vols_conf [i]->name);
+                break;
+        }
+        default:
+                default_notify (this, event, data);
+        }/* end switch */
+
+
+
+out:
+        return ret;
+}
+
+class_methods_t class_methods = {
+        .init           = qd_init,
+        .fini           = qd_fini,
+        .reconfigure    = qd_reconfigure,
+        .notify         = qd_notify,
+};
+
+struct xlator_fops fops = {
+};
+
+struct xlator_cbks cbks = {
+};
+
+struct volume_options options[] = {
+        {.key = {"*.limit-set"}},
+        {.key = {"*.soft-timeout"},
+         .type = GF_OPTION_TYPE_TIME,
+         .min = 1,
+         .max = LONG_MAX,
+         .default_value = "10",
+         .description = ""
+        },
+        {.key = {"*.hard-timeout"},
+         .type = GF_OPTION_TYPE_TIME,
+         .min = 0,
+         .max = LONG_MAX,
+         .default_value = "2",
+         .description = ""
+        },
+        {.key = {"*.alert-time"},
+         .type = GF_OPTION_TYPE_TIME,
+         .min = 0,
+         .max = LONG_MAX,
+         /* default weekly (7 * 24 * 60 *60) */
+         .default_value = "604800",
+         .description = ""
+        },
+        {.key = {"*.default-soft-limit"},
+         .type = GF_OPTION_TYPE_PERCENT,
+         .min = 0,
+         .max = 100,
+         .default_value = "90%",
+         .description = "Takes this if individual paths are not configured "
+                        "with soft limits."
+        },
+        {.key = {NULL}}
+};