LU-12780 quota: don't use ptlrpc_thread for qmt_reba_thread

Instead of ptlrpc_thread, use native kthread functionality.

- for startup, perform allocations before creating the thread so that
  once created it cannot fail.
  We still use a completion to ensure the thread function runs before
  kthread_stop is called, so that cleanup will happen.

- As lu_env_add() can fail, and needs to know which task owns
  the env, we add a new function lu_env_add_task()

- for shutdown, use kthread_stop() and kthread_should_stop()

- to alert thread of a new event, use wake_up_process()
  Do this under qmt_reba_lock so the thread cannot disappear while
  being woken.
  The thread sets TASK_IDLE, then if any tests show there is work
  to do it sets TASK_RUNNING, so the following schedule() won't
  block and the loop will check again.

Signed-off-by: Mr NeilBrown <neilb@suse.de>
Change-Id: If3f0444d5aa38ea84990d95f85ad18202f99d5df
Reviewed-on: https://review.whamcloud.com/36556
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Wang Shilong <wshilong@ddn.com>
Reviewed-by: James Simmons <jsimmons@infradead.org>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/include/lu_object.h b/lustre/include/lu_object.h
index 65180b2..bf2c445 100644 (file)
--- a/lustre/include/lu_object.h
+++ b/lustre/include/lu_object.h
@@ -1292,6 +1292,7 @@ static inline void* lu_env_info(const struct lu_env *env,
 
 struct lu_env *lu_env_find(void);
 int lu_env_add(struct lu_env *env);
+int lu_env_add_task(struct lu_env *env, struct task_struct *task);
 void lu_env_remove(struct lu_env *env);
 
 /** @} lu_context */

diff --git a/lustre/obdclass/lu_object.c b/lustre/obdclass/lu_object.c
index 7ebe04e..cd2f481 100644 (file)
--- a/lustre/obdclass/lu_object.c
+++ b/lustre/obdclass/lu_object.c
@@ -2011,7 +2011,7 @@ struct lu_env_percpu {
 
 static struct lu_env_percpu lu_env_percpu[NR_CPUS];
 
-int lu_env_add(struct lu_env *env)
+int lu_env_add_task(struct lu_env *env, struct task_struct *task)
 {
        struct lu_env_item *lei, *old;
 
@@ -2021,7 +2021,7 @@ int lu_env_add(struct lu_env *env)
        if (!lei)
                return -ENOMEM;
 
-       lei->lei_task = current;
+       lei->lei_task = task;
        lei->lei_env = env;
 
        old = rhashtable_lookup_get_insert_fast(&lu_env_rhash,
@@ -2031,6 +2031,12 @@ int lu_env_add(struct lu_env *env)
 
        return 0;
 }
+EXPORT_SYMBOL(lu_env_add_task);
+
+int lu_env_add(struct lu_env *env)
+{
+       return lu_env_add_task(env, current);
+}
 EXPORT_SYMBOL(lu_env_add);
 
 static void lu_env_item_free(struct rcu_head *head)

diff --git a/lustre/quota/qmt_dev.c b/lustre/quota/qmt_dev.c
index d876cd3..cab5481 100644 (file)
--- a/lustre/quota/qmt_dev.c
+++ b/lustre/quota/qmt_dev.c
@@ -241,8 +241,6 @@ static int qmt_device_init0(const struct lu_env *env, struct qmt_device *qmt,
                GOTO(out, rc);
 
        /* set up and start rebalance thread */
-       thread_set_flags(&qmt->qmt_reba_thread, SVC_STARTING);
-       init_waitqueue_head(&qmt->qmt_reba_thread.t_ctl_waitq);
        INIT_LIST_HEAD(&qmt->qmt_reba_list);
        spin_lock_init(&qmt->qmt_reba_lock);
        if (!qmt->qmt_child->dd_rdonly) {

diff --git a/lustre/quota/qmt_internal.h b/lustre/quota/qmt_internal.h
index a3ab56f..cd68fa8 100644 (file)
--- a/lustre/quota/qmt_internal.h
+++ b/lustre/quota/qmt_internal.h
@@ -76,7 +76,7 @@ struct qmt_device {
        struct proc_dir_entry   *qmt_proc;
 
        /* dedicated thread in charge of space rebalancing */
-       struct ptlrpc_thread     qmt_reba_thread;
+       struct task_struct      *qmt_reba_task;
 
        /* list of lqe entry which need space rebalancing */
        struct list_head         qmt_reba_list;

diff --git a/lustre/quota/qmt_lock.c b/lustre/quota/qmt_lock.c
index 0e20f9c..5b744ec 100644 (file)
--- a/lustre/quota/qmt_lock.c
+++ b/lustre/quota/qmt_lock.c
@@ -722,16 +722,26 @@ void qmt_id_lock_notify(struct qmt_device *qmt, struct lquota_entry *lqe)
        if (!qmt->qmt_stopping && list_empty(&lqe->lqe_link)) {
                list_add_tail(&lqe->lqe_link, &qmt->qmt_reba_list);
                added = true;
+               if (qmt->qmt_reba_task)
+                       wake_up_process(qmt->qmt_reba_task);
        }
        spin_unlock(&qmt->qmt_reba_lock);
 
-       if (added)
-               wake_up(&qmt->qmt_reba_thread.t_ctl_waitq);
-       else
+       if (!added)
                lqe_putref(lqe);
        EXIT;
 }
 
+struct qmt_reba_args {
+       struct qmt_device       *qra_dev;
+       struct lu_env            qra_env;
+       struct completion       *qra_started;
+};
+
+#ifndef TASK_IDLE
+#define TASK_IDLE TASK_INTERRUPTIBLE
+#endif
+
 /*
  * The rebalance thread is in charge of sending glimpse callbacks on per-ID
  * quota locks owned by slaves in order to notify them of:
@@ -742,63 +752,40 @@ void qmt_id_lock_notify(struct qmt_device *qmt, struct lquota_entry *lqe)
  *   try to acquire quota from the master since this latter has already
  *   distributed all the space.
  */
-static int qmt_reba_thread(void *arg)
+static int qmt_reba_thread(void *_args)
 {
-       struct qmt_device       *qmt = (struct qmt_device *)arg;
-       struct ptlrpc_thread    *thread = &qmt->qmt_reba_thread;
-       struct lu_env           *env;
+       struct qmt_reba_args    *args = _args;
+       struct qmt_device       *qmt = args->qra_dev;
+       struct lu_env           *env = &args->qra_env;
        struct lquota_entry     *lqe, *tmp;
-       int                      rc;
        ENTRY;
 
-       OBD_ALLOC_PTR(env);
-       if (env == NULL) {
-               thread_set_flags(thread, SVC_STOPPED);
-               RETURN(-ENOMEM);
-       }
-
-       rc = lu_env_init(env, LCT_MD_THREAD);
-       if (rc) {
-               CERROR("%s: failed to init env.", qmt->qmt_svname);
-               GOTO(out_env, rc);
-       }
-       rc = lu_env_add(env);
-       if (rc)
-               GOTO(out_env_fini, rc);
-
-       thread_set_flags(thread, SVC_RUNNING);
-       wake_up(&thread->t_ctl_waitq);
-
-       while (1) {
-               wait_event_idle(thread->t_ctl_waitq,
-                               !list_empty(&qmt->qmt_reba_list) ||
-                               !thread_is_running(thread));
+       complete(args->qra_started);
+       while (({set_current_state(TASK_IDLE);
+                !kthread_should_stop(); })) {
 
                spin_lock(&qmt->qmt_reba_lock);
                list_for_each_entry_safe(lqe, tmp, &qmt->qmt_reba_list,
                                         lqe_link) {
+                       __set_current_state(TASK_RUNNING);
                        list_del_init(&lqe->lqe_link);
                        spin_unlock(&qmt->qmt_reba_lock);
 
-                       if (thread_is_running(thread))
+                       if (!kthread_should_stop())
                                qmt_id_lock_glimpse(env, qmt, lqe, NULL);
 
                        lqe_putref(lqe);
                        spin_lock(&qmt->qmt_reba_lock);
                }
                spin_unlock(&qmt->qmt_reba_lock);
-
-               if (!thread_is_running(thread))
-                       break;
+               schedule();
        }
+       __set_current_state(TASK_RUNNING);
+
        lu_env_remove(env);
-out_env_fini:
        lu_env_fini(env);
-out_env:
-       OBD_FREE_PTR(env);
-       thread_set_flags(thread, SVC_STOPPED);
-       wake_up(&thread->t_ctl_waitq);
-       RETURN(rc);
+       OBD_FREE_PTR(args);
+       RETURN(0);
 }
 
 /*
@@ -806,23 +793,47 @@ out_env:
  */
 int qmt_start_reba_thread(struct qmt_device *qmt)
 {
-       struct ptlrpc_thread    *thread = &qmt->qmt_reba_thread;
-       struct task_struct              *task;
+       struct task_struct *task;
+       struct qmt_reba_args *args;
+       DECLARE_COMPLETION_ONSTACK(started);
+       int rc;
        ENTRY;
 
-       task = kthread_run(qmt_reba_thread, (void *)qmt,
-                              "qmt_reba_%s", qmt->qmt_svname);
+       OBD_ALLOC_PTR(args);
+       if (args == NULL)
+               RETURN(-ENOMEM);
+       args->qra_dev = qmt;
+       args->qra_started = &started;
+
+       rc = lu_env_init(&args->qra_env, LCT_MD_THREAD);
+       if (rc) {
+               CERROR("%s: failed to init env.", qmt->qmt_svname);
+               GOTO(out_env, rc);
+       }
+
+       task = kthread_create(qmt_reba_thread, args,
+                             "qmt_reba_%s", qmt->qmt_svname);
        if (IS_ERR(task)) {
                CERROR("%s: failed to start rebalance thread (%ld)\n",
                       qmt->qmt_svname, PTR_ERR(task));
-               thread_set_flags(thread, SVC_STOPPED);
-               RETURN(PTR_ERR(task));
+               GOTO(out_env_fini, rc = PTR_ERR(task));
        }
 
-       wait_event_idle(thread->t_ctl_waitq,
-                       thread_is_running(thread) || thread_is_stopped(thread));
+       rc = lu_env_add_task(&args->qra_env, task);
+       if (rc) {
+               kthread_stop(task);
+               GOTO(out_env_fini, rc);
+       }
+       qmt->qmt_reba_task = task;
+       wake_up_process(task);
+       wait_for_completion(&started);
 
        RETURN(0);
+out_env_fini:
+       lu_env_fini(&args->qra_env);
+out_env:
+       OBD_FREE_PTR(args);
+       RETURN(rc);
 }
 
 /*
@@ -830,14 +841,15 @@ int qmt_start_reba_thread(struct qmt_device *qmt)
  */
 void qmt_stop_reba_thread(struct qmt_device *qmt)
 {
-       struct ptlrpc_thread *thread = &qmt->qmt_reba_thread;
+       struct task_struct *task;
 
-       if (!thread_is_stopped(thread)) {
+       spin_lock(&qmt->qmt_reba_lock);
+       task = qmt->qmt_reba_task;
+       qmt->qmt_reba_task = NULL;
+       spin_unlock(&qmt->qmt_reba_lock);
 
-               thread_set_flags(thread, SVC_STOPPING);
-               wake_up(&thread->t_ctl_waitq);
+       if (task)
+               kthread_stop(task);
 
-               wait_event_idle(thread->t_ctl_waitq, thread_is_stopped(thread));
-       }
        LASSERT(list_empty(&qmt->qmt_reba_list));
 }