LU-16772 quota: protect lqe_glbl_data in qmt_site_recalc_cb

lqe_glbl_data should be protected with lqe_glbl_data_lock in
qmt_site_recalc_sb like it did in other places. Otherwise it
may cause following panic:

  BUG: unable to handle kernel NULL pointer at 00000000000000f8
  qmt_site_recalc_cb+0x2f8/0x790 [lquota]
  cfs_hash_for_each_tight+0x121/0x310 [libcfs]
  qmt_pool_recalc+0x372/0x9f0 [lquota]

Also protect lqe_glbl_data access with lqe_glbl_data_lock in
qmt_lvbo_free().

Fixes: 1dbcbd70f8 ("LU-15021 quota: protect lqe_glbl_data in lqe")
Signed-off-by: Sergey Cheremencev <scherementsev@ddn.com>
Change-Id: I030f14b02062151f1708a03ac7414a9991f798f6
Reviewed-on: https://review.whamcloud.com/c/fs/lustre-release/+/50748
Tested-by: jenkins <devops@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/quota/qmt_entry.c b/lustre/quota/qmt_entry.c
index b17b19a..cb3dca2 100644 (file)
--- a/lustre/quota/qmt_entry.c
+++ b/lustre/quota/qmt_entry.c
@@ -1118,6 +1118,9 @@ struct lqe_glbl_data *qmt_alloc_lqe_gd(struct qmt_pool_info *pool, int qtype)
 
 void qmt_free_lqe_gd(struct lqe_glbl_data *lgd)
 {
+       if (unlikely(!lgd))
+               return;
+
        OBD_FREE(lgd->lqeg_arr,
                 sizeof(struct lqe_glbl_entry) * lgd->lqeg_num_alloc);
        OBD_FREE(lgd, sizeof(struct lqe_glbl_data));

diff --git a/lustre/quota/qmt_lock.c b/lustre/quota/qmt_lock.c
index 9d2e01a..67cd526 100644 (file)
--- a/lustre/quota/qmt_lock.c
+++ b/lustre/quota/qmt_lock.c
@@ -402,10 +402,13 @@ int qmt_lvbo_update(struct lu_device *ld, struct ldlm_resource *res,
        if (rc)
                GOTO(out_exp, rc);
 
-       if (need_revoke && qmt_set_revoke(env, lqe, stype, idx) &&
-           lqe->lqe_glbl_data) {
-               qmt_seed_glbe_edquot(env, lqe->lqe_glbl_data);
-               qmt_id_lock_notify(qmt, lqe);
+       if (need_revoke && qmt_set_revoke(env, lqe, stype, idx)) {
+               mutex_lock(&lqe->lqe_glbl_data_lock);
+               if (lqe->lqe_glbl_data) {
+                       qmt_seed_glbe_edquot(env, lqe->lqe_glbl_data);
+                       qmt_id_lock_notify(qmt, lqe);
+               }
+               mutex_unlock(&lqe->lqe_glbl_data_lock);
        }
 
        if (lvb->lvb_id_rel) {
@@ -520,12 +523,13 @@ int qmt_lvbo_free(struct lu_device *ld, struct ldlm_resource *res)
 
        if (res->lr_name.name[LUSTRE_RES_ID_QUOTA_SEQ_OFF] != 0) {
                struct lquota_entry *lqe = res->lr_lvb_data;
-               struct lqe_glbl_data *lgd = lqe->lqe_glbl_data;
+               struct lqe_glbl_data *lgd;
 
                mutex_lock(&lqe->lqe_glbl_data_lock);
+               lgd = lqe->lqe_glbl_data;
                lqe->lqe_glbl_data = NULL;
-               qmt_free_lqe_gd(lgd);
                mutex_unlock(&lqe->lqe_glbl_data_lock);
+               qmt_free_lqe_gd(lgd);
 
                /* release lqe reference */
                lqe_putref(lqe);
@@ -573,6 +577,7 @@ static int qmt_alloc_lock_array(struct ldlm_resource *res,
                                struct qmt_gl_lock_array *array,
                                qmt_glimpse_cb_t cb, void *arg)
 {
+       struct lquota_entry *lqe = arg;
        struct list_head *pos;
        unsigned long count = 0;
        int fail_cnt = 0;
@@ -580,6 +585,8 @@ static int qmt_alloc_lock_array(struct ldlm_resource *res,
 
        LASSERT(!array->q_max && !array->q_cnt && !array->q_locks);
 again:
+       if (cb)
+               mutex_lock(&lqe->lqe_glbl_data_lock);
        lock_res(res);
        /* scan list of granted locks */
        list_for_each(pos, &res->lr_granted) {
@@ -603,6 +610,8 @@ again:
                }
        }
        unlock_res(res);
+       if (cb)
+               mutex_unlock(&lqe->lqe_glbl_data_lock);
 
        if (count > array->q_max) {
                qmt_free_lock_array(array);
@@ -630,7 +639,6 @@ void qmt_setup_id_desc(struct ldlm_lock *lock, union ldlm_gl_desc *desc,
                       struct lquota_entry *lqe)
 {
        struct obd_uuid *uuid = &(lock)->l_export->exp_client_uuid;
-       struct lqe_glbl_data *lgd = lqe->lqe_glbl_data;
        int idx, stype;
        __u64 qunit;
        bool edquot;
@@ -643,8 +651,18 @@ void qmt_setup_id_desc(struct ldlm_lock *lock, union ldlm_gl_desc *desc,
                edquot = lqe->lqe_edquot;
                qunit = lqe->lqe_qunit;
        } else {
-               edquot = lgd->lqeg_arr[idx].lge_edquot;
-               qunit = lgd->lqeg_arr[idx].lge_qunit;
+               struct lqe_glbl_data *lgd;
+
+               mutex_lock(&lqe->lqe_glbl_data_lock);
+               lgd = lqe->lqe_glbl_data;
+               if (lgd) {
+                       edquot = lgd->lqeg_arr[idx].lge_edquot;
+                       qunit = lgd->lqeg_arr[idx].lge_qunit;
+               } else {
+                       edquot = lqe->lqe_edquot;
+                       qunit = lqe->lqe_qunit;
+               }
+               mutex_unlock(&lqe->lqe_glbl_data_lock);
        }
 
        /* fill glimpse descriptor with lqe settings */
@@ -673,7 +691,6 @@ static int qmt_glimpse_lock(const struct lu_env *env, struct qmt_device *qmt,
                            qmt_glimpse_cb_t cb, struct lquota_entry *lqe)
 {
        union ldlm_gl_desc *descs = NULL;
-       struct lqe_glbl_data *gld;
        struct list_head *tmp, *pos;
        LIST_HEAD(gl_list);
        struct qmt_gl_lock_array locks;
@@ -681,7 +698,6 @@ static int qmt_glimpse_lock(const struct lu_env *env, struct qmt_device *qmt,
        int rc = 0;
        ENTRY;
 
-       gld = lqe ? lqe->lqe_glbl_data : NULL;
        memset(&locks, 0, sizeof(locks));
        rc = qmt_alloc_lock_array(res, &locks, cb, lqe);
        if (rc) {
@@ -698,7 +714,7 @@ static int qmt_glimpse_lock(const struct lu_env *env, struct qmt_device *qmt,
        locks_count = locks.q_cnt;
 
        /* Use one desc for all works, when called from qmt_glb_lock_notify */
-       if (gld && locks.q_cnt > 1) {
+       if (cb && locks.q_cnt > 1) {
                /* TODO: think about to store this preallocated descs
                 * in lqe_global in lqeg_arr as a part of lqe_glbl_entry.
                 * The benefit is that we don't need to allocate/free
@@ -725,7 +741,7 @@ static int qmt_glimpse_lock(const struct lu_env *env, struct qmt_device *qmt,
                        continue;
                }
 
-               if (gld) {
+               if (cb) {
                        if (descs)
                                desc = &descs[i - 1];
                        qmt_setup_id_desc(locks.q_locks[i - 1], desc, lqe);
@@ -849,8 +865,8 @@ static int qmt_id_lock_cb(struct ldlm_lock *lock, struct lquota_entry *lqe)
        if (qmt_dom(lqe_rtype(lqe), stype))
                return 1;
        else
-               return lgd->lqeg_arr[idx].lge_edquot_nu ||
-                      lgd->lqeg_arr[idx].lge_qunit_nu;
+               return lgd ? lgd->lqeg_arr[idx].lge_edquot_nu ||
+                      lgd->lqeg_arr[idx].lge_qunit_nu : 0;
 }
 
 

diff --git a/lustre/quota/qmt_pool.c b/lustre/quota/qmt_pool.c
index ad04b4c..92bcf60 100644 (file)
--- a/lustre/quota/qmt_pool.c
+++ b/lustre/quota/qmt_pool.c
@@ -1088,10 +1088,14 @@ static int qmt_site_recalc_cb(struct cfs_hash *hs, struct cfs_hash_bd *bd,
                        /* Find all lqes with lqe_id to reseed lgd array */
                        rc = qmt_pool_lqes_lookup_spec(env, qmt, lqe_rtype(lqe),
                                                lqe_qtype(lqe), &lqe->lqe_id);
-                       if (!rc && qti_lqes_glbl(env)->lqe_glbl_data) {
-                               qmt_seed_glbe(env,
-                                       qti_lqes_glbl(env)->lqe_glbl_data);
-                               qmt_id_lock_notify(qmt, qti_lqes_glbl(env));
+                       if (!rc) {
+                               struct lquota_entry *lqeg = qti_lqes_glbl(env);
+
+                               mutex_lock(&lqeg->lqe_glbl_data_lock);
+                               if (lqeg->lqe_glbl_data)
+                                       qmt_seed_glbe(env, lqeg->lqe_glbl_data);
+                               mutex_unlock(&lqeg->lqe_glbl_data_lock);
+                               qmt_id_lock_notify(qmt, lqeg);
                        }
                        qti_lqes_fini(env);
                }