features/quota: Skip quota when it is disabled

Add quota always in the server graph but make it work only when the quota is
enabled.

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

2 files changed, 366 insertions, 82 deletions

diff --git a/xlators/features/quota/src/quota.c b/xlators/features/quota/src/quota.c
index 4ea54cca..8e0e5208 100644
--- a/xlators/features/quota/src/quota.c
+++ b/xlators/features/quota/src/quota.c
@@ -306,6 +306,7 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
         char                  just_validated = 0;
         uuid_t                trav_uuid      = {0,};
 
+
         GF_VALIDATE_OR_GOTO ("quota", this, out);
         GF_VALIDATE_OR_GOTO (this->name, frame, out);
         GF_VALIDATE_OR_GOTO (this->name, inode, out);
@@ -457,43 +458,6 @@ loc_fill_failed:
 }
 
 
-int32_t
-quota_get_limit_value (inode_t *inode, xlator_t *this, int64_t *n)
-{
-        int32_t       ret        = 0;
-        char         *path       = NULL;
-        limits_t     *limit_node = NULL;
-        quota_priv_t *priv       = NULL;
-
-        if (inode == NULL || n == NULL) {
-                ret = -1;
-                goto out;
-        }
-
-        *n = 0;
-
-        ret = inode_path (inode, NULL, &path);
-        if (ret < 0) {
-                ret = -1;
-                goto out;
-        }
-
-        priv = this->private;
-
-        list_for_each_entry (limit_node, &priv->limit_head, limit_list) {
-                if (strcmp (limit_node->path, path) == 0) {
-                        *n = limit_node->value;
-                        break;
-                }
-        }
-
-out:
-        GF_FREE (path);
-
-        return ret;
-}
-
-
 static int32_t
 __quota_init_inode_ctx (inode_t *inode, int64_t limit, xlator_t *this,
                         dict_t *dict, struct iatt *buf,
@@ -581,7 +545,6 @@ quota_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         quota_priv_t      *priv       = NULL;
 
         local = frame->local;
-
         priv = this->private;
 
         inode_ctx_get (inode, this, &value);
@@ -692,6 +655,9 @@ quota_lookup (call_frame_t *frame, xlator_t *this, loc_t *loc,
 
         priv = this->private;
 
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, lookup, loc,
+                                xattr_req);
+
         list_for_each_entry (limit_node, &priv->limit_head, limit_list) {
                 if (strcmp (limit_node->path, loc->path) == 0) {
                         limit = limit_node->value;
@@ -908,6 +874,11 @@ quota_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
         call_stub_t       *stub    = NULL;
         quota_dentry_t    *dentry  = NULL;
 
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, writev, fd, vector,
+                                count, off, flags, iobref, xdata);
+
         GF_ASSERT (frame);
         GF_VALIDATE_OR_GOTO ("quota", this, unwind);
         GF_VALIDATE_OR_GOTO (this->name, fd, unwind);
@@ -1032,6 +1003,12 @@ quota_mkdir (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
         int32_t        ret            = 0, op_errno = 0;
         quota_local_t *local          = NULL;
         call_stub_t   *stub           = NULL;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, mkdir, loc, mode,
+                                umask, xdata);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -1083,7 +1060,6 @@ quota_mkdir (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
 err:
         QUOTA_STACK_UNWIND (mkdir, frame, -1, op_errno, NULL, NULL, NULL,
                             NULL, NULL);
-
         return 0;
 }
 
@@ -1178,6 +1154,12 @@ quota_create (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
         int32_t            ret            = -1;
         quota_local_t     *local          = NULL;
         call_stub_t       *stub           = NULL;
+        quota_priv_t      *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, create, loc, flags,
+                                mode, umask, fd, xdata);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -1224,7 +1206,6 @@ quota_create (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
 err:
         QUOTA_STACK_UNWIND (create, frame, -1, ENOMEM, NULL, NULL, NULL, NULL,
                             NULL, NULL);
-
         return 0;
 }
 
@@ -1271,6 +1252,12 @@ quota_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
 {
         int32_t        ret = 0;
         quota_local_t *local = NULL;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, unlink, loc, xflag,
+                                xdata);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -1412,6 +1399,12 @@ quota_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
         quota_local_t     *local = NULL;
         call_stub_t       *stub  = NULL;
         quota_inode_ctx_t *ctx = NULL;
+        quota_priv_t      *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, link, oldloc,
+                                newloc, xdata);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -1629,6 +1622,12 @@ quota_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
         quota_local_t     *local          = NULL;
         call_stub_t       *stub           = NULL;
         quota_inode_ctx_t *ctx            = NULL;
+        quota_priv_t      *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, rename, oldloc,
+                                newloc, xdata);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -1797,6 +1796,12 @@ quota_symlink (call_frame_t *frame, xlator_t *this, const char *linkpath,
         int32_t          op_errno = ENOMEM;
         quota_local_t   *local    = NULL;
         call_stub_t     *stub     = NULL;
+        quota_priv_t    *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, symlink, linkpath,
+                                loc, umask, xdata);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -1902,6 +1907,12 @@ quota_truncate (call_frame_t *frame, xlator_t *this, loc_t *loc, off_t offset,
 {
         int32_t          ret   = -1;
         quota_local_t   *local = NULL;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, truncate, loc,
+                                offset, xdata);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -1977,6 +1988,12 @@ quota_ftruncate (call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
                  dict_t *xdata)
 {
         quota_local_t   *local = NULL;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, ftruncate, fd,
+                                offset, xdata);
 
         local = quota_local_new ();
         if (local == NULL)
@@ -2118,6 +2135,11 @@ quota_stat (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
 {
         quota_local_t *local = NULL;
         int32_t        ret   = -1;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, stat, loc, xdata);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -2185,6 +2207,11 @@ int32_t
 quota_fstat (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
 {
         quota_local_t *local = NULL;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, fstat, fd, xdata);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -2250,6 +2277,12 @@ quota_readlink (call_frame_t *frame, xlator_t *this, loc_t *loc, size_t size,
 {
         quota_local_t *local = NULL;
         int32_t        ret   = -1;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, readlink, loc,
+                                size, xdata);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -2320,6 +2353,12 @@ 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_local_t *local = NULL;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, readv, fd, size,
+                                offset, flags, xdata);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -2386,6 +2425,12 @@ quota_fsync (call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t flags,
              dict_t *xdata)
 {
         quota_local_t *local = NULL;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, fsync, fd, flags,
+                                xdata);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -2454,6 +2499,12 @@ quota_setattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
 {
         quota_local_t *local = NULL;
         int32_t        ret   = -1;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, setattr, loc,
+                                stbuf, valid, xdata);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -2469,7 +2520,8 @@ quota_setattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
         }
 
         STACK_WIND (frame, quota_setattr_cbk, FIRST_CHILD (this),
-                    FIRST_CHILD (this)->fops->setattr, loc, stbuf, valid, xdata);
+                    FIRST_CHILD (this)->fops->setattr, loc, stbuf, valid,
+                    xdata);
         return 0;
 
 unwind:
@@ -2523,6 +2575,12 @@ quota_fsetattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
                 struct iatt *stbuf, int32_t valid, dict_t *xdata)
 {
         quota_local_t *local = NULL;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, fsetattr, fd,
+                                stbuf, valid, xdata);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -2534,7 +2592,8 @@ 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);
+                    FIRST_CHILD (this)->fops->fsetattr, fd, stbuf, valid,
+                    xdata);
         return 0;
 
 unwind:
