rpc-clnt: separate out connection_cleanup() from destroy()v3.3.0qa30

noticed that there are possibilities where one would like to do a connection_cleanup() before destroying a RPC connection itself, also current code is such that, rpc_clnt_connection_cleanup() does rpc_clnt_ref() and unref(), creating a race window/double unref possibilities in the code. by separating out the functions, this race window/double fault can be prevented. Change-Id: I7ebd3392efa891232857b6db9108b0b19e40fc12 Signed-off-by: Amar Tumballi <amarts@redhat.com> BUG: 802403 Reviewed-on: http://review.gluster.com/2979 Tested-by: Gluster Build System <jenkins@build.gluster.com> Reviewed-by: Vijay Bellur <vijay@gluster.com>
author: Amar Tumballi <amarts@redhat.com> 2012-03-19 18:10:34 +0530
committer: Vijay Bellur <vijay@gluster.com> 2012-03-19 09:14:40 -0700
commit: d98c3e19342be3b8b3e2c7e3d7e544de34143bdf (patch)
tree: ba1d1d18d33694c99749dd005d0d46e91020e7e1 /xlators/cluster/unify
parent: 55c22fbfae259bff3b0e0cca192f709b74d8bee5 (diff)
0 files changed, 0 insertions, 0 deletions
diff --git a/xlators/Makefile.am b/xlators/Makefile.am
index 4c94f5e44..f60fa85ce 100644
--- a/xlators/Makefile.am
+++ b/xlators/Makefile.am
@@ -1,3 +1,4 @@
-SUBDIRS = cluster storage protocol performance debug features encryption mount nfs mgmt
+SUBDIRS = cluster storage protocol performance debug features encryption mount nfs mgmt system \
+          playground
 
 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 3310a2115..9b96790de 100644
