features/quota: Quota related changes to statfs fop

The code adjusts the statvfs values if the quota-deem-statfs option is enabled.

i.   Adjust statvfs based on limit configured on root.
ii.  If limit is set on the inode passed, use size/limits on that inode to
     populate statvfs. Otherwise, use size/limits configured on root.
iii. Upon statvfs, update the ctx->size on the inode.
iv.  Don't let DHT aggregate, instead take the maximum of the usages from the
     subvols of the DHT, since each of it contains the complete information.

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

2 files changed, 208 insertions, 29 deletions

diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index 4f02d18f..722d891d 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -2891,11 +2891,16 @@ int
 dht_statfs_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 int op_ret, int op_errno, struct statvfs *statvfs, dict_t *xdata)
 {
-        dht_local_t *local         = NULL;
-        int          this_call_cnt = 0;
-        int          bsize         = 0;
-        int          frsize        = 0;
+        dht_local_t *local              = NULL;
+        int          this_call_cnt      = 0;
+        int          bsize              = 0;
+        int          frsize             = 0;
+        int8_t       quota_deem_statfs  = 0;
+        GF_UNUSED int     ret           = 0;
+        unsigned long     new_usage     = 0;
+        unsigned long     cur_usage     = 0;
 
+        ret = dict_get_int8 (xdata, "quota-deem-statfs", &quota_deem_statfs);
 
         local = frame->local;
 
@@ -2905,8 +2910,22 @@ dht_statfs_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                         local->op_errno = op_errno;
                         goto unlock;
                 }
+                if (!statvfs) {
+                        op_errno = EINVAL;
+                        local->op_ret = -1;
+                        goto unlock;
+                }
                 local->op_ret = 0;
 
+                if (quota_deem_statfs) {
+                        new_usage = statvfs->f_blocks - statvfs->f_bfree;
+                        cur_usage = local->statvfs.f_blocks - local->statvfs.f_bfree;
+                        /* We take the maximux of the usage from the subvols */
+                        if (new_usage >= cur_usage)
+                                local->statvfs = *statvfs;
+                        goto unlock;
+                }
+
                 if (local->statvfs.f_bsize != 0) {
                         bsize = max(local->statvfs.f_bsize, statvfs->f_bsize);
                         frsize = max(local->statvfs.f_frsize, statvfs->f_frsize);
@@ -2927,6 +2946,7 @@ dht_statfs_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 local->statvfs.f_flag     = statvfs->f_flag;
                 local->statvfs.f_namemax  = statvfs->f_namemax;
 
+
         }
 unlock:
         UNLOCK (&frame->lock);
diff --git a/xlators/features/quota/src/quota.c b/xlators/features/quota/src/quota.c
index 07e5adb4..c1fd0ee1 100644
--- a/xlators/features/quota/src/quota.c
+++ b/xlators/features/quota/src/quota.c
@@ -512,7 +512,8 @@ err:
 }
 
 int
