LU-13227 LDLM: update LVB if it is server lock

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

diff --git a/lustre/ldlm/ldlm_lock.c b/lustre/ldlm/ldlm_lock.c
index 66304f1..aa2eeea 100644 (file)
--- a/lustre/ldlm/ldlm_lock.c
+++ b/lustre/ldlm/ldlm_lock.c
@@ -190,6 +190,15 @@ struct ldlm_lock *ldlm_lock_get(struct ldlm_lock *lock)
 }
 EXPORT_SYMBOL(ldlm_lock_get);
 
+static void lock_handle_free(struct rcu_head *rcu)
+{
+       struct ldlm_lock *lock = container_of(rcu, struct ldlm_lock,
+                                             l_handle.h_rcu);
+
+       OBD_FREE_PRE(lock, sizeof(*lock), "slab-freed");
+       kmem_cache_free(ldlm_lock_slab, lock);
+}
+
 /**
  * Release lock reference.
  *
@@ -234,7 +243,7 @@ void ldlm_lock_put(struct ldlm_lock *lock)
                ldlm_resource_putref(res);
                lock->l_resource = NULL;
                 lu_ref_fini(&lock->l_reference);
-               OBD_FREE_RCU(lock, sizeof(*lock), &lock->l_handle);
+               call_rcu(&lock->l_handle.h_rcu, lock_handle_free);
         }
 
         EXIT;
@@ -438,16 +447,7 @@ void ldlm_lock_destroy_nolock(struct ldlm_lock *lock)
         EXIT;
 }
 
-static void lock_handle_free(void *lock, int size)
-{
-       LASSERT(size == sizeof(struct ldlm_lock));
-       OBD_SLAB_FREE(lock, ldlm_lock_slab, size);
-}
-
-static struct portals_handle_ops lock_handle_ops = {
-       .hop_free   = lock_handle_free,
-       .hop_type       = "ldlm",
-};
+static const char lock_handle_owner[] = "ldlm";
 
 /**
  *
@@ -489,8 +489,8 @@ static struct ldlm_lock *ldlm_lock_new(struct ldlm_resource *resource)
 
         lprocfs_counter_incr(ldlm_res_to_ns(resource)->ns_stats,
                              LDLM_NSS_LOCKS);
-       INIT_LIST_HEAD_RCU(&lock->l_handle.h_link);
-       class_handle_hash(&lock->l_handle, &lock_handle_ops);
+       INIT_HLIST_NODE(&lock->l_handle.h_link);
+       class_handle_hash(&lock->l_handle, lock_handle_owner);
 
         lu_ref_init(&lock->l_reference);
         lu_ref_add(&lock->l_reference, "hash", lock);
@@ -600,7 +600,10 @@ struct ldlm_lock *__ldlm_handle2lock(const struct lustre_handle *handle,
 
        LASSERT(handle);
 
-       lock = class_handle2object(handle->cookie, &lock_handle_ops);
+       if (!lustre_handle_is_used(handle))
+               RETURN(NULL);
+
+       lock = class_handle2object(handle->cookie, lock_handle_owner);
        if (lock == NULL)
                RETURN(NULL);
 
@@ -614,7 +617,7 @@ struct ldlm_lock *__ldlm_handle2lock(const struct lustre_handle *handle,
        /* It's unlikely but possible that someone marked the lock as
         * destroyed after we did handle2object on it */
        if ((flags == 0) && !ldlm_is_destroyed(lock)) {
-               lu_ref_add(&lock->l_reference, "handle", current);
+               lu_ref_add_atomic(&lock->l_reference, "handle", current);
                RETURN(lock);
        }
 
