LU-16772 quota: access lqe_glbl_data only under mutex

Hold mutex to protect lqe_glbl_data against freeing in
qmt_lvbo_update, qmt_allock_lock_array and qmt_setup_id_desc.
This patch should help against below(LU-14434) and similar panics.

  BUG: unable to handle kernel NULL pointer dereference at 0000000000000000
  RIP: 0010:qmt_id_lock_cb+0x69/0x100 [lquota]
  qmt_id_lock_cb+0x69/0x100 [lquota]
  qmt_glimpse_lock.isra.19+0x27e/0xfb0 [lquota]
  qmt_reba_thread+0x5da/0x9b0 [lquota]
  kthread+0x112/0x130
  ret_from_fork+0x35/0x40

It is the 2nd part of 50ff4d1da63e8. The latest patchset of
https://review.whamcloud.com/c/fs/lustre-release/+/50748/ has extra
changes, but occasionally has been landed without inspection.
So it's not simple porting from upstream.

Fixes: 50ff4d1da63 ("LU-16772 quota: protect lqe_glbl_data in qmt_site_recalc_cb")
Signed-off-by: Sergey Cheremencev <scherementsev@ddn.com>
Change-Id: I5753ace94a739c6df70dd8ea3bde828e2b5ed812
Reviewed-on: https://review.whamcloud.com/c/ex/lustre-release/+/51368
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Andreas Dilger <adilger@whamcloud.com>
Reviewed-by: Hongchao Zhang <hongchao@whamcloud.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>

diff --git a/lustre/quota/qmt_lock.c b/lustre/quota/qmt_lock.c
index a21d466..52f5d8b 100644 (file)
--- a/lustre/quota/qmt_lock.c
+++ b/lustre/quota/qmt_lock.c
@@ -401,10 +401,11 @@ 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, rc, idx) &&
-           lqe->lqe_glbl_data) {
+       if (need_revoke && qmt_set_revoke(env, lqe, rc, idx)) {
+               mutex_lock(&lqe->lqe_glbl_data_lock);
                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) {
@@ -581,6 +582,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;
@@ -588,6 +590,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) {
@@ -611,6 +615,8 @@ again:
                }
        }
        unlock_res(res);
+       if (cb)
+               mutex_unlock(&lqe->lqe_glbl_data_lock);
 
        if (count > array->q_max) {
                qmt_free_lock_array(array);
@@ -638,7 +644,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;
@@ -651,8 +656,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 */
@@ -681,7 +696,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;
@@ -689,7 +703,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) {
@@ -706,7 +719,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
@@ -733,7 +746,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);
@@ -857,8 +870,8 @@ static int qmt_id_lock_cb(struct ldlm_lock *lock, struct lquota_entry *lqe)
        if (lqe_rtype(lqe) == LQUOTA_RES_DT && stype == QMT_STYPE_MDT)
                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;
 }