--- a/xlators/bindings/python/src/python.c
+++ b/xlators/bindings/python/src/python.c
@@ -1,22 +1,12 @@
 /*
-   Copyright (c) 2007-2010 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;
@@ -148,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;
   }
@@ -166,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);
@@ -188,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())
   {
@@ -217,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/afr/src/Makefile.am b/xlators/cluster/afr/src/Makefile.am
index 4e4b4c752..35d18a6c0 100644
--- a/xlators/cluster/afr/src/Makefile.am
+++ b/xlators/cluster/afr/src/Makefile.am
@@ -1,20 +1,31 @@
 xlator_LTLIBRARIES = afr.la pump.la
 xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/cluster
 
-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_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_LDFLAGS = -module -avoidversion
+afr_la_LDFLAGS = -module -avoid-version
 afr_la_SOURCES = $(afr_common_source) afr.c
 afr_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la
 
-pump_la_LDFLAGS = -module -avoidversion
+pump_la_LDFLAGS = -module -avoid-version
 pump_la_SOURCES =  $(afr_common_source) pump.c
 pump_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.h afr-mem-types.h afr-common.c
+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
 
-AM_CFLAGS = -fPIC -D_FILE_OFFSET_BITS=64 -D_GNU_SOURCE -Wall -D$(GF_HOST_OS) \
-	    -I$(top_srcdir)/libglusterfs/src -I$(top_srcdir)/contrib/md5 -shared -nostartfiles $(GF_CFLAGS)
+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 =
 
diff --git a/xlators/cluster/afr/src/afr-common.c b/xlators/cluster/afr/src/afr-common.c
index 81f01c125..af01f2ef2 100644
--- a/xlators/cluster/afr/src/afr-common.c
+++ b/xlators/cluster/afr/src/afr-common.c
@@ -1,20 +1,11 @@
 /*
-   Copyright (c) 2007-2010 Gluster, Inc. <http://www.gluster.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 Affero 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
-   Affero General Public License for more details.
-
-   You should have received a copy of the GNU Affero General Public License
-   along with this program.  If not, see
-   <http://www.gnu.org/licenses/>.
+  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>
@@ -44,6 +35,7 @@
 #include "compat.h"
 #include "byte-order.h"
 #include "statedump.h"
+#include "inode.h"
 
 #include "fd.h"
 
@@ -54,301 +46,827 @@
 #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     0x0000000200000000ULL
-#define AFR_ICTX_SPLIT_BRAIN_MASK      0x0000000100000000ULL
+#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;
+        int ret       = 0;
+        uuid_t *pgfid = NULL;
 
         GF_ASSERT (gfid);
 
-        ret = dict_set_static_bin (dict, "gfid-req", gfid, 16);
+        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_DEBUG, "gfid set failed");
+                gf_log (THIS->name, GF_LOG_ERROR, "gfid set failed");
+
+out:
+        if (ret && pgfid)
+                GF_FREE (pgfid);
 
         return ret;
 }
 
-uint64_t
-afr_is_split_brain (xlator_t *this, inode_t *inode)
+void
+afr_inode_ctx_destroy (afr_inode_ctx_t *ctx)
 {
-        int ret = 0;
+        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;
 
-        uint64_t ctx         = 0;
-        uint64_t split_brain = 0;
+fail:
+        afr_inode_ctx_destroy (ctx);
+        return NULL;
+}
 
-        VALIDATE_OR_GOTO (inode, out);
+afr_inode_ctx_t*
+afr_inode_ctx_get (inode_t *inode, xlator_t *this)
+{
+        afr_inode_ctx_t *ctx = NULL;
 
         LOCK (&inode->lock);
         {
-                ret = __inode_ctx_get (inode, this, &ctx);
+                ctx = __afr_inode_ctx_get (inode, this);
+        }
+        UNLOCK (&inode->lock);
+        return ctx;
+}
 
-                if (ret < 0)
-                        goto unlock;
+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;
 
-                split_brain = ctx & AFR_ICTX_SPLIT_BRAIN_MASK;
+        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 split_brain;
+        return spb;
 }
 
+gf_boolean_t
+afr_is_opendir_done (xlator_t *this, inode_t *inode)
+{
+        afr_inode_params_t params = {0};
 
-void
-afr_set_split_brain (xlator_t *this, inode_t *inode, gf_boolean_t set)
+        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)
 {
-        uint64_t ctx = 0;
-        int      ret = 0;
+        afr_inode_params_t      params = {0};
 
-        VALIDATE_OR_GOTO (inode, out);
+        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;
+}
 
-        LOCK (&inode->lock);
-        {
-                ret = __inode_ctx_get (inode, this, &ctx);
+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;
 
-                if (ret < 0) {
-                        ctx = 0;
-                }
+        remaining_mask = (~AFR_ICTX_READ_CHILD_MASK & ctx->masks);
+        mask = (AFR_ICTX_READ_CHILD_MASK & read_child);
+        ctx->masks = remaining_mask | mask;
+}
 
-                if (set) {
-                        ctx = (~AFR_ICTX_SPLIT_BRAIN_MASK & ctx)
-                                | (0xFFFFFFFFFFFFFFFFULL & AFR_ICTX_SPLIT_BRAIN_MASK);
-                } else {
-                        ctx = (~AFR_ICTX_SPLIT_BRAIN_MASK & ctx);
-                }
-                __inode_ctx_put (inode, this, ctx);
+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;
         }
-        UNLOCK (&inode->lock);
-out:
-        return;
 }
 
+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;
 
-uint64_t
-afr_is_opendir_done (xlator_t *this, inode_t *inode)
+        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)
 {
-        int ret = 0;
+        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;
+}
 
-        uint64_t ctx          = 0;
-        uint64_t opendir_done = 0;
+void
+afr_inode_set_ctx_params (xlator_t *this, inode_t *inode,
+                          afr_inode_params_t *params)
+{
+        GF_ASSERT (inode);
+        GF_ASSERT (params);
 
-        VALIDATE_OR_GOTO (inode, out);
+        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);
         {
-                ret = __inode_ctx_get (inode, this, &ctx);
-
-                if (ret < 0)
+                ctx = __afr_inode_ctx_get (inode, this);
+                if (!ctx)
                         goto unlock;
-
-                opendir_done = ctx & AFR_ICTX_OPENDIR_DONE_MASK;
+                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);
-
-out:
-        return opendir_done;
 }
 
+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)
 {
-        uint64_t ctx = 0;
-        int      ret = 0;
+        afr_inode_params_t params = {0};
 
-        VALIDATE_OR_GOTO (inode, out);
+        params.op = AFR_INODE_SET_OPENDIR_DONE;
+        afr_inode_set_ctx_params (this, inode, &params);
+}
 
-        LOCK (&inode->lock);
-        {
-                ret = __inode_ctx_get (inode, this, &ctx);
+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;
 
-                if (ret < 0) {
-                        ctx = 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;
+}
 
-                ctx = (~AFR_ICTX_OPENDIR_DONE_MASK & ctx)
-                        | (0xFFFFFFFFFFFFFFFFULL & AFR_ICTX_OPENDIR_DONE_MASK);
+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;
 
-                __inode_ctx_put (inode, this, ctx);
+        if (child < 0) {
+                return _gf_false;
         }
-        UNLOCK (&inode->lock);
+
+        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;
+        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;
+}
 
-uint64_t
-afr_read_child (xlator_t *this, inode_t *inode)
+/* 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)
 {
-        int ret = 0;
+        int32_t                  read_child   = -1;
+        int                      i            = 0;
 
-        uint64_t ctx         = 0;
-        uint64_t read_child  = 0;
+        GF_ASSERT (success_children);
 
-        VALIDATE_OR_GOTO (inode, out);
+        read_child = config_read_child;
+        if (afr_is_read_child (success_children, sources, child_count,
+                               read_child))
+                goto out;
 
-        LOCK (&inode->lock);
-        {
-                ret = __inode_ctx_get (inode, this, &ctx);
+        read_child = prev_read_child;
+        if (afr_is_read_child (success_children, sources, child_count,
+                               read_child))
+                goto out;
 
-                if (ret < 0)
-                        goto unlock;
+        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;
+        }
 
-                read_child = ctx & AFR_ICTX_READ_CHILD_MASK;
+        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;
         }
-unlock:
-        UNLOCK (&inode->lock);
+        read_child = -1;
 
 out:
         return read_child;
 }
 
-
+/* This function should be used when all the success_children are sources
+ */
 void
-afr_set_read_child (xlator_t *this, inode_t *inode, int32_t read_child)
+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)
 {
-        uint64_t ctx = 0;
-        int      ret = 0;
+        int                      read_child = -1;
+        afr_private_t            *priv = NULL;
 
-        VALIDATE_OR_GOTO (inode, out);
+        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);
+}
 
