LU-874 ldlm: Fix ldlm_bl_* thread creation

[fs/lustre-release.git] / lustre / ldlm / ldlm_lockd.c

diff --git a/lustre/ldlm/ldlm_lockd.c b/lustre/ldlm/ldlm_lockd.c
index 2990edc..2be5aa9 100644 (file)
--- a/lustre/ldlm/ldlm_lockd.c
+++ b/lustre/ldlm/ldlm_lockd.c
@@ -279,7 +279,7 @@ static int ldlm_lock_busy(struct ldlm_lock *lock)
 /* This is called from within a timer interrupt and cannot schedule */
 static void waiting_locks_callback(unsigned long unused)
 {
-        struct ldlm_lock *lock, *last = NULL;
+        struct ldlm_lock *lock;
 
 repeat:
         cfs_spin_lock_bh(&waiting_locks_spinlock);
@@ -355,8 +355,6 @@ repeat:
                            libcfs_nid2str(
                                    lock->l_export->exp_connection->c_peer.nid));
 
-                last = lock;
-
                 /* no needs to take an extra ref on the lock since it was in
                  * the waiting_locks_list and ldlm_add_waiting_lock()
                  * already grabbed a ref */
@@ -649,9 +647,9 @@ static int ldlm_handle_ast_error(struct ldlm_lock *lock,
 static int ldlm_cb_interpret(const struct lu_env *env,
                              struct ptlrpc_request *req, void *data, int rc)
 {
-        struct ldlm_cb_async_args *ca = data;
-        struct ldlm_cb_set_arg *arg = ca->ca_set_arg;
-        struct ldlm_lock *lock = ca->ca_lock;
+        struct ldlm_cb_async_args *ca   = data;
+        struct ldlm_lock          *lock = ca->ca_lock;
+        struct ldlm_cb_set_arg    *arg  = ca->ca_set_arg;
         ENTRY;
 
         LASSERT(lock != NULL);
@@ -659,17 +657,17 @@ static int ldlm_cb_interpret(const struct lu_env *env,
                 rc = ldlm_handle_ast_error(lock, req, rc,
                                            arg->type == LDLM_BL_CALLBACK
                                            ? "blocking" : "completion");
+                if (rc == -ERESTART)
+                        cfs_atomic_inc(&arg->restart);
         }
-
         LDLM_LOCK_RELEASE(lock);
 
-        if (rc == -ERESTART)
-                cfs_atomic_set(&arg->restart, 1);
-
+        if (cfs_atomic_dec_return(&arg->rpcs) < arg->threshold)
+                cfs_waitq_signal(&arg->waitq);
         RETURN(0);
 }
 
-static inline int ldlm_bl_and_cp_ast_fini(struct ptlrpc_request *req,
+static inline int ldlm_bl_and_cp_ast_tail(struct ptlrpc_request *req,
                                           struct ldlm_cb_set_arg *arg,
                                           struct ldlm_lock *lock,
                                           int instant_cancel)
@@ -681,12 +679,11 @@ static inline int ldlm_bl_and_cp_ast_fini(struct ptlrpc_request *req,
                 rc = ptl_send_rpc(req, 1);
                 ptlrpc_req_finished(req);
                 if (rc == 0)
-                        /* If we cancelled the lock, we need to restart
-                         * ldlm_reprocess_queue */
-                        cfs_atomic_set(&arg->restart, 1);
+                        cfs_atomic_inc(&arg->restart);
         } else {
                 LDLM_LOCK_GET(lock);
-                ptlrpc_set_add_req(arg->set, req);
+                ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
+                cfs_atomic_inc(&arg->rpcs);
         }
 
         RETURN(rc);
@@ -810,7 +807,7 @@ int ldlm_server_blocking_ast(struct ldlm_lock *lock,
                 lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
                                      LDLM_BL_CALLBACK - LDLM_FIRST_OPC);
 
-        rc = ldlm_bl_and_cp_ast_fini(req, arg, lock, instant_cancel);
+        rc = ldlm_bl_and_cp_ast_tail(req, arg, lock, instant_cancel);
 
         RETURN(rc);
 }
@@ -925,7 +922,7 @@ int ldlm_server_completion_ast(struct ldlm_lock *lock, int flags, void *data)
                 lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
                                      LDLM_CP_CALLBACK - LDLM_FIRST_OPC);
 
