features/bit-rot: stub changes for showing bad objects in the status

Change-Id: If905132f6f1df4aebd9ab255e1e8c59902f84fe5
BUG: 1207627
Signed-off-by: Raghavendra Bhat <raghavendra@redhat.com>
Reviewed-on: http://review.gluster.org/12503
Tested-by: NetBSD Build System <jenkins@build.gluster.org>
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Venky Shankar <vshankar@redhat.com>

6 files changed, 955 insertions, 91 deletions

diff --git a/xlators/features/bit-rot/src/stub/Makefile.am b/xlators/features/bit-rot/src/stub/Makefile.am
index 5b5253c4ad5..1746344f1b0 100644
--- a/xlators/features/bit-rot/src/stub/Makefile.am
+++ b/xlators/features/bit-rot/src/stub/Makefile.am
@@ -3,13 +3,14 @@ xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/features
 
 bitrot_stub_la_LDFLAGS = $(GF_XLATOR_DEFAULT_LDFLAGS)
 
-bitrot_stub_la_SOURCES = bit-rot-stub.c
+bitrot_stub_la_SOURCES = bit-rot-stub-helpers.c bit-rot-stub.c
 bitrot_stub_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la
 
 noinst_HEADERS = bit-rot-stub.h bit-rot-common.h bit-rot-stub-mem-types.h \
                  bit-rot-object-version.h bit-rot-stub-messages.h
 
-AM_CPPFLAGS = $(GF_CPPFLAGS) -I$(top_srcdir)/libglusterfs/src
+AM_CPPFLAGS = $(GF_CPPFLAGS) -I$(top_srcdir)/libglusterfs/src \
+	      -I$(top_srcdir)/rpc/xdr/src -I$(top_srcdir)/rpc/rpc-lib/src
 
 AM_CFLAGS = -Wall $(GF_CFLAGS)
 
diff --git a/xlators/features/bit-rot/src/stub/bit-rot-common.h b/xlators/features/bit-rot/src/stub/bit-rot-common.h
index bcf931a2b0b..2afc9f47c29 100644
--- a/xlators/features/bit-rot/src/stub/bit-rot-common.h
+++ b/xlators/features/bit-rot/src/stub/bit-rot-common.h
@@ -21,6 +21,8 @@
 #define BR_VXATTR_ALL_MISSING                           \
         (BR_VXATTR_VERSION | BR_VXATTR_SIGNATURE)
 