-        LOCK (&inode->lock);
-        {
-                ret = __inode_ctx_get (inode, this, &ctx);
+/* 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;
 
-                if (ret < 0) {
-                        ctx = 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;
+                        }
                 }
 
-                ctx = (~AFR_ICTX_READ_CHILD_MASK & ctx)
-                        | (AFR_ICTX_READ_CHILD_MASK & read_child);
+                if (*call_child == -1) {
+                        ret = -ENOTCONN;
+                        goto out;
+                }
 
-                __inode_ctx_put (inode, this, ctx);
+                *last_index = i;
         }
-        UNLOCK (&inode->lock);
+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;
 }
 
-
-/**
- * afr_local_cleanup - cleanup everything in frame->local
- */
+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;
-        int              i = 0;
-
 
         sh = &local->self_heal;
         priv = this->private;
 
-        if (sh->buf)
-                GF_FREE (sh->buf);
+        if (sh->data_sh_info && strcmp (sh->data_sh_info, ""))
+                GF_FREE (sh->data_sh_info);
 
-        if (sh->xattr) {
-                for (i = 0; i < priv->child_count; i++) {
-                        if (sh->xattr[i]) {
-                                dict_unref (sh->xattr[i]);
-                                sh->xattr[i] = NULL;
-                        }
-                }
-                GF_FREE (sh->xattr);
-        }
+        if (sh->metadata_sh_info && strcmp (sh->metadata_sh_info, ""))
+                GF_FREE (sh->metadata_sh_info);
 
-        if (sh->child_errno)
-                GF_FREE (sh->child_errno);
+        GF_FREE (sh->buf);
 
-        if (sh->pending_matrix) {
-                for (i = 0; i < priv->child_count; i++) {
-                        GF_FREE (sh->pending_matrix[i]);
-                }
-                GF_FREE (sh->pending_matrix);
-        }
+        GF_FREE (sh->parentbufs);
 
-        if (sh->delta_matrix) {
-                for (i = 0; i < priv->child_count; i++) {
-                        GF_FREE (sh->delta_matrix[i]);
-                }
-                GF_FREE (sh->delta_matrix);
-        }
+        if (sh->inode)
+                inode_unref (sh->inode);
 
-        if (sh->sources)
-                GF_FREE (sh->sources);
+        afr_xattr_array_destroy (sh->xattr, priv->child_count);
 
-        if (sh->success)
-                GF_FREE (sh->success);
+        GF_FREE (sh->child_errno);
 
-        if (sh->locked_nodes)
-                GF_FREE (sh->locked_nodes);
+        afr_matrix_cleanup (sh->pending_matrix, priv->child_count);
+        afr_matrix_cleanup (sh->delta_matrix, priv->child_count);
 
-        if (sh->healing_fd && !sh->healing_fd_opened) {
+        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;
         }
 
-        if (sh->linkname)
-                GF_FREE ((char *)sh->linkname);
+        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)
 {
-        int             i = 0;
-        afr_private_t * priv = NULL;
+        afr_private_t *priv    = NULL;
+        int           i        = 0;
 
         priv = this->private;
 
-        for (i = 0; i < priv->child_count; i++) {
-                if (local->pending && local->pending[i])
-                        GF_FREE (local->pending[i]);
-        }
-
-        GF_FREE (local->pending);
-
-        if (local->internal_lock.locked_nodes)
-                GF_FREE (local->internal_lock.locked_nodes);
+        afr_matrix_cleanup (local->pending, priv->child_count);
+        afr_matrix_cleanup (local->transaction.txn_changelog,
+                            priv->child_count);
 
-        if (local->internal_lock.inode_locked_nodes)
-                GF_FREE (local->internal_lock.inode_locked_nodes);
+        GF_FREE (local->internal_lock.locked_nodes);
 
-        if (local->internal_lock.entry_locked_nodes)
-                GF_FREE (local->internal_lock.entry_locked_nodes);
+        for (i = 0; local->internal_lock.inodelk[i].domain; i++) {
+                GF_FREE (local->internal_lock.inodelk[i].locked_nodes);
+        }
 
-        if (local->internal_lock.lower_locked_nodes)
-                GF_FREE (local->internal_lock.lower_locked_nodes);
+        GF_FREE (local->internal_lock.lower_locked_nodes);
 
+        afr_entry_lockee_cleanup (&local->internal_lock);
 
-        GF_FREE (local->transaction.child_errno);
-        GF_FREE (local->child_errno);
+        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)
 {
-        int i;
         afr_private_t * priv = NULL;
 
         if (!local)
@@ -369,16 +887,21 @@ afr_local_cleanup (afr_local_t *local, xlator_t *this)
         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) {
-                        for (i = 0; i < priv->child_count; i++) {
-                                if (local->cont.lookup.xattrs[i]) {
-                                        dict_unref (local->cont.lookup.xattrs[i]);
-                                        local->cont.lookup.xattrs[i] = NULL;
-                                }
-                        }
+                        afr_reset_xattr (local->cont.lookup.xattrs,
+                                         priv->child_count);
                         GF_FREE (local->cont.lookup.xattrs);
                         local->cont.lookup.xattrs = NULL;
                 }