@@ -2632,6 +2691,12 @@ quota_mknod (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
         int32_t            ret            = -1;
         quota_local_t     *local          = NULL;
         call_stub_t       *stub           = NULL;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, mknod, loc, mode,
+                                rdev, umask, xdata);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -2696,6 +2761,12 @@ quota_setxattr (call_frame_t *frame, xlator_t *this,
 {
         int             op_errno = EINVAL;
         int             op_ret   = -1;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, setxattr, loc,
+                                dict, flags, xdata);
 
         VALIDATE_OR_GOTO (frame, err);
         VALIDATE_OR_GOTO (this, err);
@@ -2704,10 +2775,8 @@ 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);
+        STACK_WIND (frame, quota_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);
@@ -2728,6 +2797,12 @@ quota_fsetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
 {
         int32_t         op_ret   = -1;
         int32_t         op_errno = EINVAL;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, fsetxattr, fd,
+                                dict, flags, xdata);
 
         VALIDATE_OR_GOTO (frame, err);
         VALIDATE_OR_GOTO (this, err);
@@ -2736,10 +2811,8 @@ 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);
+        STACK_WIND (frame, quota_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);
@@ -2760,6 +2833,12 @@ quota_removexattr (call_frame_t *frame, xlator_t *this,
                    loc_t *loc, const char *name, dict_t *xdata)
 {
         int32_t         op_errno = EINVAL;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, removexattr, loc,
+                                name, xdata);
 
         VALIDATE_OR_GOTO (this, err);
 
