From 5dadc2d8f1193cbffa1aa1f83776725747ec177b Mon Sep 17 00:00:00 2001

From: Raghavendra Bhat <raghavendra@redhat.com>

Date: Thu, 12 May 2016 16:40:04 -0400

Subject: [PATCH 386/393] features/bit-rot-stub: bring in optional versioning

  As of now bit-rot-stub does versioning always. This leads

  lots of getxattr calls being made in lookups. So make

  object versioning optional.

> BUG: 1359599

> Signed-off-by: Raghavendra Bhat <raghavendra@redhat.com>

> Signed-off-by: Kotresh HR <khiremat@redhat.com>

> Reviewed-on: https://review.gluster.org/14442

> Smoke: Gluster Build System <jenkins@build.gluster.org>

> NetBSD-regression: NetBSD Build System <jenkins@build.gluster.org>

> CentOS-regression: Gluster Build System <jenkins@build.gluster.org>

> Reviewed-by: Vijay Bellur <vbellur@redhat.com>

Change-Id: I83713e45ae59fb28004bb3cfa008f2d69edebbfa

BUG: 1298258

Signed-off-by: Kotresh HR <khiremat@redhat.com>

Reviewed-on: https://code.engineering.redhat.com/gerrit/103743

Reviewed-by: Atin Mukherjee <amukherj@redhat.com>

---

 tests/bitrot/br-stub.t                             |   3 +

 tests/bitrot/bug-1221914.t                         |   3 +

 tests/bitrot/bug-1244613.t                         |   3 +

 .../bit-rot/src/stub/bit-rot-stub-helpers.c        |   3 +

 xlators/features/bit-rot/src/stub/bit-rot-stub.c   | 318 +++++++++++++++++----

 xlators/features/bit-rot/src/stub/bit-rot-stub.h   |  26 +-

 xlators/mgmt/glusterd/src/glusterd-volgen.c        |  17 +-

 7 files changed, 314 insertions(+), 59 deletions(-)

diff --git a/tests/bitrot/br-stub.t b/tests/bitrot/br-stub.t

index 8ce170d..5013398 100644

--- a/tests/bitrot/br-stub.t

+++ b/tests/bitrot/br-stub.t

@@ -27,6 +27,9 @@ TEST $CLI volume set $V0 performance.write-behind off

 TEST $CLI volume start $V0;

 EXPECT 'Started' volinfo_field $V0 'Status';

+## enable bitrot

+TEST $CLI volume bitrot $V0 enable;

+

 ## Wait for gluster nfs to come up

 EXPECT_WITHIN $NFS_EXPORT_TIMEOUT "1" is_nfs_export_available

diff --git a/tests/bitrot/bug-1221914.t b/tests/bitrot/bug-1221914.t

index 85715e5..7f6c10c 100644

--- a/tests/bitrot/bug-1221914.t

+++ b/tests/bitrot/bug-1221914.t

@@ -25,6 +25,9 @@ TEST $CLI volume set $V0 performance.write-behind off

 TEST $CLI volume start $V0;

 EXPECT 'Started' volinfo_field $V0 'Status';

+## Enable bitrot

+TEST $CLI volume bitrot $V0 enable;

+

 ## Mount the volume

 TEST $GFS --volfile-server=$H0 --volfile-id=$V0 $M0;

diff --git a/tests/bitrot/bug-1244613.t b/tests/bitrot/bug-1244613.t

index 7cb57f0..5674e6f 100644

--- a/tests/bitrot/bug-1244613.t

+++ b/tests/bitrot/bug-1244613.t

@@ -40,6 +40,9 @@ TEST $CLI volume set $V0 performance.nfs.write-behind off

 TEST $CLI volume start $V0;

 EXPECT 'Started' volinfo_field $V0 'Status';

+## Enable bitrot

+TEST $CLI volume bitrot $V0 enable;

+

 ## Wait for gluster nfs to come up

 EXPECT_WITHIN $NFS_EXPORT_TIMEOUT "1" is_nfs_export_available