@@ -390,16 +913,24 @@ afr_local_cleanup (afr_local_t *local, xlator_t *this)
                 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 */
-                if (local->cont.getxattr.name)
-                        GF_FREE (local->cont.getxattr.name);
+                GF_FREE (local->cont.getxattr.name);
         }
 
         { /* lk */
-                if (local->cont.lk.locked_nodes)
-                        GF_FREE (local->cont.lk.locked_nodes);
+                GF_FREE (local->cont.lk.locked_nodes);
         }
 
         { /* create */
@@ -433,18 +964,40 @@ afr_local_cleanup (afr_local_t *local, xlator_t *this)
                         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 */
-                if (local->cont.opendir.checksum)
-                        GF_FREE (local->cont.opendir.checksum);
+                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);
 }
 
 
@@ -465,107 +1018,227 @@ afr_frame_return (call_frame_t *frame)
         return call_count;
 }
 
-
-/**
- * up_children_count - return the number of children that are up
- */
-
 int
-afr_up_children_count (int child_count, unsigned char *child_up)
+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 (child_up[i])
+                if (elems[i])
                         ret++;
         return ret;
 }
 
+/**
+ * up_children_count - return the number of children that are up
+ */
 
-ino64_t
-afr_itransform (ino64_t ino, int child_count, int child_index)
+unsigned int
+afr_up_children_count (unsigned char *child_up, unsigned int child_count)
 {
-        ino64_t scaled_ino = -1;
-
-        if (ino == ((uint64_t) -1)) {
-                scaled_ino = ((uint64_t) -1);
-                goto out;
-        }
-
-        scaled_ino = (ino * child_count) + child_index;
+        return afr_set_elem_count_get (child_up, child_count);
+}
 
-out:
-        return scaled_ino;
+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);
+}
 
-int
-afr_deitransform_orig (ino64_t ino, int child_count)
+gf_boolean_t
+afr_is_fresh_lookup (loc_t *loc, xlator_t *this)
 {
-        int index = -1;
+        uint64_t          ctx = 0;
+        int32_t           ret = 0;
 
-        index = ino % child_count;
+        GF_ASSERT (loc);
+        GF_ASSERT (this);
+        GF_ASSERT (loc->inode);
 
-        return index;
+        ret = inode_ctx_get (loc->inode, this, &ctx);
+        if (0 == ret)
+                return _gf_false;
+        return _gf_true;
 }
 
-
-int
-afr_deitransform (ino64_t ino, int child_count)
+void
+afr_update_loc_gfids (loc_t *loc, struct iatt *buf, struct iatt *postparent)
 {
-        return 0;
+        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'.
+ * */
 
-int
-afr_self_heal_lookup_unwind (call_frame_t *frame, xlator_t *this)
+static void
+afr_handle_quota_size (afr_local_t *local, xlator_t *this,
+                       dict_t *rsp_dict)
 {
-        afr_local_t *local = NULL;
+        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;
+        }
 
-        local = frame->local;
+        for (i = 0; i < priv->child_count; i++) {
+                if (sources[i]) {
+                        source_present = _gf_true;
+                        break;
+                }
+        }
 
-        if (local->govinda_gOvinda) {
-                afr_set_split_brain (this, local->cont.lookup.inode, _gf_true);
+        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);
+                }
         }
 
-        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);
+        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);
+                }
 
-        return 0;
+                data_unref (max_data);
+        }
 }
 
