Whamcloud - gitweb
git://git.whamcloud.com / fs / lustre-release.git / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | tree
raw | inline | side by side
b=17323
[fs/lustre-release.git] / lustre / ldlm / ldlm_lockd.c
diff --git a/lustre/ldlm/ldlm_lockd.c b/lustre/ldlm/ldlm_lockd.c
index 870ae23..83b08d0 100644 (file)
--- a/lustre/ldlm/ldlm_lockd.c
+++ b/lustre/ldlm/ldlm_lockd.c
@@ -486,7 +486,14 @@ static void ldlm_failed_ast(struct ldlm_lock *lock, int rc,
 
         if (obd_dump_on_timeout)
                 libcfs_debug_dumplog();
+#ifdef __KERNEL__
+        spin_lock_bh(&waiting_locks_spinlock);
+        list_add(&lock->l_pending_chain, &expired_lock_thread.elt_expired_locks);
+        cfs_waitq_signal(&expired_lock_thread.elt_waitq);
+        spin_unlock_bh(&waiting_locks_spinlock);
+#else
         class_fail_export(lock->l_export);
+#endif
 }
 
 static int ldlm_handle_ast_error(struct ldlm_lock *lock,
@@ -536,7 +543,8 @@ static int ldlm_handle_ast_error(struct ldlm_lock *lock,
         return rc;
 }
 
-static int ldlm_cb_interpret(struct ptlrpc_request *req, void *data, int rc)
+static int ldlm_cb_interpret(const struct lu_env *env,
+                             struct ptlrpc_request *req, void *data, int rc)
 {
         struct ldlm_cb_set_arg *arg;
         struct ldlm_lock *lock;
@@ -548,18 +556,18 @@ static int ldlm_cb_interpret(struct ptlrpc_request *req, void *data, int rc)
         lock = req->rq_async_args.pointer_arg[1];
         LASSERT(lock != NULL);
         if (rc != 0) {
-                /* If client canceled the lock but the cancel has not 
+                /* If client canceled the lock but the cancel has not
                  * been recieved yet, we need to update lvbo to have the
                  * proper attributes cached. */
                 if (rc == -EINVAL && arg->type == LDLM_BL_CALLBACK)
-                        ldlm_res_lvbo_update(lock->l_resource, NULL, 
+                        ldlm_res_lvbo_update(lock->l_resource, NULL,
                                              0, 1);
-                rc = ldlm_handle_ast_error(lock, req, rc, 
+                rc = ldlm_handle_ast_error(lock, req, rc,
                                            arg->type == LDLM_BL_CALLBACK
                                            ? "blocking" : "completion");
-        }                
+        }
 
-        LDLM_LOCK_PUT(lock);
+        LDLM_LOCK_RELEASE(lock);
 
         if (rc == -ERESTART)
                 atomic_set(&arg->restart, 1);
@@ -585,7 +593,7 @@ static inline int ldlm_bl_and_cp_ast_fini(struct ptlrpc_request *req,
         } else {
                 LDLM_LOCK_GET(lock);
                 ptlrpc_set_add_req(arg->set, req);
-        }       
+        }
 
         RETURN(rc);
 }
@@ -672,8 +680,9 @@ int ldlm_server_blocking_ast(struct ldlm_lock *lock,
         if (AT_OFF)
                 req->rq_timeout = ldlm_get_rq_timeout();
 
-        if (lock->l_export && lock->l_export->exp_ldlm_stats)
-                lprocfs_counter_incr(lock->l_export->exp_ldlm_stats,
+        if (lock->l_export && lock->l_export->exp_nid_stats &&
+            lock->l_export->exp_nid_stats->nid_ldlm_stats)
+                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);
@@ -704,10 +713,10 @@ int ldlm_server_completion_ast(struct ldlm_lock *lock, int flags, void *data)
                 LDLM_ERROR(lock, "enqueue wait took %luus from "CFS_TIME_T,
                            total_enqueue_wait, lock->l_enqueued_time.tv_sec);
 
-         req = ptlrpc_request_alloc(lock->l_export->exp_imp_reverse,
+        req = ptlrpc_request_alloc(lock->l_export->exp_imp_reverse,
                                     &RQF_LDLM_CP_CALLBACK);
-         if (req == NULL)
-                 RETURN(-ENOMEM);
+        if (req == NULL)
+                RETURN(-ENOMEM);
 
         lock_res_and_lock(lock);
         if (lock->l_resource->lr_lvb_len)
@@ -777,8 +786,9 @@ int ldlm_server_completion_ast(struct ldlm_lock *lock, int flags, void *data)
         }
         unlock_res_and_lock(lock);
 
-        if (lock->l_export && lock->l_export->exp_ldlm_stats)
-                lprocfs_counter_incr(lock->l_export->exp_ldlm_stats,
+        if (lock->l_export && lock->l_export->exp_nid_stats &&
+            lock->l_export->exp_nid_stats->nid_ldlm_stats)
+                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);
@@ -820,8 +830,9 @@ int ldlm_server_glimpse_ast(struct ldlm_lock *lock, void *data)
         if (AT_OFF)
                 req->rq_timeout = ldlm_get_rq_timeout();
 
-        if (lock->l_export && lock->l_export->exp_ldlm_stats)
-                lprocfs_counter_incr(lock->l_export->exp_ldlm_stats,
+        if (lock->l_export && lock->l_export->exp_nid_stats &&
+            lock->l_export->exp_nid_stats->nid_ldlm_stats)
+                lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
                                      LDLM_GL_CALLBACK - LDLM_FIRST_OPC);
 
         rc = ptlrpc_queue_wait(req);
@@ -907,8 +918,9 @@ int ldlm_handle_enqueue0(struct ldlm_namespace *ns,
                 ldlm_svc_get_eopc(dlm_req,
                                   req->rq_rqbd->rqbd_service->srv_stats);
 
-        if (req->rq_export->exp_ldlm_stats)
-                lprocfs_counter_incr(req->rq_export->exp_ldlm_stats,
+        if (req->rq_export && req->rq_export->exp_nid_stats &&
+            req->rq_export->exp_nid_stats->nid_ldlm_stats)
+                lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
                                      LDLM_ENQUEUE - LDLM_FIRST_OPC);
 
         if (unlikely(dlm_req->lock_desc.l_resource.lr_type < LDLM_MIN_TYPE ||
@@ -973,8 +985,7 @@ int ldlm_handle_enqueue0(struct ldlm_namespace *ns,
         lock = ldlm_lock_create(ns, &dlm_req->lock_desc.l_resource.lr_name,
                                 dlm_req->lock_desc.l_resource.lr_type,
                                 dlm_req->lock_desc.l_req_mode,
-                                cbs->lcs_blocking, cbs->lcs_completion,
-                                cbs->lcs_glimpse, NULL, 0);
+                                cbs, NULL, 0);
 
         if (!lock)
                 GOTO(out, rc = -ENOMEM);
@@ -994,7 +1005,7 @@ int ldlm_handle_enqueue0(struct ldlm_namespace *ns,
 
         if (lock->l_export->exp_lock_hash)
                 lustre_hash_add(lock->l_export->exp_lock_hash,
-                                &lock->l_remote_handle, 
+                                &lock->l_remote_handle,
                                 &lock->l_exp_hash);
 
 existing_lock:
@@ -1132,7 +1143,7 @@ existing_lock:
                 if (!err && dlm_req->lock_desc.l_resource.lr_type != LDLM_FLOCK)
                         ldlm_reprocess_all(lock->l_resource);
 
-                LDLM_LOCK_PUT(lock);
+                LDLM_LOCK_RELEASE(lock);
         }
 
         LDLM_DEBUG_NOLOCK("server-side enqueue handler END (lock %p, rc %d)",
@@ -1172,8 +1183,9 @@ int ldlm_handle_convert0(struct ptlrpc_request *req,
         int rc;
         ENTRY;
 
-        if (req->rq_export && req->rq_export->exp_ldlm_stats)
-                lprocfs_counter_incr(req->rq_export->exp_ldlm_stats,
+        if (req->rq_export && req->rq_export->exp_nid_stats &&
+            req->rq_export->exp_nid_stats->nid_ldlm_stats)
+                lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
                                      LDLM_CONVERT - LDLM_FIRST_OPC);
 
         rc = req_capsule_server_pack(&req->rq_pill);
@@ -1265,10 +1277,12 @@ int ldlm_request_cancel(struct ptlrpc_request *req,
                 if (res != pres) {
                         if (pres != NULL) {
                                 ldlm_reprocess_all(pres);
+                                LDLM_RESOURCE_DELREF(pres);
                                 ldlm_resource_putref(pres);
                         }
                         if (res != NULL) {
                                 ldlm_resource_getref(res);
+                                LDLM_RESOURCE_ADDREF(res);
                                 ldlm_res_lvbo_update(res, NULL, 0, 1);
                         }
                         pres = res;
@@ -1278,6 +1292,7 @@ int ldlm_request_cancel(struct ptlrpc_request *req,
         }
         if (pres != NULL) {
                 ldlm_reprocess_all(pres);
+                LDLM_RESOURCE_DELREF(pres);
                 ldlm_resource_putref(pres);
         }
         LDLM_DEBUG_NOLOCK("server-side cancel handler END");
@@ -1296,8 +1311,9 @@ int ldlm_handle_cancel(struct ptlrpc_request *req)
                 RETURN(-EFAULT);
         }
 
-        if (req->rq_export && req->rq_export->exp_ldlm_stats)
-                lprocfs_counter_incr(req->rq_export->exp_ldlm_stats,
+        if (req->rq_export && req->rq_export->exp_nid_stats &&
+            req->rq_export->exp_nid_stats->nid_ldlm_stats)
+                lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
                                      LDLM_CANCEL - LDLM_FIRST_OPC);
 
         rc = req_capsule_server_pack(&req->rq_pill);
@@ -1342,7 +1358,7 @@ void ldlm_handle_bl_callback(struct ldlm_namespace *ns,
         }
 
         LDLM_DEBUG(lock, "client blocking callback handler END");
-        LDLM_LOCK_PUT(lock);
+        LDLM_LOCK_RELEASE(lock);
         EXIT;
 }
 
@@ -1372,7 +1388,7 @@ static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
                 /* bug 11300: the lock has already been granted */
                 unlock_res_and_lock(lock);
                 LDLM_DEBUG(lock, "Double grant race happened");
-                LDLM_LOCK_PUT(lock);
+                LDLM_LOCK_RELEASE(lock);
                 EXIT;
                 return;
         }
@@ -1394,10 +1410,10 @@ static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
                    &lock->l_resource->lr_name,
                    sizeof(lock->l_resource->lr_name)) != 0) {
                 unlock_res_and_lock(lock);
-                if (ldlm_lock_change_resource(ns, lock, 
+                if (ldlm_lock_change_resource(ns, lock,
                                 &dlm_req->lock_desc.l_resource.lr_name) != 0) {
                         LDLM_ERROR(lock, "Failed to allocate resource");
-                        LDLM_LOCK_PUT(lock);
+                        LDLM_LOCK_RELEASE(lock);
                         EXIT;
                         return;
                 }
@@ -1436,7 +1452,7 @@ static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
 
         LDLM_DEBUG_NOLOCK("client completion callback handler END (lock %p)",
                           lock);
-        LDLM_LOCK_PUT(lock);
+        LDLM_LOCK_RELEASE(lock);
         EXIT;
 }
 
@@ -1474,7 +1490,7 @@ static void ldlm_handle_gl_callback(struct ptlrpc_request *req,
                 return;
         }
         unlock_res_and_lock(lock);
-        LDLM_LOCK_PUT(lock);
+        LDLM_LOCK_RELEASE(lock);
         EXIT;
 }
 
@@ -1614,6 +1630,8 @@ static int ldlm_callback_handler(struct ptlrpc_request *req)
                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOG_CANCEL_NET))
                         RETURN(0);
                 rc = llog_origin_handle_cancel(req);
+                if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOG_CANCEL_REP))
+                        RETURN(0);
                 ldlm_callback_reply(req, rc);
                 RETURN(0);
         case OBD_QC_CALLBACK:
@@ -1685,7 +1703,7 @@ static int ldlm_callback_handler(struct ptlrpc_request *req)
                         CERROR("ldlm_cli_cancel: %d\n", rc);
         }
 
