LU-947 ptlrpc: allow stopping threads above threads_max

If a service "threads_max" parameter is set below the number of
running threads, stop each highest-numbered running thread until
the running thread count is below threads_max.  Stopping nly the
last thread ensures the thread t_id numbers are always contiguous
rather than having gaps.  If the threads are started again they
will again be assigned contiguous t_id values.

Each thread is stopped only after it has finished processing an
incoming request, so running threads may not immediately stop
when the tunable is changed.

Also fix function declarations in file to match proper coding style.

Signed-off-by: Andreas Dilger <adilger@whamcloud.com>
Change-Id: I106f841e62c26b488ae837564c858a44263ebbe5
Reviewed-on: https://review.whamcloud.com/34400
Tested-by: Jenkins
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Wang Shilong <wshilong@ddn.com>
Reviewed-by: Hongchao Zhang <hongchao@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/ptlrpc/service.c b/lustre/ptlrpc/service.c
index 66efefd..c241279 100644 (file)
--- a/lustre/ptlrpc/service.c
+++ b/lustre/ptlrpc/service.c
@@ -98,8 +98,7 @@ ptlrpc_alloc_rqbd(struct ptlrpc_service_part *svcpt)
        return rqbd;
 }
 
-static void
-ptlrpc_free_rqbd(struct ptlrpc_request_buffer_desc *rqbd)
+static void ptlrpc_free_rqbd(struct ptlrpc_request_buffer_desc *rqbd)
 {
        struct ptlrpc_service_part *svcpt = rqbd->rqbd_svcpt;
 
@@ -115,13 +114,12 @@ ptlrpc_free_rqbd(struct ptlrpc_request_buffer_desc *rqbd)
        OBD_FREE_PTR(rqbd);
 }
 