-
-static void
-afr_lookup_collect_xattr (afr_local_t *local, xlator_t *this,
-                          int child_index, dict_t *xattr)
+int
+afr_lookup_build_response_params (afr_local_t *local, xlator_t *this)
 {
-        uint32_t        inodelk_count = 0;
-        uint32_t        entrylk_count = 0;
+        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;
 
-        int ret = 0;
+        GF_ASSERT (local);
 
-        if (afr_sh_has_metadata_pending (xattr, child_index, this)) {
-                local->self_heal.need_metadata_self_heal = _gf_true;
-                gf_log(this->name, GF_LOG_DEBUG,
-                       "metadata self-heal is pending for %s.",
-                       local->loc.path);
-        }
+        buf = &local->cont.lookup.buf;
+        postparent = &local->cont.lookup.postparent;
+        xattr = &local->cont.lookup.xattr;
+        priv = this->private;
 
-        if (afr_sh_has_entry_pending (xattr, child_index, this)) {
-                local->self_heal.need_entry_self_heal = _gf_true;
-                gf_log(this->name, GF_LOG_DEBUG,
-                       "entry self-heal is pending for %s.", local->loc.path);
+        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;
         }
 
-        if (afr_sh_has_data_pending (xattr, child_index, this)) {
-                local->self_heal.need_data_self_heal = _gf_true;
-                gf_log(this->name, GF_LOG_DEBUG,
-                       "data self-heal is pending for %s.", local->loc.path);
+        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);
@@ -576,504 +1249,1248 @@ afr_lookup_collect_xattr (afr_local_t *local, xlator_t *this,
                                &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_self_heal_check (xlator_t *this, afr_local_t *local,
-                            struct iatt *buf, struct iatt *lookup_buf)
+afr_lookup_set_self_heal_params_by_xattr (afr_local_t *local, xlator_t *this,
+                                          dict_t *xattr)
 {
-        if (FILETYPE_DIFFERS (buf, lookup_buf)) {
-                /* mismatching filetypes with same name
-                */
+        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);
 
-                gf_log (this->name, GF_LOG_NORMAL,
-                        "filetype differs for %s ", local->loc.path);
+        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;
+                }
 
-                local->govinda_gOvinda = 1;
+                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_NORMAL,
+                gf_log (this->name, GF_LOG_DEBUG,
                         "permissions differ for %s ", local->loc.path);
-                local->self_heal.need_metadata_self_heal = _gf_true;
+                local->self_heal.do_metadata_self_heal = _gf_true;
         }
 
         if (OWNERSHIP_DIFFERS (buf, lookup_buf)) {
                 /* mismatching permissions */
-                local->self_heal.need_metadata_self_heal = _gf_true;
-                gf_log (this->name, GF_LOG_NORMAL,
+                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_NORMAL,
+                gf_log (this->name, GF_LOG_DEBUG,
                         "size differs for %s ", local->loc.path);
-                local->self_heal.need_data_self_heal = _gf_true;
+                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_lookup_done (call_frame_t *frame, xlator_t *this, struct iatt *lookup_buf)
+afr_detect_self_heal_by_split_brain_status (afr_local_t *local, xlator_t *this)
 {
-        int  unwind = 1;
-        int  source = -1;
-        int  up_count = 0;
-        char sh_type_str[256] = {0,};
+        gf_boolean_t split_brain = _gf_false;
+        afr_self_heal_t *sh = NULL;
 
-        afr_private_t *priv  = NULL;
-        afr_local_t   *local = NULL;
+        sh = &local->self_heal;
 
-        priv  = this->private;
-        local = frame->local;
+        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);
+        }
+}
 
-        local->cont.lookup.postparent.ia_ino  = local->cont.lookup.parent_ino;
+static void
+afr_detect_self_heal_by_lookup_status (afr_local_t *local, xlator_t *this)
+{
+        GF_ASSERT (local);
+        GF_ASSERT (this);
 
-        if (local->cont.lookup.ino) {
-                local->cont.lookup.buf.ia_ino = local->cont.lookup.ino;
+        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);
         }
 
-        if (local->op_ret == 0) {
-                /* KLUDGE: assuming DHT will not itransform in
-                   revalidate */
-                if (local->cont.lookup.inode->ino) {
-                        local->cont.lookup.buf.ia_ino =
-                                local->cont.lookup.inode->ino;
-                }
-        }
-        up_count = afr_up_children_count (priv->child_count, priv->child_up);
-        if (up_count == 1) {
-                gf_log (this->name, GF_LOG_DEBUG,
-                        "Only 1 child up - do not attempt to detect self heal");
+        return;
+}
 
-                goto unwind;
-        }
+gf_boolean_t
+afr_can_self_heal_proceed (afr_self_heal_t *sh, afr_private_t *priv)
+{
+        GF_ASSERT (sh);
+        GF_ASSERT (priv);
 
-        if (local->success_count && local->enoent_count) {
-                local->self_heal.need_metadata_self_heal = _gf_true;
-                local->self_heal.need_data_self_heal     = _gf_true;
-                local->self_heal.need_entry_self_heal    = _gf_true;
-                gf_log(this->name, GF_LOG_NORMAL,
-                       "entries are missing in lookup of %s.",
-                       local->loc.path);
+        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;
+}
 
-        if (local->success_count) {
-                /* check for split-brain case in previous lookup */
-                if (afr_is_split_brain (this,
-                                        local->cont.lookup.inode)) {
-                        local->self_heal.need_data_self_heal = _gf_true;
-                        gf_log(this->name, GF_LOG_NORMAL,
-                               "split brain detected during lookup of "
-                               "%s.", local->loc.path);
-                }
+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;
         }
 
-        if ((local->self_heal.need_metadata_self_heal
-             || local->self_heal.need_data_self_heal
-             || local->self_heal.need_entry_self_heal)
-            && ((!local->cont.lookup.is_revalidate)
-                || (local->op_ret != -1))) {
+        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);
 
-                if (local->inodelk_count || local->entrylk_count) {
+        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);
+}
 
-                        /* Someone else is doing self-heal on this file.
-                           So just make a best effort to set the read-subvolume
-                           and return */
+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;
 
-                        if (IA_ISREG (local->cont.lookup.inode->ia_type)) {
-                                source = afr_self_heal_get_source (this, local, local->cont.lookup.xattrs);
+        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++;
+                }
+        }
 
-                                if (source >= 0) {
-                                        afr_set_read_child (this,
-                                                            local->cont.lookup.inode,
-                                                            source);
-                                }
-                        }
-                } else {
-                        if (!local->cont.lookup.inode->ia_type) {
-                                /* fix for RT #602 */
-                                local->cont.lookup.inode->ia_type =
-                                        lookup_buf->ia_type;
-                        }
+        return gfid_miss_count;
+}
 
-                        local->self_heal.background = _gf_true;
-                        local->self_heal.type       = local->cont.lookup.buf.ia_type;
-                        local->self_heal.unwind     = afr_self_heal_lookup_unwind;
+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;
 
