features/quota: Improvements to quota

Old implementation
* Client side implementation of quota
    - Not secure
    - Increased traffic in updating the ctx

New Implementation
* 2 stages of quota implementation is done: Soft and hard quota
    Upon reaching soft quota limit on the directory it logs/alerts in the quota
    daemon log (ie DEFAULT_LOG_DIR/quotad.log) and no more writes allowed after
    hard quota limit. After reaching the soft-limit the daemon alerts the
    user/admin repeatively for every 'alert-time', which is configurable.

* Quota is moved to server-side.

    There will be 2 quota xlators
    i. Quota Server
        It takes care of the enforcing the quota and maintains the context
        specific to the brick. Since this doesn't have the complete picture of
        the cluster, cluster wide usage is updated from the quota daemon. This
        updated context is saved and used for the enforcement.

        It updates its context by searching the QUOTA_UPDATE_KEY from the dict
        in the setxattr call, and is updated from nowhere else.

        The quota is always loaded in the server graph and is by passed if the
        feature is not enabled.

        Options specific to quota-server:
        server-quota    - Specifies whether the features is on/off. It is used
                          to by pass the quota if turned off.
        deem-statfs     - If set to on, it takes quota limits into
                          consideration while estimating fs size. (df command)

    ii. Quota Daemon
        This is the new xlator introduced with this patch. Its the
        *gluster client* process with no mount point, started upon enabling
        quota or restarting the volume. This is a single process for all the
        volumes in the cluster. Its volfile stored in
        GLUSTERD_DEFAULT_WORKI_DIR/quotad/quotad.vol.

        It queries for the sizes on all the bricks, aggregates the size and
        sends back the updated size, periodically. The timeout between
        successive updation is configurable and typically/by default more for
        below-soft-quota usage and less for above-soft-quota usage. It
        maintains the timeout inside the limit structure based on the usage;
        below soft limit and above soft limit.

        There will be thread running per volume which iterates through the list
        and decides whether the size to be queried in the current iteration
        based on its timeout. It takes the next iteration time taking the least
        of the timeouts in the list of entries.

        Maintains a separate inode table for each volume in the quotad. In the
        first iteration it builds the table for quota-dirs (dirs on which limit
        is set) and its components.

        Options specific to quotad:
        hard-timeout       - Timeout for updation of usage to the quota-server
                             when the usage is crosses the soft-limit.
        soft-timeout       - Timeout for the updation of usage to the
                             quota-server when the usage is below soft-limit.
        alert-time         - Frequency of logging after the usage reached
                             soft limit.
   Options common to both:
   default-soft-limit - This is used when individual paths are not
                        configured with soft-limit and default value of
                        this option is 90% of the hard-limit.
   limit-set          - String containing all the limits.

   Thus in the current implementation we'll have 2 quota xlators: one in server
   graph and one in trusted client (quota daemon) of which the sole
   purpose will be to aggregate the quota size xattrs from all the bricks and
   send the same to server quota xlator.

* Changes in glusterd and CLI

   A single volfile is created for all the volumes, similar to nfs volfile.
   All files related to quota client (volfile, pid etc) are stored in
   GLUSTERD_DEFAULT_WORK_DIR/quotad/.

   The new pattern of the quota limit stores in
   limit-set = <single-dir-limit>[,<single-dir-limit>]
   single-dir-limit = <abs-path>:<hard-limit>[:<soft-limit-in-percent>]

   It also introduces new options:
   volume quota <VOLNAME> {enable|disable|list [<path> ...]|remove <path>| default-soft-limit <percent>} |
   volume quota <VOLNAME> {limit-usage <path> <size> |soft-limit <path> <percent>} |
   volume quota <VOLNAME> {alert-time|soft-timeout|hard-timeout} {<time>}

Credit:
Raghavendra Bhat        <rabhat@redhat.com>
Varun Shastry           <vshastry@redhat.com>
Shishir Gowda           <sgowda@redhat.com>
Kruthika Dhananjay      <kdhananj@redhat.com>
Brian Foster            <bfoster@redhat.com>
Krishnan Parthasarathi  <kparthas@redhat.com>

Change-Id: I16ec5be0c2faaf42b14034b9ccaf17796adef082
BUG: 969461
Signed-off-by: Varun Shastry <vshastry@redhat.com>

30 files changed, 3441 insertions, 1243 deletions

diff --git a/cli/src/cli-cmd-parser.c b/cli/src/cli-cmd-parser.c
index 484c4a34..78f680da 100644
--- a/cli/src/cli-cmd-parser.c
+++ b/cli/src/cli-cmd-parser.c
@@ -532,8 +532,13 @@ cli_cmd_quota_parse (const char **words, int wordcount, dict_t **options)
         uint64_t         value   = 0;
         gf_quota_type    type    = GF_QUOTA_OPTION_TYPE_NONE;
         char           *opwords[] = { "enable", "disable", "limit-usage",
-                                      "remove", "list", "version", NULL };
+                                      "remove", "list", "soft-limit",
+                                      "alert-time", "soft-timeout",
+                                      "hard-timeout", "default-soft-limit",
+                                      NULL};
         char            *w       = NULL;
+        uint32_t         time    = 0;
+        double           percent = 0;
 
         GF_ASSERT (words);
         GF_ASSERT (options);
@@ -614,8 +619,8 @@ cli_cmd_quota_parse (const char **words, int wordcount, dict_t **options)
 
                 if (words[4][0] != '/') {
                         cli_err ("Please enter absolute path");
-
-                        return -2;
+                        ret = -1;
+                        goto out;
                 }
                 ret = dict_set_str (dict, "path", (char *) words[4]);
                 if (ret)
@@ -623,14 +628,14 @@ cli_cmd_quota_parse (const char **words, int wordcount, dict_t **options)
 
                 if (!words[5]) {
                         cli_err ("Please enter the limit value to be set");
-
-                        return -2;
+                        ret = -1;
+                        goto out;
                 }
 
                 ret = gf_string2bytesize (words[5], &value);
                 if (ret != 0) {
                         cli_err ("Please enter a correct value");
-                        return -1;
+                        goto out;
                 }
 
                 ret = dict_set_str (dict, "limit", (char *) words[5]);
@@ -649,8 +654,8 @@ cli_cmd_quota_parse (const char **words, int wordcount, dict_t **options)
 
                 if (words[4][0] != '/') {
                         cli_err ("Please enter absolute path");
-
-                        return -2;
+                        ret = -1;
+                        goto out;
                 }
 
                 ret = dict_set_str (dict, "path", (char *) words[4]);
@@ -683,8 +688,109 @@ cli_cmd_quota_parse (const char **words, int wordcount, dict_t **options)
                 goto set_type;
         }
 
