features/marker: quota friendly changes

* handles renames on dht linkfiles correctly

* nameless lookup friendly changes. uses gfid-to-path conversion
  functionality from storage/posix to build ancestry till root.

* log message cleanup.

* build inode contexts in readdirp

* Accounting still not correct with hardlinks.

Credits:
========
Vijay Bellur <vbellur@redhat.com>
Raghavendra Bhat <rabhat@redhat.com>

Change-Id: I415b6fbbc9691f5a38d9fd3c5d083a61e578bb81
BUG: 969461
Signed-off-by: Raghavendra G <rgowdapp@redhat.com>
Reviewed-on: http://review.gluster.org/5953
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Anand Avati <avati@redhat.com>

5 files changed, 386 insertions, 89 deletions

diff --git a/xlators/features/marker/src/marker-quota-helper.c b/xlators/features/marker/src/marker-quota-helper.c
index af5fed13271..ec0d83316c7 100644
--- a/xlators/features/marker/src/marker-quota-helper.c
+++ b/xlators/features/marker/src/marker-quota-helper.c
@@ -154,15 +154,17 @@ out:
 
 
 inode_contribution_t *
-__mq_add_new_contribution_node (xlator_t *this, quota_inode_ctx_t *ctx, loc_t *loc)
+__mq_add_new_contribution_node (xlator_t *this, quota_inode_ctx_t *ctx,
+                                loc_t *loc)
 {
-        int32_t ret = 0;
+        int32_t               ret          = 0;
         inode_contribution_t *contribution = NULL;
 
         if (!loc->parent) {
                 if (!uuid_is_null (loc->pargfid))
                         loc->parent = inode_find (loc->inode->table,
                                                   loc->pargfid);
+
                 if (!loc->parent)
                         loc->parent = inode_parent (loc->inode, loc->pargfid,
                                                     loc->name);
@@ -170,9 +172,10 @@ __mq_add_new_contribution_node (xlator_t *this, quota_inode_ctx_t *ctx, loc_t *l
                         goto out;
         }
 
-        list_for_each_entry (contribution, &ctx->contribution_head, contri_list) {
+        list_for_each_entry (contribution, &ctx->contribution_head,
+                             contri_list) {
                 if (loc->parent &&
-                     uuid_compare (contribution->gfid, loc->parent->gfid) == 0) {
+                    uuid_compare (contribution->gfid, loc->parent->gfid) == 0) {
                         goto out;
                 }
         }
@@ -196,14 +199,16 @@ out:
 
 
 inode_contribution_t *
-mq_add_new_contribution_node (xlator_t *this, quota_inode_ctx_t *ctx, loc_t *loc)
+mq_add_new_contribution_node (xlator_t *this, quota_inode_ctx_t *ctx,
+                              loc_t *loc)
 {
         inode_contribution_t *contribution = NULL;
 
         if ((ctx == NULL) || (loc == NULL))
                 return NULL;
 
-        if (strcmp (loc->path, "/") == 0)
+        if (((loc->path) && (strcmp (loc->path, "/") == 0))
+            || (!loc->path && uuid_is_null (loc->pargfid)))
                 return NULL;
 
         LOCK (&ctx->lock);
@@ -226,12 +231,16 @@ mq_dict_set_contribution (xlator_t *this, dict_t *dict,
         GF_VALIDATE_OR_GOTO ("marker", this, out);
         GF_VALIDATE_OR_GOTO ("marker", dict, out);
         GF_VALIDATE_OR_GOTO ("marker", loc, out);
-        GF_VALIDATE_OR_GOTO ("marker", loc->parent, out);
 
-        GET_CONTRI_KEY (contri_key, loc->parent->gfid, ret);
-        if (ret < 0) {
-                ret = -1;
-                goto out;
+        if (loc->parent) {
+                GET_CONTRI_KEY (contri_key, loc->parent->gfid, ret);
+                if (ret < 0) {
+                        ret = -1;
+                        goto out;
+                }
+        } else {
+                /* nameless lookup, fetch contributions to all parents */
+                GET_CONTRI_KEY (contri_key, NULL, ret);
         }
 
         ret = dict_set_int64 (dict, contri_key, 0);
diff --git a/xlators/features/marker/src/marker-quota.c b/xlators/features/marker/src/marker-quota.c
index 6f9af6e1335..d972d7f85b0 100644
--- a/xlators/features/marker/src/marker-quota.c
+++ b/xlators/features/marker/src/marker-quota.c
@@ -30,7 +30,8 @@ mq_loc_copy (loc_t *dst, loc_t *src)
         GF_VALIDATE_OR_GOTO ("marker", src, out);
 
         if (src->inode == NULL ||
-            src->path == NULL) {
+            ((src->parent == NULL) && (uuid_is_null (src->pargfid))
+             && !__is_root_gfid (src->inode->gfid))) {
                 gf_log ("marker", GF_LOG_WARNING,
                         "src loc is not valid");
                 goto out;
@@ -1038,7 +1039,11 @@ mq_create_xattr (xlator_t *this, call_frame_t *frame)
                         goto free_size;
         }
 
-        if (strcmp (local->loc.path, "/") != 0) {
+        if ((local->loc.path && strcmp (local->loc.path, "/") != 0)
+            || (local->loc.inode && !uuid_is_null (local->loc.inode->gfid) &&
+                !__is_root_gfid (local->loc.inode->gfid))
+            || (!uuid_is_null (local->loc.gfid)
+                && !__is_root_gfid (local->loc.gfid))) {
                 contri = mq_add_new_contribution_node (this, ctx, &local->loc);
                 if (contri == NULL)
                         goto err;
@@ -1107,7 +1112,12 @@ mq_check_n_set_inode_xattr (call_frame_t *frame, void *cookie,
                 goto create_xattr;
 
         //check contribution xattr if not root
-        if (strcmp (local->loc.path, "/") != 0) {
+        if ((local->loc.path && strcmp (local->loc.path, "/") != 0)
+            || (!uuid_is_null (local->loc.gfid)
+                && !__is_root_gfid (local->loc.gfid))
+            || (local->loc.inode
+                && !uuid_is_null (local->loc.inode->gfid)
+                && !__is_root_gfid (local->loc.inode->gfid))) {
                 GET_CONTRI_KEY (contri_key, local->loc.parent->gfid, ret);
                 if (ret < 0)
                         goto out;
@@ -1234,6 +1244,7 @@ mq_get_parent_inode_local (xlator_t *this, quota_local_t *local)
 {
         int32_t            ret = -1;
         quota_inode_ctx_t *ctx = NULL;
+        inode_contribution_t *contribution = NULL;
 
         GF_VALIDATE_OR_GOTO ("marker", this, out);
         GF_VALIDATE_OR_GOTO ("marker", local, out);
@@ -1263,7 +1274,7 @@ mq_get_parent_inode_local (xlator_t *this, quota_local_t *local)
         ret = mq_inode_ctx_get (local->loc.inode, this, &ctx);
         if (ret < 0) {
                 gf_log_callingfn (this->name, GF_LOG_WARNING,
-                        "inode ctx get failed");
+                                  "inode ctx get failed");
                 goto out;
         }
 
@@ -1277,7 +1288,31 @@ mq_get_parent_inode_local (xlator_t *this, quota_local_t *local)
                 goto out;
         }
 
-        local->contri = (inode_contribution_t *) ctx->contribution_head.next;
+        /* Earlier we used to get the next entry in the list maintained
+           by the context. In a good situation it works. i.e the next
+           parent in the directory hierarchy for this path is obtained.
+
+           But consider the below situation:
+           mount-point: /mnt/point
+           quota enabled directories within mount point: /a, /b, /c
+
+           Now when some file (file1) in the directory /c is written some data,
+           then to update the directories, marker has to get the contribution
+           object for the parent inode, i.e /c.
+           Beefore, it was being done by
+           local->contri = (inode_contribution_t *) ctx->contribution_head.next;
+           It works in the normal situations. But suppose /c is moved to /b.
+           Now /b's contribution object is added to the end of the list of
+           parents that the file file1 within /b/c is maintaining. Now if
+           the file /b/c/file1 is copied to /b/c/new, to update the parent in
+           the order c, b and / we cannot go to the next element in the list,
+           as in this case the next contribution object would be / and /b's
+           contribution will be at the end of the list. So get the proper
+           parent's contribution, by searching the entire list.
+        */
+        contribution = mq_get_contribution_node (local->loc.parent, ctx);
+        GF_ASSERT (contribution != NULL);
+        local->contri = contribution;
 
         ret = 0;
 out:
@@ -1521,9 +1556,10 @@ mq_update_parent_size (call_frame_t *frame,
         }
         UNLOCK (&local->contri->lock);
 
-        gf_log (this->name, GF_LOG_DEBUG, "%s %"PRId64 "%"PRId64,
-                local->loc.path, local->ctx->size,
-                local->contri->contribution);
+        gf_log_callingfn (this->name, GF_LOG_DEBUG, "path: %s size: %"PRId64
+                          " contribution:%"PRId64,
+                          local->loc.path, local->ctx->size,
+                          local->contri->contribution);
 
         if (dict == NULL) {
                 op_errno = EINVAL;
@@ -1730,7 +1766,8 @@ mq_fetch_child_size_and_contri (call_frame_t *frame, void *cookie,
         VALIDATE_OR_GOTO (local->ctx, err);
         VALIDATE_OR_GOTO (local->contri, err);
 
-        gf_log (this->name, GF_LOG_DEBUG, "%s marked dirty", local->parent_loc.path);
+        gf_log (this->name, GF_LOG_DEBUG, "%s marked dirty",
+                local->parent_loc.path);
 
         //update parent ctx
         ret = mq_inode_ctx_get (local->parent_loc.inode, this, &ctx);
@@ -1901,15 +1938,18 @@ fr_destroy:
         return -1;
 }
 
-
 int
-mq_start_quota_txn (xlator_t *this, loc_t *loc,
-                    quota_inode_ctx_t *ctx,
-                    inode_contribution_t *contri)
+mq_prepare_txn_frame (xlator_t *this, loc_t *loc,
+                      quota_inode_ctx_t *ctx,
+                      inode_contribution_t *contri,
+                      call_frame_t **new_frame)
 {
-        int32_t        ret      = -1;
-        call_frame_t  *frame    = NULL;
-        quota_local_t *local    = NULL;
+        call_frame_t    *frame = NULL;
+        int              ret   = -1;
+        quota_local_t   *local = NULL;
+
+        if (!this || !loc || !new_frame)
+                goto err;
 
         frame = create_frame (this, this->ctx->pool);
         if (frame == NULL)
@@ -1935,14 +1975,36 @@ mq_start_quota_txn (xlator_t *this, loc_t *loc,
         local->ctx = ctx;
         local->contri = contri;
 
+        ret = 0;
+        *new_frame = frame;
+
+        return ret;
+
+fr_destroy:
+        QUOTA_STACK_DESTROY (frame, this);
+err:
+        return ret;
+}
+
+int
+mq_start_quota_txn (xlator_t *this, loc_t *loc,
+                    quota_inode_ctx_t *ctx,
+                    inode_contribution_t *contri)
+{
+        int32_t        ret      = -1;
+        call_frame_t  *frame    = NULL;
+
+        ret = mq_prepare_txn_frame (this, loc, ctx,
+                                    contri, &frame);
+        if (ret)
+                goto err;
+
         ret = mq_get_lock_on_parent (frame, this);
         if (ret == -1)
                 goto err;
 
         return 0;
 
-fr_destroy:
-        QUOTA_STACK_DESTROY (frame, this);
 err:
         mq_set_ctx_updation_status (ctx, _gf_false);
 
@@ -1970,11 +2032,46 @@ mq_initiate_quota_txn (xlator_t *this, loc_t *loc)
                 goto out;
         }
 
+        /* Create the contribution node if its absent. Is it right to
+           assume that if the contribution node is not there, then
+           create one and proceed instead of returning?
+           Reason for this assumption is for hard links. Suppose
+           hard link for a file f1 present in a directory d1 is
+           created in the directory d2 (as f2). Now, since d2's
+           contribution is not there in f1's inode ctx, d2's
+           contribution xattr wont be created and will create problems
+           for quota operations.
+        */
         contribution = mq_get_contribution_node (loc->parent, ctx);
-        if (contribution == NULL)
-                goto out;
+        if (!contribution) {
+                if ((loc->path && strcmp (loc->path, "/"))
+                    || (!uuid_is_null (loc->gfid)
+                        && !__is_root_gfid (loc->gfid))
+                    || (loc->inode && !uuid_is_null (loc->inode->gfid)
+                        && !__is_root_gfid (loc->inode->gfid)))
+                        gf_log_callingfn (this->name, GF_LOG_TRACE,
+                                          "contribution node for the "
+                                          "path (%s) with parent (%s) "
+                                          "not found", loc->path,
+                                          loc->parent?
+                                          uuid_utoa (loc->parent->gfid):
+                                          NULL);
+
+                contribution = mq_add_new_contribution_node (this, ctx, loc);
+                if (!contribution) {
+                        if(loc->path && strcmp (loc->path, "/"))
+                                gf_log_callingfn (this->name, GF_LOG_WARNING,
+                                                  "could not allocate "
+                                                  " contribution node for (%s) "
+                                                  "parent: (%s)", loc->path,
+                                                  loc->parent?
+                                                  uuid_utoa (loc->parent->gfid):
+                                                  NULL);
+                        goto out;
+                }
+        }
 
-        /* To improve performance, donot start another transaction
+        /* To improve performance, do not start another transaction
          * if one is already in progress for same inode
          */
         status = _gf_true;
@@ -1993,16 +2090,7 @@ out:
 }
 
 
-/* int32_t */
-/* validate_inode_size_contribution (xlator_t *this, loc_t *loc, int64_t size, */
-/*                                int64_t contribution) */
-/* { */
-/*   if (size != contribution) { */
-/*     mq_initiate_quota_txn (this, loc); */
-/*   } */
 
-/*   return 0; */
-/* } */
 
 
 int32_t
@@ -2031,12 +2119,13 @@ mq_inspect_directory_xattr (xlator_t *this,
                 }
         }
 
-        if (strcmp (loc->path, "/") != 0) {
+        if (!loc->path || (loc->path && strcmp (loc->path, "/") != 0)) {
                 contribution = mq_add_new_contribution_node (this, ctx, loc);
                 if (contribution == NULL) {
                         if (!uuid_is_null (loc->inode->gfid))
-                                gf_log (this->name, GF_LOG_WARNING,
-                                        "cannot add a new contribution node");
+                                gf_log (this->name, GF_LOG_DEBUG,
+                                        "cannot add a new contribution node "
+                                        "(%s)", uuid_utoa (loc->inode->gfid));
                         ret = -1;
                         goto err;
                 }
@@ -2050,7 +2139,10 @@ mq_inspect_directory_xattr (xlator_t *this,
         if (ret < 0)
                 goto out;
 
-        if (strcmp (loc->path, "/") != 0) {
+        if ((loc->path && strcmp (loc->path, "/") != 0)
+            || (!uuid_is_null (loc->gfid) && !__is_root_gfid (loc->gfid))
+            || (loc->inode && !uuid_is_null (loc->inode->gfid) &&
+                !__is_root_gfid (loc->inode->gfid))) {
                 not_root = _gf_true;
 
                 GET_CONTRI_KEY (contri_key, contribution->gfid, ret);
@@ -2122,8 +2214,11 @@ mq_inspect_file_xattr (xlator_t *this,
         }
 
         contribution = mq_add_new_contribution_node (this, ctx, loc);
-        if (contribution == NULL)
+        if (contribution == NULL) {
+                gf_log_callingfn (this->name, GF_LOG_DEBUG, "cannot allocate "
+                                  "contribution node (path:%s)", loc->path);
                 goto out;
+        }
 
         LOCK (&ctx->lock);
         {
@@ -2155,8 +2250,12 @@ mq_inspect_file_xattr (xlator_t *this,
                         if (size != contri_int) {
                                 mq_initiate_quota_txn (this, loc);
                         }
-                } else
-                        mq_initiate_quota_txn (this, loc);
+                } else {
+                        if (size)
+                                mq_initiate_quota_txn (this, loc);
+                        else
+                                mq_set_inode_xattr (this, loc);
+                }
         }
 
 out:
@@ -2192,7 +2291,7 @@ mq_req_xattr (xlator_t *this,
                 goto set_size;
 
         //if not "/" then request contribution
-        if (strcmp (loc->path, "/") == 0)
+        if (loc->path && strcmp (loc->path, "/") == 0)
                 goto set_size;
 
         ret = mq_dict_set_contribution (this, dict, loc);
@@ -2235,6 +2334,12 @@ _mq_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this,
         int32_t        ret                = 0;
         char           contri_key [512]   = {0, };
         quota_local_t *local              = NULL;
+        inode_t       *inode              = NULL;
+        dentry_t      *tmp                = NULL;
+        gf_boolean_t  last_dentry         = _gf_true;
+        loc_t         loc                 = {0, };
+        dentry_t      *other_dentry       = NULL;
+        gf_boolean_t  remove              = _gf_false;
 
         local = (quota_local_t *) frame->local;
 
@@ -2245,17 +2350,84 @@ _mq_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this,
 
         frame->local = NULL;
 
-        if (local->hl_count > 1) {
-                GET_CONTRI_KEY (contri_key, local->contri->gfid, ret);
+        GET_CONTRI_KEY (contri_key, local->contri->gfid, ret);
 
-                STACK_WIND (frame, mq_removexattr_cbk, FIRST_CHILD(this),
-                            FIRST_CHILD(this)->fops->removexattr,
-                            &local->loc, contri_key, NULL);
-                ret = 0;
-        } else {
-                mq_removexattr_cbk (frame, NULL, this, 0, 0, NULL);
+        if (!local->loc.inode)
+                inode = inode_grep (local->loc.parent->table, local->loc.parent,
+                                    local->loc.name);
+        else
+                inode = inode_ref (local->loc.inode);
+
+        /* Suppose there are 2 directories dir1 and dir2. Quota limit is set on
+           both the directories. There is a file (f1) in dir1. A hark link is
+           created for that file inside the directory dir2 (say f2). Now one
+           more xattr is set in the inode as a new hard link is created in a
+           separate directory.
+           i.e trusted.glusterfs.quota.<gfid of dir2>.contri=<contribution>
+
+           Now when the hardlink f2 is removed, then the new xattr added (i.e
+           the xattr indicating its contribution to ITS parent directory) should
+           be removed (IFF there is not another hardlink for that file in the
+           same directory).
+
+           To do that upon getting unlink first check whether any other hard
+           links for the same inode exists in the same directory. If so do not
+           do anything and proceed for quota transaction.
+           Otherwise, if the removed entry was the only link for that inode
+           within that directory, then get another dentry for the inode
+           (by traversing the list of dentries for the inode) and using the
+           the dentry's parent and name, send removexattr so that the xattr
+           is removed.
+
+           If it is not done, then if the volume is restarted or the brick
+           process is restarted, then wrong quota usage will be shown for the
+           directory dir2.
+        */
+        if (inode) {
+                tmp = NULL;
+                list_for_each_entry (tmp, &inode->dentry_list, inode_list) {
+                        if (local->loc.parent == tmp->parent) {
+                                if (strcmp (local->loc.name, local->loc.name)) {
+                                        last_dentry = _gf_false;
+                                        break;
+                                }
+                        }
+                }
+                remove = last_dentry;
         }
 
+        if (remove) {
+                if (!other_dentry)    {
+                        list_for_each_entry (tmp, &inode->dentry_list,
+                                             inode_list) {
+                                if (local->loc.parent != tmp->parent) {
+                                        other_dentry = tmp;
+                                        break;
+                                }
+                        }
+                }
+
+                if (!other_dentry)
+                        mq_removexattr_cbk (frame, NULL, this, 0, 0, NULL);
+                else {
+                        loc.parent = inode_ref (other_dentry->parent);
+                        loc.name = gf_strdup (other_dentry->name);
+                        uuid_copy (loc.pargfid , other_dentry->parent->gfid);
+                        loc.inode = inode_ref (inode);
+                        uuid_copy (loc.gfid, inode->gfid);
+                        inode_path (other_dentry->parent, other_dentry->name,
+                                    (char **)&loc.path);
+
+                        STACK_WIND (frame, mq_removexattr_cbk,
+                                    FIRST_CHILD(this),
+                                    FIRST_CHILD(this)->fops->removexattr,
+                                    &loc, contri_key, NULL);
+                }
+        } else
+                mq_removexattr_cbk (frame, NULL, this, 0, 0, NULL);
+
+        ret = 0;
+
         if (strcmp (local->parent_loc.path, "/") != 0) {
                 ret = mq_get_parent_inode_local (this, local);
                 if (ret < 0)
@@ -2266,6 +2438,8 @@ _mq_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this,
 out:
         mq_local_unref (this, local);
 
+        loc_wipe (&loc);
+        inode_unref (inode);
         return 0;
 }
 
@@ -2392,8 +2566,11 @@ mq_reduce_parent_size (xlator_t *this, loc_t *loc, int64_t contri)
                 goto out;
 
         contribution = mq_get_contribution_node (loc->parent, ctx);
-        if (contribution == NULL)
+        if (contribution == NULL) {
+                gf_log_callingfn (this->name, GF_LOG_WARNING, "contribution for"
+                                  " the node %s is NULL", loc->path);
                 goto out;
+        }
 
         local = mq_local_new ();
         if (local == NULL) {
@@ -2412,6 +2589,8 @@ mq_reduce_parent_size (xlator_t *this, loc_t *loc, int64_t contri)
         }
 
         if (local->size == 0) {
+                gf_log_callingfn (this->name, GF_LOG_TRACE,
+                                  "local->size is 0 " "path: (%s)", loc->path);
                 ret = 0;
                 goto out;
         }
@@ -2424,8 +2603,12 @@ mq_reduce_parent_size (xlator_t *this, loc_t *loc, int64_t contri)
         local->contri = contribution;
 
         ret = mq_inode_loc_fill (NULL, loc->parent, &local->parent_loc);
-        if (ret < 0)
+        if (ret < 0) {
+                gf_log_callingfn (this->name, GF_LOG_INFO, "building parent loc"
+                                  " failed. (gfid: %s)",
+                                  uuid_utoa (loc->parent->gfid));
                 goto out;
+        }
 
         frame = create_frame (this, this->ctx->pool);
         if (!frame) {
diff --git a/xlators/features/marker/src/marker-quota.h b/xlators/features/marker/src/marker-quota.h
index 385760ac4dd..42def9d22dc 100644
--- a/xlators/features/marker/src/marker-quota.h
+++ b/xlators/features/marker/src/marker-quota.h
@@ -42,8 +42,6 @@
                 var = GF_CALLOC (sizeof (type), 1,      \
                                  gf_marker_mt_##type);  \
                 if (!var) {                             \
-                        gf_log ("", GF_LOG_ERROR,       \
-                                "out of memory");       \
                         ret = -1;                       \
                 }                                       \
         } while (0);
@@ -61,13 +59,20 @@
                 ret = 0;                                \
         } while (0);
 
-#define GET_CONTRI_KEY(var, _gfid, _ret)        \
-        do {                                    \
-                char _gfid_unparsed[40];        \
-                uuid_unparse (_gfid, _gfid_unparsed); \
-                _ret = snprintf (var, CONTRI_KEY_MAX, QUOTA_XATTR_PREFIX \
-                                 ".%s.%s." CONTRIBUTION, "quota", \
-                                 _gfid_unparsed); \
+#define GET_CONTRI_KEY(var, _gfid, _ret)              \
+        do {                                          \
+                if (_gfid != NULL) {                  \
+                        char _gfid_unparsed[40];      \
+                        uuid_unparse (_gfid, _gfid_unparsed);           \
+                        _ret = snprintf (var, CONTRI_KEY_MAX,           \
+                                         QUOTA_XATTR_PREFIX             \
+                                         ".%s.%s." CONTRIBUTION, "quota", \
+                                         _gfid_unparsed);               \
+                } else {                                                \
+                        _ret = snprintf (var, CONTRI_KEY_MAX,           \
+                                         QUOTA_XATTR_PREFIX             \
+                                         ".%s.." CONTRIBUTION, "quota"); \
+                }                                                       \
         } while (0);
 
 #define QUOTA_SAFE_INCREMENT(lock, var)         \
diff --git a/xlators/features/marker/src/marker.c b/xlators/features/marker/src/marker.c
index 6df1f5688e2..e448bc08f67 100644
--- a/xlators/features/marker/src/marker.c
+++ b/xlators/features/marker/src/marker.c
@@ -303,13 +303,11 @@ marker_getxattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
 
         priv = this->private;
 
-        if (priv == NULL || (priv->feature_enabled & GF_XTIME) == 0)
-                goto wind;
-
         gf_log (this->name, GF_LOG_DEBUG, "USER:PID = %d", frame->root->pid);
 
-        ret = call_from_special_client (frame, this, name);
-wind:
+        if (priv && priv->feature_enabled & GF_XTIME)
+                ret = call_from_special_client (frame, this, name);
+
         if (ret == _gf_false) {
                 if (name == NULL) {
                         /* Signifies that marker translator
@@ -807,8 +805,10 @@ marker_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         priv = this->private;
 
-        if ((priv->feature_enabled & GF_QUOTA) && (local->ia_nlink == 1))
-                mq_reduce_parent_size (this, &local->loc, -1);
+        if (priv->feature_enabled & GF_QUOTA) {
+                if (!local->skip_txn)
+                        mq_reduce_parent_size (this, &local->loc, -1);
+        }
 
         if (priv->feature_enabled & GF_XTIME)
                 marker_xtime_update_marks (this, local);
@@ -874,6 +874,11 @@ marker_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
         if (ret == -1)
                 goto err;
 
+        if (xdata && dict_get (xdata, GLUSTERFS_INTERNAL_FOP_KEY)) {
+                local->skip_txn = 1;
+                goto unlink_wind;
+        }
+
         if (uuid_is_null (loc->gfid) && loc->inode)
                 uuid_copy (loc->gfid, loc->inode->gfid);
 
@@ -919,8 +924,11 @@ marker_link_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         priv = this->private;
 
-        if (priv->feature_enabled & GF_QUOTA)
-                mq_initiate_quota_txn (this, &local->loc);
+        if (priv->feature_enabled & GF_QUOTA) {
+                if (!local->skip_txn)
+                        mq_set_inode_xattr (this, &local->loc);
+        }
+
 
         if (priv->feature_enabled & GF_XTIME)
                 marker_xtime_update_marks (this, local);
@@ -951,6 +959,9 @@ marker_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
 
         if (ret == -1)
                 goto err;
+
+        if (xdata && dict_get (xdata, GLUSTERFS_INTERNAL_FOP_KEY))
+                local->skip_txn = 1;
 wind:
         STACK_WIND (frame, marker_link_cbk, FIRST_CHILD(this),
                     FIRST_CHILD(this)->fops->link, oldloc, newloc, xdata);
@@ -1011,7 +1022,7 @@ marker_rename_done (call_frame_t *frame, void *cookie, xlator_t *this,
                 newloc.name++;
         newloc.parent = inode_ref (local->loc.parent);
 
-        mq_rename_update_newpath (this, &newloc);
+        mq_set_inode_xattr (this, &newloc);
 
         loc_wipe (&newloc);
 
@@ -2490,22 +2501,94 @@ err:
         return 0;
 }
 
+
+int
+marker_build_ancestry_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                           int op_ret, int op_errno, gf_dirent_t *entries,
+                           dict_t *xdata)
+{
+        gf_dirent_t *entry = NULL;
+        loc_t        loc    = {0, };
+        inode_t     *parent = NULL;
+
+        if ((op_ret <= 0) || (entries == NULL)) {
+                goto out;
+        }
+
+
+        list_for_each_entry (entry, &entries->list, list) {
+                if (entry->inode == entry->inode->table->root) {
+                        loc.path = gf_strdup ("/");
+                        inode_unref (parent);
+                        parent = NULL;
+                }
+
+                loc.inode = inode_ref (entry->inode);
+
+                if (parent != NULL) {
+                        loc.parent = inode_ref (parent);
+                        uuid_copy (loc.pargfid, parent->gfid);
+                }
+
+                uuid_copy (loc.gfid, entry->d_stat.ia_gfid);
+
+                mq_xattr_state (this, &loc, entry->dict, entry->d_stat);
+
+                inode_unref (parent);
+                parent = inode_ref (entry->inode);
+                loc_wipe (&loc);
+        }
+
+        if (parent)
+                inode_unref (parent);
+
+out:
+        STACK_UNWIND_STRICT (readdirp, frame, op_ret, op_errno, entries, xdata);
+        return 0;
+}
+
 int
 marker_readdirp_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                      int op_ret, int op_errno, gf_dirent_t *entries,
                      dict_t *xdata)
 {
-        gf_dirent_t *entry = NULL;
+        gf_dirent_t    *entry = NULL;
+        marker_conf_t  *priv  = NULL;
+        marker_local_t *local = NULL;
+        loc_t           loc   = {0, };
 
         if (op_ret <= 0)
                 goto unwind;
 
+        priv = this->private;
+        local = frame->local;
+
+        if (!(priv->feature_enabled & GF_QUOTA) || (local == NULL)) {
+                goto unwind;
+        }
+
         list_for_each_entry (entry, &entries->list, list) {
-                /* TODO: fill things */
+                if ((strcmp (entry->d_name, ".") == 0) ||
+                    (strcmp (entry->d_name, "..") == 0))
+                        continue;
+
+                loc.inode = inode_ref (entry->inode);
+                loc.parent = inode_ref (local->loc.inode);
+
+                uuid_copy (loc.gfid, entry->d_stat.ia_gfid);
+                uuid_copy (loc.pargfid, loc.parent->gfid);
+
+                mq_xattr_state (this, &loc, entry->dict, entry->d_stat);
+
+                loc_wipe (&loc);
         }
 
 unwind:
+        local = frame->local;
+        frame->local = NULL;
+
         STACK_UNWIND_STRICT (readdirp, frame, op_ret, op_errno, entries, xdata);
+        marker_local_unref (local);
 
         return 0;
 }
@@ -2514,20 +2597,36 @@ int
 marker_readdirp (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
                  off_t offset, dict_t *dict)
 {
-        marker_conf_t  *priv    = NULL;
+        marker_conf_t  *priv  = NULL;
+        loc_t           loc   = {0, };
+        marker_local_t *local = NULL;
 
         priv = this->private;
 
-        if (priv->feature_enabled == 0)
-                goto wind;
+        if ((dict != NULL) && dict_get (dict, GET_ANCESTRY_DENTRY_KEY)) {
+                STACK_WIND (frame, marker_build_ancestry_cbk,
+                            FIRST_CHILD(this),
+                            FIRST_CHILD(this)->fops->readdirp,
+                            fd, size, offset, dict);
+        } else {
+                if (priv->feature_enabled & GF_QUOTA) {
+                        local = mem_get0 (this->local_pool);
 
-        if ((priv->feature_enabled & GF_QUOTA) && dict)
-                mq_req_xattr (this, NULL, dict);
+                        MARKER_INIT_LOCAL (frame, local);
 
-wind:
-        STACK_WIND (frame, marker_readdirp_cbk,
-                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->readdirp,
-                    fd, size, offset, dict);
+                        loc.parent = local->loc.inode = inode_ref (fd->inode);
+
+                        if (dict == NULL)
+                                dict = dict_new ();
+
+                        mq_req_xattr (this, &loc, dict);
+                }
+
+                STACK_WIND (frame, marker_readdirp_cbk,
+                            FIRST_CHILD(this),
+                            FIRST_CHILD(this)->fops->readdirp,
+                            fd, size, offset, dict);
+        }
 
         return 0;
 }
diff --git a/xlators/features/marker/src/marker.h b/xlators/features/marker/src/marker.h
index 1a58f8cfcbc..23d1580f0e5 100644
--- a/xlators/features/marker/src/marker.h
+++ b/xlators/features/marker/src/marker.h
@@ -112,6 +112,7 @@ struct marker_local{
 
         int xflag;
         dict_t *xdata;
+        gf_boolean_t skip_txn;
 };
 typedef struct marker_local marker_local_t;