+#define BR_BAD_OBJ_CONTAINER (uuid_t){0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 8}
+
 typedef enum br_vxattr_state {
         BR_VXATTR_STATUS_FULL     = 0,
         BR_VXATTR_STATUS_MISSING  = 1,
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
new file mode 100644
index 00000000000..f637e150a57
--- /dev/null
+++ b/xlators/features/bit-rot/src/stub/bit-rot-stub-helpers.c
@@ -0,0 +1,606 @@
+/*
+   Copyright (c) 2015 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 "bit-rot-stub.h"
+
+br_stub_fd_t *
+br_stub_fd_new (void)
+{
+        br_stub_fd_t    *br_stub_fd = NULL;
+
+        br_stub_fd = GF_CALLOC (1, sizeof (*br_stub_fd),
+                                gf_br_stub_mt_br_stub_fd_t);
+
+        return br_stub_fd;
+}
+
+int
+__br_stub_fd_ctx_set (xlator_t *this, fd_t *fd, br_stub_fd_t *br_stub_fd)
+{
+        uint64_t    value = 0;
+        int         ret   = -1;
+
+        GF_VALIDATE_OR_GOTO ("bit-rot-stub", this, out);
+        GF_VALIDATE_OR_GOTO (this->name, fd, out);
+        GF_VALIDATE_OR_GOTO (this->name, br_stub_fd, out);
+
+        value = (uint64_t)(long) br_stub_fd;
+
+        ret = __fd_ctx_set (fd, this, value);
+
+out:
+        return ret;
+}
+
+br_stub_fd_t *
+__br_stub_fd_ctx_get (xlator_t *this, fd_t *fd)
+{
+        br_stub_fd_t *br_stub_fd = NULL;
+        uint64_t  value  = 0;
+        int       ret    = -1;
+
+        GF_VALIDATE_OR_GOTO ("bit-rot-stub", this, out);
+        GF_VALIDATE_OR_GOTO (this->name, fd, out);
+
+        ret = __fd_ctx_get (fd, this, &value);
+        if (ret)
+                return NULL;
+
+        br_stub_fd = (br_stub_fd_t *) ((long) value);
+
+out:
+        return br_stub_fd;
+}
+
+br_stub_fd_t *
+br_stub_fd_ctx_get (xlator_t *this, fd_t *fd)
+{
+        br_stub_fd_t *br_stub_fd = NULL;
+
+        GF_VALIDATE_OR_GOTO ("bit-rot-stub", this, out);
+        GF_VALIDATE_OR_GOTO (this->name, fd, out);
+
+        LOCK (&fd->lock);
+        {
+                br_stub_fd = __br_stub_fd_ctx_get (this, fd);
+        }
+        UNLOCK (&fd->lock);
+
+out:
+        return br_stub_fd;
+}
+
+int32_t
+br_stub_fd_ctx_set (xlator_t *this, fd_t *fd, br_stub_fd_t *br_stub_fd)
+{
+        int32_t    ret = -1;
+
+        GF_VALIDATE_OR_GOTO ("bit-rot-stub", this, out);
+        GF_VALIDATE_OR_GOTO (this->name, fd, out);
+        GF_VALIDATE_OR_GOTO (this->name, br_stub_fd, out);
+
+        LOCK (&fd->lock);
+        {
+                ret = __br_stub_fd_ctx_set (this, fd, br_stub_fd);
+        }
+        UNLOCK (&fd->lock);
+
+out:
+        return ret;
+}
+
+
+/**
+ * prints the path to the bad object's entry into the buffer provided.
+ * @priv: xlator private
+ * @filename: gfid of the bad object.
+ * @file_path: buffer provided into which path of the bad object is printed
+ *            using above 2 arguments.
+ */
+static void
+br_stub_link_path (br_stub_private_t *priv, const char *filename,
+                        char *file_path, size_t len)
+{
+        snprintf (file_path, len, "%s/%s", priv->stub_basepath, filename);
+}
+
+/**
+ * Prints the path of the object which acts as a container for all the bad
+ * objects. Each new entry corresponding to a bad object is a hard link to
+ * the object with name "stub-0000000000000008".
+ * @priv: xlator's private
+ * @stub_gfid_path: buffer into which the path to the container of bad objects
+ *                  is printed.
+ */
+static void
+br_stub_container_entry (br_stub_private_t *priv, char *stub_gfid_path,
+                         size_t len)
+{
+
+        snprintf (stub_gfid_path, len, "%s/stub-%s", priv->stub_basepath,
+                  uuid_utoa (priv->bad_object_dir_gfid));
+}
+
+/**
+ * Prints the path to the bad object's entry into the buffer provided.
+ * @priv: xlator private
+ * @gfid: gfid of the bad object.
+ * @gfid_path: buffer provided into which path of the bad object is printed
+ *            using above 2 arguments.
+ * This function is same as br_stub_link_path. But in this function the
+ * gfid of the bad object is obtained as an argument (i.e. uuid_t gfid),
+ * where as in br_stub_link_path, the gfid is received as filename
+ * (i.e. char *filename)
+ */
+static void
+br_stub_linked_entry (br_stub_private_t *priv, char *gfid_path, uuid_t gfid,
+                        size_t len)
+{
+        snprintf (gfid_path, len, "%s/%s", priv->stub_basepath,
+                  uuid_utoa (gfid));
+}
+
+/**
+ * Adds an entry to the bad objects directory.
+ * @gfid: gfid of the bad object being added to the bad objects directory
+ */
+int
+br_stub_add (xlator_t *this, uuid_t gfid)
+{
+        char              gfid_path[PATH_MAX] = {0};
+        char              bad_gfid_path[PATH_MAX] = {0};
+        int               ret = 0;
+        uuid_t            index = {0};
+        br_stub_private_t *priv = NULL;
+        struct stat       st = {0};
+        int               fd = 0;
+
+        priv = this->private;
+        GF_ASSERT_AND_GOTO_WITH_ERROR (this->name, !gf_uuid_is_null (gfid),
+                                       out, errno, EINVAL);
+
+        br_stub_linked_entry (priv, gfid_path, gfid, sizeof (gfid_path));
+
+        ret = sys_stat (gfid_path, &st);
+        if (!ret)
+                goto out;
+        br_stub_container_entry (priv, bad_gfid_path, sizeof (bad_gfid_path));
+
+        ret = sys_link (bad_gfid_path, gfid_path);
+        if (ret) {
+                if ((errno != ENOENT) && (errno != EMLINK) && (errno != EEXIST))
+                        goto out;
+
+                /*
+                 * Continue with success. At least we'll have half of the
+                 * functionality, in the sense, object is marked bad and
+                 * would be inaccessible. It's only scrub status that would
+                 * show up less number of objects. That's fine as we'll have
+                 * the log files that will have the missing information.
+                 */
+                gf_msg (this->name, GF_LOG_WARNING, errno, BRS_MSG_LINK_FAIL,
+                        "failed to record  gfid [%s]", uuid_utoa (gfid));
+        }
+
+        return 0;
+out:
+        return -1;
+}
+
+static int
+br_stub_check_stub_directory (xlator_t *this, char *fullpath)
+{
+        int ret = 0;
+        struct stat st = {0,};
+
+        ret = stat (fullpath, &st);
+        if (!ret && !S_ISDIR (st.st_mode))
+                goto error_return;
+        if (ret) {
+                if (errno != ENOENT)
+                        goto error_return;
+                ret = mkdir_p (fullpath, 0600, _gf_true);
+        }
+
+        if (ret)
+                gf_msg (this->name, GF_LOG_ERROR, errno,
+                        BRS_MSG_BAD_OBJECT_DIR_FAIL,
+                        "failed to create stub directory [%s]", fullpath);
+        return ret;
+
+error_return:
+        gf_msg (this->name, GF_LOG_ERROR, errno,
+                BRS_MSG_BAD_OBJECT_DIR_FAIL,
+                "Failed to verify stub directory [%s]", fullpath);
+        return -1;
+}
+
+/**
+ * Function to create the container for the bad objects within the bad objects
+ * directory.
+ */
+static int
+br_stub_check_stub_file (xlator_t *this, char *path)
+{
+        int ret = 0;
+        int fd = -1;
+        struct stat st = {0,};
+
+        ret = stat (path, &st);
+        if (!ret && !S_ISREG (st.st_mode))
+                goto error_return;
+        if (ret) {
+                if (errno != ENOENT)
+                        goto error_return;
+                fd = sys_creat (path, 0);
+                if (fd < 0)
+                        gf_msg (this->name, GF_LOG_ERROR, errno,
+                                BRS_MSG_BAD_OBJECT_DIR_FAIL,
+                                "Failed ot create stub file [%s]", path);
+        }
+
+        if (fd >= 0) {
+                sys_close (fd);
+                ret = 0;
+        }
+
+        return ret;
+
+error_return:
+        gf_msg (this->name, GF_LOG_ERROR, errno,
+                BRS_MSG_BAD_OBJECT_DIR_FAIL, "Failed ot verify stub file [%s]", path);
+        return -1;
+}
+
+int
+br_stub_dir_create (xlator_t *this, br_stub_private_t *priv)
+{
+        int          ret = -1;
+        char         fullpath[PATH_MAX] = {0};
+        char         stub_gfid_path[PATH_MAX] = {0, };
+        char         path[PATH_MAX] = {0};
+        size_t       len = 0;
+
+        gf_uuid_copy (priv->bad_object_dir_gfid, BR_BAD_OBJ_CONTAINER);
+
+        snprintf (fullpath, sizeof (fullpath), "%s", priv->stub_basepath);
+
+        br_stub_container_entry (priv, stub_gfid_path, sizeof (stub_gfid_path));
+
+        ret = br_stub_check_stub_directory (this, fullpath);
+        if (ret)
+                goto out;
+        ret = br_stub_check_stub_file (this, stub_gfid_path);
+        if (ret)
+                goto out;
+
+        return 0;
+
+out:
+        return -1;
+}
+
+call_stub_t *
+__br_stub_dequeue (struct list_head *callstubs)
+{
+        call_stub_t *stub = NULL;
+
+        if (!list_empty (callstubs)) {
+                stub = list_entry (callstubs->next, call_stub_t, list);
+                list_del_init (&stub->list);
+        }
+
+        return stub;
+}
+
+void
+__br_stub_enqueue (struct list_head *callstubs, call_stub_t *stub)
+{
+        list_add_tail (&stub->list, callstubs);
+}
+
+void
+br_stub_worker_enqueue (xlator_t *this, call_stub_t *stub)
+{
+        br_stub_private_t    *priv = NULL;
+
+        priv = this->private;
+        pthread_mutex_lock (&priv->container.bad_lock);
+        {
+                __br_stub_enqueue (&priv->container.bad_queue, stub);
+                pthread_cond_signal (&priv->container.bad_cond);
+        }
+        pthread_mutex_unlock (&priv->container.bad_lock);
+}
+
+void *
+br_stub_worker (void *data)
+{
+        br_stub_private_t     *priv = NULL;
+        xlator_t         *this = NULL;
+        call_stub_t      *stub = NULL;
+        int               ret = 0;
+
+
+        THIS = data;
+        this = data;
+        priv = this->private;
+
+        for (;;) {
+                pthread_mutex_lock (&priv->container.bad_lock);
+                {
+                        while (list_empty (&priv->container.bad_queue)) {
+                              ret = pthread_cond_wait (&priv->container.bad_cond,
+                                                       &priv->container.bad_lock);
+                        }
+
+                        stub = __br_stub_dequeue (&priv->container.bad_queue);
+                }
+                pthread_mutex_unlock (&priv->container.bad_lock);
+
+                if (stub) /* guard against spurious wakeups */
+                        call_resume (stub);
+        }
+
+        return NULL;
+}
+
+int32_t
+br_stub_lookup_wrapper (call_frame_t *frame, xlator_t *this,
+                      loc_t *loc, dict_t *xattr_req)
+{
+        br_stub_private_t *priv        = NULL;
+        struct stat        lstatbuf    = {0};
+        int                ret         = 0;
+        int32_t            op_errno    = EINVAL;
+        int32_t            op_ret      = -1;
+        struct iatt        stbuf       = {0, };
+        struct iatt        postparent  = {0,};
+        dict_t            *xattr       = NULL;
+
+        priv = this->private;
+
+        VALIDATE_OR_GOTO (loc, done);
+        if (gf_uuid_compare (loc->gfid, priv->bad_object_dir_gfid))
+                goto done;
+
+        ret = sys_lstat (priv->stub_basepath, &lstatbuf);
+        if (ret) {
+                gf_msg_debug (this->name, errno, "Stat failed on stub bad "
+                              "object dir");
+                op_errno = errno;
+                goto done;
+        } else if (!S_ISDIR (lstatbuf.st_mode)) {
+                gf_msg_debug (this->name, errno, "bad object container is not "
+                              "a directory");
+                op_errno = ENOTDIR;
+                goto done;
+        }
+
+        iatt_from_stat (&stbuf, &lstatbuf);
+        gf_uuid_copy (stbuf.ia_gfid, priv->bad_object_dir_gfid);
+
+        op_ret = op_errno = 0;
+        xattr = dict_new ();
+        if (!xattr) {
+                op_ret = -1;
+                op_errno = ENOMEM;
+        }
+
+done:
+        STACK_UNWIND_STRICT (lookup, frame, op_ret, op_errno,
+                             loc->inode, &stbuf, xattr, &postparent);
+        if (xattr)
+                dict_unref (xattr);
+        return 0;
+}
+
+static int
+is_bad_gfid_file_current (char *filename, uuid_t gfid)
+{
+        char current_stub_gfid[GF_UUID_BUF_SIZE + 16] = {0, };
+
+        snprintf (current_stub_gfid, sizeof current_stub_gfid,
+                  "stub-%s", uuid_utoa(gfid));
+        return (!strcmp(filename, current_stub_gfid));
+}
+
+static void
+check_delete_stale_bad_file (xlator_t *this, char *filename)
+{
+        int             ret = 0;
+        struct stat     st = {0};
+        char            filepath[PATH_MAX] = {0};
+        br_stub_private_t    *priv = NULL;
+
+        priv = this->private;
+
+        if (is_bad_gfid_file_current (filename, priv->bad_object_dir_gfid))
+                return;
+
+        br_stub_link_path (priv, filename, filepath, sizeof (filepath));
+
+        ret = sys_stat (filepath, &st);
+        if (!ret && st.st_nlink == 1)
+                sys_unlink (filepath);
+}
+
+static int
+br_stub_fill_readdir (fd_t *fd, br_stub_fd_t *fctx, DIR *dir, off_t off,
+                      size_t size, gf_dirent_t *entries)
+{
+        off_t     in_case = -1;
+        off_t     last_off = 0;
+        size_t    filled = 0;
+        int       count = 0;
+        char      entrybuf[sizeof(struct dirent) + 256 + 8];
+        struct dirent  *entry          = NULL;
+        int32_t              this_size      = -1;
+        gf_dirent_t          *this_entry     = NULL;
+        xlator_t             *this = NULL;
+
+        this = THIS;
+        if (!off) {
+                rewinddir (dir);
+        } else {
+                seekdir (dir, off);
+#ifndef GF_LINUX_HOST_OS
+                if ((u_long)telldir(dir) != off &&
+                    off != fctx->bad_object.dir_eof) {
+                        gf_msg (THIS->name, GF_LOG_ERROR, 0,
+                                BRS_MSG_BAD_OBJECT_DIR_SEEK_FAIL,
+                                "seekdir(0x%llx) failed on dir=%p: "
+				"Invalid argument (offset reused from "
+				"another DIR * structure?)", off, dir);
+                        errno = EINVAL;
+                        count = -1;
+                        goto out;
+                }
+#endif /* GF_LINUX_HOST_OS */
+        }
+
+        while (filled <= size) {
+                in_case = (u_long)telldir (dir);
+
+                if (in_case == -1) {
+                        gf_msg (THIS->name, GF_LOG_ERROR, 0,
+                                BRS_MSG_BAD_OBJECT_DIR_TELL_FAIL,
+                                "telldir failed on dir=%p: %s",
+                                dir, strerror (errno));
+                        goto out;
+                }
+
+                errno = 0;
+                entry = NULL;
+                readdir_r (dir, (struct dirent *)entrybuf, &entry);
+
+                if (!entry) {
+                        if (errno == EBADF) {
+                                gf_msg (THIS->name, GF_LOG_WARNING, 0,
+                                        BRS_MSG_BAD_OBJECT_DIR_READ_FAIL,
+                                        "readdir failed on dir=%p: %s",
+                                        dir, strerror (errno));
+                                goto out;
+                        }
+                        break;
+                }
+
+                if (!strcmp (entry->d_name, ".") ||
+                    !strcmp (entry->d_name, ".."))
+                        continue;
+
+                if (!strncmp (entry->d_name, "stub-",
+                              strlen ("stub-"))) {
+                        check_delete_stale_bad_file (this, entry->d_name);
+                        continue;
+                }
+
+                this_size = max (sizeof (gf_dirent_t),
+                                 sizeof (gfs3_dirplist))
+                        + strlen (entry->d_name) + 1;
+
+                if (this_size + filled > size) {
+                        seekdir (dir, in_case);
+#ifndef GF_LINUX_HOST_OS
+                        if ((u_long)telldir(dir) != in_case &&
+                            in_case != fctx->bad_object.dir_eof) {
+				gf_msg (THIS->name, GF_LOG_ERROR, 0,
+                                        BRS_MSG_BAD_OBJECT_DIR_SEEK_FAIL,
+					"seekdir(0x%llx) failed on dir=%p: "
+					"Invalid argument (offset reused from "
+					"another DIR * structure?)",
+					in_case, dir);
+				errno = EINVAL;
+				count = -1;
+				goto out;
+                        }
+#endif /* GF_LINUX_HOST_OS */
+                        break;
+                }
+
+                this_entry = gf_dirent_for_name (entry->d_name);
+
+                if (!this_entry) {
+                        gf_msg (THIS->name, GF_LOG_ERROR, 0,
+                                BRS_MSG_NO_MEMORY,
+                                "could not create gf_dirent for entry %s: (%s)",
+                                entry->d_name, strerror (errno));
+                        goto out;
+                }
+                /*
+                 * we store the offset of next entry here, which is
+                 * probably not intended, but code using syncop_readdir()
+                 * (glfs-heal.c, afr-self-heald.c, pump.c) rely on it
+                 * for directory read resumption.
+                 */
+                last_off = (u_long)telldir(dir);
+                this_entry->d_off = last_off;
+                this_entry->d_ino = entry->d_ino;
+
+                list_add_tail (&this_entry->list, &entries->list);
+
+                filled += this_size;
+                count++;
+        }
+
+        if ((!sys_readdir (dir) && (errno == 0))) {
+                /* Indicate EOF */
+                errno = ENOENT;
+                /* Remember EOF offset for later detection */
+                fctx->bad_object.dir_eof = last_off;
+        }
+out:
+        return count;
+}
+
+int32_t
+br_stub_readdir_wrapper (call_frame_t *frame, xlator_t *this,
+                         fd_t *fd, size_t size, off_t off, dict_t *xdata)
+{
+        br_stub_fd_t         *fctx           = NULL;
+        DIR                  *dir            = NULL;
+        int                   ret            = -1;
+        int32_t               op_ret         = -1;
+        int32_t               op_errno       = 0;
+        int                   count          = 0;
+        gf_dirent_t           entries;
+
+        INIT_LIST_HEAD (&entries.list);
+
+        fctx = br_stub_fd_ctx_get (this, fd);
+        if (!fctx) {
+                gf_msg (this->name, GF_LOG_WARNING, 0,
+                        BRS_MSG_GET_FD_CONTEXT_FAILED,
+                        "pfd is NULL, fd=%p", fd);
+                op_errno = -ret;
+                goto done;
+        }
+
+        dir = fctx->bad_object.dir;
+
+        if (!dir) {
+                gf_msg (this->name, GF_LOG_WARNING, 0,
+                        BRS_MSG_BAD_HANDLE_DIR_NULL,
+                        "dir is NULL for fd=%p", fd);
+                op_errno = EINVAL;
+                goto done;
+        }
+
+        count = br_stub_fill_readdir (fd, fctx, dir, off, size, &entries);
+
+        /* pick ENOENT to indicate EOF */
+        op_errno = errno;
+        op_ret = count;
+done:
+        STACK_UNWIND_STRICT (readdir, frame, op_ret, op_errno, &entries, xdata);
+        gf_dirent_free (&entries);
+        return 0;
+}
+
diff --git a/xlators/features/bit-rot/src/stub/bit-rot-stub-messages.h b/xlators/features/bit-rot/src/stub/bit-rot-stub-messages.h
index 532c2beb5c1..ee39e4c6d9f 100644
--- a/xlators/features/bit-rot/src/stub/bit-rot-stub-messages.h
+++ b/xlators/features/bit-rot/src/stub/bit-rot-stub-messages.h
@@ -40,7 +40,7 @@
  */
 
 #define GLFS_BITROT_STUB_BASE                   GLFS_MSGID_COMP_BITROT_STUB
-#define GLFS_BITROT_STUB_NUM_MESSAGES           15
+#define GLFS_BITROT_STUB_NUM_MESSAGES           30
 #define GLFS_MSGID_END         (GLFS_BITROT_STUB_BASE + \
                                 GLFS_BITROT_STUB_NUM_MESSAGES + 1)
 /* Messaged with message IDs */
@@ -188,6 +188,76 @@
  * @recommendedaction
  *
  */
+#define BRS_MSG_BAD_CONTAINER_FAIL          (GLFS_BITROT_STUB_BASE + 21)
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ *
+ */
+#define BRS_MSG_BAD_OBJECT_DIR_FAIL        (GLFS_BITROT_STUB_BASE + 22)
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ *
+ */
+#define BRS_MSG_BAD_OBJECT_DIR_SEEK_FAIL   (GLFS_BITROT_STUB_BASE + 23)
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ *
+ */
+#define BRS_MSG_BAD_OBJECT_DIR_TELL_FAIL   (GLFS_BITROT_STUB_BASE + 24)
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ *
+ */
+#define BRS_MSG_BAD_OBJECT_DIR_READ_FAIL   (GLFS_BITROT_STUB_BASE + 25)
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ *
+ */
+#define BRS_MSG_GET_FD_CONTEXT_FAILED      (GLFS_BITROT_STUB_BASE + 26)
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ *
+ */
+#define BRS_MSG_BAD_HANDLE_DIR_NULL        (GLFS_BITROT_STUB_BASE + 27)
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ *
+ */
+#define BRS_MSG_BAD_OBJ_THREAD_FAIL        (GLFS_BITROT_STUB_BASE + 28)
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ *
+ */
+#define BRS_MSG_BAD_OBJ_DIR_CLOSE_FAIL     (GLFS_BITROT_STUB_BASE + 29)
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ *
+ */
+#define BRS_MSG_LINK_FAIL                  (GLFS_BITROT_STUB_BASE + 30)
+/*!
+ * @messageid
+ * @diagnosis
+ * @recommendedaction
+ *
+ */
 /*------------*/
 
 #define glfs_msg_end_x GLFS_MSGID_END, "Invalid: End of messages"
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 85fad6925c1..83a78604665 100644
--- a/xlators/features/bit-rot/src/stub/bit-rot-stub.c
+++ b/xlators/features/bit-rot/src/stub/bit-rot-stub.c
@@ -55,6 +55,66 @@ mem_acct_init (xlator_t *this)
 }
 
 int32_t
+br_stub_bad_object_container_init (xlator_t *this, br_stub_private_t *priv)
+{
+        pthread_attr_t  w_attr;
+        int32_t         ret          = -1;
+
+        ret = pthread_cond_init(&priv->container.bad_cond, NULL);
+        if (ret != 0) {
+                gf_msg (this->name, GF_LOG_ERROR, 0,
+                        BRS_MSG_BAD_OBJ_THREAD_FAIL,
+                        "pthread_cond_init failed (%d)", ret);
+                goto out;
+        }
+
+        ret = pthread_mutex_init(&priv->container.bad_lock, NULL);
+        if (ret != 0) {
+                gf_msg (this->name, GF_LOG_ERROR, 0,
+                        BRS_MSG_BAD_OBJ_THREAD_FAIL,
+                        "pthread_mutex_init failed (%d)", ret);
+                goto cleanup_cond;
+        }
+
+        ret = pthread_attr_init (&w_attr);
+        if (ret != 0) {
+                gf_msg (this->name, GF_LOG_ERROR, 0,
+                        BRS_MSG_BAD_OBJ_THREAD_FAIL,
+                        "pthread_attr_init failed (%d)", ret);
+                goto cleanup_lock;
+        }
+
+        ret = pthread_attr_setstacksize (&w_attr, BAD_OBJECT_THREAD_STACK_SIZE);
+        if (ret == EINVAL) {
+                gf_msg (this->name, GF_LOG_WARNING, 0,
+                        BRS_MSG_BAD_OBJ_THREAD_FAIL,
+                        "Using default thread stack size");
+        }
+
+        INIT_LIST_HEAD (&priv->container.bad_queue);
+        ret = br_stub_dir_create (this, priv);
+        if (ret < 0)
+                goto cleanup_lock;
+
+        ret = gf_thread_create (&priv->container.thread, &w_attr, br_stub_worker, this);
+        if (ret)
+                goto cleanup_attr;
+
+        return 0;
+
+cleanup_attr:
+        pthread_attr_destroy (&w_attr);
+cleanup_lock:
+        pthread_mutex_destroy (&priv->container.bad_lock);
+cleanup_cond:
+        pthread_cond_destroy (&priv->container.bad_cond);
+out:
+        return -1;
+}
+
+#define BR_STUB_QUARANTINE_DIR GF_HIDDEN_PATH"/quanrantine"
+
+int32_t
 init (xlator_t *this)
 {
         int32_t ret = 0;
@@ -81,6 +141,9 @@ init (xlator_t *this)
         GF_OPTION_INIT ("export", tmp, str, free_mempool);
         memcpy (priv->export, tmp, strlen (tmp) + 1);
 
+        (void) snprintf (priv->stub_basepath, PATH_MAX,
+                         "%s/%s", priv->export, BR_STUB_QUARANTINE_DIR);
+
         (void) gettimeofday (&tv, NULL);
 
         /* boot time is in network endian format */
@@ -95,8 +158,16 @@ init (xlator_t *this)
         if (ret != 0)
                 goto cleanup_lock;
 
+        ret = br_stub_bad_object_container_init (this, priv);
+        if (ret) {
+                gf_msg (this->name, GF_LOG_ERROR, 0, BRS_MSG_BAD_CONTAINER_FAIL,
+                        "failed to launch the thread for storing bad gfids");
+                goto cleanup_lock;
+        }
+
+        this->private = priv;
+
         gf_msg_debug (this->name, 0, "bit-rot stub loaded");
-	this->private = priv;
 
         return 0;
 
@@ -117,6 +188,7 @@ fini (xlator_t *this)
         int32_t ret = 0;
 	br_stub_private_t *priv = this->private;
         struct br_stub_signentry *sigstub = NULL;
+        call_stub_t *stub = NULL;
 
         if (!priv)
                 return;
@@ -141,6 +213,24 @@ fini (xlator_t *this)
         pthread_mutex_destroy (&priv->lock);
         pthread_cond_destroy (&priv->cond);
 
+        ret = gf_thread_cleanup_xint (priv->container.thread);
+        if (ret) {
+                gf_msg (this->name, GF_LOG_ERROR, 0,
+                        BRS_MSG_CANCEL_SIGN_THREAD_FAILED,
+                        "Could not cancel sign serializer thread");
+                goto out;
+        }
+
+        while (!list_empty (&priv->container.bad_queue)) {
+                stub = list_first_entry (&priv->container.bad_queue, call_stub_t,
+                                         list);
+                list_del_init (&stub->list);
+                call_stub_destroy (stub);
+        };
+
+        pthread_mutex_destroy (&priv->container.bad_lock);
+        pthread_cond_destroy (&priv->container.bad_cond);
+
         this->private = NULL;
 	GF_FREE (priv);
 
@@ -1018,6 +1108,8 @@ br_stub_fsetxattr_bad_object_cbk (call_frame_t *frame, void *cookie,
                         "failed to mark object %s as bad",
                         uuid_utoa (local->u.context.inode->gfid));
 
+        ret = br_stub_add (this, local->u.context.inode->gfid);
+
 unwind:
         STACK_UNWIND_STRICT (fsetxattr, frame, op_ret, op_errno, xdata);
         br_stub_cleanup_local (local);
@@ -2306,6 +2398,74 @@ br_stub_lookup_version (xlator_t *this,
 
 /** {{{ */
 
+int32_t
+br_stub_opendir (call_frame_t *frame, xlator_t *this,
+               loc_t *loc, fd_t *fd, dict_t *xdata)
+{
+        br_stub_private_t    *priv = NULL;
+        br_stub_fd_t         *fd_ctx = NULL;
+        int32_t               op_ret = -1;
+        int32_t               op_errno = EINVAL;
+
+        priv = this->private;
+        if (gf_uuid_compare (fd->inode->gfid, priv->bad_object_dir_gfid))
+                goto normal;
+
+        fd_ctx = br_stub_fd_new ();
+        if (!fd_ctx) {
+                op_errno = ENOMEM;
+                goto unwind;
+        }
+
+        fd_ctx->bad_object.dir_eof = -1;
+        fd_ctx->bad_object.dir = sys_opendir (priv->stub_basepath);
+        if (!fd_ctx->bad_object.dir) {
+                op_errno = errno;
+                goto err_freectx;
+        }
+
+        op_ret = br_stub_fd_ctx_set (this, fd, fd_ctx);
+        if (!op_ret)
+                goto unwind;
+
+        sys_closedir (fd_ctx->bad_object.dir);
+
+err_freectx:
+        GF_FREE (fd_ctx);
+unwind:
+        STACK_UNWIND_STRICT (opendir, frame, op_ret, op_errno, fd, NULL);
+        return 0;
+
+normal:
+        STACK_WIND (frame, default_opendir_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->opendir, loc, fd, xdata);
+        return 0;
+}
+
+int32_t
+br_stub_readdir (call_frame_t *frame, xlator_t *this,
+               fd_t *fd, size_t size, off_t off, dict_t *xdata)
+{
+        call_stub_t     *stub = NULL;
+        br_stub_private_t    *priv = NULL;
+
+        priv = this->private;
+        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,
+                                 xdata);
+        if (!stub) {
+                STACK_UNWIND_STRICT (readdir, frame, -1, ENOMEM, NULL, NULL);
+                return 0;
+        }
+        br_stub_worker_enqueue (this, stub);
+        return 0;
+out:
+        STACK_WIND (frame, default_readdir_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->readdir, fd, size, off, xdata);
+        return 0;
+}
+
 int
 br_stub_readdirp_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                       int op_ret, int op_errno, gf_dirent_t *entries,
@@ -2480,11 +2640,28 @@ br_stub_lookup (call_frame_t *frame,
         void *cookie = NULL;
         uint64_t ctx_addr = 0;
         gf_boolean_t xref = _gf_false;
+        br_stub_private_t *priv = NULL;
+        call_stub_t       *stub = NULL;
 
         GF_VALIDATE_OR_GOTO ("bit-rot-stub", this, unwind);
         GF_VALIDATE_OR_GOTO (this->name, loc, unwind);
         GF_VALIDATE_OR_GOTO (this->name, loc->inode, unwind);
 
+        priv = this->private;
+
+        if (!gf_uuid_compare (loc->gfid, priv->bad_object_dir_gfid) ||
+            !gf_uuid_compare (loc->pargfid, priv->bad_object_dir_gfid)) {
+
+                stub = fop_lookup_stub (frame, br_stub_lookup_wrapper, loc,
+                                        xdata);
+                if (!stub) {
+                        op_errno = ENOMEM;
+                        goto unwind;
+                }
+                br_stub_worker_enqueue (this, stub);
+                return 0;
+        }
+
         ret = br_stub_get_inode_ctx (this, loc->inode, &ctx_addr);
         if (ret < 0)
                 ctx_addr = 0;
@@ -2727,6 +2904,31 @@ br_stub_release (xlator_t *this, fd_t *fd)
         return 0;
 }
 
+int32_t
+br_stub_releasedir (xlator_t *this, fd_t *fd)
+{
+        br_stub_fd_t   *fctx = NULL;
+        uint64_t        ctx  = 0;
+        int             ret  = 0;
+
+        ret = fd_ctx_del (fd, this, &ctx);
+        if (ret < 0)
+                goto out;
+
+        fctx = (br_stub_fd_t *) (long) ctx;
+        if (fctx->bad_object.dir) {
+                ret = sys_closedir (fctx->bad_object.dir);
+                if (ret)
+                        gf_msg (this->name, GF_LOG_ERROR, 0,
+                                BRS_MSG_BAD_OBJ_DIR_CLOSE_FAIL,
+                                "closedir error: %s", strerror (errno));
+        }
+
+        GF_FREE (fctx);
+out:
+        return 0;
+}
+
 /** }}} */
 
 /** {{{ */
@@ -2790,6 +2992,8 @@ struct xlator_fops fops = {
         .removexattr = br_stub_removexattr,
         .fremovexattr = br_stub_fremovexattr,
         .setxattr  = br_stub_setxattr,
+        .opendir   = br_stub_opendir,
+        .readdir   = br_stub_readdir,
 };
 
 struct xlator_cbks cbks = {
diff --git a/xlators/features/bit-rot/src/stub/bit-rot-stub.h b/xlators/features/bit-rot/src/stub/bit-rot-stub.h
index 9362c129303..26ed5fe0bdb 100644
--- a/xlators/features/bit-rot/src/stub/bit-rot-stub.h
+++ b/xlators/features/bit-rot/src/stub/bit-rot-stub.h
@@ -17,9 +17,12 @@
 #include "defaults.h"
 #include "call-stub.h"
 #include "bit-rot-stub-mem-types.h"
-
+#include "syscall.h"
 #include "bit-rot-common.h"
 #include "bit-rot-stub-messages.h"
+#include "glusterfs3-xdr.h"
+
+#define BAD_OBJECT_THREAD_STACK_SIZE   ((size_t)(1024*1024))
 
 typedef int (br_stub_version_cbk) (call_frame_t *, void *,
                                    xlator_t *, int32_t, int32_t, dict_t *);
@@ -38,6 +41,10 @@ typedef struct br_stub_inode_ctx {
 typedef struct br_stub_fd {
         fd_t *fd;
         struct list_head list;
+        struct bad_object_dir {
+                DIR *dir;
+                off_t dir_eof;
+        } bad_object;
 } br_stub_fd_t;
 
 #define I_DIRTY  (1<<0)        /* inode needs writeback */
@@ -77,10 +84,35 @@ typedef struct br_stub_private {
 
         struct list_head squeue;      /* ordered signing queue */
         pthread_t signth;
-
+        struct bad_objects_container {
+                pthread_t thread;
+                pthread_mutex_t bad_lock;
+                pthread_cond_t  bad_cond;
+                struct list_head bad_queue;
+        } container;
         struct mem_pool *local_pool;
+
+        char stub_basepath[PATH_MAX];
+
+        uuid_t bad_object_dir_gfid;
 } br_stub_private_t;
 
+br_stub_fd_t *
+br_stub_fd_new (void);
+
+
+int
+__br_stub_fd_ctx_set (xlator_t *this, fd_t *fd, br_stub_fd_t *br_stub_fd);
+
+br_stub_fd_t *
+__br_stub_fd_ctx_get (xlator_t *this, fd_t *fd);
+
+br_stub_fd_t *
+br_stub_fd_ctx_get (xlator_t *this, fd_t *fd);
+
+int32_t
+br_stub_fd_ctx_set (xlator_t *this, fd_t *fd, br_stub_fd_t *br_stub_fd);
+
 static inline gf_boolean_t
 __br_stub_is_bad_object (br_stub_inode_ctx_t *ctx)
 {
@@ -131,91 +163,6 @@ __br_stub_is_inode_modified (br_stub_inode_ctx_t *ctx)
         return (ctx->need_writeback & I_MODIFIED);
 }
 
-br_stub_fd_t *
-br_stub_fd_new (void)
-{
-        br_stub_fd_t    *br_stub_fd = NULL;
-
-        br_stub_fd = GF_CALLOC (1, sizeof (*br_stub_fd),
-                                gf_br_stub_mt_br_stub_fd_t);
-
-        return br_stub_fd;
-}
-
-int
-__br_stub_fd_ctx_set (xlator_t *this, fd_t *fd, br_stub_fd_t *br_stub_fd)
-{
-        uint64_t    value = 0;
-        int         ret   = -1;
-
-        GF_VALIDATE_OR_GOTO ("bit-rot-stub", this, out);
-        GF_VALIDATE_OR_GOTO (this->name, fd, out);
-        GF_VALIDATE_OR_GOTO (this->name, br_stub_fd, out);
-
-        value = (uint64_t)(long) br_stub_fd;
-
-        ret = __fd_ctx_set (fd, this, value);
-
-out:
-        return ret;
-}
-
-br_stub_fd_t *
-__br_stub_fd_ctx_get (xlator_t *this, fd_t *fd)
-{
-        br_stub_fd_t *br_stub_fd = NULL;
-        uint64_t  value  = 0;
-        int       ret    = -1;
-
-        GF_VALIDATE_OR_GOTO ("bit-rot-stub", this, out);
-        GF_VALIDATE_OR_GOTO (this->name, fd, out);
-
-        ret = __fd_ctx_get (fd, this, &value);
-        if (ret)
-                return NULL;
-
-        br_stub_fd = (br_stub_fd_t *) ((long) value);
-
-out:
-        return br_stub_fd;
-}
-
-br_stub_fd_t *
-br_stub_fd_ctx_get (xlator_t *this, fd_t *fd)
-{
-        br_stub_fd_t *br_stub_fd = NULL;
-
-        GF_VALIDATE_OR_GOTO ("bit-rot-stub", this, out);
-        GF_VALIDATE_OR_GOTO (this->name, fd, out);
-
-        LOCK (&fd->lock);
-        {
-                br_stub_fd = __br_stub_fd_ctx_get (this, fd);
-        }
-        UNLOCK (&fd->lock);
-
-out:
-        return br_stub_fd;
-}
-
-int32_t
-br_stub_fd_ctx_set (xlator_t *this, fd_t *fd, br_stub_fd_t *br_stub_fd)
-{
-        int32_t    ret = -1;
-
-        GF_VALIDATE_OR_GOTO ("bit-rot-stub", this, out);
-        GF_VALIDATE_OR_GOTO (this->name, fd, out);
-        GF_VALIDATE_OR_GOTO (this->name, br_stub_fd, out);
-
-        LOCK (&fd->lock);
-        {
-                ret = __br_stub_fd_ctx_set (this, fd, br_stub_fd);
-        }
-        UNLOCK (&fd->lock);
-
-out:
-        return ret;
-}
 
 static inline int
 br_stub_require_release_call (xlator_t *this, fd_t *fd, br_stub_fd_t **fd_ctx)
@@ -477,4 +424,38 @@ br_stub_add_fd_to_inode (xlator_t *this, fd_t *fd, br_stub_inode_ctx_t *ctx);
 br_sign_state_t
 __br_stub_inode_sign_state (br_stub_inode_ctx_t *ctx, glusterfs_fop_t fop,
                             fd_t *fd);
+
+int
+br_stub_dir_create (xlator_t *this, br_stub_private_t *priv);
+
+int
+br_stub_add (xlator_t *this, uuid_t gfid);
+
+int32_t
+br_stub_create_stub_gfid (xlator_t *this, char *stub_gfid_path, uuid_t gfid);
+
+int
+br_stub_dir_create (xlator_t *this, br_stub_private_t *priv);
+
+call_stub_t *
+__br_stub_dequeue (struct list_head *callstubs);
+
+void
+__br_stub_enqueue (struct list_head *callstubs, call_stub_t *stub);
+
+void
+br_stub_worker_enqueue (xlator_t *this, call_stub_t *stub);
+
+void *
+br_stub_worker (void *data);
+
+int32_t
+br_stub_lookup_wrapper (call_frame_t *frame, xlator_t *this,
+                        loc_t *loc, dict_t *xattr_req);
+
+int32_t
+br_stub_readdir_wrapper (call_frame_t *frame, xlator_t *this,
+                         fd_t *fd, size_t size, off_t off, dict_t *xdata);
+
+
 #endif /* __BIT_ROT_STUB_H__ */