LU-1057 quota: speed up lookup in osc_quota_chkdq

This patch replace the global hash table used to store uid/gid
about to run out of quota space with a per-OSC cfs_hash.

Signed-off-by: Gregoire Pichon <gregoire.pichon@bull.net>
Change-Id: If291ce125d79312ba7abc182a5561657fa63b6bd
Reviewed-on: http://review.whamcloud.com/4184
Tested-by: Hudson
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Johann Lombardi <johann.lombardi@intel.com>
Reviewed-by: Niu Yawei <niu@whamcloud.com>

diff --git a/lustre/include/obd.h b/lustre/include/obd.h
index 6c3ec4f..041cf3a 100644 (file)
--- a/lustre/include/obd.h
+++ b/lustre/include/obd.h
@@ -490,6 +490,9 @@ struct client_obd {
         struct lu_client_seq    *cl_seq;
 
         cfs_atomic_t             cl_resends; /* resend count */
+
+        /* hash tables for osc_quota_info */
+        cfs_hash_t              *cl_quota_hash[MAXQUOTAS];
 };
 #define obd2cli_tgt(obd) ((char *)(obd)->u.cli.cl_target_uuid.uuid)
 

diff --git a/lustre/osc/osc_request.c b/lustre/osc/osc_request.c
index 190d247..545dc4b 100644 (file)
--- a/lustre/osc/osc_request.c
+++ b/lustre/osc/osc_request.c
@@ -4468,7 +4468,7 @@ int osc_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
         rc = client_obd_setup(obd, lcfg);
         if (rc) {
                 ptlrpcd_decref();
-        } else {
+        } else if ((rc = lquota_setup(quota_interface, obd)) == 0) {
                 struct lprocfs_static_vars lvars = { 0 };
                 struct client_obd *cli = &obd->u.cli;
 

diff --git a/lustre/quota/quota_interface.c b/lustre/quota/quota_interface.c
index 01dc6ac..1ba9faa 100644 (file)
--- a/lustre/quota/quota_interface.c
+++ b/lustre/quota/quota_interface.c
@@ -734,192 +734,233 @@ static int quota_acquire_common(struct obd_device *obd, const unsigned int id[],
 #endif /* __KERNEL__ */
 
 struct osc_quota_info {
-        cfs_list_t              oqi_hash;       /* hash list */
-        struct client_obd      *oqi_cli;        /* osc obd */
-        unsigned int            oqi_id;         /* uid/gid of a file */
-        short                   oqi_type;       /* quota type */
+        /** linkage for quota hash table */
+        cfs_hlist_node_t oqi_hash;
+        obd_uid          oqi_id;
 };
 
-cfs_spinlock_t qinfo_list_lock = CFS_SPIN_LOCK_UNLOCKED;
-
-static cfs_list_t qinfo_hash[NR_DQHASH];
 /* SLAB cache for client quota context */
 cfs_mem_cache_t *qinfo_cachep = NULL;
 
-static inline int hashfn(struct client_obd *cli, unsigned long id, int type)
-                         __attribute__((__const__));
-
-static inline int hashfn(struct client_obd *cli, unsigned long id, int type)
-{
-        unsigned long tmp = ((unsigned long)cli>>6) ^ id;
-        tmp = (tmp * (MAXQUOTAS - type)) % NR_DQHASH;
-        return tmp;
-}
-
-/* caller must hold qinfo_list_lock */
-static inline void insert_qinfo_hash(struct osc_quota_info *oqi)
+static inline struct osc_quota_info *osc_oqi_alloc(obd_uid id)
 {
-        cfs_list_t *head = qinfo_hash +
-                hashfn(oqi->oqi_cli, oqi->oqi_id, oqi->oqi_type);
+        struct osc_quota_info *oqi;
 
-        LASSERT_SPIN_LOCKED(&qinfo_list_lock);
-        cfs_list_add(&oqi->oqi_hash, head);
-}
+        OBD_SLAB_ALLOC_PTR(oqi, qinfo_cachep);
+        if (oqi != NULL)
+                oqi->oqi_id = id;
 
-/* caller must hold qinfo_list_lock */
-static inline void remove_qinfo_hash(struct osc_quota_info *oqi)
-{
-        LASSERT_SPIN_LOCKED(&qinfo_list_lock);
-        cfs_list_del_init(&oqi->oqi_hash);
+        return oqi;
 }
 
-/* caller must hold qinfo_list_lock */
-static inline struct osc_quota_info *find_qinfo(struct client_obd *cli,
-                                                unsigned int id, int type)
+int osc_quota_chkdq(struct client_obd *cli, const unsigned int qid[])
 {
-        unsigned int hashent = hashfn(cli, id, type);
-        struct osc_quota_info *oqi;
+        int type;
         ENTRY;
 
-        LASSERT_SPIN_LOCKED(&qinfo_list_lock);
-        cfs_list_for_each_entry(oqi, &qinfo_hash[hashent], oqi_hash) {
-                if (oqi->oqi_cli == cli &&
-                    oqi->oqi_id == id && oqi->oqi_type == type)
-                        return oqi;
+        for (type = 0; type < MAXQUOTAS; type++) {
+                struct osc_quota_info *oqi;
+
+                oqi = cfs_hash_lookup(cli->cl_quota_hash[type], &qid[type]);
+                if (oqi) {
+                        obd_uid id = oqi->oqi_id;
+
+                        LASSERTF(id == qid[type],
+                                 "The ids don't match %u != %u\n",
+                                 id, qid[type]);
+
+                        /* the slot is busy, the user is about to run out of
+                         * quota space on this OST */
+                        CDEBUG(D_QUOTA, "chkdq found noquota for %s %d\n",
+                               type == USRQUOTA ? "user" : "grout", qid[type]);
+                        RETURN(NO_QUOTA);
+                }
         }
-        RETURN(NULL);
+
+        RETURN(QUOTA_OK);
 }
 
-static struct osc_quota_info *alloc_qinfo(struct client_obd *cli,
-                                          unsigned int id, int type)
+#define MD_QUOTA_FLAG(type) ((type == USRQUOTA) ? OBD_MD_FLUSRQUOTA \
+                                                : OBD_MD_FLGRPQUOTA)
+#define FL_QUOTA_FLAG(type) ((type == USRQUOTA) ? OBD_FL_NO_USRQUOTA \
+                                                : OBD_FL_NO_GRPQUOTA)
+
+int osc_quota_setdq(struct client_obd *cli, const unsigned int qid[],
+                    obd_flag valid, obd_flag flags)
 {
-        struct osc_quota_info *oqi;
+        int type;
+        int rc = 0;
         ENTRY;
 
-        OBD_SLAB_ALLOC(oqi, qinfo_cachep, CFS_ALLOC_IO, sizeof(*oqi));
-        if(!oqi)
-                RETURN(NULL);
+        if ((valid & (OBD_MD_FLUSRQUOTA | OBD_MD_FLGRPQUOTA)) == 0)
+                RETURN(0);
+
+        for (type = 0; type < MAXQUOTAS; type++) {
+                struct osc_quota_info *oqi;
+
+                if ((valid & MD_QUOTA_FLAG(type)) == 0)
+                        continue;
+
+                /* lookup the ID in the per-type hash table */
+                oqi = cfs_hash_lookup(cli->cl_quota_hash[type], &qid[type]);
+                if ((flags & FL_QUOTA_FLAG(type)) != 0) {
+                        /* This ID is getting close to its quota limit, let's
+                         * switch to sync I/O */
+                        if (oqi != NULL)
+                                continue;
+
+                        oqi = osc_oqi_alloc(qid[type]);
+                        if (oqi == NULL) {
+                                rc = -ENOMEM;
+                                break;
+                        }
+
+                        rc = cfs_hash_add_unique(cli->cl_quota_hash[type],
+                                                 &qid[type], &oqi->oqi_hash);
+                        /* race with others? */
+                        if (rc == -EALREADY) {
+                                rc = 0;
+                                OBD_SLAB_FREE_PTR(oqi, qinfo_cachep);
+                        }
+
+                        CDEBUG(D_QUOTA, "%s: setdq to insert for %s %d (%d)\n",
+                               cli->cl_import->imp_obd->obd_name,
+                               type == USRQUOTA ? "user" : "group",
+                               qid[type], rc);
+                } else {
+                        /* This ID is now off the hook, let's remove it from
+                         * the hash table */
+                        if (oqi == NULL)
+                                continue;
 
-        CFS_INIT_LIST_HEAD(&oqi->oqi_hash);
-        oqi->oqi_cli = cli;
-        oqi->oqi_id = id;
-        oqi->oqi_type = type;
+                        oqi = cfs_hash_del_key(cli->cl_quota_hash[type],
+                                               &qid[type]);
+                        if (oqi)
+                                OBD_SLAB_FREE_PTR(oqi, qinfo_cachep);
 
-        RETURN(oqi);
+                        CDEBUG(D_QUOTA, "%s: setdq to remove for %s %d (%p)\n",
+                               cli->cl_import->imp_obd->obd_name,
+                               type == USRQUOTA ? "user" : "group",
+                               qid[type], oqi);
+                }
+        }
+
+        RETURN(rc);
 }
 
-static void free_qinfo(struct osc_quota_info *oqi)
+/*
+ * Hash operations for uid/gid <-> osc_quota_info
+ */
+static unsigned
+oqi_hashfn(cfs_hash_t *hs, const void *key, unsigned mask)
 {
-        OBD_SLAB_FREE(oqi, qinfo_cachep, sizeof(*oqi));
+        return cfs_hash_u32_hash(*((__u32*)key), mask);
 }
 
-int osc_quota_chkdq(struct client_obd *cli, const unsigned int qid[])
+static int
+oqi_keycmp(const void *key, cfs_hlist_node_t *hnode)
 {
-        unsigned int id;
-        int cnt, rc = QUOTA_OK;
-        ENTRY;
+        struct osc_quota_info *oqi;
+        obd_uid uid;
 
-        cfs_spin_lock(&qinfo_list_lock);
-        for (cnt = 0; cnt < MAXQUOTAS; cnt++) {
-                struct osc_quota_info *oqi = NULL;
+        LASSERT(key != NULL);
+        uid = *((obd_uid*)key);
+        oqi = cfs_hlist_entry(hnode, struct osc_quota_info, oqi_hash);
 
-                id = (cnt == USRQUOTA) ? qid[USRQUOTA] : qid[GRPQUOTA];
-                oqi = find_qinfo(cli, id, cnt);
-                if (oqi) {
-                        rc = NO_QUOTA;
-                        break;
-                }
-        }
-        cfs_spin_unlock(&qinfo_list_lock);
+        return uid == oqi->oqi_id;
+}
 
-        if (rc == NO_QUOTA)
-                CDEBUG(D_QUOTA, "chkdq found noquota for %s %d\n",
-                       cnt == USRQUOTA ? "user" : "group", id);
-        RETURN(rc);
+static void *
+oqi_key(cfs_hlist_node_t *hnode)
+{
+        struct osc_quota_info *oqi;
+        oqi = cfs_hlist_entry(hnode, struct osc_quota_info, oqi_hash);
+        return &oqi->oqi_id;
 }
 
-int osc_quota_setdq(struct client_obd *cli, const unsigned int qid[],
-                    obd_flag valid, obd_flag flags)
+static void *
+oqi_object(cfs_hlist_node_t *hnode)
 {
-        unsigned int id;
-        obd_flag noquota;
-        int cnt, rc = 0;
-        ENTRY;
+        return cfs_hlist_entry(hnode, struct osc_quota_info, oqi_hash);
+}
 
+static void
+oqi_get(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
+{
+}
 
-        for (cnt = 0; cnt < MAXQUOTAS; cnt++) {
-                struct osc_quota_info *oqi = NULL, *old;
+static void
+oqi_put_locked(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
+{
+}
 
-                if (!(valid & ((cnt == USRQUOTA) ?
-                    OBD_MD_FLUSRQUOTA : OBD_MD_FLGRPQUOTA)))
-                        continue;
+static void
+oqi_exit(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
+{
+        struct osc_quota_info *oqi;
 
-                id = (cnt == USRQUOTA) ? qid[USRQUOTA] : qid[GRPQUOTA];
-                noquota = (cnt == USRQUOTA) ?
-                    (flags & OBD_FL_NO_USRQUOTA) : (flags & OBD_FL_NO_GRPQUOTA);
+        oqi = cfs_hlist_entry(hnode, struct osc_quota_info, oqi_hash);
 
-                if (noquota) {
-                        oqi = alloc_qinfo(cli, id, cnt);
-                        if (!oqi) {
-                                rc = -ENOMEM;
-                                CDEBUG(D_QUOTA, "setdq for %s %d failed, "
-                                       "(rc = %d)\n",
-                                       cnt == USRQUOTA ? "user" : "group",
-                                       id, rc);
-                                break;
-                        }
-                }
+        OBD_SLAB_FREE_PTR(oqi, qinfo_cachep);
+}
 
-                cfs_spin_lock(&qinfo_list_lock);
-                old = find_qinfo(cli, id, cnt);
-                if (old && !noquota)
-                        remove_qinfo_hash(old);
-                else if (!old && noquota)
-                        insert_qinfo_hash(oqi);
-                cfs_spin_unlock(&qinfo_list_lock);
-
-                if (old && !noquota)
-                        CDEBUG(D_QUOTA, "setdq to remove for %s %d\n",
-                               cnt == USRQUOTA ? "user" : "group", id);
-                else if (!old && noquota)
-                        CDEBUG(D_QUOTA, "setdq to insert for %s %d\n",
-                               cnt == USRQUOTA ? "user" : "group", id);
-
-                if (old) {
-                        if (noquota)
-                                free_qinfo(oqi);
-                        else
-                                free_qinfo(old);
-                }
+#define HASH_QUOTA_BKT_BITS 5
+#define HASH_QUOTA_CUR_BITS 5
+#define HASH_QUOTA_MAX_BITS 15
+
+static cfs_hash_ops_t quota_hash_ops = {
+        .hs_hash       = oqi_hashfn,
+        .hs_keycmp     = oqi_keycmp,
+        .hs_key        = oqi_key,
+        .hs_object     = oqi_object,
+        .hs_get        = oqi_get,
+        .hs_put_locked = oqi_put_locked,
+        .hs_exit       = oqi_exit,
+};
+
+int osc_quota_setup(struct obd_device *obd)
+{
+        struct client_obd *cli = &obd->u.cli;
+        int i, type;
+        ENTRY;
+
+        for (type = 0; type < MAXQUOTAS; type++) {
+                cli->cl_quota_hash[type] = cfs_hash_create("QUOTA_HASH",
+                                                           HASH_QUOTA_CUR_BITS,
+                                                           HASH_QUOTA_MAX_BITS,
+                                                           HASH_QUOTA_BKT_BITS,
+                                                           0,
+                                                           CFS_HASH_MIN_THETA,
+                                                           CFS_HASH_MAX_THETA,
+                                                           &quota_hash_ops,
+                                                           CFS_HASH_DEFAULT);
+                if (cli->cl_quota_hash[type] == NULL)
+                        break;
         }
 
-        RETURN(rc);
+        if (type == MAXQUOTAS)
+                RETURN(0);
+
+        for (i = 0; i < type; i++)
+                cfs_hash_putref(cli->cl_quota_hash[i]);
+
+        RETURN(-ENOMEM);
 }
 
 int osc_quota_cleanup(struct obd_device *obd)
 {
         struct client_obd *cli = &obd->u.cli;
-        struct osc_quota_info *oqi, *n;
-        int i;
+        int type;
         ENTRY;
 
-        cfs_spin_lock(&qinfo_list_lock);
-        for (i = 0; i < NR_DQHASH; i++) {
-                cfs_list_for_each_entry_safe(oqi, n, &qinfo_hash[i], oqi_hash) {
-                        if (oqi->oqi_cli != cli)
-                                continue;
-                        remove_qinfo_hash(oqi);
-                        free_qinfo(oqi);
-                }
-        }
-        cfs_spin_unlock(&qinfo_list_lock);
+        for (type = 0; type < MAXQUOTAS; type++)
+                cfs_hash_putref(cli->cl_quota_hash[type]);
 
         RETURN(0);
 }
 
 int osc_quota_init(void)
 {
-        int i;
         ENTRY;
 
         LASSERT(qinfo_cachep == NULL);
@@ -929,27 +970,14 @@ int osc_quota_init(void)
         if (!qinfo_cachep)
                 RETURN(-ENOMEM);
 
-        for (i = 0; i < NR_DQHASH; i++)
-                CFS_INIT_LIST_HEAD(qinfo_hash + i);
-
         RETURN(0);
 }
 
 int osc_quota_exit(void)
 {
-        struct osc_quota_info *oqi, *n;
-        int i, rc;
+        int rc;
         ENTRY;
 
-        cfs_spin_lock(&qinfo_list_lock);
-        for (i = 0; i < NR_DQHASH; i++) {
-                cfs_list_for_each_entry_safe(oqi, n, &qinfo_hash[i], oqi_hash) {
-                        remove_qinfo_hash(oqi);
-                        free_qinfo(oqi);
-                }
-        }
-        cfs_spin_unlock(&qinfo_list_lock);
-
         rc = cfs_mem_cache_destroy(qinfo_cachep);
         LASSERTF(rc == 0, "couldn't destory qinfo_cachep slab\n");
         qinfo_cachep = NULL;
@@ -957,6 +985,7 @@ int osc_quota_exit(void)
         RETURN(0);
 }
 
+
 #ifdef __KERNEL__
 #ifdef HAVE_QUOTA_SUPPORT
 quota_interface_t mds_quota_interface = {
@@ -1010,6 +1039,7 @@ quota_interface_t osc_quota_interface = {
         .quota_poll_check = client_quota_poll_check,
         .quota_init     = osc_quota_init,
         .quota_exit     = osc_quota_exit,
+        .quota_setup    = osc_quota_setup,
         .quota_chkdq    = osc_quota_chkdq,
         .quota_setdq    = osc_quota_setdq,
         .quota_cleanup  = osc_quota_cleanup,