-                        unwind = 0;
+        priv = this->private;
+        bufs = local->cont.lookup.bufs;
+        success_children = local->cont.lookup.success_children;
 
-                        afr_self_heal_type_str_get(&local->self_heal,
-                                                   sh_type_str,
-                                                   sizeof(sh_type_str));
+        miss_count =  afr_gfid_missing_count (this->name, success_children,
+                                              bufs, priv->child_count,
+                                              local->loc.path);
+        return miss_count;
+}
 
-                        gf_log (this->name, GF_LOG_NORMAL, "background %s "
-                                "self-heal triggered. path: %s",
-                                sh_type_str, local->loc.path);
+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;
 
-                        afr_self_heal (frame, this);
+        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;
 
-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);
+        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;
 
-/*
- * 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 > ENOENT > others
- *
- */
+        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;
+}
 
-static gf_boolean_t
-__error_more_important (int32_t old_errno, int32_t new_errno)
+gf_boolean_t
+afr_open_only_data_self_heal (char *data_self_heal)
 {
-        gf_boolean_t ret = _gf_true;
+        return !strcmp (data_self_heal, "open");
+}
 
-        /* Nothing should ever overwrite ESTALE */
-        if (old_errno == ESTALE)
-                ret = _gf_false;
+gf_boolean_t
+afr_data_self_heal_enabled (char *data_self_heal)
+{
+        gf_boolean_t    enabled = _gf_false;
 
-        /* Nothing should overwrite ENOENT, except ESTALE */
-        else if ((old_errno == ENOENT) && (new_errno != ESTALE))
-                ret = _gf_false;
+        if (gf_string2boolean (data_self_heal, &enabled) == -1) {
+                enabled = !strcmp (data_self_heal, "open");
+                GF_ASSERT (enabled);
+        }
 
-        return ret;
+        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_fresh_lookup_cbk (call_frame_t *frame, void *cookie,
-                      xlator_t *this,  int32_t op_ret,  int32_t op_errno,
-                      inode_t *inode,   struct iatt *buf, dict_t *xattr,
-                      struct iatt *postparent)
+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;
-        afr_private_t * priv  = NULL;
-        struct iatt *   lookup_buf = NULL;
+        afr_local_t *local = NULL;
+        int         ret    = -1;
+        dict_t      *xattr = NULL;
 
-        int             call_count      = -1;
-        int             child_index     = -1;
-        int             first_up_child  = -1;
+        local = frame->local;
 
-        child_index = (long) cookie;
+        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;
 
-        LOCK (&frame->lock);
-        {
-                local = frame->local;
+        memcpy (lookup_bufs, sh->buf, priv->child_count * sizeof (*sh->buf));
+        memcpy (lookup_parentbufs, sh->parentbufs,
+                priv->child_count * sizeof (*sh->parentbufs));
 
-                lookup_buf = &local->cont.lookup.buf;
+        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;
+        }
 
-                if (op_ret == -1) {
-                        if (op_errno == ENOENT)
-                                local->enoent_count++;
+        for (i = 0; i < priv->child_count; i++) {
+                if (sh->xattr[i])
+                        local->cont.lookup.xattrs[i] = dict_ref (sh->xattr[i]);
+        }
 
-                        if (__error_more_important (local->op_errno, op_errno))
-                                local->op_errno = op_errno;
+        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);
+}
 
-                        if (local->op_errno == ESTALE) {
-                                local->op_ret = -1;
-                        }
+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;
 
-                        goto unlock;
-                }
+        GF_ASSERT (sh_launched);
+        *sh_launched = _gf_false;
+        priv         = this->private;
+        local        = frame->local;
 
-                afr_lookup_collect_xattr (local, this, child_index, xattr);
+        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;
+        }
 
-                first_up_child = afr_first_up_child (priv);
+        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;
+}
 
-                if (child_index == first_up_child) {
-                        local->cont.lookup.ino =
-                                afr_itransform (buf->ia_ino,
-                                                priv->child_count,
-                                                first_up_child);
+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++;
                 }
+        }
+}
 
