features/marker: QUOTA related changes in marker translator.

Signed-off-by: Junaid <junaid@gluster.com>
Signed-off-by: Vijay Bellur <vijay@dev.gluster.com>

BUG: 2473 (Support for volume and directory level quota)
URL: http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=2473

10 files changed, 2986 insertions, 91 deletions

diff --git a/xlators/features/marker/src/Makefile.am b/xlators/features/marker/src/Makefile.am
index 58b12b3f594..501586a76b6 100644
--- a/xlators/features/marker/src/Makefile.am
+++ b/xlators/features/marker/src/Makefile.am
@@ -3,10 +3,10 @@ xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/features
 
 marker_la_LDFLAGS = -module -avoidversion
 
-marker_la_SOURCES = marker.c
+marker_la_SOURCES = marker.c marker-quota.c marker-quota-helper.c marker-common.c
 marker_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la
 
-noinst_HEADERS = marker-mem-types.h marker.h $(top_builddir)/xlators/lib/src/libxlator.h
+noinst_HEADERS = marker-mem-types.h marker.h marker-quota.h marker-quota-helper.h marker-common.h $(top_builddir)/xlators/lib/src/libxlator.h
 
 AM_CFLAGS = -fPIC -D_FILE_OFFSET_BITS=64 -D_GNU_SOURCE -Wall -fno-strict-aliasing -D$(GF_HOST_OS) \
         -I$(top_srcdir)/libglusterfs/src -I$(top_srcdir)/xlators/lib/src $(GF_CFLAGS) -shared -nostartfiles