@@ -2769,10 +2848,8 @@ 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,
-                    loc, name, xdata);
+        STACK_WIND (frame, quota_removexattr_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->removexattr, loc, name, xdata);
         return 0;
 err:
         QUOTA_STACK_UNWIND (removexattr, frame, -1,  op_errno, NULL);
@@ -2794,6 +2871,12 @@ quota_fremovexattr (call_frame_t *frame, xlator_t *this,
 {
         int32_t         op_ret   = -1;
         int32_t         op_errno = EINVAL;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, fremovexattr, fd,
+                                name, xdata);
 
         VALIDATE_OR_GOTO (frame, err);
         VALIDATE_OR_GOTO (this, err);
@@ -2802,10 +2885,8 @@ 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,
-                    fd, name, xdata);
+        STACK_WIND (frame, quota_fremovexattr_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->fremovexattr, fd, name, xdata);
         return 0;
  err:
         QUOTA_STACK_UNWIND (fremovexattr, frame, op_ret, op_errno, NULL);
@@ -2854,6 +2935,7 @@ quota_statfs_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 		goto unwind;
 	}
         ctx = (quota_inode_ctx_t *)(unsigned long)value;
+
 	usage = (ctx->size) / buf->f_bsize;
         priv = this->private;
 
@@ -2865,21 +2947,21 @@ quota_statfs_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                                 break;
                         }
 
-			buf->f_blocks = blocks;
-			avail = buf->f_blocks - usage;
-			if (buf->f_bfree > avail) {
-				buf->f_bfree = avail;
-			}
-			/*
-			 * We have to assume that the total assigned quota
-			 * won't cause us to dip into the reserved space,
-			 * because dealing with the overcommitted cases is
-			 * just too hairy (especially when different bricks
-			 * might be using different reserved percentages and
-			 * such).
-			 */
-			buf->f_bavail = buf->f_bfree;
-			break;
+                        buf->f_blocks = blocks;
+                        avail = buf->f_blocks - usage;
+                        if (buf->f_bfree > avail) {
+                                buf->f_bfree = avail;
+                        }
+                        /*
+                         * We have to assume that the total assigned quota
+                         * won't cause us to dip into the reserved space,
+                         * because dealing with the overcommitted cases is
+                         * just too hairy (especially when different bricks
+                         * might be using different reserved percentages and
+                         * such).
+                         */
+                        buf->f_bavail = buf->f_bfree;
+                        break;
                 }
         }
 
