LU-2624 ptlrpc: improve stop of ptlrpcd threads

This fix improves the stop of ptlrpcd threads so that
it occurs in parallel. On machines with a large number
of cores this significantly reduce the time of unload of
ptlrpc module.

Signed-off-by: Gregoire Pichon <gregoire.pichon@bull.net>
Change-Id: I744de5d9887bfaa4c7a742d34923680e3a536ee7
Reviewed-on: http://review.whamcloud.com/5039
Tested-by: Hudson
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Faccini Bruno <bruno.faccini@intel.com>
Reviewed-by: Niu Yawei <yawei.niu@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/include/lustre_net.h b/lustre/include/lustre_net.h
index dbeea10..d3a51f4 100644 (file)
--- a/lustre/include/lustre_net.h
+++ b/lustre/include/lustre_net.h
@@ -3045,6 +3045,7 @@ typedef enum {
 
 /* ptlrpc/ptlrpcd.c */
 void ptlrpcd_stop(struct ptlrpcd_ctl *pc, int force);
 
 /* ptlrpc/ptlrpcd.c */
 void ptlrpcd_stop(struct ptlrpcd_ctl *pc, int force);

+void ptlrpcd_free(struct ptlrpcd_ctl *pc);

 void ptlrpcd_wake(struct ptlrpc_request *req);
 void ptlrpcd_add_req(struct ptlrpc_request *req, pdl_policy_t policy, int idx);
 void ptlrpcd_add_rqset(struct ptlrpc_request_set *set);
 void ptlrpcd_wake(struct ptlrpc_request *req);
 void ptlrpcd_add_req(struct ptlrpc_request *req, pdl_policy_t policy, int idx);
 void ptlrpcd_add_rqset(struct ptlrpc_request_set *set);

diff --git a/lustre/ptlrpc/ptlrpcd.c b/lustre/ptlrpc/ptlrpcd.c
index 8a842cb..669b0d7 100644 (file)
--- a/lustre/ptlrpc/ptlrpcd.c
+++ b/lustre/ptlrpc/ptlrpcd.c
@@ -479,11 +479,6 @@ static int ptlrpcd(void *arg)
                 ptlrpc_set_wait(set);
         lu_context_fini(&env.le_ctx);
 
                 ptlrpc_set_wait(set);
         lu_context_fini(&env.le_ctx);
 

-       clear_bit(LIOD_START, &pc->pc_flags);
-       clear_bit(LIOD_STOP, &pc->pc_flags);
-       clear_bit(LIOD_FORCE, &pc->pc_flags);
-       clear_bit(LIOD_BIND, &pc->pc_flags);
-

        complete(&pc->pc_finishing);
 
         return 0;
        complete(&pc->pc_finishing);
 
         return 0;


@@ -758,18 +753,32 @@ out:
 
 void ptlrpcd_stop(struct ptlrpcd_ctl *pc, int force)
 {
 
 void ptlrpcd_stop(struct ptlrpcd_ctl *pc, int force)
 {

-       struct ptlrpc_request_set *set = pc->pc_set;
-        ENTRY;
+       ENTRY;

 
        if (!test_bit(LIOD_START, &pc->pc_flags)) {
 
        if (!test_bit(LIOD_START, &pc->pc_flags)) {

-                CWARN("Thread for pc %p was not started\n", pc);
-                goto out;
-        }
+               CWARN("Thread for pc %p was not started\n", pc);
+               goto out;
+       }

 
        set_bit(LIOD_STOP, &pc->pc_flags);
        if (force)
                set_bit(LIOD_FORCE, &pc->pc_flags);
        cfs_waitq_signal(&pc->pc_set->set_waitq);
 
        set_bit(LIOD_STOP, &pc->pc_flags);
        if (force)
                set_bit(LIOD_FORCE, &pc->pc_flags);
        cfs_waitq_signal(&pc->pc_set->set_waitq);

+
+out:
+       EXIT;
+}
+
+void ptlrpcd_free(struct ptlrpcd_ctl *pc)
+{
+       struct ptlrpc_request_set *set = pc->pc_set;
+       ENTRY;
+
+       if (!test_bit(LIOD_START, &pc->pc_flags)) {
+               CWARN("Thread for pc %p was not started\n", pc);
+               goto out;
+       }
+

 #ifdef __KERNEL__
        wait_for_completion(&pc->pc_finishing);
 #else
 #ifdef __KERNEL__
        wait_for_completion(&pc->pc_finishing);
 #else


@@ -783,6 +792,11 @@ void ptlrpcd_stop(struct ptlrpcd_ctl *pc, int force)
        spin_unlock(&pc->pc_lock);
        ptlrpc_set_destroy(set);
 
        spin_unlock(&pc->pc_lock);
        ptlrpc_set_destroy(set);
 

+       clear_bit(LIOD_START, &pc->pc_flags);
+       clear_bit(LIOD_STOP, &pc->pc_flags);
+       clear_bit(LIOD_FORCE, &pc->pc_flags);
+       clear_bit(LIOD_BIND, &pc->pc_flags);
+

 out:
 #ifdef __KERNEL__
         if (pc->pc_npartners > 0) {
 out:
 #ifdef __KERNEL__
         if (pc->pc_npartners > 0) {


@@ -799,18 +813,21 @@ out:
 
 static void ptlrpcd_fini(void)
 {
 
 static void ptlrpcd_fini(void)
 {

-        int i;
-        ENTRY;
+       int i;
+       ENTRY;

 
 

-        if (ptlrpcds != NULL) {
-                for (i = 0; i < ptlrpcds->pd_nthreads; i++)
-                        ptlrpcd_stop(&ptlrpcds->pd_threads[i], 0);
-                ptlrpcd_stop(&ptlrpcds->pd_thread_rcv, 0);
-                OBD_FREE(ptlrpcds, ptlrpcds->pd_size);
-                ptlrpcds = NULL;
-        }
+       if (ptlrpcds != NULL) {
+               for (i = 0; i < ptlrpcds->pd_nthreads; i++)
+                       ptlrpcd_stop(&ptlrpcds->pd_threads[i], 0);
+               for (i = 0; i < ptlrpcds->pd_nthreads; i++)
+                       ptlrpcd_free(&ptlrpcds->pd_threads[i]);
+               ptlrpcd_stop(&ptlrpcds->pd_thread_rcv, 0);
+               ptlrpcd_free(&ptlrpcds->pd_thread_rcv);
+               OBD_FREE(ptlrpcds, ptlrpcds->pd_size);
+               ptlrpcds = NULL;
+       }

 
 

-        EXIT;
+       EXIT;

 }
 
 static int ptlrpcd_init(void)
 }
 
 static int ptlrpcd_init(void)


@@ -871,7 +888,10 @@ out:
         if (rc != 0 && ptlrpcds != NULL) {
                 for (j = 0; j <= i; j++)
                         ptlrpcd_stop(&ptlrpcds->pd_threads[j], 0);
         if (rc != 0 && ptlrpcds != NULL) {
                 for (j = 0; j <= i; j++)
                         ptlrpcd_stop(&ptlrpcds->pd_threads[j], 0);

-                ptlrpcd_stop(&ptlrpcds->pd_thread_rcv, 0);
+               for (j = 0; j <= i; j++)
+                       ptlrpcd_free(&ptlrpcds->pd_threads[j]);
+               ptlrpcd_stop(&ptlrpcds->pd_thread_rcv, 0);
+               ptlrpcd_free(&ptlrpcds->pd_thread_rcv);

                 OBD_FREE(ptlrpcds, size);
                 ptlrpcds = NULL;
         }
                 OBD_FREE(ptlrpcds, size);
                 ptlrpcds = NULL;
         }

diff --git a/lustre/ptlrpc/recov_thread.c b/lustre/ptlrpc/recov_thread.c
index 63fef2b..dee77cd 100644 (file)
--- a/lustre/ptlrpc/recov_thread.c
+++ b/lustre/ptlrpc/recov_thread.c
@@ -425,6 +425,7 @@ void llog_recov_thread_stop(struct llog_commit_master *lcm, int force)
          * for processing now.
          */
         ptlrpcd_stop(&lcm->lcm_pc, force);
          * for processing now.
          */
         ptlrpcd_stop(&lcm->lcm_pc, force);

+       ptlrpcd_free(&lcm->lcm_pc);

 
         /*
          * By this point no alive inflight llcds should be left. Only
 
         /*
          * By this point no alive inflight llcds should be left. Only