features/quota : Introducing inode quota

==========================================================================
                             Inode quota
==========================================================================
= Currently, the only way to retrieve the number of files/objects in a   =
= directory or volume is to do a crawl of the entire directory/volume.   =
= This is expensive and is not scalable.                                 =
=                                                                        =
= The proposed mechanism will provide an easier alternative to determine =
= the count of files/objects in a directory or volume.                   =
=                                                                        =
= The new mechanism proposes to store count of objects/files as part of  =
= an extended attribute of a directory. Each directory's extended        =
= attribute value will indicate the number of files/objects present      =
= in a tree with the directory being considered as the root of the tree. =
=                                                                        =
= The count value can be accessed by performing a getxattr().            =
= Cluster translators like afr, dht and stripe will perform aggregation  =
= of count values from various bricks when getxattr() happens on the key =
= associated with file/object count.                                     =

A new interface is introduced:
------------------------------
        limit-objects  : limit the number of inodes at directory level
        list-objects   : list the directories where the limit is set
        remove-objects : remove the limit from the directory

==========================================================================

CLI COMMAND:
gluster volume quota <volname> limit-objects <path> <number> [<percent>]

* <number> is a hard-limit for number of objects limitation for path "<path>"
  If hard-limit is exceeded, creation of file/directory is no longer
permitted.

* <percent> is a soft-limit for number of objects creation for path "<path>"
  If soft-limit is exceeded, a warning is issued for each creation.

CLI COMMAND:
gluster volume quota <volname> remove-objects [path]

==========================================================================

CLI COMMAND:
gluster volume quota <volname> list-objects [path] ...

Sample output:
------------------
  Path                   Hard-limit Soft-limit   Used  Available
Soft-limit exceeded?
Hard-limit exceeded?
  ------------------------------------------------------------------------
--------------------------------------
  /dir                      10       80%          10       0
Yes
        Yes

==========================================================================