-quota_validate (call_frame_t *frame, inode_t *inode, xlator_t *this)
+quota_validate (call_frame_t *frame, inode_t *inode, xlator_t *this,
+                fop_lookup_cbk_t cbk_fn)
 {
         quota_local_t     *local = NULL;
         int                ret   = 0;
@@ -562,7 +563,7 @@ quota_validate (call_frame_t *frame, inode_t *inode, xlator_t *this)
         }
 
         ret = quota_enforcer_lookup (frame, this, &local->validate_loc, xdata,
-                                     quota_validate_cbk);
+                                     cbk_fn);
 
         if (ret < 0) {
                 ret = -ENOTCONN;
@@ -664,7 +665,8 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
                         quota_log_usage (this, ctx, _inode, delta);
 
                         if (need_validate) {
-                                ret = quota_validate (frame, _inode, this);
+                                ret = quota_validate (frame, _inode, this,
+                                                      quota_validate_cbk);
                                 if (ret < 0) {
                                         op_errno = -ret;
                                         goto err;
@@ -3241,9 +3243,16 @@ quota_statfs_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                   int32_t op_ret, int32_t op_errno, struct statvfs *buf,
                   dict_t *xdata)
 {
-	inode_t           *root_inode = NULL;
+	inode_t           *inode  = NULL;
+        uint64_t           value  = 0;
+	int64_t            usage  = -1;
+	int64_t            avail  = -1;
+        int64_t            blocks = 0;
+        quota_inode_ctx_t *ctx    = NULL;
+        int                ret    = 0;
+        gf_boolean_t       dict_created = _gf_false;
 
-	root_inode = cookie;
+        inode = cookie;
 
         /* This fop will fail mostly in case of client disconnect's,
          * which is already logged. Hence, not logging here */
@@ -3254,49 +3263,192 @@ quota_statfs_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 	 * cookie, and it would only do so if the value was non-NULL.  This
 	 * check is therefore just routine defensive coding.
 	 */
-	if (!root_inode) {
+	if (!inode) {
 		gf_log(this->name,GF_LOG_WARNING,
 		       "null inode, cannot adjust for quota");
 		goto unwind;
 	}
-	if (!root_inode->table || (root_inode != root_inode->table->root)) {
-		gf_log(this->name,GF_LOG_WARNING,
-		       "non-root inode, cannot adjust for quota");
-		goto unwind;
-	}
-/*
-        inode_ctx_get (root_inode, this, &value);
+
+        inode_ctx_get (inode, this, &value);
 	if (!value) {
 		goto unwind;
 	}
+
+        /* if limit is set on this inode, report statfs based on this inode
+         * else report based on root.
+         */
         ctx = (quota_inode_ctx_t *)(unsigned long)value;
-*/
+        if (ctx->hard_lim <= 0) {
+                inode_ctx_get (inode->table->root, this, &value);
+                ctx = (quota_inode_ctx_t *)(unsigned long) value;
+                if (!ctx)
+                        goto unwind;
+        }
+
+	usage = (ctx->size) / buf->f_bsize;
+
+        if (ctx->hard_lim > 0) {
+                blocks = ctx->hard_lim / buf->f_bsize;
+                buf->f_blocks = blocks;
+
+                avail = buf->f_blocks - usage;
+                avail = (avail >= 0) ? avail : 0;
+
+                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;
+        }
+
+        if (!xdata) {
+                xdata = dict_new ();
+                if (!xdata)
+                        goto unwind;
+                dict_created = _gf_true;
+        }
+
+        ret = dict_set_int8 (xdata, "quota-deem-statfs", 1);
+        if (-1 == ret)
+                gf_log (this->name, GF_LOG_ERROR, "Dict set failed, "
+                        "deem-statfs option may have no effect");
 
 unwind:
-	if (root_inode) {
-		inode_unref(root_inode);
-	}
-        STACK_UNWIND_STRICT (statfs, frame, op_ret, op_errno, buf, xdata);
+        QUOTA_STACK_UNWIND (statfs, frame, op_ret, op_errno, buf, xdata);
+
+        if (dict_created)
+                dict_unref (xdata);
+        return 0;
+}
+
+
+int32_t
+quota_statfs_helper (call_frame_t *frame, xlator_t *this, loc_t *loc,
+                     dict_t *xdata)
+{
+        quota_local_t   *local          = NULL;
+        int              op_errno       = EINVAL;
+
+        GF_VALIDATE_OR_GOTO ("quota", (local = frame->local), err);
+
+        if (-1 == local->op_ret) {
+                op_errno = local->op_errno;
+                goto err;
+        }
+
+        STACK_WIND_COOKIE (frame, quota_statfs_cbk, loc->inode,
+                           FIRST_CHILD(this),
+                           FIRST_CHILD(this)->fops->statfs, loc, xdata);
+        return 0;
+err:
+        QUOTA_STACK_UNWIND (statfs, frame, -1, op_errno, NULL, NULL);
         return 0;
 }
 
+int32_t
+quota_statfs_validate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                           int32_t op_ret, int32_t op_errno, inode_t *inode,
+                           struct iatt *buf, dict_t *xdata,
+                           struct iatt *postparent)
+{
+        quota_local_t     *local      = NULL;
+        int32_t            ret        = 0;
+        quota_inode_ctx_t *ctx        = NULL;
+        int64_t           *size       = 0;
+        uint64_t           value      = 0;
+
+        local = frame->local;
+
+        if (op_ret < 0)
+                goto resume;
+
+        GF_ASSERT (local);
+        GF_ASSERT (frame);
+        GF_VALIDATE_OR_GOTO_WITH_ERROR ("quota", this, resume, op_errno,
+                                        EINVAL);
+        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, xdata, resume, 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 resume;
+        }
+
+        ret = dict_get_bin (xdata, QUOTA_SIZE_KEY, (void **) &size);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "size key not present in dict");
+                op_errno = EINVAL;
+                goto resume;
+        }
+
+        LOCK (&ctx->lock);
+        {
+                ctx->size = ntoh64 (*size);
+                gettimeofday (&ctx->tv, NULL);
+        }
+        UNLOCK (&ctx->lock);
+
+resume:
+        --local->link_count;
+
+        quota_resume_fop_if_validation_done (local);
+        return 0;
+}
 
 int32_t
 quota_statfs (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
 {
-        quota_priv_t       *priv        = NULL;
-	inode_t *root_inode = NULL;
+        quota_local_t   *local  = NULL;
+        int              op_errno       = 0;
+        call_stub_t     *stub           = NULL;
+        quota_priv_t *priv  = NULL;
+        int           ret       = 0;
 
         priv = this->private;
 
         WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
 
 	if (priv->consider_statfs && loc->inode) {
-		root_inode = loc->inode->table->root;
-		inode_ref(root_inode);
-		STACK_WIND_COOKIE (frame, quota_statfs_cbk, root_inode,
-				   FIRST_CHILD(this),
-				   FIRST_CHILD(this)->fops->statfs, loc, xdata);
+                local = quota_local_new ();
+                if (!local) {
+                        op_errno = ENOMEM;
+                        goto err;
+                }
+                frame->local = local;
+
+                local->inode = inode_ref (loc->inode);
+                local->link_count = 1;
+
+                stub = fop_statfs_stub (frame, quota_statfs_helper, loc, xdata);
+                if (!stub) {
+                        op_errno = ENOMEM;
+                        goto err;
+                }
+
+                local->stub = stub;
+
+                ret = quota_validate (frame, local->inode, this,
+                                      quota_statfs_validate_cbk);
+                if (0 > ret) {
+                        op_errno = -ret;
+                        --local->link_count;
+                }
+
+                quota_resume_fop_if_validation_done (local);
 	}
 	else {
 		/*
@@ -3317,6 +3469,13 @@ wind:
 			    FIRST_CHILD(this)->fops->statfs, loc, xdata);
 	}
         return 0;
+
+err:
+        STACK_UNWIND_STRICT (statfs, frame, -1, op_errno, NULL, NULL);
+
+        if (local)
+                quota_local_cleanup (this, local);
+        return 0;
 }
 
 int