-        lock = ldlm_handle2lock_ns(ns, &dlm_req->lock_handle[0]);
+        lock = ldlm_handle2lock_long(&dlm_req->lock_handle[0], 0);
         if (!lock) {
                 CDEBUG(D_DLMTRACE, "callback on lock "LPX64" - lock "
                        "disappeared\n", dlm_req->lock_handle[0].cookie);
@@ -1706,7 +1724,7 @@ static int ldlm_callback_handler(struct ptlrpc_request *req)
                                    LPX64" - lock disappeared\n",
                                    dlm_req->lock_handle[0].cookie);
                         unlock_res_and_lock(lock);
-                        LDLM_LOCK_PUT(lock);
+                        LDLM_LOCK_RELEASE(lock);
                         ldlm_callback_reply(req, -EINVAL);
                         RETURN(0);
                 }
@@ -1799,6 +1817,8 @@ static int ldlm_cancel_handler(struct ptlrpc_request *req)
                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOG_CANCEL_NET))
                         RETURN(0);
                 rc = llog_origin_handle_cancel(req);
+                if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOG_CANCEL_REP))
+                        RETURN(0);
                 ldlm_callback_reply(req, rc);
                 RETURN(0);
         default:
@@ -2024,8 +2044,8 @@ void ldlm_put_ref(void)
         EXIT;
 }
 
