LU-12780 osp: don't use ptlrpc_thread for opd_update_thread

rather than ptlrpc_thread, use native kthreads functionality.

- There is no need to synchornized on startup - do all the startup
  that can fail before starting the thread.  This involves puting the
  lu_env in the per-thread struct osp_updates.

- Synchronization on shutdown is down with kthread_stop() and
  kthread_should_stop().  lu_env_put() call is moved to after
  kthread_stop() call, as it is theoretically possible that the
  thread function never gets called, so it isn't safe for it to
  be responsible for cleanup.

- opd_update_thread is replace with ou_update_task in struct
  osp_updates.

Signed-off-by: Mr NeilBrown <neilb@suse.de>
Change-Id: I78dafbadf419ee9b80a9bd0046152fb6f293f191
Reviewed-on: https://review.whamcloud.com/36264
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>
Reviewed-by: Alex Zhuravlev <bzzz@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/osp/osp_dev.c b/lustre/osp/osp_dev.c
index 4315731..20eef80 100644 (file)
--- a/lustre/osp/osp_dev.c
+++ b/lustre/osp/osp_dev.c
@@ -501,7 +501,8 @@ static int osp_disconnect(struct osp_device *d)
  */
 static int osp_update_init(struct osp_device *osp)
 {
  */
 static int osp_update_init(struct osp_device *osp)
 {

-       struct task_struct      *task;
+       struct task_struct *task;
+       int rc;

 
        ENTRY;
 
 
        ENTRY;
 


@@ -514,7 +515,6 @@ static int osp_update_init(struct osp_device *osp)
        if (osp->opd_update == NULL)
                RETURN(-ENOMEM);
 
        if (osp->opd_update == NULL)
                RETURN(-ENOMEM);
 

-       init_waitqueue_head(&osp->opd_update_thread.t_ctl_waitq);

        init_waitqueue_head(&osp->opd_update->ou_waitq);
        spin_lock_init(&osp->opd_update->ou_lock);
        INIT_LIST_HEAD(&osp->opd_update->ou_list);
        init_waitqueue_head(&osp->opd_update->ou_waitq);
        spin_lock_init(&osp->opd_update->ou_lock);
        INIT_LIST_HEAD(&osp->opd_update->ou_list);


@@ -522,12 +522,22 @@ static int osp_update_init(struct osp_device *osp)
        osp->opd_update->ou_version = 1;
        osp->opd_update->ou_generation = 0;
 
        osp->opd_update->ou_version = 1;
        osp->opd_update->ou_generation = 0;
 

+       rc = lu_env_init(&osp->opd_update->ou_env,
+                        osp->opd_dt_dev.dd_lu_dev.ld_type->ldt_ctx_tags);
+       if (rc < 0) {
+               CERROR("%s: init env error: rc = %d\n", osp->opd_obd->obd_name,
+                      rc);
+               OBD_FREE_PTR(osp->opd_update);
+               osp->opd_update = NULL;
+               RETURN(rc);
+       }

        /* start thread handling sending updates to the remote MDT */
        /* start thread handling sending updates to the remote MDT */

-       task = kthread_run(osp_send_update_thread, osp,
-                          "osp_up%u-%u", osp->opd_index, osp->opd_group);
+       task = kthread_create(osp_send_update_thread, osp,
+                             "osp_up%u-%u", osp->opd_index, osp->opd_group);

        if (IS_ERR(task)) {
                int rc = PTR_ERR(task);
 
        if (IS_ERR(task)) {
                int rc = PTR_ERR(task);
 

+               lu_env_fini(&osp->opd_update->ou_env);

                OBD_FREE_PTR(osp->opd_update);
                osp->opd_update = NULL;
                CERROR("%s: can't start precreate thread: rc = %d\n",
                OBD_FREE_PTR(osp->opd_update);
                osp->opd_update = NULL;
                CERROR("%s: can't start precreate thread: rc = %d\n",


@@ -535,9 +545,8 @@ static int osp_update_init(struct osp_device *osp)
                RETURN(rc);
        }
 
                RETURN(rc);
        }
 

-       wait_event_idle(osp->opd_update_thread.t_ctl_waitq,
-                       osp_send_update_thread_running(osp) ||
-                       osp_send_update_thread_stopped(osp));
+       osp->opd_update->ou_update_task = task;
+       wake_up_process(task);

 
        RETURN(0);
 }
 
        RETURN(0);
 }


@@ -559,11 +568,8 @@ static void osp_update_fini(const struct lu_env *env, struct osp_device *osp)
        if (ou == NULL)
                return;
 
        if (ou == NULL)
                return;
 

-       osp->opd_update_thread.t_flags = SVC_STOPPING;
-       wake_up(&ou->ou_waitq);
-
-       wait_event(osp->opd_update_thread.t_ctl_waitq,
-                  osp->opd_update_thread.t_flags & SVC_STOPPED);
+       kthread_stop(ou->ou_update_task);
+       lu_env_fini(&ou->ou_env);

 
        /* Remove the left osp thandle from the list */
        spin_lock(&ou->ou_lock);
 
        /* Remove the left osp thandle from the list */
        spin_lock(&ou->ou_lock);

diff --git a/lustre/osp/osp_internal.h b/lustre/osp/osp_internal.h
index 16ea991..02c153d 100644 (file)
--- a/lustre/osp/osp_internal.h
+++ b/lustre/osp/osp_internal.h
@@ -142,6 +142,10 @@ struct osp_updates {
         * those stale RPC(with older generation) will not be sent, otherwise it
         * will cause update lllog corruption */
        __u64                   ou_generation;
         * those stale RPC(with older generation) will not be sent, otherwise it
         * will cause update lllog corruption */
        __u64                   ou_generation;

+
+       /* dedicate update thread */
+       struct task_struct      *ou_update_task;
+       struct lu_env           ou_env;

 };
 
 struct osp_rpc_lock {
 };
 
 struct osp_rpc_lock {


@@ -201,8 +205,6 @@ struct osp_device {
 
        /* send update thread */
        struct osp_updates              *opd_update;
 
        /* send update thread */
        struct osp_updates              *opd_update;

-       /* dedicate update thread */
-       struct ptlrpc_thread             opd_update_thread;

 
        /*
         * OST synchronization thread
 
        /*
         * OST synchronization thread


@@ -738,16 +740,6 @@ int osp_object_update_request_create(struct osp_update_request *our,
        ret;                                                            \
 })
 
        ret;                                                            \
 })
 

-static inline bool osp_send_update_thread_running(struct osp_device *osp)
-{
-       return osp->opd_update_thread.t_flags & SVC_RUNNING;
-}
-
-static inline bool osp_send_update_thread_stopped(struct osp_device *osp)
-{
-       return osp->opd_update_thread.t_flags & SVC_STOPPED;
-}
-

 typedef int (*osp_update_interpreter_t)(const struct lu_env *env,
                                        struct object_update_reply *rep,
                                        struct ptlrpc_request *req,
 typedef int (*osp_update_interpreter_t)(const struct lu_env *env,
                                        struct object_update_reply *rep,
                                        struct ptlrpc_request *req,

diff --git a/lustre/osp/osp_trans.c b/lustre/osp/osp_trans.c
index f156b3a..f0cad02 100644 (file)
--- a/lustre/osp/osp_trans.c
+++ b/lustre/osp/osp_trans.c
@@ -1431,49 +1431,41 @@ void osp_invalidate_request(struct osp_device *osp)
  */
 int osp_send_update_thread(void *arg)
 {
  */
 int osp_send_update_thread(void *arg)
 {

-       struct lu_env           env;
+       struct lu_env           *env;

        struct osp_device       *osp = arg;
        struct osp_updates      *ou = osp->opd_update;
        struct osp_device       *osp = arg;
        struct osp_updates      *ou = osp->opd_update;

-       struct ptlrpc_thread    *thread = &osp->opd_update_thread;

        struct osp_update_request *our = NULL;
        int                     rc;
        ENTRY;
 
        LASSERT(ou != NULL);
        struct osp_update_request *our = NULL;
        int                     rc;
        ENTRY;
 
        LASSERT(ou != NULL);

-       rc = lu_env_init(&env, osp->opd_dt_dev.dd_lu_dev.ld_type->ldt_ctx_tags);
-       if (rc < 0) {
-               CERROR("%s: init env error: rc = %d\n", osp->opd_obd->obd_name,
-                      rc);
-               RETURN(rc);
-       }
+       env = &ou->ou_env;

 
 

-       thread->t_flags = SVC_RUNNING;
-       wake_up(&thread->t_ctl_waitq);

        while (1) {
                our = NULL;
                wait_event_idle(ou->ou_waitq,
        while (1) {
                our = NULL;
                wait_event_idle(ou->ou_waitq,

-                               !osp_send_update_thread_running(osp) ||
+                               kthread_should_stop() ||

                                osp_get_next_request(ou, &our));
 
                                osp_get_next_request(ou, &our));
 

-               if (!osp_send_update_thread_running(osp)) {
+               if (kthread_should_stop()) {

                        if (our != NULL) {
                        if (our != NULL) {

-                               osp_trans_callback(&env, our->our_th, -EINTR);
-                               osp_thandle_put(&env, our->our_th);
+                               osp_trans_callback(env, our->our_th, -EINTR);
+                               osp_thandle_put(env, our->our_th);

                        }
                        break;
                }
 
                LASSERT(our->our_th != NULL);
                if (our->our_th->ot_super.th_result != 0) {
                        }
                        break;
                }
 
                LASSERT(our->our_th != NULL);
                if (our->our_th->ot_super.th_result != 0) {

-                       osp_trans_callback(&env, our->our_th,
+                       osp_trans_callback(env, our->our_th,

                                our->our_th->ot_super.th_result);
                        rc = our->our_th->ot_super.th_result;
                } else if (ou->ou_generation != our->our_generation ||
                           OBD_FAIL_CHECK(OBD_FAIL_INVALIDATE_UPDATE)) {
                        rc = -EIO;
                                our->our_th->ot_super.th_result);
                        rc = our->our_th->ot_super.th_result;
                } else if (ou->ou_generation != our->our_generation ||
                           OBD_FAIL_CHECK(OBD_FAIL_INVALIDATE_UPDATE)) {
                        rc = -EIO;

-                       osp_trans_callback(&env, our->our_th, rc);
+                       osp_trans_callback(env, our->our_th, rc);

                } else {
                } else {

-                       rc = osp_send_update_req(&env, osp, our);
+                       rc = osp_send_update_req(env, osp, our);

                }
 
                /* Update the rpc version */
                }
 
                /* Update the rpc version */


@@ -1490,13 +1482,9 @@ int osp_send_update_thread(void *arg)
                        osp_invalidate_request(osp);
 
                /* Balanced for thandle_get in osp_check_and_set_rpc_version */
                        osp_invalidate_request(osp);
 
                /* Balanced for thandle_get in osp_check_and_set_rpc_version */

-               osp_thandle_put(&env, our->our_th);
+               osp_thandle_put(env, our->our_th);

        }
 
        }
 

-       thread->t_flags = SVC_STOPPED;
-       lu_env_fini(&env);
-       wake_up(&thread->t_ctl_waitq);
-

        RETURN(0);
 }
 
        RETURN(0);
 }
 


@@ -1576,8 +1564,7 @@ int osp_trans_stop(const struct lu_env *env, struct dt_device *dt,
                GOTO(out, rc);
        }
 
                GOTO(out, rc);
        }
 

-       if (osp->opd_update == NULL ||
-           !osp_send_update_thread_running(osp)) {
+       if (osp->opd_update == NULL) {

                osp_trans_callback(env, oth, -EIO);
                GOTO(out, rc = -EIO);
        }
                osp_trans_callback(env, oth, -EIO);
                GOTO(out, rc = -EIO);
        }