-                if (local->success_count == 0) {
-                        if (local->op_errno != ESTALE)
-                                local->op_ret = op_ret;
+static int
+afr_lookup_set_read_ctx (afr_local_t *local, xlator_t *this, int32_t read_child)
+{
+        afr_private_t           *priv = NULL;
 
-                        local->cont.lookup.inode               = inode_ref (inode);
-                        local->cont.lookup.xattr               = dict_ref (xattr);
-                        local->cont.lookup.xattrs[child_index] = dict_ref (xattr);
-                        local->cont.lookup.postparent          = *postparent;
+        GF_ASSERT (read_child >= 0);
 
-                        if (priv->first_lookup && inode->ino == 1) {
-                                gf_log (this->name, GF_LOG_NORMAL,
-                                        "added root inode");
-                                priv->root_inode = inode_ref (inode);
-                                priv->first_lookup = 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);
 
-                        *lookup_buf = *buf;
+        return 0;
+}
 
-                        lookup_buf->ia_ino = afr_itransform (buf->ia_ino,
-                                                             priv->child_count,
-                                                             child_index);
-                        if (priv->read_child >= 0) {
-                                afr_set_read_child (this,
-                                                    local->cont.lookup.inode,
-                                                    priv->read_child);
-                        } else {
-                                afr_set_read_child (this,
-                                                    local->cont.lookup.inode,
-                                                    child_index);
-                        }
+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;
 
-                } else {
-                        afr_lookup_self_heal_check (this, local, buf, lookup_buf);
-
-                        if (child_index == local->read_child_index) {
-                                /*
-                                   lookup has succeeded on the read child.
-                                   So use its inode number
-                                */
-                                if (local->cont.lookup.xattr)
-                                        dict_unref (local->cont.lookup.xattr);
-
-                                local->cont.lookup.xattr = dict_ref (xattr);
-                                local->cont.lookup.xattrs[child_index] = dict_ref (xattr);
-                                local->cont.lookup.postparent          = *postparent;
-
-                                *lookup_buf = *buf;
-
-                                if (priv->read_child >= 0) {
-                                        afr_set_read_child (this,
-                                                            local->cont.lookup.inode,
-                                                            priv->read_child);
-                                } else {
-                                        afr_set_read_child (this,
-                                                            local->cont.lookup.inode,
-                                                            local->read_child_index);
-                                }
-                        }
+        local   = frame->local;
+        fresh_lookup = local->cont.lookup.fresh_lookup;
 
-                }
+        if (local->loc.parent == NULL)
+                fail_conflict = _gf_true;
 
-                local->success_count++;
+        if (afr_lookup_conflicting_entries (local, this)) {
+                if (fail_conflict == _gf_false)
+                        ret = 0;
+                goto out;
         }
-unlock:
-        UNLOCK (&frame->lock);
 
-        call_count = afr_frame_return (frame);
+        ret = afr_lookup_select_read_child (local, this, &read_child);
+        if (!afr_is_transaction_running (local) || fresh_lookup) {
+                if (read_child < 0)
+                        goto out;
 
-        if (call_count == 0) {
-                afr_lookup_done (frame, this, lookup_buf);
+                ret = afr_lookup_set_read_ctx (local, this, read_child);
+                if (ret)
+                        goto out;
         }
 
-        return 0;
-}
+        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_revalidate_lookup_cbk (call_frame_t *frame, void *cookie,
-                           xlator_t *this, int32_t op_ret, int32_t op_errno,
-                           inode_t *inode, struct iatt *buf, dict_t *xattr,
-                           struct iatt *postparent)
+afr_lookup_get_latest_subvol (afr_local_t *local, xlator_t *this)
 {
-        afr_local_t *   local = NULL;
-        afr_private_t * priv  = NULL;
-        struct iatt *   lookup_buf = NULL;
+        afr_private_t *priv = NULL;
+        int32_t       *success_children = NULL;
+        struct iatt   *bufs = NULL;
+        int           i = 0;
+        int           child = 0;
+        int           lsubvol = -1;
 
-        int             call_count      = -1;
-        int             child_index     = -1;
-        int             first_up_child  = -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;
 
-        child_index = (long) cookie;
         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);
+}
 
-        LOCK (&frame->lock);
-        {
-                local = frame->local;
+void
+afr_succeed_lookup_on_latest_iatt (afr_local_t *local, xlator_t *this)
+{
+        int    lsubvol = 0;
 
-                lookup_buf = &local->cont.lookup.buf;
+        if (!afr_lookup_conflicting_entries (local, this))
+                goto out;
 
-                if (op_ret == -1) {
-                        if (op_errno == ENOENT)
-                                local->enoent_count++;
+        lsubvol = afr_lookup_get_latest_subvol (local, this);
+        if (lsubvol < 0)
+                goto out;
+        afr_lookup_mark_other_entries_stale (local, this, lsubvol);
+out:
+        return;
+}
 
-                        if (__error_more_important (local->op_errno, op_errno))
-                            local->op_errno = op_errno;
+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;
+}
 
-                            if (local->op_errno == ESTALE) {
-                                    local->op_ret = -1;
-                            }
 
-                            goto unlock;
-                }
+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;
 
-                afr_lookup_collect_xattr (local, this, child_index, xattr);
+        priv  = this->private;
+        local = frame->local;
 
-                first_up_child = afr_first_up_child (priv);
+	if (afr_is_entry_possibly_under_creation (local, this)) {
+		local->op_ret = -1;
+		local->op_errno = ENOENT;
+		goto unwind;
+	}
 
-                if (child_index == first_up_child) {
-                        local->cont.lookup.ino =
-                                afr_itransform (buf->ia_ino,
-                                                priv->child_count,
-                                                first_up_child);
-                }
+        if (local->op_ret < 0)
+                goto unwind;
 
-                /* in case of revalidate, we need to send stat of the
-                 * child whose stat was sent during the first lookup.
-                 * (so that time stamp does not vary with revalidate.
-                 * in case it is down, stat of the fist success will
-                 * be replied */
+        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;
+        }
 
-                /* inode number should be preserved across revalidates */
+        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 (local->success_count == 0) {
-                        if (local->op_errno != ESTALE)
-                                local->op_ret = op_ret;
+        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);
 
-                        local->cont.lookup.inode               = inode_ref (inode);
-                        local->cont.lookup.xattr               = dict_ref (xattr);
-                        local->cont.lookup.xattrs[child_index] = dict_ref (xattr);
-                        local->cont.lookup.postparent          = *postparent;
+        afr_lookup_perform_self_heal (frame, this, &sh_launched);
+        if (sh_launched) {
+                unwind = 0;
+                goto unwind;
+        }
 