@@ -1454,7 +1457,6 @@ enum ldlm_mode ldlm_lock_match_with_skip(struct ldlm_namespace *ns,
                    (!ldlm_is_lvb_ready(lock))) {
                        __u64 wait_flags = LDLM_FL_LVB_READY |
                                LDLM_FL_DESTROYED | LDLM_FL_FAIL_NOTIFIED;
-                       struct l_wait_info lwi;
 
                        if (lock->l_completion_ast) {
                                int err = lock->l_completion_ast(lock,
@@ -1464,12 +1466,11 @@ enum ldlm_mode ldlm_lock_match_with_skip(struct ldlm_namespace *ns,
                                        GOTO(out_fail_match, matched = 0);
                        }
 
-                       lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(obd_timeout),
-                                              NULL, LWI_ON_SIGNAL_NOOP, NULL);
+                       wait_event_idle_timeout(
+                               lock->l_waitq,
+                               lock->l_flags & wait_flags,
+                               cfs_time_seconds(obd_timeout));
 
-                       /* XXX FIXME see comment on CAN_MATCH in lustre_dlm.h */
-                       l_wait_event(lock->l_waitq, lock->l_flags & wait_flags,
-                                    &lwi);
                        if (!ldlm_is_lvb_ready(lock))
                                GOTO(out_fail_match, matched = 0);
                }
@@ -1759,6 +1760,9 @@ enum ldlm_error ldlm_lock_enqueue(const struct lu_env *env,
        int local = ns_is_client(ldlm_res_to_ns(res));
        enum ldlm_error rc = ELDLM_OK;
        struct ldlm_interval *node = NULL;
+#ifdef HAVE_SERVER_SUPPORT
+       bool reconstruct = false;
+#endif
        ENTRY;
 
         /* policies are not executed on the client or during replay */
@@ -1814,6 +1818,19 @@ enum ldlm_error ldlm_lock_enqueue(const struct lu_env *env,
        if (!local && (*flags & LDLM_FL_REPLAY) && res->lr_type == LDLM_EXTENT)
                OBD_SLAB_ALLOC_PTR_GFP(node, ldlm_interval_slab, GFP_NOFS);
 
+#ifdef HAVE_SERVER_SUPPORT
+       reconstruct = !local && res->lr_type == LDLM_FLOCK &&
+                     !(*flags & LDLM_FL_TEST_LOCK);
+       if (reconstruct) {
+               rc = req_can_reconstruct(cookie, NULL);
+               if (rc != 0) {
+                       if (rc == 1)
+                               rc = 0;
+                       RETURN(rc);
+               }
+       }
+#endif
+
        lock_res_and_lock(lock);
        if (local && ldlm_is_granted(lock)) {
                 /* The server returned a blocked lock, but it was granted
@@ -1887,6 +1904,16 @@ enum ldlm_error ldlm_lock_enqueue(const struct lu_env *env,
 
 out:
         unlock_res_and_lock(lock);
+
+#ifdef HAVE_SERVER_SUPPORT
+       if (reconstruct) {
+               struct ptlrpc_request *req = cookie;
+
+               tgt_mk_reply_data(NULL, NULL,
+                                 &req->rq_export->exp_target_data,
+                                 req, 0, NULL, false, 0);
+       }
+#endif
         if (node)
                 OBD_SLAB_FREE(node, ldlm_interval_slab, sizeof(*node));
         return rc;
@@ -2177,8 +2204,11 @@ int ldlm_work_gl_ast_lock(struct ptlrpc_request_set *rqset, void *opaq)
        arg->gl_interpret_data = gl_work->gl_interpret_data;
 
        /* invoke the actual glimpse callback */
-       if (lock->l_glimpse_ast(lock, (void*)arg) == 0)
-               rc = 1;
+       rc = lock->l_glimpse_ast(lock, (void *)arg);
+       if (rc == 0)
+               rc = 1; /* update LVB if this is server lock */
+       else if (rc == -ELDLM_NO_LOCK_DATA)
+               ldlm_lvbo_update(lock->l_resource, lock, NULL, 1);
 
        LDLM_LOCK_RELEASE(lock);
        if (gl_work->gl_flags & LDLM_GL_WORK_SLAB_ALLOCATED)
@@ -2313,7 +2343,7 @@ static void __ldlm_reprocess_all(struct ldlm_resource *res,
                                 enum ldlm_process_intention intention,
                                 struct ldlm_lock *hint)
 {
-       struct list_head rpc_list;
+       LIST_HEAD(rpc_list);
 #ifdef HAVE_SERVER_SUPPORT
        ldlm_reprocessing_policy reprocess;
        struct obd_device *obd;
@@ -2321,7 +2351,6 @@ static void __ldlm_reprocess_all(struct ldlm_resource *res,
 
        ENTRY;
 
-       INIT_LIST_HEAD(&rpc_list);
        /* Local lock trees don't get reprocessed. */
        if (ns_is_client(ldlm_res_to_ns(res))) {
                EXIT;
@@ -2350,7 +2379,6 @@ restart:
 #else
        ENTRY;
 
-       INIT_LIST_HEAD(&rpc_list);
        if (!ns_is_client(ldlm_res_to_ns(res))) {
                CERROR("This is client-side-only module, cannot handle "
                       "LDLM_NAMESPACE_SERVER resource type lock.\n");