-static int
-ptlrpc_grow_req_bufs(struct ptlrpc_service_part *svcpt, int post)
+static int ptlrpc_grow_req_bufs(struct ptlrpc_service_part *svcpt, int post)
 {
-       struct ptlrpc_service             *svc = svcpt->scp_service;
-        struct ptlrpc_request_buffer_desc *rqbd;
-        int                                rc = 0;
-        int                                i;
+       struct ptlrpc_service *svc = svcpt->scp_service;
+       struct ptlrpc_request_buffer_desc *rqbd;
+       int rc = 0;
+       int i;
 
        if (svcpt->scp_rqbd_allocating)
                goto try_post;
@@ -180,9 +178,8 @@ ptlrpc_grow_req_bufs(struct ptlrpc_service_part *svcpt, int post)
  * Part of Rep-Ack logic.
  * Puts a lock and its mode into reply state assotiated to request reply.
  */
-void
-ptlrpc_save_lock(struct ptlrpc_request *req, struct lustre_handle *lock,
-                int mode, bool no_ack, bool convert_lock)
+void ptlrpc_save_lock(struct ptlrpc_request *req, struct lustre_handle *lock,
+                     int mode, bool no_ack, bool convert_lock)
 {
        struct ptlrpc_reply_state *rs = req->rq_reply_state;
        int idx;
@@ -268,8 +265,8 @@ static void rs_batch_init(struct rs_batch *b)
 /**
  * Choose an hr thread to dispatch requests to.
  */
-static struct ptlrpc_hr_thread *
-ptlrpc_hr_select(struct ptlrpc_service_part *svcpt)
+static
+struct ptlrpc_hr_thread *ptlrpc_hr_select(struct ptlrpc_service_part *svcpt)
 {
        struct ptlrpc_hr_partition      *hrp;
        unsigned int                    rotor;
@@ -381,8 +378,7 @@ void ptlrpc_dispatch_difficult_reply(struct ptlrpc_reply_state *rs)
        EXIT;
 }
 
-void
-ptlrpc_schedule_difficult_reply(struct ptlrpc_reply_state *rs)
+void ptlrpc_schedule_difficult_reply(struct ptlrpc_reply_state *rs)
 {
        ENTRY;
 
@@ -430,8 +426,7 @@ void ptlrpc_commit_replies(struct obd_export *exp)
        EXIT;
 }
 
-static int
-ptlrpc_server_post_idle_rqbds(struct ptlrpc_service_part *svcpt)
+static int ptlrpc_server_post_idle_rqbds(struct ptlrpc_service_part *svcpt)
 {
        struct ptlrpc_request_buffer_desc *rqbd;
        int                               rc;
@@ -488,9 +483,8 @@ static void ptlrpc_at_timer(cfs_timer_cb_arg_t data)
        wake_up(&svcpt->scp_waitq);
 }
 
-static void
-ptlrpc_server_nthreads_check(struct ptlrpc_service *svc,
-                            struct ptlrpc_service_conf *conf)
+static void ptlrpc_server_nthreads_check(struct ptlrpc_service *svc,
+                                        struct ptlrpc_service_conf *conf)
 {
        struct ptlrpc_service_thr_conf  *tc = &conf->psc_thr;
        unsigned                        init;
@@ -593,9 +587,8 @@ ptlrpc_server_nthreads_check(struct ptlrpc_service *svc,
 /**
  * Initialize percpt data for a service
  */
-static int
-ptlrpc_service_part_init(struct ptlrpc_service *svc,
-                        struct ptlrpc_service_part *svcpt, int cpt)
+static int ptlrpc_service_part_init(struct ptlrpc_service *svc,
+                                   struct ptlrpc_service_part *svcpt, int cpt)
 {
        struct ptlrpc_at_array  *array;
        int                     size;
@@ -687,10 +680,9 @@ ptlrpc_service_part_init(struct ptlrpc_service *svc,
  * This includes starting serving threads , allocating and posting rqbds and
  * so on.
  */
-struct ptlrpc_service *
-ptlrpc_register_service(struct ptlrpc_service_conf *conf,
-                       struct kset *parent,
-                       struct dentry *debugfs_entry)
+struct ptlrpc_service *ptlrpc_register_service(struct ptlrpc_service_conf *conf,
+                                              struct kset *parent,
+                                              struct dentry *debugfs_entry)
 {
        struct ptlrpc_service_cpt_conf  *cconf = &conf->psc_cpt;
        struct ptlrpc_service           *service;
@@ -1274,8 +1266,7 @@ static int ptlrpc_at_add_timed(struct ptlrpc_request *req)
        return 0;
 }
 
-static void
-ptlrpc_at_remove_timed(struct ptlrpc_request *req)
+static void ptlrpc_at_remove_timed(struct ptlrpc_request *req)
 {
        struct ptlrpc_at_array *array;
 
@@ -1567,7 +1558,7 @@ static int ptlrpc_at_check_timed(struct ptlrpc_service_part *svcpt)
 
 /* Check if we are already handling earlier incarnation of this request.
  * Called under &req->rq_export->exp_rpc_lock locked */
-static struct ptlrpc_request*
+static struct ptlrpc_request *
 ptlrpc_server_check_resend_in_progress(struct ptlrpc_request *req)
 {
        struct ptlrpc_request   *tmp = NULL;
@@ -1848,8 +1839,9 @@ static bool ptlrpc_server_normal_pending(struct ptlrpc_service_part *svcpt,
  * \see ptlrpc_server_allow_normal
  * \see ptlrpc_server_allow high
  */
-static inline bool
-ptlrpc_server_request_pending(struct ptlrpc_service_part *svcpt, bool force)
+static inline
+bool ptlrpc_server_request_pending(struct ptlrpc_service_part *svcpt,
+                                  bool force)
 {
        return ptlrpc_server_high_pending(svcpt, force) ||
               ptlrpc_server_normal_pending(svcpt, force);
@@ -1906,9 +1898,8 @@ got_request:
  * All incoming requests pass through here before getting into
  * ptlrpc_server_handle_req later on.
  */
-static int
-ptlrpc_server_handle_req_in(struct ptlrpc_service_part *svcpt,
-                           struct ptlrpc_thread *thread)
+static int ptlrpc_server_handle_req_in(struct ptlrpc_service_part *svcpt,
+                                      struct ptlrpc_thread *thread)
 {
        struct ptlrpc_service   *svc = svcpt->scp_service;
        struct ptlrpc_request   *req;
@@ -2070,9 +2061,8 @@ err_req:
  * Main incoming request handling logic.
  * Calls handler function from service to do actual processing.
  */
-static int
-ptlrpc_server_handle_request(struct ptlrpc_service_part *svcpt,
-                            struct ptlrpc_thread *thread)
+static int ptlrpc_server_handle_request(struct ptlrpc_service_part *svcpt,
+                                       struct ptlrpc_thread *thread)
 {
        struct ptlrpc_service *svc = svcpt->scp_service;
        struct ptlrpc_request *request;
@@ -2221,8 +2211,7 @@ put_conn:
 /**
  * An internal function to process a single reply state object.
  */
-static int
-ptlrpc_handle_rs(struct ptlrpc_reply_state *rs)
+static int ptlrpc_handle_rs(struct ptlrpc_reply_state *rs)
 {
        struct ptlrpc_service_part *svcpt = rs->rs_svcpt;
        struct ptlrpc_service     *svc = svcpt->scp_service;
@@ -2371,8 +2360,7 @@ ptlrpc_handle_rs(struct ptlrpc_reply_state *rs)
 }
 
 
-static void
-ptlrpc_check_rqbd_pool(struct ptlrpc_service_part *svcpt)
+static void ptlrpc_check_rqbd_pool(struct ptlrpc_service_part *svcpt)
 {
        int avail = svcpt->scp_nrqbds_posted;
        int low_water = test_req_buffer_pressure ? 0 :
@@ -2394,8 +2382,7 @@ ptlrpc_check_rqbd_pool(struct ptlrpc_service_part *svcpt)
        }
 }
 
-static int
-ptlrpc_retry_rqbds(void *arg)
+static int ptlrpc_retry_rqbds(void *arg)
 {
        struct ptlrpc_service_part *svcpt = (struct ptlrpc_service_part *)arg;
 
@@ -2403,8 +2390,7 @@ ptlrpc_retry_rqbds(void *arg)
        return -ETIMEDOUT;
 }
 
-static inline int
-ptlrpc_threads_enough(struct ptlrpc_service_part *svcpt)
+static inline int ptlrpc_threads_enough(struct ptlrpc_service_part *svcpt)
 {
        return svcpt->scp_nreqs_active <
               svcpt->scp_nthrs_running - 1 -
@@ -2416,8 +2402,7 @@ ptlrpc_threads_enough(struct ptlrpc_service_part *svcpt)
  * user can call it w/o any lock but need to hold
  * ptlrpc_service_part::scp_lock to get reliable result
  */
-static inline int
-ptlrpc_threads_increasable(struct ptlrpc_service_part *svcpt)
+static inline int ptlrpc_threads_increasable(struct ptlrpc_service_part *svcpt)
 {
        return svcpt->scp_nthrs_running +
               svcpt->scp_nthrs_starting <
@@ -2427,22 +2412,47 @@ ptlrpc_threads_increasable(struct ptlrpc_service_part *svcpt)
 /**
  * too many requests and allowed to create more threads
  */
-static inline int
-ptlrpc_threads_need_create(struct ptlrpc_service_part *svcpt)
+static inline int ptlrpc_threads_need_create(struct ptlrpc_service_part *svcpt)
 {
        return !ptlrpc_threads_enough(svcpt) &&
                ptlrpc_threads_increasable(svcpt);
 }
 
-static inline int
-ptlrpc_thread_stopping(struct ptlrpc_thread *thread)
+static inline int ptlrpc_thread_stopping(struct ptlrpc_thread *thread)
 {
        return thread_is_stopping(thread) ||
               thread->t_svcpt->scp_service->srv_is_stopping;
 }
 
-static inline int
-ptlrpc_rqbd_pending(struct ptlrpc_service_part *svcpt)
+/* stop the highest numbered thread if there are too many threads running */
+static inline bool ptlrpc_thread_should_stop(struct ptlrpc_thread *thread)
+{
+       struct ptlrpc_service_part *svcpt = thread->t_svcpt;
+
+       return thread->t_id >= svcpt->scp_service->srv_nthrs_cpt_limit &&
+               thread->t_id == svcpt->scp_thr_nextid - 1;
+}
+
+static void ptlrpc_stop_thread(struct ptlrpc_thread *thread)
+{
+       CDEBUG(D_INFO, "Stopping thread %s #%u\n",
+              thread->t_svcpt->scp_service->srv_thread_name, thread->t_id);
+       thread_add_flags(thread, SVC_STOPPING);
+}
+
+static inline void ptlrpc_thread_stop(struct ptlrpc_thread *thread)
+{
+       struct ptlrpc_service_part *svcpt = thread->t_svcpt;
+
+       spin_lock(&svcpt->scp_lock);
+       if (ptlrpc_thread_should_stop(thread)) {
+               ptlrpc_stop_thread(thread);
+               svcpt->scp_thr_nextid--;
+       }
+       spin_unlock(&svcpt->scp_lock);
+}
+
+static inline int ptlrpc_rqbd_pending(struct ptlrpc_service_part *svcpt)
 {
        return !list_empty(&svcpt->scp_rqbd_idle) &&
               svcpt->scp_rqbd_timeout == 0;
@@ -2693,25 +2703,30 @@ static int ptlrpc_main(void *arg)
                        svcpt->scp_rqbd_timeout = cfs_time_seconds(1) / 10;
                        CDEBUG(D_RPCTRACE, "Posted buffers: %d\n",
                               svcpt->scp_nrqbds_posted);
-                }
-        }
+               }
+
+               /* If the number of threads has been tuned downward and this
+                * thread should be stopped, then stop in reverse order so the
+                * the threads always have contiguous thread index values.
+                */
+               if (unlikely(ptlrpc_thread_should_stop(thread)))
+                       ptlrpc_thread_stop(thread);
+       }
 
        ptlrpc_watchdog_disable(&thread->t_watchdog);
 
 out_srv_fini:
-        /*
-         * deconstruct service specific state created by ptlrpc_start_thread()
-         */
+       /* deconstruct service thread state created by ptlrpc_start_thread() */
        if (svc->srv_ops.so_thr_done != NULL)
                svc->srv_ops.so_thr_done(thread);
 
-        if (env != NULL) {
-                lu_context_fini(&env->le_ctx);
-                OBD_FREE_PTR(env);
-        }
+       if (env != NULL) {
+               lu_context_fini(&env->le_ctx);
+               OBD_FREE_PTR(env);
+       }
 out:
-        CDEBUG(D_RPCTRACE, "service thread [ %p : %u ] %d exiting: rc %d\n",
-               thread, thread->t_pid, thread->t_id, rc);
+       CDEBUG(D_RPCTRACE, "%s: service thread [%p:%u] %d exiting: rc = %d\n",
+              thread->t_name, thread, thread->t_pid, thread->t_id, rc);
 
        spin_lock(&svcpt->scp_lock);
        if (thread_test_and_clear_flags(thread, SVC_STARTING))
@@ -2869,11 +2884,8 @@ static void ptlrpc_svcpt_stop_threads(struct ptlrpc_service_part *svcpt)
        INIT_LIST_HEAD(&zombie);
        spin_lock(&svcpt->scp_lock);
        /* let the thread know that we would like it to stop asap */
-       list_for_each_entry(thread, &svcpt->scp_threads, t_link) {
-               CDEBUG(D_INFO, "Stopping thread %s #%u\n",
-                      svcpt->scp_service->srv_thread_name, thread->t_id);
-               thread_add_flags(thread, SVC_STOPPING);
-       }
+       list_for_each_entry(thread, &svcpt->scp_threads, t_link)
+               ptlrpc_stop_thread(thread);
 
        wake_up_all(&svcpt->scp_waitq);