@@ -2896,10 +2978,14 @@ int32_t
 quota_statfs (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
 {
 	inode_t *root_inode = NULL;
-        quota_priv_t    *priv = NULL;
+        quota_priv_t       *priv        = NULL;
 
         priv = this->private;
 
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, statfs,
+                                loc, xdata);
+
+
 	if (priv->consider_statfs && loc->inode) {
 		root_inode = loc->inode->table->root;
 		inode_ref(root_inode);
@@ -2952,23 +3038,203 @@ quota_readdirp (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
                 off_t offset, dict_t *dict)
 {
         int ret = 0;
+        quota_priv_t       *priv        = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, readdirp, fd, size,
+                                offset, dict);
 
         if (dict) {
                 ret = dict_set_uint64 (dict, QUOTA_SIZE_KEY, 0);
-                if (ret < 0) {
+                if (ret < 0)
                         goto err;
-                }
         }
 
-        STACK_WIND (frame, quota_readdirp_cbk,
-                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->readdirp,
-                    fd, size, offset, dict);
+        STACK_WIND (frame, quota_readdirp_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->readdirp, fd, size, offset, dict);
         return 0;
 err:
         STACK_UNWIND_STRICT (readdirp, frame, -1, EINVAL, NULL, NULL);
         return 0;
 }
 
+int32_t
+quota_fallocate_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
+                    int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
+                    struct iatt *postbuf, dict_t *xdata)
+{
+        int32_t                  ret            = 0;
+        uint64_t                 ctx_int        = 0;
+        quota_inode_ctx_t       *ctx            = NULL;
+        quota_local_t           *local          = NULL;
+        quota_dentry_t          *dentry         = NULL;
+        int64_t                  delta          = 0;
+
+        local = frame->local;
+
+        if ((op_ret < 0) || (local == NULL)) {
+                goto out;
+        }
+
+        ret = inode_ctx_get (local->loc.inode, this, &ctx_int);
+        if (ret) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "%s: failed to get the context", local->loc.path);
+                goto out;
+        }
+
+        ctx = (quota_inode_ctx_t *)(unsigned long) ctx_int;
+
+        if (ctx == NULL) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "quota context not set in %s (gfid:%s)",
+                        local->loc.path, uuid_utoa (local->loc.inode->gfid));
+                goto out;
+        }
+
+        LOCK (&ctx->lock);
+        {
+                ctx->buf = *postbuf;
+        }
+        UNLOCK (&ctx->lock);
+
+        list_for_each_entry (dentry, &ctx->parents, next) {
+                delta = (postbuf->ia_blocks - prebuf->ia_blocks) * 512;
+                quota_update_size (this, local->loc.inode,
+                                   dentry->name, dentry->par, delta);
+        }
+
+out:
+        QUOTA_STACK_UNWIND (fallocate, frame, op_ret, op_errno, prebuf, postbuf,
+                            xdata);
+
+        return 0;
+}
+
+int32_t
+quota_fallocate_helper(call_frame_t *frame, xlator_t *this, fd_t *fd,
+		       int32_t mode, off_t offset, size_t len, dict_t *xdata)
+{
+        quota_local_t *local    = NULL;
+        int32_t        op_errno = EINVAL;
+
+        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)
+{
+        int32_t            ret     = -1, op_errno = EINVAL;
+        int32_t            parents = 0;
+        quota_local_t     *local   = NULL;
+        quota_inode_ctx_t *ctx     = NULL;
+        quota_priv_t      *priv    = NULL;
+        call_stub_t       *stub    = NULL;
+        quota_dentry_t    *dentry  = NULL;
+
+        priv = this->private;
+
+        QUOTA_WIND_IF_DISABLED (priv->quota_on, frame, this, fallocate, fd,
+                                mode, offset, len, xdata);
+
+        GF_ASSERT (frame);
+        GF_VALIDATE_OR_GOTO ("quota", this, unwind);
+        GF_VALIDATE_OR_GOTO (this->name, fd, unwind);
+
+        local = quota_local_new ();
+        if (local == NULL) {
+                goto unwind;
+        }
+
+        frame->local = local;
+        local->loc.inode = inode_ref (fd->inode);
+
+        ret = quota_inode_ctx_get (fd->inode, -1, this, NULL, NULL, &ctx, 0);
+        if (ctx == NULL) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "quota context not set in inode (gfid:%s)",
+                        uuid_utoa (fd->inode->gfid));
+                goto unwind;
+        }
+
+        stub = fop_fallocate_stub(frame, quota_fallocate_helper, fd, mode, offset, len,
+				  xdata);
+        if (stub == NULL) {
+                op_errno = ENOMEM;
+                goto unwind;
+        }
+
+        priv = this->private;
+        GF_VALIDATE_OR_GOTO (this->name, priv, unwind);
+
+        LOCK (&ctx->lock);
+        {
+                list_for_each_entry (dentry, &ctx->parents, next) {
+                        parents++;
+                }
+        }
+        UNLOCK (&ctx->lock);
+
+	/*
+	 * Note that by using len as the delta we're assuming the range from
+	 * offset to offset+len has not already been allocated. This can result
+	 * in ENOSPC errors attempting to allocate an already allocated range.
+	 */
+        local->delta = len;
+        local->stub = stub;
+        local->link_count = parents;
+
+        list_for_each_entry (dentry, &ctx->parents, next) {
+                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;
+                }
+        }
+        UNLOCK (&local->lock);
+
+        if (stub != NULL) {
+                call_resume (stub);
+        }
+
+        return 0;
+
+unwind:
+        QUOTA_STACK_UNWIND (fallocate, frame, -1, op_errno, NULL, NULL, NULL);
+        return 0;
+}
+
 
 int32_t
 mem_acct_init (xlator_t *this)