-/* 
- * Export handle<->lock hash operations. 
+/*
+ * Export handle<->lock hash operations.
  */
 static unsigned
 ldlm_export_lock_hash(lustre_hash_t *lh, void *key, unsigned mask)
@@ -2069,7 +2089,7 @@ ldlm_export_lock_put(struct hlist_node *hnode)
         ENTRY;
 
         lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
-        LDLM_LOCK_PUT(lock);
+        LDLM_LOCK_RELEASE(lock);
 
         RETURN(lock);
 }
@@ -2248,7 +2268,7 @@ static int ldlm_cleanup(void)
 #endif
         ENTRY;
 
-        if (!list_empty(ldlm_namespace_list(LDLM_NAMESPACE_SERVER)) || 
+        if (!list_empty(ldlm_namespace_list(LDLM_NAMESPACE_SERVER)) ||
             !list_empty(ldlm_namespace_list(LDLM_NAMESPACE_CLIENT))) {
                 CERROR("ldlm still has namespaces; clean these up first.\n");
                 ldlm_dump_all_namespaces(LDLM_NAMESPACE_SERVER, D_DLMTRACE);
@@ -2356,6 +2376,7 @@ EXPORT_SYMBOL(ldlm_lock_fast_release);
 EXPORT_SYMBOL(ldlm_lock_match);
 EXPORT_SYMBOL(ldlm_lock_cancel);
 EXPORT_SYMBOL(ldlm_lock_addref);
+EXPORT_SYMBOL(ldlm_lock_addref_try);
 EXPORT_SYMBOL(ldlm_lock_decref);
 EXPORT_SYMBOL(ldlm_lock_decref_and_cancel);
 EXPORT_SYMBOL(ldlm_lock_change_resource);
@@ -2368,6 +2389,7 @@ EXPORT_SYMBOL(ldlm_reprocess_all_ns);
 EXPORT_SYMBOL(ldlm_lock_allow_match);
 
 /* ldlm_request.c */
+EXPORT_SYMBOL(ldlm_completion_ast_async);
 EXPORT_SYMBOL(ldlm_completion_ast);
 EXPORT_SYMBOL(ldlm_blocking_ast);
 EXPORT_SYMBOL(ldlm_glimpse_ast);
@@ -2382,7 +2404,6 @@ EXPORT_SYMBOL(ldlm_cli_cancel);
 EXPORT_SYMBOL(ldlm_cli_cancel_unused);
 EXPORT_SYMBOL(ldlm_cli_cancel_unused_resource);
 EXPORT_SYMBOL(ldlm_cli_cancel_req);
-EXPORT_SYMBOL(ldlm_cli_join_lru);
 EXPORT_SYMBOL(ldlm_replay_locks);
 EXPORT_SYMBOL(ldlm_resource_foreach);
 EXPORT_SYMBOL(ldlm_namespace_foreach);
Lustre primary development and releases