b=24320 do not fork a new thread in mem pressure

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

diff --git a/lustre/ldlm/ldlm_lockd.c b/lustre/ldlm/ldlm_lockd.c
index 98f038c..f5fc1f8 100644 (file)
--- a/lustre/ldlm/ldlm_lockd.c
+++ b/lustre/ldlm/ldlm_lockd.c
@@ -827,11 +827,11 @@ int ldlm_server_completion_ast(struct ldlm_lock *lock, int flags, void *data)
         if (req == NULL)
                 RETURN(-ENOMEM);
 
-        lock_res_and_lock(lock);
-        if (lock->l_resource->lr_lvb_len)
+        /* server namespace, doesn't need lock */
+        if (lock->l_resource->lr_lvb_len) {
                  req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_CLIENT,
                                       lock->l_resource->lr_lvb_len);
-        unlock_res_and_lock(lock);
+        }
 
         rc = ptlrpc_request_pack(req, LUSTRE_DLM_VERSION, LDLM_CP_CALLBACK);
         if (rc) {
@@ -851,10 +851,10 @@ int ldlm_server_completion_ast(struct ldlm_lock *lock, int flags, void *data)
         if (lock->l_resource->lr_lvb_len) {
                 void *lvb = req_capsule_client_get(&req->rq_pill, &RMF_DLM_LVB);
 
-                lock_res_and_lock(lock);
+                cfs_down(&lock->l_resource->lr_lvb_sem);
                 memcpy(lvb, lock->l_resource->lr_lvb_data,
                        lock->l_resource->lr_lvb_len);
-                unlock_res_and_lock(lock);
+                cfs_up(&lock->l_resource->lr_lvb_sem);
         }
 
         LDLM_DEBUG(lock, "server preparing completion AST (after %lds wait)",
@@ -938,10 +938,9 @@ int ldlm_server_glimpse_ast(struct ldlm_lock *lock, void *data)
         body->lock_handle[0] = lock->l_remote_handle;
         ldlm_lock2desc(lock, &body->lock_desc);
 
-        lock_res_and_lock(lock);
+        /* server namespace, doesn't need lock */
         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
                              lock->l_resource->lr_lvb_len);
-        unlock_res_and_lock(lock);
         res = lock->l_resource;
         ptlrpc_request_set_replen(req);
 
@@ -1116,10 +1115,12 @@ int ldlm_handle_enqueue0(struct ldlm_namespace *ns,
         LDLM_DEBUG(lock, "server-side enqueue handler, new lock created");
 
         OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_ENQUEUE_BLOCKED, obd_timeout * 2);
-        /* Don't enqueue a lock onto the export if it has already
-         * been evicted.  Cancel it now instead. (bug 3822) */
-        if (req->rq_export->exp_failed) {
-                LDLM_ERROR(lock, "lock on destroyed export %p", req->rq_export);
+        /* Don't enqueue a lock onto the export if it is been disonnected
+         * due to eviction (bug 3822) or server umount (bug 24324).
+         * Cancel it now instead. */
+        if (req->rq_export->exp_disconnected) {
+                LDLM_ERROR(lock, "lock on disconnected export %p",
+                           req->rq_export);
                 GOTO(out, rc = -ENOTCONN);
         }
 
@@ -1136,13 +1137,11 @@ existing_lock:
                  * local_lock_enqueue by the policy function. */
                 cookie = req;
         } else {
-                lock_res_and_lock(lock);
                 if (lock->l_resource->lr_lvb_len) {
                         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB,
                                              RCL_SERVER,
                                              lock->l_resource->lr_lvb_len);
                 }
-                unlock_res_and_lock(lock);
 
                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE_EXTENT_ERR))
                         GOTO(out, rc = -ENOMEM);
@@ -1176,9 +1175,10 @@ existing_lock:
         dlm_rep->lock_flags |= dlm_req->lock_flags & LDLM_INHERIT_FLAGS;
         lock->l_flags |= dlm_req->lock_flags & LDLM_INHERIT_FLAGS;
 