-        rc = ldlm_bl_and_cp_ast_fini(req, arg, lock, instant_cancel);
+        rc = ldlm_bl_and_cp_ast_tail(req, arg, lock, instant_cancel);
 
         RETURN(rc);
 }
@@ -1604,7 +1601,7 @@ static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
          * l_ast_data */
         OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_ENQ_RACE, 2);
 
-        ldlm_run_ast_work(&ast_list, LDLM_WORK_CP_AST);
+        ldlm_run_ast_work(ns, &ast_list, LDLM_WORK_CP_AST);
 
         LDLM_DEBUG_NOLOCK("client completion callback handler END (lock %p)",
                           lock);
@@ -2147,7 +2144,8 @@ void ldlm_revoke_export_locks(struct obd_export *exp)
         CFS_INIT_LIST_HEAD(&rpc_list);
         cfs_hash_for_each_empty(exp->exp_lock_hash,
                                 ldlm_revoke_lock_cb, &rpc_list);
-        ldlm_run_ast_work(&rpc_list, LDLM_WORK_REVOKE_AST);
+        ldlm_run_ast_work(exp->exp_obd->obd_namespace, &rpc_list,
+                          LDLM_WORK_REVOKE_AST);
 
         EXIT;
 }
@@ -2232,33 +2230,31 @@ static int ldlm_bl_thread_main(void *arg)
         while (1) {
                 struct l_wait_info lwi = { 0 };
                 struct ldlm_bl_work_item *blwi = NULL;
+                int busy;
 
                 blwi = ldlm_bl_get_work(blp);
 
                 if (blwi == NULL) {
-                        int busy;
-
                         cfs_atomic_dec(&blp->blp_busy_threads);
                         l_wait_event_exclusive(blp->blp_waitq,
                                          (blwi = ldlm_bl_get_work(blp)) != NULL,
                                          &lwi);
                         busy = cfs_atomic_inc_return(&blp->blp_busy_threads);
-
-                        if (blwi->blwi_ns == NULL)
-                                /* added by ldlm_cleanup() */
-                                break;
-
-                        /* Not fatal if racy and have a few too many threads */
-                        if (unlikely(busy < blp->blp_max_threads &&
-                            busy >= cfs_atomic_read(&blp->blp_num_threads) &&
-                            !blwi->blwi_mem_pressure))
-                                /* discard the return value, we tried */
-                                ldlm_bl_thread_start(blp);
                 } else {
-                        if (blwi->blwi_ns == NULL)
-                                /* added by ldlm_cleanup() */
-                                break;
+                        busy = cfs_atomic_read(&blp->blp_busy_threads);
                 }
+
+                if (blwi->blwi_ns == NULL)
+                        /* added by ldlm_cleanup() */
+                        break;
+
+                /* Not fatal if racy and have a few too many threads */
+                if (unlikely(busy < blp->blp_max_threads &&
+                             busy >= cfs_atomic_read(&blp->blp_num_threads) &&
+                             !blwi->blwi_mem_pressure))
+                        /* discard the return value, we tried */
+                        ldlm_bl_thread_start(blp);
+
                 if (blwi->blwi_mem_pressure)
                         cfs_memory_pressure_set();
 
@@ -2616,7 +2612,7 @@ static int ldlm_cleanup(void)
         RETURN(0);
 }
 
-int __init ldlm_init(void)
+int ldlm_init(void)
 {
         cfs_init_mutex(&ldlm_ref_sem);
         cfs_init_mutex(ldlm_namespace_lock(LDLM_NAMESPACE_SERVER));
@@ -2649,7 +2645,7 @@ int __init ldlm_init(void)
         return 0;
 }
 
-void __exit ldlm_exit(void)
+void ldlm_exit(void)
 {
         int rc;
         if (ldlm_refcount)