LU-15283 quota: deadlock between reint & lquota_wb

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

diff --git a/lustre/quota/qsd_lib.c b/lustre/quota/qsd_lib.c
index f50ca84..6b827cb 100644 (file)
--- a/lustre/quota/qsd_lib.c
+++ b/lustre/quota/qsd_lib.c
@@ -69,7 +69,7 @@ struct lu_kmem_descr qsd_caches[] = {
 
 /* define qsd thread key */
 LU_KEY_INIT_FINI(qsd, struct qsd_thread_info);
-LU_CONTEXT_KEY_DEFINE(qsd, LCT_MD_THREAD | LCT_DT_THREAD | LCT_LOCAL);
+LU_CONTEXT_KEY_DEFINE(qsd, LCT_MD_THREAD | LCT_MG_THREAD | LCT_DT_THREAD | LCT_LOCAL);
 LU_KEY_INIT_GENERIC(qsd);
 
 /* some procfs helpers */
@@ -312,7 +312,16 @@ static int qsd_conn_callback(void *data)
         * step 3) will have to wait for qsd_start() to be called */
        for (type = USRQUOTA; type < LL_MAXQUOTAS; type++) {
                struct qsd_qtype_info *qqi = qsd->qsd_type_array[type];
-               wake_up(&qqi->qqi_reint_thread.t_ctl_waitq);
+               struct task_struct *t;
+
+               /* qqi_reint_task can be set to NULL at any time,
+                * so we need to be careful.
+                */
+               rcu_read_lock();
+               t = rcu_dereference(qqi->qqi_reint_task);
+               if (t)
+                       wake_up_process(t);
+               rcu_read_unlock();
        }
 
        RETURN(0);
@@ -472,8 +481,6 @@ static int qsd_qtype_init(const struct lu_env *env, struct qsd_instance *qsd,
        qqi->qqi_glb_uptodate = false;
        qqi->qqi_slv_uptodate = false;
        qqi->qqi_reint        = false;
-       init_waitqueue_head(&qqi->qqi_reint_thread.t_ctl_waitq);
-       thread_set_flags(&qqi->qqi_reint_thread, SVC_STOPPED);
        INIT_LIST_HEAD(&qqi->qqi_deferred_glb);
        INIT_LIST_HEAD(&qqi->qqi_deferred_slv);
        lquota_generate_fid(&qqi->qqi_fid, QSD_RES_TYPE(qsd), qtype);
@@ -644,7 +651,8 @@ EXPORT_SYMBOL(qsd_fini);
  */
 struct qsd_instance *qsd_init(const struct lu_env *env, char *svname,
                              struct dt_device *dev,
-                             struct proc_dir_entry *osd_proc, bool is_md)
+                             struct proc_dir_entry *osd_proc,
+                             bool is_md, bool excl)
 {
        struct qsd_thread_info  *qti = qsd_info(env);
        struct qsd_instance     *qsd;
@@ -664,14 +672,14 @@ struct qsd_instance *qsd_init(const struct lu_env *env, char *svname,
        /* generic initializations */
        rwlock_init(&qsd->qsd_lock);
        INIT_LIST_HEAD(&qsd->qsd_link);
-       thread_set_flags(&qsd->qsd_upd_thread, SVC_STOPPED);
-       init_waitqueue_head(&qsd->qsd_upd_thread.t_ctl_waitq);
        INIT_LIST_HEAD(&qsd->qsd_upd_list);
        spin_lock_init(&qsd->qsd_adjust_lock);
        INIT_LIST_HEAD(&qsd->qsd_adjust_list);
        qsd->qsd_prepared = false;
        qsd->qsd_started = false;
        qsd->qsd_is_md = is_md;
+       qsd->qsd_updating = false;
+       qsd->qsd_exclusive = excl;
 
        /* copy service name */
        if (strlcpy(qsd->qsd_svname, svname, sizeof(qsd->qsd_svname))
@@ -889,7 +897,16 @@ int qsd_start(const struct lu_env *env, struct qsd_instance *qsd)
         * up to usage; If usage < granted, release down to usage.  */
        for (type = USRQUOTA; type < LL_MAXQUOTAS; type++) {
                struct qsd_qtype_info   *qqi = qsd->qsd_type_array[type];
-               wake_up(&qqi->qqi_reint_thread.t_ctl_waitq);
+               struct task_struct *t;
+
+               /* qqi_reint_task can be set to NULL at any time,
+                * so we need to be careful.
+                */
+               rcu_read_lock();
+               t = rcu_dereference(qqi->qqi_reint_task);
+               if (t)
+                       wake_up_process(t);
+               rcu_read_unlock();
        }
 
        RETURN(rc);