@@ -3122,7 +3388,6 @@ init (xlator_t *this)
         INIT_LIST_HEAD (&priv->limit_head);
 
         LOCK_INIT (&priv->lock);
-
         this->private = priv;
 
         ret = quota_parse_limits (priv, this, this->options, NULL);
@@ -3133,6 +3398,7 @@ init (xlator_t *this)
 
         GF_OPTION_INIT ("timeout", priv->timeout, int64, err);
         GF_OPTION_INIT ("deem-statfs", priv->consider_statfs, bool, err);
+        GF_OPTION_INIT ("server-quota", priv->quota_on, bool, err);
 
         this->local_pool = mem_pool_new (quota_local_t, 64);
         if (!this->local_pool) {
@@ -3213,6 +3479,7 @@ reconfigure (xlator_t *this, dict_t *options)
         char              found = 0;
 
         priv = this->private;
+        GF_VALIDATE_OR_GOTO (this->name, priv, out);
 
         INIT_LIST_HEAD (&head);
 
@@ -3263,6 +3530,7 @@ reconfigure (xlator_t *this, dict_t *options)
         GF_OPTION_RECONF ("timeout", priv->timeout, options, int64, out);
         GF_OPTION_RECONF ("deem-statfs", priv->consider_statfs, options, bool,
                           out);
+        GF_OPTION_RECONF ("server-quota", priv->quota_on, options, bool, out);
 
         ret = 0;
 out:
@@ -3327,5 +3595,11 @@ struct volume_options options[] = {
                         "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 = {NULL}}
 };
diff --git a/xlators/features/quota/src/quota.h b/xlators/features/quota/src/quota.h
index 67a195b2..2ba339aa 100644
--- a/xlators/features/quota/src/quota.h
+++ b/xlators/features/quota/src/quota.h
@@ -27,7 +27,17 @@
 #define READDIR_BUF             4096
 #define QUOTA_UPDATE_USAGE_KEY  "quota-update-usage"
 
-#define DID_REACH_LIMIT(lim, prev_size, cur_size)       \
+#define QUOTA_WIND_IF_DISABLED(quota_on, frame, this, fop, params ...)  \
+        do {                                                            \
+                if (!quota_on) {                                        \
+                        STACK_WIND_TAIL (frame, FIRST_CHILD(this),      \
+                                         FIRST_CHILD(this)->fops->fop,  \
+                                         params);                       \
+                        return 0;                                       \
+                }                                                       \
+        } while (0)
+
+#define DID_REACH_LIMIT(lim, prev_size, cur_size)               \
         ((cur_size) >= (lim) && (prev_size) < (lim))
 
 #define QUOTA_SAFE_INCREMENT(lock, var)         \
@@ -154,7 +164,7 @@ typedef struct qd_vols_conf qd_vols_conf_t;
 struct quota_priv {
         int64_t                   timeout;
         double                    default_soft_lim;
-        gf_boolean_t              is_quota_on;
+        gf_boolean_t              quota_on;
         gf_boolean_t              consider_statfs;
         struct list_head          limit_head;
         qd_vols_conf_t          **qd_vols_conf;