geo-rep / gsyncd: log sync failures as warnings

Syncing of certain files can fail naturally if changes happen on master (eg. file gets deleted). Therefore logging an error is misleading. Change-Id: I7b54904e5ec7c85e4e0fa1e330123d2c44c78ac5 BUG: 764510 Reported-by: Vijaykumar Koppad <vkoppad@redhat.com> Signed-off-by: Csaba Henk <csaba@redhat.com> Reviewed-on: http://review.gluster.com/3113 Tested-by: Gluster Build System <jenkins@build.gluster.com> Reviewed-by: Vijaykumar <vkoppad@redhat.com> Reviewed-by: Venky Shankar <vshankar@redhat.com>
author: Csaba Henk <csaba@redhat.com> 2012-04-10 15:34:02 +0200
committer: Vijay Bellur <vijay@gluster.com> 2012-04-13 00:25:47 -0700
commit: c238e5073a8d60fede1f88e584c14de42885276e (patch)
tree: b7316b1929c67c7b28d808fea8b2c654b5870796 /cli/src/cli-cmd.c
parent: ecfe2d0a94dd6626e2e6c1e7707549adf550a5af (diff)
0 files changed, 0 insertions, 0 deletions
diff --git a/xlators/Makefile.am b/xlators/Makefile.am
index 2abb52194..f60fa85ce 100644
--- a/xlators/Makefile.am
+++ b/xlators/Makefile.am
@@ -1,3 +1,4 @@
-SUBDIRS = cluster storage protocol performance debug features encryption mount
+SUBDIRS = cluster storage protocol performance debug features encryption mount nfs mgmt system \
+          playground
 
-CLEANFILES = 
+CLEANFILES =
diff --git a/xlators/bindings/python/src/Makefile.am b/xlators/bindings/python/src/Makefile.am
index c0b9141c6..90370d861 100644
--- a/xlators/bindings/python/src/Makefile.am
+++ b/xlators/bindings/python/src/Makefile.am
@@ -9,7 +9,7 @@ pythondir = $(xlatordir)/python
 
 python_so_SOURCES = python.c
 
-AM_CFLAGS = -fPIC -D_FILE_OFFSET_BITS=64 -D_GNU_SOURCE -Wall \
+AM_CFLAGS = -fPIC $(GF_CPPFLAGS) -Wall \
 	-I$(top_srcdir)/libglusterfs/src -shared -nostartfiles \
 	$(PYTHON_CPPLAGS) -DGLUSTER_PYTHON_PATH=\"$(pythondir)\"
 
diff --git a/xlators/bindings/python/src/gluster.py b/xlators/bindings/python/src/gluster.py
index ee0eb1310..337c983ec 100644
--- a/xlators/bindings/python/src/gluster.py
+++ b/xlators/bindings/python/src/gluster.py
@@ -1,19 +1,12 @@
-#   Copyright (c) 2007 Chris AtLee <chris@atlee.ca>
-#   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.
+
+#  Copyright (c) 2007-2012 Red Hat, Inc. <http://www.redhat.com>
+#  This file is part of GlusterFS.
 #
-#   You should have received a copy of the GNU General Public License
-#   along with this program.  If not, see
-#   <http://www.gnu.org/licenses/>.
+#  This file is licensed to you under your choice of the GNU Lesser
+#  General Public License, version 3 or any later version (LGPLv3 or
+#  later), or the GNU General Public License, version 2 (GPLv2), in all
+#  cases as published by the Free Software Foundation.
+
 from ctypes import *
 from glustertypes import *
 from glusterstack import *
diff --git a/xlators/bindings/python/src/glusterstack.py b/xlators/bindings/python/src/glusterstack.py
index ba24c8165..0c071ae98 100644
--- a/xlators/bindings/python/src/glusterstack.py
+++ b/xlators/bindings/python/src/glusterstack.py
@@ -1,19 +1,12 @@
-#   Copyright (c) 2007 Chris AtLee <chris@atlee.ca>
-#   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.
+
+#  Copyright (c) 2007-2012 Red Hat, Inc. <http://www.redhat.com>
+#  This file is part of GlusterFS.
 #
-#   You should have received a copy of the GNU General Public License
-#   along with this program.  If not, see
-#   <http://www.gnu.org/licenses/>.
+#  This file is licensed to you under your choice of the GNU Lesser
+#  General Public License, version 3 or any later version (LGPLv3 or
+#  later), or the GNU General Public License, version 2 (GPLv2), in all
+#  cases as published by the Free Software Foundation.
+
 from ctypes import *
 from glustertypes import *
 
diff --git a/xlators/bindings/python/src/glustertypes.py b/xlators/bindings/python/src/glustertypes.py
index e9069d07c..98437d22e 100644
--- a/xlators/bindings/python/src/glustertypes.py
+++ b/xlators/bindings/python/src/glustertypes.py
@@ -1,19 +1,12 @@
-#   Copyright (c) 2007 Chris AtLee <chris@atlee.ca>
-#   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.
+
+#  Copyright (c) 2007-2012 Red Hat, Inc. <http://www.redhat.com>
+#  This file is part of GlusterFS.
 #
-#   You should have received a copy of the GNU General Public License
-#   along with this program.  If not, see
-#   <http://www.gnu.org/licenses/>.
+#  This file is licensed to you under your choice of the GNU Lesser
+#  General Public License, version 3 or any later version (LGPLv3 or
+#  later), or the GNU General Public License, version 2 (GPLv2), in all
+#  cases as published by the Free Software Foundation.
+
 from ctypes import *
 import collections
 
diff --git a/xlators/bindings/python/src/python.c b/xlators/bindings/python/src/python.c
index 7f32d7f29..9b96790de 100644
--- a/xlators/bindings/python/src/python.c
+++ b/xlators/bindings/python/src/python.c
@@ -1,22 +1,12 @@
 /*
-   Copyright (c) 2007-2009 Chris AtLee <chris@atlee.ca>
+   Copyright (c) 2008-2012 Red Hat, Inc. <http://www.redhat.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/>.
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
 */
-
 #include <Python.h>
 
 #ifndef _CONFIG_H