-        if (strcmp (w, "version") == 0) {
-                type = GF_QUOTA_OPTION_TYPE_VERSION;
+        if (strcmp (w, "soft-limit") == 0) {
+                if (wordcount != 6) {
+                        ret = -1;
+                        goto out;
+                }
+
+                type = GF_QUOTA_OPTION_TYPE_SOFT_LIMIT;
+                if (words[4][0] != '/') {
+                        cli_err ("Please enter absolute path");
+                        ret = -1;
+                        goto out;
+                }
+                ret = dict_set_str (dict, "path", (char *) words[4]);
+                if (ret)
+                        goto out;
+
+                if (!words[5]) {
+                        cli_err ("Please enter the limit value to be set");
+                        ret = -1;
+                        goto out;
+                }
+
+                ret = gf_string2percent (words[5], &percent);
+                if (ret != 0) {
+                        cli_err ("Please enter a correct value");
+                        goto out;
+                }
+
+                ret = dict_set_str (dict, "limit", (char *) words[5]);
+                if (ret < 0)
+                        goto out;
+
+                goto set_type;
+
+        }
+        if (strcmp (w, "alert-time") == 0) {
+                if (wordcount != 5) {
+                        ret = -1;
+                        goto out;
+                }
+                type = GF_QUOTA_OPTION_TYPE_ALERT_TIME;
+
+                ret = gf_string2time (words[4], &time);
+                if (ret) {
+                        cli_err ("Invalid argument %s. Please enter a valid "
+                                 "string", words[4]);
+                        goto out;
+                }
+
+                ret = dict_set_str (dict, "value", (char *)words[4]);
+                if (ret < 0)
+                        goto out;
+                goto set_type;
+        }
+        if (strcmp (w, "soft-timeout") == 0) {
+                if (wordcount != 5) {
+                        ret = -1;
+                        goto out;
+                }
+                type = GF_QUOTA_OPTION_TYPE_SOFT_TIMEOUT;
+
+                ret = gf_string2time (words[4], &time);
+                if (ret) {
+                        cli_err ("Invalid argument %s. Please enter a valid "
+                                 "string", words[4]);
+                        goto out;
+                }
+
+                ret = dict_set_str (dict, "value", (char *)words[4]);
+                if (ret < 0)
+                        goto out;
+                goto set_type;
+        }
+        if (strcmp (w, "hard-timeout") == 0) {
+                if(wordcount != 5) {
+                        ret = -1;
+                        goto out;
+                }
+                type = GF_QUOTA_OPTION_TYPE_HARD_TIMEOUT;
+
+                ret = gf_string2time (words[4], &time);
+                if (ret) {
+                        cli_err ("Invalid argument %s. Please enter a valid "
+                                 "string", words[4]);
+                        goto out;
+                }
+
+                ret = dict_set_str (dict, "value", (char *)words[4]);
+                if (ret < 0)
+                        goto out;
+                goto set_type;
+        }
+        if (strcmp (w, "default-soft-limit") == 0) {
+                if(wordcount != 5) {
+                        ret = -1;
+                        goto out;
+                }
+                type = GF_QUOTA_OPTION_TYPE_DEFAULT_SOFT_LIMIT;
+
+                ret = dict_set_str (dict, "value", (char *)words[4]);
+                if (ret < 0)
+                        goto out;
+                goto set_type;
         } else {
                 GF_ASSERT (!"opword mismatch");
         }
@@ -2241,7 +2347,7 @@ cli_cmd_validate_dumpoption (const char *arg, char **option)
 {
         char    *opwords[] = {"all", "nfs", "mem", "iobuf", "callpool", "priv",
                               "fd", "inode", "history", "inodectx", "fdctx",
-                              NULL};
+                              "quotad", NULL};
         char    *w = NULL;
 
         w = str_getunamb (arg, opwords);
@@ -2273,6 +2379,10 @@ cli_cmd_volume_statedump_options_parse (const char **words, int wordcount,
                 strncat (option_str, option, strlen (option));
                 strncat (option_str, " ", 1);
         }
+        if((strstr (option_str, "nfs")) && strstr (option_str, "quotad")) {
+                ret = -1;
+                goto out;
+        }
 
         dict = dict_new ();
         if (!dict)
diff --git a/cli/src/cli-cmd-volume.c b/cli/src/cli-cmd-volume.c
index aa1c7755..6465d577 100644
--- a/cli/src/cli-cmd-volume.c
+++ b/cli/src/cli-cmd-volume.c
@@ -1897,7 +1897,9 @@ struct cli_cmd volume_cmds[] = {
            cli_cmd_volume_profile_cbk,
            "volume profile operations"},
 
-        { "volume quota <VOLNAME> <enable|disable|limit-usage|list|remove> [path] [value]",
+        { "volume quota <VOLNAME> {enable|disable|list [<path> ...]|remove <path>| default-soft-limit <percent>} |\n"
+          "volume quota <VOLNAME> {limit-usage <path> <size> |soft-limit <path> <percent>} |\n"
+          "volume quota <VOLNAME> {alert-time|soft-timeout|hard-timeout} {<time>}",
           cli_cmd_quota_cbk,
           "quota translator specific operations"},
 
@@ -1915,7 +1917,7 @@ struct cli_cmd volume_cmds[] = {
           cli_cmd_volume_heal_cbk,
           "self-heal commands on volume specified by <VOLNAME>"},
 
-        {"volume statedump <VOLNAME> [nfs] [all|mem|iobuf|callpool|priv|fd|"
+        {"volume statedump <VOLNAME> [nfs|quotad] [all|mem|iobuf|callpool|priv|fd|"
          "inode|history]...",
          cli_cmd_volume_statedump_cbk,
          "perform statedump on bricks"},
diff --git a/cli/src/cli-rpc-ops.c b/cli/src/cli-rpc-ops.c
index 28b92ee2..05f2e957 100644
--- a/cli/src/cli-rpc-ops.c
+++ b/cli/src/cli-rpc-ops.c
@@ -2322,101 +2322,288 @@ out:
         return ret;
 }
 
-int32_t
-gf_cli_print_limit_list (char *volname, char *limit_list,
-                            char *op_errstr)
+static void
+print_quota_output_for_version_1 (char *path, char *hl, char *mountdir)
 {
-        int64_t  size            = 0;
-        int64_t  limit_value     = 0;
-        int32_t  i, j;
-        int32_t  len = 0, ret    = -1;
-        char     *size_str       = NULL;
-        char     path [PATH_MAX] = {0, };
-        char     ret_str [1024]  = {0, };
-        char     value [1024]    = {0, };
-        char     mountdir []     = "/tmp/mntXXXXXX";
-        char     abspath [PATH_MAX] = {0, };
-        char     *colon_ptr      = NULL;
-        runner_t runner          = {0,};
+        uint64_t used_space       = 0;
+        uint64_t limit_value      = 0;
+        char    *used_str         = NULL;
+        char    abspath[PATH_MAX] = {0,};
+        char    ret_str[1024]     = {0,};
+        int     ret               = -1;
 
-        GF_VALIDATE_OR_GOTO ("cli", volname, out);
-        GF_VALIDATE_OR_GOTO ("cli", limit_list, out);
+        snprintf (abspath, sizeof (abspath) - 1, "%s/%s", mountdir, path);
 
-        if (!connected)
-                goto out;
+        ret = sys_lgetxattr (abspath, "trusted.limit.list", (void *) ret_str,
+                             4096);
+        if (ret < 0) {
+                cli_out ("%-20s %10s", path, hl);
+        } else {
+                sscanf (ret_str, "%"PRIu64",%"PRIu64, &used_space,
+                        &limit_value);
 
-        len = strlen (limit_list);
-        if (len == 0) {
-                cli_err ("%s", op_errstr?op_errstr:"quota limit not set ");
-                goto out;
+                used_str = gf_uint64_2human_readable (used_space);
+
+                if (used_str == NULL)
+                        cli_out ("%-20s %10s %20"PRIu64, path, hl,
+                                 used_space);
+                else
+                        cli_out ("%-20s %10s %20s", path, hl, used_str);
         }
 
-        if (mkdtemp (mountdir) == NULL) {
-                gf_log ("cli", GF_LOG_WARNING, "failed to create a temporary "
-                        "mount directory");
-                ret = -1;
+        GF_FREE (used_str);
+        return;
+}
+
+static void
+print_quota_output_for_version_2 (char *path, char *hl, char *sl,
+                                  char *mountdir)
+{
+        uint64_t used_space       = 0;
+        uint64_t limit_value      = 0;
+        uint64_t hard_limit       = 0;
+        uint64_t avail            = 0;
+        char    *used_str         = NULL;
+        char    *avail_str        = NULL;
+        char    abspath[PATH_MAX] = {0,};
+        char    ret_str[1024]     = {0,};
+        int     ret               = -1;
+
+
+        snprintf (abspath, sizeof (abspath) - 1, "%s/%s", mountdir, path);
+
+        ret = sys_lgetxattr (abspath, "trusted.limit.list", (void *) ret_str,
+                             4096);
+
+        if (ret < 0) {
+                cli_out ("%-40s %7s %9s %11s %7s", path, hl, sl, "N/A", "N/A");
+        } else {
+                sscanf (ret_str, "%"PRIu64",%"PRIu64, &used_space,
+                        &limit_value);
+
+                used_str = gf_uint64_2human_readable (used_space);
+
+                ret = gf_string2bytesize (hl, &hard_limit);
+
+                if (hard_limit > used_space)
+                        avail = hard_limit - used_space;
+                else
+                        avail = 0;
+
+                avail_str = gf_uint64_2human_readable (avail);
+                if (used_str == NULL)
+                        cli_out ("%-40s %7s %9s %11"PRIu64
+                                                 "%9"PRIu64, path, hl,
+                                                 sl, used_space, avail);
+                else
+                        cli_out ("%-40s %7s %9s %11s %7s", path, hl, sl,
+                                 used_str, avail_str);
+        }
+
+        GF_FREE (used_str);
+        GF_FREE (avail_str);
+        return;
+}
+
+static void
+gf_quota_print_limit_list (char *path, char *hl, char *sl, char *mountdir,
+                           uint32_t op_version)
+{
+        if (op_version == 1)
+                print_quota_output_for_version_1 (path, hl, mountdir);
+        else
+                print_quota_output_for_version_2 (path, hl, sl, mountdir);
+}
+
+void
+gf_cli_quota_print_first_row (uint32_t op_version)
+{
+        if (op_version == 1) {
+                cli_out ("\tpath\t\t  limit_set \t\t     size");
+                cli_out ("-----------------------------------------------------"
+                         "-----------------------------");
+        } else {
+                cli_out ("                  Path                   Hard-limit "
+                         "Soft-limit   Used  Available");
+                cli_out ("-----------------------------------------------------"
+                         "---------------------------");
+        }
+}
+
+int
+gf_cli_print_limit_list_from_dict (char *volname, uint32_t op_version,
+                                   dict_t *dict, char *default_sl, int count,
+                                   char *mountdir, int entry_count,
+                                   char *op_errstr)
+{
+        int  ret               = -1;
+        int  i                 = 0;
+        int  j                 = 0;
+        char key[1024]         = {0,};
+        char *limit            = NULL;
+        char path[PATH_MAX]    = {0,};
+        char *hl               = NULL;
+        char *sl               = NULL;
+        char *sl_final         = NULL;
+
+        if (!dict|| count <= 0)
+                goto out;
+
+        if (entry_count == 0) {
+                if (strlen (op_errstr) == 0)
+                        cli_err ("Limit not set on specified directories");
                 goto out;
         }
 
-        /* Mount a temporary client to fetch the disk usage
-         * of the directory on which the limit is set.
-         */
-        ret = runcmd (SBIN_DIR"/glusterfs", "-s",
-                      "localhost", "--volfile-id", volname, "-l",
-                      DEFAULT_LOG_FILE_DIRECTORY"/quota-list.log",
-                      mountdir, NULL);
-        if (ret) {
-                gf_log ("cli", GF_LOG_WARNING, "failed to mount glusterfs client");
-                ret = -1;
-                goto rm_dir;
+        gf_cli_quota_print_first_row (op_version);
+
+        while (count--) {
+                j = 0;
+                sl = NULL;
+                hl = NULL;
+                snprintf (key, sizeof (key), "path%d", i++);
+
+                ret = dict_get_str (dict, key, &limit);
+                if (ret < 0) {
+                        gf_log ("cli", GF_LOG_DEBUG, "Path not present in limit"
+                                " list");
+                        continue;
+                }
+                if (!strcmp (limit, "Not set"))
+                        continue;
+
+                ret = gf_get_hard_limit (limit, &hl);
+
+                ret = gf_get_soft_limit (limit, &sl);
+                if (ret == 1)
+                        sl_final = sl;
+                else
+                        sl_final = default_sl;
+
+               while (limit[j] != ':') {
+                       path[j] = limit[j];
+                       j = j + 1;
+                }
+
+                path[j] = '\0';
+
+                ret = gf_canonicalize_path (path);
+                if (ret)
+                        goto out;
+
+                gf_quota_print_limit_list (path, hl, sl_final, mountdir,
+                                           op_version);
+
+                GF_FREE (hl);
+                GF_FREE (sl);
         }
+out:
+        return ret;
+}
+
+int32_t
+gf_cli_print_limit_list_from_string (char *volname, char *limit_list,
+                                     uint32_t op_version, char *default_sl,
+                                     char *mountdir)
+{
+        int32_t   i                      = 0;
+        int32_t   j                      = 0;
+        int32_t   len                    = 0;
+        int       ret                    = -1;
+        char      path[PATH_MAX]         = {0,};
+        char     *colon_ptr              = NULL;
+        char     *sl_final               = NULL;
+        char     *hl                     = NULL;
+        char     *sl                     = NULL;
 
         len = strlen (limit_list);
         if (len == 0) {
-                cli_err ("quota limit not set ");
-                goto unmount;
+                ret = -1;
+                goto out;
         }
 
-        i = 0;
+        gf_cli_quota_print_first_row (op_version);
 
-        cli_out ("\tpath\t\t  limit_set\t     size");
-        cli_out ("-----------------------------------------------------------"
-                 "-----------------------");
         while (i < len) {
                 j = 0;
+                sl = hl = NULL;
 
                 while (limit_list [i] != ',' && limit_list [i] != '\0') {
                         path [j++] = limit_list[i++];
                 }
                 path [j] = '\0';
-                //here path[] contains both path and limit value
+                        //here path[] contains both path and limit value
 
-                colon_ptr = strrchr (path, ':');
+                ret = gf_get_hard_limit (path, &hl);
+                ret = gf_get_soft_limit (path, &sl);
+
+                if (ret == 1)
+                        sl_final = sl;
+                else
+                        sl_final = default_sl;
+
+                colon_ptr = strchr (path, ':');
                 *colon_ptr = '\0';
-                strcpy (value, ++colon_ptr);
 
-                snprintf (abspath, sizeof (abspath), "%s/%s", mountdir, path);
+                gf_quota_print_limit_list (path, hl, sl_final, mountdir,
+                                           op_version);
 
-                ret = sys_lgetxattr (abspath, "trusted.limit.list", (void *) ret_str, 4096);
-                if (ret < 0) {
-                        cli_out ("%-20s %10s", path, value);
-                } else {
-                        sscanf (ret_str, "%"PRId64",%"PRId64, &size,
-                                &limit_value);
-                        size_str = gf_uint64_2human_readable ((uint64_t) size);
-                        if (size_str == NULL) {
-                                cli_out ("%-20s %10s %20"PRId64, path,
-                                         value, size);
-                        } else {
-                                cli_out ("%-20s %10s %20s", path,
-                                         value, size_str);
-                                GF_FREE (size_str);
-                        }
-                }
                 i++;
+
+                GF_FREE (hl);
+                GF_FREE (sl);
         }
 
-unmount:
+out:
+        return ret;
+}
+
+int
+gf_cli_quota_list_run_crawler (char *volname, char *limit_list, dict_t *dict,
+                               int count, char *op_errstr, uint32_t op_version,
+                               char *default_sl, int entry_count)
+{
+        int       ret                    = -1;
+        runner_t  runner                 = {0,};
+        char      mountdir []            = "/tmp/mntXXXXXX";
+
+        GF_VALIDATE_OR_GOTO ("cli", volname, out);
+
+        if (!connected)
+                goto out;
+
+        if ((!limit_list) && (count <= 0))
+                goto out;
+
+        if (mkdtemp (mountdir) == NULL) {
+                gf_log ("cli", GF_LOG_WARNING, "failed to create a temporary "
+                        "mount directory");
+                ret = -1;
+                goto out;
+        }
+
+        /* Mount a temporary client to fetch the disk usage
+         * of the directory on which the limit is set.
+         */
+        ret = runcmd (SBIN_DIR"/glusterfs", "-s",
+                      "localhost", "--volfile-id", volname, "-l",
+                      DEFAULT_LOG_FILE_DIRECTORY"/quota-list.log",
+                      mountdir, NULL);
+        if (ret) {
+                gf_log ("cli", GF_LOG_WARNING, "failed to mount glusterfs "
+                        "client");
+                ret = -1;
+                goto rm_dir;
+        }
+
+        if (limit_list)
+                gf_cli_print_limit_list_from_string (volname, limit_list,
+                                                     op_version, default_sl,
+                                                     mountdir);
+        else if (count > 0)
+                gf_cli_print_limit_list_from_dict (volname, op_version, dict,
+                                                   default_sl, count, mountdir,
+                                                   entry_count, op_errstr);
 
         runinit (&runner);
         runner_add_args (&runner, "umount",
@@ -2433,20 +2620,24 @@ rm_dir:
         rmdir (mountdir);
 out:
         return ret;
+
 }
 
 int
 gf_cli_quota_cbk (struct rpc_req *req, struct iovec *iov,
                      int count, void *myframe)
 {
-        gf_cli_rsp         rsp        = {0,};
-        int                ret        = -1;
-        dict_t            *dict       = NULL;
-        char              *volname    = NULL;
-        char              *limit_list = NULL;
-        int32_t            type       = 0;
-        char               msg[1024]  = {0,};
-        call_frame_t      *frame      = NULL;
+        gf_cli_rsp         rsp         = {0,};
+        int                ret         = -1;
+        dict_t            *dict        = NULL;
+        char              *volname     = NULL;
+        char              *limit_list  = NULL;
+        int32_t            type        = 0;
+        char               msg[5120]   = {0,};
+        call_frame_t      *frame       = NULL;
+        uint32_t           op_version  = 1;
+        char              *default_sl  = NULL;
+        int                entry_count = 0;
 
         if (-1 == req->rpc_status) {
                 goto out;
@@ -2479,7 +2670,7 @@ gf_cli_quota_cbk (struct rpc_req *req, struct iovec *iov,
                                         rsp.dict.dict_len,
                                         &dict);
                 if (ret < 0) {
-                        gf_log ("glusterd", GF_LOG_ERROR,
+                        gf_log ("cli", GF_LOG_ERROR,
                                 "failed to "
                                 "unserialize req-buffer to dictionary");
                         goto out;
@@ -2496,11 +2687,30 @@ gf_cli_quota_cbk (struct rpc_req *req, struct iovec *iov,
                 gf_log (frame->this->name, GF_LOG_TRACE,
                         "failed to get limit_list");
 
+        ret = dict_get_uint32 (dict, "op-version", &op_version);
+        if (ret)
+                gf_log (frame->this->name, GF_LOG_TRACE, "failed to get "
+                        "op-version");
+
+        ret = dict_get_str (dict, "default-soft-limit", &default_sl);
+        if (ret)
+                gf_log (frame->this->name, GF_LOG_TRACE, "failed to get "
+                        "default soft limit");
+
         ret = dict_get_int32 (dict, "type", &type);
         if (ret)
                 gf_log (frame->this->name, GF_LOG_TRACE,
                         "failed to get type");
 
+        ret = dict_get_int32 (dict, "count", &count);
+        if (ret)
+                gf_log (frame->this->name, GF_LOG_TRACE, "failed to get count");
+
+        ret = dict_get_int32 (dict, "entry-count", &entry_count);
+        if (ret)
+                gf_log (frame->this->name, GF_LOG_TRACE, "failed to get entry "
+                        "count");
+
         if (type == GF_QUOTA_OPTION_TYPE_LIST) {
                 if (global_state->mode & GLUSTER_MODE_XML) {
                         ret = cli_xml_output_vol_quota_limit_list
@@ -2513,19 +2723,15 @@ gf_cli_quota_cbk (struct rpc_req *req, struct iovec *iov,
 
                 }
 
-                if (limit_list) {
-                        gf_cli_print_limit_list (volname,
-                                                    limit_list,
-                                                    rsp.op_errstr);
-                } else {
-                        gf_log ("cli", GF_LOG_INFO, "Received resp to quota "
-                                "command ");
-                        if (rsp.op_errstr)
-                                snprintf (msg, sizeof (msg), "%s",
-                                          rsp.op_errstr);
-                }
+                gf_log ("cli", GF_LOG_INFO, "Received resp to quota command");
+
+                gf_cli_quota_list_run_crawler (volname, limit_list, dict, count,
+                                               rsp.op_errstr, op_version,
+                                               default_sl, entry_count);
+                if (rsp.op_errstr)
+                        snprintf (msg, sizeof (msg), "%s", rsp.op_errstr);
         } else {
-                gf_log ("cli", GF_LOG_INFO, "Received resp to quota command ");
+                gf_log ("cli", GF_LOG_INFO, "Received resp to quota command");
                 if (rsp.op_errstr)
                         snprintf (msg, sizeof (msg), "%s", rsp.op_errstr);
                 else
diff --git a/libglusterfs/src/common-utils.c b/libglusterfs/src/common-utils.c
index 6eb886b3..bec3249f 100644
--- a/libglusterfs/src/common-utils.c
+++ b/libglusterfs/src/common-utils.c
@@ -787,12 +787,44 @@ gf_string2time (const char *str, uint32_t *n)
         if (errno == 0)
                 errno = old_errno;
 
-        if (!((tail[0] == '\0') ||
+        if (((tail[0] == '\0') ||
               ((tail[0] == 's') && (tail[1] == '\0')) ||
               ((tail[0] == 's') && (tail[1] == 'e') &&
 	       (tail[2] == 'c') && (tail[3] == '\0'))))
+               goto out;
+
+        else if (((tail[0] == 'm') && (tail[1] == '\0')) ||
+                 ((tail[0] == 'm') && (tail[1] == 'i') &&
+                  (tail[2] == 'n') && (tail[3] == '\0'))) {
+                value = value * GF_MINUTE_IN_SECONDS;
+                goto out;
+        }
+
+        else if (((tail[0] == 'h') && (tail[1] == '\0')) ||
+                 ((tail[0] == 'h') && (tail[1] == 'r') &&
+	         (tail[2] == '\0'))) {
+                value = value * GF_HOUR_IN_SECONDS;
+                goto out;
+        }
+
+        else if (((tail[0] == 'd') && (tail[1] == '\0')) ||
+                 ((tail[0] == 'd') && (tail[1] == 'a') &&
+	         (tail[2] == 'y') && (tail[3] == 's') &&
+                 (tail[4] == '\0'))) {
+                value = value * GF_DAY_IN_SECONDS;
+                goto out;
+        }
+
+        else if (((tail[0] == 'w') && (tail[1] == '\0')) ||
+                 ((tail[0] == 'w') && (tail[1] == 'k') &&
+	         (tail[2] == '\0'))) {
+                value = value * GF_WEEK_IN_SECONDS;
+                goto out;
+        } else {
                 return -1;
+        }
 
+out:
         *n = value;
 
         return 0;
@@ -2283,6 +2315,9 @@ gf_canonicalize_path (char *path)
         if (!path || *path != '/')
                 goto out;
 
+        if (!strcmp (path, "/"))
+                return 0;
+
         tmppath = gf_strdup (path);
         if (!tmppath)
                 goto out;
@@ -2807,3 +2842,69 @@ out:
 
 }
 
+int
+gf_get_soft_limit (char *limit, char **soft_limit)
+{
+        int   colon_count   = 0;
+        int   i             = 0;
+        int   len           = 0;
+        char *sl            = NULL;
+
+        len = strlen (limit);
+        for (i = 0; i < len; i++) {
+                if (limit[i] == ':')
+                        colon_count++;
+                if (colon_count == 2)
+                        break;
+        }
+
+        if (colon_count != 2) {
+                gf_log ("common-utils", GF_LOG_DEBUG, "Soft-limit absent");
+                return 0;
+        }
+
+        sl = GF_CALLOC (len - i, sizeof (char), gf_common_mt_char);
+        if (!sl)
+                return -1;
+        strncpy (sl, &limit[i+1], len - i - 1);
+        *soft_limit = sl;
+
+        return 1;
+}
+
+int
+gf_get_hard_limit (char *limit, char **hard_limit)
+{
+        int    i                 = 0;
+        int    hlbegin           = 0;
+        int    len               = 0;
+        char  *hl                = NULL;
+
+        len = strlen (limit);
+
+        for (i = 0; i < len; i++) {
+                if (limit[i] == ':')
+                        break;
+        }
+
+        if (i == len) {
+                gf_log ("common-utils", GF_LOG_ERROR, "Hard limit not found");
+                return -1;
+        }
+
+        hlbegin = i + 1;
+        i++;
+
+        while ((limit[i] != '\0') && (limit[i] != ':')) {
+                i++;
+        }
+
+        hl = GF_CALLOC (i - hlbegin + 1, sizeof (char), gf_common_mt_char);
+        if (!hl)
+                return -1;
+
+        strncpy (hl, &limit[hlbegin], i - hlbegin);
+        *hard_limit = hl;
+
+        return 0;
+}
diff --git a/libglusterfs/src/common-utils.h b/libglusterfs/src/common-utils.h
index bf41444d..f8db8e30 100644
--- a/libglusterfs/src/common-utils.h
+++ b/libglusterfs/src/common-utils.h
@@ -81,6 +81,11 @@ void trap (void);
 #define GF_NFS3_PORT    2049
 #define GF_CLIENT_PORT_CEILING 1024
 
+#define GF_MINUTE_IN_SECONDS 60
+#define GF_HOUR_IN_SECONDS (60*60)
+#define GF_DAY_IN_SECONDS (24*60*60)
+#define GF_WEEK_IN_SECONDS (7*24*60*60)
+
 enum _gf_boolean
 {
 	_gf_false = 0,
@@ -587,4 +592,7 @@ gf_boolean_t gf_is_local_addr (char *hostname);
 gf_boolean_t gf_is_same_address (char *host1, char *host2);
 void md5_wrapper(const unsigned char *data, size_t len, char *md5);
 
+int gf_get_soft_limit (char *limit, char **soft_limit);
+int gf_get_hard_limit (char *limit, char **hard_limit);
+
 #endif /* _COMMON_UTILS_H */
diff --git a/libglusterfs/src/inode.h b/libglusterfs/src/inode.h
index cf766a31..89a5f196 100644
--- a/libglusterfs/src/inode.h
+++ b/libglusterfs/src/inode.h
@@ -140,6 +140,9 @@ inode_rename (inode_table_t *table, inode_t *olddir, const char *oldname,
 	      inode_t *newdir, const char *newname,
 	      inode_t *inode, struct iatt *stbuf);
 
+dentry_t *
+__dentry_grep (inode_table_t *table, inode_t *parent, const char *name);
+
 inode_t *
 inode_grep (inode_table_t *table, inode_t *parent, const char *name);
 
diff --git a/libglusterfs/src/options.c b/libglusterfs/src/options.c
index 842b6413..31e5a681 100644
--- a/libglusterfs/src/options.c
+++ b/libglusterfs/src/options.c
@@ -1110,7 +1110,7 @@ DEFINE_INIT_OPT(gf_boolean_t, bool, gf_string2boolean);
 DEFINE_INIT_OPT(xlator_t *, xlator, xl_by_name);
 DEFINE_INIT_OPT(char *, path, pass);
 DEFINE_INIT_OPT(double, double, gf_string2double);
-
+DEFINE_INIT_OPT(uint32_t, time, gf_string2time);
 
 
 DEFINE_RECONF_OPT(char *, str, pass);
@@ -1125,3 +1125,4 @@ DEFINE_RECONF_OPT(gf_boolean_t, bool, gf_string2boolean);
 DEFINE_RECONF_OPT(xlator_t *, xlator, xl_by_name);
 DEFINE_RECONF_OPT(char *, path, pass);
 DEFINE_RECONF_OPT(double, double, gf_string2double);
+DEFINE_RECONF_OPT(uint32_t, time, gf_string2time);
diff --git a/libglusterfs/src/options.h b/libglusterfs/src/options.h
index e2a25baa..71a11c06 100644
--- a/libglusterfs/src/options.h
+++ b/libglusterfs/src/options.h
@@ -114,6 +114,7 @@ DECLARE_INIT_OPT(gf_boolean_t, bool);
 DECLARE_INIT_OPT(xlator_t *, xlator);
 DECLARE_INIT_OPT(char *, path);
 DECLARE_INIT_OPT(double, double);
+DECLARE_INIT_OPT(uint32_t, time);
 
 
 #define DEFINE_INIT_OPT(type_t, type, conv)                             \
diff --git a/libglusterfs/src/xlator.c b/libglusterfs/src/xlator.c
index b1c090c1..57360e1d 100644
--- a/libglusterfs/src/xlator.c
+++ b/libglusterfs/src/xlator.c
@@ -655,7 +655,6 @@ loc_copy (loc_t *dst, loc_t *src)
 
         uuid_copy (dst->gfid, src->gfid);
         uuid_copy (dst->pargfid, src->pargfid);
-        uuid_copy (dst->gfid, src->gfid);
 
         if (src->inode)
                 dst->inode = inode_ref (src->inode);
diff --git a/rpc/xdr/src/cli1-xdr.h b/rpc/xdr/src/cli1-xdr.h
index e80d6f70..edbfddba 100644
--- a/rpc/xdr/src/cli1-xdr.h
+++ b/rpc/xdr/src/cli1-xdr.h
@@ -88,6 +88,11 @@ enum gf_quota_type {
 	GF_QUOTA_OPTION_TYPE_REMOVE = 0 + 4,
 	GF_QUOTA_OPTION_TYPE_LIST = 0 + 5,
 	GF_QUOTA_OPTION_TYPE_VERSION = 0 + 6,
+	GF_QUOTA_OPTION_TYPE_SOFT_LIMIT = 0 + 7,
+	GF_QUOTA_OPTION_TYPE_ALERT_TIME = 0 + 8,
+	GF_QUOTA_OPTION_TYPE_SOFT_TIMEOUT = 0 + 9,
+	GF_QUOTA_OPTION_TYPE_HARD_TIMEOUT = 0 + 10,
+	GF_QUOTA_OPTION_TYPE_DEFAULT_SOFT_LIMIT = 0 + 11,
 };
 typedef enum gf_quota_type gf_quota_type;
 
diff --git a/rpc/xdr/src/cli1-xdr.x b/rpc/xdr/src/cli1-xdr.x
index 0bee1967..1585fb86 100644
--- a/rpc/xdr/src/cli1-xdr.x
+++ b/rpc/xdr/src/cli1-xdr.x
@@ -47,7 +47,12 @@ enum gf_quota_type {
         GF_QUOTA_OPTION_TYPE_LIMIT_USAGE,
         GF_QUOTA_OPTION_TYPE_REMOVE,
         GF_QUOTA_OPTION_TYPE_LIST,
-        GF_QUOTA_OPTION_TYPE_VERSION
+        GF_QUOTA_OPTION_TYPE_VERSION,
+        GF_QUOTA_OPTION_TYPE_SOFT_LIMIT,
+        GF_QUOTA_OPTION_TYPE_ALERT_TIME,
+        GF_QUOTA_OPTION_TYPE_SOFT_TIMEOUT,
+        GF_QUOTA_OPTION_TYPE_HARD_TIMEOUT,
+        GF_QUOTA_OPTION_TYPE_DEFAULT_SOFT_LIMIT
 };
 
 enum gf1_cli_friends_list {
diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index 8b34d1a7..4f02d18f 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -2269,6 +2269,15 @@ dht_getxattr (call_frame_t *frame, xlator_t *this,
                 return 0;
         }
 
+        // Handle the quota limit list command.
+        if (key && !strcmp (GF_XATTR_QUOTA_LIMIT_LIST, key)) {
+                local->call_cnt = 1;
+                subvol = dht_first_up_subvol (this);
+                STACK_WIND (frame, dht_getxattr_cbk, subvol,
+                            subvol->fops->getxattr, loc, key, xdata);
+                return 0;
+        }
+
         if (key && *conf->vol_uuid) {
                 if ((match_uuid_local (key, conf->vol_uuid) == 0) &&
                     (GF_CLIENT_PID_GSYNCD == frame->root->pid)) {
diff --git a/xlators/features/marker/src/marker-quota.c b/xlators/features/marker/src/marker-quota.c
index 6f9af6e1..d1e59509 100644
--- a/xlators/features/marker/src/marker-quota.c
+++ b/xlators/features/marker/src/marker-quota.c
@@ -1234,6 +1234,7 @@ mq_get_parent_inode_local (xlator_t *this, quota_local_t *local)
 {
         int32_t            ret = -1;
         quota_inode_ctx_t *ctx = NULL;
+        inode_contribution_t *contribution = NULL;
 
         GF_VALIDATE_OR_GOTO ("marker", this, out);
         GF_VALIDATE_OR_GOTO ("marker", local, out);
@@ -1263,7 +1264,7 @@ mq_get_parent_inode_local (xlator_t *this, quota_local_t *local)
         ret = mq_inode_ctx_get (local->loc.inode, this, &ctx);
         if (ret < 0) {
                 gf_log_callingfn (this->name, GF_LOG_WARNING,
-                        "inode ctx get failed");
+                                  "inode ctx get failed");
                 goto out;
         }
 
@@ -1277,7 +1278,32 @@ mq_get_parent_inode_local (xlator_t *this, quota_local_t *local)
                 goto out;
         }
 
-        local->contri = (inode_contribution_t *) ctx->contribution_head.next;
+        /* Earlier we used to get the next entry in the list maintained
+           by the context. In a good situation it works. i.e the next
+           parent in the directory hierarchy for this path is obtained.
+
+           But consider the below situation:
+           mount-point: /mnt/point
+           quota enabled directories within mount point: /a, /b, /c
+
+           Now when some file (file1) in the directory /c is written some data,
+           then to update the directories, marker has to get the contribution
+           object for the parent inode, i.e /c.
+           Beefore, it was being done by
+           local->contri = (inode_contribution_t *) ctx->contribution_head.next;
+           It works in the normal situations. But suppose /c is moved to /b.
+           Now /b's contribution object is added to the end of the list of
+           parents that the file file1 within /b/c is maintaining. Now if
+           the file /b/c/file1 is copied to /b/c/new, to update the parent in
+           the order c, b and / we cannot go to the next element in the list,
+           as in this case the next contribution object would be / and /b's
+           contribution will be at the end of the list. So get the proper
+           parent's contribution, by searching the entire list.
+        */
+
+        contribution = mq_get_contribution_node (local->loc.parent, ctx);
+        GF_ASSERT (contribution != NULL);
+        local->contri = contribution;
 
         ret = 0;
 out:
@@ -1521,9 +1547,10 @@ mq_update_parent_size (call_frame_t *frame,
         }
         UNLOCK (&local->contri->lock);
 
-        gf_log (this->name, GF_LOG_DEBUG, "%s %"PRId64 "%"PRId64,
-                local->loc.path, local->ctx->size,
-                local->contri->contribution);
+        gf_log_callingfn (this->name, GF_LOG_DEBUG, "path: %s size: %"PRId64
+                          " contribution:%"PRId64,
+                          local->loc.path, local->ctx->size,
+                          local->contri->contribution);
 
         if (dict == NULL) {
                 op_errno = EINVAL;
@@ -1970,9 +1997,39 @@ mq_initiate_quota_txn (xlator_t *this, loc_t *loc)
                 goto out;
         }
 
+        /* Create the contribution node if its absent. Is it right to
+           assume that if the contribution node is not there, then
+           create one and proceed instead of returning?
+           Reason for this assumption is for hard links. Suppose
+           hard link for a file f1 present in a directory d1 is
+           created in the directory d2 (as f2). Now, since d2's
+           contribution is not there in f1's inode ctx, d2's
+           contribution xattr wont be created and will create problems
+           for quota operations.
+        */
         contribution = mq_get_contribution_node (loc->parent, ctx);
-        if (contribution == NULL)
-                goto out;
+        if (!contribution) {
+                if (strcmp (loc->path, "/"))
+                        gf_log_callingfn (this->name, GF_LOG_TRACE,
+                                          "contribution node for the "
+                                          "path (%s) with parent (%s) "
+                                          "not found", loc->path,
+                                          loc->parent?
+                                          uuid_utoa (loc->parent->gfid):
+                                          NULL);
+                contribution = mq_add_new_contribution_node (this, ctx, loc);
+                if (!contribution) {
+                        if(strcmp (loc->path, "/"))
+                                gf_log_callingfn (this->name, GF_LOG_WARNING,
+                                                  "could not allocate "
+                                                  " contribution node for (%s) "
+                                                  "parent: (%s)", loc->path,
+                                                  loc->parent?
+                                                  uuid_utoa (loc->parent->gfid):
+                                                  NULL);
+                        goto out;
+                }
+        }
 
         /* To improve performance, donot start another transaction
          * if one is already in progress for same inode
@@ -2122,8 +2179,11 @@ mq_inspect_file_xattr (xlator_t *this,
         }
 
         contribution = mq_add_new_contribution_node (this, ctx, loc);
-        if (contribution == NULL)
+        if (contribution == NULL) {
+                gf_log_callingfn (this->name, GF_LOG_WARNING, "cannot allocate "
+                                  "contribution node (path:%s)", loc->path);
                 goto out;
+        }
 
         LOCK (&ctx->lock);
         {
@@ -2235,6 +2295,12 @@ _mq_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this,
         int32_t        ret                = 0;
         char           contri_key [512]   = {0, };
         quota_local_t *local              = NULL;
+        inode_t       *inode              = NULL;
+        dentry_t      *tmp                = NULL;
+        gf_boolean_t  last_dentry         = _gf_true;
+        loc_t         loc                 = {0, };
+        dentry_t      *other_dentry       = NULL;
+        gf_boolean_t  remove              = _gf_false;
 
         local = (quota_local_t *) frame->local;
 
@@ -2245,17 +2311,84 @@ _mq_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this,
 
         frame->local = NULL;
 
-        if (local->hl_count > 1) {
-                GET_CONTRI_KEY (contri_key, local->contri->gfid, ret);
+        GET_CONTRI_KEY (contri_key, local->contri->gfid, ret);
 
-                STACK_WIND (frame, mq_removexattr_cbk, FIRST_CHILD(this),
-                            FIRST_CHILD(this)->fops->removexattr,
-                            &local->loc, contri_key, NULL);
-                ret = 0;
-        } else {
-                mq_removexattr_cbk (frame, NULL, this, 0, 0, NULL);
+        if (!local->loc.inode)
+                inode = inode_grep (local->loc.parent->table, local->loc.parent,
+                                    local->loc.name);
+        else
+                inode = inode_ref (local->loc.inode);
+
+        /* Suppose there are 2 directories dir1 and dir2. Quota limit is set on
+           both the directories. There is a file (f1) in dir1. A hark link is
+           created for that file inside the directory dir2 (say f2). Now one
+           more xattr is set in the inode as a new hard link is created in a
+           separate directory.
+           i.e trusted.glusterfs.quota.<gfid of dir2>.contri=<contribution>
+
+           Now when the hardlink f2 is removed, then the new xattr added (i.e
+           the xattr indicating its contribution to ITS parent directory) should
+           be removed (IFF there is not another hardlink for that file in the
+           same directory).
+
+           To do that upon getting unlink first check whether any other hard
+           links for the same inode exists in the same directory. If so do not
+           do anything and proceed for quota transaction.
+           Otherwise, if the removed entry was the only link for that inode
+           within that directory, then get another dentry for the inode
+           (by traversing the list of dentries for the inode) and using the
+           the dentry's parent and name, send removexattr so that the xattr
+           is removed.
+
+           If it is not done, then if the volume is restarted or the brick
+           process is restarted, then wrong quota usage will be shown for the
+           directory dir2.
+        */
+        if (inode) {
+                tmp = NULL;
+                list_for_each_entry (tmp, &inode->dentry_list, inode_list) {
+                        if (local->loc.parent == tmp->parent) {
+                                if (strcmp (local->loc.name, local->loc.name)) {
+                                        last_dentry = _gf_false;
+                                        break;
+                                }
+                        }
+                }
+                remove = last_dentry;
         }
 
+        if (remove) {
+                if (!other_dentry)    {
+                        list_for_each_entry (tmp, &inode->dentry_list,
+                                             inode_list) {
+                                if (local->loc.parent != tmp->parent) {
+                                        other_dentry = tmp;
+                                        break;
+                                }
+                        }
+                }
+
+                if (!other_dentry)
+                        mq_removexattr_cbk (frame, NULL, this, 0, 0, NULL);
+                else {
+                        loc.parent = inode_ref (other_dentry->parent);
+                        loc.name = gf_strdup (other_dentry->name);
+                        uuid_copy (loc.pargfid , other_dentry->parent->gfid);
+                        loc.inode = inode_ref (inode);
+                        uuid_copy (loc.gfid, inode->gfid);
+                        inode_path (other_dentry->parent, other_dentry->name,
+                                    (char **)&loc.path);
+
+                        STACK_WIND (frame, mq_removexattr_cbk,
+                                    FIRST_CHILD(this),
+                                    FIRST_CHILD(this)->fops->removexattr,
+                                    &loc, contri_key, NULL);
+                }
+        } else
+                mq_removexattr_cbk (frame, NULL, this, 0, 0, NULL);
+
+        ret = 0;
+
         if (strcmp (local->parent_loc.path, "/") != 0) {
                 ret = mq_get_parent_inode_local (this, local);
                 if (ret < 0)
@@ -2266,6 +2399,8 @@ _mq_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this,
 out:
         mq_local_unref (this, local);
 
+        loc_wipe (&loc);
+        inode_unref (inode);
         return 0;
 }
 
@@ -2392,8 +2527,11 @@ mq_reduce_parent_size (xlator_t *this, loc_t *loc, int64_t contri)
                 goto out;
 
         contribution = mq_get_contribution_node (loc->parent, ctx);
-        if (contribution == NULL)
+        if (contribution == NULL) {
+                gf_log_callingfn (this->name, GF_LOG_WARNING, "contribution for"
+                                  " the node %s is NULL", loc->path);
                 goto out;
+        }
 
         local = mq_local_new ();
         if (local == NULL) {
@@ -2412,6 +2550,8 @@ mq_reduce_parent_size (xlator_t *this, loc_t *loc, int64_t contri)
         }
 
         if (local->size == 0) {
+                gf_log_callingfn (this->name, GF_LOG_INFO, "local->size is 0 "
+                                  "path: (%s)", loc->path);
                 ret = 0;
                 goto out;
         }
@@ -2424,8 +2564,12 @@ mq_reduce_parent_size (xlator_t *this, loc_t *loc, int64_t contri)
         local->contri = contribution;
 
         ret = mq_inode_loc_fill (NULL, loc->parent, &local->parent_loc);
-        if (ret < 0)
+        if (ret < 0) {
+                gf_log_callingfn (this->name, GF_LOG_INFO, "building parent loc"
+                                  " failed. (gfid: %s)",
+                                  uuid_utoa (loc->parent->gfid));
                 goto out;
+        }
 
         frame = create_frame (this, this->ctx->pool);
         if (!frame) {
diff --git a/xlators/features/marker/src/marker.c b/xlators/features/marker/src/marker.c
index 3325d8af..cd798b3f 100644
--- a/xlators/features/marker/src/marker.c
+++ b/xlators/features/marker/src/marker.c
@@ -802,7 +802,7 @@ marker_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         priv = this->private;
 
-        if ((priv->feature_enabled & GF_QUOTA) && (local->ia_nlink == 1))
+        if (priv->feature_enabled & GF_QUOTA)
                 mq_reduce_parent_size (this, &local->loc, -1);
 
         if (priv->feature_enabled & GF_XTIME)
diff --git a/xlators/features/quota/src/Makefile.am b/xlators/features/quota/src/Makefile.am
index 9546f427..605c198e 100644
--- a/xlators/features/quota/src/Makefile.am
+++ b/xlators/features/quota/src/Makefile.am
@@ -1,11 +1,15 @@
-xlator_LTLIBRARIES = quota.la
+xlator_LTLIBRARIES = quota.la quotad.la
 xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/features
 
 quota_la_LDFLAGS = -module -avoid-version
+quotad_la_LDFLAGS = -module -avoid-version
 
 quota_la_SOURCES = quota.c
 quota_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la
 
+quotad_la_SOURCES = quotad.c
+quotad_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la
+
 noinst_HEADERS = quota-mem-types.h quota.h
 
 AM_CPPFLAGS = $(GF_CPPFLAGS) -I$(top_srcdir)/libglusterfs/src \
@@ -13,5 +17,5 @@ AM_CPPFLAGS = $(GF_CPPFLAGS) -I$(top_srcdir)/libglusterfs/src \
 
 AM_CFLAGS = -Wall $(GF_CFLAGS)
 
-CLEANFILES = 
+CLEANFILES =
 
diff --git a/xlators/features/quota/src/quota-mem-types.h b/xlators/features/quota/src/quota-mem-types.h
index 3082865d..4831476d 100644
--- a/xlators/features/quota/src/quota-mem-types.h
+++ b/xlators/features/quota/src/quota-mem-types.h
@@ -21,6 +21,8 @@ enum gf_quota_mem_types_ {
         gf_quota_mt_int32_t,
         gf_quota_mt_limits_t,
         gf_quota_mt_quota_dentry_t,
+        gf_quota_mt_quota_limits_level_t,
+        gf_quota_mt_qd_vols_conf_t,
         gf_quota_mt_end
 };
 #endif
diff --git a/xlators/features/quota/src/quota.c b/xlators/features/quota/src/quota.c
index c527e7ca..744748fd 100644
--- a/xlators/features/quota/src/quota.c
+++ b/xlators/features/quota/src/quota.c
@@ -18,6 +18,54 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
                    char *name, uuid_t par);
 struct volume_options options[];
 
+
+static int32_t
+__quota_init_inode_ctx (inode_t *inode, int64_t hard_lim, int64_t soft_lim,
+                        xlator_t *this, dict_t *dict, struct iatt *buf,
+                        quota_inode_ctx_t **context)
+{
+        int32_t            ret  = -1;
+        int64_t           *size = 0;
+        quota_inode_ctx_t *ctx  = NULL;
+
+        if (inode == NULL) {
+                goto out;
+        }
+
+        QUOTA_ALLOC_OR_GOTO (ctx, quota_inode_ctx_t, out);
+
+        ctx->hard_lim = hard_lim;
+        ctx->soft_lim = soft_lim;
+
+        if (buf)
+                ctx->buf = *buf;
+
+        LOCK_INIT(&ctx->lock);
+
+        if (context != NULL) {
+                *context = ctx;
+        }
+
+        INIT_LIST_HEAD (&ctx->parents);
+
+        if (dict != NULL) {
+                ret = dict_get_bin (dict, QUOTA_SIZE_KEY, (void **) &size);
+                if (ret == 0) {
+                        ctx->size = ntoh64 (*size);
+                        gettimeofday (&ctx->tv, NULL);
+                }
+        }
+
+        ret = __inode_ctx_put (inode, this, (uint64_t )(long)ctx);
+        if (ret == -1) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "cannot set quota context in inode (gfid:%s)",
+                        uuid_utoa (inode->gfid));
+        }
+out:
+        return ret;
+}
+
 int
 quota_loc_fill (loc_t *loc, inode_t *inode, inode_t *parent, char *path)
 {
@@ -182,128 +230,16 @@ out:
         return;
 }
 
-
-int32_t
-quota_validate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
-                    int32_t op_ret, int32_t op_errno, dict_t *dict,
-                    dict_t *xdata)
-{
-        quota_local_t     *local          = NULL;
-        uint32_t           validate_count = 0, link_count = 0;
-        int32_t            ret            = 0;
-        quota_inode_ctx_t *ctx            = NULL;
-        int64_t           *size           = 0;
-        uint64_t           value          = 0;
-        call_stub_t       *stub           = NULL;
-
-        local = frame->local;
-
-        if (op_ret < 0) {
-                goto unwind;
-        }
-
-        GF_ASSERT (local);
-        GF_ASSERT (frame);
-        GF_VALIDATE_OR_GOTO_WITH_ERROR ("quota", this, unwind, op_errno,
-                                        EINVAL);
-        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, dict, unwind, op_errno,
-                                        EINVAL);
-
-        ret = inode_ctx_get (local->validate_loc.inode, this, &value);
-
-        ctx = (quota_inode_ctx_t *)(unsigned long)value;
-        if ((ret == -1) || (ctx == NULL)) {
-                gf_log (this->name, GF_LOG_WARNING,
-                        "quota context is not present in inode (gfid:%s)",
-                        uuid_utoa (local->validate_loc.inode->gfid));
-                op_errno = EINVAL;
-                goto unwind;
-        }
-
-        ret = dict_get_bin (dict, QUOTA_SIZE_KEY, (void **) &size);
-        if (ret < 0) {
-                gf_log (this->name, GF_LOG_WARNING,
-                        "size key not present in dict");
-                op_errno = EINVAL;
-                goto unwind;
-        }
-
-        local->just_validated = 1; /* so that we don't go into infinite
-                                    * loop of validation and checking
-                                    * limit when timeout is zero.
-                                    */
-        LOCK (&ctx->lock);
-        {
-                ctx->size = ntoh64 (*size);
-                gettimeofday (&ctx->tv, NULL);
-        }
-        UNLOCK (&ctx->lock);
-
-        quota_check_limit (frame, local->validate_loc.inode, this, NULL, NULL);
-        return 0;
-
-unwind:
-        LOCK (&local->lock);
-        {
-                local->op_ret = -1;
-                local->op_errno = op_errno;
-
-                validate_count = --local->validate_count;
-                link_count = local->link_count;
-
-                if ((validate_count == 0) && (link_count == 0)) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
-        }
-        UNLOCK (&local->lock);
-
-        if (stub != NULL) {
-                call_resume (stub);
-        }
-
-        return 0;
-}
-
-
-static inline uint64_t
-quota_time_elapsed (struct timeval *now, struct timeval *then)
-{
-        return (now->tv_sec - then->tv_sec);
-}
-
-
-int32_t
-quota_timeout (struct timeval *tv, int32_t timeout)
-{
-        struct timeval now       = {0,};
-        int32_t        timed_out = 0;
-
-        gettimeofday (&now, NULL);
-
-        if (quota_time_elapsed (&now, tv) >= timeout) {
-                timed_out = 1;
-        }
-
-        return timed_out;
-}
-
-
 int32_t
 quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
                    char *name, uuid_t par)
 {
-        int32_t               ret            = -1;
         inode_t              *_inode         = NULL, *parent = NULL;
         quota_inode_ctx_t    *ctx            = NULL;
-        quota_priv_t         *priv           = NULL;
         quota_local_t        *local          = NULL;
-        char                  need_validate  = 0, need_unwind = 0;
+        char                  need_unwind = 0;
         int64_t               delta          = 0;
-        call_stub_t          *stub           = NULL;
-        int32_t               validate_count = 0, link_count = 0;
         uint64_t              value          = 0;
-        char                  just_validated = 0;
         uuid_t                trav_uuid      = {0,};
 
         GF_VALIDATE_OR_GOTO ("quota", this, out);
@@ -315,25 +251,11 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
 
         delta = local->delta;
 
-        GF_VALIDATE_OR_GOTO (this->name, local->stub, out);
-
-        priv = this->private;
-
         inode_ctx_get (inode, this, &value);
         ctx = (quota_inode_ctx_t *)(unsigned long)value;
 
         _inode = inode_ref (inode);
 
-        LOCK (&local->lock);
-        {
-                just_validated = local->just_validated;
-                local->just_validated = 0;
-
-                if (just_validated) {
-                        local->validate_count--;
-                }
-        }
-        UNLOCK (&local->lock);
 
         if ( par != NULL ) {
                 uuid_copy (trav_uuid, par);
@@ -343,13 +265,9 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
                 if (ctx != NULL) {
                         LOCK (&ctx->lock);
                         {
-                                if (ctx->limit >= 0) {
-                                        if (!just_validated
-                                            && quota_timeout (&ctx->tv,
-                                                              priv->timeout)) {
-                                                need_validate = 1;
-                                        } else if ((ctx->size + delta)
-                                                   >= ctx->limit) {
+                                if (ctx->hard_lim >= 0) {
+                                        if ((ctx->size + delta)
+                                                   >= ctx->hard_lim) {
                                                 local->op_ret = -1;
                                                 local->op_errno = EDQUOT;
                                                 need_unwind = 1;
@@ -358,10 +276,6 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
                         }
                         UNLOCK (&ctx->lock);
 
-                        if (need_validate) {
-                                goto validate;
-                        }
-
                         if (need_unwind) {
                                 break;
                         }
@@ -387,7 +301,6 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
 
                 inode_unref (_inode);
                 _inode = parent;
-                just_validated = 0;
 
                 if (_inode == NULL) {
                         break;
@@ -398,62 +311,10 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
                 ctx = (quota_inode_ctx_t *)(unsigned long)value;
         } while (1);
 
-        ret = 0;
-
-        if (_inode != NULL) {
-                inode_unref (_inode);
-        }
-
-        LOCK (&local->lock);
-        {
-                validate_count = local->validate_count;
-                link_count = local->link_count;
-                if ((validate_count == 0) && (link_count == 0)) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
-        }
-        UNLOCK (&local->lock);
-
-        if (stub != NULL) {
-                call_resume (stub);
-        }
+        inode_unref (_inode);
 
 out:
-        return ret;
-
-validate:
-        LOCK (&local->lock);
-        {
-                loc_wipe (&local->validate_loc);
-
-                if (just_validated) {
-                        local->validate_count--;
-                }
-
-                local->validate_count++;
-                ret = quota_inode_loc_fill (_inode, &local->validate_loc);
-                if (ret < 0) {
-                        gf_log (this->name, GF_LOG_WARNING,
-                                "cannot fill loc for inode (gfid:%s), hence "
-                                "aborting quota-checks and continuing with fop",
-                                uuid_utoa (_inode->gfid));
-                        local->validate_count--;
-                }
-        }
-        UNLOCK (&local->lock);
-
-        if (ret < 0) {
-                goto loc_fill_failed;
-        }
-
-        STACK_WIND (frame, quota_validate_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->getxattr, &local->validate_loc,
-                    QUOTA_SIZE_KEY, NULL);
-
-loc_fill_failed:
-        inode_unref (_inode);
-        return 0;
+        return local->op_ret;
 }
 
 
@@ -482,7 +343,7 @@ quota_get_limit_value (inode_t *inode, xlator_t *this, int64_t *n)
 
         list_for_each_entry (limit_node, &priv->limit_head, limit_list) {
                 if (strcmp (limit_node->path, path) == 0) {
-                        *n = limit_node->value;
+                        *n = limit_node->hard_lim;
                         break;
                 }
         }
@@ -495,55 +356,9 @@ out:
 
 
 static int32_t
-__quota_init_inode_ctx (inode_t *inode, int64_t limit, xlator_t *this,
-                        dict_t *dict, struct iatt *buf,
-                        quota_inode_ctx_t **context)
-{
-        int32_t            ret  = -1;
-        int64_t           *size = 0;
-        quota_inode_ctx_t *ctx  = NULL;
-
-        if (inode == NULL) {
-                goto out;
-        }
-
-        QUOTA_ALLOC_OR_GOTO (ctx, quota_inode_ctx_t, out);
-
-        ctx->limit = limit;
-        if (buf)
-                ctx->buf = *buf;
-
-        LOCK_INIT(&ctx->lock);
-
-        if (context != NULL) {
-                *context = ctx;
-        }
-
-        INIT_LIST_HEAD (&ctx->parents);
-
-        if (dict != NULL) {
-                ret = dict_get_bin (dict, QUOTA_SIZE_KEY, (void **) &size);
-                if (ret == 0) {
-                        ctx->size = ntoh64 (*size);
-                        gettimeofday (&ctx->tv, NULL);
-                }
-        }
-
-        ret = __inode_ctx_put (inode, this, (uint64_t )(long)ctx);
-        if (ret == -1) {
-                gf_log (this->name, GF_LOG_WARNING,
-                        "cannot set quota context in inode (gfid:%s)",
-                        uuid_utoa (inode->gfid));
-        }
-out:
-        return ret;
-}
-
-
-static int32_t
-quota_inode_ctx_get (inode_t *inode, int64_t limit, xlator_t *this,
-                     dict_t *dict, struct iatt *buf, quota_inode_ctx_t **ctx,
-                     char create_if_absent)
+quota_inode_ctx_get (inode_t *inode, int64_t hard_lim, int64_t soft_lim,
+                     xlator_t *this, dict_t *dict, struct iatt *buf,
+                     quota_inode_ctx_t **ctx, char create_if_absent)
 {
         int32_t  ret = 0;
         uint64_t ctx_int;
@@ -555,8 +370,8 @@ quota_inode_ctx_get (inode_t *inode, int64_t limit, xlator_t *this,
                 if ((ret == 0) && (ctx != NULL)) {
                         *ctx = (quota_inode_ctx_t *) (unsigned long)ctx_int;
                 } else if (create_if_absent) {
-                        ret = __quota_init_inode_ctx (inode, limit, this, dict,
-                                                      buf, ctx);
+                        ret = __quota_init_inode_ctx (inode, hard_lim, soft_lim,
+                                                      this, dict, buf, ctx);
                 }
         }
         UNLOCK (&inode->lock);
@@ -575,10 +390,10 @@ quota_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         quota_local_t     *local      = NULL;
         quota_inode_ctx_t *ctx        = NULL;
         quota_dentry_t    *dentry     = NULL;
-        int64_t           *size       = 0;
         uint64_t           value      = 0;
         limits_t          *limit_node = NULL;
         quota_priv_t      *priv       = NULL;
+        int64_t           *size       = NULL;
 
         local = frame->local;
 
@@ -588,8 +403,8 @@ quota_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         ctx = (quota_inode_ctx_t *)(unsigned long)value;
 
         if ((op_ret < 0) || (local == NULL)
-            || (((ctx == NULL) || (ctx->limit == local->limit))
-                && (local->limit < 0) && !((IA_ISREG (buf->ia_type))
+            || (((ctx == NULL) || (ctx->hard_lim == local->hard_lim))
+                && (local->hard_lim < 0) && !((IA_ISREG (buf->ia_type))
                                            || (IA_ISLNK (buf->ia_type))))) {
                 goto unwind;
         }
@@ -606,8 +421,8 @@ quota_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         }
         UNLOCK (&priv->lock);
 
-        ret = quota_inode_ctx_get (local->loc.inode, local->limit, this, dict,
-                                   buf, &ctx, 1);
+        ret = quota_inode_ctx_get (local->loc.inode, local->hard_lim,
+                                   local->soft_lim, this, dict, buf, &ctx, 1);
         if ((ret == -1) || (ctx == NULL)) {
                 gf_log (this->name, GF_LOG_WARNING, "cannot create quota "
                         "context in inode(gfid:%s)",
@@ -619,22 +434,18 @@ quota_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         LOCK (&ctx->lock);
         {
-
-                if (dict != NULL) {
-                        ret = dict_get_bin (dict, QUOTA_SIZE_KEY,
-                                            (void **) &size);
-                        if (ret == 0) {
-                                ctx->size = ntoh64 (*size);
-                                gettimeofday (&ctx->tv, NULL);
-                        }
-                }
-
-                if (local->limit != ctx->limit) {
-                        ctx->limit = local->limit;
-                }
+                ctx->hard_lim = local->hard_lim;
+                ctx->soft_lim = local->soft_lim;
 
                 ctx->buf = *buf;
 
+                /* will be useful for quotad to determine whether quota xlator's
+                   context is maintaining the correct size.
+                */
+                size = GF_CALLOC (1, sizeof (*size), gf_quota_mt_int64_t);
+                *size = hton64 (ctx->size);
+                ret = dict_set_bin (dict, "trusted.limit.set", size, 8);
+
                 if (!(IA_ISREG (buf->ia_type) || IA_ISLNK (buf->ia_type))) {
                         goto unlock;
                 }
@@ -683,18 +494,23 @@ int32_t
 quota_lookup (call_frame_t *frame, xlator_t *this, loc_t *loc,
               dict_t *xattr_req)
 {
+        quota_priv_t       *priv        = NULL;
         int32_t             ret         = -1;
-        int64_t             limit       = -1;
         limits_t           *limit_node  = NULL;
-        gf_boolean_t        dict_newed  = _gf_false;
-        quota_priv_t       *priv        = NULL;
         quota_local_t      *local       = NULL;
+        int64_t             hard_lim    = -1;
+        int64_t             soft_lim    = -1;
 
         priv = this->private;
 
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
+
         list_for_each_entry (limit_node, &priv->limit_head, limit_list) {
                 if (strcmp (limit_node->path, loc->path) == 0) {
-                        limit = limit_node->value;
+                        hard_lim = limit_node->hard_lim;
+                        soft_lim = limit_node->soft_lim;
+                        break;
                 }
         }
 
@@ -710,25 +526,18 @@ quota_lookup (call_frame_t *frame, xlator_t *this, loc_t *loc,
 
         frame->local = local;
 
-        local->limit = limit;
+        local->hard_lim = hard_lim;
+        local->soft_lim = soft_lim;
 
-        if (limit < 0) {
+        if (hard_lim < 0) {
                 goto wind;
         }
 
-        if (xattr_req == NULL) {
-                xattr_req  = dict_new ();
-                dict_newed = _gf_true;
-        }
-
-        ret = dict_set_uint64 (xattr_req, QUOTA_SIZE_KEY, 0);
-        if (ret < 0) {
-                goto err;
-        }
-
 wind:
-        STACK_WIND (frame, quota_lookup_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->lookup, loc, xattr_req);
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_lookup_cbk: default_lookup_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->lookup, loc,
+                    xattr_req);
 
         ret = 0;
 
@@ -738,10 +547,6 @@ err:
                                     NULL, NULL, NULL, NULL);
         }
 
-        if (dict_newed == _gf_true) {
-                dict_unref (xattr_req);
-        }
-
         return 0;
 }
 
@@ -769,10 +574,12 @@ quota_update_size (xlator_t *this, inode_t *inode, char *name, uuid_t par,
         }
 
         do {
-                if ((ctx != NULL) && (ctx->limit >= 0)) {
+                if ((ctx != NULL) && (ctx->hard_lim >= 0)) {
                         LOCK (&ctx->lock);
                         {
                                 ctx->size += delta;
+                                if (ctx->size < 0)
+                                        ctx->size = 0;
                         }
                         UNLOCK (&ctx->lock);
                 }
@@ -801,6 +608,8 @@ quota_update_size (xlator_t *this, inode_t *inode, char *name, uuid_t par,
                         break;
                 }
 
+                value = 0;
+                ctx = NULL;
                 inode_ctx_get (_inode, this, &value);
                 ctx = (quota_inode_ctx_t *)(unsigned long)value;
         } while (1);
@@ -852,8 +661,8 @@ quota_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         list_for_each_entry (dentry, &ctx->parents, next) {
                 delta = (postbuf->ia_blocks - prebuf->ia_blocks) * 512;
-                quota_update_size (this, local->loc.inode,
-                                   dentry->name, dentry->par, delta);
+                quota_update_size (this, local->loc.inode, dentry->name,
+                                   dentry->par, delta);
         }
 
 out:
@@ -865,49 +674,22 @@ out:
 
 
 int32_t
-quota_writev_helper (call_frame_t *frame, xlator_t *this, fd_t *fd,
-                     struct iovec *vector, int32_t count, off_t off,
-                     uint32_t flags, struct iobref *iobref, dict_t *xdata)
-{
-        quota_local_t *local    = NULL;
-        int32_t        op_errno = EINVAL;
-
-        local = frame->local;
-        if (local == NULL) {
-                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
-                goto unwind;
-        }
-
-        if (local->op_ret == -1) {
-                op_errno = local->op_errno;
-                goto unwind;
-        }
-
-        STACK_WIND (frame, quota_writev_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->writev, fd, vector, count, off,
-                    flags, iobref, xdata);
-        return 0;
-
-unwind:
-        QUOTA_STACK_UNWIND (writev, frame, -1, op_errno, NULL, NULL, NULL);
-        return 0;
-}
-
-
-int32_t
 quota_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
               struct iovec *vector, int32_t count, off_t off,
               uint32_t flags, struct iobref *iobref, dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         int32_t            ret     = -1, op_errno = EINVAL;
         int32_t            parents = 0;
         uint64_t           size    = 0;
         quota_local_t     *local   = NULL;
         quota_inode_ctx_t *ctx     = NULL;
-        quota_priv_t      *priv    = NULL;
-        call_stub_t       *stub    = NULL;
         quota_dentry_t    *dentry  = NULL;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
         GF_ASSERT (frame);
         GF_VALIDATE_OR_GOTO ("quota", this, unwind);
         GF_VALIDATE_OR_GOTO (this->name, fd, unwind);
@@ -920,7 +702,8 @@ quota_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
         frame->local = local;
         local->loc.inode = inode_ref (fd->inode);
 
-        ret = quota_inode_ctx_get (fd->inode, -1, this, NULL, NULL, &ctx, 0);
+        ret = quota_inode_ctx_get (fd->inode, -1, -1, this, NULL, NULL, &ctx,
+                                   0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
                         "quota context not set in inode (gfid:%s)",
@@ -928,13 +711,6 @@ quota_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
                 goto unwind;
         }
 
-        stub = fop_writev_stub (frame, quota_writev_helper, fd, vector, count,
-                                off, flags, iobref, xdata);
-        if (stub == NULL) {
-                op_errno = ENOMEM;
-                goto unwind;
-        }
-
         priv = this->private;
         GF_VALIDATE_OR_GOTO (this->name, priv, unwind);
 
@@ -948,32 +724,22 @@ quota_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
         UNLOCK (&ctx->lock);
 
         local->delta = size;
-        local->stub = stub;
         local->link_count = parents;
 
         list_for_each_entry (dentry, &ctx->parents, next) {
                 ret = quota_check_limit (frame, fd->inode, this, dentry->name,
                                          dentry->par);
                 if (ret == -1) {
-                        break;
-                }
-        }
-
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                local->link_count = 0;
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
+                        op_errno = EDQUOT;
+                        goto unwind;
                 }
         }
-        UNLOCK (&local->lock);
 
-        if (stub != NULL) {
-                call_resume (stub);
-        }
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_writev_cbk: default_writev_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->writev, fd,
+                    vector, count, off, flags, iobref, xdata);
 
         return 0;
 
@@ -996,42 +762,16 @@ quota_mkdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
 
 int32_t
-quota_mkdir_helper (call_frame_t *frame, xlator_t *this, loc_t *loc,
-                    mode_t mode, mode_t umask, dict_t *xdata)
-{
-        quota_local_t *local    = NULL;
-        int32_t        op_errno = EINVAL;
-
-        local = frame->local;
-        if (local == NULL) {
-                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
-                goto unwind;
-        }
-
-        op_errno = local->op_errno;
-
-        if (local->op_ret == -1) {
-                goto unwind;
-        }
-
-        STACK_WIND (frame, quota_mkdir_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->mkdir, loc, mode, umask, xdata);
-        return 0;
-
-unwind:
-        QUOTA_STACK_UNWIND (mkdir, frame, -1, op_errno, NULL, NULL,
-                            NULL, NULL, NULL);
-        return 0;
-}
-
-
-int32_t
 quota_mkdir (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
              mode_t umask, dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         int32_t        ret            = 0, op_errno = 0;
         quota_local_t *local          = NULL;
-        call_stub_t   *stub           = NULL;
+
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -1050,34 +790,19 @@ quota_mkdir (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
                 goto err;
         }
 
-        stub = fop_mkdir_stub (frame, quota_mkdir_helper, loc, mode, umask,
-                               xdata);
-        if (stub == NULL) {
-                op_errno = ENOMEM;
-                goto err;
-        }
-
-        local->stub = stub;
         local->delta = 0;
 
-        quota_check_limit (frame, loc->parent, this, NULL, NULL);
-
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
-
-                local->link_count = 0;
+        ret = quota_check_limit (frame, loc->parent, this, NULL, NULL);
+        if (ret == -1) {
+                op_errno = EDQUOT;
+                goto err;
         }
-        UNLOCK (&local->lock);
 
-        if (stub != NULL) {
-                call_resume (stub);
-        }
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_mkdir_cbk: default_mkdir_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->mkdir, loc,
+                    mode, umask, xdata);
 
         return 0;
 err:
@@ -1104,7 +829,7 @@ quota_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto unwind;
         }
 
-        ret = quota_inode_ctx_get (inode, -1, this, NULL, buf, &ctx, 1);
+        ret = quota_inode_ctx_get (inode, -1, -1, this, NULL, buf, &ctx, 1);
         if ((ret == -1) || (ctx == NULL)) {
                 gf_log (this->name, GF_LOG_WARNING, "cannot create quota "
                         "context in inode(gfid:%s)",
@@ -1141,46 +866,21 @@ unwind:
 
 
 int32_t
-quota_create_helper (call_frame_t *frame, xlator_t *this, loc_t *loc,
-                     int32_t flags, mode_t mode, mode_t umask, fd_t *fd,
-                     dict_t *xdata)
-{
-        quota_local_t *local    = NULL;
-        int32_t        op_errno = EINVAL;
-
-        local = frame->local;
-        if (local == NULL) {
-                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
-                goto unwind;
-        }
-
-        if (local->op_ret == -1) {
-                op_errno = local->op_errno;
-                goto unwind;
-        }
-
-        STACK_WIND (frame, quota_create_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->create, loc, flags, mode, umask,
-                    fd, xdata);
-        return 0;
-
-unwind:
-        QUOTA_STACK_UNWIND (create, frame, -1, op_errno, NULL, NULL,
-                            NULL, NULL, NULL, NULL);
-        return 0;
-}
-
-
-int32_t
 quota_create (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
               mode_t mode, mode_t umask, fd_t *fd, dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         int32_t            ret            = -1;
         quota_local_t     *local          = NULL;
-        call_stub_t       *stub           = NULL;
+        int32_t            op_errno       = 0;
+
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
 
         local = quota_local_new ();
         if (local == NULL) {
+                op_errno = ENOMEM;
                 goto err;
         }
 
@@ -1189,41 +889,28 @@ quota_create (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
         ret = loc_copy (&local->loc, loc);
         if (ret) {
                 gf_log (this->name, GF_LOG_WARNING, "loc_copy failed");
+                op_errno = ENOMEM;
                 goto err;
         }
 
-        stub = fop_create_stub (frame, quota_create_helper, loc, flags, mode,
-                                umask, fd, xdata);
-        if (stub == NULL) {
-                goto err;
-        }
-
-        local->link_count = 1;
-        local->stub = stub;
         local->delta = 0;
 
-        quota_check_limit (frame, loc->parent, this, NULL, NULL);
-
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                local->link_count = 0;
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
+        ret = quota_check_limit (frame, loc->parent, this, NULL, NULL);
+        if (ret == -1) {
+                op_errno = EDQUOT;
+                goto err;
         }
-        UNLOCK (&local->lock);
 
-        if (stub != NULL) {
-                call_resume (stub);
-        }
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_create_cbk: default_create_cbk,
+                    FIRST_CHILD (this), FIRST_CHILD (this)->fops->create, loc,
+                    flags, mode, umask, fd, xdata);
 
         return 0;
 err:
-        QUOTA_STACK_UNWIND (create, frame, -1, ENOMEM, NULL, NULL, NULL, NULL,
-                            NULL, NULL);
+        QUOTA_STACK_UNWIND (create, frame, -1, op_errno, NULL, NULL, NULL,
+                            NULL, NULL, NULL);
 
         return 0;
 }
@@ -1237,6 +924,8 @@ quota_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         quota_local_t     *local = NULL;
         quota_inode_ctx_t *ctx   = NULL;
         uint64_t           value = 0;
+        quota_dentry_t    *dentry = NULL;
+        quota_dentry_t    *old_dentry = NULL;
 
         if (op_ret < 0) {
                 goto out;
@@ -1258,6 +947,21 @@ quota_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                            local->loc.parent->gfid,
                            (-(ctx->buf.ia_blocks * 512)));
 
+        LOCK (&ctx->lock);
+        {
+                list_for_each_entry (dentry, &ctx->parents, next) {
+                        if ((strcmp (dentry->name, local->loc.name) == 0) &&
+                            (uuid_compare (local->loc.parent->gfid,
+                                           dentry->par) == 0)) {
+                                old_dentry = dentry;
+                                break;
+                        }
+                }
+                if (old_dentry)
+                        __quota_dentry_free (old_dentry);
+        }
+        UNLOCK (&ctx->lock);
+
 out:
         QUOTA_STACK_UNWIND (unlink, frame, op_ret, op_errno, preparent,
                             postparent, xdata);
@@ -1269,9 +973,14 @@ int32_t
 quota_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
               dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         int32_t        ret = 0;
         quota_local_t *local = NULL;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
         local = quota_local_new ();
         if (local == NULL) {
                 goto err;
@@ -1285,8 +994,11 @@ quota_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
                 goto err;
         }
 
-        STACK_WIND (frame, quota_unlink_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->unlink, loc, xflag, xdata);
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_unlink_cbk: default_unlink_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->unlink, loc,
+                    xflag, xdata);
 
         ret = 0;
 
@@ -1320,7 +1032,7 @@ quota_link_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         quota_update_size (this, local->loc.parent, NULL, NULL,
                            (buf->ia_blocks * 512));
 
-        ret = quota_inode_ctx_get (inode, -1, this, NULL, NULL, &ctx, 0);
+        ret = quota_inode_ctx_get (inode, -1, -1, this, NULL, NULL, &ctx, 0);
         if ((ret == -1) || (ctx == NULL)) {
                 gf_log (this->name, GF_LOG_WARNING, "cannot find quota "
                         "context in %s (gfid:%s)", local->loc.path,
@@ -1375,44 +1087,19 @@ out:
 
 
 int32_t
-quota_link_helper (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
-                   loc_t *newloc, dict_t *xdata)
-{
-        quota_local_t *local    = NULL;
-        int32_t        op_errno = EINVAL;
-
-        local = frame->local;
-        if (local == NULL) {
-                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
-                goto unwind;
-        }
-
-        op_errno = local->op_errno;
-
-        if (local->op_ret == -1) {
-                goto unwind;
-        }
-
-        STACK_WIND (frame, quota_link_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->link, oldloc, newloc, xdata);
-        return 0;
-
-unwind:
-        QUOTA_STACK_UNWIND (link, frame, -1, op_errno, NULL, NULL,
-                            NULL, NULL, NULL);
-        return 0;
-}
-
-
-int32_t
 quota_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
             dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         int32_t            ret   = -1, op_errno = ENOMEM;
         quota_local_t     *local = NULL;
-        call_stub_t       *stub  = NULL;
         quota_inode_ctx_t *ctx = NULL;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
+
         local = quota_local_new ();
         if (local == NULL) {
                 goto err;
@@ -1426,16 +1113,8 @@ quota_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
                 goto err;
         }
 
-        stub = fop_link_stub (frame, quota_link_helper, oldloc, newloc, xdata);
-        if (stub == NULL) {
-                goto err;
-        }
-
-        local->link_count = 1;
-        local->stub = stub;
-
-        ret = quota_inode_ctx_get (oldloc->inode, -1, this, NULL, NULL, &ctx,
-                                   0);
+        ret = quota_inode_ctx_get (oldloc->inode, -1, -1, this, NULL, NULL,
+                                   &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
                         "quota context not set in inode (gfid:%s)",
@@ -1446,24 +1125,17 @@ quota_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc,
 
         local->delta = ctx->buf.ia_blocks * 512;
 
-        quota_check_limit (frame, newloc->parent, this, NULL, NULL);
-
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
-
-                local->link_count = 0;
+        ret = quota_check_limit (frame, newloc->parent, this, NULL, NULL);
+        if (ret == -1) {
+                op_errno = EDQUOT;
+                goto err;
         }
-        UNLOCK (&local->lock);
 
-        if (stub != NULL) {
-                call_resume (stub);
-        }
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_link_cbk: default_link_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->link, oldloc,
+                    newloc, xdata);
 
         ret = 0;
 err:
@@ -1484,11 +1156,11 @@ quota_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                   dict_t *xdata)
 {
         int32_t               ret              = -1;
+        int64_t               size             = 0;
         quota_local_t        *local            = NULL;
         quota_inode_ctx_t    *ctx              = NULL;
         quota_dentry_t       *old_dentry       = NULL, *dentry = NULL;
         char                  new_dentry_found = 0;
-        int64_t               size             = 0;
 
         if (op_ret < 0) {
                 goto out;
@@ -1508,8 +1180,10 @@ quota_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         }
 
         if (local->oldloc.parent != local->newloc.parent) {
-                quota_update_size (this, local->oldloc.parent, NULL, NULL, (-size));
-                quota_update_size (this, local->newloc.parent, NULL, NULL, size);
+                quota_update_size (this, local->oldloc.parent, NULL, NULL,
+                                   (-size));
+                quota_update_size (this, local->newloc.parent, NULL, NULL,
+                                   size);
         }
 
         if (!(IA_ISREG (local->oldloc.inode->ia_type)
@@ -1517,8 +1191,8 @@ quota_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        ret = quota_inode_ctx_get (local->oldloc.inode, -1, this, NULL, NULL,
-                                   &ctx, 0);
+        ret = quota_inode_ctx_get (local->oldloc.inode, -1, -1, this, NULL,
+                                   NULL, &ctx, 0);
         if ((ret == -1) || (ctx == NULL)) {
                 gf_log (this->name, GF_LOG_WARNING, "quota context not"
                         "set in inode(gfid:%s)",
@@ -1570,7 +1244,8 @@ quota_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                         if (dentry == NULL) {
                                 gf_log (this->name, GF_LOG_WARNING,
                                         "cannot create a new dentry (name:%s) "
-                                        "for inode(gfid:%s)", local->newloc.name,
+                                        "for inode(gfid:%s)",
+                                        local->newloc.name,
                                         uuid_utoa (local->newloc.inode->gfid));
                                 op_ret = -1;
                                 op_errno = ENOMEM;
@@ -1592,44 +1267,18 @@ out:
 
 
 int32_t
-quota_rename_helper (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
-                     loc_t *newloc, dict_t *xdata)
-{
-        quota_local_t *local    = NULL;
-        int32_t        op_errno = EINVAL;
-
-        local = frame->local;
-        if (local == NULL) {
-                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
-                goto unwind;
-        }
-
-        op_errno = local->op_errno;
-
-        if (local->op_ret == -1) {
-                goto unwind;
-        }
-
-        STACK_WIND (frame, quota_rename_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->rename, oldloc, newloc, xdata);
-        return 0;
-
-unwind:
-        QUOTA_STACK_UNWIND (rename, frame, -1, op_errno, NULL, NULL,
-                            NULL, NULL, NULL, NULL);
-        return 0;
-}
-
-
-int32_t
 quota_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
               loc_t *newloc, dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         int32_t            ret            = -1, op_errno = ENOMEM;
         quota_local_t     *local          = NULL;
-        call_stub_t       *stub           = NULL;
         quota_inode_ctx_t *ctx            = NULL;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
         local = quota_local_new ();
         if (local == NULL) {
                 goto err;
@@ -1649,19 +1298,10 @@ quota_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
                 goto err;
         }
 
-        stub = fop_rename_stub (frame, quota_rename_helper, oldloc, newloc,
-                                xdata);
-        if (stub == NULL) {
-                goto err;
-        }
-
-        local->link_count = 1;
-        local->stub = stub;
-
         if (IA_ISREG (oldloc->inode->ia_type)
             || IA_ISLNK (oldloc->inode->ia_type)) {
-                ret = quota_inode_ctx_get (oldloc->inode, -1, this, NULL, NULL,
-                                           &ctx, 0);
+                ret = quota_inode_ctx_get (oldloc->inode, -1, -1, this, NULL,
+                                           NULL, &ctx, 0);
                 if (ctx == NULL) {
                         gf_log (this->name, GF_LOG_WARNING,
                                 "quota context not set in inode (gfid:%s)",
@@ -1675,24 +1315,17 @@ quota_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
                 local->delta = 0;
         }
 
-        quota_check_limit (frame, newloc->parent, this, NULL, NULL);
-
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
-
-                local->link_count = 0;
+        ret = quota_check_limit (frame, newloc->parent, this, NULL, NULL);
+        if (ret == -1) {
+                op_errno = EDQUOT;
+                goto err;
         }
-        UNLOCK (&local->lock);
 
-        if (stub != NULL) {
-                call_resume (stub);
-        }
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_rename_cbk: default_rename_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->rename, oldloc,
+                    newloc, xdata);
 
         ret = 0;
 err:
@@ -1725,7 +1358,7 @@ quota_symlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         quota_update_size (this, local->loc.parent, NULL, NULL, size);
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 1);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -1760,43 +1393,18 @@ out:
 
 
 int
-quota_symlink_helper (call_frame_t *frame, xlator_t *this, const char *linkpath,
-                      loc_t *loc, mode_t umask, dict_t *xdata)
-{
-        quota_local_t *local    = NULL;
-        int32_t        op_errno = EINVAL;
-
-        local = frame->local;
-        if (local == NULL) {
-                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
-                goto unwind;
-        }
-
-        if (local->op_ret == -1) {
-                op_errno = local->op_errno;
-                goto unwind;
-        }
-
-        STACK_WIND (frame, quota_symlink_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->symlink, linkpath, loc, umask,
-                    xdata);
-        return 0;
-
-unwind:
-        QUOTA_STACK_UNWIND (symlink, frame, -1, op_errno, NULL, NULL,
-                            NULL, NULL, NULL);
-        return 0;
-}
-
-
-int
 quota_symlink (call_frame_t *frame, xlator_t *this, const char *linkpath,
                loc_t *loc, mode_t umask, dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         int32_t          ret      = -1;
         int32_t          op_errno = ENOMEM;
         quota_local_t   *local    = NULL;
-        call_stub_t     *stub     = NULL;
+
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -1811,36 +1419,19 @@ quota_symlink (call_frame_t *frame, xlator_t *this, const char *linkpath,
                 goto err;
         }
 
-        local->link_count = 1;
-
-        stub = fop_symlink_stub (frame, quota_symlink_helper, linkpath, loc,
-                                 umask, xdata);
-        if (stub == NULL) {
-                goto err;
-        }
-
-        local->stub = stub;
         local->delta = strlen (linkpath);
 
-        quota_check_limit (frame, loc->parent, this, NULL, NULL);
-
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
-
-                local->link_count = 0;
-        }
-        UNLOCK (&local->lock);
-
-        if (stub != NULL) {
-                call_resume (stub);
+        ret = quota_check_limit (frame, loc->parent, this, NULL, NULL);
+        if (ret == -1) {
+                op_errno = EDQUOT;
+                goto err;
         }
 
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_symlink_cbk: default_symlink_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->symlink,
+                    linkpath, loc, umask, xdata);
         return 0;
 
 err:
@@ -1857,8 +1448,8 @@ quota_truncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                     struct iatt *postbuf, dict_t *xdata)
 {
         quota_local_t     *local = NULL;
-        int64_t            delta = 0;
         quota_inode_ctx_t *ctx   = NULL;
+        int64_t            delta = 0;
 
         if (op_ret < 0) {
                 goto out;
@@ -1874,7 +1465,7 @@ quota_truncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         quota_update_size (this, local->loc.inode, NULL, NULL, delta);
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -1900,9 +1491,15 @@ int32_t
 quota_truncate (call_frame_t *frame, xlator_t *this, loc_t *loc, off_t offset,
                 dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         int32_t          ret   = -1;
         quota_local_t   *local = NULL;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
+
         local = quota_local_new ();
         if (local == NULL) {
                 goto err;
@@ -1916,8 +1513,11 @@ quota_truncate (call_frame_t *frame, xlator_t *this, loc_t *loc, off_t offset,
                 goto err;
         }
 
-        STACK_WIND (frame, quota_truncate_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->truncate, loc, offset, xdata);
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_truncate_cbk: default_truncate_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->truncate, loc,
+                    offset, xdata);
 
         return 0;
 err:
@@ -1933,8 +1533,8 @@ quota_ftruncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                      struct iatt *postbuf, dict_t *xdata)
 {
         quota_local_t     *local = NULL;
-        int64_t            delta = 0;
         quota_inode_ctx_t *ctx   = NULL;
+        int64_t            delta = 0;
 
         if (op_ret < 0) {
                 goto out;
@@ -1950,7 +1550,7 @@ quota_ftruncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         quota_update_size (this, local->loc.inode, NULL, NULL, delta);
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -1976,8 +1576,13 @@ int32_t
 quota_ftruncate (call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
                  dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         quota_local_t   *local = NULL;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
         local = quota_local_new ();
         if (local == NULL)
                 goto err;
@@ -1986,8 +1591,11 @@ quota_ftruncate (call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
 
         local->loc.inode = inode_ref (fd->inode);
 
-        STACK_WIND (frame, quota_ftruncate_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->ftruncate, fd, offset, xdata);
+wind:
+        STACK_WIND (frame, priv->is_quota_on?
+                    quota_ftruncate_cbk: default_ftruncate_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->ftruncate, fd,
+                    offset, xdata);
 
         return 0;
 err:
@@ -2006,14 +1614,23 @@ quota_send_dir_limit_to_cli (call_frame_t *frame, xlator_t *this,
         dict_t            *dict              = NULL;
         quota_inode_ctx_t *ctx               = NULL;
         uint64_t           value             = 0;
+        quota_priv_t      *priv              = NULL;
+
+        priv = this->private;
+        if (!priv->is_quota_on) {
+                snprintf (dir_limit, 1024, "Quota is disabled please turn on");
+                goto dict_set;
+        }
 
         ret = inode_ctx_get (inode, this, &value);
         if (ret < 0)
                 goto out;
 
         ctx = (quota_inode_ctx_t *)(unsigned long)value;
-        snprintf (dir_limit, 1024, "%"PRId64",%"PRId64, ctx->size, ctx->limit);
+        snprintf (dir_limit, 1024, "%"PRId64",%"PRId64, ctx->size,
+                  ctx->hard_lim);
 
+dict_set:
         dict = dict_new ();
         if (dict == NULL) {
                 ret = -1;
@@ -2024,7 +1641,7 @@ quota_send_dir_limit_to_cli (call_frame_t *frame, xlator_t *this,
         if (ret < 0)
                 goto out;
 
-        gf_log (this->name, GF_LOG_INFO, "str = %s", dir_limit);
+        gf_log (this->name, GF_LOG_DEBUG, "str = %s", dir_limit);
 
         QUOTA_STACK_UNWIND (getxattr, frame, 0, 0, dict, NULL);
 
@@ -2076,7 +1693,8 @@ quota_getxattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
 
 int32_t
 quota_stat_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
-                int32_t op_ret, int32_t op_errno, struct iatt *buf, dict_t *xdata)
+                int32_t op_ret, int32_t op_errno, struct iatt *buf,
+                dict_t *xdata)
 {
         quota_local_t     *local = NULL;
         quota_inode_ctx_t *ctx   = NULL;
@@ -2091,7 +1709,7 @@ quota_stat_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_DEBUG,
@@ -2116,9 +1734,15 @@ out:
 int32_t
 quota_stat (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         quota_local_t *local = NULL;
         int32_t        ret   = -1;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
+
         local = quota_local_new ();
         if (local == NULL) {
                 goto unwind;
@@ -2131,8 +1755,10 @@ quota_stat (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
                 goto unwind;
         }
 
-        STACK_WIND (frame, quota_stat_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->stat, loc, xdata);
+wind:
+        STACK_WIND (frame, priv->is_quota_on? quota_stat_cbk: default_stat_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->stat, loc,
+                    xdata);
         return 0;
 
 unwind:
@@ -2159,7 +1785,7 @@ quota_fstat_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -2184,8 +1810,13 @@ out:
 int32_t
 quota_fstat (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         quota_local_t *local = NULL;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
         local = quota_local_new ();
         if (local == NULL) {
                 goto unwind;
@@ -2195,8 +1826,11 @@ quota_fstat (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
 
         local->loc.inode = inode_ref (fd->inode);
 
-        STACK_WIND (frame, quota_fstat_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->fstat, fd, xdata);
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_fstat_cbk: default_fstat_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->fstat, fd,
+                    xdata);
         return 0;
 
 unwind:
@@ -2223,7 +1857,7 @@ quota_readlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -2239,7 +1873,8 @@ quota_readlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         UNLOCK (&ctx->lock);
 
 out:
-        QUOTA_STACK_UNWIND (readlink, frame, op_ret, op_errno, path, buf, xdata);
+        QUOTA_STACK_UNWIND (readlink, frame, op_ret, op_errno, path, buf,
+                            xdata);
         return 0;
 }
 
@@ -2248,9 +1883,14 @@ int32_t
 quota_readlink (call_frame_t *frame, xlator_t *this, loc_t *loc, size_t size,
                 dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         quota_local_t *local = NULL;
         int32_t        ret   = -1;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
         local = quota_local_new ();
         if (local == NULL) {
                 goto unwind;
@@ -2264,8 +1904,11 @@ quota_readlink (call_frame_t *frame, xlator_t *this, loc_t *loc, size_t size,
                 goto unwind;
         }
 
-        STACK_WIND (frame, quota_readlink_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->readlink, loc, size, xdata);
+wind:
+        STACK_WIND (frame, priv->is_quota_on?
+                    quota_readlink_cbk: default_readlink_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->readlink, loc,
+                    size, xdata);
         return 0;
 
 unwind:
@@ -2293,7 +1936,7 @@ quota_readv_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -2319,8 +1962,13 @@ int32_t
 quota_readv (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
              off_t offset, uint32_t flags, dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         quota_local_t *local = NULL;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
         local = quota_local_new ();
         if (local == NULL) {
                 goto unwind;
@@ -2330,13 +1978,16 @@ quota_readv (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
 
         local->loc.inode = inode_ref (fd->inode);
 
-        STACK_WIND (frame, quota_readv_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->readv, fd, size, offset, flags,
-                    xdata);
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_readv_cbk: default_readv_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->readv, fd,
+                    size, offset, flags, xdata);
         return 0;
 
 unwind:
-        QUOTA_STACK_UNWIND (readv, frame, -1, ENOMEM, NULL, -1, NULL, NULL, NULL);
+        QUOTA_STACK_UNWIND (readv, frame, -1, ENOMEM, NULL, -1, NULL, NULL,
+                            NULL);
         return 0;
 }
 
@@ -2359,7 +2010,7 @@ quota_fsync_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -2385,8 +2036,13 @@ int32_t
 quota_fsync (call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t flags,
              dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         quota_local_t *local = NULL;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
         local = quota_local_new ();
         if (local == NULL) {
                 goto unwind;
@@ -2396,8 +2052,11 @@ quota_fsync (call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t flags,
 
         frame->local = local;
 
-        STACK_WIND (frame, quota_fsync_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->fsync, fd, flags, xdata);
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_fsync_cbk: default_fsync_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->fsync, fd,
+                    flags, xdata);
         return 0;
 
 unwind:
@@ -2425,7 +2084,7 @@ quota_setattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_DEBUG,
@@ -2452,9 +2111,14 @@ int32_t
 quota_setattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
                struct iatt *stbuf, int32_t valid, dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         quota_local_t *local = NULL;
         int32_t        ret   = -1;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
         local = quota_local_new ();
         if (local == NULL) {
                 goto unwind;
@@ -2468,8 +2132,11 @@ quota_setattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
                 goto unwind;
         }
 
-        STACK_WIND (frame, quota_setattr_cbk, FIRST_CHILD (this),
-                    FIRST_CHILD (this)->fops->setattr, loc, stbuf, valid, xdata);
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_setattr_cbk: default_setattr_cbk,
+                    FIRST_CHILD (this), FIRST_CHILD (this)->fops->setattr, loc,
+                    stbuf, valid, xdata);
         return 0;
 
 unwind:
@@ -2496,7 +2163,7 @@ quota_fsetattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-        quota_inode_ctx_get (local->loc.inode, -1, this, NULL, NULL,
+        quota_inode_ctx_get (local->loc.inode, -1, -1, this, NULL, NULL,
                              &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
@@ -2522,8 +2189,14 @@ int32_t
 quota_fsetattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
                 struct iatt *stbuf, int32_t valid, dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         quota_local_t *local = NULL;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
+
         local = quota_local_new ();
         if (local == NULL) {
                 goto unwind;
@@ -2533,8 +2206,11 @@ quota_fsetattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
 
         local->loc.inode = inode_ref (fd->inode);
 
-        STACK_WIND (frame, quota_fsetattr_cbk, FIRST_CHILD (this),
-                    FIRST_CHILD (this)->fops->fsetattr, fd, stbuf, valid, xdata);
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_fsetattr_cbk: default_fsetattr_cbk,
+                    FIRST_CHILD (this), FIRST_CHILD (this)->fops->fsetattr, fd,
+                    stbuf, valid, xdata);
         return 0;
 
 unwind:
@@ -2559,7 +2235,7 @@ quota_mknod_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto unwind;
         }
 
-        ret = quota_inode_ctx_get (inode, -1, this, NULL, buf, &ctx, 1);
+        ret = quota_inode_ctx_get (inode, -1, -1, this, NULL, buf, &ctx, 1);
         if ((ret == -1) || (ctx == NULL)) {
                 gf_log (this->name, GF_LOG_WARNING, "cannot create quota "
                         "context in inode (gfid:%s)", uuid_utoa (inode->gfid));
@@ -2595,43 +2271,17 @@ unwind:
 
 
 int
-quota_mknod_helper (call_frame_t *frame, xlator_t *this, loc_t *loc,
-                    mode_t mode, dev_t rdev, mode_t umask, dict_t *xdata)
-{
-        quota_local_t *local    = NULL;
-        int32_t        op_errno = EINVAL;
-
-        local = frame->local;
-        if (local == NULL) {
-                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
-                goto unwind;
-        }
-
-        if (local->op_ret == -1) {
-                op_errno = local->op_errno;
-                goto unwind;
-        }
-
-        STACK_WIND (frame, quota_mknod_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->mknod, loc, mode, rdev, umask,
-                    xdata);
-
-        return 0;
-
-unwind:
-        QUOTA_STACK_UNWIND (mknod, frame, -1, op_errno, NULL, NULL,
-                            NULL, NULL, NULL);
-        return 0;
-}
-
-
-int
 quota_mknod (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
              dev_t rdev, mode_t umask, dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         int32_t            ret            = -1;
         quota_local_t     *local          = NULL;
-        call_stub_t       *stub           = NULL;
+        int32_t            op_errno       = 0;
+
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
 
         local = quota_local_new ();
         if (local == NULL) {
@@ -2646,37 +2296,24 @@ quota_mknod (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
                 goto err;
         }
 
-        stub = fop_mknod_stub (frame, quota_mknod_helper, loc, mode, rdev,
-                               umask, xdata);
-        if (stub == NULL) {
-                goto err;
-        }
-
-        local->link_count = 1;
-        local->stub = stub;
         local->delta = 0;
 
-        quota_check_limit (frame, loc->parent, this, NULL, NULL);
-
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                local->link_count = 0;
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
+        ret = quota_check_limit (frame, loc->parent, this, NULL, NULL);
+        if (ret == -1) {
+                op_errno = EDQUOT;
+                goto err;
         }
-        UNLOCK (&local->lock);
 
-        if (stub != NULL) {
-                call_resume (stub);
-        }
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_mknod_cbk: default_mknod_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->mknod, loc,
+                    mode, rdev, umask, xdata);
+
 
         return 0;
 err:
-        QUOTA_STACK_UNWIND (mknod, frame, -1, ENOMEM, NULL, NULL, NULL, NULL,
+        QUOTA_STACK_UNWIND (mknod, frame, -1, op_errno, NULL, NULL, NULL, NULL,
                             NULL);
 
         return 0;
@@ -2694,8 +2331,17 @@ int
 quota_setxattr (call_frame_t *frame, xlator_t *this,
                 loc_t *loc, dict_t *dict, int flags, dict_t *xdata)
 {
-        int             op_errno = EINVAL;
-        int             op_ret   = -1;
+        quota_priv_t       *priv        = NULL;
+        int                      op_errno       = EINVAL;
+        int                      op_ret         = -1;
+        int64_t                 *size           = 0;
+        uint64_t                 value          = 0;
+        quota_inode_ctx_t       *ctx            = NULL;
+        int                      ret            = -1;
+
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
 
         VALIDATE_OR_GOTO (frame, err);
         VALIDATE_OR_GOTO (this, err);
@@ -2704,10 +2350,36 @@ quota_setxattr (call_frame_t *frame, xlator_t *this,
         GF_IF_INTERNAL_XATTR_GOTO ("trusted.glusterfs.quota*", dict,
                                    op_errno, err);
 
-        STACK_WIND (frame, quota_setxattr_cbk,
-                    FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->setxattr,
-                    loc, dict, flags, xdata);
+        ret = dict_get_bin (dict, QUOTA_UPDATE_USAGE_KEY, (void **) &size);
+        if (0 == ret) {
+
+                inode_ctx_get (loc->inode, this, &value);
+                ctx = (quota_inode_ctx_t *)(unsigned long) value;
+                if (NULL == ctx) {
+                        gf_log (this->name, GF_LOG_ERROR, "Couldn't get the "
+                                "context for %s. Usage may cross the limit.",
+                                loc->path);
+                        op_ret = -1;
+                        goto err;
+                }
+
+                LOCK (&ctx->lock);
+                {
+                        ctx->size = ntoh64 (*size);
+                        if (ctx->size < 0)
+                                ctx->size = 0;
+                }
+                UNLOCK (&ctx->lock);
+
+                QUOTA_STACK_UNWIND (setxattr, frame, ret, 0, xdata);
+                return 0;
+        }
+
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_setxattr_cbk: default_setxattr_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->setxattr, loc,
+                    dict, flags, xdata);
         return 0;
 err:
         QUOTA_STACK_UNWIND (setxattr, frame, op_ret, op_errno, NULL);
@@ -2726,9 +2398,14 @@ int
 quota_fsetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
                  dict_t *dict, int flags, dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         int32_t         op_ret   = -1;
         int32_t         op_errno = EINVAL;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
         VALIDATE_OR_GOTO (frame, err);
         VALIDATE_OR_GOTO (this, err);
         VALIDATE_OR_GOTO (fd, err);
@@ -2736,10 +2413,11 @@ quota_fsetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
         GF_IF_INTERNAL_XATTR_GOTO ("trusted.glusterfs.quota*", dict,
                                    op_errno, err);
 
-        STACK_WIND (frame, quota_fsetxattr_cbk,
-                    FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->fsetxattr,
-                    fd, dict, flags, xdata);
+wind:
+        STACK_WIND (frame, priv->is_quota_on?
+                    quota_fsetxattr_cbk: default_fsetxattr_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->fsetxattr, fd,
+                    dict, flags, xdata);
         return 0;
  err:
         QUOTA_STACK_UNWIND (fsetxattr, frame, op_ret, op_errno, NULL);
@@ -2759,8 +2437,13 @@ int
 quota_removexattr (call_frame_t *frame, xlator_t *this,
                    loc_t *loc, const char *name, dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         int32_t         op_errno = EINVAL;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
         VALIDATE_OR_GOTO (this, err);
 
         GF_IF_NATIVE_XATTR_GOTO ("trusted.quota*",
@@ -2769,9 +2452,10 @@ quota_removexattr (call_frame_t *frame, xlator_t *this,
         VALIDATE_OR_GOTO (frame, err);
         VALIDATE_OR_GOTO (loc, err);
 
-        STACK_WIND (frame, quota_removexattr_cbk,
-                    FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->removexattr,
+wind:
+        STACK_WIND (frame, priv->is_quota_on?
+                    quota_removexattr_cbk: default_removexattr_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->removexattr,
                     loc, name, xdata);
         return 0;
 err:
@@ -2792,9 +2476,14 @@ int
 quota_fremovexattr (call_frame_t *frame, xlator_t *this,
                     fd_t *fd, const char *name, dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
         int32_t         op_ret   = -1;
         int32_t         op_errno = EINVAL;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
         VALIDATE_OR_GOTO (frame, err);
         VALIDATE_OR_GOTO (this, err);
         VALIDATE_OR_GOTO (fd, err);
@@ -2802,9 +2491,10 @@ quota_fremovexattr (call_frame_t *frame, xlator_t *this,
         GF_IF_NATIVE_XATTR_GOTO ("trusted.quota*",
                                  name, op_errno, err);
 
-        STACK_WIND (frame, quota_fremovexattr_cbk,
-                    FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->fremovexattr,
+wind:
+        STACK_WIND (frame, priv->is_quota_on?
+                    quota_fremovexattr_cbk: default_fremovexattr_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->fremovexattr,
                     fd, name, xdata);
         return 0;
  err:
@@ -2860,7 +2550,7 @@ quota_statfs_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         list_for_each_entry (limit_node, &priv->limit_head, limit_list) {
 		/* Notice that this only works for volume-level quota. */
                 if (strcmp (limit_node->path, "/") == 0) {
-                        blocks = limit_node->value / buf->f_bsize;
+                        blocks = limit_node->hard_lim / buf->f_bsize;
                         if (usage > blocks) {
                                 break;
                         }
@@ -2895,11 +2585,13 @@ unwind:
 int32_t
 quota_statfs (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
 {
+        quota_priv_t       *priv        = NULL;
 	inode_t *root_inode = NULL;
-        quota_priv_t    *priv = NULL;
 
         priv = this->private;
 
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
 	if (priv->consider_statfs && loc->inode) {
 		root_inode = loc->inode->table->root;
 		inode_ref(root_inode);
@@ -2921,6 +2613,7 @@ quota_statfs (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xdata)
                 if (priv->consider_statfs)
                         gf_log(this->name,GF_LOG_WARNING,
                                "missing inode, cannot adjust for quota");
+wind:
 		STACK_WIND (frame, default_statfs_cbk, FIRST_CHILD(this),
 			    FIRST_CHILD(this)->fops->statfs, loc, xdata);
 	}
@@ -2951,8 +2644,13 @@ int
 quota_readdirp (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
                 off_t offset, dict_t *dict)
 {
+        quota_priv_t       *priv        = NULL;
         int ret = 0;
 
+        priv = this->private;
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
         if (dict) {
                 ret = dict_set_uint64 (dict, QUOTA_SIZE_KEY, 0);
                 if (ret < 0) {
@@ -2960,9 +2658,11 @@ quota_readdirp (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
                 }
         }
 
-        STACK_WIND (frame, quota_readdirp_cbk,
-                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->readdirp,
-                    fd, size, offset, dict);
+wind:
+        STACK_WIND (frame,
+                    priv->is_quota_on? quota_readdirp_cbk: default_readdirp_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->readdirp, fd,
+                    size, offset, dict);
         return 0;
 err:
         STACK_UNWIND_STRICT (readdirp, frame, -1, EINVAL, NULL, NULL);
@@ -3023,34 +2723,6 @@ out:
 }
 
 int32_t
-quota_fallocate_helper(call_frame_t *frame, xlator_t *this, fd_t *fd,
-		       int32_t mode, off_t offset, size_t len, dict_t *xdata)
-{
-        quota_local_t *local    = NULL;
-        int32_t        op_errno = EINVAL;
-
-        local = frame->local;
-        if (local == NULL) {
-                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
-                goto unwind;
-        }
-
-        if (local->op_ret == -1) {
-                op_errno = local->op_errno;
-                goto unwind;
-        }
-
-        STACK_WIND (frame, quota_fallocate_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->fallocate, fd, mode, offset, len,
-		    xdata);
-        return 0;
-
-unwind:
-        QUOTA_STACK_UNWIND (fallocate, frame, -1, op_errno, NULL, NULL, NULL);
-        return 0;
-}
-
-int32_t
 quota_fallocate(call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t mode,
 		off_t offset, size_t len, dict_t *xdata)
 {
@@ -3059,9 +2731,13 @@ quota_fallocate(call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t mode,
         quota_local_t     *local   = NULL;
         quota_inode_ctx_t *ctx     = NULL;
         quota_priv_t      *priv    = NULL;
-        call_stub_t       *stub    = NULL;
         quota_dentry_t    *dentry  = NULL;
 
+        priv = this->private;
+        GF_VALIDATE_OR_GOTO (this->name, priv, unwind);
+
+        WIND_IF_QUOTAOFF (priv->is_quota_on, wind);
+
         GF_ASSERT (frame);
         GF_VALIDATE_OR_GOTO ("quota", this, unwind);
         GF_VALIDATE_OR_GOTO (this->name, fd, unwind);
@@ -3074,7 +2750,7 @@ quota_fallocate(call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t mode,
         frame->local = local;
         local->loc.inode = inode_ref (fd->inode);
 
-        ret = quota_inode_ctx_get (fd->inode, -1, this, NULL, NULL, &ctx, 0);
+        ret = quota_inode_ctx_get (fd->inode, -1, -1, this, NULL, NULL, &ctx, 0);
         if (ctx == NULL) {
                 gf_log (this->name, GF_LOG_WARNING,
                         "quota context not set in inode (gfid:%s)",
@@ -3082,15 +2758,6 @@ quota_fallocate(call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t mode,
                 goto unwind;
         }
 
-        stub = fop_fallocate_stub(frame, quota_fallocate_helper, fd, mode, offset, len,
-				  xdata); 
-        if (stub == NULL) {
-                op_errno = ENOMEM;
-                goto unwind;
-        }
-
-        priv = this->private;
-        GF_VALIDATE_OR_GOTO (this->name, priv, unwind);
 
         LOCK (&ctx->lock);
         {
@@ -3106,33 +2773,22 @@ quota_fallocate(call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t mode,
 	 * in ENOSPC errors attempting to allocate an already allocated range.
 	 */
         local->delta = len;
-        local->stub = stub;
         local->link_count = parents;
 
         list_for_each_entry (dentry, &ctx->parents, next) {
                 ret = quota_check_limit (frame, fd->inode, this, dentry->name,
                                          dentry->par);
                 if (ret == -1) {
-                        break;
+                        op_errno = EDQUOT;
+                        goto unwind;
                 }
         }
 
-        stub = NULL;
-
-        LOCK (&local->lock);
-        {
-                local->link_count = 0;
-                if (local->validate_count == 0) {
-                        stub = local->stub;
-                        local->stub = NULL;
-                }
-        }
-        UNLOCK (&local->lock);
-
-        if (stub != NULL) {
-                call_resume (stub);
-        }
-
+wind:
+        STACK_WIND (frame, priv->is_quota_on?
+                    quota_fallocate_cbk: default_fallocate_cbk,
+                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->fallocate, fd,
+                    mode, offset, len, xdata);
         return 0;
 
 unwind:
@@ -3164,11 +2820,16 @@ mem_acct_init (xlator_t *this)
 int32_t
 quota_forget (xlator_t *this, inode_t *inode)
 {
+        quota_priv_t         *priv    = NULL;
         int32_t               ret     = 0;
         uint64_t              ctx_int = 0;
         quota_inode_ctx_t    *ctx     = NULL;
         quota_dentry_t       *dentry  = NULL, *tmp;
 
+        priv = this->private;
+        if (!priv->is_quota_on)
+                return 0;
+
         ret = inode_ctx_del (inode, this, &ctx_int);
 
         if (ret < 0) {
@@ -3203,68 +2864,93 @@ quota_parse_limits (quota_priv_t *priv, xlator_t *this, dict_t *xl_options,
         char         *path      = NULL, *saveptr = NULL;
         uint64_t      value     = 0;
         limits_t     *quota_lim = NULL, *old = NULL;
-        char         *last_colon= NULL;
+        double        soft_l    = 0;
+        char         *limit_dir = NULL;
+        char         *saveptr_dir       = NULL;
+        char         *path_str  = NULL;
 
         ret = dict_get_str (xl_options, "limit-set", &str);
 
-        if (str) {
-                path = strtok_r (str, ",", &saveptr);
+        if (ret) {
+                gf_log (this->name, GF_LOG_INFO, "could not get the limits");
+                /* limit may not be set at all on the volume yet */
+                ret = 0;
+                goto err;
+        }
 
-                while (path) {
-                        last_colon = strrchr (path, ':');
-                        *last_colon = '\0';
-                        str_val = last_colon + 1;
+        path_str = gf_strdup (str);
+        if (!path_str)
+                goto err;
 
-                        ret = gf_string2bytesize (str_val, &value);
-                        if (ret != 0)
-                                goto err;
 
-                        QUOTA_ALLOC_OR_GOTO (quota_lim, limits_t, err);
+        limit_dir = strtok_r (path_str, ",", &saveptr);
 
-                        quota_lim->path = path;
+        while (limit_dir) {
+                QUOTA_ALLOC_OR_GOTO (quota_lim, limits_t, err);
+                saveptr_dir = NULL;
 
-                        quota_lim->value = value;
+                path = strtok_r (limit_dir, ":", &saveptr_dir);
 
-                        gf_log (this->name, GF_LOG_INFO, "%s:%"PRId64,
-                                quota_lim->path, quota_lim->value);
-
-                        if (old_list != NULL) {
-                                list_for_each_entry (old, old_list,
-                                                     limit_list) {
-                                        if (strcmp (old->path, quota_lim->path)
-                                            == 0) {
-                                                uuid_copy (quota_lim->gfid,
-                                                           old->gfid);
-                                                break;
-                                        }
-                                }
-                        }
+                str_val = strtok_r (NULL, ":", &saveptr_dir);
 
-                        LOCK (&priv->lock);
-                        {
-                                list_add_tail (&quota_lim->limit_list,
-                                               &priv->limit_head);
+                ret = gf_string2bytesize (str_val, &value);
+                if (ret != 0)
+                        goto err;
+
+                quota_lim->hard_lim = value;
+
+                str_val = strtok_r (NULL, ",", &saveptr_dir);
+
+                soft_l = priv->default_soft_lim;
+                if (str_val) {
+                        ret = gf_string2percent (str_val, &soft_l);
+                        if (ret)
+                                gf_log (this->name, GF_LOG_WARNING,
+                                        "Failed to convert str to "
+                                        "percent. Using default soft "
+                                        "limit");
+                }
+
+                quota_lim->soft_lim = soft_l;
+
+                quota_lim->path = gf_strdup (path);
+
+                gf_log (this->name, GF_LOG_INFO, "%s:%"PRId64,
+                        quota_lim->path, quota_lim->hard_lim);
+
+                if (old_list != NULL) {
+                        list_for_each_entry (old, old_list,
+                                             limit_list) {
+                                if (strcmp (old->path, quota_lim->path) == 0) {
+                                        uuid_copy (quota_lim->gfid,
+                                                   old->gfid);
+                                        break;
+                                }
                         }
-                        UNLOCK (&priv->lock);
+                }
 
-                        path = strtok_r (NULL, ",", &saveptr);
+                LOCK (&priv->lock);
+                {
+                        list_add_tail (&quota_lim->limit_list,
+                                       &priv->limit_head);
                 }
-        } else {
-                gf_log (this->name, GF_LOG_INFO,
-                        "no \"limit-set\" option provided");
+                UNLOCK (&priv->lock);
+
+                limit_dir = strtok_r (NULL, ",", &saveptr);
         }
 
         LOCK (&priv->lock);
         {
                 list_for_each_entry (quota_lim, &priv->limit_head, limit_list) {
                         gf_log (this->name, GF_LOG_INFO, "%s:%"PRId64,
-                                quota_lim->path, quota_lim->value);
+                                quota_lim->path, quota_lim->hard_lim);
                 }
         }
         UNLOCK (&priv->lock);
 
         ret = 0;
 err:
+        GF_FREE (path_str);
         return ret;
 }
 
@@ -3302,8 +2988,10 @@ init (xlator_t *this)
                 goto err;
         }
 
-        GF_OPTION_INIT ("timeout", priv->timeout, int64, err);
         GF_OPTION_INIT ("deem-statfs", priv->consider_statfs, bool, err);
+        GF_OPTION_INIT ("server-quota", priv->is_quota_on, bool, err);
+        GF_OPTION_INIT ("default-soft-limit", priv->default_soft_lim, percent,
+                        err);
 
         this->local_pool = mem_pool_new (quota_local_t, 64);
         if (!this->local_pool) {
@@ -3329,8 +3017,8 @@ __quota_reconfigure_inode_ctx (xlator_t *this, inode_t *inode, limits_t *limit)
         GF_VALIDATE_OR_GOTO (this->name, inode, out);
         GF_VALIDATE_OR_GOTO (this->name, limit, out);
 
-        ret = quota_inode_ctx_get (inode, limit->value, this, NULL, NULL, &ctx,
-                                   1);
+        ret = quota_inode_ctx_get (inode, limit->hard_lim, limit->soft_lim,
+                                   this, NULL, NULL, &ctx, 1);
         if ((ret == -1) || (ctx == NULL)) {
                 gf_log (this->name, GF_LOG_WARNING, "cannot create quota "
                         "context in inode(gfid:%s)",
@@ -3340,7 +3028,8 @@ __quota_reconfigure_inode_ctx (xlator_t *this, inode_t *inode, limits_t *limit)
 
         LOCK (&ctx->lock);
         {
-                ctx->limit = limit->value;
+                ctx->hard_lim = limit->hard_lim;
+                ctx->soft_lim = limit->soft_lim;
         }
         UNLOCK (&ctx->lock);
 
@@ -3385,6 +3074,13 @@ reconfigure (xlator_t *this, dict_t *options)
 
         priv = this->private;
 
+        GF_OPTION_RECONF ("deem-statfs", priv->consider_statfs, options, bool,
+                          out);
+        GF_OPTION_RECONF ("server-quota", priv->is_quota_on, options, bool,
+                          out);
+        GF_OPTION_RECONF ("default-soft-limit", priv->default_soft_lim,
+                          options, percent, out);
+
         INIT_LIST_HEAD (&head);
 
         LOCK (&priv->lock);
@@ -3421,7 +3117,7 @@ reconfigure (xlator_t *this, dict_t *options)
                         }
 
                         if (!found) {
-                                limit->value = -1;
+                                limit->hard_lim = -1;
                                 __quota_reconfigure (this, top->itable, limit);
                         }
 
@@ -3431,9 +3127,7 @@ reconfigure (xlator_t *this, dict_t *options)
         }
         UNLOCK (&priv->lock);
 
-        GF_OPTION_RECONF ("timeout", priv->timeout, options, int64, out);
-        GF_OPTION_RECONF ("deem-statfs", priv->consider_statfs, options, bool,
-                          out);
+
 
         ret = 0;
 out:
@@ -3484,6 +3178,25 @@ struct xlator_cbks cbks = {
 
 struct volume_options options[] = {
         {.key = {"limit-set"}},
+        {.key = {"deem-statfs"},
+         .type = GF_OPTION_TYPE_BOOL,
+         .default_value = "off",
+         .description = "If set to on, it takes quota limits into"
+                        "consideration while estimating fs size. (df command)"
+                        " (Default is off)."
+        },
+        {.key = {"server-quota"},
+         .type = GF_OPTION_TYPE_BOOL,
+         .default_value = "off",
+         .description = "Skip the quota if xlator if the feature is not "
+                        "turned on. This is not a user exposed option."
+        },
+        {.key = {"default-soft-limit"},
+         .type = GF_OPTION_TYPE_PERCENT,
+         .default_value = "90%",
+         .min = 0,
+         .max = LONG_MAX,
+        },
         {.key = {"timeout"},
          .type = GF_OPTION_TYPE_SIZET,
          .min = 0,
@@ -3492,12 +3205,5 @@ struct volume_options options[] = {
          .description = "quota caches the directory sizes on client. Timeout "
                         "indicates the timeout for the cache to be revalidated."
         },
-        {.key = {"deem-statfs"},
-         .type = GF_OPTION_TYPE_BOOL,
-         .default_value = "off",
-         .description = "If set to on, it takes quota limits into"
-                        "consideration while estimating fs size. (df command)"
-                        " (Default is off)."
-        },
         {.key = {NULL}}
 };
diff --git a/xlators/features/quota/src/quota.h b/xlators/features/quota/src/quota.h
index 84ecbb30..de9f6f16 100644
--- a/xlators/features/quota/src/quota.h
+++ b/xlators/features/quota/src/quota.h
@@ -25,6 +25,14 @@
 #define CONTRIBUTION            "contri"
 #define VAL_LENGTH              8
 #define READDIR_BUF             4096
+#define QUOTA_UPDATE_USAGE_KEY  "quota-update-usage"
+
+#define WIND_IF_QUOTAOFF(is_quota_on, label)     \
+        if (!is_quota_on)                       \
+                goto label;
+
+#define DID_REACH_LIMIT(lim, prev_size, cur_size)               \
+        ((cur_size) >= (lim) && (prev_size) < (lim))
 
 #define QUOTA_SAFE_INCREMENT(lock, var)         \
         do {                                    \
@@ -46,7 +54,7 @@
                                  gf_quota_mt_##type);   \
                 if (!var) {                             \
                         gf_log ("", GF_LOG_ERROR,       \
-                                "out of memory :(");    \
+                                "out of memory");    \
                         ret = -1;                       \
                         goto label;                     \
                 }                                       \
@@ -96,6 +104,8 @@
                         goto label;                             \
         } while (0)
 
+
+
 struct quota_dentry {
         char            *name;
         uuid_t           par;
@@ -105,7 +115,8 @@ typedef struct quota_dentry quota_dentry_t;
 
 struct quota_inode_ctx {
         int64_t          size;
-        int64_t          limit;
+        int64_t          hard_lim;
+        int64_t          soft_lim;
         struct iatt      buf;
         struct list_head parents;
         struct timeval   tv;
@@ -125,27 +136,52 @@ struct quota_local {
         int32_t      op_ret;
         int32_t      op_errno;
         int64_t      size;
-        int64_t      limit;
+        int64_t      hard_lim;
+        int64_t      soft_lim;
         char         just_validated;
         inode_t     *inode;
         call_stub_t *stub;
 };
 typedef struct quota_local quota_local_t;
 
+
+struct qd_vols_conf {
+        char                    *name;
+        inode_table_t           *itable;
+        uint32_t                 log_timeout;
+        gf_boolean_t             threads_status;
+        double                   default_soft_lim;
+        gf_lock_t                lock;
+        loc_t                    root_loc;
+	uint32_t		soft_timeout;
+	uint32_t		hard_timeout;
+	struct list_head	limit_head;
+	call_frame_t		*frame;
+};
+typedef struct qd_vols_conf qd_vols_conf_t;
+
+
 struct quota_priv {
-        int64_t           timeout;
-        gf_boolean_t      consider_statfs;
-        struct list_head  limit_head;
-        gf_lock_t         lock;
+        int64_t                 timeout;
+        double                  default_soft_lim;
+        gf_boolean_t            is_quota_on;
+        gf_boolean_t            consider_statfs;
+        struct list_head        limit_head;
+        qd_vols_conf_t        **qd_vols_conf;
+        gf_lock_t               lock;
 };
 typedef struct quota_priv quota_priv_t;
 
+
 struct limits {
         struct list_head  limit_list;
         char             *path;
-        int64_t           value;
         uuid_t            gfid;
+        int64_t           prev_size;
+        struct timeval    prev_log_tv;
+        int64_t           hard_lim;
+        int64_t           soft_lim;
+	struct timeval	  expire;
+	uint32_t	  timeout;
 };
 typedef struct limits     limits_t;
-
-uint64_t cn = 1;
diff --git a/xlators/features/quota/src/quotad.c b/xlators/features/quota/src/quotad.c
new file mode 100644
index 00000000..a5e51add
--- /dev/null
+++ b/xlators/features/quota/src/quotad.c
@@ -0,0 +1,986 @@
+/*
+   Copyright (c) 2013 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+#include <fnmatch.h>
+
+#include "quota.h"
+#include "common-utils.h"
+#include "defaults.h"
+#include "syncop.h"
+#include "libgen.h"
+#include "timer.h"
+
+
+inline uint64_t
+quota_time_elapsed (struct timeval *now, struct timeval *then)
+{
+        return (now->tv_sec - then->tv_sec);
+}
+
+
+int32_t
+quota_timeout (struct timeval *tv, uint64_t timeout)
+{
+        struct timeval now       = {0,};
+        int32_t        timed_out = 0;
+
+        gettimeofday (&now, NULL);
+
+        if (quota_time_elapsed (&now, tv) >= timeout) {
+                timed_out = 1;
+        }
+
+        return timed_out;
+}
+
+/* Returns itable->root, also creates itable if not present */
+inode_t *
+qd_build_root_inode (xlator_t *this, qd_vols_conf_t *this_vol)
+{
+        if (!this_vol->itable) {
+                this_vol->itable = inode_table_new (0, this);
+                if (!this_vol->itable)
+                        return NULL;
+        }
+
+        return inode_ref (this_vol->itable->root);
+}
+
+int
+qd_resolve_root (xlator_t *subvol, loc_t *root_loc,
+                 struct iatt *iatt, dict_t *dict_req, dict_t **dict_rsp)
+{
+        int     ret             = -1;
+
+        ret = syncop_lookup (subvol, root_loc, dict_req, iatt, dict_rsp, NULL);
+        if (-1 == ret) {
+                gf_log (subvol->name, GF_LOG_ERROR, "Received %s for lookup "
+                         "on / (vol:%s)", strerror (errno), subvol->name);
+        }
+        return ret;
+}
+
+int
+qd_build_root_loc (xlator_t *this, xlator_t *subvol, inode_t *inode, loc_t *loc)
+{
+
+        loc->path = gf_strdup ("/");
+        loc->inode = inode;
+        uuid_clear (loc->gfid);
+        loc->gfid[15] = 1;
+
+        return qd_resolve_root (subvol, loc, NULL, NULL, NULL);
+}
+
+
+int32_t
+mem_acct_init (xlator_t *this)
+{
+        int     ret = -1;
+
+        if (!this)
+                return ret;
+
+        ret = xlator_mem_acct_init (this, gf_quota_mt_end + 1);
+
+        if (0 != ret) {
+                gf_log (this->name, GF_LOG_WARNING, "Memory accounting "
+                        "init failed");
+                return ret;
+        }
+
+        return ret;
+}
+
+/**
+ * Takes the limit string, parse it, fill limits_t struct and insert into
+ * above-soft list.
+ *
+ * Format for limit string
+ * <limit-string> = <limit-on-single-dir>[,<limit-on-single-dir>]*
+ * <limit-on-single-dir> =
+ *                    <abs-path-from-volume-root>:<hard-limit>[:<soft-limit-%>]
+ */
+int
+qd_parse_limits (quota_priv_t *priv, xlator_t *this, char *limit_str,
+                    struct list_head *old_list, qd_vols_conf_t *this_vol)
+{
+        int32_t       ret       = -1;
+        char         *str_val   = NULL;
+        char         *path      = NULL, *saveptr = NULL;
+        uint64_t      value     = 0;
+        limits_t     *quota_lim = NULL;
+        char         *str       = NULL;
+        double        soft_l    = -1;
+        char         *limit_on_dir      = NULL;
+        char         *saveptr_dir       = NULL;
+
+        str = gf_strdup (limit_str);
+
+        if (str) {
+                limit_on_dir = strtok_r (str, ",", &saveptr);
+
+                while (limit_on_dir) {
+                        QUOTA_ALLOC_OR_GOTO (quota_lim, limits_t, err);
+                        gettimeofday (&quota_lim->prev_log_tv, NULL);
+
+                        saveptr_dir = NULL;
+
+                        path = strtok_r (limit_on_dir, ":", &saveptr_dir);
+
+                        str_val = strtok_r (NULL, ":", &saveptr_dir);
+
+                        ret = gf_string2bytesize (str_val, &value);
+                        if (0 != ret)
+                                goto err;
+
+                        quota_lim->hard_lim = value;
+
+                        str_val = strtok_r (NULL, ",", &saveptr_dir);
+
+                        soft_l = this_vol->default_soft_lim;
+                        if (str_val) {
+                                ret = gf_string2percent (str_val, &soft_l);
+                                if (ret)
+                                        gf_log (this->name, GF_LOG_WARNING,
+                                                "Failed to convert str to "
+                                                "percent. Using default soft "
+                                                "limit");
+                        }
+
+                        quota_lim->soft_lim = (int64_t) quota_lim->hard_lim *
+                                                                (soft_l / 100);
+
+                        quota_lim->path = gf_strdup (path);
+                        if (!quota_lim->path) {
+                                gf_log (this->name, GF_LOG_ERROR, "ENOMEM "
+                                        "Received copying the path");
+                                goto err;
+                        }
+
+                        quota_lim->prev_size = -1;
+
+                        gf_log (this->name, GF_LOG_INFO, "%s:%"PRId64":%"PRId32,
+                                quota_lim->path, quota_lim->hard_lim,
+                                (int)quota_lim->soft_lim);
+
+                        /* This is used in the reconfigure path, so not used
+                         * by quotad as of now.
+                        if (NULL != old_list) {
+                                list_for_each_entry (old, old_list,
+                                                     limit_list) {
+                                        if (0 ==
+                                            strcmp (old->path, quota_lim->path)) {
+                                                uuid_copy (quota_lim->gfid,
+                                                           old->gfid);
+                                                break;
+                                        }
+                                }
+                        } */
+
+                        LOCK (&this_vol->lock);
+                        {
+                                list_add_tail (&quota_lim->limit_list,
+                                               &this_vol->limit_head);
+                        }
+                        UNLOCK (&this_vol->lock);
+
+                        limit_on_dir = strtok_r (NULL, ",", &saveptr);
+                }
+        } else {
+                gf_log (this->name, GF_LOG_INFO,
+                        "no \"limit-set\" option provided");
+        }
+
+        ret = 0;
+err:
+        GF_FREE (str);
+        return ret;
+}
+
+
+xlator_t *
+qd_get_subvol (xlator_t *this, qd_vols_conf_t *this_vol)
+{
+        xlator_list_t   *subvol        = NULL;
+
+        for (subvol = this->children; subvol; subvol = subvol->next)
+                if (0 == strcmp (subvol->xlator->name, this_vol->name))
+                        return subvol->xlator;
+
+        return NULL;
+}
+
+/* Logs if
+ *  i.   Usage crossed soft limit
+ *  ii.  Usage above soft limit and alert-time timed out
+ */
+void
+qd_log_usage (xlator_t *this, qd_vols_conf_t *conf, limits_t *entry,
+              int64_t cur_size)
+{
+        struct timeval           cur_time       = {0,};
+        char                    *usage_str      = NULL;
+
+        gettimeofday (&cur_time, NULL);
+
+        /* Usage crossed/reached soft limit? */
+        if (DID_REACH_LIMIT (entry->soft_lim, entry->prev_size, cur_size)) {
+                entry->prev_log_tv = cur_time;
+                usage_str = gf_uint64_2human_readable (entry->soft_lim);
+                gf_log (this->name, GF_LOG_ALERT, "Usage crossed soft limit: "
+                        "%s for %s (vol:%s)", usage_str, entry->path,
+                        conf->name);
+        }
+        /* Usage crossed/reached hard limit? */
+        else if (DID_REACH_LIMIT (entry->hard_lim, entry->prev_size,
+                                                                    cur_size)) {
+                entry->prev_log_tv = cur_time;
+                usage_str = gf_uint64_2human_readable (entry->hard_lim);
+                gf_log (this->name, GF_LOG_ALERT, "Usage reached hard limit: "
+                        "%s for %s (vol:%s)", usage_str, entry->path,
+                        conf->name);
+        }
+        /* Usage is above soft limit and 'alert-time' timed out */
+        else if (cur_size > entry->soft_lim &&
+                       quota_timeout (&entry->prev_log_tv, conf->log_timeout)) {
+                entry->prev_log_tv = cur_time;
+                usage_str = gf_uint64_2human_readable (cur_size);
+                gf_log (this->name, GF_LOG_ALERT, "Usage %s %s limit for %s "
+                        "(vol:%s)", usage_str, (cur_size < entry->hard_lim)?
+                        "is above soft": "has reached hard", entry->path,
+                        conf->name);
+        }
+        if (usage_str)
+                GF_FREE (usage_str);
+}
+
+int
+qd_build_child_loc (loc_t *child, loc_t *parent, char *name)
+{
+        if (!child) {
+                goto err;
+        }
+
+        if (strcmp (parent->path, "/") == 0)
+                gf_asprintf ((char **)&child->path, "/%s", name);
+        else
+                gf_asprintf ((char **)&child->path, "%s/%s", parent->path,
+                             name);
+
+        if (!child->path) {
+                goto err;
+        }
+
+        child->name = strrchr (child->path, '/');
+        if (child->name)
+                child->name++;
+
+        child->parent = inode_ref (parent->inode);
+        if (!child->inode)
+                child->inode = inode_new (parent->inode->table);
+
+        if (!child->inode) {
+                goto err;
+        }
+        if (!uuid_is_null(parent->gfid))
+                uuid_copy (child->pargfid, parent->gfid);
+        else
+                uuid_copy (child->pargfid, parent->inode->gfid);
+
+        return 0;
+err:
+        loc_wipe (child);
+        return -1;
+}
+
+#define QUOTA_IDLE_TIMEOUT 30
+int
+qd_check_enforce (qd_vols_conf_t *conf, dict_t *dict, limits_t *entry,
+                  loc_t *entry_loc, xlator_t *subvol)
+{
+        xlator_t        *this = THIS;
+        int64_t         *size = NULL;
+        int              ret  = -1;
+        int64_t          cur_size       = 0;
+        int64_t          tmp_size       = 0;
+        dict_t          *setxattr_dict  = NULL;
+        int64_t         *quota_xlator_size = NULL;
+
+        ret = dict_get_bin (dict, QUOTA_SIZE_KEY, (void **)&size);
+        if (ret) {
+                gf_log (this->name, GF_LOG_WARNING, "Couldn't get size"
+                        " from the dict (%s)", entry->path);
+                goto out;
+        }
+
+        cur_size = ntoh64 (*size);
+
+        qd_log_usage (this, conf, entry, cur_size);
+
+        /* if size hasn't changed, skip the update and bump the timeout */
+        if (entry->prev_size == cur_size) {
+                /* Because of dht we might be getting the correct aggregated
+                   size. But the quota xlator in the server graph might not
+                   be having the right value. So get the size that quota xlator
+                   is maintaining in its context. If it is different than the
+                   total aggregated size, then update the quota xlator with
+                   the correct size through setxattr.
+                */
+                ret = dict_get_bin (dict, "trusted.limit.set",
+                                    (void **)&quota_xlator_size);
+                if (ret)
+                        gf_log (this->name, GF_LOG_WARNING, "Couldn't get size"
+                                " from the dict (%s)", entry->path);
+                else
+                        tmp_size = ntoh64 (*quota_xlator_size);
+
+                if (entry->prev_size == tmp_size) {
+                        entry->timeout = min((entry->timeout << 1),
+                                             QUOTA_IDLE_TIMEOUT);
+                        goto out;
+                }
+	}
+
+        QUOTA_ALLOC_OR_GOTO (size, int64_t, out);
+        *size = hton64 (cur_size);
+
+        setxattr_dict = dict_new();
+        ret = dict_set_bin (setxattr_dict, QUOTA_UPDATE_USAGE_KEY, size,
+                           sizeof (int64_t));
+        if (-1 == ret) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "Couldn't set dict");
+                goto out;
+        }
+
+
+        /* There is a possibility that, after a setxattr is done,
+           a rename might happen and the resolve will fail again.
+        */
+        ret = syncop_setxattr (subvol, entry_loc, setxattr_dict, 0);
+        if (ret) {
+		if (errno == ESTALE || errno == ENOENT)
+			inode_forget (entry_loc->inode, 0);
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Received ERROR:%s in updating quota value %s "
+                        " (vol:%s). Quota enforcement may not be"
+                        " accurate", strerror (errno), entry->path,
+                         subvol->name);
+                goto out;
+        }
+
+	LOCK(&conf->lock);
+	/* set the timeout based on usage */
+	if (cur_size < entry->soft_lim)
+		entry->timeout = conf->soft_timeout;
+	else
+		entry->timeout = conf->hard_timeout;
+
+	entry->prev_size = cur_size;
+	UNLOCK(&conf->lock);
+
+out:
+        if (setxattr_dict)
+                dict_unref (setxattr_dict);
+
+        return ret;
+}
+
+int
+qd_resolve_handle_error (loc_t *comp_loc, inode_t *inode,
+                         int op_ret, int op_errno)
+{
+
+        if (op_ret) {
+                switch (op_errno) {
+                case ESTALE:
+                case ENOENT:
+			if (comp_loc->inode)
+				inode_forget (comp_loc->inode, 0);
+			break;
+                default:
+                        gf_log ("", GF_LOG_ERROR, "lookup on %s returned %s",
+                                comp_loc->path, strerror(op_errno));
+                }
+        }
+
+        return 0;
+}
+
+int
+qd_resolve_entry (qd_vols_conf_t *conf, xlator_t *subvol, limits_t *entry,
+                   dict_t *dict_req, loc_t *loc, dict_t **dict_rsp,
+                   int force)
+{
+        char            *component      = NULL;
+        char            *next_comp      = NULL;
+        char            *saveptr        = NULL;
+        char            *path           = NULL;
+        int              ret            = -1;
+        struct iatt      iatt           = {0,};
+        inode_t         *inode          = NULL;
+        dict_t          *tmp_dict       = NULL;
+        loc_t            comp_loc       = {0,};
+        loc_t            par_loc        = {0,};
+        int              need_lookup    = 0;
+
+        path = gf_strdup (entry->path);
+
+        loc_copy (&par_loc, &conf->root_loc);
+        for (component = strtok_r (path, "/", &saveptr);
+	     component; component = next_comp) {
+
+		next_comp = strtok_r (NULL, "/", &saveptr);
+
+                inode = inode_grep (conf->itable,
+                                    par_loc.inode,
+                                    component);
+
+                /* No need to forget and unref the inode if revalidate flag
+                   is set (i.e @force). Reason:
+                   Just because we want to revalidate, does not mean
+                   the inode is not valid. For invalidating the inode,
+                   (i.e forgetting it) let lookup be sent, and the decision
+                   should be made depending upon lookup's reply.
+                */
+
+                /* if inode is found, then skip lookup unless last component */
+                if (inode) {
+                        comp_loc.inode = inode;
+                        need_lookup = 0;
+                } else {
+                        need_lookup = 1;
+                }
+
+                qd_build_child_loc (&comp_loc, &par_loc, component);
+                /* Get the xattrs in lookup for the last component */
+                if (!next_comp) {
+                        tmp_dict = dict_req;
+                        need_lookup = 1;
+                }
+
+                if (need_lookup || force) {
+                        ret = syncop_lookup (subvol, &comp_loc, tmp_dict, &iatt,
+                                             dict_rsp, NULL);
+                        if (ret) {
+                                /* invalidate inode got from inode_grep if
+                                 * ESTALE/ENOENT */
+                                qd_resolve_handle_error (&comp_loc, inode,
+                                                         ret, errno);
+                                goto out;
+                        }
+
+                        if (!IA_ISDIR (iatt.ia_type)) {
+			        gf_log (subvol->name, GF_LOG_ERROR,
+                                        "%s is not a directory",
+                                        comp_loc.path);
+			        goto out;
+		        }
+
+                        inode = inode_link (comp_loc.inode, par_loc.inode,
+                                            component, &iatt);
+                        comp_loc.inode = inode;
+                        inode_lookup (comp_loc.inode);
+                        inode_unref (inode);
+                }
+                inode = NULL;
+                loc_wipe (&par_loc);
+                loc_copy (&par_loc, &comp_loc);
+                loc_wipe (&comp_loc);
+	}
+        ret = 0;
+        loc_copy (loc, &par_loc);
+
+out:
+        GF_FREE(path);
+        loc_wipe (&par_loc);
+        if (ret)
+                loc_wipe (&comp_loc);
+        return ret;
+}
+
+int
+qd_handle_entry (qd_vols_conf_t *conf, xlator_t *subvol, limits_t *entry,
+                 dict_t *dict_req, int revalidate)
+{
+        dict_t          *dict_rsp  = NULL;
+        loc_t            entry_loc = {0,};
+        int              ret       = -1;
+
+        if (!strcmp (entry->path, "/")) {
+                ret = qd_resolve_root (subvol, &conf->root_loc, NULL, dict_req,
+                                       &dict_rsp);
+                if (ret) {
+                        gf_log (subvol->name, GF_LOG_ERROR, "lookup on / "
+                                "(%s)", strerror(errno));
+                        goto err;
+                }
+                loc_copy (&entry_loc, &conf->root_loc);
+
+        } else {
+                ret = qd_resolve_entry (conf, subvol, entry, dict_req,
+                                        &entry_loc, &dict_rsp, revalidate);
+                /* if resolve failed, force resolve from "/" once */
+                if (ret) {
+                        if (errno == ENOENT)
+                                entry->timeout = QUOTA_IDLE_TIMEOUT;
+			else
+				gf_log (subvol->name, GF_LOG_ERROR,
+					"Quota check on %s failed ERR:%s",
+					entry->path, strerror (errno));
+                        if (!revalidate) {
+                                ret = qd_handle_entry (conf, subvol, entry,
+                                                       dict_req, 1);
+                                if (ret)
+                                        goto err;
+                        }
+                        goto err;
+                }
+        }
+        ret = qd_check_enforce (conf, dict_rsp, entry, &entry_loc, subvol);
+        if (ret)
+                gf_log (subvol->name, GF_LOG_ERROR,
+                        "Failed to enforce quota on %s", entry_loc.path);
+err:
+        loc_wipe (&entry_loc);
+        if (dict_rsp)
+                dict_unref (dict_rsp);
+        return 0;
+}
+
+int
+qd_iterator (qd_vols_conf_t  *conf, xlator_t *subvol)
+{
+        limits_t        *entry          = NULL;
+        limits_t        *next           = NULL;
+        int32_t          ret;
+        dict_t          *dict_req       = NULL;
+	struct timeval	now;
+	struct timeval	next_expire = {0,};
+
+        GF_VALIDATE_OR_GOTO ("qd-iterator", conf, out);
+        GF_VALIDATE_OR_GOTO ("qd-iterator", subvol, out);
+
+        dict_req = dict_new ();
+        GF_VALIDATE_OR_GOTO ("qd-iterator", dict_req, out);
+        ret = dict_set_uint64 (dict_req, QUOTA_SIZE_KEY, 0);
+        if (ret) {
+                gf_log ("qd-iterator", GF_LOG_ERROR, "dict set failed for "
+                        "QUOTA SIZE key");
+                goto out;
+        }
+
+	gettimeofday(&now, NULL);
+        list_for_each_entry_safe (entry, next, &conf->limit_head, limit_list) {
+		/* just use the same start time across the entire iteration */
+		if (timercmp(&now, &entry->expire, <))
+			goto check_next_expire;
+
+                ret = qd_handle_entry (conf, subvol, entry, dict_req, 0);
+                if (ret) {
+                        gf_log ("qd-iterator", GF_LOG_ERROR, "Failed to check "
+                                "quota limit on %s", entry->path);
+                }
+
+		gettimeofday(&entry->expire, NULL);
+		entry->expire.tv_sec += entry->timeout;
+		gf_log("qd-iterator", GF_LOG_TRACE,
+			"volume %s path %s usage %lu (hard %lu soft %lu) expire in %ds",
+			conf->name, entry->path, entry->prev_size,
+			entry->hard_lim, entry->soft_lim, entry->timeout);
+
+check_next_expire:
+		if (!next_expire.tv_sec ||
+		    timercmp(&entry->expire, &next_expire, <))
+			next_expire = entry->expire;
+        }
+
+	ret = 0;
+	/* return a hint as to when we'll next have something to do */
+	gettimeofday(&now, NULL);
+	if (timercmp(&now, &next_expire, <)) {
+		timersub(&next_expire, &now, &next_expire);
+		ret = next_expire.tv_sec;
+                if (next_expire.tv_usec)
+                        ret++;
+	}
+out:
+        if (dict_req)
+                dict_unref (dict_req);
+
+        return ret;
+}
+
+int
+qd_trigger_periodically (void *args)
+{
+        int                      ret            = -1;
+        xlator_t                *this           = NULL;
+        xlator_t                *subvol         = NULL;
+        inode_t                 *root_inode     = NULL;
+        qd_vols_conf_t          *conf           = NULL;
+
+        this = THIS;
+        conf = args;
+
+        subvol = qd_get_subvol (this, conf);
+        if (!subvol) {
+                gf_log (this->name, GF_LOG_ERROR, "No subvol found");
+                return -1;
+        }
+
+        if (!conf->root_loc.path) {
+                root_inode = qd_build_root_inode (this, conf);
+                if (!root_inode) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "New itable creation failed");
+                        return -1;
+                }
+
+                ret = qd_build_root_loc (this, subvol, root_inode,
+                        &conf->root_loc);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Failed to build root_loc for %s", conf->name);
+                        return -1;
+                }
+        }
+
+	if (list_empty(&conf->limit_head))
+		return QUOTA_IDLE_TIMEOUT;
+
+        ret = qd_iterator (conf, subvol);
+        if (ret < 0)
+                gf_log (this->name, GF_LOG_WARNING,
+                        "Couldn't update the usage, frequent "
+                        "log may lead usage to cross beyond "
+                        "limit");
+
+        return ret;
+}
+
+static void create_iter_task(void *);
+
+int
+qd_trigger_periodically_cbk(int ret, call_frame_t *frame, void *args)
+{
+	qd_vols_conf_t	*conf = args;
+	struct timeval delta = { 0, };
+
+        if (ret < 0)
+                gf_log ("quotad", GF_LOG_ERROR,
+                        "Synctask stopped unexpectedly, trying to restart");
+        else
+		delta.tv_sec = ret;
+
+	conf->frame = frame;
+	gf_timer_call_after(THIS->ctx, delta, create_iter_task, conf);
+
+        return ret;
+}
+
+static void
+create_iter_task(void *data)
+{
+	qd_vols_conf_t *conf = data;
+	int ret;
+
+	ret = synctask_new(THIS->ctx->env, qd_trigger_periodically,
+			   qd_trigger_periodically_cbk, conf->frame,
+			   conf);
+	if (ret < 0)
+		gf_log("quotad", GF_LOG_ERROR,
+		       "Synctask creation failed for %s", conf->name);
+}
+
+int
+qd_start_threads (xlator_t *this, int subvol_idx)
+{
+        quota_priv_t            *priv           = NULL;
+        int                      ret            = 0;
+        qd_vols_conf_t          *this_vol       = NULL;
+
+        priv = this->private;
+
+        this_vol = priv->qd_vols_conf[subvol_idx];
+
+        if (list_empty (&this_vol->limit_head)) {
+                gf_log (this->name, GF_LOG_DEBUG, "No limit is set on "
+                        "volume %s", this_vol->name);
+                /* Dafault ret is 0 */
+                goto err;
+        }
+
+        ret = synctask_new (this->ctx->env,
+                            qd_trigger_periodically,
+                            qd_trigger_periodically_cbk,
+                            NULL, (void *) this_vol);
+        if (-1 == ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Synctask creation "
+                        "failed for %s", this_vol->name);
+                goto err;
+        }
+err:
+        return ret;
+}
+
+int
+qd_reconfigure (xlator_t *this, dict_t *options)
+{
+        /* As of now quotad is restarted upon alteration of volfile */
+        return 0;
+}
+
+void
+qd_fini (xlator_t *this)
+{
+        return;
+}
+
+int32_t
+qd_init (xlator_t *this)
+{
+        int32_t          ret            = -1;
+        quota_priv_t    *priv           = NULL;
+        int              i              = 0;
+        char            *option_str     = NULL;
+        xlator_list_t   *subvol         = NULL;
+        char            *limits         = NULL;
+        int              subvol_cnt     = 0;
+        qd_vols_conf_t  *this_vol       = NULL;
+
+        if (NULL == this->children) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "FATAL: quota (%s) not configured for min of 1 child",
+                        this->name);
+                ret = -1;
+                goto err;
+        }
+
+        QUOTA_ALLOC_OR_GOTO (priv, quota_priv_t, err);
+
+        LOCK_INIT (&priv->lock);
+
+        this->private = priv;
+
+        for (subvol_cnt = 0, subvol = this->children;
+             subvol;
+             subvol_cnt++, subvol = subvol->next);
+
+        priv->qd_vols_conf = GF_CALLOC (sizeof (qd_vols_conf_t *), subvol_cnt,
+                                        gf_quota_mt_qd_vols_conf_t);
+        if (!priv->qd_vols_conf) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to allocate memory");
+                goto err;
+        }
+
+        for (i = 0, subvol = this->children;
+             subvol;
+             subvol = subvol->next, i++) {
+
+                QUOTA_ALLOC_OR_GOTO (priv->qd_vols_conf[i],
+                                     qd_vols_conf_t, err);
+
+                this_vol = priv->qd_vols_conf[i];
+
+                LOCK_INIT (&this_vol->lock);
+                INIT_LIST_HEAD (&this_vol->limit_head);
+
+                this_vol->name = subvol->xlator->name;
+
+                ret = gf_asprintf (&option_str, "%s.default-soft-limit",
+                                   this_vol->name);
+                if (0 > ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "gf_asprintf failed");
+                        goto err;
+                }
+                GF_OPTION_INIT (option_str, this_vol->default_soft_lim, percent,
+                                err);
+                GF_FREE (option_str);
+
+                ret = gf_asprintf (&option_str, "%s.alert-time",
+                                   this_vol->name);
+                if (0 > ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "gf_asprintf failed");
+                        goto err;
+                }
+                GF_OPTION_INIT (option_str, this_vol->log_timeout, time, err);
+                GF_FREE (option_str);
+
+                ret = gf_asprintf (&option_str, "%s.limit-set", this_vol->name);
+                if (0 > ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "gf_asprintf failed");
+                        goto err;
+                }
+                ret = dict_get_str (this->options, option_str, &limits);
+                if (ret < 0) {
+                        gf_log (this->name, GF_LOG_ERROR, "dict get failed or "
+                                "no limits set");
+                        continue;
+                }
+
+                ret = qd_parse_limits (priv, this, limits, NULL, this_vol);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Couldn't parse limits for %s", this_vol->name);
+                        goto err;
+                }
+                GF_FREE (option_str);
+
+                ret = gf_asprintf (&option_str, "%s.soft-timeout",
+                                   this_vol->name);
+                if (0 > ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "gf_asprintf failed");
+                        goto err;
+                }
+                GF_OPTION_INIT (option_str, this_vol->soft_timeout,
+                                time, err);
+                GF_FREE (option_str);
+
+                ret = gf_asprintf (&option_str, "%s.hard-timeout",
+                                   this_vol->name);
+                if (0 > ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "gf_asprintf failed");
+                        goto err;
+                }
+                GF_OPTION_INIT (option_str, this_vol->hard_timeout,
+                                time, err);
+                GF_FREE (option_str);
+        }
+
+        this->local_pool = mem_pool_new (quota_local_t, 64);
+        if (!this->local_pool) {
+                ret = -1;
+                gf_log (this->name, GF_LOG_ERROR,
+                        "failed to create local_t's memory pool");
+                goto err;
+        }
+
+        ret = 0;
+err:
+        /* Free all allocated variables */
+        if (ret) {
+                /* can reach here from GF_OPTION_INIT, so cleaning opt_str */
+                GF_FREE (option_str);
+
+                for (i = 0; i < subvol_cnt; i++)
+                        GF_FREE (priv->qd_vols_conf[i]);
+                GF_FREE (priv->qd_vols_conf);
+
+                GF_FREE (priv);
+        }
+        return ret;
+}
+
+int
+qd_notify (xlator_t *this, int event, void *data, ...)
+{
+        xlator_list_t   *subvol                 = NULL;
+        xlator_t        *subvol_rec             = NULL;
+        quota_priv_t    *priv                   = NULL;
+        int              i                      = 0;
+        int              ret                    = 0;
+
+        subvol_rec = data;
+        priv = this->private;
+
+        for (i=0, subvol = this->children; subvol; i++, subvol = subvol->next) {
+                if (! strcmp (priv->qd_vols_conf[i]->name, subvol_rec->name))
+                        break;
+        }
+        if (!subvol) {
+                default_notify (this, event, data);
+                goto out;
+        }
+
+        switch (event) {
+        case GF_EVENT_CHILD_UP:
+        {
+                /* handle spurious CHILD_UP and DOWN events */
+                if (!priv->qd_vols_conf[i]->threads_status) {
+                        ret = qd_start_threads (this, i);
+                        if (-1 == ret) {
+                                gf_log (this->name, GF_LOG_ERROR, "Couldn't "
+                                        "start the threads for volumes");
+                                goto out;
+                        }
+
+                        priv->qd_vols_conf[i]->threads_status = _gf_true;
+                }
+                break;
+        }
+        case GF_EVENT_CHILD_DOWN:
+        {
+                gf_log (this->name, GF_LOG_ERROR, "vol %s is down.",
+                        priv->qd_vols_conf [i]->name);
+                break;
+        }
+        default:
+                default_notify (this, event, data);
+        }/* end switch */
+
+
+
+out:
+        return ret;
+}
+
+class_methods_t class_methods = {
+        .init           = qd_init,
+        .fini           = qd_fini,
+        .reconfigure    = qd_reconfigure,
+        .notify         = qd_notify,
+};
+
+struct xlator_fops fops = {
+};
+
+struct xlator_cbks cbks = {
+};
+
+struct volume_options options[] = {
+        {.key = {"*.limit-set"}},
+        {.key = {"*.soft-timeout"},
+         .type = GF_OPTION_TYPE_TIME,
+         .min = 1,
+         .max = LONG_MAX,
+         .default_value = "10",
+         .description = ""
+        },
+        {.key = {"*.hard-timeout"},
+         .type = GF_OPTION_TYPE_TIME,
+         .min = 0,
+         .max = LONG_MAX,
+         .default_value = "2",
+         .description = ""
+        },
+        {.key = {"*.alert-time"},
+         .type = GF_OPTION_TYPE_TIME,
+         .min = 0,
+         .max = LONG_MAX,
+         /* default weekly (7 * 24 * 60 *60) */
+         .default_value = "604800",
+         .description = ""
+        },
+        {.key = {"*.default-soft-limit"},
+         .type = GF_OPTION_TYPE_PERCENT,
+         .min = 0,
+         .max = 100,
+         .default_value = "90%",
+         .description = "Takes this if individual paths are not configured "
+                        "with soft limits."
+        },
+        {.key = {NULL}}
+};
diff --git a/xlators/lib/src/libxlator.h b/xlators/lib/src/libxlator.h
index 1d5e1657..08bd77b9 100644
--- a/xlators/lib/src/libxlator.h
+++ b/xlators/lib/src/libxlator.h
@@ -32,6 +32,7 @@
 #define MARKER_UUID_TYPE    1
 #define MARKER_XTIME_TYPE   2
 #define GF_XATTR_QUOTA_SIZE_KEY "trusted.glusterfs.quota.size"
+#define GF_XATTR_QUOTA_LIMIT_LIST "trusted.limit.list"
 
 
 typedef int32_t (*xlator_specf_unwind_t) (call_frame_t *frame,
diff --git a/xlators/mgmt/glusterd/src/glusterd-handler.c b/xlators/mgmt/glusterd/src/glusterd-handler.c
index 480ee01b..fbc48001 100644
--- a/xlators/mgmt/glusterd/src/glusterd-handler.c
+++ b/xlators/mgmt/glusterd/src/glusterd-handler.c
@@ -54,6 +54,8 @@
 #include <lvm2app.h>
 #endif
 
+extern glusterd_op_info_t opinfo;
+
 int glusterd_big_locked_notify (struct rpc_clnt *rpc, void *mydata,
                                 rpc_clnt_event_t event,
                                 void *data, rpc_clnt_notify_t notify_fn)
@@ -307,10 +309,23 @@ _build_option_key (dict_t *d, char *k, data_t *v, void *tmp)
         char                    reconfig_key[256] = {0, };
         struct args_pack        *pack             = NULL;
         int                     ret               = -1;
+        xlator_t                *this             = NULL;
+        glusterd_conf_t         *priv             = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+        priv = this->private;
+        GF_ASSERT (priv);
 
         pack = tmp;
         if (strcmp (k, GLUSTERD_GLOBAL_OPT_VERSION) == 0)
                 return 0;
+
+        if (priv->op_version > GD_OP_VERSION_MIN) {
+                if ((strcmp (k, "features.limit-usage") == 0) ||
+                    (strcmp (k, "features.soft-limit") == 0))
+                        return 0;
+        }
         snprintf (reconfig_key, 256, "volume%d.option.%s",
                   pack->vol_count, k);
         ret = dict_set_str (pack->dict, reconfig_key, v->data);
@@ -3777,6 +3792,7 @@ __glusterd_peer_rpc_notify (struct rpc_clnt *rpc, void *mydata,
         glusterd_peerinfo_t  *peerinfo    = NULL;
         glusterd_peerctx_t   *peerctx     = NULL;
         gf_boolean_t         quorum_action = _gf_false;
+        uuid_t               uuid;
 
         peerctx = mydata;
         if (!peerctx)
@@ -3818,6 +3834,13 @@ __glusterd_peer_rpc_notify (struct rpc_clnt *rpc, void *mydata,
                         glusterd_friend_remove_notify (peerctx);
                         goto out;
                 }
+                glusterd_get_lock_owner (&uuid);
+                if (!uuid_is_null (uuid) &&
+                    !uuid_compare (peerinfo->uuid, uuid)) {
+                        glusterd_unlock (peerinfo->uuid);
+                        if (opinfo.state.state != GD_OP_STATE_DEFAULT)
+                                opinfo.state.state = GD_OP_STATE_DEFAULT;
+                }
 
                 peerinfo->connected = 0;
                 break;
diff --git a/xlators/mgmt/glusterd/src/glusterd-quota.c b/xlators/mgmt/glusterd/src/glusterd-quota.c
index 31826719..ec4a6eb8 100644
--- a/xlators/mgmt/glusterd/src/glusterd-quota.c
+++ b/xlators/mgmt/glusterd/src/glusterd-quota.c
@@ -24,6 +24,22 @@
 
 #include <sys/wait.h>
 
+
+const char *gd_quota_op_list[GF_QUOTA_OPTION_TYPE_DEFAULT_SOFT_LIMIT+1] = {
+        [GF_QUOTA_OPTION_TYPE_NONE]               = "none",
+        [GF_QUOTA_OPTION_TYPE_ENABLE]             = "enable",
+        [GF_QUOTA_OPTION_TYPE_DISABLE]            = "disable",
+        [GF_QUOTA_OPTION_TYPE_LIMIT_USAGE]        = "limit-usage",
+        [GF_QUOTA_OPTION_TYPE_REMOVE]             = "remove",
+        [GF_QUOTA_OPTION_TYPE_LIST]               = "list",
+        [GF_QUOTA_OPTION_TYPE_VERSION]            = "version",
+        [GF_QUOTA_OPTION_TYPE_SOFT_LIMIT]         = "soft-limit",
+        [GF_QUOTA_OPTION_TYPE_ALERT_TIME]         = "alert-time",
+        [GF_QUOTA_OPTION_TYPE_SOFT_TIMEOUT]       = "soft-timeout",
+        [GF_QUOTA_OPTION_TYPE_HARD_TIMEOUT]       = "hard-timeout",
+        [GF_QUOTA_OPTION_TYPE_DEFAULT_SOFT_LIMIT] = "default-soft-limit",
+};
+
 int
 __glusterd_handle_quota (rpcsvc_request_t *req)
 {
@@ -31,15 +47,17 @@ __glusterd_handle_quota (rpcsvc_request_t *req)
         gf_cli_req                      cli_req = {{0,}};
         dict_t                         *dict = NULL;
         glusterd_op_t                   cli_op = GD_OP_QUOTA;
-        char                            operation[256] = {0, };
         char                           *volname = NULL;
         int32_t                         type = 0;
         char                            msg[2048] = {0,};
         xlator_t                       *this = NULL;
+        glusterd_conf_t                *conf = NULL;
 
         GF_ASSERT (req);
         this = THIS;
         GF_ASSERT (this);
+        conf = this->private;
+        GF_ASSERT (conf);
 
         ret = xdr_to_generic (req->msg[0], &cli_req, (xdrproc_t)xdr_gf_cli_req);
         if (ret < 0) {
@@ -82,23 +100,16 @@ __glusterd_handle_quota (rpcsvc_request_t *req)
                goto out;
         }
 
-        switch (type) {
-        case GF_QUOTA_OPTION_TYPE_ENABLE:
-                strncpy (operation, "enable", sizeof (operation));
-                break;
-
-        case GF_QUOTA_OPTION_TYPE_DISABLE:
-                strncpy (operation, "disable", sizeof (operation));
-                break;
-
-        case GF_QUOTA_OPTION_TYPE_LIMIT_USAGE:
-                strncpy (operation, "limit-usage", sizeof (operation));
-                break;
+        if ((conf->op_version == GD_OP_VERSION_MIN) &&
+            (type > GF_QUOTA_OPTION_TYPE_VERSION)) {
+                snprintf (msg, sizeof (msg), "Cannot execute command. The "
+                         "cluster is operating at version 1. Executing command "
+                         "%s is disallowed in this state",
+                         gd_quota_op_list[type]);
+                ret = -1;
+                goto out;
+       }
 
-        case GF_QUOTA_OPTION_TYPE_REMOVE:
-                strncpy (operation, "remove", sizeof (operation));
-                break;
-        }
         ret = glusterd_op_begin_synctask (req, GD_OP_QUOTA, dict);
 
 out:
@@ -132,7 +143,6 @@ glusterd_check_if_quota_trans_enabled (glusterd_volinfo_t *volinfo)
         }
 
         if (flag == _gf_false) {
-                gf_log ("", GF_LOG_ERROR, "first enable the quota translator");
                 ret = -1;
                 goto out;
         }
@@ -141,13 +151,21 @@ out:
         return ret;
 }
 
-/* At the end of the function, the variable found will be set
+/* At the end of the function, the variable @found will be set
  * to true if the path to be removed was present in the limit-list,
  * else will be false.
+ *
+ * In addition, the function does the following things:
+ *
+ * a. places the path to be removed, if found, in @removed_path,
+ * b. places the new limit list formed after removing @path's entry, in
+ *    @new_list. If @path is not found, the input limit string @quota_limits is
+ *    dup'd as is and placed in @new_list.
  */
 int32_t
-_glusterd_quota_remove_limits (char **quota_limits, char *path,
-                               gf_boolean_t *found)
+_glusterd_quota_remove_limits (char *quota_limits, char *path,
+                               gf_boolean_t *found, char **new_list,
+                               char **removed_path)
 {
         int      ret      = 0;
         int      i        = 0;
@@ -158,14 +176,15 @@ _glusterd_quota_remove_limits (char **quota_limits, char *path,
         int      flag     = 0;
         char    *limits   = NULL;
         char    *qlimits  = NULL;
+        char    *rp       = NULL;
 
         if (found != NULL)
                 *found = _gf_false;
 
-        if (*quota_limits == NULL)
+        if (quota_limits == NULL)
                 return -1;
 
-        qlimits = *quota_limits;
+        qlimits = quota_limits;
 
         pathlen = strlen (path);
 
@@ -192,6 +211,15 @@ _glusterd_quota_remove_limits (char **quota_limits, char *path,
                 } else {
                         skiplen = size + 1;
                         size = len - i - size;
+                        if (removed_path) {
+                                rp = GF_CALLOC (skiplen, sizeof (char), gf_gld_mt_char);
+                                if (!rp) {
+                                        ret = -1;
+                                        goto out;
+                                }
+                                strncpy (rp, &qlimits[i], skiplen - 1);
+                                *removed_path = rp;
+                        }
                         memcpy ((void *) &limits [i], (void *) &qlimits [i + skiplen], size);
                         break;
                 }
@@ -200,42 +228,27 @@ _glusterd_quota_remove_limits (char **quota_limits, char *path,
                 size = 0;
         }
 
-        if (!flag) {
-                ret = 1;
-        } else {
-                len = strlen (limits);
-
-                if (len == 0) {
-                        GF_FREE (qlimits);
-
-                        *quota_limits = NULL;
-
-                        goto out;
-                }
-
-                if (limits[len - 1] == ',') {
-                        limits[len - 1] = '\0';
-                        len --;
-                }
-
-                GF_FREE (qlimits);
-
-                qlimits = GF_CALLOC (len + 1, sizeof (char), gf_gld_mt_char);
-
-                if (!qlimits) {
-                        ret = -1;
-                        goto out;
-                }
-
-                memcpy ((void *) qlimits, (void *) limits, len + 1);
+        len = strlen (limits);
+        if (len == 0)
+                goto out;
 
-                *quota_limits = qlimits;
+        if (limits[len - 1] == ',') {
+                limits[len - 1] = '\0';
+                len --;
+        }
 
-                ret = 0;
+        *new_list = GF_CALLOC (len + 1, sizeof (char), gf_gld_mt_char);
+        if (!*new_list) {
+                ret = -1;
+                goto out;
         }
 
+        memcpy ((void *) *new_list, (void *) limits, len + 1);
+        ret = 0;
 out:
         GF_FREE (limits);
+        if (ret != -1)
+                ret = flag ? 0 : 1;
 
         return ret;
 }
@@ -381,22 +394,30 @@ _glusterd_quota_get_limit_usages (glusterd_volinfo_t *volinfo,
 
 int32_t
 glusterd_quota_get_limit_usages (glusterd_conf_t *priv,
-                                 glusterd_volinfo_t *volinfo,
-                                 char *volname,
-                                 dict_t *dict,
-                                 char **op_errstr,
+                                 glusterd_volinfo_t *volinfo, char *volname,
+                                 dict_t *dict, char **op_errstr,
                                  dict_t *rsp_dict)
 {
-        int32_t i               = 0;
-        int32_t ret             = 0;
-        int32_t count           = 0;
-        char    *path           = NULL;
-        char    cmd_str [1024]  = {0, };
-        char   *ret_str         = NULL;
+        int32_t            i               = 0;
+        int32_t            ret             = 0;
+        int32_t            count           = 0;
+        int                entry_count     = 0;
+        char              *path            = NULL;
+        char               cmd_str [1024]  = {0, };
+        char              *ret_str         = NULL;
+        xlator_t          *this            = NULL;
+        glusterd_conf_t   *conf            = NULL;
+        char              *default_limit   = NULL;
+        char              *val             = NULL;
 
         if (rsp_dict == NULL)
                 return 0;
 
+        this = THIS;
+        GF_ASSERT (this);
+        conf = this->private;
+        GF_ASSERT (conf);
+
         ret = dict_get_int32 (dict, "count", &count);
         if (ret < 0)
                 goto out;
@@ -404,22 +425,60 @@ glusterd_quota_get_limit_usages (glusterd_conf_t *priv,
         if (count == 0) {
                 ret_str = _glusterd_quota_get_limit_usages (volinfo, NULL,
                                                             op_errstr);
+                ret = dict_set_dynstr (rsp_dict, "limit_list", ret_str);
+                if (ret)
+                        goto out;
         } else {
                 i = 0;
                 while (count--) {
-                        snprintf (cmd_str, 1024, "path%d", i++);
+                        snprintf (cmd_str, sizeof (cmd_str), "path%d", i++);
 
                         ret = dict_get_str (dict, cmd_str, &path);
                         if (ret < 0)
                                 goto out;
+                        ret = gf_canonicalize_path (path);
+                        if (ret) {
+                                goto out;
+                        }
 
-                        ret_str = _glusterd_quota_get_limit_usages (volinfo, path, op_errstr);
+                        ret_str = _glusterd_quota_get_limit_usages (volinfo,
+                                                                    path,
+                                                                    op_errstr);
+                        /* Despite quota limits being absent on @path, we go
+                         * ahead and place it in @rsp_dict with
+                         * value = "Not set". This is because after commit op,
+                         * as part of aggregation of @rsp_dict with @op_ctx,
+                         * when we copy the rsp_dict into op_ctx, op_ctx would
+                         * still be containing the old key (same as @cmd_str)
+                         * with the old value. In order to overwrite the old
+                         * value, we replace it with "Not set", something that
+                         * the cli can easily interpret as a case of quota
+                         * limits not having been set on the given path.
+                         */
+                        if (!ret_str) {
+                                ret = dict_set_str (rsp_dict, cmd_str,
+                                                    "Not set");
+                        } else {
+                                ret = dict_set_dynstr (rsp_dict, cmd_str,
+                                                       ret_str);
+                                entry_count = entry_count + 1;
+                        }
                 }
         }
+        ret = dict_set_int32 (rsp_dict, "entry-count", entry_count);
+        if (ret)
+                goto out;
 
-        if (ret_str) {
-                ret = dict_set_dynstr (rsp_dict, "limit_list", ret_str);
-        }
+        ret = dict_set_uint32 (rsp_dict, "op-version", conf->op_version);
+
+        ret = glusterd_volinfo_get (volinfo, "features.default-soft-limit",
+                                    &default_limit);
+        if (default_limit)
+                val = gf_strdup (default_limit);
+        else
+                val = gf_strdup ("90%");
+
+        ret = dict_set_dynstr (rsp_dict, "default-soft-limit", val);
 out:
         return ret;
 }
@@ -430,10 +489,14 @@ glusterd_quota_enable (glusterd_volinfo_t *volinfo, char **op_errstr,
 {
         int32_t         ret     = -1;
         char            *quota_status = NULL;
+        xlator_t        *this         = NULL;
 
-        GF_VALIDATE_OR_GOTO ("glusterd", volinfo, out);
-        GF_VALIDATE_OR_GOTO ("glusterd", crawl, out);
-        GF_VALIDATE_OR_GOTO ("glusterd", op_errstr, out);
+        this = THIS;
+        GF_ASSERT (this);
+
+        GF_VALIDATE_OR_GOTO (this->name, volinfo, out);
+        GF_VALIDATE_OR_GOTO (this->name, crawl, out);
+        GF_VALIDATE_OR_GOTO (this->name, op_errstr, out);
 
         if (glusterd_is_volume_started (volinfo) == 0) {
                 *op_errstr = gf_strdup ("Volume is stopped, start volume "
@@ -449,15 +512,15 @@ glusterd_quota_enable (glusterd_volinfo_t *volinfo, char **op_errstr,
 
         quota_status = gf_strdup ("on");
         if (!quota_status) {
-                gf_log ("", GF_LOG_ERROR, "memory allocation failed");
-                *op_errstr = gf_strdup ("Enabling quota has been unsuccessful");
+                gf_log (this->name, GF_LOG_ERROR, "memory allocation failed");
+                ret = -1;
                 goto out;
         }
 
-        ret = dict_set_dynstr (volinfo->dict, VKEY_FEATURES_QUOTA, quota_status);
+        ret = dict_set_dynstr (volinfo->dict, VKEY_FEATURES_QUOTA,
+                               quota_status);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "dict set failed");
-                *op_errstr = gf_strdup ("Enabling quota has been unsuccessful");
+                gf_log (this->name, GF_LOG_ERROR, "dict set failed");
                 goto out;
         }
 
@@ -467,17 +530,34 @@ glusterd_quota_enable (glusterd_volinfo_t *volinfo, char **op_errstr,
 
         ret = 0;
 out:
+        if (ret && op_errstr && !*op_errstr)
+                gf_asprintf (op_errstr, "Enabling quota on volume %s has been "
+                             "unsuccessful", volinfo->volname);
         return ret;
 }
 
 int32_t
 glusterd_quota_disable (glusterd_volinfo_t *volinfo, char **op_errstr)
 {
-        int32_t  ret            = -1;
-        char    *quota_status   = NULL, *quota_limits = NULL;
+        int32_t    ret            = -1;
+        int        i              =  0;
+        char      *quota_status   = NULL;
+        char      *value          = NULL;
+        xlator_t  *this           = NULL;
+        glusterd_conf_t *conf     = NULL;
+        char *quota_options[]     = {VKEY_FEATURES_LIMIT_USAGE,
+                                     "features.soft-timeout",
+                                     "features.hard-timeout",
+                                     "features.alert-time",
+                                     "features.default-soft-limit", NULL};
 
-        GF_VALIDATE_OR_GOTO ("glusterd", volinfo, out);
-        GF_VALIDATE_OR_GOTO ("glusterd", op_errstr, out);
+        this = THIS;
+        GF_ASSERT (this);
+        conf = this->private;
+        GF_ASSERT (conf);
+
+        GF_VALIDATE_OR_GOTO (this->name, volinfo, out);
+        GF_VALIDATE_OR_GOTO (this->name, op_errstr, out);
 
         ret = glusterd_check_if_quota_trans_enabled (volinfo);
         if (ret == -1) {
@@ -487,47 +567,84 @@ glusterd_quota_disable (glusterd_volinfo_t *volinfo, char **op_errstr)
 
         quota_status = gf_strdup ("off");
         if (!quota_status) {
-                gf_log ("", GF_LOG_ERROR, "memory allocation failed");
-                *op_errstr = gf_strdup ("Disabling quota has been unsuccessful");
+                gf_log (this->name, GF_LOG_ERROR, "memory allocation failed");
+                ret = -1;
                 goto out;
         }
 
         ret = dict_set_dynstr (volinfo->dict, VKEY_FEATURES_QUOTA, quota_status);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "dict set failed");
-                *op_errstr = gf_strdup ("Disabling quota has been unsuccessful");
+                gf_log (this->name, GF_LOG_ERROR, "dict set failed");
                 goto out;
         }
 
+        for (i = 0; quota_options [i]; i++) {
+                ret = glusterd_volinfo_get (volinfo, quota_options[i], &value);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_INFO, "failed to get option"
+                                                         " %s",
+                                                         quota_options[i]);
+                } else {
+                dict_del (volinfo->dict, quota_options[i]);
+                }
+                if ((i == 0) && (conf->op_version == GD_OP_VERSION_MIN))
+                        break;
+         }
+
         *op_errstr = gf_strdup ("Disabling quota has been successful");
+        ret = 0;
+out:
+        if (ret && op_errstr && !*op_errstr)
+                gf_asprintf (op_errstr, "Disabling quota on volume %s has been "
+                             "unsuccessful", volinfo->volname);
+        return ret;
+}
 
-        ret = glusterd_volinfo_get (volinfo, VKEY_FEATURES_LIMIT_USAGE,
-                                    &quota_limits);
-        if (ret) {
-                gf_log ("", GF_LOG_WARNING, "failed to get the quota limits");
+static void
+gd_quota_get_formatted_limit_list (char **value, char *existing_list,
+                                   char *path, char *hard_limit,
+                                   char *soft_limit)
+{
+        if (!existing_list) {
+                if (!soft_limit)
+                        gf_asprintf (value, "%s:%s", path, hard_limit);
+                else
+                        gf_asprintf (value, "%s:%s:%s", path, hard_limit,
+                                     soft_limit);
         } else {
-                GF_FREE (quota_limits);
+                if (!soft_limit)
+                        gf_asprintf (value, "%s,%s:%s", existing_list, path,
+                                     hard_limit);
+                else
+                        gf_asprintf (value, "%s,%s:%s:%s", existing_list, path,
+                                     hard_limit, soft_limit);
         }
-
-        dict_del (volinfo->dict, VKEY_FEATURES_LIMIT_USAGE);
-
-out:
-        return ret;
 }
 
 int32_t
-glusterd_quota_limit_usage (glusterd_volinfo_t *volinfo, dict_t *dict, char **op_errstr)
+glusterd_quota_limit_usage (glusterd_volinfo_t *volinfo, dict_t *dict,
+                            char **op_errstr)
 {
-        int32_t          ret    = -1;
-        char            *path   = NULL;
-        char            *limit  = NULL;
-        char            *value  = NULL;
-        char             msg [1024] = {0,};
-        char            *quota_limits = NULL;
+        int32_t          ret                = -1;
+        char            *path               = NULL;
+        char            *limit              = NULL;
+        char            *value              = NULL;
+        char            *sl                 = NULL;
+        char             msg[5120]          = {0,};
+        char            *quota_limits       = NULL;
+        char            *new_list           = NULL;
+        xlator_t        *this               = NULL;
+        glusterd_conf_t *priv               = NULL;
+        char            *removed_path       = NULL;
 
-        GF_VALIDATE_OR_GOTO ("glusterd", dict, out);
-        GF_VALIDATE_OR_GOTO ("glusterd", volinfo, out);
-        GF_VALIDATE_OR_GOTO ("glusterd", op_errstr, out);
+        this = THIS;
+        GF_ASSERT (this);
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        GF_VALIDATE_OR_GOTO (this->name, dict, out);
+        GF_VALIDATE_OR_GOTO (this->name, volinfo, out);
+        GF_VALIDATE_OR_GOTO (this->name, op_errstr, out);
 
         ret = glusterd_check_if_quota_trans_enabled (volinfo);
         if (ret == -1) {
@@ -539,109 +656,222 @@ glusterd_quota_limit_usage (glusterd_volinfo_t *volinfo, dict_t *dict, char **op
         ret = glusterd_volinfo_get (volinfo, VKEY_FEATURES_LIMIT_USAGE,
                                     &quota_limits);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "failed to get the quota limits");
-                *op_errstr = gf_strdup ("failed to set limit");
+                gf_log (this->name, GF_LOG_ERROR, "failed to get quota limits");
                 goto out;
         }
 
         ret = dict_get_str (dict, "path", &path);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to fetch quota limits" );
-                *op_errstr = gf_strdup ("failed to set limit");
+                gf_log (this->name, GF_LOG_ERROR, "Unable to fetch path");
                 goto out;
         }
+        ret = gf_canonicalize_path (path);
+        if (ret)
+                goto out;
 
         ret = dict_get_str (dict, "limit", &limit);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to fetch quota limits" );
-                *op_errstr = gf_strdup ("failed to set limit");
+                gf_log (this->name, GF_LOG_ERROR, "Unable to fetch limit");
                 goto out;
         }
 
-        if (quota_limits) {
-                ret = _glusterd_quota_remove_limits (&quota_limits, path, NULL);
-                if (ret == -1) {
-                        gf_log ("", GF_LOG_ERROR, "Unable to allocate memory");
-                        *op_errstr = gf_strdup ("failed to set limit");
-                        goto out;
-                }
-        }
-
         if (quota_limits == NULL) {
-                ret = gf_asprintf (&value, "%s:%s", path, limit);
-                if (ret == -1) {
-                        gf_log ("", GF_LOG_ERROR, "Unable to allocate memory");
-                        *op_errstr = gf_strdup ("failed to set limit");
-                        goto out;
-                }
+                ; //do nothing and go past the else block
         } else {
-                ret = gf_asprintf (&value, "%s,%s:%s",
-                                   quota_limits, path, limit);
+                ret = _glusterd_quota_remove_limits (quota_limits, path, NULL,
+                                                     &new_list, &removed_path);
                 if (ret == -1) {
-                        gf_log ("", GF_LOG_ERROR, "Unable to allocate memory");
-                        *op_errstr = gf_strdup ("failed to set limit");
+                        gf_log (this->name, GF_LOG_ERROR, "Unable to remove "
+                                                          "limit");
                         goto out;
                 }
 
-                GF_FREE (quota_limits);
+                if (removed_path && (priv->op_version > GD_OP_VERSION_MIN)) {
+                        ret = gf_get_soft_limit (removed_path, &sl);
+                        if (ret == -1)
+                                goto out;
+                }
         }
-
-        quota_limits = value;
+        gd_quota_get_formatted_limit_list (&value, new_list, path, limit, sl);
 
         ret = dict_set_str (volinfo->dict, VKEY_FEATURES_LIMIT_USAGE,
-                            quota_limits);
+                            value);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to set quota limits" );
-                *op_errstr = gf_strdup ("failed to set limit");
+                gf_log (this->name, GF_LOG_ERROR, "Unable to set quota limits");
                 goto out;
         }
-        snprintf (msg, 1024, "limit set on %s", path);
-        *op_errstr = gf_strdup (msg);
 
+        snprintf (msg, sizeof (msg), "hard limit set on %s", path);
+        *op_errstr = gf_strdup (msg);
         ret = 0;
 out:
+        GF_FREE (sl);
+        GF_FREE (removed_path);
+        GF_FREE (new_list);
+
+        if (ret && op_errstr && !*op_errstr)
+                gf_asprintf (op_errstr, "Failed to set hard limit on path %s "
+                             "for volume %s", path, volinfo->volname);
+        return ret;
+}
+
+int
+glusterd_quota_soft_limit (glusterd_volinfo_t *volinfo, dict_t *dict,
+                           char **op_errstr)
+{
+        int     ret                = 0;
+        char   *path               = NULL;
+        char   *limit              = NULL;
+        char   *quota_limits       = NULL;
+        char   *new_list           = NULL;
+        char   *removed_path       = NULL;
+        char    msg[1024]          = {0,};
+        char   *hl                 = NULL;
+        char   *value              = NULL;
+        xlator_t *this             = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+
+        if (!volinfo || !dict || !op_errstr) {
+                ret = -1;
+                goto out;
+        }
+
+        ret = glusterd_check_if_quota_trans_enabled (volinfo);
+        if (ret == -1) {
+                *op_errstr = gf_strdup ("Quota is disabled, please enable "
+                                        "quota");
+                goto out;
+        }
+
+        ret = glusterd_volinfo_get (volinfo, VKEY_FEATURES_LIMIT_USAGE,
+                                    &quota_limits);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "failed to get quota limits");
+                goto out;
+        }
+
+        ret = dict_get_str (dict, "path", &path);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to fetch path");
+                goto out;
+        }
+
+        ret = gf_canonicalize_path (path);
+        if (ret)
+                goto out;
+
+        ret = dict_get_str (dict, "limit", &limit);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to fetch limit");
+                goto out;
+        }
+
+        if (quota_limits == NULL) {
+                gf_asprintf (op_errstr, "Soft-limit cannot be set on path %s, "
+                             "without setting hard-limit on it first.", path);
+                ret = -1;
+                goto out;
+        } else {
+                ret = _glusterd_quota_remove_limits (quota_limits, path, NULL,
+                                                     &new_list, &removed_path);
+                 if (ret == -1) {
+                        gf_log (this->name, GF_LOG_ERROR, "Failed to remove "
+                                                          "limit");
+                         goto out;
+                 }
+
+                if (!removed_path) {
+                        gf_asprintf (op_errstr, "Soft-limit cannot be set on "
+                                     "path %s without setting hard-limit on it "
+                                     "first.", path);
+                        ret = -1;
+                        goto out;
+                } else {
+                        ret = gf_get_hard_limit (removed_path, &hl);
+                        if (ret)
+                                goto out;
+                }
+         }
+
+         gd_quota_get_formatted_limit_list (&value, new_list, path, hl, limit);
+
+         ret = dict_set_str (volinfo->dict, VKEY_FEATURES_LIMIT_USAGE,
+                             value);
+         if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to set quota limits");
+                 goto out;
+         }
+
+         snprintf (msg, sizeof (msg), "soft limit set on %s", path);
+         *op_errstr = gf_strdup (msg);
+         ret = 0;
+
+out:
+        GF_FREE (hl);
+        GF_FREE (removed_path);
+        GF_FREE (new_list);
+
+        if (ret && op_errstr && !*op_errstr)
+                gf_asprintf (op_errstr, "Failed to set soft limit for path %s "
+                                        "on volume %s", path, volinfo->volname);
         return ret;
 }
 
 int32_t
-glusterd_quota_remove_limits (glusterd_volinfo_t *volinfo, dict_t *dict, char **op_errstr)
+glusterd_quota_remove_limits (glusterd_volinfo_t *volinfo, dict_t *dict,
+                              char **op_errstr)
 {
         int32_t         ret                   = -1;
         char            str [PATH_MAX + 1024] = {0,};
         char            *quota_limits         = NULL;
+        char            *new_list             = NULL;
+        char            *value                = NULL;
         char            *path                 = NULL;
         gf_boolean_t     flag                 = _gf_false;
+        xlator_t        *this                 = NULL;
 
-        GF_VALIDATE_OR_GOTO ("glusterd", dict, out);
-        GF_VALIDATE_OR_GOTO ("glusterd", volinfo, out);
-        GF_VALIDATE_OR_GOTO ("glusterd", op_errstr, out);
+        this = THIS;
+        GF_ASSERT (this);
+
+        GF_VALIDATE_OR_GOTO (this->name, dict, out);
+        GF_VALIDATE_OR_GOTO (this->name, volinfo, out);
+        GF_VALIDATE_OR_GOTO (this->name, op_errstr, out);
 
         ret = glusterd_check_if_quota_trans_enabled (volinfo);
         if (ret == -1) {
-                *op_errstr = gf_strdup ("Quota is disabled, please enable quota");
+                *op_errstr = gf_strdup ("Quota is disabled, please enable "
+                                        "quota");
                 goto out;
         }
 
         ret = glusterd_volinfo_get (volinfo, VKEY_FEATURES_LIMIT_USAGE,
                                     &quota_limits);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "failed to get the quota limits");
+                gf_log (this->name, GF_LOG_ERROR, "failed to get quota limits");
                 goto out;
         }
 
         ret = dict_get_str (dict, "path", &path);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to fetch quota limits" );
+                gf_log (this->name, GF_LOG_ERROR, "Unable to fetch path");
                 goto out;
         }
 
-        ret = _glusterd_quota_remove_limits (&quota_limits, path, &flag);
+        ret = gf_canonicalize_path (path);
+        if (ret)
+                goto out;
+
+        ret = _glusterd_quota_remove_limits (quota_limits, path, &flag,
+                                             &new_list, NULL);
         if (ret == -1) {
                 if (flag == _gf_true)
                         snprintf (str, sizeof (str), "Removing limit on %s has "
                                   "been unsuccessful", path);
                 else
-                        snprintf (str, sizeof (str), "%s has no limit set", path);
+                        snprintf (str, sizeof (str), "%s has no limit set",
+                                  path);
                 *op_errstr = gf_strdup (str);
                 goto out;
         } else {
@@ -654,11 +884,13 @@ glusterd_quota_remove_limits (glusterd_volinfo_t *volinfo, dict_t *dict, char **
                 *op_errstr = gf_strdup (str);
         }
 
-        if (quota_limits) {
+        if (new_list) {
+                value = gf_strdup (new_list);
                 ret = dict_set_str (volinfo->dict, VKEY_FEATURES_LIMIT_USAGE,
-                                    quota_limits);
+                                    value);
                 if (ret) {
-                        gf_log ("", GF_LOG_ERROR, "Unable to set quota limits" );
+                        gf_log (this->name, GF_LOG_ERROR, "Unable to set quota "
+                                                          "limits");
                         goto out;
                 }
         } else {
@@ -668,9 +900,46 @@ glusterd_quota_remove_limits (glusterd_volinfo_t *volinfo, dict_t *dict, char **
         ret = 0;
 
 out:
+        GF_FREE (new_list);
         return ret;
 }
 
+int
+glusterd_set_quota_option (glusterd_volinfo_t *volinfo, dict_t *dict,
+                           char *key, char **op_errstr)
+{
+        int        ret    = 0;
+        char      *value  = NULL;
+        xlator_t  *this   = NULL;
+        char      *option = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+
+        ret = glusterd_check_if_quota_trans_enabled (volinfo);
+        if (ret == -1) {
+                gf_asprintf (op_errstr, "Cannot set %s. Quota on volume %s is "
+                                        "disabled", key, volinfo->volname);
+                return -1;
+        }
+
+        ret = dict_get_str (dict, "value", &value);
+        if(ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Option value absent.");
+                return -1;
+        }
+
+        option = gf_strdup (value);
+        ret = dict_set_dynstr (volinfo->dict, key, option);
+        if(ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to set option %s",
+                        key);
+                return -1;
+        }
+        gf_asprintf (op_errstr, "%s on volume %s set", key, volinfo->volname);
+
+        return 0;
+}
 
 int
 glusterd_op_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
@@ -681,26 +950,41 @@ glusterd_op_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
         int                     type         = -1;
         gf_boolean_t            start_crawl  = _gf_false;
         glusterd_conf_t        *priv         = NULL;
+        xlator_t               *this         = NULL;
 
         GF_ASSERT (dict);
         GF_ASSERT (op_errstr);
 
-        priv = THIS->private;
+        this = THIS;
+        GF_ASSERT (this);
+        priv = this->private;
+        GF_ASSERT (priv);
 
         ret = dict_get_str (dict, "volname", &volname);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to get volume name " );
+                gf_log (this->name, GF_LOG_ERROR, "Unable to get volume name");
                 goto out;
         }
 
         ret = glusterd_volinfo_find (volname, &volinfo);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to allocate memory");
+                gf_asprintf (op_errstr, FMTSTR_CHECK_VOL_EXISTS, volname);
                 goto out;
         }
 
         ret = dict_get_int32 (dict, "type", &type);
 
+        if ((priv->op_version == GD_OP_VERSION_MIN) &&
+            (type > GF_QUOTA_OPTION_TYPE_VERSION)) {
+                gf_asprintf (op_errstr, "Volume quota failed. The cluster is "
+                                        "operating at version %d. Option %s "
+                                        "is disallowed in this state.",
+                                        priv->op_version,
+                                        gd_quota_op_list[type]);
+                ret = -1;
+                goto out;
+        }
+
         if (type == GF_QUOTA_OPTION_TYPE_ENABLE) {
                 ret = glusterd_quota_enable (volinfo, op_errstr, &start_crawl);
                 if (ret < 0)
@@ -708,7 +992,6 @@ glusterd_op_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
 
                 goto create_vol;
         }
-
         if (type == GF_QUOTA_OPTION_TYPE_DISABLE) {
                 ret = glusterd_quota_disable (volinfo, op_errstr);
                 if (ret < 0)
@@ -742,15 +1025,59 @@ glusterd_op_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
                 }
 
                 ret = glusterd_quota_get_limit_usages (priv, volinfo, volname,
-                                                       dict, op_errstr, rsp_dict);
+                                                       dict, op_errstr,
+                                                       rsp_dict);
 
                 goto out;
         }
+
+        if (type == GF_QUOTA_OPTION_TYPE_SOFT_LIMIT) {
+                ret = glusterd_quota_soft_limit (volinfo, dict, op_errstr);
+	                if (ret < 0)
+                                goto out;
+                goto create_vol;
+        }
+
+        if (type == GF_QUOTA_OPTION_TYPE_SOFT_TIMEOUT) {
+                ret = glusterd_set_quota_option (volinfo, dict,
+                                                 "features.soft-timeout",
+                                                 op_errstr);
+                if (ret)
+                        goto out;
+                goto create_vol;
+        }
+
+        if (type == GF_QUOTA_OPTION_TYPE_HARD_TIMEOUT) {
+                ret = glusterd_set_quota_option (volinfo, dict,
+                                                 "features.hard-timeout",
+                                                 op_errstr);
+                if (ret)
+                        goto out;
+                goto create_vol;
+        }
+
+        if (type == GF_QUOTA_OPTION_TYPE_ALERT_TIME) {
+                ret = glusterd_set_quota_option (volinfo, dict,
+                                                 "features.alert-time",
+                                                 op_errstr);
+                if (ret)
+                        goto out;
+                goto create_vol;
+        }
+        if (type == GF_QUOTA_OPTION_TYPE_DEFAULT_SOFT_LIMIT) {
+                ret = glusterd_set_quota_option (volinfo, dict,
+                                                 "features.default-soft-limit",
+                                                 op_errstr);
+                if (ret)
+                        goto out;
+                goto create_vol;
+        }
+
 create_vol:
         ret = glusterd_create_volfiles_and_notify_services (volinfo);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to re-create volfile for"
-                                          " 'quota'");
+                gf_log (this->name, GF_LOG_ERROR, "Unable to re-create "
+                                                  "volfiles");
                 ret = -1;
                 goto out;
         }
@@ -759,20 +1086,31 @@ create_vol:
         if (ret)
                 goto out;
 
-        if (GLUSTERD_STATUS_STARTED == volinfo->status)
-                ret = glusterd_check_generate_start_nfs ();
-
+        if (GLUSTERD_STATUS_STARTED == volinfo->status) {
+                if (priv->op_version == GD_OP_VERSION_MIN)
+                        ret = glusterd_check_generate_start_nfs ();
+        }
         ret = 0;
 
 out:
         if (rsp_dict && start_crawl == _gf_true)
                 glusterd_quota_initiate_fs_crawl (priv, volname);
 
+        if (priv->op_version > GD_OP_VERSION_MIN &&
+            is_origin_glusterd ()) {
+                if (type != GF_QUOTA_OPTION_TYPE_LIST) {
+                        if (glusterd_all_volumes_with_quota_stopped ())
+                                ret = glusterd_quotad_stop ();
+                        else
+                                ret = glusterd_check_generate_start_quotad ();
+                }
+        }
+
         if (rsp_dict && *op_errstr) {
                 ret = dict_set_dynstr (rsp_dict, "errstr", *op_errstr);
                 if (ret) {
                         GF_FREE (*op_errstr);
-                        gf_log ("", GF_LOG_DEBUG,
+                        gf_log (this->name, GF_LOG_DEBUG,
                                 "failed to set error message in ctx");
                 }
                 *op_errstr = NULL;
@@ -789,51 +1127,63 @@ glusterd_op_stage_quota (dict_t *dict, char **op_errstr)
         gf_boolean_t       exists        = _gf_false;
         int                type          = 0;
         dict_t            *ctx           = NULL;
+        xlator_t          *this          = NULL;
+        glusterd_conf_t   *priv          = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+        priv = this->private;
+        GF_ASSERT (priv);
 
         GF_ASSERT (dict);
         GF_ASSERT (op_errstr);
 
         ret = dict_get_str (dict, "volname", &volname);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to get volume name");
+                gf_log (this->name, GF_LOG_ERROR, "Unable to get volume name");
                 goto out;
         }
 
         exists = glusterd_check_volume_exists (volname);
         if (!exists) {
-                gf_log ("", GF_LOG_ERROR, "Volume with name: %s "
-                                "does not exist",
-                                volname);
-                *op_errstr = gf_strdup ("Invalid volume name");
+                gf_asprintf (op_errstr, FMTSTR_CHECK_VOL_EXISTS, volname);
                 ret = -1;
                 goto out;
         }
 
         ret = dict_get_int32 (dict, "type", &type);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to get 'type' for quota op");
-                *op_errstr = gf_strdup ("Volume quota failed, internal error "
-                                        ", unable to get type of operation");
+                *op_errstr = gf_strdup ("Volume quota failed, internal error, "
+                                        "unable to get type of operation");
                 goto out;
         }
 
-
-        ctx = glusterd_op_get_ctx();
-        if (ctx && (type == GF_QUOTA_OPTION_TYPE_ENABLE
-                    || type == GF_QUOTA_OPTION_TYPE_LIST)) {
-                /* Fuse mount req. only for enable & list-usage options*/
-                if (!glusterd_is_fuse_available ()) {
-                        gf_log ("glusterd", GF_LOG_ERROR, "Unable to open /dev/"
-                                "fuse (%s), quota command failed",
-                                strerror (errno));
-                        *op_errstr = gf_strdup ("Fuse unavailable");
-                        ret = -1;
-                        goto out;
-                }
+        if ((priv->op_version == GD_OP_VERSION_MIN) &&
+            (type > GF_QUOTA_OPTION_TYPE_VERSION)) {
+                gf_asprintf (op_errstr, "Volume quota failed. The cluster is "
+                                        "operating at version %d. Option %s "
+                                        "is disallowed in this state.",
+                                         priv->op_version,
+                                         gd_quota_op_list[type]);
+                ret = -1;
+                goto out;
         }
 
-out:
-        gf_log ("", GF_LOG_DEBUG, "Returning %d", ret);
-
-        return ret;
+         ctx = glusterd_op_get_ctx();
+         if (ctx && (type == GF_QUOTA_OPTION_TYPE_ENABLE
+                     || type == GF_QUOTA_OPTION_TYPE_LIST)) {
+                 /* Fuse mount req. only for enable & list-usage options*/
+                 if (!glusterd_is_fuse_available ()) {
+                         *op_errstr = gf_strdup ("Fuse unavailable");
+                         ret = -1;
+                         goto out;
+                 }
+         }
+
+ out:
+        if (ret && op_errstr && *op_errstr)
+                gf_log (this->name, GF_LOG_ERROR, "%s", *op_errstr);
+        gf_log (this->name, GF_LOG_DEBUG, "Returning %d", ret);
+
+         return ret;
 }
diff --git a/xlators/mgmt/glusterd/src/glusterd-utils.c b/xlators/mgmt/glusterd/src/glusterd-utils.c
index be594183..0958d527 100644
--- a/xlators/mgmt/glusterd/src/glusterd-utils.c
+++ b/xlators/mgmt/glusterd/src/glusterd-utils.c
@@ -3033,6 +3033,7 @@ glusterd_compare_friend_data (dict_t  *vols, int32_t *status, char *hostname)
         gf_boolean_t            update = _gf_false;
         gf_boolean_t            stale_nfs = _gf_false;
         gf_boolean_t            stale_shd = _gf_false;
+        gf_boolean_t            stale_qd  = _gf_false;
 
         GF_ASSERT (vols);
         GF_ASSERT (status);
@@ -3062,6 +3063,8 @@ glusterd_compare_friend_data (dict_t  *vols, int32_t *status, char *hostname)
                         stale_nfs = _gf_true;
                 if (glusterd_is_nodesvc_running ("glustershd"))
                         stale_shd = _gf_true;
+                if (glusterd_is_nodesvc_running ("quotad"))
+                        stale_qd  = _gf_true;
                 ret = glusterd_import_global_opts (vols);
                 if (ret)
                         goto out;
@@ -3075,6 +3078,8 @@ glusterd_compare_friend_data (dict_t  *vols, int32_t *status, char *hostname)
                                 glusterd_nfs_server_stop ();
                         if (stale_shd)
                                 glusterd_shd_stop ();
+                        if (stale_qd)
+                                glusterd_quotad_stop ();
                 }
         }
 
@@ -3157,7 +3162,10 @@ glusterd_get_nodesvc_volfile (char *server, char *workdir,
         GF_ASSERT (len == PATH_MAX);
 
         glusterd_get_nodesvc_dir (server, workdir, dir, sizeof (dir));
-        snprintf (volfile, len, "%s/%s-server.vol", dir, server);
+        if (strcmp ("quotad", server) != 0)
+                snprintf (volfile, len, "%s/%s-server.vol", dir, server);
+        else
+                snprintf (volfile, len, "%s/%s.vol", dir, server);
 }
 
 void
@@ -3170,11 +3178,14 @@ glusterd_nodesvc_set_online_status (char *server, gf_boolean_t status)
         GF_ASSERT (priv);
         GF_ASSERT (priv->shd);
         GF_ASSERT (priv->nfs);
+        GF_ASSERT (priv->quotad);
 
         if (!strcmp("glustershd", server))
                 priv->shd->online = status;
         else if (!strcmp ("nfs", server))
                 priv->nfs->online = status;
+        else if (!strcmp ("quotad", server))
+                priv->quotad->online = status;
 }
 
 gf_boolean_t
@@ -3188,11 +3199,14 @@ glusterd_is_nodesvc_online (char *server)
         GF_ASSERT (conf);
         GF_ASSERT (conf->shd);
         GF_ASSERT (conf->nfs);
+        GF_ASSERT (conf->quotad);
 
         if (!strcmp (server, "glustershd"))
                 online = conf->shd->online;
         else if (!strcmp (server, "nfs"))
                 online = conf->nfs->online;
+        else if (!strcmp (server, "quotad"))
+                online = conf->quotad->online;
 
         return online;
 }
@@ -3258,11 +3272,14 @@ glusterd_nodesvc_get_rpc (char *server)
         GF_ASSERT (priv);
         GF_ASSERT (priv->shd);
         GF_ASSERT (priv->nfs);
+        GF_ASSERT (priv->quotad);
 
         if (!strcmp (server, "glustershd"))
                 rpc = priv->shd->rpc;
         else if (!strcmp (server, "nfs"))
                 rpc = priv->nfs->rpc;
+        else if (!strcmp (server, "quotad"))
+                rpc = priv->quotad->rpc;
 
         return rpc;
 }
@@ -3280,11 +3297,14 @@ glusterd_nodesvc_set_rpc (char *server, struct rpc_clnt *rpc)
         GF_ASSERT (priv);
         GF_ASSERT (priv->shd);
         GF_ASSERT (priv->nfs);
+        GF_ASSERT (priv->quotad);
 
         if (!strcmp ("glustershd", server))
                 priv->shd->rpc = rpc;
         else if (!strcmp ("nfs", server))
                 priv->nfs->rpc = rpc;
+        else if (!strcmp ("quotad", server))
+                priv->quotad->rpc = rpc;
 
         return ret;
 }
@@ -3411,6 +3431,14 @@ glusterd_nodesvc_start (char *server)
                 runner_add_args (&runner, "--xlator-option",
                                  glusterd_uuid_option, NULL);
         }
+        if (!strcmp (server, "quotad")) {
+                runner_add_args (&runner, "--xlator-option",
+                                 "*replicate*.data-self-heal=off",
+                                 "--xlator-option",
+                                 "*replicate*.metadata-self-heal=off",
+                                 "--xlator-option",
+                                 "*replicate*.entry-self-heal=off", NULL);
+        }
         runner_log (&runner, "", GF_LOG_DEBUG,
                     "Starting the nfs/glustershd services");
 
@@ -3434,6 +3462,12 @@ glusterd_shd_start ()
         return glusterd_nodesvc_start ("glustershd");
 }
 
+int
+glusterd_quotad_start ()
+{
+        return glusterd_nodesvc_start ("quotad");
+}
+
 gf_boolean_t
 glusterd_is_nodesvc_running (char *server)
 {
@@ -3552,6 +3586,12 @@ glusterd_shd_stop ()
 }
 
 int
+glusterd_quotad_stop ()
+{
+        return glusterd_nodesvc_stop ("quotad", SIGTERM);
+}
+
+int
 glusterd_add_node_to_dict (char *server, dict_t *dict, int count,
                            dict_t *vol_opts)
 {
@@ -3701,6 +3741,12 @@ glusterd_reconfigure_shd ()
 }
 
 int
+glusterd_reconfigure_quotad ()
+{
+        return glusterd_reconfigure_nodesvc (glusterd_create_quotad_volfile);
+}
+
+int
 glusterd_reconfigure_nfs ()
 {
         int             ret             = -1;
@@ -3747,21 +3793,54 @@ glusterd_check_generate_start_shd ()
 }
 
 int
-glusterd_nodesvcs_batch_op (glusterd_volinfo_t *volinfo,
-                             int (*nfs_op) (), int (*shd_op) ())
+glusterd_check_generate_start_quotad ()
 {
+        int ret = 0;
+
+        ret = glusterd_check_generate_start_service (glusterd_create_quotad_volfile,
+                                                     glusterd_quotad_stop,
+                                                     glusterd_quotad_start);
+        if (ret == -EINVAL)
+                ret = 0;
+        return ret;
+}
+
+int
+glusterd_nodesvcs_batch_op (glusterd_volinfo_t *volinfo, int (*nfs_op) (),
+                            int (*shd_op) (), int (*qd_op) ())
+ {
         int     ret = 0;
+        xlator_t *this = THIS;
+        glusterd_conf_t *conf = NULL;
+
+        GF_ASSERT (this);
+        conf = this->private;
+        GF_ASSERT (conf);
 
         ret = nfs_op ();
         if (ret)
                 goto out;
 
         if (volinfo && !glusterd_is_volume_replicate (volinfo))
-                goto out;
+                goto quotad_op;
 
         ret = shd_op ();
         if (ret)
                 goto out;
+
+quotad_op:
+
+        if (conf->op_version == GD_OP_VERSION_MIN)
+                goto out;
+
+        if (is_origin_glusterd ()) {
+                if (volinfo && !glusterd_is_volume_quota_enabled (volinfo))
+                        goto out;
+                ret = qd_op ();
+                if (ret)
+                        goto out;
+        }
+
 out:
         return ret;
 }
@@ -3771,7 +3850,8 @@ glusterd_nodesvcs_start (glusterd_volinfo_t *volinfo)
 {
         return glusterd_nodesvcs_batch_op (volinfo,
                                            glusterd_nfs_server_start,
-                                           glusterd_shd_start);
+                                           glusterd_shd_start,
+                                           glusterd_quotad_start);
 }
 
 int
@@ -3779,7 +3859,8 @@ glusterd_nodesvcs_stop (glusterd_volinfo_t *volinfo)
 {
         return glusterd_nodesvcs_batch_op (volinfo,
                                             glusterd_nfs_server_stop,
-                                            glusterd_shd_stop);
+                                            glusterd_shd_stop,
+                                            glusterd_quotad_stop);
 }
 
 gf_boolean_t
@@ -3825,21 +3906,53 @@ glusterd_all_replicate_volumes_stopped ()
         return _gf_true;
 }
 
+gf_boolean_t
+glusterd_all_volumes_with_quota_stopped ()
+{
+        glusterd_conf_t                   *priv     = NULL;
+        xlator_t                          *this     = NULL;
+        glusterd_volinfo_t                *voliter  = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        list_for_each_entry (voliter, &priv->volumes, vol_list) {
+                if (!glusterd_is_volume_quota_enabled (voliter))
+                        continue;
+                if (voliter->status == GLUSTERD_STATUS_STARTED)
+                        return _gf_false;
+        }
+
+        return _gf_true;
+}
+
+
 int
 glusterd_nodesvcs_handle_graph_change (glusterd_volinfo_t *volinfo)
 {
         int (*shd_op) () = NULL;
         int (*nfs_op) () = NULL;
+        int (*qd_op)  () = NULL;
 
         shd_op = glusterd_check_generate_start_shd;
         nfs_op = glusterd_check_generate_start_nfs;
+        qd_op  = glusterd_check_generate_start_quotad;
         if (glusterd_are_all_volumes_stopped ()) {
                 shd_op = glusterd_shd_stop;
                 nfs_op = glusterd_nfs_server_stop;
-        } else if (glusterd_all_replicate_volumes_stopped()) {
-                shd_op = glusterd_shd_stop;
+                qd_op  = glusterd_quotad_stop;
+        } else {
+                if (glusterd_all_replicate_volumes_stopped()) {
+                        shd_op = glusterd_shd_stop;
+                }
+                if (glusterd_all_volumes_with_quota_stopped ()) {
+                        qd_op = glusterd_quotad_stop;
+                }
         }
-        return glusterd_nodesvcs_batch_op (volinfo, nfs_op, shd_op);
+
+        return glusterd_nodesvcs_batch_op (volinfo, nfs_op, shd_op, qd_op);
 }
 
 int
@@ -3847,7 +3960,8 @@ glusterd_nodesvcs_handle_reconfigure (glusterd_volinfo_t *volinfo)
 {
         return glusterd_nodesvcs_batch_op (volinfo,
                                            glusterd_reconfigure_nfs,
-                                           glusterd_reconfigure_shd);
+                                           glusterd_reconfigure_shd,
+                                           glusterd_reconfigure_quotad);
 }
 
 int
@@ -5782,6 +5896,82 @@ out:
         return ret;
 }
 
+int
+glusterd_quotad_statedump (char *options, int option_cnt, char **op_errstr)
+{
+        int                     ret = -1;
+        xlator_t                *this = NULL;
+        glusterd_conf_t         *conf = NULL;
+        char                    pidfile_path[PATH_MAX] = {0,};
+        char                    path[PATH_MAX] = {0,};
+        FILE                    *pidfile = NULL;
+        pid_t                   pid = -1;
+        char                    dumpoptions_path[PATH_MAX] = {0,};
+        char                    *option = NULL;
+        char                    *tmpptr = NULL;
+        char                    *dup_options = NULL;
+        char                    msg[256] = {0,};
+
+        this = THIS;
+        GF_ASSERT (this);
+        conf = this->private;
+        GF_ASSERT (conf);
+
+        dup_options = gf_strdup (options);
+        option = strtok_r (dup_options, " ", &tmpptr);
+        if (strcmp (option, "quotad")) {
+                snprintf (msg, sizeof (msg), "for quotad statedump, options "
+                          "should be after the key 'quotad'");
+                *op_errstr = gf_strdup (msg);
+                ret = -1;
+                goto out;
+        }
+
+        GLUSTERD_GET_QUOTAD_DIR (path, conf);
+        GLUSTERD_GET_QUOTAD_PIDFILE (pidfile_path, path);
+
+        pidfile = fopen (pidfile_path, "r");
+        if (!pidfile) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to open pidfile: %s",
+                        pidfile_path);
+                ret = -1;
+                goto out;
+        }
+
+        ret = fscanf (pidfile, "%d", &pid);
+        if (ret <= 0) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to get pid of quotad "
+                        "process");
+                ret = -1;
+                goto out;
+        }
+
+        snprintf (dumpoptions_path, sizeof (dumpoptions_path),
+                  DEFAULT_VAR_RUN_DIRECTORY"/glusterdump.%d.options", pid);
+        ret = glusterd_set_dump_options (dumpoptions_path, options, option_cnt);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_ERROR, "error while parsing "
+                        "statedump options");
+                ret = -1;
+                goto out;
+        }
+
+        gf_log (this->name, GF_LOG_INFO, "Performing statedump on quotad with "
+                "pid %d", pid);
+
+        kill (pid, SIGUSR1);
+
+        sleep (1);
+
+        ret = 0;
+out:
+        if (pidfile)
+                fclose (pidfile);
+        unlink (dumpoptions_path);
+        GF_FREE (dup_options);
+        return ret;
+}
+
 /* Checks if the given peer contains all the bricks belonging to the
  * given volume. Returns true if it does else returns false
  */
@@ -7404,3 +7594,9 @@ gd_is_remove_brick_committed (glusterd_volinfo_t *volinfo)
 
         return _gf_true;
 }
+
+int
+glusterd_is_volume_quota_enabled (glusterd_volinfo_t *volinfo)
+{
+        return (glusterd_volinfo_get_boolean (volinfo, VKEY_FEATURES_QUOTA));
+}
diff --git a/xlators/mgmt/glusterd/src/glusterd-utils.h b/xlators/mgmt/glusterd/src/glusterd-utils.h
index 912b394b..c1291819 100644
--- a/xlators/mgmt/glusterd/src/glusterd-utils.h
+++ b/xlators/mgmt/glusterd/src/glusterd-utils.h
@@ -188,6 +188,12 @@ glusterd_shd_start ();
 int32_t
 glusterd_shd_stop ();
 
+int32_t
+glusterd_quotad_start ();
+
+int32_t
+glusterd_quotad_stop ();
+
 void
 glusterd_set_socket_filepath (char *sock_filepath, char *sockpath, size_t len);
 
@@ -228,6 +234,9 @@ int
 glusterd_check_generate_start_shd (void);
 
 int
+glusterd_check_generate_start_quotad (void);
+
+int
 glusterd_nodesvcs_handle_graph_change (glusterd_volinfo_t *volinfo);
 
 int
@@ -395,8 +404,13 @@ glusterd_brick_statedump (glusterd_volinfo_t *volinfo,
                           char *options, int option_cnt, char **op_errstr);
 int
 glusterd_nfs_statedump (char *options, int option_cnt, char **op_errstr);
+
+int
+glusterd_quotad_statedump (char *options, int option_cnt, char **op_errstr);
+
 gf_boolean_t
 glusterd_is_volume_replicate (glusterd_volinfo_t *volinfo);
+
 gf_boolean_t
 glusterd_is_brick_decommissioned (glusterd_volinfo_t *volinfo, char *hostname,
                                   char *path);
@@ -531,4 +545,13 @@ gd_peer_uuid_str (glusterd_peerinfo_t *peerinfo);
 
 gf_boolean_t
 gd_is_remove_brick_committed (glusterd_volinfo_t *volinfo);
+
+int
+glusterd_is_volume_quota_enabled (glusterd_volinfo_t *volinfo);
+
+gf_boolean_t
+glusterd_all_volumes_with_quota_stopped ();
+
+int
+glusterd_reconfigure_quotad ();
 #endif
diff --git a/xlators/mgmt/glusterd/src/glusterd-volgen.c b/xlators/mgmt/glusterd/src/glusterd-volgen.c
index 127e4c46..148cea9c 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volgen.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volgen.c
@@ -1423,6 +1423,7 @@ server_graph_builder (volgen_graph_t *graph, glusterd_volinfo_t *volinfo,
         char     *vg                      = NULL;
         glusterd_brickinfo_t *brickinfo   = NULL;
         char changelog_basepath[PATH_MAX] = {0,};
+        char     *value                   = NULL;
 
         brickinfo = param;
         path      = brickinfo->path;
@@ -1607,7 +1608,18 @@ server_graph_builder (volgen_graph_t *graph, glusterd_volinfo_t *volinfo,
         if (ret)
                 return -1;
 
-        if (dict_get_str_boolean (set_dict, "features.read-only", 0) &&
+        xl = volgen_graph_add (graph, "features/quota", volname);
+        if (!xl)
+                return -1;
+
+        ret = glusterd_volinfo_get (volinfo, VKEY_FEATURES_QUOTA, &value);
+        if (value) {
+                ret = xlator_set_option (xl, "server-quota", value);
+                if (ret)
+                        return -1;
+        }
+
+         if (dict_get_str_boolean (set_dict, "features.read-only", 0) &&
             dict_get_str_boolean (set_dict, "features.worm",0)) {
                 gf_log (THIS->name, GF_LOG_ERROR,
                         "read-only and worm cannot be set together");
@@ -2315,7 +2327,8 @@ out:
 
 static int
 volgen_graph_build_dht_cluster (volgen_graph_t *graph,
-                                glusterd_volinfo_t *volinfo, size_t child_count)
+                                glusterd_volinfo_t *volinfo, size_t child_count,
+                                gf_boolean_t is_quotad)
 {
         int32_t                 clusters                 = 0;
         int                     ret                      = -1;
@@ -2323,17 +2336,23 @@ volgen_graph_build_dht_cluster (volgen_graph_t *graph,
         xlator_t                *dht                     = NULL;
         char                    *optstr                  = NULL;
         gf_boolean_t             use_nufa                = _gf_false;
+        char                    *name_fmt                = NULL;
 
         if (dict_get_str(volinfo->dict,"cluster.nufa",&optstr) == 0) {
                 /* Keep static analyzers quiet by "using" the value. */
                 ret = gf_string2boolean(optstr,&use_nufa);
         }
 
+        if (is_quotad)
+                name_fmt = "%s";
+        else
+                name_fmt = "%s-dht";
+
         clusters = volgen_graph_build_clusters (graph,  volinfo,
                                                 use_nufa
                                                         ? "cluster/nufa"
                                                         : "cluster/distribute",
-                                                "%s-dht",
+                                                name_fmt,
                                                 child_count, child_count);
         if (clusters < 0)
                 goto out;
@@ -2356,7 +2375,8 @@ out:
 
 static int
 volume_volgen_graph_build_clusters (volgen_graph_t *graph,
-                                    glusterd_volinfo_t *volinfo)
+                                    glusterd_volinfo_t *volinfo,
+                                    gf_boolean_t is_quotad)
 {
         char                    *replicate_args[]   = {"cluster/replicate",
                                                        "%s-replicate-%d"};
@@ -2429,7 +2449,7 @@ build_distribute:
         }
 
         ret = volgen_graph_build_dht_cluster (graph, volinfo,
-                                              dist_count);
+                                              dist_count, is_quotad);
         if (ret)
                 goto out;
 
@@ -2446,25 +2466,30 @@ client_graph_builder (volgen_graph_t *graph, glusterd_volinfo_t *volinfo,
         xlator_t *xl       = NULL;
         char     *volname  = NULL;
         data_t   *tmp_data = NULL;
+        glusterd_conf_t *conf = THIS->private;
+        GF_ASSERT (conf);
 
         volname = volinfo->volname;
         ret = volgen_graph_build_clients (graph, volinfo, set_dict, param);
         if (ret)
                 goto out;
 
-        ret = volume_volgen_graph_build_clusters (graph, volinfo);
+        ret = volume_volgen_graph_build_clusters (graph, volinfo, _gf_false);
         if (ret)
                 goto out;
 
-        ret = glusterd_volinfo_get_boolean (volinfo, VKEY_FEATURES_QUOTA);
-        if (ret == -1)
-                goto out;
-        if (ret) {
-                xl = volgen_graph_add (graph, "features/quota", volname);
-
-                if (!xl) {
-                        ret = -1;
+        if (conf->op_version == GD_OP_VERSION_MIN) {
+                ret = glusterd_volinfo_get_boolean (volinfo,
+                                                    VKEY_FEATURES_QUOTA);
+                if (ret == -1)
                         goto out;
+                if (ret) {
+                        xl = volgen_graph_add (graph, "features/quota",
+                                              volname);
+                        if (!xl) {
+                                ret = -1;
+                                goto out;
+                        }
                 }
         }
 
@@ -2734,6 +2759,60 @@ nfs_option_handler (volgen_graph_t *graph,
 }
 
 static int
+set_quotad_xlator_option (char *fmt_str, char *opt_name, char *volname,
+                          xlator_t *xl, char *value)
+{
+        int ret = -1;
+        char *opt_str = NULL;
+
+        ret = gf_asprintf (&opt_str, fmt_str, volname, opt_name);
+        if (opt_str == NULL) {
+                ret = -1;
+                goto out;
+        }
+        ret = xlator_set_option (xl, opt_str, value);
+
+out:
+        GF_FREE (opt_str);
+        return ret;
+}
+
+int
+quotad_option_handler (volgen_graph_t *graph, struct volopt_map_entry *vme,
+                       void *param)
+{
+        xlator_t            *xl        = NULL;
+        char                *opt_str   = NULL;
+        int                  ret       = 0;
+        glusterd_volinfo_t  *volinfo   = NULL;
+        int                  i         = 0;
+        char                *ptr       = NULL;
+        char                *options[] = {"!*.soft-timeout", "!*.hard-timeout",
+                                          "limit-set", "!*.alert-time",
+                                          "default-soft-limit", NULL};
+
+        volinfo = param;
+        xl = first_of (graph);
+
+        for (i = 0; options[i]; i++) {
+                if (!strcmp (vme->option, options[i])) {
+                        ptr = strchr (options[i], '.');
+                        if (!ptr)
+                                opt_str = options[i];
+                        else
+                                opt_str = ptr + 1;
+                        ret = set_quotad_xlator_option ("%s.%s", opt_str,
+                                                        volinfo->volname, xl,
+                                                        vme->value);
+                        if (ret)
+                                return -1;
+                }
+        }
+
+        return 0;
+}
+
+static int
 volgen_graph_set_iam_shd (volgen_graph_t *graph)
 {
         xlator_t        *trav;
@@ -3018,9 +3097,6 @@ build_nfs_graph (volgen_graph_t *graph, dict_t *mod_dict)
         return ret;
 }
 
-
-
-
 /****************************
  *
  * Volume generation interface
@@ -3107,7 +3183,109 @@ glusterd_generate_brick_volfile (glusterd_volinfo_t *volinfo,
         return ret;
 }
 
+static int
+build_quotad_graph (volgen_graph_t *graph, dict_t *mod_dict)
+{
+        volgen_graph_t     cgraph         = {0};
+        glusterd_volinfo_t *voliter       = NULL;
+        xlator_t           *this          = NULL;
+        glusterd_conf_t    *priv          = NULL;
+        dict_t             *set_dict      = NULL;
+        int                ret            = 0;
+        xlator_t           *quotad_xl     = NULL;
+
+        this = THIS;
+        priv = this->private;
+
+        set_dict = dict_new ();
+        if (!set_dict) {
+                ret = -ENOMEM;
+                goto out;
+        }
+
+        quotad_xl = volgen_graph_add_as (graph, "features/quotad", "quotad");
+        if (!quotad_xl) {
+                ret = -1;
+                goto out;
+        }
+
+        list_for_each_entry (voliter, &priv->volumes, vol_list) {
+                if (voliter->status != GLUSTERD_STATUS_STARTED)
+                        continue;
+
+                if (1 != glusterd_is_volume_quota_enabled (voliter))
+                        continue;
+
+                ret = dict_set_uint32 (set_dict, "trusted-client",
+                                       GF_CLIENT_TRUSTED);
+                if (ret)
+                        goto out;
+
+                dict_copy (voliter->dict, set_dict);
+                if (mod_dict)
+                        dict_copy (mod_dict, set_dict);
+
+                memset (&cgraph, 0, sizeof (cgraph));
+                ret = volgen_graph_build_clients (&cgraph, voliter, set_dict,
+                                                  NULL);
+                if (ret)
+                        goto out;
+
+                ret = volume_volgen_graph_build_clusters (&cgraph, voliter,
+                                                          _gf_true);
+                if (ret) {
+                        ret = -1;
+                        goto out;
+                }
+
+                if (mod_dict) {
+                        dict_copy (mod_dict, set_dict);
+                        ret = volgen_graph_set_options_generic (&cgraph, set_dict,
+                                                        voliter,
+                                                        basic_option_handler);
+                } else {
+                        ret = volgen_graph_set_options_generic (&cgraph,
+                                                                voliter->dict,
+                                                                voliter,
+                                                                basic_option_handler);
+                }
+                if (ret)
+                        goto out;
+
+                ret = volgen_graph_merge_sub (graph, &cgraph, 1);
+                if (ret)
+                        goto out;
+
+                ret = dict_reset (set_dict);
+                if (ret)
+                        goto out;
+        }
 
+        list_for_each_entry (voliter, &priv->volumes, vol_list) {
+                if (voliter->status != GLUSTERD_STATUS_STARTED)
+                        continue;
+
+                if (1 != glusterd_is_volume_quota_enabled (voliter))
+                        continue;
+                if (mod_dict) {
+                        ret = volgen_graph_set_options_generic (graph, mod_dict,
+                                                                voliter,
+                                                        quotad_option_handler);
+                } else {
+                        ret = volgen_graph_set_options_generic (graph,
+                                                         voliter->dict, voliter,
+                                                         quotad_option_handler);
+                }
+
+                if (ret)
+                        gf_log (THIS->name, GF_LOG_ERROR, "Failed to set "
+                                "options for the volume %s", voliter->volname);
+        }
+out:
+        if (set_dict)
+                dict_unref (set_dict);
+        return ret;
+}
 
 static void
 get_vol_tstamp_file (char *filename, glusterd_volinfo_t *volinfo)
@@ -3375,6 +3553,18 @@ out:
 }
 
 int
+glusterd_create_quotad_volfile ()
+{
+        char             filepath[PATH_MAX] = {0,};
+        glusterd_conf_t *conf               = THIS->private;
+
+        glusterd_get_nodesvc_volfile ("quotad", conf->workdir,
+                                            filepath, sizeof (filepath));
+        return glusterd_create_global_volfile (build_quotad_graph,
+                                               filepath, NULL);
+}
+
+int
 glusterd_check_nfs_volfile_identical (gf_boolean_t *identical)
 {
         char            nfsvol[PATH_MAX]        = {0,};
diff --git a/xlators/mgmt/glusterd/src/glusterd-volgen.h b/xlators/mgmt/glusterd/src/glusterd-volgen.h
index 6a41f47c..deb4da85 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volgen.h
+++ b/xlators/mgmt/glusterd/src/glusterd-volgen.h
@@ -22,6 +22,7 @@
 #define VKEY_DIAG_CNT_FOP_HITS    "diagnostics.count-fop-hits"
 #define VKEY_DIAG_LAT_MEASUREMENT "diagnostics.latency-measurement"
 #define VKEY_FEATURES_LIMIT_USAGE "features.limit-usage"
+#define VKEY_FEATURES_SOFT_LIMIT  "features.soft-limit"
 #define VKEY_MARKER_XTIME         GEOREP".indexing"
 #define VKEY_FEATURES_QUOTA       "features.quota"
 
@@ -118,6 +119,7 @@ void glusterd_get_shd_filepath (char *filename);
 
 int glusterd_create_nfs_volfile ();
 int glusterd_create_shd_volfile ();
+int glusterd_create_quotad_volfile ();
 
 int glusterd_delete_volfile (glusterd_volinfo_t *volinfo,
                              glusterd_brickinfo_t *brickinfo);
diff --git a/xlators/mgmt/glusterd/src/glusterd-volume-ops.c b/xlators/mgmt/glusterd/src/glusterd-volume-ops.c
index 17b3f618..2907dcfe 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volume-ops.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volume-ops.c
@@ -526,9 +526,12 @@ __glusterd_handle_cli_statedump_volume (rpcsvc_request_t *req)
         glusterd_op_t                   cli_op = GD_OP_STATEDUMP_VOLUME;
         char                            err_str[2048] = {0,};
         xlator_t                        *this = NULL;
+        glusterd_conf_t                 *priv = NULL;
 
         this = THIS;
         GF_ASSERT (this);
+        priv = this->private;
+        GF_ASSERT (priv);
 
         GF_ASSERT (req);
 
@@ -577,6 +580,14 @@ __glusterd_handle_cli_statedump_volume (rpcsvc_request_t *req)
                 goto out;
         }
 
+        if (priv->op_version == GD_OP_VERSION_MIN &&
+            strstr (options, "quotad")) {
+                snprintf (err_str, sizeof (err_str), "The cluster is operating "
+                          "at op-version 1. Taking quotad's statedump is "
+                          "disallowed in this state");
+                ret = -1;
+                goto out;
+        }
 
         gf_log (this->name, GF_LOG_INFO, "Received statedump request for "
                 "volume %s with options %s", volname, options);
@@ -1236,6 +1247,13 @@ glusterd_op_stage_statedump_volume (dict_t *dict, char **op_errstr)
         gf_boolean_t            is_running = _gf_false;
         glusterd_volinfo_t      *volinfo = NULL;
         char                    msg[2408] = {0,};
+        xlator_t                *this     = NULL;
+        glusterd_conf_t         *priv     = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+        priv = this->private;
+        GF_ASSERT (priv);
 
         ret = glusterd_op_statedump_volume_args_get (dict, &volname, &options,
                                                      &option_cnt);
@@ -1244,10 +1262,7 @@ glusterd_op_stage_statedump_volume (dict_t *dict, char **op_errstr)
 
         ret = glusterd_volinfo_find (volname, &volinfo);
         if (ret) {
-                snprintf (msg, sizeof(msg), "Volume %s does not exist",
-                          volname);
-                gf_log ("", GF_LOG_ERROR, "%s", msg);
-                *op_errstr = gf_strdup (msg);
+                snprintf (msg, sizeof(msg), FMTSTR_CHECK_VOL_EXISTS, volname);
                 goto out;
         }
 
@@ -1257,16 +1272,31 @@ glusterd_op_stage_statedump_volume (dict_t *dict, char **op_errstr)
 
         is_running = glusterd_is_volume_started (volinfo);
         if (!is_running) {
-                snprintf (msg, sizeof(msg), "Volume %s is not in a started"
+                snprintf (msg, sizeof(msg), "Volume %s is not in the started"
                           " state", volname);
-                gf_log ("", GF_LOG_ERROR, "%s", msg);
-                *op_errstr = gf_strdup (msg);
                 ret = -1;
                 goto out;
         }
 
+        if (priv->op_version == GD_OP_VERSION_MIN &&
+            strstr (options, "quotad")) {
+                snprintf (msg, sizeof (msg), "The cluster is operating "
+                          "at op-version 1. Taking quotad's statedump is "
+                          "disallowed in this state");
+                ret = -1;
+                goto out;
+        }
+        if ((strstr (options, "quotad")) &&
+            (!glusterd_is_volume_quota_enabled (volinfo))) {
+                    snprintf (msg, sizeof (msg), "Quota is not enabled on "
+                              "volume %s", volname);
+                    ret = -1;
+                    goto out;
+        }
 out:
-        gf_log ("", GF_LOG_DEBUG, "Returning %d", ret);
+        if (ret && msg[0] != '\0')
+                *op_errstr = gf_strdup (msg);
+        gf_log (this->name, GF_LOG_DEBUG, "Returning %d", ret);
         return ret;
 }
 
@@ -1827,6 +1857,18 @@ glusterd_op_statedump_volume (dict_t *dict, char **op_errstr)
                 ret = glusterd_nfs_statedump (options, option_cnt, op_errstr);
                 if (ret)
                         goto out;
+
+        } else if (strstr (options, "quotad")) {
+                if (is_origin_glusterd()) {
+                        ret = glusterd_quotad_statedump (options, option_cnt,
+                                                         op_errstr);
+                if (ret)
+                        goto out;
+                } else {
+                        ret = 0;
+                        goto out;
+                }
+
         } else {
                 list_for_each_entry (brickinfo, &volinfo->bricks,
                                                         brick_list) {
diff --git a/xlators/mgmt/glusterd/src/glusterd-volume-set.c b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
index 245b29af..600100aa 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volume-set.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
@@ -920,29 +920,51 @@ struct volopt_map_entry glusterd_volopt_map[] = {
         },
 
         /* Quota xlator options */
-        { .key        = VKEY_FEATURES_LIMIT_USAGE,
-          .voltype    = "features/quota",
-          .option     = "limit-set",
-          .type       = NO_DOC,
-          .op_version = 1,
-          .flags      = OPT_FLAG_CLIENT_OPT
-        },
-        { .key         = "features.quota-timeout",
-          .voltype     = "features/quota",
-          .option      = "timeout",
-          .value       = "0",
-          .op_version  = 1,
-          .validate_fn = validate_quota,
-          .flags       = OPT_FLAG_CLIENT_OPT
-        },
-        { .key         = "features.quota-deem-statfs",
-          .voltype     = "features/quota",
-          .option      = "deem-statfs",
-          .value       = "off",
-          .type        = DOC,
-          .op_version  = 3,
-          .validate_fn = validate_quota,
-          .flags       = OPT_FLAG_CLIENT_OPT
+        { .key           = VKEY_FEATURES_LIMIT_USAGE,
+          .voltype       = "features/quota",
+          .option        = "limit-set",
+          .type          = NO_DOC,
+          .op_version    = 1,
+        },
+        {
+          .key           = "features.quota-timeout",
+          .voltype       = "features/quota",
+          .option        = "timeout",
+          .value         = "0",
+          .op_version    = 1,
+          .validate_fn   = validate_quota,
+        },
+        { .key           = "features.default-soft-limit",
+          .voltype       = "features/quota",
+          .option        = "default-soft-limit",
+          .type          = NO_DOC,
+          .op_version    = 3,
+        },
+        { .key           = "features.soft-timeout",
+          .voltype       = "features/quotad",
+          .option        = "!*.soft-timeout",
+          .type          = NO_DOC,
+          .op_version    = 3,
+        },
+        { .key           = "features.hard-timeout",
+          .voltype       = "features/quotad",
+          .option        = "!*.hard-timeout",
+          .type          = NO_DOC,
+          .op_version    = 3,
+        },
+        { .key           = "features.alert-time",
+          .voltype       = "features/quotad",
+          .option        = "!*.alert-time",
+          .type          = NO_DOC,
+          .op_version    = 3,
+        },
+        { .key           = "features.quota-deem-statfs",
+          .voltype       = "features/quota",
+          .option        = "deem-statfs",
+          .value         = "off",
+          .type          = DOC,
+          .op_version    = 2,
+          .validate_fn   = validate_quota,
         },
 
         /* Marker xlator options */
diff --git a/xlators/mgmt/glusterd/src/glusterd.c b/xlators/mgmt/glusterd/src/glusterd.c
index 84c72657..1ac28bc8 100644
--- a/xlators/mgmt/glusterd/src/glusterd.c
+++ b/xlators/mgmt/glusterd/src/glusterd.c
@@ -1019,6 +1019,15 @@ init (xlator_t *this)
                 exit (1);
         }
 
+        snprintf (storedir, PATH_MAX, "%s/quotad", workdir);
+        ret = mkdir (storedir, 0777);
+        if ((-1 == ret) && (errno != EEXIST)) {
+                gf_log (this->name, GF_LOG_CRITICAL,
+                        "Unable to create quotad directory %s"
+                        " ,errno = %d", storedir, errno);
+                exit (1);
+        }
+
         snprintf (storedir, PATH_MAX, "%s/groups", workdir);
         ret = mkdir (storedir, 0777);
         if ((-1 == ret) && (errno != EEXIST)) {
@@ -1072,12 +1081,14 @@ init (xlator_t *this)
         conf = GF_CALLOC (1, sizeof (glusterd_conf_t),
                           gf_gld_mt_glusterd_conf_t);
         GF_VALIDATE_OR_GOTO(this->name, conf, out);
-        conf->shd = GF_CALLOC (1, sizeof (nodesrv_t),
-                               gf_gld_mt_nodesrv_t);
+
+        conf->shd = GF_CALLOC (1, sizeof (nodesrv_t), gf_gld_mt_nodesrv_t);
         GF_VALIDATE_OR_GOTO(this->name, conf->shd, out);
-        conf->nfs = GF_CALLOC (1, sizeof (nodesrv_t),
-                               gf_gld_mt_nodesrv_t);
+        conf->nfs = GF_CALLOC (1, sizeof (nodesrv_t), gf_gld_mt_nodesrv_t);
         GF_VALIDATE_OR_GOTO(this->name, conf->nfs, out);
+        conf->quotad = GF_CALLOC (1, sizeof (nodesrv_t),
+                               gf_gld_mt_nodesrv_t);
+        GF_VALIDATE_OR_GOTO(this->name, conf->quotad, out);
 
         INIT_LIST_HEAD (&conf->peers);
         INIT_LIST_HEAD (&conf->volumes);
diff --git a/xlators/mgmt/glusterd/src/glusterd.h b/xlators/mgmt/glusterd/src/glusterd.h
index 6bd9431e..055df1fb 100644
--- a/xlators/mgmt/glusterd/src/glusterd.h
+++ b/xlators/mgmt/glusterd/src/glusterd.h
@@ -127,6 +127,7 @@ typedef struct {
         rpcsvc_t          *rpc;
         nodesrv_t         *shd;
         nodesrv_t         *nfs;
+        nodesrv_t         *quotad;
         struct pmap_registry *pmap;
         struct list_head  volumes;
         pthread_mutex_t   xprt_lock;
@@ -306,6 +307,7 @@ typedef enum gd_node_type_ {
         GD_NODE_SHD,
         GD_NODE_REBALANCE,
         GD_NODE_NFS,
+        GD_NODE_QUOTAD,
 } gd_node_type;
 
 typedef struct glusterd_pending_node_ {
@@ -361,6 +363,9 @@ typedef ssize_t (*gd_serialize_t) (struct iovec outmsg, void *args);
 #define GLUSTERD_GET_NFS_DIR(path, priv) \
         snprintf (path, PATH_MAX, "%s/nfs", priv->workdir);
 
+#define GLUSTERD_GET_QUOTAD_DIR(path, priv) \
+        snprintf (path, PATH_MAX, "%s/quotad", priv->workdir);
+
 #define GLUSTERD_REMOVE_SLASH_FROM_PATH(path,string) do {               \
                 int i = 0;                                              \
                 for (i = 1; i < strlen (path); i++) {                   \
@@ -384,6 +389,11 @@ typedef ssize_t (*gd_serialize_t) (struct iovec outmsg, void *args);
                           nfspath);                                     \
         }
 
+#define GLUSTERD_GET_QUOTAD_PIDFILE(pidfile,quotadpath) {                     \
+                snprintf (pidfile, PATH_MAX, "%s/run/quotad.pid",          \
+                          quotadpath);                                     \
+        }
+
 #define GLUSTERD_STACK_DESTROY(frame) do {\
 		frame->local = NULL;	  \
 		STACK_DESTROY (frame->root);\