gluster: add fallocate fop support

Implement support for the fallocate file operation. fallocate
allocates blocks for a particular inode such that future writes
to the associated region of the file are guaranteed not to fail
with ENOSPC.

This patch adds fallocate support to the following areas:

	- libglusterfs
	- mount/fuse
	- io-stats
	- performance/md-cache,open-behind
	- quota
	- cluster/afr,dht,stripe
	- rpc/xdr
	- protocol/client,server
	- io-threads
	- marker
	- storage/posix
	- libgfapi

BUG: 949242
Change-Id: Ice8e61351f9d6115c5df68768bc844abbf0ce8bd
Signed-off-by: Brian Foster <bfoster@redhat.com>
Reviewed-on: http://review.gluster.org/4969
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Anand Avati <avati@redhat.com>

2 files changed, 241 insertions, 0 deletions

diff --git a/xlators/features/marker/src/marker.c b/xlators/features/marker/src/marker.c
index 82d9066d515..2dc03f93195 100644
--- a/xlators/features/marker/src/marker.c
+++ b/xlators/features/marker/src/marker.c
@@ -1808,6 +1808,74 @@ err:
 }
 
 
+int32_t
+marker_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)
+{
+        marker_local_t     *local   = NULL;
+        marker_conf_t      *priv    = NULL;
+
+        if (op_ret == -1) {
+                gf_log (this->name, GF_LOG_TRACE, "%s occurred while "
+                        "fallocating a file ", strerror (op_errno));
+        }
+
+        local = (marker_local_t *) frame->local;
+
+        frame->local = NULL;
+
+        STACK_UNWIND_STRICT (fallocate, frame, op_ret, op_errno, prebuf,
+                             postbuf, xdata);
+
+        if (op_ret == -1 || local == NULL)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled & GF_QUOTA)
+                mq_initiate_quota_txn (this, &local->loc);
+
+        if (priv->feature_enabled & GF_XTIME)
+                marker_xtime_update_marks (this, local);
+out:
+        marker_local_unref (local);
+
+        return 0;
+}
+
+int32_t
+marker_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   = 0;
+        marker_local_t  *local = NULL;
+        marker_conf_t   *priv  = NULL;
+
+        priv = this->private;
+
+        if (priv->feature_enabled == 0)
+                goto wind;
+
+        local = mem_get0 (this->local_pool);
+
+        MARKER_INIT_LOCAL (frame, local);
+
+        ret = marker_inode_loc_fill (fd->inode, &local->loc);
+
+        if (ret == -1)
+                goto err;
+wind:
+        STACK_WIND (frame, marker_fallocate_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->fallocate, fd, mode, offset, len,
+		    xdata);
+        return 0;
+err:
+        STACK_UNWIND_STRICT (fallocate, frame, -1, ENOMEM, NULL, NULL, NULL);
+
+        return 0;
+}
+
 /* when a call from the special client is received on
  * key trusted.glusterfs.volume-mark with value "RESET"
  * or if the value is 0length, update the change the
@@ -2617,6 +2685,7 @@ struct xlator_fops fops = {
         .removexattr = marker_removexattr,
         .getxattr    = marker_getxattr,
         .readdirp    = marker_readdirp,
+	.fallocate   = marker_fallocate,
 };
 
 struct xlator_cbks cbks = {
diff --git a/xlators/features/quota/src/quota.c b/xlators/features/quota/src/quota.c
index 4ea54cca896..c527e7ca71c 100644
--- a/xlators/features/quota/src/quota.c
+++ b/xlators/features/quota/src/quota.c
@@ -2969,6 +2969,177 @@ err:
         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;
+
+        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)
@@ -3304,6 +3475,7 @@ struct xlator_fops fops = {
         .removexattr  = quota_removexattr,
         .fremovexattr = quota_fremovexattr,
         .readdirp     = quota_readdirp,
+	.fallocate    = quota_fallocate,
 };
 
 struct xlator_cbks cbks = {