diff --git a/xlators/features/marker/src/marker-common.c b/xlators/features/marker/src/marker-common.c
new file mode 100644
index 00000000000..3e2d7f00047
--- /dev/null
+++ b/xlators/features/marker/src/marker-common.c
@@ -0,0 +1,70 @@
+/*Copyright (c) 2008-2010 Gluster, Inc. <http://www.gluster.com>
+  This file is part of GlusterFS.
+
+  GlusterFS is free software; you can redistribute it and/or modify
+  it under the terms of the GNU 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 "marker-common.h"
+
+marker_inode_ctx_t *
+marker_inode_ctx_new ()
+{
+        marker_inode_ctx_t *ctx = NULL;
+
+        ctx = GF_CALLOC (1, sizeof (marker_inode_ctx_t),
+                         gf_marker_mt_marker_inode_ctx_t);
+        if (ctx == NULL)
+                goto out;
+
+        ctx->quota_ctx = NULL;
+out:
+        return ctx;
+}
+
+int32_t
+marker_force_inode_ctx_get (inode_t *inode, xlator_t *this,
+                            marker_inode_ctx_t **ctx)
+{
+        int32_t  ret = -1;
+        uint64_t ctx_int;
+
+        LOCK (&inode->lock);
+        {
+                ret = __inode_ctx_get (inode, this, &ctx_int);
+                if (ret == 0)
+                        *ctx = (marker_inode_ctx_t *) ctx_int;
+                else {
+                        *ctx = marker_inode_ctx_new ();
+                        if (*ctx == NULL)
+                                goto unlock;
+
+                        ret = __inode_ctx_put (inode, this, (uint64_t ) *ctx);
+                        if (ret == -1) {
+                                GF_FREE (*ctx);
+                                goto unlock;
+                        }
+                        ret = 0;
+                }
+        }
+unlock: UNLOCK (&inode->lock);
+
+        return ret;
+}
+
diff --git a/xlators/features/marker/src/marker-common.h b/xlators/features/marker/src/marker-common.h
new file mode 100644
index 00000000000..311e4e23832
--- /dev/null
+++ b/xlators/features/marker/src/marker-common.h
@@ -0,0 +1,33 @@
+/*Copyright (c) 2008-2010 Gluster, Inc. <http://www.gluster.com>
+  This file is part of GlusterFS.
+
+  GlusterFS is free software; you can redistribute it and/or modify
+  it under the terms of the GNU 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 _MARKER_COMMON_H
+#define _MARKER_COMMON_H
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "inode.h"
+#include "xlator.h"
+#include "marker.h"
+
+int32_t
+marker_force_inode_ctx_get (inode_t *, xlator_t *, marker_inode_ctx_t **);
+#endif
diff --git a/xlators/features/marker/src/marker-mem-types.h b/xlators/features/marker/src/marker-mem-types.h
index 3936ef794fc..847bfa67c05 100644
--- a/xlators/features/marker/src/marker-mem-types.h
+++ b/xlators/features/marker/src/marker-mem-types.h
@@ -27,6 +27,11 @@ enum gf_marker_mem_types_ {
         gf_marker_mt_marker_conf_t,
         gf_marker_mt_loc_t,
         gf_marker_mt_volume_mark,
+        gf_marker_mt_int64_t,
+        gf_marker_mt_quota_inode_ctx_t,
+        gf_marker_mt_marker_inode_ctx_t,
+        gf_marker_mt_quota_local_t,
+        gf_marker_mt_inode_contribution_t,
         gf_marker_mt_end
 };
 #endif
diff --git a/xlators/features/marker/src/marker-quota-helper.c b/xlators/features/marker/src/marker-quota-helper.c
new file mode 100644
index 00000000000..b6a87698eed
--- /dev/null
+++ b/xlators/features/marker/src/marker-quota-helper.c
@@ -0,0 +1,366 @@
+/*Copyright (c) 2008-2010 Gluster, Inc. <http://www.gluster.com>
+  This file is part of GlusterFS.
+
+  GlusterFS is free software; you can redistribute it and/or modify
+  it under the terms of the GNU 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 "locking.h"
+#include "marker-quota.h"
+#include "marker-common.h"
+#include "marker-quota-helper.h"
+#include "marker-mem-types.h"
+
+int
+quota_loc_fill (loc_t *loc, inode_t *inode, inode_t *parent, char *path)
+{
+        int ret = -1;
+
+        if (!loc)
+                return ret;
+
+        if (inode) {
+                loc->inode = inode_ref (inode);
+                loc->ino = inode->ino;
+        }
+
+        if (parent)
+                loc->parent = inode_ref (parent);
+
+        loc->path = gf_strdup (path);
+        if (!loc->path) {
+                gf_log ("loc fill", GF_LOG_ERROR, "strdup failed");
+                goto loc_wipe;
+        }
+
+        loc->name = strrchr (loc->path, '/');
+        if (loc->name)
+                loc->name++;
+        else
+                goto loc_wipe;
+
+        ret = 0;
+loc_wipe:
+        if (ret < 0)
+                loc_wipe (loc);
+
+        return ret;
+}
+
+
+int32_t
+quota_inode_loc_fill (const char *parent_gfid, inode_t *inode, loc_t *loc)
+{
+        char            *resolvedpath = NULL;
+        inode_t         *parent       = NULL;
+        int              ret          = -1;
+
+        if ((!inode) || (!loc))
+                return ret;
+
+        if ((inode) && (inode->ino == 1)) {
+                loc->parent = NULL;
+                goto ignore_parent;
+        }
+
+        if (parent_gfid == NULL)
+                parent = inode_parent (inode, 0, NULL);
+        else
+                parent = inode_find (inode->table,
+                                     (unsigned char *) parent_gfid);
+
+        if (parent == NULL)
+                goto err;
+
+ignore_parent:
+        ret = inode_path (inode, NULL, &resolvedpath);
+        if (ret < 0)
+                goto err;
+
+        ret = quota_loc_fill (loc, inode, parent, resolvedpath);
+        if (ret < 0)
+                goto err;
+
+err:
+        if (parent)
+                inode_unref (parent);
+
+        GF_FREE (resolvedpath);
+
+        return ret;
+}
+
+
+quota_inode_ctx_t *
+quota_alloc_inode_ctx ()
+{
+        int32_t                  ret    = -1;
+        quota_inode_ctx_t       *ctx    = NULL;
+
+        QUOTA_ALLOC (ctx, quota_inode_ctx_t, ret);
+        if (ret == -1)
+                goto out;
+
+        ctx->size = 0;
+        ctx->dirty = 0;
+        LOCK_INIT (&ctx->lock);
+        INIT_LIST_HEAD (&ctx->contribution_head);
+out:
+        return ctx;
+}
+
+inode_contribution_t *
+get_contribution_node (inode_t *inode, quota_inode_ctx_t *ctx)
+{
+        inode_contribution_t    *contri = NULL;
+        inode_contribution_t    *temp   = NULL;
+
+        GF_VALIDATE_OR_GOTO ("marker", inode, out);
+        GF_VALIDATE_OR_GOTO ("marker", ctx, out);
+
+        list_for_each_entry (temp, &ctx->contribution_head, contri_list) {
+                if (uuid_compare (temp->gfid, inode->gfid) == 0) {
+                        contri = temp;
+                        goto out;
+                }
+        }
+out:
+        return contri;
+}
+
+
+int32_t
+delete_contribution_node (dict_t *dict, char *key,
+                          inode_contribution_t *contribution)
+{
+        if (dict_get (dict, key) != NULL)
+                goto out;
+
+        QUOTA_FREE_CONTRIBUTION_NODE (contribution);
+out:
+        return 0;
+}
+
+
+inode_contribution_t *
+__add_new_contribution_node (xlator_t *this, quota_inode_ctx_t *ctx, loc_t *loc)
+{
+        int32_t ret = 0;
+        inode_contribution_t *contribution = NULL;
+
+        list_for_each_entry (contribution, &ctx->contribution_head, contri_list) {
+                if (uuid_compare (contribution->gfid, loc->parent->gfid) == 0) {
+                        goto out;
+                }
+        }
+
+        QUOTA_ALLOC (contribution, inode_contribution_t, ret);
+        if (ret == -1)
+                goto out;
+
+        contribution->contribution = 0;
+
+        uuid_copy (contribution->gfid, loc->parent->gfid);
+
+        list_add_tail (&contribution->contri_list, &ctx->contribution_head);
+
+out:
+        return contribution;
+}
+
+
+inode_contribution_t *
+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)
+                return NULL;
+
+        LOCK (&ctx->lock);
+        {
+                contribution = __add_new_contribution_node (this, ctx, loc);
+        }
+        UNLOCK (&ctx->lock);
+
+        return contribution;
+}
+
+
+int32_t
+dict_set_contribution (xlator_t *this, dict_t *dict,
+                       loc_t *loc)
+{
+        int32_t ret              = -1;
+        char    contri_key [512] = {0, };
+
+        GET_CONTRI_KEY (contri_key, loc->parent->gfid, ret);
+        if (ret < 0) {
+                ret = -1;
+                goto out;
+        }
+
+        ret = dict_set_int64 (dict, contri_key, 0);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "unable to set dict value.");
+                goto out;
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
+
+
+int32_t
+quota_inode_ctx_get (inode_t *inode, xlator_t *this,
+                     quota_inode_ctx_t **ctx)
+{
+        int32_t             ret      = -1;
+        uint64_t            ctx_int  = 0;
+        marker_inode_ctx_t *mark_ctx = NULL;
+
+        GF_VALIDATE_OR_GOTO ("marker", inode, out);
+        GF_VALIDATE_OR_GOTO ("marker", this, out);
+        GF_VALIDATE_OR_GOTO ("marker", ctx, out);
+
+        ret = inode_ctx_get (inode, this, &ctx_int);
+        if (ret < 0) {
+                ret = -1;
+                *ctx = NULL;
+                goto out;
+        }
+
+        mark_ctx = (marker_inode_ctx_t *) ctx_int;
+        if (mark_ctx->quota_ctx == NULL) {
+                ret = -1;
+                goto out;
+        }
+
+        *ctx = mark_ctx->quota_ctx;
+
+        ret = 0;
+
+out:
+        return ret;
+}
+
+
+quota_inode_ctx_t *
+__quota_inode_ctx_new (inode_t *inode, xlator_t *this)
+{
+        int32_t               ret        = -1;
+        quota_inode_ctx_t    *quota_ctx  = NULL;
+        marker_inode_ctx_t   *mark_ctx   = NULL;
+
+        ret = marker_force_inode_ctx_get (inode, this, &mark_ctx);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "marker_force_inode_ctx_get() failed");
+                goto out;
+        }
+
+        LOCK (&inode->lock);
+        {
+                if (mark_ctx->quota_ctx == NULL) {
+                        quota_ctx = quota_alloc_inode_ctx ();
+                        if (quota_ctx == NULL) {
+                                ret = -1;
+                                goto unlock;
+                        }
+                        mark_ctx->quota_ctx = quota_ctx;
+                } else
+                        quota_ctx = mark_ctx->quota_ctx;
+
+                ret = 0;
+        }
+unlock: UNLOCK (&inode->lock);
+
+out:
+        return quota_ctx;
+}
+
+
+quota_inode_ctx_t *
+quota_inode_ctx_new (inode_t * inode, xlator_t *this)
+{
+        return __quota_inode_ctx_new (inode, this);
+}
+
+quota_local_t *
+quota_local_new ()
+{
+        int32_t         ret     = -1;
+        quota_local_t  *local   = NULL;
+
+        QUOTA_ALLOC (local, quota_local_t, ret);
+        if (ret < 0)
+                goto out;
+
+        local->ref = 1;
+        local->delta = 0;
+        local->err = 0;
+        LOCK_INIT (&local->lock);
+
+        memset (&local->loc, 0, sizeof (loc_t));
+        memset (&local->parent_loc, 0, sizeof (loc_t));
+
+        local->ctx = NULL;
+        local->contri = NULL;
+
+out:
+        return local;
+}
+
+quota_local_t *
+quota_local_ref (quota_local_t *local)
+{
+        LOCK (&local->lock);
+        {
+                local->ref ++;
+        }
+        UNLOCK (&local->lock);
+
+        return local;
+}
+
+
+int32_t
+quota_local_unref (xlator_t *this, quota_local_t *local)
+{
+        if (local == NULL)
+                goto out;
+
+        QUOTA_SAFE_DECREMENT (&local->lock, local->ref);
+
+        if (local->ref > 0)
+                goto out;
+
+        loc_wipe (&local->loc);
+
+        loc_wipe (&local->parent_loc);
+
+        LOCK_DESTROY (&local->lock);
+out:
+        return 0;
+}
diff --git a/xlators/features/marker/src/marker-quota-helper.h b/xlators/features/marker/src/marker-quota-helper.h
new file mode 100644
index 00000000000..9a24c8c3d94
--- /dev/null
+++ b/xlators/features/marker/src/marker-quota-helper.h
@@ -0,0 +1,76 @@
+/*Copyright (c) 2008-2010 Gluster, Inc. <http://www.gluster.com>
+  This file is part of GlusterFS.
+
+  GlusterFS is free software; you can redistribute it and/or modify
+  it under the terms of the GNU 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 _MARKER_QUOTA_HELPER_H
+#define _MARKER_QUOTA_HELPER
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+#include "marker-quota.h"
+
+#define QUOTA_FREE_CONTRIBUTION_NODE(_contribution)     \
+        do {                                            \
+                list_del (&_contribution->contri_list); \
+                GF_FREE (_contribution);                \
+        } while (0)
+
+#define QUOTA_SAFE_INCREMENT(lock, var)                 \
+        do {                                            \
+                LOCK (lock);                            \
+                        var ++;                         \
+                UNLOCK (lock);                          \
+        } while (0)
+
+#define QUOTA_SAFE_DECREMENT(lock, var)                 \
+        do {                                            \
+                LOCK (lock);                            \
+                        var --;                         \
+                UNLOCK (lock);                          \
+        } while (0)
+
+inode_contribution_t *
+add_new_contribution_node (xlator_t *, quota_inode_ctx_t *, loc_t *);
+
+int32_t
+dict_set_contribution (xlator_t *, dict_t *, loc_t *);
+
+quota_inode_ctx_t *
+quota_inode_ctx_new (inode_t *, xlator_t *);
+
+int32_t
+quota_inode_ctx_get (inode_t *, xlator_t *, quota_inode_ctx_t **);
+
+int32_t
+delete_contribution_node (dict_t *, char *, inode_contribution_t *);
+
+int32_t
+quota_inode_loc_fill (const char *, inode_t *, loc_t *);
+
+quota_local_t *
+quota_local_new ();
+
+quota_local_t *
+quota_local_ref (quota_local_t *);
+
+int32_t
+quota_local_unref (xlator_t *, quota_local_t *);
+
+inode_contribution_t *
+get_contribution_node (inode_t *, quota_inode_ctx_t *);
+#endif
diff --git a/xlators/features/marker/src/marker-quota.c b/xlators/features/marker/src/marker-quota.c
new file mode 100644
index 00000000000..f7b397a5a84
--- /dev/null
+++ b/xlators/features/marker/src/marker-quota.c
@@ -0,0 +1,1815 @@
+/*Copyright (c) 2008-2010 Gluster, Inc. <http://www.gluster.com>
+  This file is part of GlusterFS.
+
+  GlusterFS is free software; you can redistribute it and/or modify
+  it under the terms of the GNU 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 "dict.h"
+#include "xlator.h"
+#include "defaults.h"
+#include "libxlator.h"
+#include "common-utils.h"
+#include "byte-order.h"
+#include "marker-quota.h"
+#include "marker-quota-helper.h"
+
+int32_t
+loc_fill_from_name (xlator_t *this, loc_t *newloc, loc_t *oldloc, uint64_t ino, char *name)
+{
+        int32_t   ret  = 0;
+        int32_t   len  = 0;
+        char     *path = NULL;
+
+
+        newloc->ino = ino;
+
+        newloc->inode = inode_new (oldloc->inode->table);
+
+        if (!newloc->inode) {
+                ret = -1;
+                goto out;
+        }
+
+        newloc->parent = inode_ref (oldloc->inode);
+
+        len = strlen (oldloc->path);
+
+        if (oldloc->path [len - 1] == '/')
+                ret = gf_asprintf ((char **) &path, "%s%s",
+                                   oldloc->path, name);
+        else
+                ret = gf_asprintf ((char **) &path, "%s/%s",
+                                   oldloc->path, name);
+
+        if (ret < 0)
+                goto out;
+
+        newloc->path = path;
+
+        newloc->name = strrchr (newloc->path, '/');
+
+        if (newloc->name)
+                newloc->name++;
+
+        gf_log (this->name, GF_LOG_INFO, "path = %s name =%s",newloc->path, newloc->name);
+out:
+        return ret;
+}
+
+int32_t
+dirty_inode_updation_done (call_frame_t *frame, void *cookie, xlator_t *this,
+                           int32_t op_ret, int32_t op_errno)
+{
+        quota_local_t *local = NULL;
+
+        local = frame->local;
+
+        if (!local->err)
+                QUOTA_SAFE_DECREMENT (&local->lock, local->ref);
+        else
+                frame->local = NULL;
+
+        QUOTA_STACK_DESTROY (frame, this);
+
+        return 0;
+}
+
+int32_t
+release_lock_on_dirty_inode (call_frame_t *frame, void *cookie, xlator_t *this,
+                             int32_t op_ret, int32_t op_errno)
+{
+        struct gf_flock   lock;
+        quota_local_t    *local = NULL;
+
+        if (op_ret == -1) {
+                local->err = -1;
+
+                dirty_inode_updation_done (frame, NULL, this, 0, 0);
+
+                return 0;
+        }
+
+        local = frame->local;
+
+        if (op_ret == 0)
+                local->ctx->dirty = 0;
+
+        lock.l_type   = F_UNLCK;
+        lock.l_whence = SEEK_SET;
+        lock.l_start  = 0;
+        lock.l_len    = 0;
+        lock.l_pid    = 0;
+
+        STACK_WIND (frame,
+                    dirty_inode_updation_done,
+                    FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->inodelk,
+                    this->name, &local->loc, F_SETLKW, &lock);
+
+        return 0;
+}
+
+int32_t
+mark_inode_undirty (call_frame_t *frame, void *cookie, xlator_t *this,
+                     int32_t op_ret, int32_t op_errno, dict_t *dict)
+{
+        int32_t        ret       = -1;
+        int64_t       *size      = NULL;
+        dict_t        *newdict   = NULL;
+        quota_local_t *local     = NULL;
+        marker_conf_t  *priv      = NULL;
+
+        if (op_ret == -1)
+                goto err;
+
+        priv = (marker_conf_t *) this->private;
+
+        local = (quota_local_t *) frame->local;
+
+        if (!dict)
+                goto wind;
+
+        ret = dict_get_bin (dict, priv->size_key, (void **) &size);
+        if (ret)
+                goto wind;
+
+        local->ctx->size = ntoh64 (*size);
+
+wind:
+        newdict = dict_new ();
+        if (!newdict)
+                goto err;
+
+        ret = dict_set_int8 (newdict, priv->dirty_key, 0);
+        if (ret)
+                goto err;
+
+        STACK_WIND (frame, release_lock_on_dirty_inode,
+                    FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->setxattr,
+                    &local->loc, newdict, 0);
+        ret = 0;
+
+err:
+        if (op_ret == -1 || ret == -1) {
+                local->err = -1;
+
+                release_lock_on_dirty_inode (frame, NULL, this, 0, 0);
+        }
+
+        if (newdict)
+                dict_unref (newdict);
+
+        return 0;
+}
+
+int32_t
+update_size_xattr (call_frame_t *frame, void *cookie, xlator_t *this,
+                   int32_t op_ret, int32_t op_errno, inode_t *inode,
+                   struct iatt *buf, dict_t *dict, struct iatt *postparent)
+{
+        int32_t          ret      = -1;
+        dict_t          *new_dict = NULL;
+        int64_t         *size     = NULL;
+        int64_t         *delta    = NULL;
+        quota_local_t   *local    = NULL;
+        marker_conf_t    *priv     = NULL;
+
+        if (op_ret == -1)
+                goto err;
+
+        priv = this->private;
+
+        if (!dict)
+                goto err;
+
+        local = frame->local;
+
+        ret = dict_get_bin (dict, priv->size_key, (void **) &size);
+        if (!size)
+                goto err;
+
+        QUOTA_ALLOC_OR_GOTO (delta, int64_t, ret, err);
+
+        *delta = ntoh64 (ntoh64 (*size) - local->sum);
+
+        gf_log (this->name, GF_LOG_DEBUG, "calculated size = %ld, original size = %ld"
+                        "path = %s diff = %ld", local->sum, ntoh64 (*size),
+                        local->loc.path, ntoh64 (*delta));
+
+        new_dict = dict_new ();
+        if (!new_dict);
+
+        ret = dict_set_bin (new_dict, priv->size_key, delta, 8);
+        if (ret)
+                goto err;
+
+        STACK_WIND (frame, mark_inode_undirty, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->xattrop, &local->loc,
+                    GF_XATTROP_ADD_ARRAY64, new_dict);
+
+        ret = 0;
+
+err:
+        if (op_ret == -1 || ret == -1) {
+                local->err = -1;
+
+                release_lock_on_dirty_inode (frame, NULL, this, 0, 0);
+        }
+
+        if (new_dict)
+                dict_unref (dict);
+
+        return 0;
+}
+
+int32_t
+get_dirty_inode_size (call_frame_t *frame, xlator_t *this)
+{
+        int32_t          ret            = -1;
+        dict_t          *dict           = NULL;
+        quota_local_t   *local          = NULL;
+        marker_conf_t    *priv           = NULL;
+
+        local = (quota_local_t *) frame->local;
+
+        priv = (marker_conf_t *) this->private;
+
+        dict = dict_new ();
+        if (!dict) {
+                ret = -1;
+                goto err;
+        }
+
+        ret = dict_set_int64 (dict, priv->size_key, 0);
+        if (ret)
+                goto err;
+
+        STACK_WIND (frame, update_size_xattr, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->lookup, &local->loc, dict);
+        ret =0;
+
+err:
+        if (ret) {
+                local->err = -1;
+
+                release_lock_on_dirty_inode (frame, NULL, this, 0, 0);
+        }
+
+        return 0;
+}
+
+int32_t
+get_child_contribution (call_frame_t *frame,
+                        void *cookie,
+                        xlator_t *this,
+                        int32_t op_ret,
+                        int32_t op_errno,
+                        inode_t *inode,
+                        struct iatt *buf,
+                        dict_t *dict,
+                        struct iatt *postparent)
+{
+        int32_t          ret            = -1;
+        char             contri_key [512] = {0, };
+        int64_t         *contri         = NULL;
+        quota_local_t   *local          = NULL;
+
+        local = frame->local;
+
+        frame->local = NULL;
+
+        QUOTA_STACK_DESTROY (frame, this);
+
+        if (op_ret == -1) {
+                gf_log (this->name, GF_LOG_ERROR, "%s", strerror (op_errno));
+
+                local->err = -2;
+
+                release_lock_on_dirty_inode (local->frame, NULL, this, 0, 0);
+
+                goto out;
+        }
+
+        if (local->err)
+                goto out;
+
+        GET_CONTRI_KEY (contri_key, local->loc.inode->gfid, ret);
+        if (ret < 0)
+                goto out;
+
+        if (!dict)
+                goto out;
+
+        if (dict_get_bin (dict, contri_key, (void **) &contri) == 0)
+                local->sum += ntoh64 (*contri);
+
+out:
+        LOCK (&local->lock);
+        {
+                local->dentry_child_count--;
+        }
+        UNLOCK (&local->lock);
+
+        if (local->dentry_child_count == 0) {
+                if (local->err) {
+                        QUOTA_SAFE_DECREMENT (&local->lock, local->ref);
+
+                        quota_local_unref (this, local);
+                } else
+                        quota_dirty_inode_readdir (local->frame, NULL, this,
+                                                   0, 0, NULL);
+        }
+
+        return 0;
+}
+
+int32_t
+quota_readdir_cbk (call_frame_t *frame,
+                   void *cookie,
+                   xlator_t *this,
+                   int32_t op_ret,
+                   int32_t op_errno,
+                   gf_dirent_t *entries)
+{
+        char             contri_key [512] = {0, };
+        loc_t            loc;
+        int32_t          ret            = 0;
+        off_t            offset         = 0;
+        int32_t          count          = 0;
+        dict_t          *dict           = NULL;
+        quota_local_t   *local          = NULL;
+        gf_dirent_t     *entry          = NULL;
+        call_frame_t    *newframe      = NULL;
+
+        local = frame->local;
+
+        if (op_ret == -1) {
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "readdir failed %s", strerror (op_errno));
+                local->err = -1;
+
+                release_lock_on_dirty_inode (frame, NULL, this, 0, 0);
+
+                return 0;
+        } else if (op_ret == 0) {
+                get_dirty_inode_size (frame, this);
+
+                return 0;
+        }
+
+        local->dentry_child_count =  0;
+
+        list_for_each_entry (entry, (&entries->list), list) {
+              gf_log (this->name, GF_LOG_INFO, "entry  = %s", entry->d_name);
+
+              if ((!strcmp (entry->d_name, ".")) || (!strcmp (entry->d_name, ".."))) {
+                      gf_log (this->name, GF_LOG_INFO, "entry  = %s", entry->d_name);
+                      continue;
+              }
+              count++;
+        }
+
+        local->frame = frame;
+
+        list_for_each_entry (entry, (&entries->list), list) {
+                gf_log (this->name, GF_LOG_INFO, "entry  = %s", entry->d_name);
+
+                if ((!strcmp (entry->d_name, ".")) || (!strcmp (entry->d_name, ".."))) {
+                        gf_log (this->name, GF_LOG_INFO, "entry  = %s", entry->d_name);
+                        offset = entry->d_off;
+                        continue;
+                }
+
+                ret = loc_fill_from_name (this, &loc, &local->loc,
+                                          entry->d_ino, entry->d_name);
+                if (ret < 0)
+                        goto out;
+
+                newframe = copy_frame (frame);
+                if (!newframe) {
+                        ret = -1;
+                        goto out;
+                }
+
+                newframe->local = local;
+
+                dict = dict_new ();
+                if (!dict) {
+                        ret = -1;
+                        goto out;
+                }
+
+                GET_CONTRI_KEY (contri_key, local->loc.inode->gfid, ret);
+                if (ret < 0)
+                        goto out;
+
+                ret = dict_set_int64 (dict, contri_key, 0);
+                if (ret)
+                        goto out;
+
+                QUOTA_SAFE_INCREMENT (&local->lock, local->dentry_child_count);
+
+                STACK_WIND (newframe,
+                            get_child_contribution,
+                            FIRST_CHILD(this),
+                            FIRST_CHILD(this)->fops->lookup,
+                            &loc, dict);
+
+                offset = entry->d_off;
+
+                loc_wipe (&loc);
+
+        out:
+                if (dict) {
+                        dict_unref (dict);
+                        dict = NULL;
+                }
+
+                if (ret) {
+                        LOCK (&local->lock);
+                        {
+                                if (local->dentry_child_count == 0)
+                                        local->err = -1;
+                                else
+                                        local->err = -2;
+                        }
+                        UNLOCK (&local->lock);
+
+                        if (newframe) {
+                                newframe->local = NULL;
+
+                                QUOTA_STACK_DESTROY (newframe, this);
+                        }
+
+                        break;
+                }
+        }
+        gf_log (this->name, GF_LOG_INFO, "offset before =%lu",local->d_off);
+        local->d_off +=offset;
+        gf_log (this->name, GF_LOG_INFO, "offset after = %lu",local->d_off);
+
+        if (ret)
+                release_lock_on_dirty_inode (frame, NULL, this, 0, 0);
+
+        else if (count == 0 )
+                get_dirty_inode_size (frame, this);
+
+        return 0;
+}
+
+int32_t
+quota_dirty_inode_readdir (call_frame_t *frame,
+                           void *cookie,
+                           xlator_t *this,
+                           int32_t op_ret,
+                           int32_t op_errno,
+                           fd_t *fd)
+{
+        quota_local_t *local = NULL;
+
+        if (op_ret == -1) {
+                local->err = -1;
+                release_lock_on_dirty_inode (frame, NULL, this, 0, 0);
+                return 0;
+        }
+
+        local = frame->local;
+
+        if (local->fd == NULL)
+                local->fd = fd_ref (fd);
+
+        STACK_WIND (frame,
+                    quota_readdir_cbk,
+                    FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->readdir,
+                    local->fd, READDIR_BUF, local->d_off);
+
+        return 0;
+}
+
+int32_t
+check_if_still_dirty (call_frame_t *frame,
+                      void *cookie,
+                      xlator_t *this,
+                      int32_t op_ret,
+                      int32_t op_errno,
+                      inode_t *inode,
+                      struct iatt *buf,
+                      dict_t *dict,
+                      struct iatt *postparent)
+{
+        int8_t           dirty          = -1;
+        int32_t          ret            = -1;
+        fd_t            *fd             = NULL;
+        quota_local_t   *local          = NULL;
+        marker_conf_t   *priv           = NULL;
+
+        local = frame->local;
+
+        if (op_ret == -1) {
+                gf_log (this->name, GF_LOG_ERROR, "failed to get "
+                        "the dirty xattr for %s", local->loc.path);
+                goto err;
+        }
+
+        priv = this->private;
+
+        if (!dict) {
+              ret = -1;
+              goto err;
+        }
+
+        ret = dict_get_int8 (dict, priv->dirty_key, &dirty);
+        if (ret)
+                goto err;
+
+        //the inode is not dirty anymore
+        if (dirty == 0) {
+                release_lock_on_dirty_inode (frame, NULL, this, 0, 0);
+
+                return 0;
+        }
+
+        fd = fd_create (local->loc.inode, frame->root->pid);
+
+        local->d_off = 0;
+
+        STACK_WIND(frame,
+                   quota_dirty_inode_readdir,
+                   FIRST_CHILD(this),
+                   FIRST_CHILD(this)->fops->opendir,
+                   &local->loc, fd);
+
+        ret = 0;
+
+err:
+        if (op_ret == -1 || ret == -1) {
+                local->err = -1;
+                release_lock_on_dirty_inode (frame, NULL, this, 0, 0);
+        }
+
+        return 0;
+}
+
+int32_t
+get_dirty_xattr (call_frame_t *frame, void *cookie,
+                 xlator_t *this, int32_t op_ret, int32_t op_errno)
+{
+        int32_t         ret             = -1;
+        dict_t          *xattr_req      = NULL;
+        quota_local_t   *local          = NULL;
+        marker_conf_t   *priv           = NULL;
+
+        if (op_ret == -1) {
+                dirty_inode_updation_done (frame, NULL, this, 0, 0);
+                return 0;
+        }
+
+        priv = (marker_conf_t *) this->private;
+
+        local = frame->local;
+
+        xattr_req = dict_new ();
+        if (xattr_req == NULL) {
+                ret = -1;
+                goto err;
+        }
+
+        ret = dict_set_int8 (xattr_req, priv->dirty_key, 0);
+        if (ret)
+                goto err;
+
+        STACK_WIND (frame,
+                    check_if_still_dirty,
+                    FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->lookup,
+                    &local->loc,
+                    xattr_req);
+        ret = 0;
+
+err:
+        if (ret) {
+                local->err = -1;
+                release_lock_on_dirty_inode(frame, NULL, this, 0, 0);
+        }
+
+        if (xattr_req)
+                dict_unref (xattr_req);
+
+        return 0;
+}
+
+int32_t
+update_dirty_inode (xlator_t *this,
+                    loc_t *loc,
+                    quota_inode_ctx_t *ctx,
+                    inode_contribution_t *contribution)
+{
+        int32_t          ret    = -1;
+        quota_local_t   *local  = NULL;
+        struct gf_flock  lock;
+        call_frame_t    *frame  = NULL;
+
+        frame = create_frame (this, this->ctx->pool);
+        if (frame == NULL) {
+                ret = -1;
+                goto out;
+        }
+
+        local = quota_local_new ();
+        if (local == NULL)
+                goto fr_destroy;
+
+        frame->local = local;
+
+        ret = loc_copy (&local->loc, loc);
+        if (ret < 0)
+                goto fr_destroy;
+
+        local->ctx = ctx;
+
+        local->contri = contribution;
+
+        frame->root->lk_owner = cn++;
+
+        lock.l_type = F_WRLCK;
+        lock.l_whence = SEEK_SET;
+        lock.l_start = 0;
+        lock.l_len = 0;
+
+         STACK_WIND (frame,
+                     get_dirty_xattr,
+                     FIRST_CHILD(this),
+                     FIRST_CHILD(this)->fops->inodelk,
+                     this->name, &local->loc, F_SETLKW, &lock);
+
+fr_destroy:
+         QUOTA_STACK_DESTROY (frame, this);
+out:
+
+        return 0;
+}
+
+
+int32_t
+quota_inode_creation_done (call_frame_t *frame, void *cookie, xlator_t *this,
+                           int32_t op_ret, int32_t op_errno)
+{
+        quota_local_t *local = NULL;
+
+        if (frame == NULL)
+                return 0;
+
+        local = frame->local;
+
+        QUOTA_STACK_DESTROY (frame, this);
+
+        return 0;
+}
+
+int32_t
+create_dirty_xattr (call_frame_t *frame, void *cookie, xlator_t *this,
+                    int32_t op_ret, int32_t op_errno, dict_t *dict)
+{
+        int32_t          ret       = -1;
+        dict_t          *newdict   = NULL;
+        quota_local_t   *local     = NULL;
+        marker_conf_t   *priv      = NULL;
+
+        if (op_ret == -1 && op_errno == ENOTCONN)
+                goto err;
+
+        local = frame->local;
+
+        priv = (marker_conf_t *) this->private;
+
+        if (local->loc.inode->ia_type == IA_IFDIR) {
+                newdict = dict_new ();
+                if (!newdict)
+                        goto err;
+
+                ret = dict_set_int8 (newdict, priv->dirty_key, 0);
+                if (ret == -1)
+                        goto err;
+
+                STACK_WIND (frame, quota_inode_creation_done,
+                            FIRST_CHILD(this),
+                            FIRST_CHILD(this)->fops->setxattr,
+                            &local->loc, newdict, 0);
+        } else
+                quota_inode_creation_done (frame, NULL, this, 0, 0);
+
+        ret = 0;
+
+err:
+        if (ret == -1)
+                quota_inode_creation_done (frame, NULL, this, -1, 0);
+
+        if (newdict)
+                dict_unref (newdict);
+
+        return 0;
+}
+
+
+int32_t
+quota_set_inode_xattr (xlator_t *this, loc_t *loc)
+{
+        int32_t               ret       = 0;
+        int64_t              *value     = NULL;
+        int64_t              *size      = NULL;
+        dict_t               *dict      = NULL;
+        char                  key[512]  = {0, };
+        call_frame_t         *frame     = NULL;
+        quota_local_t        *local     = NULL;
+        marker_conf_t        *priv      = NULL;
+        quota_inode_ctx_t    *ctx       = NULL;
+        inode_contribution_t *contri    = NULL;
+
+        if (loc == NULL || this == NULL)
+                return 0;
+
+        priv = (marker_conf_t *) this->private;
+
+        ret = quota_inode_ctx_get (loc->inode, this, &ctx);
+        if (ret < 0) {
+                ctx = quota_inode_ctx_new (loc->inode, this);
+                if (ctx == NULL) {
+                        gf_log (this->name, GF_LOG_WARNING,
+                                "quota_inode_ctx_new failed");
+                        ret = -1;
+                        goto out;
+                }
+        }
+
+        dict = dict_new ();
+        if (!dict)
+                goto out;
+
+        if (loc->inode->ia_type == IA_IFDIR) {
+                QUOTA_ALLOC_OR_GOTO (size, int64_t, ret, err);
+                ret = dict_set_bin (dict, priv->size_key, size, 8);
+                if (ret < 0)
+                        goto free_size;
+        }
+
+        //if '/' then dont set contribution xattr
+        if (strcmp (loc->path, "/") == 0)
+                goto wind;
+
+        contri = add_new_contribution_node (this, ctx, loc);
+        if (contri == NULL)
+                goto err;
+
+        QUOTA_ALLOC_OR_GOTO (value, int64_t, ret, err);
+        GET_CONTRI_KEY (key, loc->parent->gfid, ret);
+
+        ret = dict_set_bin (dict, key, value, 8);
+        if (ret < 0)
+                goto free_value;
+
+wind:
+        frame = create_frame (this, this->ctx->pool);
+        if (!frame) {
+                ret = -1;
+                goto err;
+        }
+
+        local = quota_local_new ();
+        if (local == NULL)
+                goto free_size;
+
+        local->ctx = ctx;
+
+        local->contri = contri;
+
+        ret = loc_copy (&local->loc, loc);
+        if (ret < 0)
+                quota_local_unref (this, local);
+
+        frame->local = local;
+
+        STACK_WIND (frame, create_dirty_xattr, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->xattrop, &local->loc,
+                    GF_XATTROP_ADD_ARRAY64, dict);
+        ret = 0;
+
+free_size:
+        if (ret < 0)
+                GF_FREE (size);
+
+free_value:
+        if (ret < 0)
+                GF_FREE (value);
+
+err:
+        dict_unref (dict);
+
+out:
+        if (ret < 0)
+                quota_inode_creation_done (NULL, NULL, this, -1, 0);
+
+        return 0;
+}
+
+
+int32_t
+get_parent_inode_local (xlator_t *this, quota_local_t *local)
+{
+        uint32_t           ret;
+        quota_inode_ctx_t *ctx = NULL;
+
+        loc_wipe (&local->loc);
+
+        loc_copy (&local->loc, &local->parent_loc);
+
+        loc_wipe (&local->parent_loc);
+
+        quota_inode_loc_fill (NULL, local->loc.parent, &local->parent_loc);
+
+        ret = quota_inode_ctx_get (local->loc.inode, this, &ctx);
+        if (ret < 0)
+                return -1;
+
+        local->ctx = ctx;
+
+        local->contri = (inode_contribution_t *) ctx->contribution_head.next;
+
+        return 0;
+}
+
+
+int32_t
+xattr_updation_done (call_frame_t *frame,
+                     void *cookie,
+                     xlator_t *this,
+                     int32_t op_ret,
+                     int32_t op_errno,
+                     dict_t *dict)
+{
+        QUOTA_STACK_DESTROY (frame, this);
+        return 0;
+}
+
+
+int32_t
+quota_inodelk_cbk (call_frame_t *frame, void *cookie,
+                   xlator_t *this, int32_t op_ret, int32_t op_errno)
+{
+        int32_t         ret  = 0;
+        quota_local_t  *local = NULL;
+
+        trap ();
+
+        local = frame->local;
+
+        if (op_ret == -1 || local->err) {
+                gf_log (this->name, GF_LOG_INFO, "lock setting failed");
+                xattr_updation_done (frame, NULL, this, 0, 0, NULL);
+
+                return 0;
+        }
+
+        gf_log (this->name, GF_LOG_DEBUG,
+                "inodelk released on %s", local->parent_loc.path);
+
+        if (strcmp (local->parent_loc.path, "/") == 0) {
+                xattr_updation_done (frame, NULL, this, 0, 0, NULL);
+        } else {
+                ret = get_parent_inode_local (this, local);
+                if (ret < 0) {
+                        xattr_updation_done (frame, NULL, this, 0, 0, NULL);
+                        goto out;
+                }
+
+                get_lock_on_parent (frame, this);
+        }
+out:
+        return 0;
+}
+
+
+//now release lock on the parent inode
+int32_t
+quota_release_parent_lock (call_frame_t *frame, void *cookie,
+                           xlator_t *this, int32_t op_ret,
+                           int32_t op_errno)
+{
+        int32_t            ret      = 0;
+        struct gf_flock    lock;
+        quota_local_t     *local    = NULL;
+        quota_inode_ctx_t *ctx      = NULL;
+
+        trap ();
+
+        local = frame->local;
+
+        ret = quota_inode_ctx_get (local->parent_loc.inode, this, &ctx);
+        if (ret < 0)
+                goto wind;
+
+        LOCK (&ctx->lock);
+        {
+                ctx->dirty = 0;
+        }
+        UNLOCK (&ctx->lock);
+
+wind:
+        lock.l_type   = F_UNLCK;
+        lock.l_whence = SEEK_SET;
+        lock.l_start  = 0;
+        lock.l_len    = 0;
+        lock.l_pid    = 0;
+
+        STACK_WIND (frame,
+                    quota_inodelk_cbk,
+                    FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->inodelk,
+                    this->name, &local->parent_loc,
+                    F_SETLKW, &lock);
+
+        return 0;
+}
+
+
+int32_t
+quota_mark_undirty (call_frame_t *frame,
+                    void *cookie,
+                    xlator_t *this,
+                    int32_t op_ret,
+                    int32_t op_errno,
+                    dict_t *dict)
+{
+        int32_t            ret          = -1;
+        int64_t           *size         = NULL;
+        dict_t            *newdict      = NULL;
+        quota_local_t     *local        = NULL;
+        quota_inode_ctx_t *ctx          = NULL;
+        marker_conf_t     *priv         = NULL;
+
+        trap ();
+
+        local = frame->local;
+
+        if (op_ret == -1) {
+                gf_log (this->name, GF_LOG_WARNING, "%s occurred while"
+                        " updating the size of %s", strerror (op_errno),
+                        local->parent_loc.path);
+
+                goto err;
+        }
+
+        priv = this->private;
+
+       //update the size of the parent inode
+        if (dict != NULL) {
+                ret = quota_inode_ctx_get (local->parent_loc.inode, this, &ctx);
+                if (ret < 0)
+                        goto err;
+
+                ret = dict_get_bin (dict, priv->size_key, (void **) &size);
+                if (ret < 0)
+                        goto err;
+
+                LOCK (&ctx->lock);
+                {
+                        if (size)
+                                ctx->size = ntoh64 (*size);
+                }
+                UNLOCK (&ctx->lock);
+        }
+
+        newdict = dict_new ();
+
+        if (!newdict)
+                goto err;
+
+        ret = dict_set_int8 (newdict, priv->dirty_key, 0);
+
+        if (ret == -1)
+                goto err;
+
+        STACK_WIND (frame, quota_release_parent_lock,
+                    FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->setxattr,
+                    &local->parent_loc, newdict, 0);
+
+        ret = 0;
+err:
+        if (op_ret == -1 || ret == -1) {
+                local->err = 1;
+
+                quota_release_parent_lock (frame, NULL, this, 0, 0);
+        }
+
+        if (newdict)
+                dict_unref (newdict);
+
+        return 0;
+}
+
+
+int32_t
+quota_update_parent_size (call_frame_t *frame,
+                          void *cookie,
+                          xlator_t *this,
+                          int32_t op_ret,
+                          int32_t op_errno,
+                          dict_t *dict)
+{
+        int64_t             *size       = NULL;
+        int32_t              ret        = -1;
+        dict_t              *newdict    = NULL;
+        marker_conf_t       *priv       = NULL;
+        quota_local_t       *local      = NULL;
+        quota_inode_ctx_t   *ctx        = NULL;
+
+        trap ();
+
+        local = frame->local;
+
+        if (op_ret == -1) {
+                gf_log (this->name, GF_LOG_ERROR, "%s", strerror (op_errno));
+
+                goto err;
+        }
+
+        priv = this->private;
+
+        if (dict == NULL)
+                goto err;
+
+        ret = quota_inode_ctx_get (local->parent_loc.inode, this, &ctx);
+        if (ret < 0)
+                goto err;
+
+        newdict = dict_new ();
+        if (!newdict) {
+                ret = -1;
+                goto err;
+        }
+
+        QUOTA_ALLOC_OR_GOTO (size, int64_t, ret, err);
+
+        *size = ntoh64 (local->delta);
+
+        ret = dict_set_bin (newdict, priv->size_key, size, 8);
+        if (ret < 0)
+                goto err;
+
+        STACK_WIND (frame,
+                    quota_mark_undirty,
+                    FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->xattrop,
+                    &local->parent_loc,
+                    GF_XATTROP_ADD_ARRAY64,
+                    newdict);
+        ret = 0;
+err:
+        if (op_ret == -1 || ret < 0) {
+                local->err = 1;
+                quota_release_parent_lock (frame, NULL, this, 0, 0);
+        }
+
+        if (dict)
+                dict_unref (newdict);
+
+        return 0;
+}
+
+int32_t
+quota_update_inode_contribution (call_frame_t *frame, void *cookie,
+                                 xlator_t *this, int32_t op_ret,
+                                 int32_t op_errno, inode_t *inode,
+                                 struct iatt *buf, dict_t *dict,
+                                 struct iatt *postparent)
+{
+        int32_t               ret             = -1;
+        int64_t              *size            = NULL;
+        int64_t              *contri          = NULL;
+        int64_t              *delta           = NULL;
+        char                  contri_key [512] = {0, };
+        dict_t               *newdict         = NULL;
+        quota_local_t        *local           = NULL;
+        quota_inode_ctx_t    *ctx             = NULL;
+        marker_conf_t        *priv            = NULL;
+        inode_contribution_t *contribution    = NULL;
+
+        trap ();
+
+        local = frame->local;
+
+        if (op_ret == -1) {
+                gf_log (this->name, GF_LOG_ERROR, "failed to get size and "
+                        "contribution with %s error", strerror (op_errno));
+                goto err;
+        }
+
+        priv = this->private;
+
+        ctx = local->ctx;
+        contribution = local->contri;
+
+        //prepare to update size & contribution of the inode
+        GET_CONTRI_KEY (contri_key, contribution->gfid, ret);
+        if (ret == -1)
+                goto err;
+
+        LOCK (&ctx->lock);
+        {
+                if (local->loc.inode->ia_type == IA_IFDIR ) {
+                        ret = dict_get_bin (dict, priv->size_key,
+                                            (void **) &size);
+                        if (ret < 0)
+                                goto unlock;
+
+                        ctx->size = ntoh64 (*size);
+                } else
+                        ctx->size = buf->ia_size;
+
+                ret = dict_get_bin (dict, contri_key, (void **) &contri);
+                if (ret < 0)
+                        contribution->contribution = 0;
+                else
+                        contribution->contribution = ntoh64 (*contri);
+
+                ret = 0;
+        }
+unlock:
+        UNLOCK (&ctx->lock);
+
+        if (ret < 0)
+                goto err;
+
+        newdict = dict_new ();
+        if (newdict == NULL) {
+                ret = -1;
+                goto err;
+        }
+
+        local->delta = ctx->size - contribution->contribution;
+
+        QUOTA_ALLOC_OR_GOTO (delta, int64_t, ret, err);
+
+        *delta = hton64 (local->delta);
+
+        ret = dict_set_bin (newdict, contri_key, delta, 8);
+        if (ret < 0) {
+                ret = -1;
+                goto err;
+        }
+
+        STACK_WIND (frame,
+                    quota_update_parent_size,
+                    FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->xattrop,
+                    &local->loc,
+                    GF_XATTROP_ADD_ARRAY64,
+                    newdict);
+        ret = 0;
+
+err:
+        if (op_ret == -1 || ret < 0) {
+                local->err = 1;
+
+                quota_release_parent_lock (frame, NULL, this, 0, 0);
+        }
+
+        if (newdict)
+                dict_unref (newdict);
+
+        return 0;
+}
+
+int32_t
+quota_fetch_child_size_and_contri (call_frame_t *frame, void *cookie,
+                                   xlator_t *this, int32_t op_ret,
+                                   int32_t op_errno)
+{
+        int32_t            ret        = -1;
+        char               contri_key [512] = {0, };
+        dict_t            *newdict    = NULL;
+        quota_local_t     *local      = NULL;
+        marker_conf_t     *priv       = NULL;
+        quota_inode_ctx_t *ctx        = NULL;
+
+        trap ();
+
+        local = frame->local;
+
+        if (op_ret == -1) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "%s couldnt mark dirty", local->parent_loc.path);
+                goto err;
+        }
+
+        gf_log (this->name, GF_LOG_DEBUG, "%s marked dirty", local->parent_loc.path);
+
+        priv = this->private;
+
+        //update parent ctx
+        ret = quota_inode_ctx_get (local->parent_loc.inode, this, &ctx);
+        if (ret == -1)
+                goto err;
+
+        LOCK (&ctx->lock);
+        {
+                ctx->dirty = 1;
+        }
+        UNLOCK (&ctx->lock);
+
+        newdict = dict_new ();
+        if (newdict == NULL)
+                goto err;
+
+        if (local->loc.inode->ia_type == IA_IFDIR) {
+                ret = dict_set_int64 (newdict, priv->size_key, 0);
+        }
+
+        GET_CONTRI_KEY (contri_key, local->contri->gfid, ret);
+        if (ret < 0)
+                goto err;
+
+        ret = dict_set_int64 (newdict, contri_key, 0);
+
+        STACK_WIND (frame, quota_update_inode_contribution, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->lookup, &local->loc, newdict);
+
+        ret = 0;
+
+err:
+        if (op_ret == -1 || ret == -1) {
+                local->err = 1;
+
+                quota_release_parent_lock (frame, NULL, this, 0, 0);
+        }
+
+        if (newdict)
+                dict_unref (newdict);
+
+        return 0;
+}
+
+int32_t
+quota_markdirty (call_frame_t *frame, void *cookie,
+                 xlator_t *this, int32_t op_ret, int32_t op_errno)
+{
+        int32_t        ret      = -1;
+        dict_t        *dict     = NULL;
+        quota_local_t *local    = NULL;
+        marker_conf_t *priv     = NULL;
+
+        local = frame->local;
+
+        if (op_ret == -1){
+                gf_log (this->name, GF_LOG_ERROR,
+                        "lock setting failed %s", strerror (op_errno));
+
+                local->err = 1;
+
+                quota_inodelk_cbk (frame, NULL, this, 0, 0);
+
+                return 0;
+        }
+
+        gf_log (this->name, GF_LOG_TRACE,
+                "inodelk succeeded on  %s", local->parent_loc.path);
+
+        priv = this->private;
+
+        dict = dict_new ();
+        if (!dict) {
+                ret = -1;
+                goto err;
+        }
+
+        ret = dict_set_int8 (dict, priv->dirty_key, 1);
+        if (ret == -1)
+                goto err;
+
+        STACK_WIND (frame, quota_fetch_child_size_and_contri,
+                    FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->setxattr,
+                    &local->parent_loc, dict, 0);
+
+        ret = 0;
+err:
+        if (ret == -1) {
+                local->err = 1;
+
+                quota_release_parent_lock (frame, NULL, this, 0, 0);
+        }
+
+        if (dict)
+                dict_unref (dict);
+
+        return 0;
+}
+
+
+int32_t
+get_lock_on_parent (call_frame_t *frame, xlator_t *this)
+{
+        struct gf_flock  lock;
+        quota_local_t   *local = NULL;
+
+        GF_VALIDATE_OR_GOTO ("marker", frame, fr_destroy);
+
+        local = frame->local;
+
+        lock.l_len    = 0;
+        lock.l_start  = 0;
+        lock.l_type   = F_WRLCK;
+        lock.l_whence = SEEK_SET;
+
+        STACK_WIND (frame,
+                    quota_markdirty,
+                    FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->inodelk,
+                    this->name, &local->parent_loc, F_SETLKW, &lock);
+
+        return 0;
+
+fr_destroy:
+        QUOTA_STACK_DESTROY (frame, this);
+
+        return 0;
+}
+
+
+int
+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;
+        quota_local_t   *local  = NULL;
+
+        frame = create_frame (this, this->ctx->pool);
+        if (frame == NULL)
+                goto err;
+
+        frame->root->lk_owner = cn++;
+
+        local = quota_local_new ();
+        if (local == NULL)
+                goto fr_destroy;
+
+        frame->local = local;
+
+        ret = loc_copy (&local->loc, loc);
+        if (ret < 0)
+                goto local_unref;
+
+        ret = quota_inode_loc_fill (NULL, local->loc.parent,
+                                    &local->parent_loc);
+        if (ret < 0)
+                goto local_unref;
+
+        local->ctx = ctx;
+        local->contri = contri;
+
+        get_lock_on_parent (frame, this);
+
+        return 0;
+
+local_unref:
+        quota_local_unref (this, local);
+
+fr_destroy:
+        QUOTA_STACK_DESTROY (frame, this);
+
+err:
+        return -1;
+}
+
+
+int
+initiate_quota_txn (xlator_t *this, loc_t *loc)
+{
+        int32_t            ret    = -1;
+        quota_inode_ctx_t *ctx    = NULL;
+        inode_contribution_t *contribution = NULL;
+
+        trap ();
+
+        VALIDATE_OR_GOTO (loc, out);
+
+        ret = quota_inode_ctx_get (loc->inode, this, &ctx);
+        if (ret == -1) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "inode ctx get failed, aborting quota txn");
+                ret = -1;
+                goto out;
+        }
+
+        contribution = get_contribution_node (loc->parent, ctx);
+        if (contribution == NULL)
+                goto out;
+
+        start_quota_txn (this, loc, ctx, contribution);
+out:
+        return 0;
+}
+
+
+int32_t
+validate_inode_size_contribution (xlator_t *this,
+                                  loc_t *loc,
+                                  quota_inode_ctx_t *ctx,
+                                  inode_contribution_t *contribution)
+{
+        if (ctx->size != contribution->contribution)
+                initiate_quota_txn (this, loc);
+
+        return 0;
+}
+
+
+int32_t
+inspect_directory_xattr (xlator_t *this,
+                        loc_t *loc,
+                        dict_t *dict,
+                        struct iatt buf)
+{
+        int32_t                  ret            = 0;
+        int8_t                   dirty          = -1;
+        int64_t                 *size           = NULL;
+        int64_t                 *contri         = NULL;
+        char                     contri_key [512] = {0, };
+        marker_conf_t           *priv           = NULL;
+        quota_inode_ctx_t       *ctx            = NULL;
+        inode_contribution_t    *contribution   = NULL;
+
+        priv = this->private;
+
+        ret = quota_inode_ctx_get (loc->inode, this, &ctx);
+        if (ret < 0) {
+                ctx = quota_inode_ctx_new (loc->inode, this);
+                if (ctx == NULL) {
+                        gf_log (this->name, GF_LOG_WARNING,
+                                "quota_inode_ctx_new failed");
+                        ret = -1;
+                        goto out;
+                }
+        }
+
+        ret = dict_get_bin (dict, priv->size_key, (void **) &size);
+        if (ret < 0)
+                goto out;
+
+        ret = dict_get_int8 (dict, priv->dirty_key, &dirty);
+        if (ret < 0)
+                goto out;
+
+        if (strcmp (loc->path, "/") != 0) {
+                contribution = add_new_contribution_node (this, ctx, loc);
+                if (contribution == NULL) {
+                        gf_log (this->name, GF_LOG_DEBUG,
+                                "cannot add a new contributio node");
+                        goto out;
+                }
+
+                GET_CONTRI_KEY (contri_key, contribution->gfid, ret);
+                if (ret < 0)
+                        goto out;
+
+                ret = dict_get_bin (dict, contri_key, (void **) &contri);
+                if (ret < 0)
+                        goto out;
+
+                contribution->contribution = ntoh64 (*contri);
+        }
+
+        ctx->size = ntoh64 (*size);
+
+        ctx->dirty = dirty;
+        if (ctx->dirty == 1)
+                update_dirty_inode (this, loc, ctx, contribution);
+
+        ret = 0;
+out:
+        if (ret)
+                quota_set_inode_xattr (this, loc);
+
+        return 0;
+}
+
+int32_t
+inspect_file_xattr (xlator_t *this,
+                    loc_t *loc,
+                    dict_t *dict,
+                    struct iatt buf)
+{
+        int32_t               ret          = -1;
+        uint64_t              contri_int   = 0;
+        int64_t              *contri_ptr   = NULL;
+        char                  contri_key [512] = {0, };
+        marker_conf_t        *priv         = NULL;
+        quota_inode_ctx_t    *ctx          = NULL;
+        inode_contribution_t *contribution = NULL;
+
+        priv = this->private;
+
+        ret = quota_inode_ctx_get (loc->inode, this, &ctx);
+        if (ret < 0) {
+                ctx = quota_inode_ctx_new (loc->inode, this);
+                if (ctx == NULL) {
+                        gf_log (this->name, GF_LOG_WARNING,
+                                "quota_inode_ctx_new failed");
+                        ret = -1;
+                        goto out;
+                }
+        }
+
+        contribution = add_new_contribution_node (this, ctx, loc);
+        if (contribution == NULL)
+                goto out;
+
+        LOCK (&ctx->lock);
+        {
+                ctx->size = buf.ia_size;
+        }
+        UNLOCK (&ctx->lock);
+
+        list_for_each_entry (contribution, &ctx->contribution_head, contri_list) {
+                GET_CONTRI_KEY (contri_key, contribution->gfid, ret);
+                if (ret < 0)
+                        continue;
+
+                ret = dict_get_bin (dict, contri_key, (void **) &contri_int);
+                if (ret == 0) {
+                        contri_ptr = (int64_t *) contri_int;
+
+                        contribution->contribution = ntoh64 (*contri_ptr);
+
+                        ret = validate_inode_size_contribution
+                                (this, loc, ctx, contribution);
+                } else
+                      initiate_quota_txn (this, loc);
+        }
+
+out:
+        return ret;
+}
+
+int32_t
+quota_xattr_state (xlator_t *this,
+                   loc_t *loc,
+                   dict_t *dict,
+                   struct iatt buf)
+{
+        if (buf.ia_type == IA_IFREG ||
+            buf.ia_type == IA_IFLNK) {
+                k ++;
+                inspect_file_xattr (this, loc, dict, buf);
+        } else if (buf.ia_type == IA_IFDIR)
+                inspect_directory_xattr (this, loc, dict, buf);
+
+        return 0;
+}
+
+int32_t
+quota_req_xattr (xlator_t *this,
+                 loc_t *loc,
+                 dict_t *dict)
+{
+        int32_t               ret       = -1;
+        marker_conf_t        *priv      = NULL;
+
+        GF_VALIDATE_OR_GOTO ("marker", this, out);
+        GF_VALIDATE_OR_GOTO ("marker", loc, out);
+        GF_VALIDATE_OR_GOTO ("marker", dict, out);
+
+        priv = this->private;
+
+        //if not "/" then request contribution
+        if (strcmp (loc->path, "/") == 0)
+                goto set_size;
+
+        ret = dict_set_contribution (this, dict, loc);
+        if (ret == -1)
+                goto out;
+
+set_size:
+        ret = dict_set_uint64 (dict, priv->size_key, 0);
+        if (ret < 0) {
+                ret = -1;
+                goto out;
+        }
+
+        ret = dict_set_int8 (dict, priv->dirty_key, 0);
+        if (ret < 0) {
+                ret = -1;
+                goto out;
+        }
+
+        ret = 0;
+
+out:
+        return ret;
+}
+
+
+int32_t
+quota_removexattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                       int32_t op_ret, int32_t op_errno)
+{
+        QUOTA_STACK_DESTROY (frame, this);
+
+        return 0;
+}
+
+int32_t
+quota_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this,
+                         int32_t op_ret, int32_t op_errno, dict_t *dict)
+{
+        int32_t          ret        = 0;
+        char             contri_key [512] = {0, };
+        quota_local_t   *local      = NULL;
+
+        local = (quota_local_t *) frame->local;
+
+        if (local->hl_count > 1) {
+                GET_CONTRI_KEY (contri_key, local->contri->gfid, ret);
+
+                STACK_WIND (frame, quota_removexattr_cbk, FIRST_CHILD(this),
+                           FIRST_CHILD(this)->fops->removexattr,
+                           &local->loc, contri_key);
+                ret = 0;
+        }
+
+        if (strcmp (local->parent_loc.path, "/") != 0) {
+                get_parent_inode_local (this, local);
+
+                start_quota_txn (this, &local->loc, local->ctx, local->contri);
+        }
+
+        quota_local_unref (this, local);
+
+        return 0;
+}
+
+int32_t
+reduce_parent_size (xlator_t *this, loc_t *loc)
+{
+        int32_t                  ret               = -1;
+        int64_t                 *size              = NULL;
+        dict_t                  *dict              = NULL;
+        call_frame_t            *frame             = NULL;
+        marker_conf_t           *priv              = NULL;
+        quota_local_t           *local             = NULL;
+        quota_inode_ctx_t       *ctx               = NULL;
+        inode_contribution_t    *contribution      = NULL;
+
+        GF_VALIDATE_OR_GOTO ("marker", this, out);
+        GF_VALIDATE_OR_GOTO ("marker", loc, out);
+
+        priv = this->private;
+
+        ret = quota_inode_ctx_get (loc->inode, this, &ctx);
+        if (ret < 0)
+                goto out;
+
+        contribution = get_contribution_node (loc->parent, ctx);
+        if (contribution == NULL)
+                goto out;
+
+        local = quota_local_new ();
+        if (local == NULL) {
+                ret = -1;
+                goto out;
+        }
+
+        ret = quota_inode_loc_fill ((const char *) loc->parent->gfid,
+                                    loc->parent, &local->parent_loc);
+        if (ret < 0)
+                goto free_local;
+
+        dict = dict_new ();
+        if (dict == NULL) {
+                ret = -1;
+                goto free_local;
+        }
+
+        QUOTA_ALLOC_OR_GOTO (size, int64_t, ret, free_local);
+
+        *size = hton64 (-contribution->contribution);
+
+        ret = dict_set_bin (dict, priv->size_key, size, 8);
+        if (ret < 0)
+                goto free_size;
+
+        frame = create_frame (this, this->ctx->pool);
+        if (!frame) {
+                ret = -1;
+                goto free_size;
+        }
+
+        frame->local = local;
+
+        STACK_WIND (frame, quota_inode_remove_done, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->xattrop, &local->parent_loc,
+                    GF_XATTROP_ADD_ARRAY64, dict);
+        ret = 0;
+
+free_size:
+        if (ret < 0)
+                GF_FREE (size);
+free_local:
+        if (ret < 0)
+                quota_local_unref (this, local);
+out:
+        dict_unref (dict);
+
+        return ret;
+}
+
+
+int32_t
+init_quota_priv (xlator_t *this)
+{
+        marker_conf_t  *priv    = NULL;
+
+        priv = this->private;
+
+        strcpy (volname, "quota");
+
+        gf_asprintf (&priv->size_key, QUOTA_XATTR_PREFIX
+                     ".%s.size", volname);
+
+        gf_asprintf (&priv->dirty_key, QUOTA_XATTR_PREFIX
+                     ".%s.dirty", volname);
+
+        return 0;
+}
+
+
+int32_t
+quota_rename_update_newpath (xlator_t *this, loc_t *loc, inode_t *inode)
+{
+        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", loc, out);
+        GF_VALIDATE_OR_GOTO ("marker", inode, out);
+
+        if (loc->inode == NULL)
+                loc->inode = inode_ref (inode);
+
+        ret = quota_inode_ctx_get (loc->inode, this, &ctx);
+        if (ret < 0)
+                goto out;
+
+        contribution = add_new_contribution_node (this, ctx, loc);
+        if (contribution == NULL) {
+                ret = -1;
+                goto out;
+        }
+
+        initiate_quota_txn (this, loc);
+out:
+        return ret;
+}
+
+int32_t
+quota_forget (xlator_t *this, quota_inode_ctx_t *ctx)
+{
+        inode_contribution_t *contri = NULL;
+        inode_contribution_t *next   = NULL;
+
+        GF_VALIDATE_OR_GOTO ("marker", this, out);
+        GF_VALIDATE_OR_GOTO ("marker", ctx, out);
+
+        list_for_each_entry_safe (contri, next, &ctx->contribution_head,
+                                  contri_list) {
+                list_del (&contri->contri_list);
+                GF_FREE (contri);
+        }
+
+        LOCK_DESTROY (&ctx->lock);
+        GF_FREE (ctx);
+out:
+        return 0;
+}
diff --git a/xlators/features/marker/src/marker-quota.h b/xlators/features/marker/src/marker-quota.h
new file mode 100644
index 00000000000..fdec1230fb8
--- /dev/null
+++ b/xlators/features/marker/src/marker-quota.h
@@ -0,0 +1,167 @@
+/*Copyright (c) 2008-2010 Gluster, Inc. <http://www.gluster.com>
+  This file is part of GlusterFS.
+
+  GlusterFS is free software; you can redistribute it and/or modify
+  it under the terms of the GNU 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 _MARKER_QUOTA_H
+#define _MARKER_QUOTA_H
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "marker.h"
+#include "xlator.h"
+#include "marker-mem-types.h"
+
+#define QUOTA_XATTR_PREFIX "trusted.glusterfs"
+#define CONTRIBUTION "contri"
+#define VOL_NAME volname
+#define CONTRI_KEY_MAX 512
+#define READDIR_BUF 4096
+
+char volname [40];
+
+#define QUOTA_STACK_DESTROY(_frame, _this)              \
+        do {                                            \
+                quota_local_t *_local = NULL;           \
+                _local = _frame->local;                 \
+                _frame->local = NULL;                   \
+                STACK_DESTROY (_frame->root);           \
+                quota_local_unref (_this, _local);      \
+                GF_FREE (_local);                       \
+        } while (0)
+
+
+#define QUOTA_ALLOC(var, type, ret)                     \
+        do {                                            \
+                var = GF_CALLOC (sizeof (type), 1,      \
+                                gf_marker_mt_##type);   \
+                if (!var) {                             \
+                        gf_log ("", GF_LOG_ERROR,       \
+                                "out of memory");       \
+                        ret = -1;                       \
+                }                                       \
+                ret = 0;                                \
+        } while (0);
+
+#define QUOTA_ALLOC_OR_GOTO(var, type, ret, label)      \
+        do {                                            \
+                var = GF_CALLOC (sizeof (type), 1,      \
+                                gf_marker_mt_##type);   \
+                if (!var) {                             \
+                        gf_log ("", GF_LOG_ERROR,       \
+                                "out of memory");       \
+                        ret = -1;                       \
+                        goto label;                     \
+                }                                       \
+                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, VOL_NAME,       \
+                                _gfid_unparsed);                        \
+        } while (0);
+
+#define QUOTA_SAFE_INCREMENT(lock, var)                 \
+        do {                                            \
+                LOCK (lock);                            \
+                        var ++;                         \
+                UNLOCK (lock);                          \
+        } while (0)
+
+#define QUOTA_SAFE_DECREMENT(lock, var)                 \
+        do {                                            \
+                LOCK (lock);                            \
+                        var --;                         \
+                UNLOCK (lock);                          \
+        } while (0)
+
+
+struct quota_inode_ctx {
+        int64_t size;
+        int8_t  dirty;
+        gf_lock_t lock;
+        struct list_head  contribution_head;
+};
+typedef struct quota_inode_ctx quota_inode_ctx_t;
+
+struct inode_contribution {
+        struct list_head contri_list;
+        int64_t  contribution;
+        uuid_t   gfid;
+};
+typedef struct inode_contribution inode_contribution_t;
+
+struct quota_local {
+        int64_t delta;
+        int64_t d_off;
+        int32_t err;
+        int32_t ref;
+        int64_t sum;
+        int32_t hl_count;
+        int32_t dentry_child_count;
+
+        fd_t   *fd;
+        call_frame_t *frame;
+        gf_lock_t lock;
+
+        loc_t   loc;
+        loc_t   parent_loc;
+
+        quota_inode_ctx_t    *ctx;
+        inode_contribution_t *contri;
+};
+typedef struct quota_local quota_local_t;
+
+int32_t cn;
+
+int32_t
+get_lock_on_parent (call_frame_t *, xlator_t *);
+
+int32_t
+quota_req_xattr (xlator_t *, loc_t *, dict_t *);
+
+int32_t
+init_quota_priv (xlator_t *);
+
+int32_t
+quota_xattr_state (xlator_t *, loc_t *, dict_t *, struct iatt);
+
+int32_t
+quota_set_inode_xattr (xlator_t *, loc_t *);
+
+int
+initiate_quota_txn (xlator_t *, loc_t *);
+
+int32_t
+quota_dirty_inode_readdir (call_frame_t *, void *, xlator_t *,
+                           int32_t, int32_t, fd_t *);
+
+int32_t
+reduce_parent_size (xlator_t *, loc_t *);
+
+int32_t
+quota_rename_update_newpath (xlator_t *, loc_t *, inode_t *);
+
+int32_t
+quota_forget (xlator_t *, quota_inode_ctx_t *);
+#endif
diff --git a/xlators/features/marker/src/marker.c b/xlators/features/marker/src/marker.c
index f6302247e5e..83824891235 100644
--- a/xlators/features/marker/src/marker.c
+++ b/xlators/features/marker/src/marker.c
@@ -26,6 +26,7 @@
 #include "libxlator.h"
 #include "marker.h"
 #include "marker-mem-types.h"
+#include "marker-quota.h"
 
 void
 fini (xlator_t *this);
@@ -33,6 +34,22 @@ fini (xlator_t *this);
 int32_t
 marker_start_setxattr (call_frame_t *, xlator_t *);
 
+marker_local_t *
+marker_local_ref (marker_local_t *local)
+{
+        GF_VALIDATE_OR_GOTO ("marker", local, err);
+
+        LOCK (&local->lock);
+        {
+                local->ref++;
+        }
+        UNLOCK (&local->lock);
+
+        return local;
+err:
+        return NULL;
+}
+
 int
 marker_loc_fill (loc_t *loc, inode_t *inode, inode_t *parent, char *path)
 {
@@ -139,8 +156,20 @@ marker_error_handler (xlator_t *this)
 }
 
 int32_t
-marker_free_local (marker_local_t *local)
+marker_local_unref (marker_local_t *local)
 {
+        if (local == NULL)
+                return -1;
+
+        LOCK (&local->lock);
+        {
+                local->ref--;
+        }
+        UNLOCK (&local->lock);
+
+        if (local->ref != 0)
+                goto out;
+
         loc_wipe (&local->loc);
 
         if (local->oplocal) {
@@ -148,7 +177,7 @@ marker_free_local (marker_local_t *local)
                 GF_FREE (local->oplocal);
         }
         GF_FREE (local);
-
+out:
         return 0;
 }
 
@@ -263,7 +292,7 @@ marker_setxattr_done (call_frame_t *frame)
 
         STACK_DESTROY (frame->root);
 
-        marker_free_local (local);
+        marker_local_unref (local);
 
         return 0;
 }
@@ -365,32 +394,30 @@ marker_create_frame (xlator_t *this, marker_local_t *local)
 }
 
 int32_t
-update_marks (xlator_t *this, marker_local_t *local, int32_t ret)
+update_marks (xlator_t *this, marker_local_t *local)
 {
-        if (ret == -1 || local->pid < 0)
-                marker_free_local (local);
-        else {
-                marker_gettimeofday (local);
+        marker_gettimeofday (local);
 
-                marker_create_frame (this, local);
-        }
+        marker_local_ref (local);
+
+        marker_create_frame (this, local);
 
         return 0;
 }
 
+
 int32_t
 marker_mkdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                    int32_t op_ret, int32_t op_errno, inode_t *inode,
                    struct iatt *buf, struct iatt *preparent,
                    struct iatt *postparent)
 {
-        int32_t             ret     = 0;
+        marker_conf_t      *priv    = NULL;
         marker_local_t     *local   = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR, "error occurred "
                         "while Creating a file %s", strerror (op_errno));
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -400,7 +427,19 @@ marker_mkdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         STACK_UNWIND_STRICT (mkdir, frame, op_ret, op_errno, inode,
                              buf, preparent, postparent);
 
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled & GF_QUOTA)
+                quota_set_inode_xattr (this, &local->loc);
+
+        if (priv->feature_enabled & GF_GSYNC)
+                update_marks (this, local);
+
+out:
+        marker_local_unref (local);
 
         return 0;
 }
@@ -431,19 +470,19 @@ err:
         return 0;
 }
 
+
 int32_t
 marker_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                     int32_t op_ret, int32_t op_errno, fd_t *fd, inode_t *inode,
                     struct iatt *buf, struct iatt *preparent,
                     struct iatt *postparent)
 {
-        int32_t             ret     = 0;
         marker_local_t     *local   = NULL;
+        marker_conf_t      *priv    = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR, "error occurred "
                         "while Creating a file %s", strerror (op_errno));
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -453,7 +492,19 @@ marker_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         STACK_UNWIND_STRICT (create, frame, op_ret, op_errno, fd, inode, buf,
                              preparent, postparent);
 
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled & GF_QUOTA)
+                quota_set_inode_xattr (this, &local->loc);
+
+        if (priv->feature_enabled & GF_GSYNC)
+                update_marks (this, local);
+
+out:
+        marker_local_unref (local);
 
         return 0;
 }
@@ -484,18 +535,18 @@ err:
         return 0;
 }
 
+
 int32_t
 marker_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                    int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
                    struct iatt *postbuf)
 {
-        int32_t             ret     = 0;
+        marker_conf_t      *priv    = NULL;
         marker_local_t     *local   = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR, "error occurred "
                         "while write, %s", strerror (op_errno));
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -504,7 +555,19 @@ marker_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         STACK_UNWIND_STRICT (writev, frame, op_ret, op_errno, prebuf, postbuf);
 
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled & GF_QUOTA)
+                initiate_quota_txn (this, &local->loc);
+
+        if (priv->feature_enabled & GF_GSYNC)
+                update_marks (this, local);
+
+out:
+        marker_local_unref (local);
 
         return 0;
 }
@@ -540,18 +603,18 @@ err:
         return 0;
 }
 
+
 int32_t
 marker_rmdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                    int32_t op_ret, int32_t op_errno, struct iatt *preparent,
                    struct iatt *postparent)
 {
-        int32_t             ret     = 0;
+        marker_conf_t      *priv    = NULL;
         marker_local_t     *local   = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR, "error occurred "
                         "rmdir %s", strerror (op_errno));
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -561,7 +624,18 @@ marker_rmdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         STACK_UNWIND_STRICT (rmdir, frame, op_ret, op_errno, preparent,
                              postparent);
 
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled & GF_QUOTA)
+                reduce_parent_size (this, &local->loc);
+
+        if (priv->feature_enabled & GF_GSYNC)
+                update_marks (this, local);
+out:
+        marker_local_unref (local);
 
         return 0;
 }
@@ -590,19 +664,18 @@ err:
         return 0;
 }
 
+
 int32_t
 marker_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                     int32_t op_ret, int32_t op_errno, struct iatt *preparent,
                     struct iatt *postparent)
 {
-        int32_t             ret     = 0;
+        marker_conf_t      *priv    = NULL;
         marker_local_t     *local   = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR,
                         "%s occurred in unlink", strerror (op_errno));
-
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -612,7 +685,18 @@ marker_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         STACK_UNWIND_STRICT (unlink, frame, op_ret, op_errno, preparent,
                              postparent);
 
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled & GF_QUOTA)
+                reduce_parent_size (this, &local->loc);
+
+        if (priv->feature_enabled & GF_GSYNC)
+                update_marks (this, local);
+out:
+        marker_local_unref (local);
 
         return 0;
 }
@@ -641,19 +725,19 @@ err:
         return 0;
 }
 
+
 int32_t
 marker_link_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                   int32_t op_ret, int32_t op_errno, inode_t *inode,
                   struct iatt *buf, struct iatt *preparent,
                   struct iatt *postparent)
 {
-        int32_t             ret     = 0;
         marker_local_t     *local   = NULL;
+        marker_conf_t      *priv    = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR, "%s occured while "
                         "linking a file ", strerror (op_errno));
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -663,7 +747,18 @@ marker_link_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         STACK_UNWIND_STRICT (link, frame, op_ret, op_errno, inode, buf,
                              preparent, postparent);
 
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled & GF_QUOTA)
+                initiate_quota_txn (this, &local->loc);
+
+        if (priv->feature_enabled & GF_GSYNC)
+                update_marks (this, local);
+out:
+        marker_local_unref (local);
 
         return 0;
 }
@@ -692,20 +787,20 @@ err:
         return 0;
 }
 
+
 int32_t
 marker_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                     int32_t op_ret, int32_t op_errno, struct iatt *buf,
                     struct iatt *preoldparent, struct iatt *postoldparent,
                     struct iatt *prenewparent, struct iatt *postnewparent)
 {
-        int32_t             ret     = 0;
+        marker_conf_t      *priv    = NULL;
         marker_local_t     *local   = NULL;
         marker_local_t	   *oplocal = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR, "%s occured while "
                         "renaming a file ", strerror (op_errno));
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -718,10 +813,25 @@ marker_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         oplocal = local->oplocal;
         local->oplocal = NULL;
 
-        //update marks on oldpath
-        update_marks (this, oplocal, ret);
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
 
+        if (priv->feature_enabled & GF_QUOTA) {
+                reduce_parent_size (this, &oplocal->loc);
+                quota_rename_update_newpath (this, &local->loc,
+                                             oplocal->loc.inode);
+        }
+
+        if (priv->feature_enabled & GF_GSYNC) {
+                //update marks on oldpath
+                update_marks (this, oplocal);
+                update_marks (this, local);
+        }
+out:
+        marker_local_unref (local);
+        marker_local_unref (oplocal);
         return 0;
 }
 
@@ -763,18 +873,18 @@ err:
         return 0;
 }
 
+
 int32_t
 marker_truncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                       int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
                       struct iatt *postbuf)
 {
-        int32_t             ret     = 0;
         marker_local_t     *local   = NULL;
+        marker_conf_t      *priv    = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR, "%s occured while "
                         "truncating a file ", strerror (op_errno));
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -784,7 +894,19 @@ marker_truncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         STACK_UNWIND_STRICT (truncate, frame, op_ret, op_errno, prebuf,
                              postbuf);
 
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled & GF_QUOTA)
+                initiate_quota_txn (this, &local->loc);
+
+        if (priv->feature_enabled & GF_GSYNC)
+                update_marks (this, local);
+
+out:
+        marker_local_unref (local);
 
         return 0;
 }
@@ -813,18 +935,18 @@ err:
         return 0;
 }
 
+
 int32_t
 marker_ftruncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                        int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
                        struct iatt *postbuf)
 {
-        int32_t             ret     = 0;
         marker_local_t     *local   = NULL;
+        marker_conf_t      *priv    = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR, "%s occured while "
                         "truncating a file ", strerror (op_errno));
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -834,7 +956,18 @@ marker_ftruncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         STACK_UNWIND_STRICT (ftruncate, frame, op_ret, op_errno, prebuf,
                              postbuf);
 
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled & GF_QUOTA)
+                initiate_quota_txn (this, &local->loc);
+
+        if (priv->feature_enabled & GF_GSYNC)
+                update_marks (this, local);
+out:
+        marker_local_unref (local);
 
         return 0;
 }
@@ -863,19 +996,19 @@ err:
         return 0;
 }
 
+
 int32_t
 marker_symlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                      int32_t op_ret, int32_t op_errno, inode_t *inode,
                      struct iatt *buf, struct iatt *preparent,
                      struct iatt *postparent)
 {
-        int32_t             ret     = 0;
+        marker_conf_t      *priv    = NULL;
         marker_local_t     *local   = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR, "%s occured while "
                         "creating symlinks ", strerror (op_errno));
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -885,7 +1018,18 @@ marker_symlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         STACK_UNWIND_STRICT (symlink, frame, op_ret, op_errno, inode, buf,
                              preparent, postparent);
 
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled & GF_QUOTA)
+                initiate_quota_txn (this, &local->loc);
+
+        if (priv->feature_enabled & GF_GSYNC)
+                update_marks (this, local);
+out:
+        marker_local_unref (local);
 
         return 0;
 }
@@ -915,19 +1059,19 @@ err:
         return 0;
 }
 
+
 int32_t
 marker_mknod_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                    int32_t op_ret, int32_t op_errno, inode_t *inode,
                    struct iatt *buf, struct iatt *preparent,
                    struct iatt *postparent)
 {
-        int32_t             ret     = 0;
         marker_local_t     *local   = NULL;
+        marker_conf_t      *priv    = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR, "%s occured while "
                         "creating symlinks ", strerror (op_errno));
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -937,7 +1081,15 @@ marker_mknod_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         STACK_UNWIND_STRICT (mknod, frame, op_ret, op_errno, inode,
                              buf, preparent, postparent);
 
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled == GF_GSYNC)
+                update_marks (this, local);
+out:
+        marker_local_unref (local);
 
         return 0;
 }
@@ -967,6 +1119,7 @@ err:
         return 0;
 }
 
+
 /* when a call from the special client is received on
  * key trusted.glusterfs.volume-mark with value "RESET"
  * or if the value is 0length, update the change the
@@ -1027,17 +1180,17 @@ out:
         return 0;
 }
 
+
 int32_t
 marker_setxattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                       int32_t op_ret, int32_t op_errno)
 {
-        int32_t             ret     = 0;
         marker_local_t     *local   = NULL;
+        marker_conf_t      *priv    = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR, "%s occured while "
                         "creating symlinks ", strerror (op_errno));
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -1046,7 +1199,15 @@ marker_setxattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         STACK_UNWIND_STRICT (setxattr, frame, op_ret, op_errno);
 
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled == GF_GSYNC)
+                update_marks (this, local);
+out:
+        marker_local_unref (local);
 
         return 0;
 }
@@ -1080,17 +1241,17 @@ err:
         return 0;
 }
 
+
 int32_t
 marker_fsetxattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                        int32_t op_ret, int32_t op_errno)
 {
-        int32_t             ret     = 0;
         marker_local_t     *local   = NULL;
+        marker_conf_t      *priv    = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR, "%s occured while "
                         "creating symlinks ", strerror (op_errno));
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -1099,7 +1260,15 @@ marker_fsetxattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         STACK_UNWIND_STRICT (fsetxattr, frame, op_ret, op_errno);
 
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled == GF_GSYNC)
+                update_marks (this, local);
+out:
+        marker_local_unref (local);
 
         return 0;
 }
@@ -1133,18 +1302,18 @@ err:
         return 0;
 }
 
+
 int32_t
 marker_fsetattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                       int32_t op_ret, int32_t op_errno, struct iatt *statpre,
                       struct iatt *statpost)
 {
-        int32_t             ret     = 0;
         marker_local_t     *local   = NULL;
+        marker_conf_t      *priv    = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR, "%s occured while "
                         "creating symlinks ", strerror (op_errno));
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -1154,11 +1323,20 @@ marker_fsetattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         STACK_UNWIND_STRICT (fsetattr, frame, op_ret, op_errno, statpre,
                              statpost);
 
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled == GF_GSYNC)
+                update_marks (this, local);
+out:
+        marker_local_unref (local);
 
         return 0;
 }
 
+
 int32_t
 marker_fsetattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
                   struct iatt *stbuf, int32_t valid)
@@ -1184,18 +1362,18 @@ err:
         return 0;
 }
 
+
 int32_t
 marker_setattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                      int32_t op_ret, int32_t op_errno, struct iatt *statpre,
                      struct iatt *statpost)
 {
-        int32_t             ret     = 0;
         marker_local_t     *local   = NULL;
+        marker_conf_t      *priv    = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR, "%s occured while "
                         "creating symlinks ", strerror (op_errno));
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -1205,7 +1383,15 @@ marker_setattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         STACK_UNWIND_STRICT (setattr, frame, op_ret, op_errno, statpre,
                              statpost);
 
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled == GF_GSYNC)
+                update_marks (this, local);
+out:
+        marker_local_unref (local);
 
         return 0;
 }
@@ -1235,17 +1421,17 @@ err:
         return 0;
 }
 
+
 int32_t
 marker_removexattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                          int32_t op_ret, int32_t op_errno)
 {
-        int32_t             ret     = 0;
         marker_local_t     *local   = NULL;
+        marker_conf_t      *priv    = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_ERROR, "%s occured while "
                         "creating symlinks ", strerror (op_errno));
-                ret = -1;
         }
 
         local = (marker_local_t *) frame->local;
@@ -1254,7 +1440,15 @@ marker_removexattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         STACK_UNWIND_STRICT (removexattr, frame, op_ret, op_errno);
 
-        update_marks (this, local, ret);
+        if (op_ret == -1)
+                goto out;
+
+        priv = this->private;
+
+        if (priv->feature_enabled == GF_GSYNC)
+                update_marks (this, local);
+out:
+        marker_local_unref (local);
 
         return 0;
 }
@@ -1284,6 +1478,79 @@ err:
         return 0;
 }
 
+
+int32_t
+marker_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                   int32_t op_ret, int32_t op_errno, inode_t *inode,
+                   struct iatt *buf, dict_t *dict, struct iatt *postparent)
+{
+        struct iatt     stat_buf;
+        marker_conf_t  *priv    = NULL;
+        marker_local_t *local   = NULL;
+
+        if (op_ret == -1) {
+                gf_log (this->name, GF_LOG_TRACE, "lookup failed with %s",
+                        strerror (op_errno));
+        }
+
+        priv = this->private;
+
+        dict_ref (dict);
+
+        memmove (&stat_buf, buf, sizeof (struct iatt));
+
+        local = (marker_local_t *) frame->local;
+
+        frame->local = NULL;
+
+        STACK_UNWIND_STRICT (lookup, frame, op_ret, op_errno, inode, buf,
+                             dict, postparent);
+
+        if (op_ret == -1)
+                goto out;
+
+        if (priv->feature_enabled & GF_QUOTA) {
+                quota_xattr_state (this, &local->loc, dict, stat_buf);
+        }
+
+out:
+        marker_local_unref (local);
+
+        dict_unref (dict);
+
+        return 0;
+}
+
+int32_t
+marker_lookup (call_frame_t *frame, xlator_t *this,
+               loc_t *loc, dict_t *xattr_req)
+{
+        int32_t         ret     = 0;
+        marker_conf_t  *priv    = NULL;
+        marker_local_t *local   = NULL;
+
+        priv = this->private;
+
+        ALLOCATE_OR_GOTO (local, marker_local_t, err);
+
+        MARKER_INIT_LOCAL (frame, local);
+
+        ret = loc_copy (&local->loc, loc);
+        if (ret == -1)
+                goto err;
+
+        if (priv->feature_enabled & GF_QUOTA)
+                quota_req_xattr (this, loc, xattr_req);
+
+        STACK_WIND (frame, marker_lookup_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->lookup, loc, xattr_req);
+        return 0;
+err:
+        STACK_UNWIND_STRICT (lookup, frame, -1, 0, NULL, NULL, NULL, NULL);
+
+        return 0;
+}
+
 int32_t
 mem_acct_init (xlator_t *this)
 {
@@ -1303,39 +1570,24 @@ mem_acct_init (xlator_t *this)
         return ret;
 }
 
+
 int32_t
-init (xlator_t *this)
+init_gsync_priv (xlator_t *this)
 {
-        dict_t        *options = NULL;
-        data_t        *data    = NULL;
-        int32_t        ret     = 0;
-        marker_conf_t *priv    = NULL;
-
-        if (!this->children) {
-                gf_log (this->name, GF_LOG_ERROR,
-                        "marker translator needs subvolume defined.");
-                return -1;
-        }
-
-        if (!this->parents) {
-                gf_log (this->name, GF_LOG_WARNING,
-                        "Volume is dangling.");
-                return -1;
-        }
+        dict_t          *options = NULL;
+        data_t          *data    = NULL;
+        int32_t          ret     = 0;
+        marker_conf_t   *priv    = NULL;
 
         options = this->options;
 
-        ALLOCATE_OR_GOTO (this->private, marker_conf_t, err);
-
-        priv = this->private;
-
-        if( (data = dict_get (options, VOLUME_UUID)) != NULL) {
+        if((data = dict_get (options, VOLUME_UUID)) != NULL) {
                 priv->volume_uuid = data->data;
 
                 ret = uuid_parse (priv->volume_uuid, priv->volume_uuid_bin);
                 if (ret == -1) {
                         gf_log (this->name, GF_LOG_ERROR, "invalid volume uuid %s", priv->volume_uuid);
-                        goto err;
+                        goto out;
                 }
 
                 ret = gf_asprintf (& (priv->marker_xattr), "%s.%s.%s",
@@ -1346,7 +1598,7 @@ init (xlator_t *this)
 
                         gf_log (this->name, GF_LOG_ERROR,
                                 "Failed to allocate memory");
-                        goto err;
+                        goto out;
                 }
 
                 gf_log (this->name, GF_LOG_DEBUG,
@@ -1375,6 +1627,68 @@ init (xlator_t *this)
                         "please specify the timestamp-file"
                         "in the translator options");
 
+                goto out;
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
+
+int32_t
+init (xlator_t *this)
+{
+        dict_t        *options = NULL;
+        data_t        *data    = NULL;
+        int32_t        ret     = 0;
+        marker_conf_t *priv    = NULL;
+
+        if (!this->children) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "marker translator needs subvolume defined.");
+                return -1;
+        }
+
+        if (!this->parents) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "Volume is dangling.");
+                return -1;
+        }
+
+        options = this->options;
+
+        ALLOCATE_OR_GOTO (this->private, marker_conf_t, err);
+
+        priv = this->private;
+
+        priv->feature_enabled = 0;
+
+        data = dict_get (options, "quota");
+        if (data) {
+                if (strcmp (data->data, "on") == 0) {
+                        priv->feature_enabled |= GF_QUOTA;
+                        ret = init_quota_priv (this);
+                        if (ret < 0)
+                                goto err;
+                }
+        }
+
+        data = dict_get (options, "marker");
+        if (data) {
+                if (strcmp (data->data, "on") == 0) {
+                        priv->feature_enabled |= GF_GSYNC;
+                        ret = init_gsync_priv (this);
+                        if (ret < 0)
+                                goto err;
+                }
+        }
+
+        if (priv->feature_enabled == 0) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "Marker translator in the volfile "
+                        "but no functionality is enabled "
+                        "failing marker init()");
+
                 goto err;
         }
 
@@ -1385,6 +1699,21 @@ err:
         return -1;
 }
 
+int32_t
+marker_forget (xlator_t *this, inode_t *inode)
+{
+        marker_inode_ctx_t      *ctx = NULL;
+
+        if (inode_ctx_get (inode, this, (uint64_t *) &ctx) != 0)
+                goto out;
+
+        quota_forget (this, ctx->quota_ctx);
+
+        GF_FREE (ctx);
+out:
+        return 0;
+}
+
 void
 fini (xlator_t *this)
 {
@@ -1404,22 +1733,29 @@ fini (xlator_t *this)
         if (priv->marker_xattr != NULL)
                 GF_FREE (priv->marker_xattr);
 
+        if (priv->size_key != NULL)
+                GF_FREE (priv->size_key);
+
+        if (priv->dirty_key != NULL)
+                GF_FREE (priv->dirty_key);
+
         GF_FREE (priv);
 out:
         return ;
 }
 
 struct xlator_fops fops = {
+        .lookup      = marker_lookup,
         .create      = marker_create,
-        .unlink      = marker_unlink,
-        .link        = marker_link,
         .mkdir       = marker_mkdir,
-        .rmdir       = marker_rmdir,
         .writev      = marker_writev,
-        .rename      = marker_rename,
         .truncate    = marker_truncate,
         .ftruncate   = marker_ftruncate,
         .symlink     = marker_symlink,
+        .link        = marker_link,
+        .unlink      = marker_unlink,
+        .rmdir       = marker_rmdir,
+        .rename      = marker_rename,
         .mknod       = marker_mknod,
         .setxattr    = marker_setxattr,
         .fsetxattr   = marker_fsetxattr,
@@ -1430,10 +1766,13 @@ struct xlator_fops fops = {
 };
 
 struct xlator_cbks cbks = {
+         .forget = marker_forget
 };
 
 struct volume_options options[] = {
         {.key = {"volume-uuid"}},
         {.key = {"timestamp-file"}},
+        {.key = {"quota"}},
+        {.key = {"marker"}},
         {.key = {NULL}}
 };
diff --git a/xlators/features/marker/src/marker.h b/xlators/features/marker/src/marker.h
index eeb64a16a87..5602456ff5d 100644
--- a/xlators/features/marker/src/marker.h
+++ b/xlators/features/marker/src/marker.h
@@ -16,11 +16,15 @@
   <http://www.gnu.org/licenses/>.
 */
 