[root@snapshot-28 dir]# ls
a  b  file11  file12  file13  file14  file15  file16  file17
[root@snapshot-28 dir]# touch a1
touch: cannot touch `a1': Disk quota exceeded
* Nine files are created in directory "dir" and directory is included in
* the
count too. Hence the limit "10" is reached and further file creation
fails

==========================================================================

Note: We have also done some re-factoring in cli for volume name
validation. New function cli_validate_volname is created

==========================================================================

Change-Id: I1823497de4f790a2a20ebb1770293472ea33ee2b
BUG: 1190108
Signed-off-by: Sachin Pandit <spandit@redhat.com>
Signed-off-by: vmallika <vmallika@redhat.com>
Reviewed-on: http://review.gluster.org/9769
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

12 files changed, 599 insertions, 350 deletions

diff --git a/xlators/cluster/afr/src/afr-inode-read.c b/xlators/cluster/afr/src/afr-inode-read.c
index 25d576d0261..77acf0b25b9 100644
--- a/xlators/cluster/afr/src/afr-inode-read.c
+++ b/xlators/cluster/afr/src/afr-inode-read.c
@@ -35,6 +35,7 @@
 #include "common-utils.h"
 #include "compat-errno.h"
 #include "compat.h"
+#include "quota-common-utils.h"
 
 #include "afr-transaction.h"
 
@@ -50,15 +51,16 @@
 int
 afr_handle_quota_size (call_frame_t *frame, xlator_t *this)
 {
-        unsigned char *readable = NULL;
-        afr_local_t *local = NULL;
-        afr_private_t *priv = NULL;
+        unsigned char *readable   = NULL;
+        afr_local_t *local        = NULL;
+        afr_private_t *priv       = NULL;
         struct afr_reply *replies = NULL;
-        int i = 0;
-        uint64_t size = 0;
-        uint64_t max_size = 0;
-        int readable_cnt = 0;
-        int read_subvol = -1;
+        int i                     = 0;
+        int ret                   = 0;
+        quota_meta_t size         = {0, };
+        quota_meta_t max_size     = {0, };
+        int readable_cnt          = 0;
+        int read_subvol           = -1;
 
         local = frame->local;
         priv = this->private;
@@ -77,22 +79,29 @@ afr_handle_quota_size (call_frame_t *frame, xlator_t *this)
                         continue;
                 if (!replies[i].xdata)
                         continue;
-                if (dict_get_uint64 (replies[i].xdata, QUOTA_SIZE_KEY, &size))
+                ret = quota_dict_get_meta (replies[i].xdata, QUOTA_SIZE_KEY,
+                                           &size);
+                if (ret == -1)
                         continue;
-                size = ntoh64 (size);
                 if (read_subvol == -1)
                         read_subvol = i;
-                if (size > max_size) {
+                if (size.size > max_size.size ||
+                    (size.file_count + size.dir_count) >
+                    (max_size.file_count + max_size.dir_count))
                         read_subvol = i;
-                        max_size = size;
-                }
+
+                if (size.size > max_size.size)
+                        max_size.size = size.size;
+                if (size.file_count > max_size.file_count)
+                        max_size.file_count = size.file_count;
+                if (size.dir_count > max_size.dir_count)
+                        max_size.dir_count = size.dir_count;
         }
 
-        if (!max_size)
+        if (max_size.size == 0 && max_size.file_count == 0 &&
+            max_size.dir_count == 0)
                 return read_subvol;
 
-        max_size = hton64 (max_size);
-
         for (i = 0; i < priv->child_count; i++) {
                 if (!replies[i].valid || replies[i].op_ret == -1)
                         continue;
@@ -100,8 +109,8 @@ afr_handle_quota_size (call_frame_t *frame, xlator_t *this)
                         continue;
                 if (!replies[i].xdata)
                         continue;
-                if (dict_set_uint64 (replies[i].xdata, QUOTA_SIZE_KEY, max_size))
-                        continue;
+                quota_dict_set_meta (replies[i].xdata, QUOTA_SIZE_KEY,
+                                     &max_size, IA_IFDIR);
         }
 
         return read_subvol;
diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index 729ab30e672..9fda4aa07d6 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -23,6 +23,7 @@
 #include "defaults.h"
 #include "byte-order.h"
 #include "glusterfs-acl.h"
+#include "quota-common-utils.h"
 
 #include <sys/time.h>
 #include <libgen.h>
@@ -31,47 +32,90 @@
 int dht_link2 (xlator_t *this, call_frame_t *frame, int op_ret);
 
 int
-dht_aggregate (dict_t *this, char *key, data_t *value, void *data)
+dht_aggregate_quota_xattr (dict_t *dst, char *key, data_t *value)
 {
-        dict_t  *dst  = NULL;
-        int64_t *ptr  = 0, *size = NULL;
-        int32_t  ret  = -1;
-        data_t  *dict_data = NULL;
-
-        dst = data;
+        int              ret            = -1;
+        quota_meta_t    *meta_dst       = NULL;
+        quota_meta_t    *meta_src       = NULL;
+        int64_t         *size           = NULL;
+        int64_t          dst_dir_count  = 0;
+        int64_t          src_dir_count  = 0;
+
+        if (value == NULL) {
+                gf_log ("dht", GF_LOG_WARNING, "data value is NULL");
+                ret = -1;
+                goto out;
+        }
 
-        if (strcmp (key, GF_XATTR_QUOTA_SIZE_KEY) == 0) {
-                ret = dict_get_bin (dst, key, (void **)&size);
+        ret = dict_get_bin (dst, key, (void **)&meta_dst);
+        if (ret < 0) {
+                meta_dst = GF_CALLOC (1, sizeof (quota_meta_t),
+                                      gf_common_quota_meta_t);
+                if (meta_dst == NULL) {
+                        gf_msg ("dht", GF_LOG_WARNING, 0, DHT_MSG_NO_MEMORY,
+                                "Memory allocation failed");
+                        ret = -1;
+                        goto out;
+                }
+                ret = dict_set_bin (dst, key, meta_dst,
+                                    sizeof (quota_meta_t));
                 if (ret < 0) {
-                        size = GF_CALLOC (1, sizeof (int64_t),
-                                          gf_common_mt_char);
-                        if (size == NULL) {
-                                gf_msg ("dht", GF_LOG_WARNING, 0,
-                                        DHT_MSG_NO_MEMORY,
-                                        "Memory allocation failed");
-                                return -1;
-                        }
-                        ret = dict_set_bin (dst, key, size, sizeof (int64_t));
-                        if (ret < 0) {
-                                gf_log ("dht", GF_LOG_WARNING,
-                                        "dht aggregate dict set failed");
-                                GF_FREE (size);
-                                return -1;
-                        }
+                        gf_log ("dht", GF_LOG_WARNING,
+                                "dht aggregate dict set failed");
+                        GF_FREE (meta_dst);
+                        ret = -1;
+                        goto out;
                 }
+        }
+
+        if (value->len > sizeof (int64_t)) {
+                meta_src = data_to_bin (value);
+
+                meta_dst->size = hton64 (ntoh64 (meta_dst->size) +
+                                         ntoh64 (meta_src->size));
+                meta_dst->file_count = hton64 (ntoh64 (meta_dst->file_count) +
+                                               ntoh64 (meta_src->file_count));
 
-                ptr = data_to_bin (value);
-                if (ptr == NULL) {
-                        gf_log ("dht", GF_LOG_WARNING, "data to bin failed");
-                        return -1;
+                if (value->len > (2 * sizeof (int64_t))) {
+                        dst_dir_count = ntoh64 (meta_dst->dir_count);
+                        src_dir_count = ntoh64 (meta_src->dir_count);
+
+                        if (src_dir_count > dst_dir_count)
+                                meta_dst->dir_count = meta_src->dir_count;
+                } else {
+                        meta_dst->dir_count = 0;
                 }
+        } else {
+                size = data_to_bin (value);
+                meta_dst->size = hton64 (ntoh64 (meta_dst->size) +
+                                         ntoh64 (*size));
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
 
-                *size = hton64 (ntoh64 (*size) + ntoh64 (*ptr));
+int
+dht_aggregate (dict_t *this, char *key, data_t *value, void *data)
+{
+        dict_t          *dst            = NULL;
+        int32_t          ret            = -1;
+        data_t          *dict_data      = NULL;
+
+        dst = data;
 
+        if (strcmp (key, QUOTA_SIZE_KEY) == 0) {
+                ret = dht_aggregate_quota_xattr (dst, key, value);
+                if (ret) {
+                        gf_log ("dht", GF_LOG_WARNING, "Failed to "
+                                "aggregate qutoa xattr");
+                        goto out;
+                }
         } else if (fnmatch (GF_XATTR_STIME_PATTERN, key, FNM_NOESCAPE) == 0) {
                 ret = gf_get_min_stime (THIS, dst, key, value);
                 if (ret < 0)
-                        return ret;
+                        goto out;
         } else {
                 /* compare user xattrs only */
                 if (!strncmp (key, "user.", strlen ("user."))) {
@@ -85,14 +129,17 @@ dht_aggregate (dict_t *this, char *key, data_t *value, void *data)
                         }
                 }
                 ret = dict_set (dst, key, value);
-                if (ret)
+                if (ret) {
                         gf_msg ("dht", GF_LOG_WARNING, 0,
                                 DHT_MSG_DICT_SET_FAILED,
                                 "Failed to set dictionary value: key = %s",
                                 key);
+                }
         }
 
-        return 0;
+        ret = 0;
+out:
+        return ret;
 }
 
 
@@ -255,7 +302,6 @@ selfheal:
 
 }
 
-
 int
 dht_discover_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                   int op_ret, int op_errno,
@@ -2896,7 +2942,8 @@ dht_getxattr (call_frame_t *frame, xlator_t *this,
                 return 0;
         }
 
-        if (key && !strcmp (GF_XATTR_QUOTA_LIMIT_LIST, key)) {
+        if (key && (!strcmp (GF_XATTR_QUOTA_LIMIT_LIST, key) ||
+                    !strcmp (GF_XATTR_QUOTA_LIMIT_LIST_OBJECT, key))) {
                 /* quota hardlimit and aggregated size of a directory is stored
                  * in inode contexts of each brick. Hence its good enough that
                  * we send getxattr for this key to any brick.
diff --git a/xlators/features/marker/src/marker-quota.c b/xlators/features/marker/src/marker-quota.c
index cc75922b9f2..63aa4efa44d 100644
--- a/xlators/features/marker/src/marker-quota.c
+++ b/xlators/features/marker/src/marker-quota.c
@@ -21,6 +21,7 @@
 #include "marker-quota.h"
 #include "marker-quota-helper.h"
 #include "syncop.h"
+#include "quota-common-utils.h"
 
 int
 mq_loc_copy (loc_t *dst, loc_t *src)
@@ -2035,79 +2036,6 @@ err:
         return -1;
 }
 
-int32_t
-mq_dict_set_meta (dict_t *dict, char *key, const quota_meta_t *meta,
-                  ia_type_t ia_type)
-{
-        int32_t         ret      = -1;
-        quota_meta_t   *value    = NULL;
-
-        QUOTA_ALLOC_OR_GOTO (value, quota_meta_t, ret, out);
-
-        value->size = hton64 (meta->size);
-        value->file_count = hton64 (meta->file_count);
-        value->dir_count = hton64 (meta->dir_count);
-
-        if (ia_type == IA_IFDIR) {
-                ret = dict_set_bin (dict, key, value, sizeof (*value));
-        } else {
-                /* For a file we don't need to store dir_count in the
-                 * quota size xattr, so we set the len of the data in the dict
-                 * as 128bits, so when the posix xattrop reads the dict, it only
-                 * performs operations on size and file_count
-                 */
-                ret = dict_set_bin (dict, key, value,
-                                    sizeof (*value) - sizeof (int64_t));
-        }
-
-        if (ret < 0) {
-                gf_log_callingfn ("marker", GF_LOG_ERROR, "dict set failed");
-                GF_FREE (value);
-        }
-
-out:
-        return ret;
-}
-
-int32_t
-mq_dict_get_meta (dict_t *dict, char *key, quota_meta_t *meta)
-{
-        int32_t        ret      = -1;
-        data_t        *data     = NULL;
-        quota_meta_t  *value    = NULL;
-
-        if (!dict || !key || !meta)
-                goto out;
-
-        data = dict_get (dict, key);
-        if (!data || !data->data)
-                goto out;
-
-        if (data->len > sizeof (int64_t)) {
-                value = (quota_meta_t *) data->data;
-                meta->size = ntoh64 (value->size);
-                meta->file_count = ntoh64 (value->file_count);
-                if (data->len > (sizeof (int64_t)) * 2)
-                        meta->dir_count  = ntoh64 (value->dir_count);
-                else
-                        meta->dir_count = 0;
-        } else {
-                /* This can happen during software upgrade.
-                 * Older version of glusterfs will not have inode count.
-                 * Return failure, this will be healed as part of lookup
-                 */
-                gf_log_callingfn ("marker", GF_LOG_DEBUG, "Object quota xattrs "
-                                  "missing: len = %d", data->len);
-                ret = -1;
-                goto out;
-        }
-
-        ret = 0;
-out:
-
-        return ret;
-}
-
 void
 mq_compute_delta (quota_meta_t *delta, const quota_meta_t *op1,
                   const quota_meta_t *op2)
@@ -2187,7 +2115,7 @@ mq_are_xattrs_set (xlator_t *this, loc_t *loc, gf_boolean_t *result,
 
         *result = _gf_true;
         if (loc->inode->ia_type == IA_IFDIR) {
-                ret = mq_dict_get_meta (rsp_dict, QUOTA_SIZE_KEY, &meta);
+                ret = quota_dict_get_meta (rsp_dict, QUOTA_SIZE_KEY, &meta);
                 if (ret < 0 || meta.dir_count == 0) {
                         ret = 0;
                         *result = _gf_false;
@@ -2248,7 +2176,8 @@ mq_create_xattrs (xlator_t *this, loc_t *loc, gf_boolean_t objects)
                         /* Initial object count of a directory is 1 */
                         size.dir_count = 1;
                 }
-                ret = mq_dict_set_meta (dict, QUOTA_SIZE_KEY, &size, IA_IFDIR);
+                ret = quota_dict_set_meta (dict, QUOTA_SIZE_KEY, &size,
+                                           IA_IFDIR);
                 if (ret < 0)
                         goto out;
         }
@@ -2261,8 +2190,8 @@ mq_create_xattrs (xlator_t *this, loc_t *loc, gf_boolean_t objects)
                 }
 
                 GET_CONTRI_KEY (key, contribution->gfid, ret);
-                ret = mq_dict_set_meta (dict, key, &contri,
-                                        loc->inode->ia_type);
+                ret = quota_dict_set_meta (dict, key, &contri,
+                                           loc->inode->ia_type);
                 if (ret < 0)
                         goto out;
         }
@@ -2458,8 +2387,8 @@ _mq_get_metadata (xlator_t *this, loc_t *loc, quota_meta_t *contri,
 
         if (size) {
                 if (loc->inode->ia_type == IA_IFDIR) {
-                        ret = mq_dict_get_meta (rsp_dict, QUOTA_SIZE_KEY,
-                                                &meta);
+                        ret = quota_dict_get_meta (rsp_dict, QUOTA_SIZE_KEY,
+                                                   &meta);
                         if (ret < 0) {
                                 gf_log (this->name, GF_LOG_ERROR,
                                         "dict_get failed.");
@@ -2477,7 +2406,7 @@ _mq_get_metadata (xlator_t *this, loc_t *loc, quota_meta_t *contri,
         }
 
         if (contri && !loc_is_root(loc)) {
-                ret = mq_dict_get_meta (rsp_dict, contri_key, &meta);
+                ret = quota_dict_get_meta (rsp_dict, contri_key, &meta);
                 if (ret < 0) {
                         contri->size = 0;
                         contri->file_count = 0;
@@ -2679,7 +2608,8 @@ mq_update_contri (xlator_t *this, loc_t *loc, inode_contribution_t *contri,
                 goto out;
         }
 
-        ret = mq_dict_set_meta (dict, contri_key, delta, loc->inode->ia_type);
+        ret = quota_dict_set_meta (dict, contri_key, delta,
+                                   loc->inode->ia_type);
         if (ret < 0)
                 goto out;
 
@@ -2737,8 +2667,8 @@ mq_update_size (xlator_t *this, loc_t *loc, quota_meta_t *delta)
                 goto out;
         }
 
-        ret = mq_dict_set_meta (dict, QUOTA_SIZE_KEY, delta,
-                                loc->inode->ia_type);
+        ret = quota_dict_set_meta (dict, QUOTA_SIZE_KEY, delta,
+                                   loc->inode->ia_type);
         if (ret < 0)
                 goto out;
 
@@ -3484,7 +3414,7 @@ mq_inspect_directory_xattr_task (void *opaque)
         if (ret < 0)
                 goto out;
 
-        ret = mq_dict_get_meta (dict, QUOTA_SIZE_KEY, &size);
+        ret = quota_dict_get_meta (dict, QUOTA_SIZE_KEY, &size);
         if (ret < 0)
                 goto out;
 
@@ -3493,7 +3423,7 @@ mq_inspect_directory_xattr_task (void *opaque)
                 if (ret < 0)
                         goto err;
 
-                ret = mq_dict_get_meta (dict, contri_key, &contri);
+                ret = quota_dict_get_meta (dict, contri_key, &contri);
                 if (ret < 0)
                         goto out;
 
@@ -3609,7 +3539,7 @@ mq_inspect_file_xattr_task (void *opaque)
                 if (ret < 0)
                         continue;
 
-                ret = mq_dict_get_meta (dict, contri_key, &contri);
+                ret = quota_dict_get_meta (dict, contri_key, &contri);
                 if (ret < 0) {
                         ret = mq_create_xattrs_blocking_txn (this, loc);
                 } else {
diff --git a/xlators/features/marker/src/marker-quota.h b/xlators/features/marker/src/marker-quota.h
index fa132a815b7..4600954c6a2 100644
--- a/xlators/features/marker/src/marker-quota.h
+++ b/xlators/features/marker/src/marker-quota.h
@@ -93,13 +93,6 @@ struct quota_inode_ctx {
 };
 typedef struct quota_inode_ctx quota_inode_ctx_t;
 
-struct quota_meta {
-        int64_t    size;
-        int64_t    file_count;
-        int64_t    dir_count;
-};
-typedef struct quota_meta quota_meta_t;
-
 struct quota_synctask {
         xlator_t      *this;
         loc_t          loc;
diff --git a/xlators/features/marker/src/marker.c b/xlators/features/marker/src/marker.c
index af7ec1f907f..c1db70369a8 100644
--- a/xlators/features/marker/src/marker.c
+++ b/xlators/features/marker/src/marker.c
@@ -30,6 +30,7 @@
 static char   *quota_external_xattrs[] = {
         QUOTA_SIZE_KEY,
         QUOTA_LIMIT_KEY,
+        QUOTA_LIMIT_OBJECTS_KEY,
         NULL,
 };
 
diff --git a/xlators/features/quota/src/quota.c b/xlators/features/quota/src/quota.c
index 0e6ad6f8f30..5fcc65b7243 100644
--- a/xlators/features/quota/src/quota.c
+++ b/xlators/features/quota/src/quota.c
@@ -13,6 +13,7 @@
 #include "common-utils.h"
 #include "defaults.h"
 #include "statedump.h"
+#include "quota-common-utils.h"
 
 struct volume_options options[];
 
@@ -535,8 +536,10 @@ quota_validate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         quota_local_t     *local      = NULL;
         int32_t            ret        = 0;
         quota_inode_ctx_t *ctx        = NULL;
-        int64_t           *size       = 0;
+        int64_t           *object_size = 0;
         uint64_t           value      = 0;
+        data_t            *data       = NULL;
+        quota_meta_t       size       = {0,};
 
         local = frame->local;
 
@@ -562,12 +565,11 @@ quota_validate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto unwind;
         }
 
-        ret = dict_get_bin (xdata, QUOTA_SIZE_KEY, (void **) &size);
-        if (ret < 0) {
-                gf_log (this->name, GF_LOG_WARNING,
-                        "size key not present in dict");
+        ret = quota_dict_get_meta (xdata, QUOTA_SIZE_KEY, &size);
+        if (ret == -1) {
+                gf_log (this->name, GF_LOG_WARNING, "dict get failed "
+                        "on quota size");
                 op_errno = EINVAL;
-                goto unwind;
         }
 
         local->just_validated = 1; /* so that we don't go into infinite
@@ -576,7 +578,9 @@ quota_validate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                                     */
         LOCK (&ctx->lock);
         {
-                ctx->size = ntoh64 (*size);
+                ctx->size = size.size;
+                ctx->file_count = size.file_count;
+                ctx->dir_count = size.dir_count;
                 gettimeofday (&ctx->tv, NULL);
         }
         UNLOCK (&ctx->lock);
@@ -995,6 +999,163 @@ out:
 }
 
 int32_t
+quota_check_object_limit (call_frame_t *frame, quota_inode_ctx_t *ctx,
+                          quota_priv_t *priv, inode_t *_inode, xlator_t *this,
+                          int32_t *op_errno, int just_validated,
+                          quota_local_t *local, gf_boolean_t *skip_check)
+{
+        int32_t         ret                     = -1;
+        uint32_t        timeout                 =  0;
+        char            need_validate           =  0;
+        gf_boolean_t    hard_limit_exceeded     =  0;
+        int64_t         object_aggr_count       =  0;
+
+        GF_ASSERT (frame);
+        GF_ASSERT (priv);
+        GF_ASSERT (_inode);
+        GF_ASSERT (this);
+        GF_ASSERT (local);
+
+        if (ctx != NULL && (ctx->object_hard_lim > 0 ||
+                            ctx->object_soft_lim)) {
+                LOCK (&ctx->lock);
+                {
+                        timeout = priv->soft_timeout;
+
+                        object_aggr_count = ctx->file_count +
+                                            ctx->dir_count + 1;
+                        if (((ctx->object_soft_lim >= 0)
+                             && (object_aggr_count) >
+                             ctx->object_soft_lim)) {
+                                timeout = priv->hard_timeout;
+                        }
+
+                        if (!just_validated
+                            && quota_timeout (&ctx->tv, timeout)) {
+                                need_validate = 1;
+                        } else if ((object_aggr_count) >
+                                  ctx->object_hard_lim) {
+                                hard_limit_exceeded = 1;
+                        }
+                }
+                UNLOCK (&ctx->lock);
+
+                if (need_validate && *skip_check != _gf_true) {
+                        *skip_check = _gf_true;
+                        ret = quota_validate (frame, _inode, this,
+                                              quota_validate_cbk);
+                        if (ret < 0) {
+                                *op_errno = -ret;
+                                *skip_check = _gf_false;
+                        }
+                        goto out;
+                }
+
+                if (hard_limit_exceeded) {
+                        local->op_ret = -1;
+                        local->op_errno = EDQUOT;
+                        *op_errno = EDQUOT;
+                }
+
+                /*We log usage only if quota limit is configured on
+                   that inode
+                */
+                quota_log_usage (this, ctx, _inode, 0);
+        }
+
+        ret = 0;
+
+out:
+        return ret;
+}
+
+
+int32_t
+quota_check_size_limit (call_frame_t *frame, quota_inode_ctx_t *ctx,
+                          quota_priv_t *priv, inode_t *_inode, xlator_t *this,
+                          int32_t *op_errno, int just_validated, int64_t delta,
+                          quota_local_t *local, gf_boolean_t *skip_check)
+{
+        int32_t         ret                     = -1;
+        uint32_t        timeout                 =  0;
+        char            need_validate           =  0;
+        gf_boolean_t    hard_limit_exceeded     =  0;
+        int64_t         space_available         =  0;
+        int64_t         wouldbe_size            =  0;
+
+        GF_ASSERT (frame);
+        GF_ASSERT (priv);
+        GF_ASSERT (_inode);
+        GF_ASSERT (this);
+        GF_ASSERT (local);
+
+        if (ctx != NULL && (ctx->hard_lim > 0 || ctx->soft_lim > 0)) {
+                wouldbe_size = ctx->size + delta;
+
+                LOCK (&ctx->lock);
+                {
+                        timeout = priv->soft_timeout;
+
+                        if ((ctx->soft_lim >= 0)
+                            && (wouldbe_size > ctx->soft_lim)) {
+                                timeout = priv->hard_timeout;
+                        }
+
+                        if (!just_validated
+                            && quota_timeout (&ctx->tv, timeout)) {
+                                need_validate = 1;
+                        } else if (wouldbe_size >= ctx->hard_lim) {
+                                hard_limit_exceeded = 1;
+                        }
+                }
+                UNLOCK (&ctx->lock);
+
+                if (need_validate && *skip_check != _gf_true) {
+                        *skip_check = _gf_true;
+                        ret = quota_validate (frame, _inode, this,
+                                              quota_validate_cbk);
+                        if (ret < 0) {
+                                *op_errno = -ret;
+                                *skip_check = _gf_false;
+                        }
+                        goto out;
+                }
+
+                if (hard_limit_exceeded) {
+                        local->op_ret = -1;
+                        local->op_errno = EDQUOT;
+
+                        space_available = ctx->hard_lim - ctx->size;
+
+                        if (space_available < 0)
+                                space_available = 0;
+
+                        if ((local->space_available < 0)
+                            || (local->space_available
+                                > space_available)){
+                                local->space_available
+                                        = space_available;
+
+                        }
+
+                        if (space_available == 0) {
+                                *op_errno = EDQUOT;
+                                goto out;
+                        }
+                }
+
+                /* We log usage only if quota limit is configured on
+                   that inode. */
+                quota_log_usage (this, ctx, _inode, delta);
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
+
+
+int32_t
 quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
                    char *name, uuid_t par)
 {
@@ -1004,13 +1165,12 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
         quota_priv_t      *priv                = NULL;
         quota_local_t     *local               = NULL;
         char               need_validate       = 0;
+        char               just_validated      = 0;
         gf_boolean_t       hard_limit_exceeded = 0;
-        int64_t            delta               = 0, wouldbe_size = 0;
-        int64_t            space_available     = 0;
+        int64_t            delta               = 0;
         uint64_t           value               = 0;
-        char               just_validated      = 0;
         uuid_t             trav_uuid           = {0,};
-        uint32_t           timeout             = 0;
+        gf_boolean_t       skip_check          = _gf_false;
 
         GF_VALIDATE_OR_GOTO ("quota", this, err);
         GF_VALIDATE_OR_GOTO (this->name, frame, err);
@@ -1063,64 +1223,28 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
                         break;
                 }
 
-                if (ctx != NULL && (ctx->hard_lim > 0 || ctx->soft_lim > 0)) {
-                        wouldbe_size = ctx->size + delta;
-
-                        LOCK (&ctx->lock);
-                        {
-                                timeout = priv->soft_timeout;
-
-                                if ((ctx->soft_lim >= 0)
-                                    && (wouldbe_size > ctx->soft_lim)) {
-                                        timeout = priv->hard_timeout;
-                                }
-
-                                if (!just_validated
-                                    && quota_timeout (&ctx->tv, timeout)) {
-                                        need_validate = 1;
-                                } else if (wouldbe_size >= ctx->hard_lim) {
-                                        hard_limit_exceeded = 1;
-                                }
-                        }
-                        UNLOCK (&ctx->lock);
-
-                        if (need_validate) {
-                                ret = quota_validate (frame, _inode, this,
-                                                      quota_validate_cbk);
-                                if (ret < 0) {
-                                        op_errno = -ret;
-                                        goto err;
-                                }
-
-                                break;
-                        }
-
-                        if (hard_limit_exceeded) {
-                                local->op_ret = -1;
-                                local->op_errno = EDQUOT;
-
-                                space_available = ctx->hard_lim - ctx->size;
+                ret = quota_check_object_limit (frame, ctx, priv, _inode, this,
+                                                &op_errno, just_validated,
+                                                local, &skip_check);
+                if (skip_check == _gf_true)
+                        goto done;
 
-                                if (space_available < 0)
-                                        space_available = 0;
-
-                                if ((local->space_available < 0)
-                                    || (local->space_available
-                                        > space_available)){
-                                        local->space_available
-                                                = space_available;
-
-                                }
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "Failed to check "
+                                "quota object limit");
+                        goto err;
+                }
 
-                                if (space_available == 0) {
-                                        op_errno = EDQUOT;
-                                        goto err;
-                                }
-                        }
+                ret = quota_check_size_limit (frame, ctx, priv, _inode, this,
+                                              &op_errno, just_validated, delta,
+                                              local, &skip_check);
+                if (skip_check == _gf_true)
+                        goto done;
 
-                        /* We log usage only if quota limit is configured on
-                           that inode. */
-                        quota_log_usage (this, ctx, _inode, delta);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "Failed to check "
+                                "quota size limit");
+                        goto err;
                 }
 
                 if (__is_root_gfid (_inode->gfid)) {
@@ -1160,12 +1284,12 @@ quota_check_limit (call_frame_t *frame, inode_t *inode, xlator_t *this,
                 ctx = (quota_inode_ctx_t *)(unsigned long)value;
         } while (1);
 
+
+done:
         if (_inode != NULL) {
                 inode_unref (_inode);
                 _inode = NULL;
         }
-
-done:
         return 0;
 
 err:
@@ -1177,12 +1301,15 @@ err:
 
 static inline int
 quota_get_limits (xlator_t *this, dict_t *dict, int64_t *hard_lim,
-                  int64_t *soft_lim)
+                  int64_t *soft_lim, int64_t *object_hard_limit,
+                  int64_t *object_soft_limit)
 {
-        quota_limit_t *limit            = NULL;
-        quota_priv_t  *priv             = NULL;
-        int64_t        soft_lim_percent = 0, *ptr = NULL;
-        int            ret              = 0;
+        quota_limits_t *limit             = NULL;
+        quota_limits_t *object_limit      = NULL;
+        quota_priv_t   *priv              = NULL;
+        int64_t         soft_lim_percent  = 0;
+        int64_t        *ptr               = NULL;
+        int             ret               = 0;
 
         if ((this == NULL) || (dict == NULL) || (hard_lim == NULL)
             || (soft_lim == NULL))
@@ -1191,11 +1318,11 @@ quota_get_limits (xlator_t *this, dict_t *dict, int64_t *hard_lim,
         priv = this->private;
 
         ret = dict_get_bin (dict, QUOTA_LIMIT_KEY, (void **) &ptr);
-        limit = (quota_limit_t *)ptr;
+        limit = (quota_limits_t *)ptr;
 
         if (limit) {
-                *hard_lim = ntoh64 (limit->hard_lim);
-                soft_lim_percent = ntoh64 (limit->soft_lim_percent);
+                *hard_lim = ntoh64 (limit->hl);
+                soft_lim_percent = ntoh64 (limit->sl);
         }
 
         if (soft_lim_percent < 0) {
@@ -1206,6 +1333,25 @@ quota_get_limits (xlator_t *this, dict_t *dict, int64_t *hard_lim,
                 *soft_lim = (soft_lim_percent * (*hard_lim))/100;
         }
 
+        ret = dict_get_bin (dict, QUOTA_LIMIT_OBJECTS_KEY, (void **) &ptr);
+        if (ret)
+                return 0;
+        object_limit = (quota_limits_t *)ptr;
+
+        if (object_limit) {
+                *object_hard_limit = ntoh64 (object_limit->hl);
+                 soft_lim_percent = ntoh64 (object_limit->sl);
+        }
+
+        if (soft_lim_percent < 0) {
+                soft_lim_percent = priv->default_soft_lim;
+        }
+
+        if ((*object_hard_limit > 0) && (soft_lim_percent > 0)) {
+                *object_soft_limit = (soft_lim_percent *
+                                     (*object_hard_limit))/100;
+        }
+
 out:
         return 0;
 }
@@ -1214,14 +1360,18 @@ int
 quota_fill_inodectx (xlator_t *this, inode_t *inode, dict_t *dict,
                      loc_t *loc, struct iatt *buf, int32_t *op_errno)
 {
-        int32_t            ret      = -1;
-        char               found    = 0;
-        quota_inode_ctx_t *ctx      = NULL;
-        quota_dentry_t    *dentry   = NULL;
-        uint64_t           value    = 0;
-        int64_t            hard_lim = -1, soft_lim = -1;
-
-        quota_get_limits (this, dict, &hard_lim, &soft_lim);
+        int32_t            ret                  = -1;
+        char               found                = 0;
+        quota_inode_ctx_t *ctx                  = NULL;
+        quota_dentry_t    *dentry               = NULL;
+        uint64_t           value                = 0;
+        int64_t            hard_lim             = 0;
+        int64_t            soft_lim             = 0;
+        int64_t            object_hard_limit    = 0;
+        int64_t            object_soft_limit    = 0;
+
+        quota_get_limits (this, dict, &hard_lim, &soft_lim, &object_hard_limit,
+                          &object_soft_limit);
 
         inode_ctx_get (inode, this, &value);
         ctx = (quota_inode_ctx_t *)(unsigned long)value;
@@ -1246,6 +1396,8 @@ quota_fill_inodectx (xlator_t *this, inode_t *inode, dict_t *dict,
         {
                 ctx->hard_lim = hard_lim;
                 ctx->soft_lim = soft_lim;
+                ctx->object_hard_lim = object_hard_limit;
+                ctx->object_soft_lim = object_soft_limit;
 
                 ctx->buf = *buf;
 
@@ -1360,6 +1512,13 @@ quota_lookup (call_frame_t *frame, xlator_t *this, loc_t *loc,
                 goto err;
         }
 
+        ret = dict_set_int8 (xattr_req, QUOTA_LIMIT_OBJECTS_KEY, 1);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "dict set of key for quota object limit failed");
+                goto err;
+        }
+
         STACK_WIND (frame, quota_lookup_cbk, FIRST_CHILD(this),
                     FIRST_CHILD(this)->fops->lookup, loc, xattr_req);
 
@@ -3671,20 +3830,26 @@ quota_setxattr_cbk (call_frame_t *frame, void *cookie,
         quota_inode_ctx_t *ctx   = NULL;
         int                ret   = 0;
 
+        if (op_ret < 0) {
+                goto out;
+        }
+
         local = frame->local;
         if (!local)
                 goto out;
 
         ret = quota_inode_ctx_get (local->loc.inode, this, &ctx, 1);
         if ((ret < 0) || (ctx == NULL)) {
-                op_errno = ENOMEM;
+                op_errno = -1;
                 goto out;
         }
 
         LOCK (&ctx->lock);
         {
-                ctx->hard_lim = local->limit.hard_lim;
-                ctx->soft_lim = local->limit.soft_lim_percent;
+                ctx->hard_lim = local->limit.hl;
+                ctx->soft_lim = local->limit.sl;
+                ctx->object_hard_lim = local->object_limit.hl;
+                ctx->object_soft_lim = local->object_limit.sl;
         }
         UNLOCK (&ctx->lock);
 
@@ -3697,11 +3862,14 @@ int
 quota_setxattr (call_frame_t *frame, xlator_t *this,
                 loc_t *loc, dict_t *dict, int flags, dict_t *xdata)
 {
-        quota_priv_t  *priv     = NULL;
-        int            op_errno = EINVAL;
-        int            op_ret   = -1;
-        int64_t        hard_lim = -1, soft_lim = -1;
-        quota_local_t *local    = NULL;
+        quota_priv_t    *priv                   = NULL;
+        int             op_errno                = EINVAL;
+        int             op_ret                  = -1;
+        int64_t         hard_lim                = -1;
+        int64_t         soft_lim                = -1;
+        int64_t         object_hard_limit       = -1;
+        int64_t         object_soft_limit       = -1;
+        quota_local_t   *local                  = NULL;
 
         priv = this->private;
 
@@ -3718,20 +3886,27 @@ quota_setxattr (call_frame_t *frame, xlator_t *this,
                                            err);
         }
 
-        quota_get_limits (this, dict, &hard_lim, &soft_lim);
+        quota_get_limits (this, dict, &hard_lim, &soft_lim, &object_hard_limit,
+                          &object_soft_limit);
 
-        if (hard_lim > 0) {
+        if (hard_lim > 0 || object_hard_limit > 0) {
                 local = quota_local_new ();
                 if (local == NULL) {
                         op_errno = ENOMEM;
                         goto err;
                 }
-
                 frame->local = local;
                 loc_copy (&local->loc, loc);
+        }
 
-                local->limit.hard_lim = hard_lim;
-                local->limit.soft_lim_percent = soft_lim;
+        if (hard_lim > 0) {
+                local->limit.hl = hard_lim;
+                local->limit.sl = soft_lim;
+        }
+
+        if (object_hard_limit > 0) {
+                local->object_limit.hl = object_hard_limit;
+                local->object_limit.sl = object_soft_limit;
         }
 
         STACK_WIND (frame, quota_setxattr_cbk,
@@ -3756,6 +3931,9 @@ quota_fsetxattr_cbk (call_frame_t *frame, void *cookie,
         quota_inode_ctx_t *ctx   = NULL;
         quota_local_t     *local = NULL;
 
+        if (op_ret < 0)
+                goto out;
+
         local = frame->local;
         if (!local)
                 goto out;
@@ -3768,8 +3946,10 @@ quota_fsetxattr_cbk (call_frame_t *frame, void *cookie,
 
         LOCK (&ctx->lock);
         {
-                ctx->hard_lim = local->limit.hard_lim;
-                ctx->soft_lim = local->limit.soft_lim_percent;
+                ctx->hard_lim = local->limit.hl;
+                ctx->soft_lim = local->limit.sl;
+                ctx->object_hard_lim = local->object_limit.hl;
+                ctx->object_soft_lim = local->object_limit.sl;
         }
         UNLOCK (&ctx->lock);
 
@@ -3782,11 +3962,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;
-        quota_local_t *local    = NULL;
-        int64_t        hard_lim = -1, soft_lim = -1;
+        quota_priv_t    *priv                   = NULL;
+        int32_t         op_ret                  = -1;
+        int32_t         op_errno                = EINVAL;
+        quota_local_t   *local                  = NULL;
+        int64_t         hard_lim                = -1;
+        int64_t         soft_lim                = -1;
+        int64_t         object_hard_limit       = -1;
+        int64_t         object_soft_limit       = -1;
 
         priv = this->private;
 
@@ -3803,9 +3986,10 @@ quota_fsetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
                                             op_errno, err);
         }
 
-        quota_get_limits (this, dict, &hard_lim, &soft_lim);
+        quota_get_limits (this, dict, &hard_lim, &soft_lim, &object_hard_limit,
+                          &object_soft_limit);
 
-        if (hard_lim > 0) {
+        if (hard_lim > 0 || object_hard_limit > 0) {
                 local = quota_local_new ();
                 if (local == NULL) {
                         op_errno = ENOMEM;
@@ -3813,9 +3997,16 @@ quota_fsetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
                 }
                 frame->local = local;
                 local->loc.inode = inode_ref (fd->inode);
+        }
+
+        if (hard_lim > 0) {
+                local->limit.hl = hard_lim;
+                local->limit.sl = soft_lim;
+        }
 
-                local->limit.hard_lim = hard_lim;
-                local->limit.soft_lim_percent = soft_lim;
+        if (object_hard_limit > 0) {
+                local->object_limit.hl = object_hard_limit;
+                local->object_limit.sl = object_soft_limit;
         }
 
         STACK_WIND (frame, quota_fsetxattr_cbk,
@@ -4040,8 +4231,9 @@ quota_statfs_validate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         quota_local_t     *local      = NULL;
         int32_t            ret        = 0;
         quota_inode_ctx_t *ctx        = NULL;
-        int64_t           *size       = 0;
         uint64_t           value      = 0;
+        data_t            *data       = NULL;
+        quota_meta_t       size       = {0,};
 
         local = frame->local;
 
@@ -4066,17 +4258,18 @@ quota_statfs_validate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto resume;
         }
 
-        ret = dict_get_bin (xdata, QUOTA_SIZE_KEY, (void **) &size);
-        if (ret < 0) {
-                gf_log (this->name, GF_LOG_WARNING,
-                        "size key not present in dict");
+        ret = quota_dict_get_meta (xdata, QUOTA_SIZE_KEY, &size);
+        if (ret == -1) {
+                gf_log (this->name, GF_LOG_WARNING, "dict get failed "
+                        "on quota size");
                 op_errno = EINVAL;
-                goto resume;
         }
 
         LOCK (&ctx->lock);
         {
-                ctx->size = ntoh64 (*size);
+                ctx->size = size.size;
+                ctx->file_count = size.file_count;
+                ctx->dir_count = size.dir_count;
                 gettimeofday (&ctx->tv, NULL);
         }
         UNLOCK (&ctx->lock);
@@ -4359,6 +4552,15 @@ quota_readdirp (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
                 }
         }
 
+        if (dict) {
+                ret = dict_set_int8 (dict, QUOTA_LIMIT_OBJECTS_KEY, 1);
+                if (ret < 0) {
+                        gf_log (this->name, GF_LOG_WARNING,
+                                "dict set of key for hard-limit failed");
+                        goto err;
+                }
+        }
+
         STACK_WIND (frame, quota_readdirp_cbk,
                     FIRST_CHILD(this), FIRST_CHILD(this)->fops->readdirp, fd,
                     size, offset, dict);
diff --git a/xlators/features/quota/src/quota.h b/xlators/features/quota/src/quota.h
index ac08a040d8a..c61c974e352 100644
--- a/xlators/features/quota/src/quota.h
+++ b/xlators/features/quota/src/quota.h
@@ -25,7 +25,6 @@
 #include "logging.h"
 #include "dict.h"
 #include "stack.h"
-#include "common-utils.h"
 #include "event.h"
 #include "globals.h"
 #include "rpcsvc.h"
@@ -36,6 +35,7 @@
 #include "xdr-generic.h"
 #include "compat-errno.h"
 #include "protocol-common.h"
+#include "quota-common-utils.h"
 
 #define DIRTY                   "dirty"
 #define SIZE                    "size"
@@ -169,6 +169,10 @@ struct quota_inode_ctx {
         int64_t          size;
         int64_t          hard_lim;
         int64_t          soft_lim;
+        int64_t          file_count;
+        int64_t          dir_count;
+        int64_t          object_hard_lim;
+        int64_t          object_soft_lim;
         struct iatt      buf;
         struct list_head parents;
         struct timeval   tv;
@@ -178,12 +182,6 @@ struct quota_inode_ctx {
 };
 typedef struct quota_inode_ctx quota_inode_ctx_t;
 
-struct quota_limit {
-        int64_t hard_lim;
-        int64_t soft_lim_percent;
-} __attribute__ ((packed));
-typedef struct quota_limit quota_limit_t;
-
 typedef void
 (*quota_ancestry_built_t) (struct list_head *parents, inode_t *inode,
                            int32_t op_ret, int32_t op_errno, void *data);
@@ -210,7 +208,8 @@ struct quota_local {
         uuid_t                  common_ancestor; /* Used by quota_rename */
         call_stub_t            *stub;
         struct iobref          *iobref;
-        quota_limit_t           limit;
+        quota_limits_t          limit;
+        quota_limits_t          object_limit;
         int64_t                 space_available;
         quota_ancestry_built_t  ancestry_cbk;
         void                   *ancestry_data;
@@ -261,4 +260,15 @@ int
 quota_fill_inodectx (xlator_t *this, inode_t *inode, dict_t *dict,
                      loc_t *loc, struct iatt *buf, int32_t *op_errno);
 
+int32_t
+quota_check_size_limit (call_frame_t *frame, quota_inode_ctx_t *ctx,
+                          quota_priv_t *priv, inode_t *_inode, xlator_t *this,
+                          int32_t *op_errno, int just_validated, int64_t delta,
+                          quota_local_t *local, gf_boolean_t *skip_check);
+
+int32_t
+quota_check_object_limit (call_frame_t *frame, quota_inode_ctx_t *ctx,
+                          quota_priv_t *priv, inode_t *_inode, xlator_t *this,
+                          int32_t *op_errno, int just_validated,
+                          quota_local_t *local, gf_boolean_t *skip_check);
 #endif
diff --git a/xlators/features/quota/src/quotad-aggregator.c b/xlators/features/quota/src/quotad-aggregator.c
index f34bdbddd45..0abe4e6fc80 100644
--- a/xlators/features/quota/src/quotad-aggregator.c
+++ b/xlators/features/quota/src/quotad-aggregator.c
@@ -126,16 +126,29 @@ int
 quotad_aggregator_getlimit_cbk (xlator_t *this, call_frame_t *frame,
                                 void *lookup_rsp)
 {
-        gfs3_lookup_rsp            *rsp   = lookup_rsp;
+        gfs3_lookup_rsp            *rsp    = lookup_rsp;
         gf_cli_rsp                 cli_rsp = {0,};
-        dict_t                     *xdata = NULL;
-        int                         ret = -1;
+        dict_t                     *xdata  = NULL;
+        quotad_aggregator_state_t  *state  = NULL;
+        int                         ret    = -1;
+        int                         type   = 0;
 
         GF_PROTOCOL_DICT_UNSERIALIZE (frame->this, xdata,
                                       (rsp->xdata.xdata_val),
                                       (rsp->xdata.xdata_len), rsp->op_ret,
                                       rsp->op_errno, out);
 
+        if (xdata) {
+                state = frame->root->state;
+                ret = dict_get_int32 (state->xdata, "type", &type);
+                if (ret < 0)
+                        goto out;
+
+                ret = dict_set_int32 (xdata, "type", type);
+                if (ret < 0)
+                        goto out;
+        }
+
         ret = 0;
 out:
         rsp->op_ret = ret;
@@ -215,10 +228,18 @@ quotad_aggregator_getlimit (rpcsvc_request_t *req)
         }
         state = frame->root->state;
         state->xdata = dict;
+
         ret = dict_set_int32 (state->xdata, QUOTA_LIMIT_KEY, 42);
         if (ret)
                 goto err;
 
+        ret = dict_set_int32 (state->xdata, QUOTA_LIMIT_OBJECTS_KEY, 42);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to set "
+                        "QUOTA_LIMIT_OBJECTS_KEY");
+                goto err;
+        }
+
         ret = dict_set_int32 (state->xdata, QUOTA_SIZE_KEY, 42);
         if (ret)
                 goto err;
diff --git a/xlators/lib/src/libxlator.h b/xlators/lib/src/libxlator.h
index 175d3141d45..404124ca7d3 100644
--- a/xlators/lib/src/libxlator.h
+++ b/xlators/lib/src/libxlator.h
@@ -33,6 +33,7 @@
 #define MARKER_XTIME_TYPE   2
 #define GF_XATTR_QUOTA_SIZE_KEY "trusted.glusterfs.quota.size"
 #define GF_XATTR_QUOTA_LIMIT_LIST "trusted.limit.list"
+#define GF_XATTR_QUOTA_LIMIT_LIST_OBJECT "trusted.limit.objects"
 
 
 typedef int32_t (*xlator_specf_unwind_t) (call_frame_t *frame,
diff --git a/xlators/mgmt/glusterd/src/glusterd-quota.c b/xlators/mgmt/glusterd/src/glusterd-quota.c
index 462064819ac..f2a73057414 100644
--- a/xlators/mgmt/glusterd/src/glusterd-quota.c
+++ b/xlators/mgmt/glusterd/src/glusterd-quota.c
@@ -26,6 +26,7 @@
 #include "syscall.h"
 #include "byte-order.h"
 #include "compat-errno.h"
+#include "quota-common-utils.h"
 
 #include <sys/wait.h>
 #include <dlfcn.h>
@@ -42,7 +43,7 @@
 /* Any negative pid to make it special client */
 #define QUOTA_CRAWL_PID "-100"
 
-const char *gd_quota_op_list[GF_QUOTA_OPTION_TYPE_DEFAULT_SOFT_LIMIT+1] = {
+const char *gd_quota_op_list[GF_QUOTA_OPTION_TYPE_MAX + 1] = {
         [GF_QUOTA_OPTION_TYPE_NONE]               = "none",
         [GF_QUOTA_OPTION_TYPE_ENABLE]             = "enable",
         [GF_QUOTA_OPTION_TYPE_DISABLE]            = "disable",
@@ -54,6 +55,10 @@ const char *gd_quota_op_list[GF_QUOTA_OPTION_TYPE_DEFAULT_SOFT_LIMIT+1] = {
         [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",
+        [GF_QUOTA_OPTION_TYPE_LIMIT_OBJECTS]      = "limit-objects",
+        [GF_QUOTA_OPTION_TYPE_LIST_OBJECTS]       = "list-objects",
+        [GF_QUOTA_OPTION_TYPE_REMOVE_OBJECTS]     = "remove-objetcs",
+        [GF_QUOTA_OPTION_TYPE_MAX]                = NULL
 };
 
 int
@@ -438,24 +443,17 @@ out:
         return ret;
 }
 
-
 static int
 glusterd_set_quota_limit (char *volname, char *path, char *hard_limit,
-                          char *soft_limit, char **op_errstr)
+                          char *soft_limit, char *key, char **op_errstr)
 {
         int               ret                = -1;
         xlator_t         *this               = NULL;
         char              abspath[PATH_MAX]  = {0,};
         glusterd_conf_t  *priv               = NULL;
-        double           soft_lim            = 0;
-
-        typedef struct quota_limits {
-                int64_t hl;
-                int64_t sl;
-        } __attribute__ ((__packed__)) quota_limits_t;
-
-	quota_limits_t existing_limit = {0,};
-	quota_limits_t new_limit = {0,};
+	quota_limits_t    existing_limit     = {0,};
+	quota_limits_t    new_limit          = {0,};
+        double            soft_limit_double  = 0;
 
         this = THIS;
         GF_ASSERT (this);
@@ -471,9 +469,7 @@ glusterd_set_quota_limit (char *volname, char *path, char *hard_limit,
         }
 
         if (!soft_limit) {
-                ret = sys_lgetxattr (abspath,
-                                     "trusted.glusterfs.quota.limit-set",
-                                     (void *)&existing_limit,
+                ret = sys_lgetxattr (abspath, key, (void *)&existing_limit,
                                      sizeof (existing_limit));
                 if (ret < 0) {
                         switch (errno) {
@@ -484,10 +480,9 @@ glusterd_set_quota_limit (char *volname, char *path, char *hard_limit,
                                 existing_limit.sl = -1;
                             break;
                         default:
-                                gf_asprintf (op_errstr, "Failed to get the xattr "
-                                             "'trusted.glusterfs.quota.limit-set' from "
-                                             "%s. Reason : %s", abspath,
-                                             strerror (errno));
+                                gf_asprintf (op_errstr, "Failed to get the "
+                                             "xattr %s from %s. Reason : %s",
+                                             key, abspath, strerror (errno));
                                 goto out;
                         }
                 } else {
@@ -497,10 +492,10 @@ glusterd_set_quota_limit (char *volname, char *path, char *hard_limit,
                 new_limit.sl = existing_limit.sl;
 
         } else {
-                ret = gf_string2percent (soft_limit, &soft_lim);
+                ret = gf_string2percent (soft_limit, &soft_limit_double);
                 if (ret)
                         goto out;
-                new_limit.sl = soft_lim;
+                new_limit.sl = soft_limit_double;
         }
 
         new_limit.sl = hton64 (new_limit.sl);
@@ -511,12 +506,11 @@ glusterd_set_quota_limit (char *volname, char *path, char *hard_limit,
 
         new_limit.hl = hton64 (new_limit.hl);
 
-        ret = sys_lsetxattr (abspath, "trusted.glusterfs.quota.limit-set",
-                             (char *)(void *)&new_limit, sizeof (new_limit), 0);
+        ret = sys_lsetxattr (abspath, key, (char *)(void *)&new_limit,
+                             sizeof (new_limit), 0);
         if (ret == -1) {
-                gf_asprintf (op_errstr, "setxattr of "
-                             "'trusted.glusterfs.quota.limit-set' failed on %s."
-                             " Reason : %s", abspath, strerror (errno));
+                gf_asprintf (op_errstr, "setxattr of %s failed on %s."
+                             " Reason : %s", key, abspath, strerror (errno));
                 goto out;
         }
         ret = 0;
@@ -728,44 +722,46 @@ glusterd_store_quota_config (glusterd_volinfo_t *volinfo, char *path,
         }
 
         switch (opcode) {
-                case GF_QUOTA_OPTION_TYPE_LIMIT_USAGE:
-                        if (!found) {
-                                ret = write (fd, gfid, 16);
-                                if (ret == -1) {
-                                        gf_log (this->name, GF_LOG_ERROR,
-                                                "write into quota.conf failed. "
-                                                "Reason : %s",
-                                                strerror (errno));
-                                        goto out;
-                                }
-                                modified = _gf_true;
+        case GF_QUOTA_OPTION_TYPE_LIMIT_USAGE:
+        case GF_QUOTA_OPTION_TYPE_LIMIT_OBJECTS:
+                if (!found) {
+                        ret = write (fd, gfid, 16);
+                        if (ret == -1) {
+                                gf_log (this->name, GF_LOG_ERROR,
+                                        "write into quota.conf failed. "
+                                        "Reason : %s",
+                                        strerror (errno));
+                                goto out;
                         }
-                        break;
+                        modified = _gf_true;
+                }
+                break;
 
-                case GF_QUOTA_OPTION_TYPE_REMOVE:
-                        if (is_file_empty) {
-                                gf_asprintf (op_errstr, "Cannot remove limit on"
-                                             " %s. The quota configuration file"
-                                             " for volume %s is empty.", path,
-                                             volinfo->volname);
+        case GF_QUOTA_OPTION_TYPE_REMOVE:
+        case GF_QUOTA_OPTION_TYPE_REMOVE_OBJECTS:
+                if (is_file_empty) {
+                        gf_asprintf (op_errstr, "Cannot remove limit on"
+                                     " %s. The quota configuration file"
+                                     " for volume %s is empty.", path,
+                                     volinfo->volname);
+                        ret = -1;
+                        goto out;
+                } else {
+                        if (!found) {
+                                gf_asprintf (op_errstr, "Error. gfid %s"
+                                             " for path %s not found in"
+                                             " store", gfid_str, path);
                                 ret = -1;
                                 goto out;
                         } else {
-                                if (!found) {
-                                        gf_asprintf (op_errstr, "Error. gfid %s"
-                                                     " for path %s not found in"
-                                                     " store", gfid_str, path);
-                                        ret = -1;
-                                        goto out;
-                                } else {
-                                        modified = _gf_true;
-                                }
+                                modified = _gf_true;
                         }
-                        break;
+                }
+                break;
 
-                default:
-                        ret = 0;
-                        break;
+        default:
+                ret = 0;
+                break;
         }
 
         if (modified)
@@ -850,9 +846,17 @@ glusterd_quota_limit_usage (glusterd_volinfo_t *volinfo, dict_t *dict,
         }
 
         if (is_origin_glusterd (dict)) {
-                ret = glusterd_set_quota_limit (volinfo->volname, path,
-                                                hard_limit, soft_limit,
-                                                op_errstr);
+                if (opcode == GF_QUOTA_OPTION_TYPE_LIMIT_USAGE) {
+                        ret = glusterd_set_quota_limit (volinfo->volname, path,
+                                                        hard_limit, soft_limit,
+                                                        QUOTA_LIMIT_KEY,
+                                                        op_errstr);
+                } else {
+                        ret = glusterd_set_quota_limit (volinfo->volname, path,
+                                                        hard_limit, soft_limit,
+                                                        QUOTA_LIMIT_OBJECTS_KEY,
+                                                        op_errstr);
+                }
                 if (ret)
                         goto out;
         }
@@ -879,7 +883,8 @@ out:
 }
 
 static int
-glusterd_remove_quota_limit (char *volname, char *path, char **op_errstr)
+glusterd_remove_quota_limit (char *volname, char *path, char **op_errstr,
+                             int type)
 {
         int               ret                = -1;
         xlator_t         *this               = NULL;
@@ -899,11 +904,24 @@ glusterd_remove_quota_limit (char *volname, char *path, char **op_errstr)
                 goto out;
         }
 
-        ret = sys_lremovexattr (abspath, "trusted.glusterfs.quota.limit-set");
-        if (ret) {
-                gf_asprintf (op_errstr, "removexattr failed on %s. Reason : %s",
-                             abspath, strerror (errno));
-                goto out;
+        if (type == GF_QUOTA_OPTION_TYPE_REMOVE) {
+                ret = sys_lremovexattr (abspath,
+                                        "trusted.glusterfs.quota.limit-set");
+                if (ret) {
+                        gf_asprintf (op_errstr, "removexattr failed on %s. "
+                                     "Reason : %s", abspath, strerror (errno));
+                        goto out;
+                }
+        }
+
+        if (type == GF_QUOTA_OPTION_TYPE_REMOVE_OBJECTS) {
+                ret = sys_lremovexattr (abspath,
+                                "trusted.glusterfs.quota.limit-objects");
+                if (ret) {
+                        gf_asprintf (op_errstr, "removexattr failed on %s. "
+                                     "Reason : %s", abspath, strerror (errno));
+                        goto out;
+                }
         }
         ret = 0;
 
@@ -913,7 +931,7 @@ out:
 
 int32_t
 glusterd_quota_remove_limits (glusterd_volinfo_t *volinfo, dict_t *dict,
-                              int opcode, char **op_errstr)
+                              int opcode, char **op_errstr, int type)
 {
         int32_t         ret                   = -1;
         char            *path                 = NULL;
@@ -946,7 +964,7 @@ glusterd_quota_remove_limits (glusterd_volinfo_t *volinfo, dict_t *dict,
 
         if (is_origin_glusterd (dict)) {
                 ret = glusterd_remove_quota_limit (volinfo->volname, path,
-                                                   op_errstr);
+                                                   op_errstr, type);
                 if (ret)
                         goto out;
         }
@@ -1100,16 +1118,19 @@ glusterd_op_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
                         break;
 
                 case GF_QUOTA_OPTION_TYPE_LIMIT_USAGE:
+                case GF_QUOTA_OPTION_TYPE_LIMIT_OBJECTS:
                         ret = glusterd_quota_limit_usage (volinfo, dict, type,
                                                           op_errstr);
                         goto out;
 
                 case GF_QUOTA_OPTION_TYPE_REMOVE:
+                case GF_QUOTA_OPTION_TYPE_REMOVE_OBJECTS:
                         ret = glusterd_quota_remove_limits (volinfo, dict, type,
-                                                            op_errstr);
+                                                            op_errstr, type);
                         goto out;
 
                 case GF_QUOTA_OPTION_TYPE_LIST:
+                case GF_QUOTA_OPTION_TYPE_LIST_OBJECTS:
                         ret = glusterd_check_if_quota_trans_enabled (volinfo);
                         if (ret == -1) {
                                 *op_errstr = gf_strdup ("Cannot list limits, "
@@ -1349,6 +1370,7 @@ glusterd_op_stage_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
         glusterd_volinfo_t *volinfo        = NULL;
         char               *hard_limit_str = NULL;
         uint64_t           hard_limit      = 0;
+        gf_boolean_t       get_gfid        = _gf_false;
 
         this = THIS;
         GF_ASSERT (this);
@@ -1420,6 +1442,7 @@ glusterd_op_stage_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
         switch (type) {
         case GF_QUOTA_OPTION_TYPE_ENABLE:
         case GF_QUOTA_OPTION_TYPE_LIST:
+        case GF_QUOTA_OPTION_TYPE_LIST_OBJECTS:
                 /* Fuse mount req. only for enable & list-usage options*/
                 if (is_origin_glusterd (dict) &&
                     !glusterd_is_fuse_available ()) {
@@ -1452,15 +1475,15 @@ glusterd_op_stage_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
                                 "greater than INT64_MAX", hard_limit_str);
                         goto out;
                 }
-                /*The break statement is missing here to allow intentional fall
-                 * through of code execution to the next switch case
-                 */
+                get_gfid = _gf_true;
+                break;
+        case GF_QUOTA_OPTION_TYPE_LIMIT_OBJECTS:
+                get_gfid = _gf_true;
+                break;
 
         case GF_QUOTA_OPTION_TYPE_REMOVE:
-                ret = glusterd_get_gfid_from_brick (dict, volinfo, rsp_dict,
-                                                    op_errstr);
-                if (ret)
-                        goto out;
+        case GF_QUOTA_OPTION_TYPE_REMOVE_OBJECTS:
+                get_gfid = _gf_true;
                 break;
 
         case GF_QUOTA_OPTION_TYPE_SOFT_TIMEOUT:
@@ -1476,6 +1499,13 @@ glusterd_op_stage_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
                 break;
         }
 
+        if (get_gfid == _gf_true) {
+                ret = glusterd_get_gfid_from_brick (dict, volinfo, rsp_dict,
+                                                    op_errstr);
+                if (ret)
+                        goto out;
+        }
+
         ret = 0;
 
  out:
diff --git a/xlators/mgmt/glusterd/src/glusterd-syncop.c b/xlators/mgmt/glusterd/src/glusterd-syncop.c
index f87a5787860..8e3f7ebc4eb 100644
--- a/xlators/mgmt/glusterd/src/glusterd-syncop.c
+++ b/xlators/mgmt/glusterd/src/glusterd-syncop.c
@@ -1350,7 +1350,8 @@ gd_commit_op_phase (struct cds_list_head *peers, glusterd_op_t op,
                 }
         }
 
-        if (((op == GD_OP_QUOTA) && (type == GF_QUOTA_OPTION_TYPE_LIST)) ||
+        if (((op == GD_OP_QUOTA) && ((type == GF_QUOTA_OPTION_TYPE_LIST) ||
+             (type == GF_QUOTA_OPTION_TYPE_LIST_OBJECTS))) ||
             ((op != GD_OP_SYNC_VOLUME) && (op != GD_OP_QUOTA))) {
 
                 ret =  glusterd_syncop_aggr_rsp_dict (op, op_ctx,
diff --git a/xlators/mgmt/glusterd/src/glusterd-utils.c b/xlators/mgmt/glusterd/src/glusterd-utils.c
index f98c3b5102c..1cd9a7c4741 100644
--- a/xlators/mgmt/glusterd/src/glusterd-utils.c
+++ b/xlators/mgmt/glusterd/src/glusterd-utils.c
@@ -8344,7 +8344,9 @@ glusterd_volume_quota_copy_to_op_ctx_dict (dict_t *dict, dict_t *rsp_dict)
         }
 
         if ((type != GF_QUOTA_OPTION_TYPE_LIMIT_USAGE) &&
-            (type != GF_QUOTA_OPTION_TYPE_REMOVE)) {
+            (type != GF_QUOTA_OPTION_TYPE_LIMIT_OBJECTS) &&
+            (type != GF_QUOTA_OPTION_TYPE_REMOVE) &&
+            (type != GF_QUOTA_OPTION_TYPE_REMOVE_OBJECTS)) {
                 dict_copy (rsp_dict, dict);
                 ret = 0;
                 goto out;
@@ -9183,7 +9185,9 @@ glusterd_validate_and_set_gfid (dict_t *op_ctx, dict_t *req_dict,
         }
 
         if ((op_code != GF_QUOTA_OPTION_TYPE_LIMIT_USAGE) &&
-            (op_code != GF_QUOTA_OPTION_TYPE_REMOVE)) {
+            (op_code != GF_QUOTA_OPTION_TYPE_LIMIT_OBJECTS) &&
+            (op_code != GF_QUOTA_OPTION_TYPE_REMOVE) &&
+            (op_code != GF_QUOTA_OPTION_TYPE_REMOVE_OBJECTS)) {
                 ret = 0;
                 goto out;
         }