-        /* Don't move a pending lock onto the export if it has already
-         * been evicted.  Cancel it now instead. (bug 5683) */
-        if (unlikely(req->rq_export->exp_failed ||
+        /* Don't move a pending lock onto the export if it has already been
+         * disconnected due to eviction (bug 5683) or server umount (bug 24324).
+         * Cancel it now instead. */
+        if (unlikely(req->rq_export->exp_disconnected ||
                      OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE_OLD_EXPORT))) {
                 LDLM_ERROR(lock, "lock on destroyed export %p", req->rq_export);
                 rc = -ENOTCONN;
@@ -1242,7 +1242,6 @@ existing_lock:
                 LDLM_DEBUG(lock, "server-side enqueue handler, sending reply"
                            "(err=%d, rc=%d)", err, rc);
 
-                lock_res_and_lock(lock);
                 if (rc == 0) {
                         if (lock->l_resource->lr_lvb_len > 0) {
                                 void *lvb;
@@ -1252,14 +1251,17 @@ existing_lock:
                                 LASSERTF(lvb != NULL, "req %p, lock %p\n",
                                          req, lock);
 
+                                cfs_down(&lock->l_resource->lr_lvb_sem);
                                 memcpy(lvb, lock->l_resource->lr_lvb_data,
                                        lock->l_resource->lr_lvb_len);
+                                cfs_up(&lock->l_resource->lr_lvb_sem);
                         }
                 } else {
+                        lock_res_and_lock(lock);
                         ldlm_resource_unlink_lock(lock);
                         ldlm_lock_destroy_nolock(lock);
+                        unlock_res_and_lock(lock);
                 }
-                unlock_res_and_lock(lock);
 
                 if (!err && dlm_req->lock_desc.l_resource.lr_type != LDLM_FLOCK)
                         ldlm_reprocess_all(lock->l_resource);
@@ -2216,7 +2218,8 @@ static int ldlm_bl_thread_main(void *arg)
 
                         /* 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)))
+                            busy >= cfs_atomic_read(&blp->blp_num_threads) &&
+                            !blwi->blwi_mem_pressure))
                                 /* discard the return value, we tried */
                                 ldlm_bl_thread_start(blp);
                 } else {
@@ -2307,10 +2310,9 @@ static void *
 ldlm_export_lock_key(cfs_hlist_node_t *hnode)
 {
         struct ldlm_lock *lock;
-        ENTRY;
 
         lock = cfs_hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
-        RETURN(&lock->l_remote_handle);
+        return &lock->l_remote_handle;
 }
 
 static void
@@ -2325,8 +2327,7 @@ ldlm_export_lock_keycpy(cfs_hlist_node_t *hnode, void *key)
 static int
 ldlm_export_lock_keycmp(void *key, cfs_hlist_node_t *hnode)
 {
-        ENTRY;
-        RETURN(lustre_handle_equal(ldlm_export_lock_key(hnode), key));
+        return lustre_handle_equal(ldlm_export_lock_key(hnode), key);
 }
 
 static void *
@@ -2335,28 +2336,22 @@ ldlm_export_lock_object(cfs_hlist_node_t *hnode)
         return cfs_hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
 }
 
-static void *
-ldlm_export_lock_get(cfs_hlist_node_t *hnode)
+static void
+ldlm_export_lock_get(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
 {
         struct ldlm_lock *lock;
-        ENTRY;
 
         lock = cfs_hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
         LDLM_LOCK_GET(lock);
-
-        RETURN(lock);
 }
 
-static void *
-ldlm_export_lock_put(cfs_hlist_node_t *hnode)
+static void
+ldlm_export_lock_put(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
 {
         struct ldlm_lock *lock;
-        ENTRY;
 
         lock = cfs_hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
         LDLM_LOCK_RELEASE(lock);
-
-        RETURN(lock);
 }
 
 static cfs_hash_ops_t ldlm_export_lock_ops = {
@@ -2487,11 +2482,11 @@ static int ldlm_setup(void)
                         GOTO(out_thread, rc);
         }
 
-        rc = ptlrpc_start_threads(NULL, ldlm_state->ldlm_cancel_service);
+        rc = ptlrpc_start_threads(ldlm_state->ldlm_cancel_service);
         if (rc)
                 GOTO(out_thread, rc);
 
-        rc = ptlrpc_start_threads(NULL, ldlm_state->ldlm_cb_service);
+        rc = ptlrpc_start_threads(ldlm_state->ldlm_cb_service);
         if (rc)
                 GOTO(out_thread, rc);
 
@@ -2690,7 +2685,6 @@ EXPORT_SYMBOL(ldlm_cli_cancel_req);
 EXPORT_SYMBOL(ldlm_replay_locks);
 EXPORT_SYMBOL(ldlm_resource_foreach);
 EXPORT_SYMBOL(ldlm_namespace_foreach);
-EXPORT_SYMBOL(ldlm_namespace_foreach_res);
 EXPORT_SYMBOL(ldlm_resource_iterate);
 EXPORT_SYMBOL(ldlm_cancel_resource_local);
 EXPORT_SYMBOL(ldlm_cli_cancel_list_local);