diff --git a/xlators/features/bit-rot/src/stub/bit-rot-stub-helpers.c b/xlators/features/bit-rot/src/stub/bit-rot-stub-helpers.c

index 7bd270d..7a308f9 100644

--- a/xlators/features/bit-rot/src/stub/bit-rot-stub-helpers.c

+++ b/xlators/features/bit-rot/src/stub/bit-rot-stub-helpers.c

@@ -393,8 +393,11 @@ br_stub_lookup_wrapper (call_frame_t *frame, xlator_t *this,

         struct iatt        stbuf       = {0, };

         struct iatt        postparent  = {0,};

         dict_t            *xattr       = NULL;

+        gf_boolean_t       ver_enabled = _gf_false;

+        BR_STUB_VER_ENABLED_IN_CALLPATH(frame, ver_enabled);

         priv = this->private;

+        BR_STUB_VER_COND_GOTO (priv, (!ver_enabled), done);

         VALIDATE_OR_GOTO (loc, done);

         if (gf_uuid_compare (loc->gfid, priv->bad_object_dir_gfid))

diff --git a/xlators/features/bit-rot/src/stub/bit-rot-stub.c b/xlators/features/bit-rot/src/stub/bit-rot-stub.c

index 4e01f5c..115b10d 100644

--- a/xlators/features/bit-rot/src/stub/bit-rot-stub.c

+++ b/xlators/features/bit-rot/src/stub/bit-rot-stub.c

@@ -136,7 +136,7 @@ init (xlator_t *this)

         if (!priv->local_pool)

                 goto free_priv;

-        GF_OPTION_INIT ("bitrot", priv->go, bool, free_mempool);

+        GF_OPTION_INIT ("bitrot", priv->do_versioning, bool, free_mempool);

         GF_OPTION_INIT ("export", tmp, str, free_mempool);

         memcpy (priv->export, tmp, strlen (tmp) + 1);

@@ -186,6 +186,34 @@ init (xlator_t *this)

         return -1;

 }

+/* TODO:

+ * As of now enabling bitrot option does 2 things.

+ * 1) Start the Bitrot Daemon which signs the objects (currently files only)

+ *    upon getting notified by the stub.

+ * 2) Enable versioning of the objects. Object versions (again files only) are

+ *    incremented upon modification.

+ * So object versioning is tied to bitrot daemon's signing. In future, object

+ * versioning might be necessary for other things as well apart from bit-rot

+ * detection (well thats the objective of bringing in object-versioning :)).

+ * In that case, better to make versioning a new option and letting it to be

+ * enabled despite bit-rot detection is not needed.

+ * Ex: ICAP.

+ */

+int32_t

+reconfigure (xlator_t *this, dict_t *options)

+{

+        int32_t            ret  = -1;

+        br_stub_private_t *priv = NULL;

+

+        priv = this->private;

+

+        GF_OPTION_RECONF ("bitrot", priv->do_versioning, options, bool, out);

+

+        ret = 0;

+out:

+        return ret;

+}