@@ -45,7 +35,7 @@ python_writev (call_frame_t *frame,
               xlator_t *this,
               fd_t *fd,
               struct iovec *vector,
-              int32_t count, 
+              int32_t count,
               off_t offset)
 {
   python_private_t *priv = (python_private_t *)this->private;
@@ -77,9 +67,6 @@ struct xlator_fops fops = {
     .writev       = python_writev
 };
 
-struct xlator_mops mops = {
-};
-
 static PyObject *
 AnonModule_FromFile (const char* fname)
 {
@@ -151,7 +138,7 @@ init (xlator_t *this)
   Py_InitializeEx(0);
 
   if (!this->children) {
-    gf_log ("python", GF_LOG_ERROR, 
+    gf_log ("python", GF_LOG_ERROR,
             "FATAL: python should have exactly one child");
     return -1;
   }
@@ -169,7 +156,7 @@ init (xlator_t *this)
   }
 
   priv->pInterp = Py_NewInterpreter();
-    
+
   // Adjust python's path
   PyObject *syspath = PySys_GetObject("path");
   PyObject *path = PyString_FromString(GLUSTER_PYTHON_PATH);
@@ -191,7 +178,7 @@ init (xlator_t *this)
   priv->pVectorType = PyObject_GetAttrString(priv->pGlusterModule, "iovec");
 
   gf_log("python", GF_LOG_DEBUG, "Loading script...%s", priv->scriptname);
-  
+
   priv->pScriptModule = AnonModule_FromFile(priv->scriptname);
   if (!priv->pScriptModule || PyErr_Occurred())
   {
@@ -220,7 +207,7 @@ init (xlator_t *this)
   return 0;
 }
 
-void 
+void
 fini (xlator_t *this)
 {
   python_private_t *priv = (python_private_t*)(this->private);
diff --git a/xlators/bindings/python/src/testxlator.py b/xlators/bindings/python/src/testxlator.py
index 507455c85..59a991dca 100644
--- a/xlators/bindings/python/src/testxlator.py
+++ b/xlators/bindings/python/src/testxlator.py
@@ -1,19 +1,12 @@
-#   Copyright (c) 2007 Chris AtLee <chris@atlee.ca>
-#   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/>.
+"""
+  Copyright (c) 2007-2012 Red Hat, Inc. <http://www.redhat.com>
+  This file is part of GlusterFS.
+
+  This file is licensed to you under your choice of the GNU Lesser
+  General Public License, version 3 or any later version (LGPLv3 or
+  later), or the GNU General Public License, version 2 (GPLv2), in all
+  cases as published by the Free Software Foundation.
+"""
 
 """
 This is a test translator written in python.
diff --git a/xlators/cluster/Makefile.am b/xlators/cluster/Makefile.am
index a6ddb3564..0990822a7 100644
--- a/xlators/cluster/Makefile.am
+++ b/xlators/cluster/Makefile.am
@@ -1,3 +1,3 @@
-SUBDIRS = unify stripe afr dht ha map
+SUBDIRS = stripe afr dht
 
 CLEANFILES = 
diff --git a/xlators/cluster/afr/src/Makefile.am b/xlators/cluster/afr/src/Makefile.am
index df284d12c..35d18a6c0 100644
--- a/xlators/cluster/afr/src/Makefile.am
+++ b/xlators/cluster/afr/src/Makefile.am
@@ -1,20 +1,37 @@
-xlator_LTLIBRARIES = afr.la
+xlator_LTLIBRARIES = afr.la pump.la
 xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/cluster
 
-afr_la_LDFLAGS = -module -avoidversion 
+afr_common_source = afr-dir-read.c afr-dir-write.c afr-inode-read.c \
+	afr-inode-write.c afr-open.c afr-transaction.c afr-self-heal-data.c \
+	afr-self-heal-common.c afr-self-heal-metadata.c afr-self-heal-entry.c \
+	afr-self-heal-algorithm.c afr-lk-common.c afr-self-heald.c \
+	$(top_builddir)/xlators/lib/src/libxlator.c
 
-afr_la_SOURCES = afr.c afr-dir-read.c afr-dir-write.c afr-inode-read.c afr-inode-write.c afr-transaction.c afr-self-heal-data.c afr-self-heal-common.c afr-self-heal-metadata.c afr-self-heal-entry.c afr-self-heal-algorithm.c
+afr_la_LDFLAGS = -module -avoid-version
+afr_la_SOURCES = $(afr_common_source) afr.c
 afr_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la
 
-noinst_HEADERS = afr.h afr-transaction.h afr-inode-write.h afr-inode-read.h afr-dir-read.h afr-dir-write.h afr-self-heal.h afr-self-heal-common.h afr-self-heal-algorithm.h
+pump_la_LDFLAGS = -module -avoid-version
+pump_la_SOURCES =  $(afr_common_source) pump.c
+pump_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la
 
-AM_CFLAGS = -fPIC -D_FILE_OFFSET_BITS=64 -D_GNU_SOURCE -Wall -D$(GF_HOST_OS) \
-	    -I$(top_srcdir)/libglusterfs/src -shared -nostartfiles $(GF_CFLAGS)
+noinst_HEADERS = afr.h afr-transaction.h afr-inode-write.h afr-inode-read.h \
+	afr-dir-read.h afr-dir-write.h afr-self-heal.h afr-self-heal-common.h \
+	afr-self-heal-algorithm.h pump.h afr-mem-types.h afr-common.c \
+	afr-self-heald.h $(top_builddir)/xlators/lib/src/libxlator.h \
+	$(top_builddir)/glusterfsd/src/glusterfsd.h
 
-CLEANFILES = 
+AM_CPPFLAGS = $(GF_CPPFLAGS) \
+	-I$(top_srcdir)/libglusterfs/src -I$(top_srcdir)/xlators/lib/src \
+	-I$(top_srcdir)/rpc/rpc-lib/src
+
+AM_CFLAGS = -Wall $(GF_CFLAGS)
+
+CLEANFILES =
 
 uninstall-local:
 	rm -f $(DESTDIR)$(xlatordir)/replicate.so
+	rm -f $(DESTDIR)$(xlatordir)/pump.so
 
 install-data-hook:
-	ln -sf afr.so $(DESTDIR)$(xlatordir)/replicate.so
-\ No newline at end of file
+	ln -sf afr.so $(DESTDIR)$(xlatordir)/replicate.so
diff --git a/xlators/cluster/afr/src/afr-common.c b/xlators/cluster/afr/src/afr-common.c
new file mode 100644
index 000000000..af01f2ef2
--- /dev/null
+++ b/xlators/cluster/afr/src/afr-common.c
@@ -0,0 +1,4591 @@
+/*
+  Copyright (c) 2008-2012 Red Hat, Inc. <http://www.redhat.com>
+  This file is part of GlusterFS.
+
+  This file is licensed to you under your choice of the GNU Lesser
+  General Public License, version 3 or any later version (LGPLv3 or
+  later), or the GNU General Public License, version 2 (GPLv2), in all
+  cases as published by the Free Software Foundation.
+*/
+
+#include <libgen.h>
+#include <unistd.h>
+#include <fnmatch.h>
+#include <sys/time.h>
+#include <stdlib.h>
+#include <signal.h>
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "glusterfs.h"
+#include "afr.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 "statedump.h"
+#include "inode.h"
+
+#include "fd.h"
+
+#include "afr-inode-read.h"
+#include "afr-inode-write.h"
+#include "afr-dir-read.h"
+#include "afr-dir-write.h"
+#include "afr-transaction.h"
+#include "afr-self-heal.h"
+#include "afr-self-heal-common.h"
+#include "afr-self-heald.h"
+#include "pump.h"
+
+#define AFR_ICTX_OPENDIR_DONE_MASK     0x0000000100000000ULL
+#define AFR_ICTX_READ_CHILD_MASK       0x00000000FFFFFFFFULL
+#define AFR_STATISTICS_HISTORY_SIZE    50
+int
+afr_lookup_done_success_action (call_frame_t *frame, xlator_t *this,
+                                gf_boolean_t fail_conflict);
+void
+afr_children_copy (int32_t *dst, int32_t *src, unsigned int child_count)
+{
+        int     i = 0;
+
+        for (i = 0; i < child_count; i++)
+                dst[i] = src[i];
+}
+
+void
+afr_xattr_req_prepare (xlator_t *this, dict_t *xattr_req, const char *path)
+{
+        int             i           = 0;
+        afr_private_t   *priv       = NULL;
+        int             ret         = 0;
+
+        priv   = this->private;
+
+        for (i = 0; i < priv->child_count; i++) {
+                ret = dict_set_uint64 (xattr_req, priv->pending_key[i],
+                                       3 * sizeof(int32_t));
+                if (ret < 0)
+                        gf_log (this->name, GF_LOG_WARNING,
+                                "%s: Unable to set dict value for %s",
+                                path, priv->pending_key[i]);
+                /* 3 = data+metadata+entry */
+        }
+        ret = dict_set_int32 (xattr_req, GF_GFIDLESS_LOOKUP, 1);
+        if (ret) {
+                gf_log (this->name, GF_LOG_DEBUG, "%s: failed to set gfidless "
+                        "lookup", path);
+        }
+}
+
+int
+afr_lookup_xattr_req_prepare (afr_local_t *local, xlator_t *this,
+                              dict_t *xattr_req, loc_t *loc, void **gfid_req)
+{
+        int     ret = -ENOMEM;
+
+        GF_ASSERT (gfid_req);
+
+        *gfid_req = NULL;
+        local->xattr_req = dict_new ();
+        if (!local->xattr_req)
+                goto out;
+        if (xattr_req)
+                dict_copy (xattr_req, local->xattr_req);
+
+        afr_xattr_req_prepare (this, local->xattr_req, loc->path);
+        ret = dict_set_uint64 (local->xattr_req, GLUSTERFS_INODELK_COUNT, 0);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "%s: Unable to set dict value for %s",
+                        loc->path, GLUSTERFS_INODELK_COUNT);
+        }
+        ret = dict_set_uint64 (local->xattr_req, GLUSTERFS_ENTRYLK_COUNT, 0);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "%s: Unable to set dict value for %s",
+                        loc->path, GLUSTERFS_ENTRYLK_COUNT);
+        }
+
+        ret = dict_set_uint32 (local->xattr_req, GLUSTERFS_PARENT_ENTRYLK, 0);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "%s: Unable to set dict value for %s",
+                        loc->path, GLUSTERFS_PARENT_ENTRYLK);
+        }
+
+        ret = dict_get_ptr (local->xattr_req, "gfid-req", gfid_req);
+        if (ret) {
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "%s: failed to get the gfid from dict", loc->path);
+                *gfid_req = NULL;
+        } else {
+                if (loc->parent != NULL)
+                        dict_del (local->xattr_req, "gfid-req");
+        }
+        ret = 0;
+out:
+        return ret;
+}
+
+void
+afr_lookup_save_gfid (uuid_t dst, void* new, const loc_t *loc)
+{
+        inode_t  *inode = NULL;
+
+        inode = loc->inode;
+        if (inode && !uuid_is_null (inode->gfid))
+                uuid_copy (dst, inode->gfid);
+        else if (!uuid_is_null (loc->gfid))
+                uuid_copy (dst, loc->gfid);
+        else if (new && !uuid_is_null (new))
+                uuid_copy (dst, new);
+}
+
+int
+afr_errno_count (int32_t *children, int *child_errno,
+                 unsigned int child_count, int32_t op_errno)
+{
+        int i = 0;
+        int errno_count = 0;
+        int child = 0;
+
+        for (i = 0; i < child_count; i++) {
+                if (children) {
+                        child = children[i];
+                        if (child == -1)
+                                break;
+                } else {
+                        child = i;
+                }
+                if (child_errno[child] == op_errno)
+                        errno_count++;
+        }
+        return errno_count;
+}
+
+int32_t
+afr_set_dict_gfid (dict_t *dict, uuid_t gfid)
+{
+        int ret       = 0;
+        uuid_t *pgfid = NULL;
+
+        GF_ASSERT (gfid);
+
+        pgfid = GF_CALLOC (1, sizeof (uuid_t), gf_common_mt_char);
+        if (!pgfid) {
+                ret = -1;
+                goto out;
+        }
+
+        uuid_copy (*pgfid, gfid);
+
+        ret = dict_set_dynptr (dict, "gfid-req", pgfid, sizeof (uuid_t));
+        if (ret)
+                gf_log (THIS->name, GF_LOG_ERROR, "gfid set failed");
+
+out:
+        if (ret && pgfid)
+                GF_FREE (pgfid);
+
+        return ret;
+}
+
+void
+afr_inode_ctx_destroy (afr_inode_ctx_t *ctx)
+{
+        if (!ctx)
+                return;
+        GF_FREE (ctx->fresh_children);
+        GF_FREE (ctx);
+}
+
+afr_inode_ctx_t*
+__afr_inode_ctx_get (inode_t *inode, xlator_t *this)
+{
+        int             ret      = 0;
+        uint64_t        ctx_addr = 0;
+        afr_inode_ctx_t *ctx     = NULL;
+        afr_private_t   *priv    = NULL;
+
+        priv = this->private;
+        ret = __inode_ctx_get (inode, this, &ctx_addr);
+        if (ret < 0)
+                ctx_addr = 0;
+        if (ctx_addr != 0) {
+                ctx = (afr_inode_ctx_t*) (long) ctx_addr;
+                goto out;
+        }
+        ctx = GF_CALLOC (1, sizeof (*ctx),
+                         gf_afr_mt_inode_ctx_t);
+        if (!ctx)
+                goto fail;
+        ctx->fresh_children = GF_CALLOC (priv->child_count,
+                                         sizeof (*ctx->fresh_children),
+                                         gf_afr_mt_int32_t);
+        if (!ctx->fresh_children)
+                goto fail;
+        ret = __inode_ctx_put (inode, this, (uint64_t)ctx);
+        if (ret) {
+                gf_log_callingfn (this->name, GF_LOG_ERROR, "failed to "
+                                  "set the inode ctx (%s)",
+                                  uuid_utoa (inode->gfid));
+                goto fail;
+        }
+
+out:
+        return ctx;
+
+fail:
+        afr_inode_ctx_destroy (ctx);
+        return NULL;
+}
+
+afr_inode_ctx_t*
+afr_inode_ctx_get (inode_t *inode, xlator_t *this)
+{
+        afr_inode_ctx_t *ctx = NULL;
+
+        LOCK (&inode->lock);
+        {
+                ctx = __afr_inode_ctx_get (inode, this);
+        }
+        UNLOCK (&inode->lock);
+        return ctx;
+}
+
+void
+afr_inode_get_ctx_params (xlator_t *this, inode_t *inode,
+                          afr_inode_params_t *params)
+{
+        GF_ASSERT (inode);
+        GF_ASSERT (params);
+
+        afr_inode_ctx_t *ctx = NULL;
+        afr_private_t   *priv = NULL;
+        int             i = 0;
+        int32_t         read_child = -1;
+        int32_t         *fresh_children = NULL;
+
+        priv = this->private;
+        LOCK (&inode->lock);
+        {
+                ctx = __afr_inode_ctx_get (inode, this);
+                if (!ctx)
+                        goto unlock;
+                switch (params->op) {
+                case AFR_INODE_GET_READ_CTX:
+                        fresh_children = params->u.read_ctx.children;
+                        read_child = (int32_t)(ctx->masks &
+                                               AFR_ICTX_READ_CHILD_MASK);
+                        params->u.read_ctx.read_child = read_child;
+                        if (!fresh_children)
+                                goto unlock;
+                        for (i = 0; i < priv->child_count; i++)
+                                fresh_children[i] = ctx->fresh_children[i];
+                        break;
+                case AFR_INODE_GET_OPENDIR_DONE:
+                        params->u.value = _gf_false;
+                        if (ctx->masks & AFR_ICTX_OPENDIR_DONE_MASK)
+                                params->u.value = _gf_true;
+                        break;
+                default:
+                        GF_ASSERT (0);
+                        break;
+                }
+        }
+unlock:
+        UNLOCK (&inode->lock);
+}
+
+gf_boolean_t
+afr_is_split_brain (xlator_t *this, inode_t *inode)
+{
+        afr_inode_ctx_t *ctx = NULL;
+        gf_boolean_t    spb  = _gf_false;
+
+        ctx = afr_inode_ctx_get (inode, this);
+        if (!ctx)
+                goto out;
+        if ((ctx->mdata_spb == SPB) || (ctx->data_spb == SPB))
+                spb = _gf_true;
+out:
+        return spb;
+}
+
+gf_boolean_t
+afr_is_opendir_done (xlator_t *this, inode_t *inode)
+{
+        afr_inode_params_t params = {0};
+
+        params.op = AFR_INODE_GET_OPENDIR_DONE;
+        afr_inode_get_ctx_params (this, inode, &params);
+        return params.u.value;
+}
+
+int32_t
+afr_inode_get_read_ctx (xlator_t *this, inode_t *inode, int32_t *fresh_children)
+{
+        afr_inode_params_t      params = {0};
+
+        params.op = AFR_INODE_GET_READ_CTX;
+        params.u.read_ctx.children = fresh_children;
+        afr_inode_get_ctx_params (this, inode, &params);
+        return params.u.read_ctx.read_child;
+}
+
+void
+afr_inode_ctx_set_read_child (afr_inode_ctx_t *ctx, int32_t read_child)
+{
+        uint64_t        remaining_mask = 0;
+        uint64_t        mask         = 0;
+
+        remaining_mask = (~AFR_ICTX_READ_CHILD_MASK & ctx->masks);
+        mask = (AFR_ICTX_READ_CHILD_MASK & read_child);
+        ctx->masks = remaining_mask | mask;
+}
+
+void
+afr_inode_ctx_set_read_ctx (afr_inode_ctx_t *ctx, int32_t read_child,
+                            int32_t *fresh_children, int32_t child_count)
+{
+        int             i            = 0;
+
+        afr_inode_ctx_set_read_child (ctx, read_child);
+        for (i = 0; i < child_count; i++) {
+                if (fresh_children)
+                        ctx->fresh_children[i] = fresh_children[i];
+                else
+                        ctx->fresh_children[i] = -1;
+        }
+}
+
+void
+afr_inode_ctx_rm_stale_children (afr_inode_ctx_t *ctx, int32_t *stale_children,
+                                 int32_t child_count)
+{
+        int             i            = 0;
+        int32_t         read_child   = -1;
+
+        GF_ASSERT (stale_children);
+        for (i = 0; i < child_count; i++) {
+                if (stale_children[i] == -1)
+                        break;
+                afr_children_rm_child (ctx->fresh_children,
+                                       stale_children[i], child_count);
+        }
+        read_child = (int32_t)(ctx->masks & AFR_ICTX_READ_CHILD_MASK);
+        if (!afr_is_child_present (ctx->fresh_children, child_count,
+                                   read_child))
+                afr_inode_ctx_set_read_child (ctx, ctx->fresh_children[0]);
+}
+
+void
+afr_inode_ctx_set_opendir_done (afr_inode_ctx_t *ctx)
+{
+        uint64_t        remaining_mask = 0;
+        uint64_t        mask = 0;
+
+        remaining_mask = (~AFR_ICTX_OPENDIR_DONE_MASK & ctx->masks);
+        mask = (0xFFFFFFFFFFFFFFFFULL & AFR_ICTX_OPENDIR_DONE_MASK);
+        ctx->masks = remaining_mask | mask;
+}
+
+void
+afr_inode_set_ctx_params (xlator_t *this, inode_t *inode,
+                          afr_inode_params_t *params)
+{
+        GF_ASSERT (inode);
+        GF_ASSERT (params);
+
+        afr_inode_ctx_t *ctx            = NULL;
+        afr_private_t   *priv           = NULL;
+        int32_t         read_child      = -1;
+        int32_t         *fresh_children = NULL;
+        int32_t         *stale_children = NULL;
+
+        priv = this->private;
+        LOCK (&inode->lock);
+        {
+                ctx = __afr_inode_ctx_get (inode, this);
+                if (!ctx)
+                        goto unlock;
+                switch (params->op) {
+                case AFR_INODE_SET_READ_CTX:
+                        read_child = params->u.read_ctx.read_child;
+                        fresh_children = params->u.read_ctx.children;
+                        afr_inode_ctx_set_read_ctx (ctx, read_child,
+                                                    fresh_children,
+                                                    priv->child_count);
+                        break;
+                case AFR_INODE_RM_STALE_CHILDREN:
+                        stale_children = params->u.read_ctx.children;
+                        afr_inode_ctx_rm_stale_children (ctx,
+                                                         stale_children,
+                                                         priv->child_count);
+                        break;
+                case AFR_INODE_SET_OPENDIR_DONE:
+                        afr_inode_ctx_set_opendir_done (ctx);
+                        break;
+                default:
+                        GF_ASSERT (0);
+                        break;
+                }
+        }
+unlock:
+        UNLOCK (&inode->lock);
+}
+
+void
+afr_set_split_brain (xlator_t *this, inode_t *inode, afr_spb_state_t mdata_spb,
+                     afr_spb_state_t data_spb)
+{
+        afr_inode_ctx_t *ctx = NULL;
+
+        ctx = afr_inode_ctx_get (inode, this);
+        if (mdata_spb != DONT_KNOW)
+                ctx->mdata_spb = mdata_spb;
+        if (data_spb != DONT_KNOW)
+                ctx->data_spb = data_spb;
+}
+
+void
+afr_set_opendir_done (xlator_t *this, inode_t *inode)
+{
+        afr_inode_params_t params = {0};
+
+        params.op = AFR_INODE_SET_OPENDIR_DONE;
+        afr_inode_set_ctx_params (this, inode, &params);
+}
+
+void
+afr_inode_set_read_ctx (xlator_t *this, inode_t *inode, int32_t read_child,
+                        int32_t *fresh_children)
+{
+        afr_inode_params_t params = {0};
+        afr_private_t      *priv  = NULL;
+
+        priv = this->private;
+        GF_ASSERT (read_child >= 0);
+        GF_ASSERT (fresh_children);
+        GF_ASSERT (afr_is_child_present (fresh_children, priv->child_count,
+                                         read_child));
+
+        params.op = AFR_INODE_SET_READ_CTX;
+        params.u.read_ctx.read_child     = read_child;
+        params.u.read_ctx.children = fresh_children;
+        afr_inode_set_ctx_params (this, inode, &params);
+}
+
+void
+afr_inode_rm_stale_children (xlator_t *this, inode_t *inode,
+                             int32_t *stale_children)
+{
+        afr_inode_params_t params = {0};
+
+        GF_ASSERT (stale_children);
+
+        params.op = AFR_INODE_RM_STALE_CHILDREN;
+        params.u.read_ctx.children = stale_children;
+        afr_inode_set_ctx_params (this, inode, &params);
+}
+
+gf_boolean_t
+afr_is_source_child (int32_t *sources, int32_t child_count, int32_t child)
+{
+        gf_boolean_t             source_xattrs = _gf_false;
+
+        GF_ASSERT (child < child_count);
+
+        if ((child >= 0) && (child < child_count) &&
+             sources[child]) {
+                source_xattrs = _gf_true;
+        }
+        return source_xattrs;
+}
+
+gf_boolean_t
+afr_is_child_present (int32_t *success_children, int32_t child_count,
+                      int32_t child)
+{
+        gf_boolean_t             success_child = _gf_false;
+        int                      i = 0;
+
+        GF_ASSERT (child < child_count);
+
+        for (i = 0; i < child_count; i++) {
+                if (success_children[i] == -1)
+                        break;
+                if (child == success_children[i]) {
+                        success_child = _gf_true;
+                        break;
+                }
+        }
+        return success_child;
+}
+
+gf_boolean_t
+afr_is_read_child (int32_t *success_children, int32_t *sources,
+                   int32_t child_count, int32_t child)
+{
+        gf_boolean_t             success_child = _gf_false;
+        gf_boolean_t             source        = _gf_false;
+
+        if (child < 0) {
+                return _gf_false;
+        }
+
+        GF_ASSERT (success_children);
+        GF_ASSERT (child_count > 0);
+
+        success_child = afr_is_child_present (success_children, child_count,
+                                              child);
+        if (!success_child)
+                goto out;
+        if (NULL == sources) {
+                source = _gf_true;
+                goto out;
+        }
+        source = afr_is_source_child (sources, child_count, child);
+out:
+        return (success_child && source);
+}
+
+int32_t
+afr_hash_child (int32_t *success_children, int32_t child_count,
+                unsigned int hmode, uuid_t gfid)
+{
+        uuid_t  gfid_copy = {0,};
+        pid_t pid;
+
+        if (!hmode) {
+                return -1;
+        }
+
+        if (gfid) {
+               uuid_copy(gfid_copy,gfid);
+        }
+        if (hmode > 1) {
+                /*
+                 * Why getpid?  Because it's one of the cheapest calls
+                 * available - faster than gethostname etc. - and returns a
+                 * constant-length value that's sure to be shorter than a UUID.
+                 * It's still very unlikely to be the same across clients, so
+                 * it still provides good mixing.  We're not trying for
+                 * perfection here.  All we need is a low probability that
+                 * multiple clients won't converge on the same subvolume.
+                 */
+                pid = getpid();
+                memcpy (gfid_copy, &pid, sizeof(pid));
+        }
+
+        return SuperFastHash((char *)gfid_copy,
+                             sizeof(gfid_copy)) % child_count;
+}
+
+/* If sources is NULL the xattrs are assumed to be of source for all
+ * success_children.
+ */
+int
+afr_select_read_child_from_policy (int32_t *success_children,
+                                   int32_t child_count, int32_t prev_read_child,
+                                   int32_t config_read_child, int32_t *sources,
+                                   unsigned int hmode, uuid_t gfid)
+{
+        int32_t                  read_child   = -1;
+        int                      i            = 0;
+
+        GF_ASSERT (success_children);
+
+        read_child = config_read_child;
+        if (afr_is_read_child (success_children, sources, child_count,
+                               read_child))
+                goto out;
+
+        read_child = prev_read_child;
+        if (afr_is_read_child (success_children, sources, child_count,
+                               read_child))
+                goto out;
+
+        read_child = afr_hash_child (success_children, child_count,
+                                     hmode, gfid);
+        if (afr_is_read_child (success_children, sources, child_count,
+                               read_child)) {
+                goto out;
+        }
+
+        for (i = 0; i < child_count; i++) {
+                read_child = success_children[i];
+                if (read_child < 0)
+                        break;
+                if (afr_is_read_child (success_children, sources, child_count,
+                                       read_child))
+                        goto out;
+        }
+        read_child = -1;
+
+out:
+        return read_child;
+}
+
+/* This function should be used when all the success_children are sources
+ */
+void
+afr_set_read_ctx_from_policy (xlator_t *this, inode_t *inode,
+                              int32_t *fresh_children, int32_t prev_read_child,
+                              int32_t config_read_child, uuid_t gfid)
+{
+        int                      read_child = -1;
+        afr_private_t            *priv = NULL;
+
+        priv = this->private;
+        read_child = afr_select_read_child_from_policy (fresh_children,
+                                                        priv->child_count,
+                                                        prev_read_child,
+                                                        config_read_child,
+                                                        NULL,
+                                                        priv->hash_mode, gfid);
+        if (read_child >= 0)
+                afr_inode_set_read_ctx (this, inode, read_child,
+                                        fresh_children);
+}
+
+/* afr_next_call_child ()
+ * This is a common function used by all the read-type fops
+ * This function should not be called with the inode's read_children array.
+ * The fop's handler should make a copy of the inode's read_children,
+ * preferred read_child into the local vars, because while this function is
+ * in execution there is a chance for inode's read_ctx to change.
+ */
+int32_t
+afr_next_call_child (int32_t *fresh_children, unsigned char *child_up,
+                     size_t child_count, int32_t *last_index,
+                     int32_t read_child)
+{
+        int             next_index      = 0;
+        int32_t         next_call_child = -1;
+
+        GF_ASSERT (last_index);
+
+        next_index = *last_index;
+retry:
+        next_index++;
+        if ((next_index >= child_count) ||
+           (fresh_children[next_index] == -1))
+                goto out;
+        if ((fresh_children[next_index] == read_child) ||
+           (!child_up[fresh_children[next_index]]))
+                goto retry;
+        *last_index = next_index;
+        next_call_child = fresh_children[next_index];
+out:
+        return next_call_child;
+}
+
+ /* This function should not be called with the inode's read_children array.
+ * The fop's handler should make a copy of the inode's read_children,
+ * preferred read_child into the local vars, because while this function is
+ * in execution there is a chance for inode's read_ctx to change.
+ */
+int32_t
+afr_get_call_child (xlator_t *this, unsigned char *child_up, int32_t read_child,
+                    int32_t *fresh_children,
+                    int32_t *call_child, int32_t *last_index)
+{
+        int             ret   = 0;
+        afr_private_t   *priv = NULL;
+        int             i     = 0;
+
+        GF_ASSERT (child_up);
+        GF_ASSERT (call_child);
+        GF_ASSERT (last_index);
+        GF_ASSERT (fresh_children);
+
+        if (read_child < 0) {
+                ret = -EIO;
+                goto out;
+        }
+        priv = this->private;
+        *call_child = -1;
+        *last_index = -1;
+
+        if (child_up[read_child]) {
+                *call_child = read_child;
+        } else {
+                for (i = 0; i < priv->child_count; i++) {
+                        if (fresh_children[i] == -1)
+                                break;
+                        if (child_up[fresh_children[i]]) {
+                                *call_child = fresh_children[i];
+                                ret = 0;
+                                break;
+                        }
+                }
+
+                if (*call_child == -1) {
+                        ret = -ENOTCONN;
+                        goto out;
+                }
+
+                *last_index = i;
+        }
+out:
+        gf_log (this->name, GF_LOG_DEBUG, "Returning %d, call_child: %d, "
+                "last_index: %d", ret, *call_child, *last_index);
+        return ret;
+}
+
+void
+afr_reset_xattr (dict_t **xattr, unsigned int child_count)
+{
+        unsigned int i = 0;
+
+        if (!xattr)
+                goto out;
+        for (i = 0; i < child_count; i++) {
+                if (xattr[i]) {
+                        dict_unref (xattr[i]);
+                        xattr[i] = NULL;
+                }
+        }
+out:
+        return;
+}
+
+void
+afr_xattr_array_destroy (dict_t **xattr, unsigned int child_count)
+{
+        afr_reset_xattr (xattr, child_count);
+        GF_FREE (xattr);
+}
+
+void
+afr_local_sh_cleanup (afr_local_t *local, xlator_t *this)
+{
+        afr_self_heal_t *sh = NULL;
+        afr_private_t   *priv = NULL;
+
+        sh = &local->self_heal;
+        priv = this->private;
+
+        if (sh->data_sh_info && strcmp (sh->data_sh_info, ""))
+                GF_FREE (sh->data_sh_info);
+
+        if (sh->metadata_sh_info && strcmp (sh->metadata_sh_info, ""))
+                GF_FREE (sh->metadata_sh_info);
+
+        GF_FREE (sh->buf);
+
+        GF_FREE (sh->parentbufs);
+
+        if (sh->inode)
+                inode_unref (sh->inode);
+
+        afr_xattr_array_destroy (sh->xattr, priv->child_count);
+
+        GF_FREE (sh->child_errno);
+
+        afr_matrix_cleanup (sh->pending_matrix, priv->child_count);
+        afr_matrix_cleanup (sh->delta_matrix, priv->child_count);
+
+        GF_FREE (sh->sources);
+
+        GF_FREE (sh->success);
+
+        GF_FREE (sh->locked_nodes);
+
+        if (sh->healing_fd) {
+                fd_unref (sh->healing_fd);
+                sh->healing_fd = NULL;
+        }
+
+        GF_FREE ((char *)sh->linkname);
+
+        GF_FREE (sh->success_children);
+
+        GF_FREE (sh->fresh_children);
+
+        GF_FREE (sh->fresh_parent_dirs);
+
+        loc_wipe (&sh->parent_loc);
+        loc_wipe (&sh->lookup_loc);
+
+        GF_FREE (sh->checksum);
+
+        GF_FREE (sh->write_needed);
+        if (sh->healing_fd)
+                fd_unref (sh->healing_fd);
+}
+
+
+void
+afr_local_transaction_cleanup (afr_local_t *local, xlator_t *this)
+{
+        afr_private_t *priv    = NULL;
+        int           i        = 0;
+
+        priv = this->private;
+
+        afr_matrix_cleanup (local->pending, priv->child_count);
+        afr_matrix_cleanup (local->transaction.txn_changelog,
+                            priv->child_count);
+
+        GF_FREE (local->internal_lock.locked_nodes);
+
+        for (i = 0; local->internal_lock.inodelk[i].domain; i++) {
+                GF_FREE (local->internal_lock.inodelk[i].locked_nodes);
+        }
+
+        GF_FREE (local->internal_lock.lower_locked_nodes);
+
+        afr_entry_lockee_cleanup (&local->internal_lock);
+
+        GF_FREE (local->transaction.pre_op);
+        GF_FREE (local->transaction.eager_lock);
+
+        GF_FREE (local->transaction.basename);
+        GF_FREE (local->transaction.new_basename);
+
+        loc_wipe (&local->transaction.parent_loc);
+        loc_wipe (&local->transaction.new_parent_loc);
+
+        GF_FREE (local->transaction.postop_piggybacked);
+}
+
+
+void
+afr_local_cleanup (afr_local_t *local, xlator_t *this)
+{
+        afr_private_t * priv = NULL;
+
+        if (!local)
+                return;
+
+        afr_local_sh_cleanup (local, this);
+
+        afr_local_transaction_cleanup (local, this);
+
+        priv = this->private;
+
+        loc_wipe (&local->loc);
+        loc_wipe (&local->newloc);
+
+        if (local->fd)
+                fd_unref (local->fd);
+
+        if (local->xattr_req)
+                dict_unref (local->xattr_req);
+
+        if (local->dict)
+                dict_unref (local->dict);
+
+	GF_FREE(local->replies);
+
+        GF_FREE (local->child_up);
+
+        GF_FREE (local->child_errno);
+
+        GF_FREE (local->fresh_children);
+
+        { /* lookup */
+                if (local->cont.lookup.xattrs) {
+                        afr_reset_xattr (local->cont.lookup.xattrs,
+                                         priv->child_count);
+                        GF_FREE (local->cont.lookup.xattrs);
+                        local->cont.lookup.xattrs = NULL;
+                }
+
+                if (local->cont.lookup.xattr) {
+                        dict_unref (local->cont.lookup.xattr);
+                }
+
+                if (local->cont.lookup.inode) {
+                        inode_unref (local->cont.lookup.inode);
+                }
+
+                GF_FREE (local->cont.lookup.postparents);
+
+                GF_FREE (local->cont.lookup.bufs);
+
+                GF_FREE (local->cont.lookup.success_children);
+
+                GF_FREE (local->cont.lookup.sources);
+                afr_matrix_cleanup (local->cont.lookup.pending_matrix,
+                                    priv->child_count);
+        }
+
+        { /* getxattr */
+                GF_FREE (local->cont.getxattr.name);
+        }
+
+        { /* lk */
+                GF_FREE (local->cont.lk.locked_nodes);
+        }
+
+        { /* create */
+                if (local->cont.create.fd)
+                        fd_unref (local->cont.create.fd);
+                if (local->cont.create.params)
+                        dict_unref (local->cont.create.params);
+        }
+
+        { /* mknod */
+                if (local->cont.mknod.params)
+                        dict_unref (local->cont.mknod.params);
+        }
+
+        { /* mkdir */
+                if (local->cont.mkdir.params)
+                        dict_unref (local->cont.mkdir.params);
+        }
+
+        { /* symlink */
+                if (local->cont.symlink.params)
+                        dict_unref (local->cont.symlink.params);
+        }
+
+        { /* writev */
+                GF_FREE (local->cont.writev.vector);
+        }
+
+        { /* setxattr */
+                if (local->cont.setxattr.dict)
+                        dict_unref (local->cont.setxattr.dict);
+        }
+
+        { /* fsetxattr */
+                if (local->cont.fsetxattr.dict)
+                        dict_unref (local->cont.fsetxattr.dict);
+        }
+
+        { /* removexattr */
+                GF_FREE (local->cont.removexattr.name);
+        }
+        { /* xattrop */
+                if (local->cont.xattrop.xattr)
+                        dict_unref (local->cont.xattrop.xattr);
+        }
+        { /* fxattrop */
+                if (local->cont.fxattrop.xattr)
+                        dict_unref (local->cont.fxattrop.xattr);
+        }
+        { /* symlink */
+                GF_FREE (local->cont.symlink.linkpath);
+        }
+
+        { /* opendir */
+                GF_FREE (local->cont.opendir.checksum);
+        }
+
+        { /* readdirp */
+                if (local->cont.readdir.dict)
+                        dict_unref (local->cont.readdir.dict);
+        }
+
+        if (local->xdata_req)
+                dict_unref (local->xdata_req);
+
+        if (local->xdata_rsp)
+                dict_unref (local->xdata_rsp);
+}
+
+
+int
+afr_frame_return (call_frame_t *frame)
+{
+        afr_local_t *local = NULL;
+        int          call_count = 0;
+
+        local = frame->local;
+
+        LOCK (&frame->lock);
+        {
+                call_count = --local->call_count;
+        }
+        UNLOCK (&frame->lock);
+
+        return call_count;
+}
+
+int
+afr_set_elem_count_get (unsigned char *elems, int child_count)
+{
+        int i   = 0;
+        int ret = 0;
+
+        for (i = 0; i < child_count; i++)
+                if (elems[i])
+                        ret++;
+        return ret;
+}
+
+/**
+ * up_children_count - return the number of children that are up
+ */
+
+unsigned int
+afr_up_children_count (unsigned char *child_up, unsigned int child_count)
+{
+        return afr_set_elem_count_get (child_up, child_count);
+}
+
+unsigned int
+afr_locked_children_count (unsigned char *children, unsigned int child_count)
+{
+        return afr_set_elem_count_get (children, child_count);
+}
+
+unsigned int
+afr_pre_op_done_children_count (unsigned char *pre_op,
+                                unsigned int child_count)
+{
+        return afr_set_elem_count_get (pre_op, child_count);
+}
+
+gf_boolean_t
+afr_is_fresh_lookup (loc_t *loc, xlator_t *this)
+{
+        uint64_t          ctx = 0;
+        int32_t           ret = 0;
+
+        GF_ASSERT (loc);
+        GF_ASSERT (this);
+        GF_ASSERT (loc->inode);
+
+        ret = inode_ctx_get (loc->inode, this, &ctx);
+        if (0 == ret)
+                return _gf_false;
+        return _gf_true;
+}
+
+void
+afr_update_loc_gfids (loc_t *loc, struct iatt *buf, struct iatt *postparent)
+{
+        GF_ASSERT (loc);
+        GF_ASSERT (buf);
+
+        uuid_copy (loc->gfid, buf->ia_gfid);
+        if (postparent)
+                uuid_copy (loc->pargfid, postparent->ia_gfid);
+}
+
+/*
+ * Quota size xattrs are not maintained by afr. There is a
+ * possibility that they differ even when both the directory changelog xattrs
+ * suggest everything is fine. So if there is at least one 'source' check among
+ * the sources which has the maximum quota size. Otherwise check among all the
+ * available ones for maximum quota size. This way if there is a source and
+ * stale copies it always votes for the 'source'.
+ * */
+
+static void
+afr_handle_quota_size (afr_local_t *local, xlator_t *this,
+                       dict_t *rsp_dict)
+{
+        int32_t       *sources       = NULL;
+        dict_t        *xattr         = NULL;
+        data_t        *max_data      = NULL;
+        int64_t       max_quota_size = -1;
+        data_t        *data          = NULL;
+        int64_t       *size          = NULL;
+        int64_t       quota_size     = -1;
+        afr_private_t *priv          = NULL;
+        int           i              = 0;
+        int           ret            = -1;
+        gf_boolean_t  source_present = _gf_false;
+
+        priv    = this->private;
+        sources = local->cont.lookup.sources;
+
+        if (rsp_dict == NULL) {
+                gf_log_callingfn (this->name, GF_LOG_ERROR, "%s: Invalid "
+                                  "response dictionary", local->loc.path);
+                return;
+        }
+
+        for (i = 0; i < priv->child_count; i++) {
+                if (sources[i]) {
+                        source_present = _gf_true;
+                        break;
+                }
+        }
+
+        for (i = 0; i < priv->child_count; i++) {
+                /*
+                 * If there is at least one source lets check
+                 * for maximum quota sizes among sources, otherwise take the
+                 * maximum of the ones present to be on the safer side.
+                 */
+                if (source_present && !sources[i])
+                        continue;
+
+                xattr = local->cont.lookup.xattrs[i];
+                if (!xattr)
+                        continue;
+
+                data = dict_get (xattr, QUOTA_SIZE_KEY);
+                if (!data)
+                        continue;
+
+                size = (int64_t*)data->data;
+                quota_size = ntoh64(*size);
+                gf_log (this->name, GF_LOG_DEBUG, "%s: %d, size: %"PRId64,
+                        local->loc.path, i, quota_size);
+                if (quota_size > max_quota_size) {
+                        if (max_data)
+                                data_unref (max_data);
+
+                        max_quota_size = quota_size;
+                        max_data = data_ref (data);
+                }
+        }
+
+        if (max_data) {
+                ret = dict_set (rsp_dict, QUOTA_SIZE_KEY, max_data);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "%s: Failed to set "
+                                "quota size", local->loc.path);
+                }
+
+                data_unref (max_data);
+        }
+}
+
+int
+afr_lookup_build_response_params (afr_local_t *local, xlator_t *this)
+{
+        struct iatt     *buf = NULL;
+        struct iatt     *postparent = NULL;
+        dict_t          **xattr = NULL;
+        int32_t         *success_children = NULL;
+        int32_t         *sources = NULL;
+        afr_private_t   *priv = NULL;
+        int32_t         read_child = -1;
+        int             ret = 0;
+        int             i = 0;
+
+        GF_ASSERT (local);
+
+        buf = &local->cont.lookup.buf;
+        postparent = &local->cont.lookup.postparent;
+        xattr = &local->cont.lookup.xattr;
+        priv = this->private;
+
+        read_child = afr_inode_get_read_ctx (this, local->cont.lookup.inode,
+                                             local->fresh_children);
+        if (read_child < 0) {
+                ret = -1;
+                goto out;
+        }
+        success_children = local->cont.lookup.success_children;
+        sources = local->cont.lookup.sources;
+        memset (sources, 0, sizeof (*sources) * priv->child_count);
+        afr_children_intersection_get (local->fresh_children, success_children,
+                                       sources, priv->child_count);
+        if (!sources[read_child]) {
+                read_child = -1;
+                for (i = 0; i < priv->child_count; i++) {
+                        if (sources[i]) {
+                                read_child = i;
+                                break;
+                        }
+                }
+        }
+        if (read_child < 0) {
+                ret = -1;
+                goto out;
+        }
+
+        gf_log (this->name, GF_LOG_DEBUG, "Building lookup response from %d",
+                read_child);
+        if (!*xattr)
+                *xattr = dict_ref (local->cont.lookup.xattrs[read_child]);
+
+        *buf = local->cont.lookup.bufs[read_child];
+        *postparent = local->cont.lookup.postparents[read_child];
+
+        if (dict_get (local->xattr_req, QUOTA_SIZE_KEY))
+                afr_handle_quota_size (local, this, *xattr);
+
+        if (IA_INVAL == local->cont.lookup.inode->ia_type) {
+                /* fix for RT #602 */
+                local->cont.lookup.inode->ia_type = buf->ia_type;
+        }
+out:
+        return ret;
+}
+
+static void
+afr_lookup_update_lk_counts (afr_local_t *local, xlator_t *this,
+                            int child_index, dict_t *xattr)
+{
+        uint32_t inodelk_count = 0;
+        uint32_t entrylk_count = 0;
+        int      ret           = -1;
+        uint32_t parent_entrylk = 0;
+
+        GF_ASSERT (local);
+        GF_ASSERT (this);
+        GF_ASSERT (xattr);
+        GF_ASSERT (child_index >= 0);
+
+        ret = dict_get_uint32 (xattr, GLUSTERFS_INODELK_COUNT,
+                               &inodelk_count);
+        if (ret == 0)
+                local->inodelk_count += inodelk_count;
+
+        ret = dict_get_uint32 (xattr, GLUSTERFS_ENTRYLK_COUNT,
+                               &entrylk_count);
+        if (ret == 0)
+                local->entrylk_count += entrylk_count;
+        ret = dict_get_uint32 (xattr, GLUSTERFS_PARENT_ENTRYLK,
+                               &parent_entrylk);
+        if (!ret)
+                local->cont.lookup.parent_entrylk += parent_entrylk;
+}
+
+/*
+ * It's important to maintain a commutative property on do_*_self_heal and
+ * found*; once set, they must not be cleared by a subsequent iteration or
+ * call, so that they represent a logical OR of all iterations and calls
+ * regardless of child/key order.  That allows the caller to call us multiple
+ * times without having to use a separate variable as a "reduce" accumulator.
+ */
+static void
+afr_lookup_set_self_heal_params_by_xattr (afr_local_t *local, xlator_t *this,
+                                          dict_t *xattr)
+{
+        afr_private_t *priv        = NULL;
+        int            i           = 0;
+        int            ret         = -1;
+        void          *pending_raw = NULL;
+        int32_t       *pending     = NULL;
+
+        GF_ASSERT (local);
+        GF_ASSERT (this);
+        GF_ASSERT (xattr);
+
+        priv = this->private;
+
+        for (i = 0; i < priv->child_count; i++) {
+                ret = dict_get_ptr (xattr, priv->pending_key[i],
+                                    &pending_raw);
+                if (ret != 0) {
+                        continue;
+                }
+                pending = pending_raw;
+
+                if (pending[AFR_METADATA_TRANSACTION]) {
+                        gf_log(this->name, GF_LOG_DEBUG,
+                               "metadata self-heal is pending for %s.",
+                               local->loc.path);
+                        local->self_heal.do_metadata_self_heal = _gf_true;
+                }
+
+                if (pending[AFR_ENTRY_TRANSACTION]) {
+                        gf_log(this->name, GF_LOG_DEBUG,
+                               "entry self-heal is pending for %s.",
+                               local->loc.path);
+                        local->self_heal.do_entry_self_heal = _gf_true;
+                }
+
+                if (pending[AFR_DATA_TRANSACTION]) {
+                        gf_log(this->name, GF_LOG_DEBUG,
+                               "data self-heal is pending for %s.",
+                               local->loc.path);
+                        local->self_heal.do_data_self_heal = _gf_true;
+                }
+        }
+}
+
+void
+afr_lookup_check_set_metadata_split_brain (afr_local_t *local, xlator_t *this)
+{
+        int32_t                  *sources = NULL;
+        afr_private_t            *priv = NULL;
+        int32_t                  subvol_status = 0;
+        int32_t                  *success_children   = NULL;
+        dict_t                   **xattrs = NULL;
+        struct iatt              *bufs = NULL;
+        int32_t                  **pending_matrix = NULL;
+
+        priv = this->private;
+
+        sources = GF_CALLOC (priv->child_count, sizeof (*sources),
+                             gf_afr_mt_int32_t);
+        if (NULL == sources)
+                goto out;
+        success_children = local->cont.lookup.success_children;
+        xattrs = local->cont.lookup.xattrs;
+        bufs = local->cont.lookup.bufs;
+        pending_matrix = local->cont.lookup.pending_matrix;
+        afr_build_sources (this, xattrs, bufs, pending_matrix,
+                           sources, success_children, AFR_METADATA_TRANSACTION,
+                           &subvol_status, _gf_false);
+        if (subvol_status & SPLIT_BRAIN)
+                local->cont.lookup.possible_spb = _gf_true;
+out:
+        GF_FREE (sources);
+}
+
+static void
+afr_detect_self_heal_by_iatt (afr_local_t *local, xlator_t *this,
+                            struct iatt *buf, struct iatt *lookup_buf)
+{
+        if (PERMISSION_DIFFERS (buf, lookup_buf)) {
+                /* mismatching permissions */
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "permissions differ for %s ", local->loc.path);
+                local->self_heal.do_metadata_self_heal = _gf_true;
+        }
+
+        if (OWNERSHIP_DIFFERS (buf, lookup_buf)) {
+                /* mismatching permissions */
+                local->self_heal.do_metadata_self_heal = _gf_true;
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "ownership differs for %s ", local->loc.path);
+        }
+
+        if (SIZE_DIFFERS (buf, lookup_buf)
+            && IA_ISREG (buf->ia_type)) {
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "size differs for %s ", local->loc.path);
+                local->self_heal.do_data_self_heal = _gf_true;
+        }
+
+        if (uuid_compare (buf->ia_gfid, lookup_buf->ia_gfid)) {
+                /* mismatching gfid */
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "%s: gfid different on subvolume", local->loc.path);
+        }
+}
+
+static void
+afr_detect_self_heal_by_split_brain_status (afr_local_t *local, xlator_t *this)
+{
+        gf_boolean_t split_brain = _gf_false;
+        afr_self_heal_t *sh = NULL;
+
+        sh = &local->self_heal;
+
+        split_brain = afr_is_split_brain (this, local->cont.lookup.inode);
+        split_brain = split_brain || local->cont.lookup.possible_spb;
+        if ((local->success_count > 0) && split_brain &&
+            IA_ISREG (local->cont.lookup.inode->ia_type)) {
+                sh->force_confirm_spb = _gf_true;
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "split brain detected during lookup of %s.",
+                        local->loc.path);
+        }
+}
+
+static void
+afr_detect_self_heal_by_lookup_status (afr_local_t *local, xlator_t *this)
+{
+        GF_ASSERT (local);
+        GF_ASSERT (this);
+
+        if ((local->success_count > 0) && (local->enoent_count > 0)) {
+                local->self_heal.do_metadata_self_heal = _gf_true;
+                local->self_heal.do_data_self_heal     = _gf_true;
+                local->self_heal.do_entry_self_heal    = _gf_true;
+                local->self_heal.do_gfid_self_heal    = _gf_true;
+                local->self_heal.do_missing_entry_self_heal    = _gf_true;
+                gf_log(this->name, GF_LOG_DEBUG,
+                       "entries are missing in lookup of %s.",
+                       local->loc.path);
+        }
+
+        return;
+}
+
+gf_boolean_t
+afr_can_self_heal_proceed (afr_self_heal_t *sh, afr_private_t *priv)
+{
+        GF_ASSERT (sh);
+        GF_ASSERT (priv);
+
+        if (sh->force_confirm_spb)
+                return _gf_true;
+        return (sh->do_gfid_self_heal
+                || sh->do_missing_entry_self_heal
+                || (afr_data_self_heal_enabled (priv->data_self_heal) &&
+                    sh->do_data_self_heal)
+                || (priv->metadata_self_heal && sh->do_metadata_self_heal)
+                || (priv->entry_self_heal && sh->do_entry_self_heal));
+}
+
+afr_transaction_type
+afr_transaction_type_get (ia_type_t ia_type)
+{
+        afr_transaction_type    type = AFR_METADATA_TRANSACTION;
+
+        GF_ASSERT (ia_type != IA_INVAL);
+
+        if (IA_ISDIR (ia_type)) {
+                type = AFR_ENTRY_TRANSACTION;
+        } else if (IA_ISREG (ia_type)) {
+                type = AFR_DATA_TRANSACTION;
+        }
+        return type;
+}
+
+int
+afr_lookup_select_read_child (afr_local_t *local, xlator_t *this,
+                              int32_t *read_child)
+{
+        ia_type_t               ia_type        = IA_INVAL;
+        int32_t                 source         = -1;
+        int                     ret            = -1;
+        dict_t                  **xattrs       = NULL;
+        int32_t                 *success_children = NULL;
+        afr_transaction_type    type           = AFR_METADATA_TRANSACTION;
+        uuid_t                  *gfid          = NULL;
+
+        GF_ASSERT (local);
+        GF_ASSERT (this);
+        GF_ASSERT (local->success_count > 0);
+
+        success_children = local->cont.lookup.success_children;
+        /*We can take the success_children[0] only because we already
+         *handle the conflicting children other wise, we could select the
+         *read_child based on wrong file type
+         */
+        ia_type = local->cont.lookup.bufs[success_children[0]].ia_type;
+        type = afr_transaction_type_get (ia_type);
+        xattrs = local->cont.lookup.xattrs;
+        gfid = &local->cont.lookup.buf.ia_gfid;
+        source = afr_lookup_select_read_child_by_txn_type (this, local, xattrs,
+                                                           type, *gfid);
+        if (source < 0) {
+                gf_log (this->name, GF_LOG_DEBUG, "failed to select source "
+                        "for %s", local->loc.path);
+                goto out;
+        }
+
+        gf_log (this->name, GF_LOG_DEBUG, "Source selected as %d for %s",
+                source, local->loc.path);
+        *read_child = source;
+        ret = 0;
+out:
+        return ret;
+}
+
+static inline gf_boolean_t
+afr_is_transaction_running (afr_local_t *local)
+{
+        GF_ASSERT (local->fop == GF_FOP_LOOKUP);
+        return ((local->inodelk_count > 0) || (local->entrylk_count > 0));
+}
+
+void
+afr_launch_self_heal (call_frame_t *frame, xlator_t *this, inode_t *inode,
+                      gf_boolean_t background, ia_type_t ia_type, char *reason,
+                      void (*gfid_sh_success_cbk) (call_frame_t *sh_frame,
+                                                   xlator_t *this),
+                      int (*unwind) (call_frame_t *frame, xlator_t *this,
+                                     int32_t op_ret, int32_t op_errno,
+                                     int32_t sh_failed))
+{
+        afr_local_t             *local = NULL;
+        char                    sh_type_str[256] = {0,};
+        char                    *bg = "";
+
+        GF_ASSERT (frame);
+        GF_ASSERT (this);
+        GF_ASSERT (inode);
+        GF_ASSERT (ia_type != IA_INVAL);
+
+        local = frame->local;
+        local->self_heal.background = background;
+        local->self_heal.type       = ia_type;
+        local->self_heal.unwind     = unwind;
+        local->self_heal.gfid_sh_success_cbk     = gfid_sh_success_cbk;
+
+        afr_self_heal_type_str_get (&local->self_heal,
+                                    sh_type_str,
+                                    sizeof (sh_type_str));
+
+        if (background)
+                bg = "background";
+        gf_log (this->name, GF_LOG_DEBUG,
+                "%s %s self-heal triggered. path: %s, reason: %s", bg,
+                sh_type_str, local->loc.path, reason);
+
+        afr_self_heal (frame, this, inode);
+}
+
+unsigned int
+afr_gfid_missing_count (const char *xlator_name, int32_t *success_children,
+                        struct iatt *bufs, unsigned int child_count,
+                        const char *path)
+{
+        unsigned int    gfid_miss_count   = 0;
+        int             i              = 0;
+        struct iatt     *child1        = NULL;
+
+        for (i = 0; i < child_count; i++) {
+                if (success_children[i] == -1)
+                        break;
+                child1 = &bufs[success_children[i]];
+                if (uuid_is_null (child1->ia_gfid)) {
+                        gf_log (xlator_name, GF_LOG_DEBUG, "%s: gfid is null"
+                                " on subvolume %d", path, success_children[i]);
+                        gfid_miss_count++;
+                }
+        }
+
+        return gfid_miss_count;
+}
+
+static int
+afr_lookup_gfid_missing_count (afr_local_t *local, xlator_t *this)
+{
+        int32_t         *success_children = NULL;
+        afr_private_t   *priv          = NULL;
+        struct iatt     *bufs          = NULL;
+        int             miss_count     = 0;
+
+        priv = this->private;
+        bufs = local->cont.lookup.bufs;
+        success_children = local->cont.lookup.success_children;
+
+        miss_count =  afr_gfid_missing_count (this->name, success_children,
+                                              bufs, priv->child_count,
+                                              local->loc.path);
+        return miss_count;
+}
+
+gf_boolean_t
+afr_conflicting_iattrs (struct iatt *bufs, int32_t *success_children,
+                        unsigned int child_count, const char *path,
+                        const char *xlator_name)
+{
+        gf_boolean_t    conflicting    = _gf_false;
+        int             i              = 0;
+        struct iatt     *child1        = NULL;
+        struct iatt     *child2        = NULL;
+        uuid_t          *gfid          = NULL;
+
+        for (i = 0; i < child_count; i++) {
+                if (success_children[i] == -1)
+                        break;
+                child1 = &bufs[success_children[i]];
+                if ((!gfid) && (!uuid_is_null (child1->ia_gfid)))
+                        gfid = &child1->ia_gfid;
+
+                if (i == 0)
+                        continue;
+
+                child2 = &bufs[success_children[i-1]];
+                if (FILETYPE_DIFFERS (child1, child2)) {
+                        gf_log (xlator_name, GF_LOG_DEBUG, "%s: filetype "
+                                "differs on subvolumes (%d, %d)", path,
+                                success_children[i-1], success_children[i]);
+                        conflicting = _gf_true;
+                        goto out;
+                }
+                if (!gfid || uuid_is_null (child1->ia_gfid))
+                        continue;
+                if (uuid_compare (*gfid, child1->ia_gfid)) {
+                       gf_log (xlator_name, GF_LOG_DEBUG, "%s: gfid differs"
+                               " on subvolume %d", path, success_children[i]);
+                       conflicting = _gf_true;
+                       goto out;
+                }
+        }
+out:
+        return conflicting;
+}
+
+/* afr_update_gfid_from_iatts: This function should be called only if the
+ * iatts are not conflicting.
+ */
+void
+afr_update_gfid_from_iatts (uuid_t uuid, struct iatt *bufs,
+                            int32_t *success_children, unsigned int child_count)
+{
+        uuid_t          *gfid = NULL;
+        int             i = 0;
+        int             child = 0;
+
+        for (i = 0; i < child_count; i++) {
+                child = success_children[i];
+                if (child == -1)
+                        break;
+                if ((!gfid) && (!uuid_is_null (bufs[child].ia_gfid))) {
+                        gfid = &bufs[child].ia_gfid;
+                } else if (gfid && (!uuid_is_null (bufs[child].ia_gfid))) {
+                        if (uuid_compare (*gfid, bufs[child].ia_gfid)) {
+                                GF_ASSERT (0);
+                                goto out;
+                        }
+                }
+        }
+        if (gfid && (!uuid_is_null (*gfid)))
+                uuid_copy (uuid, *gfid);
+out:
+        return;
+}
+
+static gf_boolean_t
+afr_lookup_conflicting_entries (afr_local_t *local, xlator_t *this)
+{
+        afr_private_t           *priv = NULL;
+        gf_boolean_t            conflict = _gf_false;
+
+        priv = this->private;
+        conflict =  afr_conflicting_iattrs (local->cont.lookup.bufs,
+                                            local->cont.lookup.success_children,
+                                            priv->child_count, local->loc.path,
+                                            this->name);
+        return conflict;
+}
+
+gf_boolean_t
+afr_open_only_data_self_heal (char *data_self_heal)
+{
+        return !strcmp (data_self_heal, "open");
+}
+
+gf_boolean_t
+afr_data_self_heal_enabled (char *data_self_heal)
+{
+        gf_boolean_t    enabled = _gf_false;
+
+        if (gf_string2boolean (data_self_heal, &enabled) == -1) {
+                enabled = !strcmp (data_self_heal, "open");
+                GF_ASSERT (enabled);
+        }
+
+        return enabled;
+}
+
+static void
+afr_lookup_set_self_heal_params (afr_local_t *local, xlator_t *this)
+{
+        int                     i = 0;
+        struct iatt             *bufs = NULL;
+        dict_t                  **xattr = NULL;
+        afr_private_t           *priv = NULL;
+        int32_t                 child1 = -1;
+        int32_t                 child2 = -1;
+        afr_self_heal_t         *sh = NULL;
+
+        priv  = this->private;
+        sh = &local->self_heal;
+
+        afr_detect_self_heal_by_lookup_status (local, this);
+
+        if (afr_lookup_gfid_missing_count (local, this))
+                local->self_heal.do_gfid_self_heal    = _gf_true;
+
+        if (_gf_true == afr_lookup_conflicting_entries (local, this))
+                local->self_heal.do_missing_entry_self_heal    = _gf_true;
+        else
+                afr_update_gfid_from_iatts (local->self_heal.sh_gfid_req,
+                                            local->cont.lookup.bufs,
+                                            local->cont.lookup.success_children,
+                                            priv->child_count);
+
+        bufs = local->cont.lookup.bufs;
+        for (i = 1; i < local->success_count; i++) {
+                child1 = local->cont.lookup.success_children[i-1];
+                child2 = local->cont.lookup.success_children[i];
+                afr_detect_self_heal_by_iatt (local, this,
+                                              &bufs[child1], &bufs[child2]);
+        }
+
+        xattr = local->cont.lookup.xattrs;
+        for (i = 0; i < local->success_count; i++) {
+                child1 = local->cont.lookup.success_children[i];
+                afr_lookup_set_self_heal_params_by_xattr (local, this,
+                                                          xattr[child1]);
+        }
+        if (afr_open_only_data_self_heal (priv->data_self_heal))
+                sh->do_data_self_heal = _gf_false;
+        if (sh->do_metadata_self_heal)
+                afr_lookup_check_set_metadata_split_brain (local, this);
+        afr_detect_self_heal_by_split_brain_status (local, this);
+}
+
+int
+afr_self_heal_lookup_unwind (call_frame_t *frame, xlator_t *this,
+                             int32_t op_ret, int32_t op_errno,
+                             int32_t sh_failed)
+{
+        afr_local_t *local = NULL;
+        int         ret    = -1;
+        dict_t      *xattr = NULL;
+
+        local = frame->local;
+
+        if (op_ret == -1) {
+                local->op_ret = -1;
+		local->op_errno = afr_most_important_error(local->op_errno,
+							   op_errno, _gf_true);
+
+                goto out;
+        } else {
+                local->op_ret = 0;
+        }
+
+        afr_lookup_done_success_action (frame, this, _gf_true);
+        xattr = local->cont.lookup.xattr;
+        if (xattr) {
+                ret = dict_set_int32 (xattr, "sh-failed", sh_failed);
+                if (ret)
+                        gf_log (this->name, GF_LOG_ERROR, "%s: Failed to set "
+                                "sh-failed to %d", local->loc.path, sh_failed);
+
+                if (local->self_heal.actual_sh_started == _gf_true &&
+                    sh_failed == 0) {
+                        ret = dict_set_int32 (xattr, "actual-sh-done", 1);
+                        if (ret)
+                                gf_log(this->name, GF_LOG_ERROR, "%s: Failed to"
+                                       " set actual-sh-done to %d",
+                                       local->loc.path,
+                                       local->self_heal.actual_sh_started);
+                }
+        }
+out:
+        AFR_STACK_UNWIND (lookup, frame, local->op_ret, local->op_errno,
+                          local->cont.lookup.inode, &local->cont.lookup.buf,
+                          local->cont.lookup.xattr,
+                          &local->cont.lookup.postparent);
+
+        return 0;
+}
+
+//TODO: At the moment only lookup needs this, so not doing any checks, in the
+// future we will have to do fop specific operations
+void
+afr_post_gfid_sh_success (call_frame_t *sh_frame, xlator_t *this)
+{
+        afr_local_t             *local = NULL;
+        afr_local_t             *sh_local = NULL;
+        afr_private_t           *priv = NULL;
+        afr_self_heal_t         *sh = NULL;
+        int                     i = 0;
+        struct iatt             *lookup_bufs = NULL;
+        struct iatt             *lookup_parentbufs = NULL;
+
+        sh_local = sh_frame->local;
+        sh       = &sh_local->self_heal;
+        local = sh->orig_frame->local;
+        lookup_bufs = local->cont.lookup.bufs;
+        lookup_parentbufs = local->cont.lookup.postparents;
+        priv = this->private;
+
+        memcpy (lookup_bufs, sh->buf, priv->child_count * sizeof (*sh->buf));
+        memcpy (lookup_parentbufs, sh->parentbufs,
+                priv->child_count * sizeof (*sh->parentbufs));
+
+        afr_reset_xattr (local->cont.lookup.xattrs, priv->child_count);
+        if (local->cont.lookup.xattr) {
+                dict_unref (local->cont.lookup.xattr);
+                local->cont.lookup.xattr = NULL;
+        }
+
+        for (i = 0; i < priv->child_count; i++) {
+                if (sh->xattr[i])
+                        local->cont.lookup.xattrs[i] = dict_ref (sh->xattr[i]);
+        }
+
+        afr_reset_children (local->cont.lookup.success_children,
+                            priv->child_count);
+        afr_children_copy (local->cont.lookup.success_children,
+                           sh->fresh_children, priv->child_count);
+}
+
+static void
+afr_lookup_perform_self_heal (call_frame_t *frame, xlator_t *this,
+                              gf_boolean_t *sh_launched)
+{
+        unsigned int         up_count = 0;
+        afr_private_t       *priv    = NULL;
+        afr_local_t         *local   = NULL;
+        char                *reason  = NULL;
+
+        GF_ASSERT (sh_launched);
+        *sh_launched = _gf_false;
+        priv         = this->private;
+        local        = frame->local;
+
+        up_count  = afr_up_children_count (local->child_up, priv->child_count);
+        if (up_count == 1) {
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "Only 1 child up - do not attempt to detect self heal");
+                goto out;
+        }
+
+        afr_lookup_set_self_heal_params (local, this);
+        if (afr_can_self_heal_proceed (&local->self_heal, priv)) {
+                if  (afr_is_transaction_running (local) &&
+                     (!local->allow_sh_for_running_transaction))
+                        goto out;
+
+                reason = "lookup detected pending operations";
+                afr_launch_self_heal (frame, this, local->cont.lookup.inode,
+                                      _gf_true, local->cont.lookup.buf.ia_type,
+                                      reason, afr_post_gfid_sh_success,
+                                      afr_self_heal_lookup_unwind);
+                *sh_launched = _gf_true;
+        }
+out:
+        return;
+}
+
+void
+afr_get_fresh_children (int32_t *success_children, int32_t *sources,
+                        int32_t *fresh_children, unsigned int child_count)
+{
+        unsigned int i = 0;
+        unsigned int j = 0;
+
+        GF_ASSERT (success_children);
+        GF_ASSERT (sources);
+        GF_ASSERT (fresh_children);
+
+        afr_reset_children (fresh_children, child_count);
+        for (i = 0; i < child_count; i++) {
+                if (success_children[i] == -1)
+                        break;
+                if (afr_is_read_child (success_children, sources, child_count,
+                                       success_children[i])) {
+                        fresh_children[j] = success_children[i];
+                        j++;
+                }
+        }
+}
+
+static int
+afr_lookup_set_read_ctx (afr_local_t *local, xlator_t *this, int32_t read_child)
+{
+        afr_private_t           *priv = NULL;
+
+        GF_ASSERT (read_child >= 0);
+
+        priv = this->private;
+        afr_get_fresh_children (local->cont.lookup.success_children,
+                                local->cont.lookup.sources,
+                                local->fresh_children, priv->child_count);
+        afr_inode_set_read_ctx (this, local->cont.lookup.inode, read_child,
+                                local->fresh_children);
+
+        return 0;
+}
+
+int
+afr_lookup_done_success_action (call_frame_t *frame, xlator_t *this,
+                                gf_boolean_t fail_conflict)
+{
+        int32_t             read_child = -1;
+        int32_t             ret        = -1;
+        afr_local_t         *local     = NULL;
+        gf_boolean_t        fresh_lookup = _gf_false;
+
+        local   = frame->local;
+        fresh_lookup = local->cont.lookup.fresh_lookup;
+
+        if (local->loc.parent == NULL)
+                fail_conflict = _gf_true;
+
+        if (afr_lookup_conflicting_entries (local, this)) {
+                if (fail_conflict == _gf_false)
+                        ret = 0;
+                goto out;
+        }
+
+        ret = afr_lookup_select_read_child (local, this, &read_child);
+        if (!afr_is_transaction_running (local) || fresh_lookup) {
+                if (read_child < 0)
+                        goto out;
+
+                ret = afr_lookup_set_read_ctx (local, this, read_child);
+                if (ret)
+                        goto out;
+        }
+
+        ret = afr_lookup_build_response_params (local, this);
+        if (ret)
+                goto out;
+        afr_update_loc_gfids (&local->loc,
+                              &local->cont.lookup.buf,
+                              &local->cont.lookup.postparent);
+
+        ret = 0;
+out:
+        if (ret) {
+                local->op_ret = -1;
+                local->op_errno = EIO;
+        }
+        return ret;
+}
+
+int
+afr_lookup_get_latest_subvol (afr_local_t *local, xlator_t *this)
+{
+        afr_private_t *priv = NULL;
+        int32_t       *success_children = NULL;
+        struct iatt   *bufs = NULL;
+        int           i = 0;
+        int           child = 0;
+        int           lsubvol = -1;
+
+        priv = this->private;
+        success_children = local->cont.lookup.success_children;
+        bufs = local->cont.lookup.bufs;
+        for (i = 0; i < priv->child_count; i++) {
+                child = success_children[i];
+                if (child == -1)
+                        break;
+                if (uuid_is_null (bufs[child].ia_gfid))
+                        continue;
+                if (lsubvol < 0) {
+                        lsubvol = child;
+                } else if (bufs[lsubvol].ia_ctime < bufs[child].ia_ctime) {
+                        lsubvol = child;
+                } else if ((bufs[lsubvol].ia_ctime == bufs[child].ia_ctime) &&
+                  (bufs[lsubvol].ia_ctime_nsec < bufs[child].ia_ctime_nsec)) {
+                        lsubvol = child;
+                }
+        }
+        return lsubvol;
+}
+
+void
+afr_lookup_mark_other_entries_stale (afr_local_t *local, xlator_t *this,
+                                     int subvol)
+{
+        afr_private_t *priv = NULL;
+        int32_t       *success_children = NULL;
+        struct iatt   *bufs = NULL;
+        int           i = 0;
+        int           child = 0;
+
+        priv = this->private;
+        success_children = local->cont.lookup.success_children;
+        bufs = local->cont.lookup.bufs;
+        memcpy (local->fresh_children, success_children,
+                sizeof (*success_children) * priv->child_count);
+        for (i = 0; i < priv->child_count; i++) {
+                child = local->fresh_children[i];
+                if (child == -1)
+                        break;
+                if (child == subvol)
+                        continue;
+                if (uuid_is_null (bufs[child].ia_gfid) &&
+                    (bufs[child].ia_type == bufs[subvol].ia_type))
+                        continue;
+                afr_children_rm_child (success_children, child,
+                                       priv->child_count);
+                local->success_count--;
+        }
+        afr_reset_children (local->fresh_children, priv->child_count);
+}
+
+void
+afr_succeed_lookup_on_latest_iatt (afr_local_t *local, xlator_t *this)
+{
+        int    lsubvol = 0;
+
+        if (!afr_lookup_conflicting_entries (local, this))
+                goto out;
+
+        lsubvol = afr_lookup_get_latest_subvol (local, this);
+        if (lsubvol < 0)
+                goto out;
+        afr_lookup_mark_other_entries_stale (local, this, lsubvol);
+out:
+        return;
+}
+
+gf_boolean_t
+afr_is_entry_possibly_under_creation (afr_local_t *local, xlator_t *this)
+{
+        /*
+         * We need to perform this test in lookup done and treat on going
+         * create/DELETE as ENOENT.
+         * Reason:
+        Multiple clients A, B and C are attempting 'mkdir -p /mnt/a/b/c'
+
+        1 Client A is in the middle of mkdir(/a). It has acquired lock.
+          It has performed mkdir(/a) on one subvol, and second one is still
+          in progress
+        2 Client B performs a lookup, sees directory /a on one,
+          ENOENT on the other, succeeds lookup.
+        3 Client B performs lookup on /a/b on both subvols, both return ENOENT
+          (one subvol because /a/b does not exist, another because /a
+          itself does not exist)
+        4 Client B proceeds to mkdir /a/b. It obtains entrylk on inode=/a with
+          basename=b on one subvol, but fails on other subvol as /a is yet to
+          be created by Client A.
+        5 Client A finishes mkdir of /a on other subvol
+        6 Client C also attempts to create /a/b, lookup returns ENOENT on
+          both subvols.
+        7 Client C tries to obtain entrylk on on inode=/a with basename=b,
+          obtains on one subvol (where B had failed), and waits for B to unlock
+          on other subvol.
+        8 Client B finishes mkdir() on one subvol with GFID-1 and completes
+          transaction and unlocks
+        9 Client C gets the lock on the second subvol, At this stage second
+          subvol already has /a/b created from Client B, but Client C does not
+          check that in the middle of mkdir transaction
+        10 Client C attempts mkdir /a/b on both subvols. It succeeds on
+           ONLY ONE (where Client B could not get lock because of
+           missing parent /a dir) with GFID-2, and gets EEXIST from ONE subvol.
+        This way we have /a/b in GFID mismatch. One subvol got GFID-1 because
+        Client B performed transaction on only one subvol (because entrylk()
+        could not be obtained on second subvol because of missing parent dir --
+        caused by premature/speculative succeeding of lookup() on /a when locks
+        are detected). Other subvol gets GFID-2 from Client C because while
+        it was waiting for entrylk() on both subvols, Client B was in the
+        middle of creating mkdir() on only one subvol, and Client C does not
+        "expect" this when it is between lock() and pre-op()/op() phase of the
+        transaction.
+         */
+	if (local->cont.lookup.parent_entrylk && local->enoent_count)
+		return _gf_true;
+
+	return _gf_false;
+}
+
+
+static void
+afr_lookup_done (call_frame_t *frame, xlator_t *this)
+{
+        int                 unwind = 1;
+        afr_private_t       *priv  = NULL;
+        afr_local_t         *local = NULL;
+        int                 ret = -1;
+        gf_boolean_t        sh_launched = _gf_false;
+        gf_boolean_t        fail_conflict = _gf_false;
+        int                 gfid_miss_count = 0;
+        int                 enotconn_count = 0;
+        int                 up_children_count = 0;
+
+        priv  = this->private;
+        local = frame->local;
+
+	if (afr_is_entry_possibly_under_creation (local, this)) {
+		local->op_ret = -1;
+		local->op_errno = ENOENT;
+		goto unwind;
+	}
+
+        if (local->op_ret < 0)
+                goto unwind;
+
+        if (local->cont.lookup.parent_entrylk && local->success_count > 1)
+                afr_succeed_lookup_on_latest_iatt (local, this);
+
+        gfid_miss_count = afr_lookup_gfid_missing_count (local, this);
+        up_children_count = afr_up_children_count (local->child_up,
+                                                   priv->child_count);
+        enotconn_count = priv->child_count - up_children_count;
+        if ((gfid_miss_count == local->success_count) &&
+            (enotconn_count > 0)) {
+                local->op_ret = -1;
+                local->op_errno = EIO;
+                gf_log (this->name, GF_LOG_ERROR, "Failing lookup for %s, "
+                        "LOOKUP on a file without gfid is not allowed when "
+                        "some of the children are down", local->loc.path);
+                goto unwind;
+        }
+
+        if ((gfid_miss_count == local->success_count) &&
+            uuid_is_null (local->cont.lookup.gfid_req)) {
+                local->op_ret = -1;
+                local->op_errno = ENODATA;
+                gf_log (this->name, GF_LOG_ERROR, "%s: No gfid present",
+                        local->loc.path);
+                goto unwind;
+        }
+
+        if (gfid_miss_count && uuid_is_null (local->cont.lookup.gfid_req))
+                fail_conflict = _gf_true;
+        ret = afr_lookup_done_success_action (frame, this, fail_conflict);
+        if (ret)
+                goto unwind;
+        uuid_copy (local->self_heal.sh_gfid_req, local->cont.lookup.gfid_req);
+
+        afr_lookup_perform_self_heal (frame, this, &sh_launched);
+        if (sh_launched) {
+                unwind = 0;
+                goto unwind;
+        }
+
+ unwind:
+         if (unwind) {
+                 AFR_STACK_UNWIND (lookup, frame, local->op_ret,
+                                   local->op_errno, local->cont.lookup.inode,
+                                   &local->cont.lookup.buf,
+                                   local->cont.lookup.xattr,
+                                   &local->cont.lookup.postparent);
+        }
+}
+
+/*
+ * During a lookup, some errors are more "important" than
+ * others in that they must be given higher priority while
+ * returning to the user.
+ *
+ * The hierarchy is ESTALE > EIO > ENOENT > others
+ */
+int32_t
+afr_most_important_error(int32_t old_errno, int32_t new_errno,
+			 gf_boolean_t eio)
+{
+	if (old_errno == ESTALE || new_errno == ESTALE)
+		return ESTALE;
+	if (eio && (old_errno == EIO || new_errno == EIO))
+		return EIO;
+	if (old_errno == ENOENT || new_errno == ENOENT)
+		return ENOENT;
+
+	return new_errno;
+}
+
+int32_t
+afr_resultant_errno_get (int32_t *children,
+                         int *child_errno, unsigned int child_count)
+{
+        int     i = 0;
+        int32_t op_errno = 0;
+        int     child = 0;
author	Csaba Henk <csaba@redhat.com>	2012-04-10 15:34:02 +0200
committer	Vijay Bellur <vijay@gluster.com>	2012-04-13 00:25:47 -0700
commit	c238e5073a8d60fede1f88e584c14de42885276e (patch)
tree	b7316b1929c67c7b28d808fea8b2c654b5870796 /cli/src/cli-cmd.c
parent	ecfe2d0a94dd6626e2e6c1e7707549adf550a5af (diff)