+#ifndef _MARKER_H
+#define _MARKER_H
+
 #ifndef _CONFIG_H
 #define _CONFIG_H
 #include "config.h"
 #endif
 
+#include "marker-quota.h"
 #include "xlator.h"
 #include "defaults.h"
 #include "uuid.h"
@@ -31,11 +35,18 @@
 #define VOLUME_UUID         "volume-uuid"
 #define TIMESTAMP_FILE      "timestamp-file"
 
+enum {
+        GF_QUOTA=1,
+        GF_GSYNC=2
+};
+
 /*initialize the local variable*/
 #define MARKER_INIT_LOCAL(_frame,_local) do {                   \
                 _frame->local = _local;                         \
                 _local->pid = _frame->root->pid;                \
                 memset (&_local->loc, 0, sizeof (loc_t));       \
+                _local->ref = 1;                                \
+                LOCK_INIT (&_local->lock);                      \
                 _local->oplocal = NULL;                         \
         } while (0)
 
@@ -54,15 +65,28 @@ struct marker_local{
         uint32_t        timebuf[2];
         pid_t           pid;
         loc_t           loc;
+        int32_t         ref;
 
+        gf_lock_t       lock;
         struct marker_local *oplocal;
 };
 typedef struct marker_local marker_local_t;
 
+struct marker_inode_ctx {
+        struct quota_inode_ctx *quota_ctx;
+};
+typedef struct marker_inode_ctx marker_inode_ctx_t;
+
 struct marker_conf{
+        char         feature_enabled;
+        char        *size_key;
+        char        *dirty_key;
         char        *volume_uuid;
         uuid_t      volume_uuid_bin;
         char        *timestamp_file;
         char        *marker_xattr;
 };
 typedef struct marker_conf marker_conf_t;
+
+int32_t k;
+#endif