LU-15097 quota: stop pool_recalc before killing pool

[fs/lustre-release.git] / lustre / quota / qmt_pool.c

diff --git a/lustre/quota/qmt_pool.c b/lustre/quota/qmt_pool.c
index d3b7704..c4398a5 100644 (file)
--- a/lustre/quota/qmt_pool.c
+++ b/lustre/quota/qmt_pool.c
@@ -58,7 +58,7 @@ static inline int qmt_sarr_pool_init(struct qmt_pool_info *qpi);
 static inline int qmt_sarr_pool_add(struct qmt_pool_info *qpi,
                                    int idx, int min);
 static inline int qmt_sarr_pool_rem(struct qmt_pool_info *qpi, int idx);
-static inline int qmt_sarr_pool_free(struct qmt_pool_info *qpi);
+static inline void qmt_sarr_pool_free(struct qmt_pool_info *qpi);
 static inline int qmt_sarr_check_idx(struct qmt_pool_info *qpi, int idx);
 static inline void qmt_stop_pool_recalc(struct qmt_pool_info *qpi);
 
@@ -79,6 +79,8 @@ static int qpi_state_seq_show(struct seq_file *m, void *data)
        int                      type;
 
        LASSERT(pool != NULL);
+       if (unlikely(!test_bit(QPI_FLAG_STATE_INITED, &pool->qpi_flags)))
+               return -ENOENT;
 
        seq_printf(m, "pool:\n"
                   "    id: %u\n"
@@ -106,6 +108,8 @@ static int qpi_soft_least_qunit_seq_show(struct seq_file *m, void *data)
 {
        struct qmt_pool_info    *pool = m->private;
        LASSERT(pool != NULL);
+       if (unlikely(!test_bit(QPI_FLAG_STATE_INITED, &pool->qpi_flags)))
+               return -ENOENT;
 
        seq_printf(m, "%lu\n", pool->qpi_soft_least_qunit);
        return 0;
@@ -121,6 +125,8 @@ qpi_soft_least_qunit_seq_write(struct file *file, const char __user *buffer,
        int rc;
 
        LASSERT(pool != NULL);
+       if (unlikely(!test_bit(QPI_FLAG_STATE_INITED, &pool->qpi_flags)))
+               return -ENOENT;
 
        /* Not tuneable for inode limit */
        if (pool->qpi_rtype != LQUOTA_RES_DT)
@@ -177,6 +183,7 @@ static int qmt_pool_alloc(const struct lu_env *env, struct qmt_device *qmt,
        init_rwsem(&pool->qpi_recalc_sem);
 
        pool->qpi_rtype = pool_type;
+       pool->qpi_flags = 0;
 
        /* initialize refcount to 1, hash table will then grab an additional
         * reference */
@@ -383,7 +390,7 @@ struct qmt_pool_info *qmt_pool_lookup(const struct lu_env *env,
                                             int idx, bool add)
 {
        struct qmt_pool_info    *pos, *pool;
-       int rc;
+       int rc = 0;
        ENTRY;
 
        down_read(&qmt->qmt_pool_lock);
@@ -409,7 +416,7 @@ struct qmt_pool_info *qmt_pool_lookup(const struct lu_env *env,
                if (idx >= 0 && !qmt_sarr_check_idx(pos, idx)) {
                        rc = qti_pools_add(env, pos);
                        if (rc)
-                               GOTO(out_err, rc);
+                               break;
                        continue;
                }
 
@@ -419,7 +426,7 @@ struct qmt_pool_info *qmt_pool_lookup(const struct lu_env *env,
                        if (add) {
                                rc = qti_pools_add(env, pos);
                                if (rc)
-                                       GOTO(out_err, rc);
+                                       break;
                        } else {
                                qpi_getref(pool);
                        }
@@ -428,6 +435,9 @@ struct qmt_pool_info *qmt_pool_lookup(const struct lu_env *env,
        }
        up_read(&qmt->qmt_pool_lock);
 
+       if (rc)
+               GOTO(out_err, rc);
+
        if (idx >= 0 && qti_pools_cnt(env))
                pool = qti_pools_env(env)[0];
 
@@ -435,7 +445,7 @@ struct qmt_pool_info *qmt_pool_lookup(const struct lu_env *env,
 out_err:
        CERROR("%s: cannot add pool %s: err = %d\n",
                qmt->qmt_svname, pos->qpi_name, rc);
-       RETURN(ERR_PTR(rc));
+       return ERR_PTR(rc);
 }
 
 /*
@@ -456,6 +466,8 @@ void qmt_pool_fini(const struct lu_env *env, struct qmt_device *qmt)
 
        /* parse list of pool and destroy each element */
        list_for_each_entry_safe(pool, tmp, &qmt->qmt_pool_list, qpi_linkage) {
+               /* stop all recalc threads - it may hold qpi reference */
+               qmt_stop_pool_recalc(pool);
                /* release extra reference taken in qmt_pool_alloc */
                qpi_putref(env, pool);
        }
@@ -667,6 +679,7 @@ int qmt_pool_prepare(const struct lu_env *env, struct qmt_device *qmt,
                                      qmt->qmt_svname, PFID(&qti->qti_fid), rc);
 #endif
                }
+               set_bit(QPI_FLAG_STATE_INITED, &pool->qpi_flags);
                if (name)
                        break;
        }
@@ -821,18 +834,6 @@ int qmt_pool_lqes_lookup(const struct lu_env *env,
                        qti_lqes_fini(env);
                        GOTO(out, rc = PTR_ERR(lqe));
                }
-               /* Only release could be done for not enforced lqe
-                * (see qmt_dqacq0). However slave could request to
-                * release more than not global lqe had granted before
-                * lqe_enforced was cleared. It is legal case,
-                * because even if current lqe is not enforced,
-                * lqes from other pools are still active and avilable
-                * for acquiring. Furthermore, skip not enforced lqe
-                * to don't make extra allocations. */
-               /*if (!lqe_is_glbl(lqe) && !lqe->lqe_enforced) {
-                       lqe_putref(lqe);
-                       continue;
-               }*/
                qti_lqes_add(env, lqe);
        }
        LASSERT(qti_lqes_glbl(env)->lqe_is_global);
@@ -1164,6 +1165,13 @@ static int qmt_pool_recalc(void *args)
 
        pool = args;
 
+       rc = lu_env_init(&env, LCT_MD_THREAD);
+       if (rc) {
+               CERROR("%s: cannot init env: rc = %d\n",
+                      pool->qpi_qmt->qmt_svname, rc);
+               GOTO(out, rc);
+       }
+
        obd = qmt_get_mgc(pool->qpi_qmt);
        if (IS_ERR(obd))
                GOTO(out, rc = PTR_ERR(obd));
@@ -1173,6 +1181,7 @@ static int qmt_pool_recalc(void *args)
                while (obd->obd_process_conf)
                        schedule_timeout_uninterruptible(cfs_time_seconds(1));
 
+       OBD_FAIL_TIMEOUT(OBD_FAIL_QUOTA_RECALC, cfs_fail_val);
        sem = qmt_sarr_rwsem(pool);
        LASSERT(sem);
        down_read(sem);
@@ -1192,15 +1201,9 @@ static int qmt_pool_recalc(void *args)
         * solution looks more complex, so leave it as it is. */
        down_write(&pool->qpi_recalc_sem);
 
-       rc = lu_env_init(&env, LCT_MD_THREAD);
-       if (rc) {
-               CERROR("%s: cannot init env: rc = %d\n", obd->obd_name, rc);
-               GOTO(out, rc);
-       }
-
        glbl_pool = qmt_pool_lookup_glb(&env, pool->qpi_qmt, pool->qpi_rtype);
        if (IS_ERR(glbl_pool))
-               GOTO(out_env, rc = PTR_ERR(glbl_pool));
+               GOTO(out, rc = PTR_ERR(glbl_pool));
 
        slaves_cnt = qmt_sarr_count(pool);
        CDEBUG(D_QUOTA, "Starting pool recalculation for %d slaves in %s\n",
@@ -1246,8 +1249,6 @@ static int qmt_pool_recalc(void *args)
        GOTO(out_stop, rc);
 out_stop:
        qpi_putref(&env, glbl_pool);
-out_env:
-       lu_env_fini(&env);
 out:
        if (xchg(&pool->qpi_recalc_task, NULL) == NULL)
                /*
@@ -1260,12 +1261,20 @@ out:
 
        clear_bit(QPI_FLAG_RECALC_OFFSET, &pool->qpi_flags);
        /* Pool can't be changed, since sem has been down.
-        * Thus until up_read, no one can restart recalc thread. */
+        * Thus until up_read, no one can restart recalc thread.
+        */
        if (sem) {
                up_read(sem);
                up_write(&pool->qpi_recalc_sem);
        }
-       qpi_putref(&env, pool);
+
+       /* qpi_getref has been called in qmt_start_pool_recalc,
+        * however we can't call qpi_putref if lu_env_init failed.
+        */
+       if (env.le_ctx.lc_state == LCS_ENTERED) {
+               qpi_putref(&env, pool);
+               lu_env_fini(&env);
+       }
 
        return rc;
 }
@@ -1358,6 +1367,9 @@ static int qmt_pool_add_rem(struct obd_device *obd, char *poolname,
        int                      rc, idx;
        ENTRY;
 
+       if (qmt->qmt_stopping)
+               RETURN(0);
+
        if (strnlen(poolname, LOV_MAXPOOLNAME + 1) > LOV_MAXPOOLNAME)
                RETURN(-ENAMETOOLONG);
 
@@ -1539,19 +1551,19 @@ static inline int qmt_sarr_pool_rem(struct qmt_pool_info *qpi, int idx)
        }
 }
 
-static inline int qmt_sarr_pool_free(struct qmt_pool_info *qpi)
+static inline void qmt_sarr_pool_free(struct qmt_pool_info *qpi)
 {
        if (qmt_pool_global(qpi))
-               return 0;
+               return;
 
        switch (qpi->qpi_rtype) {
        case LQUOTA_RES_DT:
-               if (!qpi->qpi_sarr.osts.op_array)
-                       return 0;
-               return lu_tgt_pool_free(&qpi->qpi_sarr.osts);
+               if (qpi->qpi_sarr.osts.op_array)
+                       lu_tgt_pool_free(&qpi->qpi_sarr.osts);
+               return;
        case LQUOTA_RES_MD:
        default:
-               return 0;
+               return;
        }
 }