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>

1 files changed, 168 insertions, 0 deletions

diff --git a/xlators/cluster/stripe/src/stripe.c b/xlators/cluster/stripe/src/stripe.c
index dadd3fec539..3c56464b8c1 100644
--- a/xlators/cluster/stripe/src/stripe.c
+++ b/xlators/cluster/stripe/src/stripe.c
@@ -3774,6 +3774,173 @@ err:
 
 
 int32_t
+stripe_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         callcnt = 0;
+        stripe_local_t *local = NULL;
+	stripe_local_t *mlocal = NULL;
+        call_frame_t   *prev = NULL;
+	call_frame_t   *mframe = NULL;
+
+        if (!this || !frame || !frame->local || !cookie) {
+                gf_log ("stripe", GF_LOG_DEBUG, "possible NULL deref");
+                goto out;
+        }
+
+        prev  = cookie;
+        local = frame->local;
+	mframe = local->orig_frame;
+	mlocal = mframe->local;
+
+        LOCK(&frame->lock);
+        {
+                callcnt = ++mlocal->call_count;
+
+                if (op_ret == 0) {
+                        mlocal->post_buf = *postbuf;
+                        mlocal->pre_buf = *prebuf;
+
+			mlocal->prebuf_blocks  += prebuf->ia_blocks;
+			mlocal->postbuf_blocks += postbuf->ia_blocks;
+
+			correct_file_size(prebuf, mlocal->fctx, prev);
+			correct_file_size(postbuf, mlocal->fctx, prev);
+
+			if (mlocal->prebuf_size < prebuf->ia_size)
+				mlocal->prebuf_size = prebuf->ia_size;
+			if (mlocal->postbuf_size < postbuf->ia_size)
+				mlocal->postbuf_size = postbuf->ia_size;
+                }
+
+		/* return the first failure */
+		if (mlocal->op_ret == 0) {
+			mlocal->op_ret = op_ret;
+			mlocal->op_errno = op_errno;
+		}
+        }
+        UNLOCK (&frame->lock);
+
+        if ((callcnt == mlocal->wind_count) && mlocal->unwind) {
+		mlocal->pre_buf.ia_size = mlocal->prebuf_size;
+		mlocal->pre_buf.ia_blocks = mlocal->prebuf_blocks;
+		mlocal->post_buf.ia_size = mlocal->postbuf_size;
+		mlocal->post_buf.ia_blocks = mlocal->postbuf_blocks;
+
+                STRIPE_STACK_UNWIND (fallocate, mframe, mlocal->op_ret,
+                                     mlocal->op_errno, &mlocal->pre_buf,
+                                     &mlocal->post_buf, NULL);
+        }
+out:
+	STRIPE_STACK_DESTROY(frame);
+        return 0;
+}
+
+int32_t
+stripe_fallocate(call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t mode,
+		 off_t offset, size_t len, dict_t *xdata)
+{
+        stripe_local_t   *local = NULL;
+        stripe_fd_ctx_t  *fctx = NULL;
+        int32_t           op_errno = 1;
+        int32_t           idx = 0;
+        int32_t           offset_offset = 0;
+        int32_t           remaining_size = 0;
+        off_t             fill_size = 0;
+        uint64_t          stripe_size = 0;
+        uint64_t          tmp_fctx = 0;
+	off_t		  dest_offset = 0;
+	call_frame_t	  *fframe = NULL;
+	stripe_local_t	  *flocal = NULL;
+
+        VALIDATE_OR_GOTO (frame, err);
+        VALIDATE_OR_GOTO (this, err);
+        VALIDATE_OR_GOTO (fd, err);
+        VALIDATE_OR_GOTO (fd->inode, err);
+
+        inode_ctx_get (fd->inode, this, &tmp_fctx);
+        if (!tmp_fctx) {
+                op_errno = EINVAL;
+                goto err;
+        }
+        fctx = (stripe_fd_ctx_t *)(long)tmp_fctx;
+        stripe_size = fctx->stripe_size;
+
+        STRIPE_VALIDATE_FCTX (fctx, err);
+
+        remaining_size = len;
+
+        local = mem_get0 (this->local_pool);
+        if (!local) {
+                op_errno = ENOMEM;
+                goto err;
+        }
+        frame->local = local;
+        local->stripe_size = stripe_size;
+	local->fctx = fctx;
+
+        if (!stripe_size) {
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "Wrong stripe size for the file");
+                op_errno = EINVAL;
+                goto err;
+        }
+
+        while (1) {
+		fframe = copy_frame(frame);
+		flocal = mem_get0(this->local_pool);
+		if (!flocal) {
+			op_errno = ENOMEM;
+			goto err;
+		}
+		flocal->orig_frame = frame;
+		fframe->local = flocal;
+
+		/* send fallocate request to the associated child node */
+                idx = (((offset + offset_offset) /
+                        local->stripe_size) % fctx->stripe_count);
+
+                fill_size = (local->stripe_size -
+                             ((offset + offset_offset) % local->stripe_size));
+                if (fill_size > remaining_size)
+                        fill_size = remaining_size;
+
+                remaining_size -= fill_size;
+
+                local->wind_count++;
+                if (remaining_size == 0)
+                        local->unwind = 1;
+
+		dest_offset = offset + offset_offset;
+		if (fctx->stripe_coalesce)
+			dest_offset = coalesced_offset(dest_offset,
+					local->stripe_size, fctx->stripe_count);
+
+		/*
+		 * TODO: Create a separate handler for coalesce mode that sends a
+		 * single fallocate per-child (since the ranges are linear).
+		 */
+		STACK_WIND(fframe, stripe_fallocate_cbk, fctx->xl_array[idx],
+			   fctx->xl_array[idx]->fops->fallocate, fd, mode,
+			   dest_offset, fill_size, xdata);
+
+                offset_offset += fill_size;
+                if (remaining_size == 0)
+                        break;
+        }
+
+        return 0;
+err:
+	if (fframe)
+		STRIPE_STACK_DESTROY(fframe);
+
+        STRIPE_STACK_UNWIND (fallocate, frame, -1, op_errno, NULL, NULL, NULL);
+        return 0;
+}
+
+
+int32_t
 stripe_release (xlator_t *this, fd_t *fd)
 {
 	return 0;
@@ -5221,6 +5388,7 @@ struct xlator_fops fops = {
         .removexattr    = stripe_removexattr,
         .fremovexattr   = stripe_fremovexattr,
         .readdirp       = stripe_readdirp,
+	.fallocate	= stripe_fallocate,
 };
 
 struct xlator_cbks cbks = {