+

 void

 fini (xlator_t *this)

 {

@@ -289,6 +317,9 @@ br_stub_alloc_local (xlator_t *this)

 static void

 br_stub_dealloc_local (br_stub_local_t *ptr)

 {

+        if (!ptr)

+                return;

+

         mem_put (ptr);

 }

@@ -408,6 +439,9 @@ br_stub_fill_local (br_stub_local_t *local,

 static void

 br_stub_cleanup_local (br_stub_local_t *local)

 {

+        if (!local)

+                return;

+

         local->fopstub = NULL;

         local->versioningtype = 0;

         local->u.context.version = 0;

@@ -430,17 +464,33 @@ br_stub_need_versioning (xlator_t *this,

         int32_t ret      = -1;

         uint64_t ctx_addr = 0;

         br_stub_inode_ctx_t *c = NULL;

+        unsigned long        version  = BITROT_DEFAULT_CURRENT_VERSION;

         *versioning = _gf_false;

         *modified = _gf_false;

+        /* Bitrot stub inode context was initialized only in lookup, create

+         * and mknod cbk path. Object versioning was enabled by default

+         * irrespective of bitrot enbaled or not. But it's made optional now.

+         * As a consequence there could be cases where getting inode ctx would

+         * fail because it's not set yet.

+         * e.g., If versioning (with bitrot enable) is enabled while I/O is

+         * happening, it could directly get other fops like writev without

+         * lookup, where getting inode ctx would fail. Hence initialize the

+         * inode ctx on failure to get ctx. This is done in all places where

+         * applicable.

+         */

         ret = br_stub_get_inode_ctx (this, fd->inode, &ctx_addr);

         if (ret < 0) {

-                gf_msg (this->name, GF_LOG_ERROR, 0,

-                        BRS_MSG_GET_INODE_CONTEXT_FAILED, "failed to get the "

-                        "inode context for the inode %s",

-                        uuid_utoa (fd->inode->gfid));

-                goto error_return;

+                ret = br_stub_init_inode_versions (this, fd, fd->inode, version,

+                                                   _gf_true, _gf_false);

+                if (ret) {

+                        gf_msg (this->name, GF_LOG_ERROR, 0,

+                                BRS_MSG_GET_INODE_CONTEXT_FAILED, "failed to "

+                                " init the inode context for the inode %s",

+                                uuid_utoa (fd->inode->gfid));

+                        goto error_return;

+                }

         }

         c = (br_stub_inode_ctx_t *) (long) ctx_addr;

@@ -524,12 +574,17 @@ br_stub_mark_inode_modified (xlator_t *this, br_stub_local_t *local)

         int32_t              ret      = 0;

         uint64_t             ctx_addr = 0;

         br_stub_inode_ctx_t *ctx      = NULL;

+        unsigned long        version  = BITROT_DEFAULT_CURRENT_VERSION;

         fd = local->u.context.fd;

         ret = br_stub_get_inode_ctx (this, fd->inode, &ctx_addr);

-        if (ret < 0)

-                goto error_return;

+        if (ret < 0) {

+                ret = br_stub_init_inode_versions (this, fd, fd->inode, version,

+                                                   _gf_true, _gf_false);

+                if (ret)

+                        goto error_return;

+        }

         ctx = (br_stub_inode_ctx_t *) (long) ctx_addr;

@@ -558,6 +613,7 @@ br_stub_check_bad_object (xlator_t *this, inode_t *inode, int32_t *op_ret,

                            int32_t *op_errno)

 {

         int ret = -1;

+        unsigned long        version  = BITROT_DEFAULT_CURRENT_VERSION;

         ret = br_stub_is_bad_object (this, inode);

         if (ret == -2) {

@@ -569,11 +625,16 @@ br_stub_check_bad_object (xlator_t *this, inode_t *inode, int32_t *op_ret,

         }

         if (ret == -1) {

-                gf_msg (this->name, GF_LOG_ERROR, 0,

-                        BRS_MSG_GET_INODE_CONTEXT_FAILED, "could not get inode"

-                        " context for %s", uuid_utoa (inode->gfid));

-                *op_ret = -1;

-                *op_errno = EINVAL;

+                ret = br_stub_init_inode_versions (this, NULL, inode, version,

+                                                   _gf_true, _gf_false);

+                if (ret) {

+                        gf_msg (this->name, GF_LOG_ERROR, 0,

+                                BRS_MSG_GET_INODE_CONTEXT_FAILED,

+                                "failed to init inode context for %s",

+                                uuid_utoa (inode->gfid));

+                        *op_ret = -1;

+                        *op_errno = EINVAL;

+                }

         }

         return ret;

@@ -787,6 +848,20 @@ br_stub_signth (void *arg)

         return NULL;

 }

+static gf_boolean_t

+br_stub_internal_xattr (dict_t *dict) {

+

+        if (dict_get (dict, GLUSTERFS_SET_OBJECT_SIGNATURE) ||

+            dict_get (dict, GLUSTERFS_GET_OBJECT_SIGNATURE) ||

+            dict_get (dict, BR_REOPEN_SIGN_HINT_KEY) ||

+            dict_get (dict, BITROT_OBJECT_BAD_KEY) ||

+            dict_get (dict, BITROT_SIGNING_VERSION_KEY) ||

+            dict_get (dict, BITROT_CURRENT_VERSION_KEY))

+                return _gf_true;

+

+        return _gf_false;

+}

+

 int

 orderq (struct list_head *elem1, struct list_head *elem2)

 {

@@ -1200,6 +1275,26 @@ br_stub_fsetxattr (call_frame_t *frame, xlator_t *this,

         int32_t              ret      = 0;

         uint32_t             val      = 0;

         br_isignature_t     *sign     = NULL;

+        br_stub_private_t   *priv     = NULL;

+        int32_t              op_ret   = -1;

+        int32_t              op_errno = EINVAL;

+        char                *format   = "(%s:%s)";

+        char                 dump[64*1024]  = {0,};

+

+        priv = this->private;

+

+        if ((frame->root->pid != GF_CLIENT_PID_BITD &&

+            frame->root->pid != GF_CLIENT_PID_SCRUB) &&

+            br_stub_internal_xattr (dict)) {

+                dict_dump_to_str (dict, dump, sizeof(dump), format);

+                gf_msg (this->name, GF_LOG_ERROR, 0,

+                        BRS_MSG_SET_INTERNAL_XATTR, "fsetxattr called on "

+                        "internal xattr %s", dump);

+                goto unwind;

+        }

+

+        if (!priv->do_versioning)

+                goto wind;

         if (!IA_ISREG (fd->inode->ia_type))

                 goto wind;

@@ -1251,6 +1346,11 @@ wind:

         STACK_WIND (frame, default_fsetxattr_cbk, FIRST_CHILD (this),

                     FIRST_CHILD (this)->fops->fsetxattr, fd, dict, flags,

                     xdata);

+        return 0;

+

+unwind:

+        STACK_UNWIND_STRICT (fsetxattr, frame, op_ret, op_errno, NULL);

+

 done:

         return 0;

 }

@@ -1272,12 +1372,7 @@ br_stub_setxattr (call_frame_t *frame, xlator_t *this,

         char     dump[64*1024]             = {0,};

         char    *format                    = "(%s:%s)";

-        if (dict_get (dict, GLUSTERFS_SET_OBJECT_SIGNATURE) ||

-            dict_get (dict, GLUSTERFS_GET_OBJECT_SIGNATURE) ||

-            dict_get (dict, BR_REOPEN_SIGN_HINT_KEY) ||

-            dict_get (dict, BITROT_OBJECT_BAD_KEY) ||

-            dict_get (dict, BITROT_SIGNING_VERSION_KEY) ||

-            dict_get (dict, BITROT_CURRENT_VERSION_KEY)) {

+        if (br_stub_internal_xattr (dict)) {

                 dict_dump_to_str (dict, dump, sizeof(dump), format);

                 gf_msg (this->name, GF_LOG_ERROR, 0,

                         BRS_MSG_SET_INTERNAL_XATTR, "setxattr called on "

@@ -1285,7 +1380,6 @@ br_stub_setxattr (call_frame_t *frame, xlator_t *this,

                 goto unwind;

         }

-

         STACK_WIND_TAIL (frame, FIRST_CHILD (this),

                          FIRST_CHILD (this)->fops->setxattr, loc, dict, flags,

                          xdata);

@@ -1318,7 +1412,6 @@ br_stub_removexattr (call_frame_t *frame, xlator_t *this,

                 goto unwind;

         }

-

         STACK_WIND_TAIL (frame, FIRST_CHILD(this),

                          FIRST_CHILD(this)->fops->removexattr,

                          loc, name, xdata);

@@ -1345,7 +1438,6 @@ br_stub_fremovexattr (call_frame_t *frame, xlator_t *this,

                 goto unwind;

         }

-

         STACK_WIND_TAIL (frame, FIRST_CHILD(this),

                          FIRST_CHILD(this)->fops->fremovexattr,

                          fd, name, xdata);

@@ -1593,9 +1685,12 @@ br_stub_getxattr (call_frame_t *frame, xlator_t *this,

         int32_t             op_ret   = -1;

         int32_t             op_errno = EINVAL;

         br_stub_local_t    *local    = NULL;

+        br_stub_private_t  *priv     = NULL;

+

         GF_VALIDATE_OR_GOTO ("bit-rot-stub", this, unwind);

         GF_VALIDATE_OR_GOTO (this->name, loc, unwind);

+        GF_VALIDATE_OR_GOTO (this->name, this->private, unwind);

         GF_VALIDATE_OR_GOTO (this->name, loc->inode, unwind);

         rootgfid[15] = 1;

@@ -1605,6 +1700,13 @@ br_stub_getxattr (call_frame_t *frame, xlator_t *this,

                 goto wind;

         }

+        if (br_stub_is_internal_xattr (name))

+                goto unwind;

+

+        priv = this->private;

+        if (!priv->do_versioning)

+                goto wind;

+

         /**

          * If xattr is node-uuid and the inode is marked bad, return EIO.

          * Returning EIO would result in AFR to choose correct node-uuid

@@ -1616,9 +1718,6 @@ br_stub_getxattr (call_frame_t *frame, xlator_t *this,

                 goto unwind;

         }

-        if (br_stub_is_internal_xattr (name))

-                goto unwind;

-

         /**

          * this special extended attribute is allowed only on root

          */

@@ -1671,6 +1770,7 @@ br_stub_fgetxattr (call_frame_t *frame, xlator_t *this,

         int32_t op_ret = -1;

         int32_t op_errno = EINVAL;

         br_stub_local_t *local = NULL;

+        br_stub_private_t  *priv = NULL;

         rootgfid[15] = 1;

@@ -1679,6 +1779,13 @@ br_stub_fgetxattr (call_frame_t *frame, xlator_t *this,

                 goto wind;

         }

+        if (br_stub_is_internal_xattr (name))

+                goto unwind;

+

+        priv = this->private;

+        if (!priv->do_versioning)

+                goto wind;

+

         /**

          * If xattr is node-uuid and the inode is marked bad, return EIO.

          * Returning EIO would result in AFR to choose correct node-uuid

@@ -1690,9 +1797,6 @@ br_stub_fgetxattr (call_frame_t *frame, xlator_t *this,

                 goto unwind;

         }

-        if (br_stub_is_internal_xattr (name))

-                goto unwind;

-

         /**

          * this special extended attribute is allowed only on root

          */

@@ -1741,16 +1845,23 @@ br_stub_readv (call_frame_t *frame, xlator_t *this,

         int32_t              op_ret   = -1;

         int32_t              op_errno = EINVAL;

         int32_t              ret      = -1;

+        br_stub_private_t   *priv     = NULL;

         GF_VALIDATE_OR_GOTO ("bit-rot-stub", this, unwind);

         GF_VALIDATE_OR_GOTO (this->name, frame, unwind);

+        GF_VALIDATE_OR_GOTO (this->name, this->private, unwind);

         GF_VALIDATE_OR_GOTO (this->name, fd, unwind);

         GF_VALIDATE_OR_GOTO (this->name, fd->inode, unwind);

+        priv = this->private;

+        if (!priv->do_versioning)

+                goto wind;

+

         ret = br_stub_check_bad_object (this, fd->inode, &op_ret, &op_errno);

         if (ret)

                 goto unwind;

+wind:

         STACK_WIND_TAIL (frame, FIRST_CHILD(this),

                          FIRST_CHILD(this)->fops->readv, fd, size, offset,

                          flags, xdata);

@@ -1774,8 +1885,8 @@ br_stub_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,

                     int32_t op_ret, int32_t op_errno, struct iatt *prebuf,

                     struct iatt *postbuf, dict_t *xdata)

 {

-        int32_t          ret   = 0;

-        br_stub_local_t *local = NULL;

+        int32_t            ret   = 0;

+        br_stub_local_t   *local = NULL;

         local = frame->local;

         frame->local = NULL;

@@ -1834,11 +1945,17 @@ br_stub_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,

         int32_t              ret         = -1;

         fop_writev_cbk_t     cbk         = default_writev_cbk;

         br_stub_local_t     *local       = NULL;

+        br_stub_private_t   *priv        = NULL;

         GF_VALIDATE_OR_GOTO ("bit-rot-stub", this, unwind);

+        GF_VALIDATE_OR_GOTO (this->name, this->private, unwind);

         GF_VALIDATE_OR_GOTO (this->name, frame, unwind);

         GF_VALIDATE_OR_GOTO (this->name, fd, unwind);

+        priv = this->private;

+        if (!priv->do_versioning)

+                goto wind;

+

         ret = br_stub_need_versioning (this, fd, &inc_version, &modified, &ctx;;

         if (ret)

                 goto unwind;

@@ -1909,8 +2026,8 @@ br_stub_ftruncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,

                        int32_t op_ret, int32_t op_errno, struct iatt *prebuf,

                        struct iatt *postbuf, dict_t *xdata)

 {

-        int32_t          ret   = -1;

-        br_stub_local_t *local = NULL;

+        int32_t            ret   = -1;

+        br_stub_local_t   *local = NULL;

         local = frame->local;

         frame->local = NULL;

@@ -1957,11 +2074,17 @@ br_stub_ftruncate (call_frame_t *frame, xlator_t *this, fd_t *fd,

         br_stub_inode_ctx_t *ctx         = NULL;

         int32_t              ret         = -1;

         fop_ftruncate_cbk_t  cbk         = default_ftruncate_cbk;

+        br_stub_private_t   *priv        = NULL;

         GF_VALIDATE_OR_GOTO ("bit-rot-stub", this, unwind);

+        GF_VALIDATE_OR_GOTO (this->name, this->private, unwind);

         GF_VALIDATE_OR_GOTO (this->name, frame, unwind);

         GF_VALIDATE_OR_GOTO (this->name, fd, unwind);

+        priv = this->private;

+        if (!priv->do_versioning)

+                goto wind;

+

         ret = br_stub_need_versioning (this, fd, &inc_version, &modified, &ctx;;

         if (ret)

                 goto unwind;

@@ -2081,12 +2204,18 @@ br_stub_truncate (call_frame_t *frame, xlator_t *this, loc_t *loc,

         int32_t              ret         = -1;

         fd_t                *fd          = NULL;

         fop_truncate_cbk_t   cbk         = default_truncate_cbk;

+        br_stub_private_t   *priv        = NULL;

         GF_VALIDATE_OR_GOTO ("bit-rot-stub", this, unwind);

+        GF_VALIDATE_OR_GOTO (this->name, this->private, unwind);

         GF_VALIDATE_OR_GOTO (this->name, frame, unwind);

         GF_VALIDATE_OR_GOTO (this->name, loc, unwind);

         GF_VALIDATE_OR_GOTO (this->name, loc->inode, unwind);

+        priv = this->private;

+        if (!priv->do_versioning)

+                goto wind;

+

         fd = fd_anonymous (loc->inode);

         if (!fd) {

                 gf_msg (this->name, GF_LOG_ERROR, 0,

@@ -2133,7 +2262,8 @@ br_stub_truncate (call_frame_t *frame, xlator_t *this, loc_t *loc,

  wind:

         STACK_WIND (frame, cbk, FIRST_CHILD(this),

                     FIRST_CHILD(this)->fops->truncate, loc, offset, xdata);

-        fd_unref (fd);

+        if (fd)

+                fd_unref (fd);

         return 0;

  cleanup_local:

@@ -2179,19 +2309,32 @@ br_stub_open (call_frame_t *frame, xlator_t *this,

         uint64_t             ctx_addr = 0;

         int32_t              op_ret   = -1;

         int32_t              op_errno = EINVAL;

+        br_stub_private_t   *priv     = NULL;

+        unsigned long        version  = BITROT_DEFAULT_CURRENT_VERSION;

         GF_VALIDATE_OR_GOTO ("bit-rot-stub", this, unwind);

+        GF_VALIDATE_OR_GOTO (this->name, this->private, unwind);

         GF_VALIDATE_OR_GOTO (this->name, loc, unwind);

         GF_VALIDATE_OR_GOTO (this->name, fd, unwind);

         GF_VALIDATE_OR_GOTO (this->name, fd->inode, unwind);

+        priv = this->private;

+

+        if (!priv->do_versioning)

+                goto wind;

+

         ret = br_stub_get_inode_ctx (this, fd->inode, &ctx_addr);

         if (ret) {

-                gf_msg (this->name, GF_LOG_ERROR, 0,

-                        BRS_MSG_GET_INODE_CONTEXT_FAILED, "failed to get the "

-                        "inode context for the file %s (gfid: %s)", loc->path,

-                        uuid_utoa (fd->inode->gfid));

-                goto unwind;

+                ret = br_stub_init_inode_versions (this, fd, fd->inode, version,

+                                                   _gf_true, _gf_false);

+                if (ret) {

+                        gf_msg (this->name, GF_LOG_ERROR, 0,

+                                BRS_MSG_GET_INODE_CONTEXT_FAILED,

+                                "failed to init the inode context for "

+                                "the file %s (gfid: %s)", loc->path,

+                                uuid_utoa (fd->inode->gfid));

+                        goto unwind;

+                }

         }

         ctx = (br_stub_inode_ctx_t *)(long)ctx_addr;

@@ -2272,10 +2415,16 @@ br_stub_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,

         uint64_t             ctx_addr = 0;

         br_stub_inode_ctx_t *ctx      = NULL;

         unsigned long        version  = BITROT_DEFAULT_CURRENT_VERSION;

+        br_stub_private_t   *priv     = NULL;

+

+        priv = this->private;

         if (op_ret < 0)

                 goto unwind;

+        if (!priv->do_versioning)

+                goto unwind;

+

         ret = br_stub_get_inode_ctx (this, fd->inode, &ctx_addr);

         if (ret < 0) {

                 ret = br_stub_init_inode_versions (this, fd, inode, version,

@@ -2324,10 +2473,16 @@ br_stub_mknod_cbk (call_frame_t *frame, void *cookie, xlator_t *this,

 {

         int32_t              ret      = -1;

         unsigned long        version  = BITROT_DEFAULT_CURRENT_VERSION;

+        br_stub_private_t   *priv     = NULL;

+

+        priv = this->private;

         if (op_ret < 0)

                 goto unwind;

+        if (!priv->do_versioning)

+                goto unwind;

+

         ret = br_stub_init_inode_versions (this, NULL, inode, version,

                                            _gf_true, _gf_false);

         /**

@@ -2478,6 +2633,9 @@ br_stub_readdir (call_frame_t *frame, xlator_t *this,

         br_stub_private_t    *priv = NULL;

         priv = this->private;

+        if (!priv->do_versioning)

+                goto out;

+

         if (gf_uuid_compare (fd->inode->gfid, priv->bad_object_dir_gfid))

                 goto out;

         stub = fop_readdir_stub (frame, br_stub_readdir_wrapper, fd, size, off,

@@ -2499,9 +2657,15 @@ br_stub_readdirp_cbk (call_frame_t *frame, void *cookie, xlator_t *this,

                       int op_ret, int op_errno, gf_dirent_t *entries,

                       dict_t *dict)

 {

-        int32_t      ret     = 0;

-        uint64_t     ctxaddr = 0;

-        gf_dirent_t *entry   = NULL;

+        int32_t            ret     = 0;

+        uint64_t           ctxaddr = 0;

+        gf_dirent_t       *entry   = NULL;

+        br_stub_private_t *priv    = NULL;

+        gf_boolean_t       ver_enabled = _gf_false;

+

+        BR_STUB_VER_ENABLED_IN_CALLPATH (frame, ver_enabled);

+        priv = this->private;

+        BR_STUB_VER_COND_GOTO (priv, (!ver_enabled), unwind);

         if (op_ret < 0)

                 goto unwind;

@@ -2549,9 +2713,13 @@ int

 br_stub_readdirp (call_frame_t *frame, xlator_t *this,

                   fd_t *fd, size_t size, off_t offset, dict_t *dict)

 {

-        int32_t ret = -1;

-        int op_errno = 0;

-        gf_boolean_t xref = _gf_false;

+        int32_t            ret      = -1;

+        int                op_errno = 0;

+        gf_boolean_t       xref     = _gf_false;

+        br_stub_private_t *priv     = NULL;

+

+        priv = this->private;

+        BR_STUB_VER_NOT_ACTIVE_THEN_GOTO (frame, priv, wind);

         op_errno = ENOMEM;

         if (!dict) {

@@ -2575,16 +2743,19 @@ br_stub_readdirp (call_frame_t *frame, xlator_t *this,

         if (ret)

                 goto unwind;

+wind:

         STACK_WIND (frame, br_stub_readdirp_cbk, FIRST_CHILD (this),

                     FIRST_CHILD(this)->fops->readdirp, fd, size,

                     offset, dict);

         goto unref_dict;

- unwind:

+unwind:

+        if (frame->local == (void *)0x1)

+                frame->local = NULL;

         STACK_UNWIND_STRICT (readdirp, frame, -1, op_errno, NULL, NULL);

         return 0;

- unref_dict:

+unref_dict:

         if (xref)

                 dict_unref (dict);

         return 0;

@@ -2653,6 +2824,12 @@ br_stub_lookup_cbk (call_frame_t *frame, void *cookie,

                     struct iatt *stbuf, dict_t *xattr, struct iatt *postparent)

 {

         int32_t ret = 0;

+        br_stub_private_t *priv = NULL;

+        gf_boolean_t  ver_enabled = _gf_false;

+

+        BR_STUB_VER_ENABLED_IN_CALLPATH(frame, ver_enabled);

+        priv = this->private;

+        BR_STUB_VER_COND_GOTO (priv, (!ver_enabled), unwind);

         if (op_ret < 0) {

                 (void) br_stub_handle_lookup_error (this, inode, op_errno);

@@ -2730,6 +2907,8 @@ br_stub_lookup (call_frame_t *frame,

         priv = this->private;

+        BR_STUB_VER_NOT_ACTIVE_THEN_GOTO (frame, priv, wind);

+

         if (!gf_uuid_compare (loc->gfid, priv->bad_object_dir_gfid) ||

             !gf_uuid_compare (loc->pargfid, priv->bad_object_dir_gfid)) {

@@ -2787,6 +2966,8 @@ br_stub_lookup (call_frame_t *frame,

         goto dealloc_dict;

  unwind:

+        if (frame->local == (void *) 0x1)

+                frame->local = NULL;

         STACK_UNWIND_STRICT (lookup, frame,

                              -1, op_errno, NULL, NULL, NULL, NULL);

  dealloc_dict:

@@ -2803,9 +2984,15 @@ br_stub_lookup (call_frame_t *frame,

 int

 br_stub_stat (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)

 {

-        int32_t ret      = 0;

-        int32_t op_ret   = -1;

-        int32_t op_errno = EINVAL;

+        int32_t            ret      = 0;

+        int32_t            op_ret   = -1;

+        int32_t            op_errno = EINVAL;

+        br_stub_private_t *priv     = NULL;

+

+        priv = this->private;

+

+        if (!priv->do_versioning)

+                goto wind;

         if (!IA_ISREG (loc->inode->ia_type))

                 goto wind;

@@ -2828,9 +3015,15 @@ unwind: