features/index: Index translator implementation

Change-Id: If8a11ecbdd010f64fb4409add5751080f4b59086
BUG: 763820
Signed-off-by: Pranith Kumar K <pranithk@gluster.com>
Reviewed-on: http://review.gluster.com/2722
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Jeff Darcy <jdarcy@redhat.com>
Reviewed-by: Vijay Bellur <vijay@gluster.com>

9 files changed, 1288 insertions, 15 deletions

diff --git a/configure.ac b/configure.ac
index 41777267b..bb16ee61f 100644
--- a/configure.ac
+++ b/configure.ac
@@ -117,6 +117,8 @@ AC_CONFIG_FILES([Makefile
 		xlators/features/mac-compat/src/Makefile
 		xlators/features/quiesce/Makefile
 		xlators/features/quiesce/src/Makefile
+                xlators/features/index/Makefile
+                xlators/features/index/src/Makefile
 		xlators/encryption/Makefile
 		xlators/encryption/rot-13/Makefile
 		xlators/encryption/rot-13/src/Makefile
diff --git a/libglusterfs/src/glusterfs.h b/libglusterfs/src/glusterfs.h
index ef9cdfe85..f961fd593 100644
--- a/libglusterfs/src/glusterfs.h
+++ b/libglusterfs/src/glusterfs.h
@@ -95,6 +95,9 @@
 #define GLUSTERFS_POSIXLK_COUNT "glusterfs.posixlk-count"
 #define QUOTA_SIZE_KEY "trusted.glusterfs.quota.size"
 
+/* Index xlator related */
+#define GF_XATTROP_INDEX_GFID "glusterfs.xattrop_index_gfid"
+
 /* replace-brick and pump related internal xattrs */
 #define RB_PUMP_CMD_START       "glusterfs.pump.start"
 #define RB_PUMP_CMD_PAUSE       "glusterfs.pump.pause"
diff --git a/xlators/features/Makefile.am b/xlators/features/Makefile.am
index 809bbe510..884f82d68 100644
--- a/xlators/features/Makefile.am
+++ b/xlators/features/Makefile.am
@@ -1,3 +1,3 @@
-SUBDIRS = locks trash quota read-only mac-compat quiesce marker#path-converter # filter
+SUBDIRS = locks trash quota read-only mac-compat quiesce marker index#path-converter # filter
 
 CLEANFILES =
diff --git a/xlators/features/index/Makefile.am b/xlators/features/index/Makefile.am
new file mode 100644
index 000000000..a985f42a8
--- /dev/null
+++ b/xlators/features/index/Makefile.am
@@ -0,0 +1,3 @@
+SUBDIRS = src
+
+CLEANFILES =
diff --git a/xlators/features/index/src/Makefile.am b/xlators/features/index/src/Makefile.am
new file mode 100644
index 000000000..5d037c7ec
--- /dev/null
+++ b/xlators/features/index/src/Makefile.am
@@ -0,0 +1,15 @@
+xlator_LTLIBRARIES = index.la
+xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/features
+
+index_la_LDFLAGS = -module -avoidversion
+
+index_la_SOURCES = index.c
+index_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la
+
+noinst_HEADERS = index.h index-mem-types.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)/rpc/xdr/src \
+	-I$(top_srcdir)/rpc/rpc-lib/src -shared -nostartfiles $(GF_CFLAGS)
+
+CLEANFILES =
diff --git a/xlators/features/index/src/index-mem-types.h b/xlators/features/index/src/index-mem-types.h
new file mode 100644
index 000000000..bd10f518f
--- /dev/null
+++ b/xlators/features/index/src/index-mem-types.h
@@ -0,0 +1,32 @@
+/*
+   Copyright (c) 2012-2012 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 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/>.
+*/
+
+
+#ifndef __QUIESCE_MEM_TYPES_H__
+#define __QUIESCE_MEM_TYPES_H__
+
+#include "mem-types.h"
+
+enum gf_index_mem_types_ {
+        gf_index_mt_priv_t = gf_common_mt_end + 1,
+        gf_index_inode_ctx_t = gf_common_mt_end + 2,
+        gf_index_fd_ctx_t = gf_common_mt_end + 3,
+        gf_index_mt_end
+};
+#endif
diff --git a/xlators/features/index/src/index.c b/xlators/features/index/src/index.c
new file mode 100644
index 000000000..34b0ec8e7
--- /dev/null
+++ b/xlators/features/index/src/index.c
@@ -0,0 +1,1129 @@
+/*
+  Copyright (c) 2012-2012 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 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/>.
+*/
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "index.h"
+#include "options.h"
+#include "glusterfs3-xdr.h"
+
+#define XATTROP_SUBDIR "xattrop"
+
+call_stub_t *
+__index_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;
+}
+
+inline static void
+__index_enqueue (struct list_head *callstubs, call_stub_t *stub)
+{
+        list_add_tail (&stub->list, callstubs);
+}
+
+static void
+worker_enqueue (xlator_t *this, call_stub_t *stub)
+{
+        index_priv_t    *priv = NULL;
+
+        priv = this->private;
+        pthread_mutex_lock (&priv->mutex);
+        {
+                __index_enqueue (&priv->callstubs, stub);
+                pthread_cond_signal (&priv->cond);
+        }
+        pthread_mutex_unlock (&priv->mutex);
+}
+
+void *
+index_worker (void *data)
+{
+        index_priv_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->mutex);
+                {
+                        while (list_empty (&priv->callstubs)) {
+                                ret = pthread_cond_wait (&priv->cond,
+                                                         &priv->mutex);
+                        }
+
+                        stub = __index_dequeue (&priv->callstubs);
+                }
+                pthread_mutex_unlock (&priv->mutex);
+
+                if (stub) /* guard against spurious wakeups */
+                        call_resume (stub);
+        }
+
+        return NULL;
+}
+int
+__index_inode_ctx_get (inode_t *inode, xlator_t *this, index_inode_ctx_t **ctx)
+{
+        int               ret = 0;
+        index_inode_ctx_t *ictx = NULL;
+        uint64_t          tmpctx = 0;
+
+        ret = __inode_ctx_get (inode, this, &tmpctx);
+        if (!ret) {
+                ictx = (index_inode_ctx_t*) (long) tmpctx;
+                goto out;
+        }
+        ictx = GF_CALLOC (1, sizeof (*ictx), gf_index_inode_ctx_t);
+        if (!ictx) {
+                ret = -1;
+                goto out;
+        }
+
+        INIT_LIST_HEAD (&ictx->callstubs);
+        ret = __inode_ctx_put (inode, this, (uint64_t)ictx);
+        if (ret) {
+                GF_FREE (ictx);
+                ictx = NULL;
+                goto out;
+        }
+out:
+        if (ictx)
+                *ctx = ictx;
+        return ret;
+}
+
+int
+index_inode_ctx_get (inode_t *inode, xlator_t *this, index_inode_ctx_t **ctx)
+{
+        int               ret = 0;
+
+        LOCK (&inode->lock);
+        {
+                ret = __index_inode_ctx_get (inode, this, ctx);
+        }
+        UNLOCK (&inode->lock);
+
+        return ret;
+}
+
+static void
+make_index_dir_path (char *base, const char *subdir,
+                     char *index_dir, size_t len)
+{
+        snprintf (index_dir, len, "%s/%s", base, subdir);
+}
+
+int
+index_dir_create (xlator_t *this, const char *subdir)
+{
+        int          ret = 0;
+        struct       stat st = {0};
+        char         fullpath[PATH_MAX] = {0};
+        char         path[PATH_MAX] = {0};
+        char         *dir = NULL;
+        index_priv_t *priv = NULL;
+        size_t       len = 0;
+        size_t       pathlen = 0;
+
+        priv = this->private;
+        make_index_dir_path (priv->index_basepath, subdir, fullpath,
+                             sizeof (fullpath));
+        ret = stat (fullpath, &st);
+        if (!ret) {
+                if (!S_ISDIR (st.st_mode))
+                        ret = -2;
+                goto out;
+        }
+
+        pathlen = strlen (fullpath);
+        if ((pathlen > 1) && fullpath[pathlen - 1] == '/')
+                fullpath[pathlen - 1] = '\0';
+        dir = strchr (fullpath, '/');
+        while (dir) {
+                dir = strchr (dir + 1, '/');
+                if (dir)
+                        len = pathlen - strlen (dir);
+                else
+                        len = pathlen;
+                strncpy (path, fullpath, len);
+                path[len] = '\0';
+                ret = mkdir (path, 0600);
+                if (ret && (errno != EEXIST))
+                        goto out;
+        }
+        ret = 0;
+out:
+        if (ret == -1) {
+                gf_log (this->name, GF_LOG_ERROR, "%s/%s: Failed to "
+                        "create (%s)", priv->index_basepath, subdir,
+                        strerror (errno));
+        } else if (ret == -2) {
+                gf_log (this->name, GF_LOG_ERROR, "%s/%s: Failed to create, "
+                        "path exists, not a directory ", priv->index_basepath,
+                        subdir);
+        }
+        return ret;
+}
+
+void
+index_get_index (index_priv_t *priv, uuid_t index)
+{
+        LOCK (&priv->lock);
+        {
+                uuid_copy (index, priv->index);
+        }
+        UNLOCK (&priv->lock);
+}
+
+void
+index_generate_index (index_priv_t *priv, uuid_t index)
+{
+        LOCK (&priv->lock);
+        {
+                //To prevent duplicate generates.
+                //This method fails if number of contending threads is greater
+                //than MAX_LINK count of the fs
+                if (!uuid_compare (priv->index, index))
+                        uuid_generate (priv->index);
+                uuid_copy (index, priv->index);
+        }
+        UNLOCK (&priv->lock);
+}
+
+static void
+make_index_path (char *base, const char *subdir, uuid_t index,
+                 char *index_path, size_t len)
+{
+        make_index_dir_path (base, subdir, index_path, len);
+        snprintf (index_path + strlen (index_path), len - strlen (index_path),
+                  "/%s-%s", subdir, uuid_utoa (index));
+}
+
+static void
+make_gfid_path (char *base, const char *subdir, uuid_t gfid,
+                char *gfid_path, size_t len)
+{
+        make_index_dir_path (base, subdir, gfid_path, len);
+        snprintf (gfid_path + strlen (gfid_path), len - strlen (gfid_path),
+                  "/%s", uuid_utoa (gfid));
+}
+
+static void
+make_file_path (char *base, const char *subdir, const char *filename,
+                char *file_path, size_t len)
+{
+        make_index_dir_path (base, subdir, file_path, len);
+        snprintf (file_path + strlen (file_path), len - strlen (file_path),
+                  "/%s", filename);
+}
+
+static void
+check_delete_stale_index_file (xlator_t *this, char *filename)
+{
+        int             ret = 0;
+        struct stat     st = {0};
+        char            filepath[PATH_MAX] = {0};
+        index_priv_t    *priv = NULL;
+
+        priv = this->private;
+        make_file_path (priv->index_basepath, XATTROP_SUBDIR,
+                        filename, filepath, sizeof (filepath));
+        ret = stat (filepath, &st);
+        if (!ret && st.st_nlink == 1)
+                unlink (filepath);
+}
+
+static int
+index_fill_readdir (fd_t *fd, DIR *dir, off_t off,
+                    size_t size, gf_dirent_t *entries)
+{
+        off_t     in_case = -1;
+        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);
+        }
+
+        while (filled <= size) {
+                in_case = telldir (dir);
+
+                if (in_case == -1) {
+                        gf_log (THIS->name, GF_LOG_ERROR,
+                                "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_log (THIS->name, GF_LOG_WARNING,
+                                        "readdir failed on dir=%p: %s",
+                                        dir, strerror (errno));
+                                goto out;
+                        }
+                        break;
+                }
+
+                if (!strncmp (entry->d_name, XATTROP_SUBDIR"-",
+                              strlen (XATTROP_SUBDIR"-"))) {
+                        check_delete_stale_index_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);
+                        break;
+                }
+
+                this_entry = gf_dirent_for_name (entry->d_name);
+
+                if (!this_entry) {
+                        gf_log (THIS->name, GF_LOG_ERROR,
+                                "could not create gf_dirent for entry %s: (%s)",
+                                entry->d_name, strerror (errno));
+                        goto out;
+                }
+                this_entry->d_off = telldir (dir);
+                this_entry->d_ino = entry->d_ino;
+
+                list_add_tail (&this_entry->list, &entries->list);
+
+                filled += this_size;
+                count ++;
+        }
+
+        if ((!readdir (dir) && (errno == 0)))
+                /* Indicate EOF */
+                errno = ENOENT;
+out:
+        return count;
+}
+
+int
+index_add (xlator_t *this, uuid_t gfid, const char *subdir)
+{
+        int32_t           op_errno = 0;
+        char              gfid_path[PATH_MAX] = {0};
+        char              index_path[PATH_MAX] = {0};
+        int               ret = 0;
+        uuid_t            index = {0};
+        index_priv_t      *priv = NULL;
+        struct stat       st = {0};
+
+        priv = this->private;
+        GF_ASSERT_AND_GOTO_WITH_ERROR (this->name, !uuid_is_null (gfid),
+                                       out, op_errno, EINVAL);
+
+        make_gfid_path (priv->index_basepath, subdir, gfid,
+                        gfid_path, sizeof (gfid_path));
+
+        ret = stat (gfid_path, &st);
+        if (!ret)
+                goto out;
+        index_get_index (priv, index);
+        make_index_path (priv->index_basepath, subdir,
+                         index, index_path, sizeof (index_path));
+        ret = link (index_path, gfid_path);
+        if (!ret || (errno == EEXIST))  {
+                ret = 0;
+                goto out;
+        }
+
+        op_errno = errno;
+        if (op_errno == ENOENT) {
+                ret = index_dir_create (this, subdir);
+                if (ret)
+                        goto out;
+        } else if (op_errno == EMLINK) {
+                index_generate_index (priv, index);
+                make_index_path (priv->index_basepath, subdir,
+                                 index, index_path, sizeof (index_path));
+        } else {
+                goto out;
+        }
+
+        ret = creat (index_path, 0);
+        if (ret && (errno != EEXIST)) {
+                gf_log (this->name, GF_LOG_ERROR, "%s: Not able to "
+                        "create index (%s)", uuid_utoa (gfid),
+                        strerror (errno));
+                goto out;
+        }
+
+
+        ret = link (index_path, gfid_path);
+        if (ret && (errno != EEXIST)) {
+                gf_log (this->name, GF_LOG_ERROR, "%s: Not able to "
+                        "add to index (%s)", uuid_utoa (gfid),
+                        strerror (errno));
+                goto out;
+        }
+
+out:
+        return ret;
+}
+
+int
+index_del (xlator_t *this, uuid_t gfid, const char *subdir)
+{
+        int32_t      op_errno __attribute__((unused)) = 0;
+        index_priv_t *priv = NULL;
+        int          ret = 0;
+        char         gfid_path[PATH_MAX] = {0};
+
+        priv = this->private;
+        GF_ASSERT_AND_GOTO_WITH_ERROR (this->name, !uuid_is_null (gfid),
+                                       out, op_errno, EINVAL);
+        make_gfid_path (priv->index_basepath, subdir, gfid,
+                        gfid_path, sizeof (gfid_path));
+        ret = unlink (gfid_path);
+        if (ret && (errno != ENOENT)) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "%s: failed to delete from index (%s)",
+                        gfid_path, strerror (errno));
+                ret = -errno;
+                goto out;
+        }
+        ret = 0;
+out:
+        return ret;
+}
+
+void
+_xattrop_index_action (xlator_t *this, inode_t *inode,  dict_t *xattr)
+{
+        data_pair_t       *trav = NULL;
+        gf_boolean_t      zero_xattr = _gf_true;
+        index_inode_ctx_t *ctx = NULL;
+        int               ret = 0;
+
+        trav = xattr->members_list;
+        while (trav && inode) {
+                if (mem_0filled ((const char*)trav->value->data,
+                                 trav->value->len)) {
+                        zero_xattr = _gf_false;
+                        break;
+                }
+                trav = trav->next;
+        }
+        ret = index_inode_ctx_get (inode, this, &ctx);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Not able to %s %s -> index",
+                        zero_xattr?"add":"del", uuid_utoa (inode->gfid));
+                goto out;
+        }
+        if (zero_xattr) {
+                if (ctx->state == NOTIN)
+                        goto out;
+                ret = index_del (this, inode->gfid, XATTROP_SUBDIR);
+                if (!ret)
+                        ctx->state = NOTIN;
+        } else {
+                if (ctx->state == IN)
+                        goto out;
+                ret = index_add (this, inode->gfid, XATTROP_SUBDIR);
+                if (!ret)
+                        ctx->state = IN;
+        }
+out:
+        return;
+}
+
+void
+fop_xattrop_index_action (xlator_t *this, inode_t *inode, dict_t *xattr)
+{
+        _xattrop_index_action (this, inode, xattr);
+}
+
+void
+fop_fxattrop_index_action (xlator_t *this, inode_t *inode, dict_t *xattr)
+{
+        _xattrop_index_action (this, inode, xattr);
+}
+
+inline gf_boolean_t
+index_xattrop_track (loc_t *loc, gf_xattrop_flags_t flags, dict_t *dict)
+{
+        return (flags == GF_XATTROP_ADD_ARRAY);
+}
+
+inline gf_boolean_t
+index_fxattrop_track (fd_t *fd, gf_xattrop_flags_t flags, dict_t *dict)
+{
+        return (flags == GF_XATTROP_ADD_ARRAY);
+}
+
+int
+__index_fd_ctx_get (fd_t *fd, xlator_t *this, index_fd_ctx_t **ctx)
+{
+        int               ret = 0;
+        index_fd_ctx_t    *fctx = NULL;
+        uint64_t          tmpctx = 0;
+        char              index_dir[PATH_MAX] = {0};
+        index_priv_t      *priv = NULL;
+
+        priv = this->private;
+        if (uuid_compare (fd->inode->gfid, priv->xattrop_vgfid)) {
+                ret = -EINVAL;
+                goto out;
+        }
+
+        ret = __fd_ctx_get (fd, this, &tmpctx);
+        if (!ret) {
+                fctx = (index_fd_ctx_t*) (long) tmpctx;
+                goto out;
+        }
+
+        fctx = GF_CALLOC (1, sizeof (*fctx), gf_index_fd_ctx_t);
+        if (!fctx) {
+                ret = -ENOMEM;
+                goto out;
+        }
+
+        make_index_dir_path (priv->index_basepath, XATTROP_SUBDIR,
+                             index_dir, sizeof (index_dir));
+        fctx->dir = opendir (index_dir);
+        if (!fctx->dir) {
+                ret = -errno;
+                GF_FREE (fctx);
+                fctx = NULL;
+                goto out;
+        }
+
+        ret = __fd_ctx_set (fd, this, (uint64_t)(long)fctx);
+        if (ret) {
+                GF_FREE (fctx);
+                fctx = NULL;
+                ret = -EINVAL;
+                goto out;
+        }
+out:
+        if (fctx)
+                *ctx = fctx;
+        return ret;
+}
+
+int
+index_fd_ctx_get (fd_t *fd, xlator_t *this, index_fd_ctx_t **ctx)
+{
+        int     ret = 0;
+        LOCK (&fd->lock);
+        {
+                ret = __index_fd_ctx_get (fd, this, ctx);
+        }
+        UNLOCK (&fd->lock);
+        return ret;
+}
+
+//new - Not NULL means start a fop
+//new - NULL means done processing the fop
+void
+index_queue_process (xlator_t *this, inode_t *inode, call_stub_t *new)
+{
+        call_stub_t       *stub = NULL;
+        index_inode_ctx_t *ctx = NULL;
+        int               ret = 0;
+        call_frame_t      *frame = NULL;
+
+        LOCK (&inode->lock);
+        {
+                ret = __index_inode_ctx_get (inode, this, &ctx);
+                if (ret)
+                        goto unlock;
+
+                if (new) {
+                        __index_enqueue (&ctx->callstubs, new);
+                        new = NULL;
+                } else {
+                        ctx->processing = _gf_false;
+                }
+
+                if (!ctx->processing) {
+                        stub = __index_dequeue (&ctx->callstubs);
+                        if (stub)
+                                ctx->processing = _gf_true;
+                        else
+                                ctx->processing = _gf_false;
+                }
+        }
+unlock:
+        UNLOCK (&inode->lock);
+
+        if (ret && new) {
+                frame = new->frame;
+                if (new->fop == GF_FOP_XATTROP) {
+                        INDEX_STACK_UNWIND (xattrop, frame, -1, ENOMEM,
+                                            NULL);
+                } else if (new->fop == GF_FOP_FXATTROP) {
+                        INDEX_STACK_UNWIND (fxattrop, frame, -1, ENOMEM,
+                                            NULL);
+                }
+                call_stub_destroy (new);
+        } else if (stub) {
+                call_resume (stub);
+        }
+        return;
+}
+
+int32_t
+index_xattrop_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                   int32_t op_ret, int32_t op_errno, dict_t *xattr)
+{
+        inode_t *inode = NULL;
+
+        inode = inode_ref (frame->local);
+        if (op_ret < 0)
+                goto out;
+        fop_xattrop_index_action (this, frame->local, xattr);
+out:
+        INDEX_STACK_UNWIND (xattrop, frame, op_ret, op_errno, xattr);
+        index_queue_process (this, inode, NULL);
+        inode_unref (inode);
+
+        return 0;
+}
+
+int32_t
+index_fxattrop_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                    int32_t op_ret, int32_t op_errno, dict_t *xattr)
+{
+        inode_t *inode = NULL;
+
+        inode = inode_ref (frame->local);
+        if (op_ret < 0)
+                goto out;
+
+        fop_fxattrop_index_action (this, frame->local, xattr);
+out:
+        INDEX_STACK_UNWIND (fxattrop, frame, op_ret, op_errno, xattr);
+        index_queue_process (this, inode, NULL);
+        inode_unref (inode);
+
+        return 0;
+}
+
+int
+index_xattrop_wrapper (call_frame_t *frame, xlator_t *this, loc_t *loc,
+                       gf_xattrop_flags_t optype, dict_t *xattr)
+{
+        STACK_WIND (frame, index_xattrop_cbk, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->xattrop, loc, optype, xattr);
+        return 0;
+}
+
+int
+index_fxattrop_wrapper (call_frame_t *frame, xlator_t *this, fd_t *fd,
+                        gf_xattrop_flags_t optype, dict_t *xattr)
+{
+        STACK_WIND (frame, index_fxattrop_cbk, FIRST_CHILD (this),
+                    FIRST_CHILD (this)->fops->fxattrop, fd, optype, xattr);
+        return 0;
+}
+
+int32_t
+index_xattrop (call_frame_t *frame, xlator_t *this, loc_t *loc,
+	       gf_xattrop_flags_t flags, dict_t *dict)
+{
+        call_stub_t     *stub = NULL;
+
+        if (!index_xattrop_track (loc, flags, dict))
+                goto out;
+
+        frame->local = inode_ref (loc->inode);
+        stub = fop_xattrop_stub (frame, index_xattrop_wrapper,
+                                 loc, flags, dict);
+        if (!stub) {
+                INDEX_STACK_UNWIND (xattrop, frame, -1, ENOMEM, NULL);
+                return 0;
+        }
+
+        index_queue_process (this, loc->inode, stub);
+        return 0;
+out:
+        STACK_WIND (frame, default_xattrop_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->xattrop, loc, flags, dict);
+        return 0;
+}
+
+int32_t
+index_fxattrop (call_frame_t *frame, xlator_t *this, fd_t *fd,
+		gf_xattrop_flags_t flags, dict_t *dict)
+{
+        call_stub_t    *stub = NULL;
+
+        if (!index_fxattrop_track (fd, flags, dict))
+                goto out;
+
+        frame->local = inode_ref (fd->inode);
+        stub = fop_fxattrop_stub (frame, index_fxattrop_wrapper,
+                                  fd, flags, dict);
+        if (!stub) {
+                INDEX_STACK_UNWIND (fxattrop, frame, -1, ENOMEM, NULL);
+                return 0;
+        }
+
+        index_queue_process (this, fd->inode, stub);
+        return 0;
+out:
+        STACK_WIND (frame, default_fxattrop_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->fxattrop, fd, flags, dict);
+        return 0;
+}
+
+int32_t
+index_getxattr_wrapper (call_frame_t *frame, xlator_t *this,
+                        loc_t *loc, const char *name)
+{
+        index_priv_t    *priv = NULL;
+        dict_t          *xattr = NULL;
+        int             ret = 0;
+
+        priv = this->private;
+
+        xattr = dict_new ();
+        if (!xattr) {
+                ret = -ENOMEM;
+                goto done;
+        }
+
+        ret = dict_set_static_bin (xattr, (char*)name, priv->xattrop_vgfid,
+                                   sizeof (priv->xattrop_vgfid));
+        if (ret) {
+                ret = -ENOMEM;
+                gf_log (THIS->name, GF_LOG_ERROR, "xattrop index "
+                        "gfid set failed");
+                goto done;
+        }
+done:
+        if (ret)
+                STACK_UNWIND_STRICT (getxattr, frame, -1, -ret, xattr);
+        else
+                STACK_UNWIND_STRICT (getxattr, frame, 0, 0, xattr);
+
+        if (xattr)
+                dict_unref (xattr);
+
+        return 0;
+}
+
+int32_t
+index_lookup_wrapper (call_frame_t *frame, xlator_t *this,
+                      loc_t *loc, dict_t *xattr_req)
+{
+        index_priv_t    *priv = NULL;
+        struct stat     lstatbuf = {0};
+        int             ret = 0;
+        int32_t         op_errno = EINVAL;
+        int32_t         op_ret = -1;
+        char            path[PATH_MAX] = {0};
+        struct iatt     stbuf        = {0, };
+        struct iatt     postparent = {0,};
+        dict_t          *xattr = NULL;
+        gf_boolean_t    is_dir = _gf_false;
+
+        priv = this->private;
+
+        VALIDATE_OR_GOTO (loc, done);
+        if (!uuid_compare (loc->gfid, priv->xattrop_vgfid)) {
+                make_index_dir_path (priv->index_basepath, XATTROP_SUBDIR,
+                                     path, sizeof (path));
+                is_dir = _gf_true;
+        } else if (!uuid_compare (loc->pargfid, priv->xattrop_vgfid)) {
+                make_file_path (priv->index_basepath, XATTROP_SUBDIR,
+                                loc->name, path, sizeof (path));
+        }
+
+        ret = lstat (path, &lstatbuf);
+        if (ret) {
+                gf_log (this->name, GF_LOG_DEBUG, "Stat failed on index dir "
+                        "(%s)", strerror (errno));
+                op_errno = errno;
+                goto done;
+        } else if (!S_ISDIR (lstatbuf.st_mode) && is_dir) {
+                gf_log (this->name, GF_LOG_DEBUG, "Stat failed on index dir, "
+                        "not a directory");
+                op_errno = ENOENT;
+                goto done;
+        }
+        xattr = dict_new ();
+        if (!xattr) {
+                op_errno = ENOMEM;
+                goto done;
+        }
+
+        iatt_from_stat (&stbuf, &lstatbuf);
+        if (is_dir)
+                uuid_copy (stbuf.ia_gfid, priv->xattrop_vgfid);
+        else
+                uuid_generate (stbuf.ia_gfid);
+        stbuf.ia_ino = -1;
+        op_ret = 0;
+done:
+        STACK_UNWIND_STRICT (lookup, frame, op_ret, op_errno,
+                             loc->inode, &stbuf, xattr, &postparent);
+        if (xattr)
+                dict_unref (xattr);
+        return 0;
+}
+
+int32_t
+index_readdir_wrapper (call_frame_t *frame, xlator_t *this,
+                       fd_t *fd, size_t size, off_t off)
+{
+        index_fd_ctx_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);
+
+        ret = index_fd_ctx_get (fd, this, &fctx);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "pfd is NULL, fd=%p", fd);
+                op_errno = -ret;
+                goto done;
+        }
+
+        dir = fctx->dir;
+
+        if (!dir) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "dir is NULL for fd=%p", fd);
+                op_errno = EINVAL;
+                goto done;
+        }
+
+        count = index_fill_readdir (fd, 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);
+        gf_dirent_free (&entries);
+        return 0;
+}
+
+int
+index_unlink_wrapper (call_frame_t *frame, xlator_t *this, loc_t *loc)
+{
+        index_priv_t    *priv = NULL;
+        int32_t         op_ret = 0;
+        int32_t         op_errno = 0;
+        int             ret = 0;
+        struct  iatt    preparent = {0};
+        struct  iatt    postparent = {0};
+        char            index_dir[PATH_MAX] = {0};
+        struct  stat    lstatbuf = {0};
+        uuid_t          gfid = {0};
+
+        priv = this->private;
+        make_index_dir_path (priv->index_basepath, XATTROP_SUBDIR,
+                             index_dir, sizeof (index_dir));
+        ret = lstat (index_dir, &lstatbuf);
+        if (ret < 0) {
+                op_ret = -1;
+                op_errno = errno;
+                goto done;
+        }
+
+        iatt_from_stat (&preparent, &lstatbuf);
+        uuid_copy (preparent.ia_gfid, priv->xattrop_vgfid);
+        preparent.ia_ino = -1;
+        uuid_parse (loc->name, gfid);
+        ret = index_del (this, gfid, XATTROP_SUBDIR);
+        if (ret < 0) {
+                op_ret = -1;
+                op_errno = -ret;
+                goto done;
+        }
+        memset (&lstatbuf, 0, sizeof (lstatbuf));
+        ret = lstat (index_dir, &lstatbuf);
+        if (ret < 0) {
+                op_ret = -1;
+                op_errno = errno;
+                goto done;
+        }
+        iatt_from_stat (&postparent, &lstatbuf);
+        uuid_copy (postparent.ia_gfid, priv->xattrop_vgfid);
+        postparent.ia_ino = -1;
+done:
+        INDEX_STACK_UNWIND (unlink, frame, op_ret, op_errno, &preparent,
+                            &postparent);
+        return 0;
+}
+
+int32_t
+index_getxattr (call_frame_t *frame, xlator_t *this,
+                loc_t *loc, const char *name)
+{
+        call_stub_t     *stub = NULL;
+
+        if (!name || strcmp (GF_XATTROP_INDEX_GFID, name))
+                goto out;
+
+        stub = fop_getxattr_stub (frame, index_getxattr_wrapper, loc, name);
+        if (!stub) {
+                STACK_UNWIND_STRICT (getxattr, frame, -1, ENOMEM, NULL);
+                return 0;
+        }
+        worker_enqueue (this, stub);
+        return 0;
+out:
+        STACK_WIND (frame, default_getxattr_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->getxattr, loc, name);
+        return 0;
+}
+
+int32_t
+index_lookup (call_frame_t *frame, xlator_t *this,
+              loc_t *loc, dict_t *xattr_req)
+{
+        call_stub_t     *stub = NULL;
+        index_priv_t    *priv = NULL;
+
+        priv = this->private;
+
+        if (uuid_compare (loc->gfid, priv->xattrop_vgfid) &&
+            uuid_compare (loc->pargfid, priv->xattrop_vgfid))
+                goto normal;
+
+        stub = fop_lookup_stub (frame, index_lookup_wrapper, loc, xattr_req);
+        if (!stub) {
+                STACK_UNWIND_STRICT (lookup, frame, -1, ENOMEM, loc->inode,
+                                     NULL, NULL, NULL);
+                return 0;
+        }
+        worker_enqueue (this, stub);
+        return 0;
+normal:
+        STACK_WIND (frame, default_lookup_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->lookup, loc, xattr_req);
+
+        return 0;
+}
+
+int32_t
+index_readdir (call_frame_t *frame, xlator_t *this,
+               fd_t *fd, size_t size, off_t off)
+{
+        call_stub_t     *stub = NULL;
+        index_priv_t    *priv = NULL;
+
+        priv = this->private;
+        if (uuid_compare (fd->inode->gfid, priv->xattrop_vgfid))
+                goto out;
+        stub = fop_readdir_stub (frame, index_readdir_wrapper, fd, size, off);
+        if (!stub) {
+                STACK_UNWIND_STRICT (readdir, frame, -1, ENOMEM, NULL);
+                return 0;
+        }
+        worker_enqueue (this, stub);
+        return 0;
+out:
+        STACK_WIND (frame, default_readdir_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->readdir, fd, size, off);
+        return 0;
+}
+
+int
+index_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc)
+{
+        call_stub_t     *stub = NULL;
+        index_priv_t    *priv = NULL;
+
+        priv = this->private;
+        if (uuid_compare (loc->pargfid, priv->xattrop_vgfid))
+                goto out;
+
+        stub = fop_unlink_stub (frame, index_unlink_wrapper, loc);
+        if (!stub) {
+                STACK_UNWIND_STRICT (unlink, frame, -1, ENOMEM, NULL, NULL);
+                return 0;
+        }
+        worker_enqueue (this, stub);
+        return 0;
+out:
+        STACK_WIND (frame, default_unlink_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->unlink, loc);
+        return 0;
+}
+
+int32_t
+mem_acct_init (xlator_t *this)
+{
+        int     ret = -1;
+
+        ret = xlator_mem_acct_init (this, gf_index_mt_end + 1);
+
+        return ret;
+}
+
+int
+init (xlator_t *this)
+{
+        int ret = -1;
+        index_priv_t *priv = NULL;
+        pthread_t thread;
+
+	if (!this->children || this->children->next) {
+		gf_log (this->name, GF_LOG_ERROR,
+			"'index' not configured with exactly one child");
+                goto out;
+	}
+
+	if (!this->parents) {
+		gf_log (this->name, GF_LOG_WARNING,
+			"dangling volume. check volfile ");
+	}
+
+        priv = GF_CALLOC (1, sizeof (*priv), gf_index_mt_priv_t);
+        if (!priv)
+                goto out;
+
+        LOCK_INIT (&priv->lock);
+        if ((ret = pthread_cond_init(&priv->cond, NULL)) != 0) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "pthread_cond_init failed (%d)", ret);
+                goto out;
+        }
+
+        if ((ret = pthread_mutex_init(&priv->mutex, NULL)) != 0) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "pthread_mutex_init failed (%d)", ret);
+                goto out;
+        }
+
+        pthread_attr_init (&priv->w_attr);
+        ret = pthread_attr_setstacksize (&priv->w_attr,
+                                         INDEX_THREAD_STACK_SIZE);
+        if (ret == EINVAL) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "Using default thread stack size");
+        }
+        GF_OPTION_INIT ("index-base", priv->index_basepath, path, out);
+        uuid_generate (priv->index);
+        uuid_generate (priv->xattrop_vgfid);
+        INIT_LIST_HEAD (&priv->callstubs);
+
+        this->private = priv;
+        ret = pthread_create (&thread, &priv->w_attr, index_worker, this);
+        if (ret) {
+                gf_log (this->name, GF_LOG_WARNING, "Failed to create "
+                        "worker thread, aborting");
+                goto out;
+        }
+        ret = 0;
+out:
+        return ret;
+}
+
+void
+fini (xlator_t *this)
+{
+        index_priv_t *priv = NULL;
+
+        priv = this->private;
+        if (!priv)
+                goto out;
+        this->private = NULL;
+        LOCK_DESTROY (&priv->lock);
+        GF_FREE (priv);
+out:
+        return;
+}
+
+int
+index_forget (xlator_t *this, inode_t *inode)
+{
+        uint64_t tmp_cache = 0;
+        if (!inode_ctx_del (inode, this, &tmp_cache))
+                GF_FREE ((index_inode_ctx_t*) (long)tmp_cache);
+
+        return 0;
+}
+
+int
+notify (xlator_t *this, int event, void *data, ...)
+{
+        int     ret = 0;
+        ret = default_notify (this, event, data);
+        return ret;
+}
+
+struct xlator_fops fops = {
+	.xattrop     = index_xattrop,
+	.fxattrop    = index_fxattrop,
+
+        //interface functions follow
+        .getxattr    = index_getxattr,
+        .lookup      = index_lookup,
+        .readdir     = index_readdir,
+        .unlink      = index_unlink
+};
+
+struct xlator_dumpops dumpops = {
+};
+
+struct xlator_cbks cbks = {
+        .forget = index_forget,
+};
+
+struct volume_options options[] = {
+        { .key  = {"index-base" },
+          .type = GF_OPTION_TYPE_PATH,
+          .description = "path where the index files need to be stored",
+        },
+        { .key  = {NULL} },
+};
diff --git a/xlators/features/index/src/index.h b/xlators/features/index/src/index.h
new file mode 100644
index 000000000..bbf258f30
--- /dev/null
+++ b/xlators/features/index/src/index.h
@@ -0,0 +1,70 @@
+/*
+   Copyright (c) 2010-2011 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 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/>.
+*/
+
+
+#ifndef __INDEX_H__
+#define __INDEX_H__
+
+#include "xlator.h"
+#include "call-stub.h"
+#include "defaults.h"
+#include "byte-order.h"
+#include "common-utils.h"
+#include "index-mem-types.h"
+
+#define INDEX_THREAD_STACK_SIZE   ((size_t)(1024*1024))
+
+typedef enum {
+        UNKNOWN,
+        IN,
+        NOTIN
+} index_state_t;
+
+typedef struct index_inode_ctx {
+        gf_boolean_t processing;
+        struct list_head callstubs;
+        index_state_t state;
+} index_inode_ctx_t;
+
+typedef struct index_fd_ctx {
+        DIR *dir;
+} index_fd_ctx_t;
+
+typedef struct index_priv {
+        char *index_basepath;
+        uuid_t index;
+        gf_lock_t lock;
+        uuid_t xattrop_vgfid;//virtual gfid of the xattrop index dir
+        struct list_head callstubs;
+        pthread_mutex_t mutex;
+        pthread_cond_t  cond;
+        pthread_attr_t  w_attr;
+} index_priv_t;
+
+#define INDEX_STACK_UNWIND(fop, frame, params ...)      \
+do {                                                    \
+        if (frame) {                                    \
+                inode_t *_inode = frame->local;         \
+                frame->local = NULL;                    \
+                inode_unref (_inode);                   \
+        }                                               \
+        STACK_UNWIND_STRICT (fop, frame, params);       \
+} while (0)
+
+#endif
diff --git a/xlators/mgmt/glusterd/src/glusterd-volgen.c b/xlators/mgmt/glusterd/src/glusterd-volgen.c
index ac5378b1a..f3d294341 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volgen.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volgen.c
@@ -1539,20 +1539,21 @@ static int
 server_graph_builder (volgen_graph_t *graph, glusterd_volinfo_t *volinfo,
                       dict_t *set_dict, void *param)
 {
-        char     *volname               = NULL;
-        char     *path                  = NULL;
-        int       pump                  = 0;
-        xlator_t *xl                    = NULL;
-        xlator_t *txl                   = NULL;
-        xlator_t *rbxl                  = NULL;
-        char      transt[16]            = {0,};
-        char     *ptranst               = NULL;
-        char      volume_id[64]         = {0,};
-        char      tstamp_file[PATH_MAX] = {0,};
-        int       ret                   = 0;
-        char     *xlator                = NULL;
-        char     *loglevel              = NULL;
-        char      key[1024]             = {0};
+        char     *volname                 = NULL;
+        char     *path                    = NULL;
+        int       pump                    = 0;
+        xlator_t *xl                      = NULL;
+        xlator_t *txl                     = NULL;
+        xlator_t *rbxl                    = NULL;
+        char      transt[16]              = {0,};
+        char     *ptranst                 = NULL;
+        char      volume_id[64]           = {0,};
+        char      tstamp_file[PATH_MAX]   = {0,};
+        int       ret                     = 0;
+        char     *xlator                  = NULL;
+        char     *loglevel                = NULL;
+        char     index_basepath[PATH_MAX] = {0};
+        char     key[1024]                = {0};
 
         path = param;
         volname = volinfo->volname;
@@ -1611,6 +1612,24 @@ server_graph_builder (volgen_graph_t *graph, glusterd_volinfo_t *volinfo,
         if (ret)
                 return -1;
 
+        if (glusterd_is_volume_replicate (volinfo)) {
+                xl = volgen_graph_add (graph, "features/index", volname);
+                if (!xl)
+                        return -1;
+
+                snprintf (index_basepath, sizeof (index_basepath), "%s/%s",
+                          path, ".glusterfs/indices");
+                ret = xlator_set_option (xl, "index-base", index_basepath);
+                if (ret)
+                        return -1;
+
+                ret = check_and_add_debug_xl (graph, set_dict, volname,
+                                              "index");
+                if (ret)
+                        return -1;
+
+        }
+
         ret = dict_get_int32 (volinfo->dict, "enable-pump", &pump);
         if (ret == -ENOENT)
                 ret = pump = 0;