cluster/afr: Make the self-heal algorithm pluggable.

Abstract the read/write loop part of data self-heal. This
patch has support for the "full" (i.e., read and write entire
file) algorithm.

Signed-off-by: Anand V. Avati <avati@dev.gluster.com>

1 files changed, 233 insertions, 0 deletions

diff --git a/xlators/cluster/afr/src/afr-self-heal-algorithm.c b/xlators/cluster/afr/src/afr-self-heal-algorithm.c
new file mode 100644
index 00000000000..bc3917caca0
--- /dev/null
+++ b/xlators/cluster/afr/src/afr-self-heal-algorithm.c
@@ -0,0 +1,233 @@
+/*
+   Copyright (c) 2009 Z RESEARCH, Inc. <http://www.zresearch.com>
+   This file is part of GlusterFS.
+
+   GlusterFS is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published
+   by the Free Software Foundation; either version 3 of the License,
+   or (at your option) any later version.
+
+   GlusterFS is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see
+   <http://www.gnu.org/licenses/>.
+*/
+
+
+#include "glusterfs.h"
+#include "afr.h"
+#include "xlator.h"
+#include "dict.h"
+#include "xlator.h"
+#include "hashfn.h"
+#include "logging.h"
+#include "stack.h"
+#include "list.h"
+#include "call-stub.h"
+#include "defaults.h"
+#include "common-utils.h"
+#include "compat-errno.h"
+#include "compat.h"
+#include "byte-order.h"
+#include "md5.h"
+
+#include "afr-transaction.h"
+#include "afr-self-heal.h"
+#include "afr-self-heal-common.h"
+#include "afr-self-heal-algorithm.h"
+
+/*
+  This file contains the various self-heal algorithms
+*/
+
+
+/*
+  The "full" algorithm. Copies the entire file from
+  source to sinks.
+*/
+
+static int
+sh_full_read_write_iter (call_frame_t *frame, xlator_t *this);
+
+static int
+sh_full_write_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                   int32_t op_ret, int32_t op_errno, struct stat *buf)
+{
+	afr_private_t * priv = NULL;
+	afr_local_t * local  = NULL;
+	afr_self_heal_t *sh  = NULL;
+
+	int child_index = (long) cookie;
+	int call_count = 0;
+
+	priv = this->private;
+	local = frame->local;
+	sh = &local->self_heal;
+
+	gf_log (this->name, GF_LOG_TRACE,
+		"wrote %d bytes of data from %s to child %d, offset %"PRId64"",
+		op_ret, local->loc.path, child_index, sh->offset - op_ret);
+
+	LOCK (&frame->lock);
+	{
+		if (op_ret == -1) {
+			gf_log (this->name, GF_LOG_DEBUG,
+				"write to %s failed on subvolume %s (%s)",
+				local->loc.path,
+				priv->children[child_index]->name,
+				strerror (op_errno));
+			sh->op_failed = 1;
+		}
+	}
+	UNLOCK (&frame->lock);
+
+	call_count = afr_frame_return (frame);
+
+	if (call_count == 0) {
+		sh_full_read_write_iter (frame, this);
+	}
+
+	return 0;
+}
+
+
+static int
+sh_full_read_cbk (call_frame_t *frame, void *cookie,
+                  xlator_t *this, int32_t op_ret, int32_t op_errno,
+                  struct iovec *vector, int32_t count, struct stat *buf,
+                  struct iobref *iobref)
+{
+	afr_private_t * priv = NULL;
+	afr_local_t * local  = NULL;
+	afr_self_heal_t *sh  = NULL;
+
+	int child_index = (long) cookie;
+	int i = 0;
+	int call_count = 0;
+
+	off_t offset;
+
+	priv = this->private;
+	local = frame->local;
+	sh = &local->self_heal;
+
+	call_count = sh->active_sinks;
+
+	local->call_count = call_count;
+
+	gf_log (this->name, GF_LOG_TRACE,
+		"read %d bytes of data from %s on child %d, offset %"PRId64"",
+		op_ret, local->loc.path, child_index, sh->offset);
+
+	if (op_ret <= 0) {
+		local->self_heal.algo_completion_cbk (frame, this);
+		return 0;
+	}
+
+	/* what if we read less than block size? */
+	offset = sh->offset;
+	sh->offset += op_ret;
+
+	if (sh->file_has_holes) {
+		if (iov_0filled (vector, count) == 0) {
+			/* the iter function depends on the
+			   sh->offset already being updated
+			   above
+			*/
+
+                        sh_full_read_write_iter (frame, this);
+			goto out;
+		}
+	}
+
+	for (i = 0; i < priv->child_count; i++) {
+		if (sh->sources[i] || !local->child_up[i])
+			continue;
+
+		/* this is a sink, so write to it */
+		STACK_WIND_COOKIE (frame, sh_full_write_cbk,
+				   (void *) (long) i,
+				   priv->children[i],
+				   priv->children[i]->fops->writev,
+				   sh->healing_fd, vector, count, offset,
+                                   iobref);
+
+		if (!--call_count)
+			break;
+	}
+
+out:
+	return 0;
+}
+
+
+static int
+sh_full_read_write (call_frame_t *frame, xlator_t *this)
+{
+	afr_private_t * priv = NULL;
+	afr_local_t * local  = NULL;
+	afr_self_heal_t *sh  = NULL;
+
+	priv = this->private;
+	local = frame->local;
+	sh = &local->self_heal;
+
+	STACK_WIND_COOKIE (frame, sh_full_read_cbk,
+			   (void *) (long) sh->source,
+			   priv->children[sh->source],
+			   priv->children[sh->source]->fops->readv,
+			   sh->healing_fd, sh->block_size,
+			   sh->offset);
+
+	return 0;
+}
+
+
+static int
+sh_full_read_write_iter (call_frame_t *frame, xlator_t *this)
+{
+	afr_private_t * priv = NULL;
+	afr_local_t * local  = NULL;
+	afr_self_heal_t *sh  = NULL;
+
+	priv = this->private;
+	local = frame->local;
+	sh = &local->self_heal;
+
+	if (sh->op_failed) {
+		local->self_heal.algo_abort_cbk (frame, this);
+		goto out;
+	}
+
+	if (sh->offset >= sh->file_size) {
+		gf_log (this->name, GF_LOG_TRACE,
+			"closing fd's of %s",
+			local->loc.path);
+
+		local->self_heal.algo_completion_cbk (frame, this);
+
+		goto out;
+	}
+
+	sh_full_read_write (frame, this);
+
+out:
+	return 0;
+}
+
+
+int
+afr_sh_algo_full (call_frame_t *frame, xlator_t *this)
+{
+        sh_full_read_write (frame, this);
+        return 0;
+}
+
+
+struct afr_sh_algorithm afr_self_heal_algorithms[] = {
+        {.name = "full",  .fn = afr_sh_algo_full},
+};