-                        *lookup_buf = *buf;
+ 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);
+        }
+}
 
-                        lookup_buf->ia_ino = afr_itransform (buf->ia_ino,
-                                                             priv->child_count,
-                                                             child_index);
+/*
+ * 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;
+}
 
-                        if (priv->read_child >= 0) {
-                                afr_set_read_child (this,
-                                                    local->cont.lookup.inode,
-                                                    priv->read_child);
-                        } else {
-                                afr_set_read_child (this,
-                                                    local->cont.lookup.inode,
-                                                    child_index);
-                        }
+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;
 
+        for (i = 0; i < child_count; i++) {
+                if (children) {
+                        child = children[i];
+                        if (child == -1)
+                                break;
                 } else {
-                        afr_lookup_self_heal_check (this, local, buf, lookup_buf);
+                        child = i;
+                }
+		op_errno = afr_most_important_error(op_errno,
+						    child_errno[child],
+						    _gf_false);
+        }
+        return op_errno;
+}
 
-                        if (child_index == local->read_child_index) {
+static void
+afr_lookup_handle_error (afr_local_t *local, int32_t op_ret,  int32_t op_errno)
+{
+        GF_ASSERT (local);
+        if (op_errno == ENOENT)
+                local->enoent_count++;
 
-                                /*
-                                   lookup has succeeded on the read child.
-                                   So use its inode number
-                                */
+	local->op_errno = afr_most_important_error(local->op_errno, op_errno,
+						   _gf_false);
 
-                                if (local->cont.lookup.xattr)
-                                        dict_unref (local->cont.lookup.xattr);
+        if (local->op_errno == ESTALE) {
+                local->op_ret = -1;
+        }
+}
 
-                                local->cont.lookup.xattr               = dict_ref (xattr);
-                                local->cont.lookup.xattrs[child_index] = dict_ref (xattr);
-                                local->cont.lookup.postparent          = *postparent;
+static void
+afr_set_root_inode_on_first_lookup (afr_local_t *local, xlator_t *this,
+                                    inode_t *inode)
+{
+        afr_private_t           *priv = NULL;
+        GF_ASSERT (inode);
 
-                                *lookup_buf = *buf;
+        if (!__is_root_gfid (inode->gfid))
+                goto out;
+        if (!afr_is_fresh_lookup (&local->loc, this))
+                goto out;
+        priv = this->private;
+        if ((priv->first_lookup)) {
+                gf_log (this->name, GF_LOG_INFO, "added root inode");
+                priv->root_inode = inode_ref (inode);
+                priv->first_lookup = 0;
+        }
+out:
+        return;
+}
 
-                                if (priv->read_child >= 0) {
-                                        afr_set_read_child (this,
-                                                            local->cont.lookup.inode,
-                                                            priv->read_child);
-                                } else {
-                                        afr_set_read_child (this,
-                                                            local->cont.lookup.inode,
-                                                            local->read_child_index);
-                                }
-                        }
+static void
+afr_lookup_cache_args (afr_local_t *local, int child_index, dict_t *xattr,
+                       struct iatt *buf, struct iatt *postparent)
+{
+        GF_ASSERT (child_index >= 0);
+        local->cont.lookup.xattrs[child_index] = dict_ref (xattr);
+        local->cont.lookup.postparents[child_index] = *postparent;
+        local->cont.lookup.bufs[child_index] = *buf;
+}
+
+static void
+afr_lookup_handle_first_success (afr_local_t *local, xlator_t *this,
+                                 inode_t *inode, struct iatt *buf)
+{
+        local->cont.lookup.inode      = inode_ref (inode);
+        local->cont.lookup.buf        = *buf;
+        afr_set_root_inode_on_first_lookup (local, this, inode);
+}
+
+static int32_t
+afr_discovery_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                   int32_t op_ret, int32_t op_errno, dict_t *dict,
+                   dict_t *xdata)
+{
+        int              ret            = 0;
+        char            *pathinfo       = NULL;
+        gf_boolean_t     is_local        = _gf_false;
+        afr_private_t   *priv           = NULL;
+        int32_t          child_index    = -1;
+
+        if (op_ret != 0) {
+                goto out;
+        }
+
+        ret = dict_get_str (dict, GF_XATTR_PATHINFO_KEY, &pathinfo);
+        if (ret != 0) {
+                goto out;
+        }
+
+        ret = afr_local_pathinfo (pathinfo, &is_local);
+        if (ret) {
+                goto out;
+        }
+
+        priv = this->private;
+        /*
+         * Note that one local subvolume will override another here.  The only
+         * way to avoid that would be to reta
author	Amar Tumballi <amarts@redhat.com>	2012-03-19 18:10:34 +0530
committer	Vijay Bellur <vijay@gluster.com>	2012-03-19 09:14:40 -0700
commit	d98c3e19342be3b8b3e2c7e3d7e544de34143bdf (patch)
tree	ba1d1d18d33694c99749dd005d0d46e91020e7e1 /xlators/cluster/unify
parent	55c22fbfae259bff3b0e0cca192f709b74d8bee5 (diff)