X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fldlm%2Fldlm_lock.c;h=ded6c7c285cdaa0e552b6ee0fd7a1b382705ac07;hp=302c96573ec0e6afdd738c309cdcf3da2a1e22a7;hb=6052cc88eb1232ac3b0193f0d47881887a2dcfdc;hpb=f4b93dff9a8f4a59976ea864c4e3c2c42faa5770 diff --git a/lustre/ldlm/ldlm_lock.c b/lustre/ldlm/ldlm_lock.c index 302c965..ded6c7c 100644 --- a/lustre/ldlm/ldlm_lock.c +++ b/lustre/ldlm/ldlm_lock.c @@ -15,11 +15,7 @@ * * You should have received a copy of the GNU General Public License * version 2 along with this program; If not, see - * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf - * - * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara, - * CA 95054 USA or visit www.sun.com if you need additional information or - * have any questions. + * http://www.gnu.org/licenses/gpl-2.0.html * * GPL HEADER END */ @@ -27,7 +23,7 @@ * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. * - * Copyright (c) 2010, 2014, Intel Corporation. + * Copyright (c) 2010, 2017, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -42,9 +38,15 @@ #define DEBUG_SUBSYSTEM S_LDLM #include + +#include #include + #include "ldlm_internal.h" +struct kmem_cache *ldlm_glimpse_work_kmem; +EXPORT_SYMBOL(ldlm_glimpse_work_kmem); + /* lock types */ char *ldlm_lockname[] = { [0] = "--", @@ -83,23 +85,23 @@ static ldlm_policy_local_to_wire_t ldlm_policy_local_to_wire[] = { /** * Converts lock policy from local format to on the wire lock_desc format */ -void ldlm_convert_policy_to_wire(ldlm_type_t type, - const ldlm_policy_data_t *lpolicy, - ldlm_wire_policy_data_t *wpolicy) +void ldlm_convert_policy_to_wire(enum ldlm_type type, + const union ldlm_policy_data *lpolicy, + union ldlm_wire_policy_data *wpolicy) { - ldlm_policy_local_to_wire_t convert; + ldlm_policy_local_to_wire_t convert; - convert = ldlm_policy_local_to_wire[type - LDLM_MIN_TYPE]; + convert = ldlm_policy_local_to_wire[type - LDLM_MIN_TYPE]; - convert(lpolicy, wpolicy); + convert(lpolicy, wpolicy); } /** * Converts lock policy from on the wire lock_desc format to local format */ -void ldlm_convert_policy_to_local(struct obd_export *exp, ldlm_type_t type, - const ldlm_wire_policy_data_t *wpolicy, - ldlm_policy_data_t *lpolicy) +void ldlm_convert_policy_to_local(struct obd_export *exp, enum ldlm_type type, + const union ldlm_wire_policy_data *wpolicy, + union ldlm_policy_data *lpolicy) { ldlm_policy_wire_to_local_t convert; @@ -108,36 +110,32 @@ void ldlm_convert_policy_to_local(struct obd_export *exp, ldlm_type_t type, convert(wpolicy, lpolicy); } -char *ldlm_it2str(int it) -{ - switch (it) { - case IT_OPEN: - return "open"; - case IT_CREAT: - return "creat"; - case (IT_OPEN | IT_CREAT): - return "open|creat"; - case IT_READDIR: - return "readdir"; - case IT_GETATTR: - return "getattr"; - case IT_LOOKUP: - return "lookup"; - case IT_UNLINK: - return "unlink"; - case IT_GETXATTR: - return "getxattr"; - case IT_LAYOUT: - return "layout"; - default: - CERROR("Unknown intent %d\n", it); - return "UNKNOWN"; - } +const char *ldlm_it2str(enum ldlm_intent_flags it) +{ + switch (it) { + case IT_OPEN: + return "open"; + case IT_CREAT: + return "creat"; + case (IT_OPEN | IT_CREAT): + return "open|creat"; + case IT_READDIR: + return "readdir"; + case IT_GETATTR: + return "getattr"; + case IT_LOOKUP: + return "lookup"; + case IT_GETXATTR: + return "getxattr"; + case IT_LAYOUT: + return "layout"; + default: + CERROR("Unknown intent 0x%08x\n", it); + return "UNKNOWN"; + } } EXPORT_SYMBOL(ldlm_it2str); -extern struct kmem_cache *ldlm_lock_slab; - #ifdef HAVE_SERVER_SUPPORT static ldlm_processing_policy ldlm_processing_policy_table[] = { [LDLM_PLAIN] = ldlm_process_plain_lock, @@ -151,6 +149,19 @@ ldlm_processing_policy ldlm_get_processing_policy(struct ldlm_resource *res) return ldlm_processing_policy_table[res->lr_type]; } EXPORT_SYMBOL(ldlm_get_processing_policy); + +static ldlm_reprocessing_policy ldlm_reprocessing_policy_table[] = { + [LDLM_PLAIN] = ldlm_reprocess_queue, + [LDLM_EXTENT] = ldlm_reprocess_queue, + [LDLM_FLOCK] = ldlm_reprocess_queue, + [LDLM_IBITS] = ldlm_reprocess_inodebits_queue, +}; + +ldlm_reprocessing_policy ldlm_get_reprocessing_policy(struct ldlm_resource *res) +{ + return ldlm_reprocessing_policy_table[res->lr_type]; +} + #endif /* HAVE_SERVER_SUPPORT */ void ldlm_register_intent(struct ldlm_namespace *ns, ldlm_res_policy arg) @@ -174,11 +185,20 @@ EXPORT_SYMBOL(ldlm_register_intent); */ struct ldlm_lock *ldlm_lock_get(struct ldlm_lock *lock) { - atomic_inc(&lock->l_refc); + refcount_inc(&lock->l_handle.h_ref); return 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. * @@ -189,8 +209,8 @@ void ldlm_lock_put(struct ldlm_lock *lock) ENTRY; LASSERT(lock->l_resource != LP_POISON); - LASSERT(atomic_read(&lock->l_refc) > 0); - if (atomic_dec_and_test(&lock->l_refc)) { + LASSERT(refcount_read(&lock->l_handle.h_ref) > 0); + if (refcount_dec_and_test(&lock->l_handle.h_ref)) { struct ldlm_resource *res; LDLM_DEBUG(lock, @@ -198,14 +218,13 @@ void ldlm_lock_put(struct ldlm_lock *lock) res = lock->l_resource; LASSERT(ldlm_is_destroyed(lock)); + LASSERT(list_empty(&lock->l_exp_list)); LASSERT(list_empty(&lock->l_res_link)); LASSERT(list_empty(&lock->l_pending_chain)); lprocfs_counter_decr(ldlm_res_to_ns(res)->ns_stats, LDLM_NSS_LOCKS); lu_ref_del(&res->lr_reference, "lock", lock); - ldlm_resource_putref(res); - lock->l_resource = NULL; if (lock->l_export) { class_export_lock_put(lock->l_export, lock); lock->l_export = NULL; @@ -214,9 +233,17 @@ void ldlm_lock_put(struct ldlm_lock *lock) if (lock->l_lvb_data != NULL) OBD_FREE_LARGE(lock->l_lvb_data, lock->l_lvb_len); - ldlm_interval_free(ldlm_interval_detach(lock)); + if (res->lr_type == LDLM_EXTENT) { + ldlm_interval_free(ldlm_interval_detach(lock)); + } else if (res->lr_type == LDLM_IBITS) { + if (lock->l_ibits_node != NULL) + OBD_SLAB_FREE_PTR(lock->l_ibits_node, + ldlm_inodebits_slab); + } + 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; @@ -233,6 +260,8 @@ int ldlm_lock_remove_from_lru_nolock(struct ldlm_lock *lock) struct ldlm_namespace *ns = ldlm_lock_to_ns(lock); LASSERT(lock->l_resource->lr_type != LDLM_FLOCK); + if (ns->ns_last_pos == &lock->l_lru) + ns->ns_last_pos = lock->l_lru.prev; list_del_init(&lock->l_lru); LASSERT(ns->ns_nr_unused > 0); ns->ns_nr_unused--; @@ -243,11 +272,19 @@ int ldlm_lock_remove_from_lru_nolock(struct ldlm_lock *lock) /** * Removes LDLM lock \a lock from LRU. Obtains the LRU lock first. + * + * If \a last_use is non-zero, it will remove the lock from LRU only if + * it matches lock's l_last_used. + * + * \retval 0 if \a last_use is set, the lock is not in LRU list or \a last_use + * doesn't match lock's l_last_used; + * otherwise, the lock hasn't been in the LRU list. + * \retval 1 the lock was in LRU list and removed. */ -int ldlm_lock_remove_from_lru(struct ldlm_lock *lock) +int ldlm_lock_remove_from_lru_check(struct ldlm_lock *lock, ktime_t last_use) { struct ldlm_namespace *ns = ldlm_lock_to_ns(lock); - int rc; + int rc = 0; ENTRY; if (ldlm_is_ns_srv(lock)) { @@ -256,10 +293,12 @@ int ldlm_lock_remove_from_lru(struct ldlm_lock *lock) } spin_lock(&ns->ns_lock); - rc = ldlm_lock_remove_from_lru_nolock(lock); + if (!ktime_compare(last_use, ktime_set(0, 0)) || + !ktime_compare(last_use, lock->l_last_used)) + rc = ldlm_lock_remove_from_lru_nolock(lock); spin_unlock(&ns->ns_lock); - EXIT; - return rc; + + RETURN(rc); } /** @@ -269,11 +308,10 @@ void ldlm_lock_add_to_lru_nolock(struct ldlm_lock *lock) { struct ldlm_namespace *ns = ldlm_lock_to_ns(lock); - lock->l_last_used = cfs_time_current(); + lock->l_last_used = ktime_get(); LASSERT(list_empty(&lock->l_lru)); LASSERT(lock->l_resource->lr_type != LDLM_FLOCK); list_add_tail(&lock->l_lru, &ns->ns_unused_list); - ldlm_clear_skipped(lock); LASSERT(ns->ns_nr_unused >= 0); ns->ns_nr_unused++; } @@ -370,16 +408,6 @@ static int ldlm_lock_destroy_internal(struct ldlm_lock *lock) ldlm_lock_remove_from_lru(lock); class_handle_unhash(&lock->l_handle); -#if 0 - /* Wake anyone waiting for this lock */ - /* FIXME: I should probably add yet another flag, instead of using - * l_export to only call this on clients */ - if (lock->l_export) - class_export_put(lock->l_export); - lock->l_export = NULL; - if (lock->l_export && lock->l_completion_ast) - lock->l_completion_ast(lock, 0); -#endif EXIT; return 1; } @@ -419,22 +447,7 @@ void ldlm_lock_destroy_nolock(struct ldlm_lock *lock) EXIT; } -/* this is called by portals_handle2object with the handle lock taken */ -static void lock_handle_addref(void *lock) -{ - LDLM_LOCK_GET((struct ldlm_lock *)lock); -} - -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_addref = lock_handle_addref, - .hop_free = lock_handle_free, -}; +static const char lock_handle_owner[] = "ldlm"; /** * @@ -460,7 +473,7 @@ static struct ldlm_lock *ldlm_lock_new(struct ldlm_resource *resource) lock->l_resource = resource; lu_ref_add(&resource->lr_reference, "lock", lock); - atomic_set(&lock->l_refc, 2); + refcount_set(&lock->l_handle.h_ref, 2); INIT_LIST_HEAD(&lock->l_res_link); INIT_LIST_HEAD(&lock->l_lru); INIT_LIST_HEAD(&lock->l_pending_chain); @@ -476,12 +489,13 @@ 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(&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); - lock->l_callback_timeout = 0; + lock->l_callback_timestamp = 0; + lock->l_activity = 0; #if LUSTRE_TRACKS_LOCK_EXP_REFS INIT_LIST_HEAD(&lock->l_exp_refs_link); @@ -501,16 +515,16 @@ static struct ldlm_lock *ldlm_lock_new(struct ldlm_resource *resource) int ldlm_lock_change_resource(struct ldlm_namespace *ns, struct ldlm_lock *lock, const struct ldlm_res_id *new_resid) { - struct ldlm_resource *oldres = lock->l_resource; + struct ldlm_resource *oldres; struct ldlm_resource *newres; int type; ENTRY; LASSERT(ns_is_client(ns)); - lock_res_and_lock(lock); - if (memcmp(new_resid, &lock->l_resource->lr_name, - sizeof(lock->l_resource->lr_name)) == 0) { + oldres = lock_res_and_lock(lock); + if (memcmp(new_resid, &oldres->lr_name, + sizeof(oldres->lr_name)) == 0) { /* Nothing to do */ unlock_res_and_lock(lock); RETURN(0); @@ -586,14 +600,24 @@ struct ldlm_lock *__ldlm_handle2lock(const struct lustre_handle *handle, LASSERT(handle); - lock = class_handle2object(handle->cookie, NULL); + if (!lustre_handle_is_used(handle)) + RETURN(NULL); + + lock = class_handle2object(handle->cookie, lock_handle_owner); if (lock == NULL) RETURN(NULL); + if (lock->l_export != NULL && lock->l_export->exp_failed) { + CDEBUG(D_INFO, "lock export failed: lock %p, exp %p\n", + lock, lock->l_export); + LDLM_LOCK_PUT(lock); + RETURN(NULL); + } + /* 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", lock); RETURN(lock); } @@ -601,7 +625,7 @@ struct ldlm_lock *__ldlm_handle2lock(const struct lustre_handle *handle, LASSERT(lock->l_resource != NULL); - lu_ref_add_atomic(&lock->l_reference, "handle", current); + lu_ref_add_atomic(&lock->l_reference, "handle", lock); if (unlikely(ldlm_is_destroyed(lock))) { unlock_res_and_lock(lock); CDEBUG(D_INFO, "lock already destroyed: lock %p\n", lock); @@ -655,12 +679,19 @@ static void ldlm_add_bl_work_item(struct ldlm_lock *lock, struct ldlm_lock *new, * discard dirty data, rather than writing back. */ if (ldlm_is_ast_discard_data(new)) ldlm_set_discard_data(lock); - LASSERT(list_empty(&lock->l_bl_ast)); - list_add(&lock->l_bl_ast, work_list); - LDLM_LOCK_GET(lock); - LASSERT(lock->l_blocking_lock == NULL); - lock->l_blocking_lock = LDLM_LOCK_GET(new); - } + + /* Lock can be converted from a blocking state back to granted + * after lock convert or COS downgrade but still be in an + * older bl_list because it is controlled only by + * ldlm_work_bl_ast_lock(), let it be processed there. + */ + if (list_empty(&lock->l_bl_ast)) { + list_add(&lock->l_bl_ast, work_list); + LDLM_LOCK_GET(lock); + } + LASSERT(lock->l_blocking_lock == NULL); + lock->l_blocking_lock = LDLM_LOCK_GET(new); + } } /** @@ -701,12 +732,12 @@ void ldlm_add_ast_work_item(struct ldlm_lock *lock, struct ldlm_lock *new, * r/w reference type is determined by \a mode * Calls ldlm_lock_addref_internal. */ -void ldlm_lock_addref(struct lustre_handle *lockh, __u32 mode) +void ldlm_lock_addref(const struct lustre_handle *lockh, enum ldlm_mode mode) { struct ldlm_lock *lock; lock = ldlm_handle2lock(lockh); - LASSERTF(lock != NULL, "Non-existing lock: "LPX64"\n", lockh->cookie); + LASSERTF(lock != NULL, "Non-existing lock: %#llx\n", lockh->cookie); ldlm_lock_addref_internal(lock, mode); LDLM_LOCK_PUT(lock); } @@ -719,7 +750,8 @@ EXPORT_SYMBOL(ldlm_lock_addref); * Removes lock from LRU if it is there. * Assumes the LDLM lock is already locked. */ -void ldlm_lock_addref_internal_nolock(struct ldlm_lock *lock, __u32 mode) +void ldlm_lock_addref_internal_nolock(struct ldlm_lock *lock, + enum ldlm_mode mode) { ldlm_lock_remove_from_lru(lock); if (mode & (LCK_NL | LCK_CR | LCK_PR)) { @@ -743,7 +775,7 @@ void ldlm_lock_addref_internal_nolock(struct ldlm_lock *lock, __u32 mode) * * \retval -EAGAIN lock is being canceled. */ -int ldlm_lock_addref_try(struct lustre_handle *lockh, __u32 mode) +int ldlm_lock_addref_try(const struct lustre_handle *lockh, enum ldlm_mode mode) { struct ldlm_lock *lock; int result; @@ -769,11 +801,11 @@ EXPORT_SYMBOL(ldlm_lock_addref_try); * Locks LDLM lock and calls ldlm_lock_addref_internal_nolock to do the work. * Only called for local locks. */ -void ldlm_lock_addref_internal(struct ldlm_lock *lock, __u32 mode) +void ldlm_lock_addref_internal(struct ldlm_lock *lock, enum ldlm_mode mode) { - lock_res_and_lock(lock); - ldlm_lock_addref_internal_nolock(lock, mode); - unlock_res_and_lock(lock); + lock_res_and_lock(lock); + ldlm_lock_addref_internal_nolock(lock, mode); + unlock_res_and_lock(lock); } /** @@ -783,7 +815,8 @@ void ldlm_lock_addref_internal(struct ldlm_lock *lock, __u32 mode) * Does NOT add lock to LRU if no r/w references left to accomodate flock locks * that cannot be placed in LRU. */ -void ldlm_lock_decref_internal_nolock(struct ldlm_lock *lock, __u32 mode) +void ldlm_lock_decref_internal_nolock(struct ldlm_lock *lock, + enum ldlm_mode mode) { LDLM_DEBUG(lock, "ldlm_lock_decref(%s)", ldlm_lockname[mode]); if (mode & (LCK_NL | LCK_CR | LCK_PR)) { @@ -809,87 +842,85 @@ void ldlm_lock_decref_internal_nolock(struct ldlm_lock *lock, __u32 mode) * on the namespace. * For blocked LDLM locks if r/w count drops to zero, blocking_ast is called. */ -void ldlm_lock_decref_internal(struct ldlm_lock *lock, __u32 mode) +void ldlm_lock_decref_internal(struct ldlm_lock *lock, enum ldlm_mode mode) { - struct ldlm_namespace *ns; - ENTRY; + struct ldlm_namespace *ns; - lock_res_and_lock(lock); + ENTRY; - ns = ldlm_lock_to_ns(lock); + lock_res_and_lock(lock); - ldlm_lock_decref_internal_nolock(lock, mode); + ns = ldlm_lock_to_ns(lock); - if (ldlm_is_local(lock) && - !lock->l_readers && !lock->l_writers) { - /* If this is a local lock on a server namespace and this was - * the last reference, cancel the lock. */ - CDEBUG(D_INFO, "forcing cancel of local lock\n"); - ldlm_set_cbpending(lock); - } + ldlm_lock_decref_internal_nolock(lock, mode); - if (!lock->l_readers && !lock->l_writers && - (ldlm_is_cbpending(lock) || lock->l_req_mode == LCK_GROUP)) { - /* If we received a blocked AST and this was the last reference, - * run the callback. + if ((ldlm_is_local(lock) || lock->l_req_mode == LCK_GROUP) && + !lock->l_readers && !lock->l_writers) { + /* If this is a local lock on a server namespace and this was + * the last reference, cancel the lock. + * * Group locks are special: * They must not go in LRU, but they are not called back * like non-group locks, instead they are manually released. * They have an l_writers reference which they keep until * they are manually released, so we remove them when they have * no more reader or writer references. - LU-6368 */ - if (ldlm_is_ns_srv(lock) && lock->l_export) - CERROR("FL_CBPENDING set on non-local lock--just a " - "warning\n"); + ldlm_set_cbpending(lock); + } - LDLM_DEBUG(lock, "final decref done on cbpending lock"); + if (!lock->l_readers && !lock->l_writers && ldlm_is_cbpending(lock)) { + unsigned int mask = D_DLMTRACE; - LDLM_LOCK_GET(lock); /* dropped by bl thread */ - ldlm_lock_remove_from_lru(lock); - unlock_res_and_lock(lock); + /* If we received a blocked AST and this was the last reference, + * run the callback. */ + if (ldlm_is_ns_srv(lock) && lock->l_export) + mask |= D_WARNING; + LDLM_DEBUG_LIMIT(mask, lock, + "final decref done on %sCBPENDING lock", + mask & D_WARNING ? "non-local " : ""); + + LDLM_LOCK_GET(lock); /* dropped by bl thread */ + ldlm_lock_remove_from_lru(lock); + unlock_res_and_lock(lock); if (ldlm_is_fail_loc(lock)) - OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE); + OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE); if (ldlm_is_atomic_cb(lock) || ldlm_bl_to_thread_lock(ns, NULL, lock) != 0) - ldlm_handle_bl_callback(ns, NULL, lock); + ldlm_handle_bl_callback(ns, NULL, lock); } else if (ns_is_client(ns) && - !lock->l_readers && !lock->l_writers && + !lock->l_readers && !lock->l_writers && !ldlm_is_no_lru(lock) && - !ldlm_is_bl_ast(lock)) { + !ldlm_is_bl_ast(lock) && + !ldlm_is_converting(lock)) { - LDLM_DEBUG(lock, "add lock into lru list"); - - /* If this is a client-side namespace and this was the last - * reference, put it on the LRU. */ - ldlm_lock_add_to_lru(lock); - unlock_res_and_lock(lock); + /* If this is a client-side namespace and this was the last + * reference, put it on the LRU. + */ + ldlm_lock_add_to_lru(lock); + unlock_res_and_lock(lock); + LDLM_DEBUG(lock, "add lock into lru list"); if (ldlm_is_fail_loc(lock)) - OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE); - - /* Call ldlm_cancel_lru() only if EARLY_CANCEL and LRU RESIZE - * are not supported by the server, otherwise, it is done on - * enqueue. */ - if (!exp_connect_cancelset(lock->l_conn_export) && - !ns_connect_lru_resize(ns)) - ldlm_cancel_lru(ns, 0, LCF_ASYNC, 0); - } else { - LDLM_DEBUG(lock, "do not add lock into lru list"); - unlock_res_and_lock(lock); - } + OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE); - EXIT; + ldlm_cancel_lru(ns, 0, LCF_ASYNC, 0); + } else { + LDLM_DEBUG(lock, "do not add lock into lru list"); + unlock_res_and_lock(lock); + } + + EXIT; } /** * Decrease reader/writer refcount for LDLM lock with handle \a lockh */ -void ldlm_lock_decref(struct lustre_handle *lockh, __u32 mode) +void ldlm_lock_decref(const struct lustre_handle *lockh, enum ldlm_mode mode) { struct ldlm_lock *lock = __ldlm_handle2lock(lockh, 0); - LASSERTF(lock != NULL, "Non-existing lock: "LPX64"\n", lockh->cookie); + LASSERTF(lock != NULL, "Non-existing lock: %#llx\n", lockh->cookie); ldlm_lock_decref_internal(lock, mode); LDLM_LOCK_PUT(lock); } @@ -901,7 +932,8 @@ EXPORT_SYMBOL(ldlm_lock_decref); * drops to zero instead of putting into LRU. * */ -void ldlm_lock_decref_and_cancel(struct lustre_handle *lockh, __u32 mode) +void ldlm_lock_decref_and_cancel(const struct lustre_handle *lockh, + enum ldlm_mode mode) { struct ldlm_lock *lock = __ldlm_handle2lock(lockh, 0); ENTRY; @@ -1015,7 +1047,6 @@ static void search_granted_lock(struct list_head *queue, prev->mode_link = &req->l_sl_mode; prev->policy_link = &req->l_sl_policy; EXIT; - return; } /** @@ -1060,16 +1091,14 @@ static void ldlm_granted_list_add_lock(struct ldlm_lock *lock, * Add a lock to granted list on a resource maintaining skiplist * correctness. */ -static void ldlm_grant_lock_with_skiplist(struct ldlm_lock *lock) +void ldlm_grant_lock_with_skiplist(struct ldlm_lock *lock) { - struct sl_insert_point prev; - ENTRY; + struct sl_insert_point prev; - LASSERT(lock->l_req_mode == lock->l_granted_mode); + LASSERT(ldlm_is_granted(lock)); - search_granted_lock(&lock->l_resource->lr_granted, lock, &prev); - ldlm_granted_list_add_lock(lock, &prev); - EXIT; + search_granted_lock(&lock->l_resource->lr_granted, lock, &prev); + ldlm_granted_list_add_lock(lock, &prev); } /** @@ -1079,7 +1108,6 @@ static void ldlm_grant_lock_with_skiplist(struct ldlm_lock *lock) * NOTE: called by * - ldlm_lock_enqueue * - ldlm_reprocess_queue - * - ldlm_lock_convert * * must be called with lr_lock held */ @@ -1095,114 +1123,206 @@ void ldlm_grant_lock(struct ldlm_lock *lock, struct list_head *work_list) if (work_list && lock->l_completion_ast != NULL) ldlm_add_ast_work_item(lock, NULL, work_list); - /* We should not add locks to granted list in the following cases: - * - this is an UNLOCK but not a real lock; - * - this is a TEST lock; - * - this is a F_CANCELLK lock (async flock has req_mode == 0) - * - this is a deadlock (flock cannot be granted) */ - if (lock->l_req_mode == 0 || - lock->l_req_mode == LCK_NL || - ldlm_is_test_lock(lock) || - ldlm_is_flock_deadlock(lock)) - RETURN_EXIT; - if (res->lr_type == LDLM_PLAIN || res->lr_type == LDLM_IBITS) ldlm_grant_lock_with_skiplist(lock); else if (res->lr_type == LDLM_EXTENT) ldlm_extent_add_lock(res, lock); - else - ldlm_resource_add_lock(res, &res->lr_granted, lock); - - if (lock->l_granted_mode < res->lr_most_restr) - res->lr_most_restr = lock->l_granted_mode; + else if (res->lr_type == LDLM_FLOCK) { + /* We should not add locks to granted list in the following + * cases: + * - this is an UNLOCK but not a real lock; + * - this is a TEST lock; + * - this is a F_CANCELLK lock (async flock has req_mode == 0) + * - this is a deadlock (flock cannot be granted) */ + if (lock->l_req_mode == 0 || + lock->l_req_mode == LCK_NL || + ldlm_is_test_lock(lock) || + ldlm_is_flock_deadlock(lock)) + RETURN_EXIT; + ldlm_resource_add_lock(res, &res->lr_granted, lock); + } else { + LBUG(); + } ldlm_pool_add(&ldlm_res_to_ns(res)->ns_pool, lock); EXIT; } /** - * Search for a lock with given properties in a queue. + * Check if the given @lock meets the criteria for a match. + * A reference on the lock is taken if matched. * - * \retval a referenced lock or NULL. See the flag descriptions below, in the - * comment above ldlm_lock_match + * \param lock test-against this lock + * \param data parameters */ -static struct ldlm_lock *search_queue(struct list_head *queue, - ldlm_mode_t *mode, - ldlm_policy_data_t *policy, - struct ldlm_lock *old_lock, - __u64 flags, int unref) -{ - struct ldlm_lock *lock; - struct list_head *tmp; +static int lock_matches(struct ldlm_lock *lock, struct ldlm_match_data *data) +{ + union ldlm_policy_data *lpol = &lock->l_policy_data; + enum ldlm_mode match = LCK_MINMODE; + + if (lock == data->lmd_old) + return INTERVAL_ITER_STOP; + + /* Check if this lock can be matched. + * Used by LU-2919(exclusive open) for open lease lock */ + if (ldlm_is_excl(lock)) + return INTERVAL_ITER_CONT; + + /* llite sometimes wants to match locks that will be + * canceled when their users drop, but we allow it to match + * if it passes in CBPENDING and the lock still has users. + * this is generally only going to be used by children + * whose parents already hold a lock so forward progress + * can still happen. */ + if (ldlm_is_cbpending(lock) && + !(data->lmd_flags & LDLM_FL_CBPENDING)) + return INTERVAL_ITER_CONT; + if (!(data->lmd_match & LDLM_MATCH_UNREF) && ldlm_is_cbpending(lock) && + lock->l_readers == 0 && lock->l_writers == 0) + return INTERVAL_ITER_CONT; + + if (!(lock->l_req_mode & *data->lmd_mode)) + return INTERVAL_ITER_CONT; + + /* When we search for ast_data, we are not doing a traditional match, + * so we don't worry about IBITS or extent matching. + */ + if (data->lmd_match & (LDLM_MATCH_AST | LDLM_MATCH_AST_ANY)) { + if (!lock->l_ast_data) + return INTERVAL_ITER_CONT; - list_for_each(tmp, queue) { - ldlm_mode_t match; + if (data->lmd_match & LDLM_MATCH_AST_ANY) + goto matched; + } - lock = list_entry(tmp, struct ldlm_lock, l_res_link); + match = lock->l_req_mode; - if (lock == old_lock) - break; + switch (lock->l_resource->lr_type) { + case LDLM_EXTENT: + if (lpol->l_extent.start > data->lmd_policy->l_extent.start || + lpol->l_extent.end < data->lmd_policy->l_extent.end) + return INTERVAL_ITER_CONT; - /* Check if this lock can be matched. - * Used by LU-2919(exclusive open) for open lease lock */ - if (ldlm_is_excl(lock)) - continue; + if (unlikely(match == LCK_GROUP) && + data->lmd_policy->l_extent.gid != LDLM_GID_ANY && + lpol->l_extent.gid != data->lmd_policy->l_extent.gid) + return INTERVAL_ITER_CONT; + break; + case LDLM_IBITS: + /* We match if we have existing lock with same or wider set + of bits. */ + if ((lpol->l_inodebits.bits & + data->lmd_policy->l_inodebits.bits) != + data->lmd_policy->l_inodebits.bits) + return INTERVAL_ITER_CONT; + break; + default: + ; + } - /* llite sometimes wants to match locks that will be - * canceled when their users drop, but we allow it to match - * if it passes in CBPENDING and the lock still has users. - * this is generally only going to be used by children - * whose parents already hold a lock so forward progress - * can still happen. */ - if (ldlm_is_cbpending(lock) && - !(flags & LDLM_FL_CBPENDING)) - continue; - if (!unref && ldlm_is_cbpending(lock) && - lock->l_readers == 0 && lock->l_writers == 0) - continue; + /* We match if we have existing lock with same or wider set + of bits. */ + if (!(data->lmd_match & LDLM_MATCH_UNREF) && LDLM_HAVE_MASK(lock, GONE)) + return INTERVAL_ITER_CONT; - if (!(lock->l_req_mode & *mode)) - continue; - match = lock->l_req_mode; + if (!equi(data->lmd_flags & LDLM_FL_LOCAL_ONLY, ldlm_is_local(lock))) + return INTERVAL_ITER_CONT; - if (lock->l_resource->lr_type == LDLM_EXTENT && - (lock->l_policy_data.l_extent.start > - policy->l_extent.start || - lock->l_policy_data.l_extent.end < policy->l_extent.end)) - continue; + /* Filter locks by skipping flags */ + if (data->lmd_skip_flags & lock->l_flags) + return INTERVAL_ITER_CONT; - if (unlikely(match == LCK_GROUP) && - lock->l_resource->lr_type == LDLM_EXTENT && - policy->l_extent.gid != LDLM_GID_ANY && - lock->l_policy_data.l_extent.gid != policy->l_extent.gid) +matched: + if (data->lmd_flags & LDLM_FL_TEST_LOCK) { + LDLM_LOCK_GET(lock); + ldlm_lock_touch_in_lru(lock); + } else { + ldlm_lock_addref_internal_nolock(lock, match); + } + + *data->lmd_mode = match; + data->lmd_lock = lock; + + return INTERVAL_ITER_STOP; +} + +static unsigned int itree_overlap_cb(struct interval_node *in, void *args) +{ + struct ldlm_interval *node = to_ldlm_interval(in); + struct ldlm_match_data *data = args; + struct ldlm_lock *lock; + int rc; + + list_for_each_entry(lock, &node->li_group, l_sl_policy) { + rc = lock_matches(lock, data); + if (rc == INTERVAL_ITER_STOP) + return INTERVAL_ITER_STOP; + } + return INTERVAL_ITER_CONT; +} + +/** + * Search for a lock with given parameters in interval trees. + * + * \param res search for a lock in this resource + * \param data parameters + * + * \retval a referenced lock or NULL. + */ +struct ldlm_lock *search_itree(struct ldlm_resource *res, + struct ldlm_match_data *data) +{ + struct interval_node_extent ext = { + .start = data->lmd_policy->l_extent.start, + .end = data->lmd_policy->l_extent.end + }; + int idx; + + data->lmd_lock = NULL; + + for (idx = 0; idx < LCK_MODE_NUM; idx++) { + struct ldlm_interval_tree *tree = &res->lr_itree[idx]; + + if (tree->lit_root == NULL) continue; - /* We match if we have existing lock with same or wider set - of bits. */ - if (lock->l_resource->lr_type == LDLM_IBITS && - ((lock->l_policy_data.l_inodebits.bits & - policy->l_inodebits.bits) != - policy->l_inodebits.bits)) - continue; + if (!(tree->lit_mode & *data->lmd_mode)) + continue; - if (!unref && LDLM_HAVE_MASK(lock, GONE)) - continue; + interval_search(tree->lit_root, &ext, + itree_overlap_cb, data); + if (data->lmd_lock) + return data->lmd_lock; + } - if ((flags & LDLM_FL_LOCAL_ONLY) && - !ldlm_is_local(lock)) - continue; + return NULL; +} +EXPORT_SYMBOL(search_itree); - if (flags & LDLM_FL_TEST_LOCK) { - LDLM_LOCK_GET(lock); - ldlm_lock_touch_in_lru(lock); - } else { - ldlm_lock_addref_internal_nolock(lock, match); - } - *mode = match; - return lock; - } - return NULL; +/** + * Search for a lock with given properties in a queue. + * + * \param queue search for a lock in this queue + * \param data parameters + * + * \retval a referenced lock or NULL. + */ +static struct ldlm_lock *search_queue(struct list_head *queue, + struct ldlm_match_data *data) +{ + struct ldlm_lock *lock; + int rc; + + data->lmd_lock = NULL; + + list_for_each_entry(lock, queue, l_res_link) { + rc = lock_matches(lock, data); + if (rc == INTERVAL_ITER_STOP) + return data->lmd_lock; + } + + return NULL; } void ldlm_lock_fail_match_locked(struct ldlm_lock *lock) @@ -1277,142 +1397,131 @@ EXPORT_SYMBOL(ldlm_lock_allow_match); * keep caller code unchanged), the context failure will be discovered by * caller sometime later. */ -ldlm_mode_t ldlm_lock_match(struct ldlm_namespace *ns, __u64 flags, - const struct ldlm_res_id *res_id, ldlm_type_t type, - ldlm_policy_data_t *policy, ldlm_mode_t mode, - struct lustre_handle *lockh, int unref) -{ - struct ldlm_resource *res; - struct ldlm_lock *lock, *old_lock = NULL; - int rc = 0; - ENTRY; +enum ldlm_mode ldlm_lock_match_with_skip(struct ldlm_namespace *ns, + __u64 flags, __u64 skip_flags, + const struct ldlm_res_id *res_id, + enum ldlm_type type, + union ldlm_policy_data *policy, + enum ldlm_mode mode, + struct lustre_handle *lockh, + enum ldlm_match_flags match_flags) +{ + struct ldlm_match_data data = { + .lmd_old = NULL, + .lmd_lock = NULL, + .lmd_mode = &mode, + .lmd_policy = policy, + .lmd_flags = flags, + .lmd_skip_flags = skip_flags, + .lmd_match = match_flags, + }; + struct ldlm_resource *res; + struct ldlm_lock *lock; + int matched; - if (ns == NULL) { - old_lock = ldlm_handle2lock(lockh); - LASSERT(old_lock); + ENTRY; - ns = ldlm_lock_to_ns(old_lock); - res_id = &old_lock->l_resource->lr_name; - type = old_lock->l_resource->lr_type; - mode = old_lock->l_req_mode; - } + if (ns == NULL) { + data.lmd_old = ldlm_handle2lock(lockh); + LASSERT(data.lmd_old != NULL); + + ns = ldlm_lock_to_ns(data.lmd_old); + res_id = &data.lmd_old->l_resource->lr_name; + type = data.lmd_old->l_resource->lr_type; + *data.lmd_mode = data.lmd_old->l_req_mode; + } res = ldlm_resource_get(ns, NULL, res_id, type, 0); if (IS_ERR(res)) { - LASSERT(old_lock == NULL); + LASSERT(data.lmd_old == NULL); RETURN(0); } - LDLM_RESOURCE_ADDREF(res); - lock_res(res); - - lock = search_queue(&res->lr_granted, &mode, policy, old_lock, - flags, unref); - if (lock != NULL) - GOTO(out, rc = 1); - if (flags & LDLM_FL_BLOCK_GRANTED) - GOTO(out, rc = 0); - lock = search_queue(&res->lr_converting, &mode, policy, old_lock, - flags, unref); - if (lock != NULL) - GOTO(out, rc = 1); - lock = search_queue(&res->lr_waiting, &mode, policy, old_lock, - flags, unref); - if (lock != NULL) - GOTO(out, rc = 1); - - EXIT; - out: - unlock_res(res); - LDLM_RESOURCE_DELREF(res); - ldlm_resource_putref(res); - - if (lock) { - ldlm_lock2handle(lock, lockh); - if ((flags & LDLM_FL_LVB_READY) && + LDLM_RESOURCE_ADDREF(res); + lock_res(res); + if (res->lr_type == LDLM_EXTENT) + lock = search_itree(res, &data); + else + lock = search_queue(&res->lr_granted, &data); + if (!lock && !(flags & LDLM_FL_BLOCK_GRANTED)) + lock = search_queue(&res->lr_waiting, &data); + matched = lock ? mode : 0; + unlock_res(res); + LDLM_RESOURCE_DELREF(res); + ldlm_resource_putref(res); + + if (lock) { + ldlm_lock2handle(lock, lockh); + if ((flags & LDLM_FL_LVB_READY) && (!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, - LDLM_FL_WAIT_NOREPROC, - NULL); - if (err) { - if (flags & LDLM_FL_TEST_LOCK) - LDLM_LOCK_RELEASE(lock); - else - ldlm_lock_decref_internal(lock, - mode); - rc = 0; - goto out2; - } - } - lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(obd_timeout), - NULL, LWI_ON_SIGNAL_NOOP, NULL); - - /* 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)) { - if (flags & LDLM_FL_TEST_LOCK) - LDLM_LOCK_RELEASE(lock); - else - ldlm_lock_decref_internal(lock, mode); - rc = 0; - } - } - } - out2: - if (rc) { - LDLM_DEBUG(lock, "matched ("LPU64" "LPU64")", - (type == LDLM_PLAIN || type == LDLM_IBITS) ? - res_id->name[2] : policy->l_extent.start, - (type == LDLM_PLAIN || type == LDLM_IBITS) ? - res_id->name[3] : policy->l_extent.end); - - /* check user's security context */ - if (lock->l_conn_export && - sptlrpc_import_check_ctx( - class_exp2cliimp(lock->l_conn_export))) { - if (!(flags & LDLM_FL_TEST_LOCK)) - ldlm_lock_decref_internal(lock, mode); - rc = 0; - } + if (lock->l_completion_ast) { + int err = lock->l_completion_ast(lock, + LDLM_FL_WAIT_NOREPROC, + NULL); + if (err) + GOTO(out_fail_match, matched = 0); + } - if (flags & LDLM_FL_TEST_LOCK) - LDLM_LOCK_RELEASE(lock); - - } else if (!(flags & LDLM_FL_TEST_LOCK)) {/*less verbose for test-only*/ - LDLM_DEBUG_NOLOCK("not matched ns %p type %u mode %u res " - LPU64"/"LPU64" ("LPU64" "LPU64")", ns, - type, mode, res_id->name[0], res_id->name[1], - (type == LDLM_PLAIN || type == LDLM_IBITS) ? - res_id->name[2] :policy->l_extent.start, - (type == LDLM_PLAIN || type == LDLM_IBITS) ? - res_id->name[3] : policy->l_extent.end); - } - if (old_lock) - LDLM_LOCK_PUT(old_lock); + wait_event_idle_timeout( + lock->l_waitq, + lock->l_flags & wait_flags, + cfs_time_seconds(obd_timeout)); + + if (!ldlm_is_lvb_ready(lock)) + GOTO(out_fail_match, matched = 0); + } + + /* check user's security context */ + if (lock->l_conn_export && + sptlrpc_import_check_ctx( + class_exp2cliimp(lock->l_conn_export))) + GOTO(out_fail_match, matched = 0); + + LDLM_DEBUG(lock, "matched (%llu %llu)", + (type == LDLM_PLAIN || type == LDLM_IBITS) ? + res_id->name[2] : policy->l_extent.start, + (type == LDLM_PLAIN || type == LDLM_IBITS) ? + res_id->name[3] : policy->l_extent.end); + +out_fail_match: + if (flags & LDLM_FL_TEST_LOCK) + LDLM_LOCK_RELEASE(lock); + else if (!matched) + ldlm_lock_decref_internal(lock, mode); + } + + /* less verbose for test-only */ + if (!matched && !(flags & LDLM_FL_TEST_LOCK)) { + LDLM_DEBUG_NOLOCK("not matched ns %p type %u mode %u res " + "%llu/%llu (%llu %llu)", ns, + type, mode, res_id->name[0], res_id->name[1], + (type == LDLM_PLAIN || type == LDLM_IBITS) ? + res_id->name[2] : policy->l_extent.start, + (type == LDLM_PLAIN || type == LDLM_IBITS) ? + res_id->name[3] : policy->l_extent.end); + } + if (data.lmd_old != NULL) + LDLM_LOCK_PUT(data.lmd_old); - return rc ? mode : 0; + return matched; } -EXPORT_SYMBOL(ldlm_lock_match); +EXPORT_SYMBOL(ldlm_lock_match_with_skip); -ldlm_mode_t ldlm_revalidate_lock_handle(struct lustre_handle *lockh, - __u64 *bits) +enum ldlm_mode ldlm_revalidate_lock_handle(const struct lustre_handle *lockh, + __u64 *bits) { - struct ldlm_lock *lock; - ldlm_mode_t mode = 0; - ENTRY; + struct ldlm_lock *lock; + enum ldlm_mode mode = 0; + ENTRY; - lock = ldlm_handle2lock(lockh); - if (lock != NULL) { - lock_res_and_lock(lock); + lock = ldlm_handle2lock(lockh); + if (lock != NULL) { + lock_res_and_lock(lock); if (LDLM_HAVE_MASK(lock, GONE)) - GOTO(out, mode); + GOTO(out, mode); if (ldlm_is_cbpending(lock) && lock->l_readers == 0 && lock->l_writers == 0) @@ -1526,7 +1635,7 @@ int ldlm_fill_lvb(struct ldlm_lock *lock, struct req_capsule *pill, memcpy(data, lvb, size); break; default: - LDLM_ERROR(lock, "Unknown LVB type: %d\n", lock->l_lvb_type); + LDLM_ERROR(lock, "Unknown LVB type: %d", lock->l_lvb_type); libcfs_debug_dumpstack(NULL); RETURN(-EINVAL); } @@ -1540,8 +1649,8 @@ int ldlm_fill_lvb(struct ldlm_lock *lock, struct req_capsule *pill, */ struct ldlm_lock *ldlm_lock_create(struct ldlm_namespace *ns, const struct ldlm_res_id *res_id, - ldlm_type_t type, - ldlm_mode_t mode, + enum ldlm_type type, + enum ldlm_mode mode, const struct ldlm_callback_suite *cbs, void *data, __u32 lvb_len, enum lvb_type lvb_type) @@ -1561,7 +1670,7 @@ struct ldlm_lock *ldlm_lock_create(struct ldlm_namespace *ns, lock->l_req_mode = mode; lock->l_ast_data = data; - lock->l_pid = current_pid(); + lock->l_pid = current->pid; if (ns_is_server(ns)) ldlm_set_ns_srv(lock); if (cbs) { @@ -1570,11 +1679,18 @@ struct ldlm_lock *ldlm_lock_create(struct ldlm_namespace *ns, lock->l_glimpse_ast = cbs->lcs_glimpse; } - lock->l_tree_node = NULL; - /* if this is the extent lock, allocate the interval tree node */ - if (type == LDLM_EXTENT) - if (ldlm_interval_alloc(lock) == NULL) - GOTO(out, rc = -ENOMEM); + switch (type) { + case LDLM_EXTENT: + rc = ldlm_extent_alloc_lock(lock); + break; + case LDLM_IBITS: + rc = ldlm_inodebits_alloc_lock(lock); + break; + default: + rc = 0; + } + if (rc) + GOTO(out, rc); if (lvb_len) { lock->l_lvb_len = lvb_len; @@ -1595,6 +1711,34 @@ out: RETURN(ERR_PTR(rc)); } +#ifdef HAVE_SERVER_SUPPORT +static enum ldlm_error ldlm_lock_enqueue_helper(struct ldlm_lock *lock, + __u64 *flags) +{ + struct ldlm_resource *res = lock->l_resource; + enum ldlm_error rc = ELDLM_OK; + LIST_HEAD(rpc_list); + ldlm_processing_policy policy; + + ENTRY; + + policy = ldlm_get_processing_policy(res); +restart: + policy(lock, flags, LDLM_PROCESS_ENQUEUE, &rc, &rpc_list); + if (rc == ELDLM_OK && lock->l_granted_mode != lock->l_req_mode && + res->lr_type != LDLM_FLOCK) { + rc = ldlm_handle_conflict_lock(lock, flags, &rpc_list); + if (rc == -ERESTART) + GOTO(restart, rc); + } + + if (!list_empty(&rpc_list)) + ldlm_discard_bl_list(&rpc_list); + + RETURN(rc); +} +#endif + /** * Enqueue (request) a lock. * @@ -1605,25 +1749,26 @@ out: * set, skip all the enqueueing and delegate lock processing to intent policy * function. */ -ldlm_error_t ldlm_lock_enqueue(struct ldlm_namespace *ns, - struct ldlm_lock **lockp, - void *cookie, __u64 *flags) +enum ldlm_error ldlm_lock_enqueue(const struct lu_env *env, + struct ldlm_namespace *ns, + struct ldlm_lock **lockp, + void *cookie, __u64 *flags) { - struct ldlm_lock *lock = *lockp; - struct ldlm_resource *res = lock->l_resource; - int local = ns_is_client(ldlm_res_to_ns(res)); + struct ldlm_lock *lock = *lockp; + struct ldlm_resource *res; + int local = ns_is_client(ns); + enum ldlm_error rc = ELDLM_OK; + struct ldlm_interval *node = NULL; #ifdef HAVE_SERVER_SUPPORT - ldlm_processing_policy policy; + bool reconstruct = false; #endif - ldlm_error_t rc = ELDLM_OK; - struct ldlm_interval *node = NULL; - ENTRY; + ENTRY; /* policies are not executed on the client or during replay */ if ((*flags & (LDLM_FL_HAS_INTENT|LDLM_FL_REPLAY)) == LDLM_FL_HAS_INTENT && !local && ns->ns_policy) { - rc = ns->ns_policy(ns, lockp, cookie, lock->l_req_mode, *flags, - NULL); + rc = ns->ns_policy(env, ns, lockp, cookie, lock->l_req_mode, + *flags, NULL); if (rc == ELDLM_LOCK_REPLACED) { /* The lock that was returned has already been granted, * and placed into lockp. If it's not the same as the @@ -1635,6 +1780,16 @@ ldlm_error_t ldlm_lock_enqueue(struct ldlm_namespace *ns, } *flags |= LDLM_FL_LOCK_CHANGED; RETURN(0); + } else if (rc != ELDLM_OK && + ldlm_is_granted(lock)) { + LASSERT(*flags & LDLM_FL_RESENT); + /* It may happen that ns_policy returns an error in + * resend case, object may be unlinked or just some + * error occurs. It is unclear if lock reached the + * client in the original reply, just leave the lock on + * server, not returning it again to client. Due to + * LU-6529, the server will not OOM. */ + RETURN(rc); } else if (rc != ELDLM_OK || (rc == ELDLM_OK && (*flags & LDLM_FL_INTENT_ONLY))) { ldlm_lock_destroy(lock); @@ -1649,21 +1804,39 @@ ldlm_error_t ldlm_lock_enqueue(struct ldlm_namespace *ns, * Take NO_TIMEOUT from the lock as it is inherited through * LDLM_FL_INHERIT_MASK */ *flags |= LDLM_FL_LOCK_CHANGED; - if (lock->l_req_mode != lock->l_granted_mode) + if (!ldlm_is_granted(lock)) *flags |= LDLM_FL_BLOCK_GRANTED; *flags |= lock->l_flags & LDLM_FL_NO_TIMEOUT; RETURN(ELDLM_OK); } +#ifdef HAVE_SERVER_SUPPORT /* For a replaying lock, it might be already in granted list. So * unlinking the lock will cause the interval node to be freed, we * have to allocate the interval node early otherwise we can't regrant - * this lock in the future. - jay */ - if (!local && (*flags & LDLM_FL_REPLAY) && res->lr_type == LDLM_EXTENT) + * this lock in the future. - jay + * + * The only time the ldlm_resource changes for the ldlm_lock is when + * ldlm_lock_change_resource() is called and that only happens for + * the Lustre client case. + */ + if (!local && (*flags & LDLM_FL_REPLAY) && + lock->l_resource->lr_type == LDLM_EXTENT) OBD_SLAB_ALLOC_PTR_GFP(node, ldlm_interval_slab, GFP_NOFS); - lock_res_and_lock(lock); - if (local && lock->l_req_mode == lock->l_granted_mode) { + reconstruct = !local && lock->l_resource->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 + res = lock_res_and_lock(lock); + if (local && ldlm_is_granted(lock)) { /* The server returned a blocked lock, but it was granted * before we got a chance to actually enqueue it. We don't * need to do anything else. */ @@ -1689,6 +1862,10 @@ ldlm_error_t ldlm_lock_enqueue(struct ldlm_namespace *ns, ldlm_set_ast_discard_data(lock); if (*flags & LDLM_FL_TEST_LOCK) ldlm_set_test_lock(lock); + if (*flags & LDLM_FL_COS_INCOMPAT) + ldlm_set_cos_incompat(lock); + if (*flags & LDLM_FL_COS_ENABLED) + ldlm_set_cos_enabled(lock); /* This distinction between local lock trees is very important; a client * namespace only has information about locks taken by that client, and @@ -1700,33 +1877,27 @@ ldlm_error_t ldlm_lock_enqueue(struct ldlm_namespace *ns, * more or less trusting the clients not to lie. * * FIXME (bug 268): Detect obvious lies by checking compatibility in - * granted/converting queues. */ + * granted queue. */ if (local) { - if (*flags & LDLM_FL_BLOCK_CONV) - ldlm_resource_add_lock(res, &res->lr_converting, lock); - else if (*flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED)) - ldlm_resource_add_lock(res, &res->lr_waiting, lock); - else - ldlm_grant_lock(lock, NULL); + if (*flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED)) + ldlm_resource_add_lock(res, &res->lr_waiting, lock); + else + ldlm_grant_lock(lock, NULL); GOTO(out, rc = ELDLM_OK); #ifdef HAVE_SERVER_SUPPORT - } else if (*flags & LDLM_FL_REPLAY) { - if (*flags & LDLM_FL_BLOCK_CONV) { - ldlm_resource_add_lock(res, &res->lr_converting, lock); + } else if (*flags & LDLM_FL_REPLAY) { + if (*flags & LDLM_FL_BLOCK_WAIT) { + ldlm_resource_add_lock(res, &res->lr_waiting, lock); GOTO(out, rc = ELDLM_OK); - } else if (*flags & LDLM_FL_BLOCK_WAIT) { - ldlm_resource_add_lock(res, &res->lr_waiting, lock); + } else if (*flags & LDLM_FL_BLOCK_GRANTED) { + ldlm_grant_lock(lock, NULL); GOTO(out, rc = ELDLM_OK); - } else if (*flags & LDLM_FL_BLOCK_GRANTED) { - ldlm_grant_lock(lock, NULL); - GOTO(out, rc = ELDLM_OK); - } - /* If no flags, fall through to normal enqueue path. */ - } + } + /* If no flags, fall through to normal enqueue path. */ + } - policy = ldlm_processing_policy_table[res->lr_type]; - policy(lock, flags, 1, &rc, NULL); - GOTO(out, rc); + rc = ldlm_lock_enqueue_helper(lock, flags); + GOTO(out, rc); #else } else { CERROR("This is client-side-only module, cannot handle " @@ -1737,6 +1908,16 @@ ldlm_error_t ldlm_lock_enqueue(struct ldlm_namespace *ns, 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; @@ -1750,113 +1931,223 @@ out: * Must be called with resource lock held. */ int ldlm_reprocess_queue(struct ldlm_resource *res, struct list_head *queue, - struct list_head *work_list) + struct list_head *work_list, + enum ldlm_process_intention intention, + struct ldlm_lock *hint) { struct list_head *tmp, *pos; - ldlm_processing_policy policy; + ldlm_processing_policy policy; __u64 flags; - int rc = LDLM_ITER_CONTINUE; - ldlm_error_t err; - ENTRY; + int rc = LDLM_ITER_CONTINUE; + enum ldlm_error err; + LIST_HEAD(bl_ast_list); - check_res_locked(res); + ENTRY; - policy = ldlm_processing_policy_table[res->lr_type]; - LASSERT(policy); + check_res_locked(res); + policy = ldlm_get_processing_policy(res); + LASSERT(policy); + LASSERT(intention == LDLM_PROCESS_RESCAN || + intention == LDLM_PROCESS_RECOVERY); + +restart: list_for_each_safe(tmp, pos, queue) { - struct ldlm_lock *pending; + struct ldlm_lock *pending; + LIST_HEAD(rpc_list); + pending = list_entry(tmp, struct ldlm_lock, l_res_link); CDEBUG(D_INFO, "Reprocessing lock %p\n", pending); flags = 0; - rc = policy(pending, &flags, 0, &err, work_list); - if (rc != LDLM_ITER_CONTINUE) - break; + rc = policy(pending, &flags, intention, &err, &rpc_list); + if (pending->l_granted_mode == pending->l_req_mode || + res->lr_type == LDLM_FLOCK) { + list_splice(&rpc_list, work_list); + } else { + list_splice(&rpc_list, &bl_ast_list); + } + /* + * When this is called from recovery done, we always want + * to scan the whole list no matter what 'rc' is returned. + */ + if (rc != LDLM_ITER_CONTINUE && + intention == LDLM_PROCESS_RESCAN) + break; } - RETURN(rc); + if (!list_empty(&bl_ast_list)) { + unlock_res(res); + + rc = ldlm_run_ast_work(ldlm_res_to_ns(res), &bl_ast_list, + LDLM_WORK_BL_AST); + + lock_res(res); + if (rc == -ERESTART) + GOTO(restart, rc); + } + + if (!list_empty(&bl_ast_list)) + ldlm_discard_bl_list(&bl_ast_list); + + RETURN(intention == LDLM_PROCESS_RESCAN ? rc : LDLM_ITER_CONTINUE); } -#endif /** - * Process a call to blocking AST callback for a lock in ast_work list + * Conflicting locks are detected for a lock to be enqueued, add the lock + * into waiting list and send blocking ASTs to the conflicting locks. + * + * \param[in] lock The lock to be enqueued. + * \param[out] flags Lock flags for the lock to be enqueued. + * \param[in] rpc_list Conflicting locks list. + * + * \retval -ERESTART: Some lock was instantly canceled while sending + * blocking ASTs, caller needs to re-check conflicting + * locks. + * \retval -EAGAIN: Lock was destroyed, caller should return error. + * \reval 0: Lock is successfully added in waiting list. */ -static int -ldlm_work_bl_ast_lock(struct ptlrpc_request_set *rqset, void *opaq) +int ldlm_handle_conflict_lock(struct ldlm_lock *lock, __u64 *flags, + struct list_head *rpc_list) { - struct ldlm_cb_set_arg *arg = opaq; - struct ldlm_lock_desc d; - int rc; - struct ldlm_lock *lock; + struct ldlm_resource *res = lock->l_resource; + int rc; ENTRY; - if (list_empty(arg->list)) - RETURN(-ENOENT); + check_res_locked(res); - lock = list_entry(arg->list->next, struct ldlm_lock, l_bl_ast); + /* If either of the compat_queue()s returned failure, then we + * have ASTs to send and must go onto the waiting list. + * + * bug 2322: we used to unlink and re-add here, which was a + * terrible folly -- if we goto restart, we could get + * re-ordered! Causes deadlock, because ASTs aren't sent! */ + if (list_empty(&lock->l_res_link)) + ldlm_resource_add_lock(res, &res->lr_waiting, lock); + unlock_res(res); + + rc = ldlm_run_ast_work(ldlm_res_to_ns(res), rpc_list, + LDLM_WORK_BL_AST); + + if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_OST_FAIL_RACE) && + !ns_is_client(ldlm_res_to_ns(res))) + class_fail_export(lock->l_export); + + lock_res(res); + if (rc == -ERESTART) { + /* 15715: The lock was granted and destroyed after + * resource lock was dropped. Interval node was freed + * in ldlm_lock_destroy. Anyway, this always happens + * when a client is being evicted. So it would be + * ok to return an error. -jay */ + if (ldlm_is_destroyed(lock)) + RETURN(-EAGAIN); + + /* lock was granted while resource was unlocked. */ + if (ldlm_is_granted(lock)) { + /* bug 11300: if the lock has been granted, + * break earlier because otherwise, we will go + * to restart and ldlm_resource_unlink will be + * called and it causes the interval node to be + * freed. Then we will fail at + * ldlm_extent_add_lock() */ + *flags &= ~LDLM_FL_BLOCKED_MASK; + RETURN(0); + } - /* nobody should touch l_bl_ast */ - lock_res_and_lock(lock); - list_del_init(&lock->l_bl_ast); + RETURN(rc); + } + *flags |= LDLM_FL_BLOCK_GRANTED; - LASSERT(ldlm_is_ast_sent(lock)); - LASSERT(lock->l_bl_ast_run == 0); - LASSERT(lock->l_blocking_lock); - lock->l_bl_ast_run++; - unlock_res_and_lock(lock); + RETURN(0); +} - ldlm_lock2desc(lock->l_blocking_lock, &d); +/** + * Discard all AST work items from list. + * + * If for whatever reason we do not want to send ASTs to conflicting locks + * anymore, disassemble the list with this function. + */ +void ldlm_discard_bl_list(struct list_head *bl_list) +{ + struct ldlm_lock *lock, *tmp; - rc = lock->l_blocking_ast(lock, &d, (void *)arg, LDLM_CB_BLOCKING); - LDLM_LOCK_RELEASE(lock->l_blocking_lock); - lock->l_blocking_lock = NULL; - LDLM_LOCK_RELEASE(lock); + ENTRY; - RETURN(rc); + list_for_each_entry_safe(lock, tmp, bl_list, l_bl_ast) { + LASSERT(!list_empty(&lock->l_bl_ast)); + list_del_init(&lock->l_bl_ast); + ldlm_clear_ast_sent(lock); + LASSERT(lock->l_bl_ast_run == 0); + ldlm_clear_blocking_lock(lock); + LDLM_LOCK_RELEASE(lock); + } + EXIT; } /** - * Process a call to completion AST callback for a lock in ast_work list + * Process a call to blocking AST callback for a lock in ast_work list */ static int -ldlm_work_cp_ast_lock(struct ptlrpc_request_set *rqset, void *opaq) +ldlm_work_bl_ast_lock(struct ptlrpc_request_set *rqset, void *opaq) { - struct ldlm_cb_set_arg *arg = opaq; - int rc = 0; - struct ldlm_lock *lock; - ldlm_completion_callback completion_callback; + struct ldlm_cb_set_arg *arg = opaq; + struct ldlm_lock *lock; + struct ldlm_lock_desc d; + struct ldlm_bl_desc bld; + int rc; + ENTRY; if (list_empty(arg->list)) RETURN(-ENOENT); - lock = list_entry(arg->list->next, struct ldlm_lock, l_cp_ast); - - /* It's possible to receive a completion AST before we've set - * the l_completion_ast pointer: either because the AST arrived - * before the reply, or simply because there's a small race - * window between receiving the reply and finishing the local - * enqueue. (bug 842) - * - * This can't happen with the blocking_ast, however, because we - * will never call the local blocking_ast until we drop our - * reader/writer reference, which we won't do until we get the - * reply and finish enqueueing. */ + lock = list_entry(arg->list->next, struct ldlm_lock, l_bl_ast); - /* nobody should touch l_cp_ast */ + /* nobody should touch l_bl_ast but some locks in the list may become + * granted after lock convert or COS downgrade, these locks should be + * just skipped here and removed from the list. + */ lock_res_and_lock(lock); - list_del_init(&lock->l_cp_ast); - LASSERT(ldlm_is_cp_reqd(lock)); - /* save l_completion_ast since it can be changed by - * mds_intent_policy(), see bug 14225 */ - completion_callback = lock->l_completion_ast; - ldlm_clear_cp_reqd(lock); + list_del_init(&lock->l_bl_ast); + + /* lock is not blocking lock anymore, but was kept in the list because + * it can managed only here. + */ + if (!ldlm_is_ast_sent(lock)) { + unlock_res_and_lock(lock); + LDLM_LOCK_RELEASE(lock); + RETURN(0); + } + + LASSERT(lock->l_blocking_lock); + ldlm_lock2desc(lock->l_blocking_lock, &d); + /* copy blocking lock ibits in cancel_bits as well, + * new client may use them for lock convert and it is + * important to use new field to convert locks from + * new servers only + */ + d.l_policy_data.l_inodebits.cancel_bits = + lock->l_blocking_lock->l_policy_data.l_inodebits.bits; + + /* Blocking lock is being destroyed here but some information about it + * may be needed inside l_blocking_ast() function below, + * e.g. in mdt_blocking_ast(). So save needed data in bl_desc. + */ + bld.bl_same_client = lock->l_client_cookie == + lock->l_blocking_lock->l_client_cookie; + bld.bl_cos_incompat = ldlm_is_cos_incompat(lock->l_blocking_lock); + arg->bl_desc = &bld; + + LASSERT(ldlm_is_ast_sent(lock)); + LASSERT(lock->l_bl_ast_run == 0); + lock->l_bl_ast_run++; + ldlm_clear_blocking_lock(lock); unlock_res_and_lock(lock); - if (completion_callback != NULL) - rc = completion_callback(lock, 0, (void *)arg); + rc = lock->l_blocking_ast(lock, &d, (void *)arg, LDLM_CB_BLOCKING); + LDLM_LOCK_RELEASE(lock); RETURN(rc); @@ -1913,18 +2204,71 @@ int ldlm_work_gl_ast_lock(struct ptlrpc_request_set *rqset, void *opaq) /* transfer the glimpse descriptor to ldlm_cb_set_arg */ arg->gl_desc = gl_work->gl_desc; + arg->gl_interpret_reply = gl_work->gl_interpret_reply; + 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_NOFREE) == 0) + if (gl_work->gl_flags & LDLM_GL_WORK_SLAB_ALLOCATED) + OBD_SLAB_FREE_PTR(gl_work, ldlm_glimpse_work_kmem); + else OBD_FREE_PTR(gl_work); RETURN(rc); } +#endif + +/** + * Process a call to completion AST callback for a lock in ast_work list + */ +static int +ldlm_work_cp_ast_lock(struct ptlrpc_request_set *rqset, void *opaq) +{ + struct ldlm_cb_set_arg *arg = opaq; + struct ldlm_lock *lock; + ldlm_completion_callback completion_callback; + int rc = 0; + + ENTRY; + + if (list_empty(arg->list)) + RETURN(-ENOENT); + + lock = list_entry(arg->list->next, struct ldlm_lock, l_cp_ast); + + /* It's possible to receive a completion AST before we've set + * the l_completion_ast pointer: either because the AST arrived + * before the reply, or simply because there's a small race + * window between receiving the reply and finishing the local + * enqueue. (bug 842) + * + * This can't happen with the blocking_ast, however, because we + * will never call the local blocking_ast until we drop our + * reader/writer reference, which we won't do until we get the + * reply and finish enqueueing. */ + + /* nobody should touch l_cp_ast */ + lock_res_and_lock(lock); + list_del_init(&lock->l_cp_ast); + LASSERT(ldlm_is_cp_reqd(lock)); + /* save l_completion_ast since it can be changed by + * mds_intent_policy(), see bug 14225 */ + completion_callback = lock->l_completion_ast; + ldlm_clear_cp_reqd(lock); + unlock_res_and_lock(lock); + + if (completion_callback != NULL) + rc = completion_callback(lock, 0, (void *)arg); + LDLM_LOCK_RELEASE(lock); + + RETURN(rc); +} /** * Process list of locks in need of ASTs being sent. @@ -1933,11 +2277,11 @@ int ldlm_work_gl_ast_lock(struct ptlrpc_request_set *rqset, void *opaq) * one. */ int ldlm_run_ast_work(struct ldlm_namespace *ns, struct list_head *rpc_list, - ldlm_desc_ast_t ast_type) + ldlm_desc_ast_t ast_type) { struct ldlm_cb_set_arg *arg; - set_producer_func work_ast_lock; - int rc; + set_producer_func work_ast_lock; + int rc; if (list_empty(rpc_list)) RETURN(0); @@ -1950,24 +2294,26 @@ int ldlm_run_ast_work(struct ldlm_namespace *ns, struct list_head *rpc_list, arg->list = rpc_list; switch (ast_type) { - case LDLM_WORK_BL_AST: - arg->type = LDLM_BL_CALLBACK; - work_ast_lock = ldlm_work_bl_ast_lock; - break; - case LDLM_WORK_CP_AST: - arg->type = LDLM_CP_CALLBACK; - work_ast_lock = ldlm_work_cp_ast_lock; - break; - case LDLM_WORK_REVOKE_AST: - arg->type = LDLM_BL_CALLBACK; - work_ast_lock = ldlm_work_revoke_ast_lock; - break; - case LDLM_WORK_GL_AST: - arg->type = LDLM_GL_CALLBACK; - work_ast_lock = ldlm_work_gl_ast_lock; - break; - default: - LBUG(); + case LDLM_WORK_CP_AST: + arg->type = LDLM_CP_CALLBACK; + work_ast_lock = ldlm_work_cp_ast_lock; + break; +#ifdef HAVE_SERVER_SUPPORT + case LDLM_WORK_BL_AST: + arg->type = LDLM_BL_CALLBACK; + work_ast_lock = ldlm_work_bl_ast_lock; + break; + case LDLM_WORK_REVOKE_AST: + arg->type = LDLM_BL_CALLBACK; + work_ast_lock = ldlm_work_revoke_ast_lock; + break; + case LDLM_WORK_GL_AST: + arg->type = LDLM_GL_CALLBACK; + work_ast_lock = ldlm_work_gl_ast_lock; + break; +#endif + default: + LBUG(); } /* We create a ptlrpc request set with flow control extension. @@ -1979,7 +2325,7 @@ int ldlm_run_ast_work(struct ldlm_namespace *ns, struct list_head *rpc_list, if (arg->set == NULL) GOTO(out, rc = -ENOMEM); - ptlrpc_set_wait(arg->set); + ptlrpc_set_wait(NULL, arg->set); ptlrpc_set_destroy(arg->set); rc = atomic_read(&arg->restart) ? -ERESTART : 0; @@ -1989,87 +2335,94 @@ out: return rc; } -static int reprocess_one_queue(struct ldlm_resource *res, void *closure) -{ - ldlm_reprocess_all(res); - return LDLM_ITER_CONTINUE; -} - -static int ldlm_reprocess_res(cfs_hash_t *hs, cfs_hash_bd_t *bd, - struct hlist_node *hnode, void *arg) -{ - struct ldlm_resource *res = cfs_hash_object(hs, hnode); - int rc; - - rc = reprocess_one_queue(res, arg); - - return rc == LDLM_ITER_STOP; -} - -/** - * Iterate through all resources on a namespace attempting to grant waiting - * locks. - */ -void ldlm_reprocess_all_ns(struct ldlm_namespace *ns) -{ - ENTRY; - - if (ns != NULL) { - cfs_hash_for_each_nolock(ns->ns_rs_hash, - ldlm_reprocess_res, NULL); - } - EXIT; -} - /** * Try to grant all waiting locks on a resource. * - * Calls ldlm_reprocess_queue on converting and waiting queues. + * Calls ldlm_reprocess_queue on waiting queue. * * Typically called after some resource locks are cancelled to see * if anything could be granted as a result of the cancellation. */ -void ldlm_reprocess_all(struct ldlm_resource *res) +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 - int rc; - ENTRY; + ldlm_reprocessing_policy reprocess; + struct obd_device *obd; + int rc; - INIT_LIST_HEAD(&rpc_list); - /* Local lock trees don't get reprocessed. */ - if (ns_is_client(ldlm_res_to_ns(res))) { - EXIT; - return; - } + ENTRY; + /* Local lock trees don't get reprocessed. */ + if (ns_is_client(ldlm_res_to_ns(res))) { + EXIT; + return; + } + + /* Disable reprocess during lock replay stage but allow during + * request replay stage. + */ + obd = ldlm_res_to_ns(res)->ns_obd; + if (obd->obd_recovering && + atomic_read(&obd->obd_req_replay_clients) == 0) + RETURN_EXIT; restart: - lock_res(res); - rc = ldlm_reprocess_queue(res, &res->lr_converting, &rpc_list); - if (rc == LDLM_ITER_CONTINUE) - ldlm_reprocess_queue(res, &res->lr_waiting, &rpc_list); - unlock_res(res); - - rc = ldlm_run_ast_work(ldlm_res_to_ns(res), &rpc_list, - LDLM_WORK_CP_AST); - if (rc == -ERESTART) { + lock_res(res); + reprocess = ldlm_get_reprocessing_policy(res); + reprocess(res, &res->lr_waiting, &rpc_list, intention, hint); + unlock_res(res); + + rc = ldlm_run_ast_work(ldlm_res_to_ns(res), &rpc_list, + LDLM_WORK_CP_AST); + if (rc == -ERESTART) { LASSERT(list_empty(&rpc_list)); - goto restart; - } + goto restart; + } #else - ENTRY; + 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"); - LBUG(); - } + 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"); + LBUG(); + } #endif - EXIT; + EXIT; +} + +void ldlm_reprocess_all(struct ldlm_resource *res, struct ldlm_lock *hint) +{ + __ldlm_reprocess_all(res, LDLM_PROCESS_RESCAN, hint); } EXPORT_SYMBOL(ldlm_reprocess_all); +static int ldlm_reprocess_res(struct cfs_hash *hs, struct cfs_hash_bd *bd, + struct hlist_node *hnode, void *arg) +{ + struct ldlm_resource *res = cfs_hash_object(hs, hnode); + + /* This is only called once after recovery done. LU-8306. */ + __ldlm_reprocess_all(res, LDLM_PROCESS_RECOVERY, NULL); + return 0; +} + +/** + * Iterate through all resources on a namespace attempting to grant waiting + * locks. + */ +void ldlm_reprocess_recovery_done(struct ldlm_namespace *ns) +{ + ENTRY; + + if (ns != NULL) { + cfs_hash_for_each_nolock(ns->ns_rs_hash, + ldlm_reprocess_res, NULL, 0); + } + EXIT; +} + /** * Helper function to call blocking AST for LDLM lock \a lock in a * "cancelling" mode. @@ -2080,15 +2433,24 @@ void ldlm_cancel_callback(struct ldlm_lock *lock) if (!ldlm_is_cancel(lock)) { ldlm_set_cancel(lock); if (lock->l_blocking_ast) { - unlock_res_and_lock(lock); - lock->l_blocking_ast(lock, NULL, lock->l_ast_data, - LDLM_CB_CANCELING); - lock_res_and_lock(lock); - } else { - LDLM_DEBUG(lock, "no blocking ast"); - } - } - ldlm_set_bl_done(lock); + unlock_res_and_lock(lock); + lock->l_blocking_ast(lock, NULL, lock->l_ast_data, + LDLM_CB_CANCELING); + lock_res_and_lock(lock); + } else { + LDLM_DEBUG(lock, "no blocking ast"); + } + + /* only canceller can set bl_done bit */ + ldlm_set_bl_done(lock); + wake_up_all(&lock->l_waitq); + } else if (!ldlm_is_bl_done(lock)) { + /* The lock is guaranteed to have been canceled once + * returning from this function. */ + unlock_res_and_lock(lock); + wait_event_idle(lock->l_waitq, is_bl_done(lock)); + lock_res_and_lock(lock); + } } /** @@ -2122,6 +2484,7 @@ void ldlm_lock_cancel(struct ldlm_lock *lock) * talking to me first. -phik */ if (lock->l_readers || lock->l_writers) { LDLM_ERROR(lock, "lock still has references"); + unlock_res_and_lock(lock); LBUG(); } @@ -2131,13 +2494,16 @@ void ldlm_lock_cancel(struct ldlm_lock *lock) /* Releases cancel callback. */ ldlm_cancel_callback(lock); - LASSERT(!ldlm_is_waited(lock)); + /* Yes, second time, just in case it was added again while we were + * running with no res lock in ldlm_cancel_callback */ + if (ldlm_is_waited(lock)) + ldlm_del_waiting_lock(lock); ldlm_resource_unlink_lock(lock); ldlm_lock_destroy_nolock(lock); - if (lock->l_granted_mode == lock->l_req_mode) - ldlm_pool_del(&ns->ns_pool, lock); + if (ldlm_is_granted(lock)) + ldlm_pool_del(&ns->ns_pool, lock); /* Make sure we will not be called again for same lock what is possible * if not to zero out lock->l_granted_mode */ @@ -2151,7 +2517,7 @@ EXPORT_SYMBOL(ldlm_lock_cancel); /** * Set opaque data into the lock that only makes sense to upper layer. */ -int ldlm_lock_set_data(struct lustre_handle *lockh, void *data) +int ldlm_lock_set_data(const struct lustre_handle *lockh, void *data) { struct ldlm_lock *lock = ldlm_handle2lock(lockh); int rc = -EINVAL; @@ -2169,229 +2535,200 @@ int ldlm_lock_set_data(struct lustre_handle *lockh, void *data) EXPORT_SYMBOL(ldlm_lock_set_data); struct export_cl_data { + const struct lu_env *ecl_env; struct obd_export *ecl_exp; int ecl_loop; }; +static void ldlm_cancel_lock_for_export(struct obd_export *exp, + struct ldlm_lock *lock, + struct export_cl_data *ecl) +{ + struct ldlm_resource *res; + + res = ldlm_resource_getref(lock->l_resource); + + ldlm_lvbo_update(res, lock, NULL, 1); + ldlm_lock_cancel(lock); + if (!exp->exp_obd->obd_stopping) + ldlm_reprocess_all(res, lock); + ldlm_resource_putref(res); + + ecl->ecl_loop++; + if ((ecl->ecl_loop & -ecl->ecl_loop) == ecl->ecl_loop) { + CDEBUG(D_INFO, "Export %p, %d locks cancelled.\n", + exp, ecl->ecl_loop); + } +} + /** - * Iterator function for ldlm_cancel_locks_for_export. + * Iterator function for ldlm_export_cancel_locks. * Cancels passed locks. */ -static int ldlm_cancel_locks_for_export_cb(cfs_hash_t *hs, cfs_hash_bd_t *bd, - struct hlist_node *hnode, void *data) +static int +ldlm_cancel_locks_for_export_cb(struct cfs_hash *hs, struct cfs_hash_bd *bd, + struct hlist_node *hnode, void *data) { struct export_cl_data *ecl = (struct export_cl_data *)data; struct obd_export *exp = ecl->ecl_exp; - struct ldlm_lock *lock = cfs_hash_object(hs, hnode); - struct ldlm_resource *res; + struct ldlm_lock *lock = cfs_hash_object(hs, hnode); - res = ldlm_resource_getref(lock->l_resource); - LDLM_LOCK_GET(lock); - - LDLM_DEBUG(lock, "export %p", exp); - ldlm_res_lvbo_update(res, NULL, 1); - ldlm_lock_cancel(lock); - ldlm_reprocess_all(res); - ldlm_resource_putref(res); - LDLM_LOCK_RELEASE(lock); - - ecl->ecl_loop++; - if ((ecl->ecl_loop & -ecl->ecl_loop) == ecl->ecl_loop) { - CDEBUG(D_INFO, - "Cancel lock %p for export %p (loop %d), still have " - "%d locks left on hash table.\n", - lock, exp, ecl->ecl_loop, - atomic_read(&hs->hs_count)); - } + LDLM_LOCK_GET(lock); + ldlm_cancel_lock_for_export(exp, lock, ecl); + LDLM_LOCK_RELEASE(lock); return 0; } /** - * Cancel all locks for given export. + * Cancel all blocked locks for given export. * * Typically called on client disconnection/eviction */ -void ldlm_cancel_locks_for_export(struct obd_export *exp) +int ldlm_export_cancel_blocked_locks(struct obd_export *exp) { + struct lu_env env; struct export_cl_data ecl = { .ecl_exp = exp, .ecl_loop = 0, }; + int rc; - cfs_hash_for_each_empty(exp->exp_lock_hash, - ldlm_cancel_locks_for_export_cb, &ecl); + rc = lu_env_init(&env, LCT_DT_THREAD); + if (rc) + RETURN(rc); + ecl.ecl_env = &env; + + while (!list_empty(&exp->exp_bl_list)) { + struct ldlm_lock *lock; + + spin_lock_bh(&exp->exp_bl_list_lock); + if (!list_empty(&exp->exp_bl_list)) { + lock = list_entry(exp->exp_bl_list.next, + struct ldlm_lock, l_exp_list); + LDLM_LOCK_GET(lock); + list_del_init(&lock->l_exp_list); + } else { + lock = NULL; + } + spin_unlock_bh(&exp->exp_bl_list_lock); + + if (lock == NULL) + break; + + ldlm_cancel_lock_for_export(exp, lock, &ecl); + LDLM_LOCK_RELEASE(lock); + } + + lu_env_fini(&env); + + CDEBUG(D_DLMTRACE, "Export %p, canceled %d locks, " + "left on hash table %d.\n", exp, ecl.ecl_loop, + atomic_read(&exp->exp_lock_hash->hs_count)); + + return ecl.ecl_loop; } /** - * Downgrade an exclusive lock. - * - * A fast variant of ldlm_lock_convert for convertion of exclusive - * locks. The convertion is always successful. - * Used by Commit on Sharing (COS) code. + * Cancel all locks for given export. * - * \param lock A lock to convert - * \param new_mode new lock mode + * Typically called after client disconnection/eviction */ -void ldlm_lock_downgrade(struct ldlm_lock *lock, int new_mode) +int ldlm_export_cancel_locks(struct obd_export *exp) { - ENTRY; + struct export_cl_data ecl; + struct lu_env env; + int rc; - LASSERT(lock->l_granted_mode & (LCK_PW | LCK_EX)); - LASSERT(new_mode == LCK_COS); + rc = lu_env_init(&env, LCT_DT_THREAD); + if (rc) + RETURN(rc); + ecl.ecl_env = &env; + ecl.ecl_exp = exp; + ecl.ecl_loop = 0; - lock_res_and_lock(lock); - ldlm_resource_unlink_lock(lock); - /* - * Remove the lock from pool as it will be added again in - * ldlm_grant_lock() called below. - */ - ldlm_pool_del(&ldlm_lock_to_ns(lock)->ns_pool, lock); + cfs_hash_for_each_empty(exp->exp_lock_hash, + ldlm_cancel_locks_for_export_cb, &ecl); - lock->l_req_mode = new_mode; - ldlm_grant_lock(lock, NULL); - unlock_res_and_lock(lock); - ldlm_reprocess_all(lock->l_resource); + CDEBUG(D_DLMTRACE, "Export %p, canceled %d locks, " + "left on hash table %d.\n", exp, ecl.ecl_loop, + atomic_read(&exp->exp_lock_hash->hs_count)); - EXIT; + if (ecl.ecl_loop > 0 && + atomic_read(&exp->exp_lock_hash->hs_count) == 0 && + exp->exp_obd->obd_stopping) + ldlm_reprocess_recovery_done(exp->exp_obd->obd_namespace); + + lu_env_fini(&env); + + return ecl.ecl_loop; } -EXPORT_SYMBOL(ldlm_lock_downgrade); /** - * Attempt to convert already granted lock to a different mode. + * Downgrade an PW/EX lock to COS | CR mode. + * + * A lock mode convertion from PW/EX mode to less conflict mode. The + * convertion may fail if lock was canceled before downgrade, but it doesn't + * indicate any problem, because such lock has no reader or writer, and will + * be released soon. + * + * Used by Commit on Sharing (COS) code to force object changes commit in case + * of conflict. Converted lock is considered as new lock and all blocking AST + * things are cleared, so any pending or new blocked lock on that lock will + * cause new call to blocking_ast and force resource object commit. * - * While lock conversion is not currently used, future client-side - * optimizations could take advantage of it to avoid discarding cached - * pages on a file. + * Also used by layout_change to replace EX lock to CR lock. + * + * \param lock A lock to convert + * \param new_mode new lock mode */ -struct ldlm_resource *ldlm_lock_convert(struct ldlm_lock *lock, int new_mode, - __u32 *flags) +void ldlm_lock_mode_downgrade(struct ldlm_lock *lock, enum ldlm_mode new_mode) { - struct list_head rpc_list; - struct ldlm_resource *res; - struct ldlm_namespace *ns; - int granted = 0; #ifdef HAVE_SERVER_SUPPORT - int old_mode; - struct sl_insert_point prev; -#endif - struct ldlm_interval *node; ENTRY; - INIT_LIST_HEAD(&rpc_list); - /* Just return if mode is unchanged. */ - if (new_mode == lock->l_granted_mode) { - *flags |= LDLM_FL_BLOCK_GRANTED; - RETURN(lock->l_resource); - } + LASSERT(new_mode == LCK_COS || new_mode == LCK_CR); - /* I can't check the type of lock here because the bitlock of lock - * is not held here, so do the allocation blindly. -jay */ - OBD_SLAB_ALLOC_PTR_GFP(node, ldlm_interval_slab, GFP_NOFS); - if (node == NULL) /* Actually, this causes EDEADLOCK to be returned */ - RETURN(NULL); + lock_res_and_lock(lock); - LASSERTF((new_mode == LCK_PW && lock->l_granted_mode == LCK_PR), - "new_mode %u, granted %u\n", new_mode, lock->l_granted_mode); + if (!(lock->l_granted_mode & (LCK_PW | LCK_EX))) { + unlock_res_and_lock(lock); - lock_res_and_lock(lock); + LASSERT(lock->l_granted_mode == LCK_MINMODE); + LDLM_DEBUG(lock, "lock was canceled before downgrade"); + RETURN_EXIT; + } - res = lock->l_resource; - ns = ldlm_res_to_ns(res); + ldlm_resource_unlink_lock(lock); + /* + * Remove the lock from pool as it will be added again in + * ldlm_grant_lock() called below. + */ + ldlm_pool_del(&ldlm_lock_to_ns(lock)->ns_pool, lock); -#ifdef HAVE_SERVER_SUPPORT - old_mode = lock->l_req_mode; -#endif - lock->l_req_mode = new_mode; - if (res->lr_type == LDLM_PLAIN || res->lr_type == LDLM_IBITS) { -#ifdef HAVE_SERVER_SUPPORT - /* remember the lock position where the lock might be - * added back to the granted list later and also - * remember the join mode for skiplist fixing. */ - prev.res_link = lock->l_res_link.prev; - prev.mode_link = lock->l_sl_mode.prev; - prev.policy_link = lock->l_sl_policy.prev; -#endif - ldlm_resource_unlink_lock(lock); - } else { - ldlm_resource_unlink_lock(lock); - if (res->lr_type == LDLM_EXTENT) { - /* FIXME: ugly code, I have to attach the lock to a - * interval node again since perhaps it will be granted - * soon */ - INIT_LIST_HEAD(&node->li_group); - ldlm_interval_attach(node, lock); - node = NULL; - } - } + /* Consider downgraded lock as a new lock and clear all states + * related to a previous blocking AST processing. + */ + ldlm_clear_blocking_data(lock); - /* - * Remove old lock from the pool before adding the lock with new - * mode below in ->policy() - */ - ldlm_pool_del(&ns->ns_pool, lock); + lock->l_req_mode = new_mode; + ldlm_grant_lock(lock, NULL); + unlock_res_and_lock(lock); - /* If this is a local resource, put it on the appropriate list. */ - if (ns_is_client(ldlm_res_to_ns(res))) { - if (*flags & (LDLM_FL_BLOCK_CONV | LDLM_FL_BLOCK_GRANTED)) { - ldlm_resource_add_lock(res, &res->lr_converting, lock); - } else { - /* This should never happen, because of the way the - * server handles conversions. */ - LDLM_ERROR(lock, "Erroneous flags %x on local lock\n", - *flags); - LBUG(); + ldlm_reprocess_all(lock->l_resource, lock); - ldlm_grant_lock(lock, &rpc_list); - granted = 1; - /* FIXME: completion handling not with lr_lock held ! */ - if (lock->l_completion_ast) - lock->l_completion_ast(lock, 0, NULL); - } -#ifdef HAVE_SERVER_SUPPORT - } else { - int rc; - ldlm_error_t err; - __u64 pflags = 0; - ldlm_processing_policy policy; - policy = ldlm_processing_policy_table[res->lr_type]; - rc = policy(lock, &pflags, 0, &err, &rpc_list); - if (rc == LDLM_ITER_STOP) { - lock->l_req_mode = old_mode; - if (res->lr_type == LDLM_EXTENT) - ldlm_extent_add_lock(res, lock); - else - ldlm_granted_list_add_lock(lock, &prev); - - res = NULL; - } else { - *flags |= LDLM_FL_BLOCK_GRANTED; - granted = 1; - } - } -#else - } else { - CERROR("This is client-side-only module, cannot handle " - "LDLM_NAMESPACE_SERVER resource type lock.\n"); - LBUG(); - } + EXIT; #endif - unlock_res_and_lock(lock); - - if (granted) - ldlm_run_ast_work(ns, &rpc_list, LDLM_WORK_CP_AST); - if (node) - OBD_SLAB_FREE(node, ldlm_interval_slab, sizeof(*node)); - RETURN(res); } +EXPORT_SYMBOL(ldlm_lock_mode_downgrade); /** * Print lock with lock handle \a lockh description into debug log. * * Used when printing all locks on a resource for debug purposes. */ -void ldlm_lock_dump_handle(int level, struct lustre_handle *lockh) +void ldlm_lock_dump_handle(int level, const struct lustre_handle *lockh) { struct ldlm_lock *lock; @@ -2418,125 +2755,139 @@ void _ldlm_lock_debug(struct ldlm_lock *lock, { va_list args; struct obd_export *exp = lock->l_export; - struct ldlm_resource *resource = lock->l_resource; + struct ldlm_resource *resource = NULL; + struct va_format vaf; char *nid = "local"; + /* on server-side resource of lock doesn't change */ + if ((lock->l_flags & LDLM_FL_NS_SRV) != 0) { + if (lock->l_resource != NULL) + resource = ldlm_resource_getref(lock->l_resource); + } else if (spin_trylock(&lock->l_lock)) { + if (lock->l_resource != NULL) + resource = ldlm_resource_getref(lock->l_resource); + spin_unlock(&lock->l_lock); + } + va_start(args, fmt); + vaf.fmt = fmt; + vaf.va = &args; if (exp && exp->exp_connection) { - nid = libcfs_nid2str(exp->exp_connection->c_peer.nid); + nid = obd_export_nid2str(exp); } else if (exp && exp->exp_obd != NULL) { struct obd_import *imp = exp->exp_obd->u.cli.cl_import; - nid = libcfs_nid2str(imp->imp_connection->c_peer.nid); + nid = obd_import_nid2str(imp); } if (resource == NULL) { - libcfs_debug_vmsg2(msgdata, fmt, args, - " ns: \?\? lock: %p/"LPX64" lrc: %d/%d,%d mode: %s/%s " - "res: \?\? rrc=\?\? type: \?\?\? flags: "LPX64" nid: %s " - "remote: "LPX64" expref: %d pid: %u timeout: %lu " - "lvb_type: %d\n", - lock, - lock->l_handle.h_cookie, atomic_read(&lock->l_refc), - lock->l_readers, lock->l_writers, - ldlm_lockname[lock->l_granted_mode], - ldlm_lockname[lock->l_req_mode], - lock->l_flags, nid, lock->l_remote_handle.cookie, - exp ? atomic_read(&exp->exp_refcount) : -99, - lock->l_pid, lock->l_callback_timeout, lock->l_lvb_type); + libcfs_debug_msg(msgdata, + "%pV ns: \?\? lock: %p/%#llx lrc: %d/%d,%d mode: %s/%s res: \?\? rrc=\?\? type: \?\?\? flags: %#llx nid: %s remote: %#llx expref: %d pid: %u timeout: %lld lvb_type: %d\n", + &vaf, + lock, + lock->l_handle.h_cookie, + refcount_read(&lock->l_handle.h_ref), + lock->l_readers, lock->l_writers, + ldlm_lockname[lock->l_granted_mode], + ldlm_lockname[lock->l_req_mode], + lock->l_flags, nid, + lock->l_remote_handle.cookie, + exp ? refcount_read(&exp->exp_handle.h_ref) : -99, + lock->l_pid, lock->l_callback_timestamp, + lock->l_lvb_type); va_end(args); return; } switch (resource->lr_type) { case LDLM_EXTENT: - libcfs_debug_vmsg2(msgdata, fmt, args, - " ns: %s lock: %p/"LPX64" lrc: %d/%d,%d mode: %s/%s " - "res: "DLDLMRES" rrc: %d type: %s ["LPU64"->"LPU64"] " - "(req "LPU64"->"LPU64") flags: "LPX64" nid: %s remote: " - LPX64" expref: %d pid: %u timeout: %lu lvb_type: %d\n", - ldlm_lock_to_ns_name(lock), lock, - lock->l_handle.h_cookie, atomic_read(&lock->l_refc), - lock->l_readers, lock->l_writers, - ldlm_lockname[lock->l_granted_mode], - ldlm_lockname[lock->l_req_mode], - PLDLMRES(resource), - atomic_read(&resource->lr_refcount), - ldlm_typename[resource->lr_type], - lock->l_policy_data.l_extent.start, - lock->l_policy_data.l_extent.end, - lock->l_req_extent.start, lock->l_req_extent.end, - lock->l_flags, nid, lock->l_remote_handle.cookie, - exp ? atomic_read(&exp->exp_refcount) : -99, - lock->l_pid, lock->l_callback_timeout, - lock->l_lvb_type); + libcfs_debug_msg(msgdata, + "%pV ns: %s lock: %p/%#llx lrc: %d/%d,%d mode: %s/%s res: " DLDLMRES " rrc: %d type: %s [%llu->%llu] (req %llu->%llu) flags: %#llx nid: %s remote: %#llx expref: %d pid: %u timeout: %lld lvb_type: %d\n", + &vaf, + ldlm_lock_to_ns_name(lock), lock, + lock->l_handle.h_cookie, + refcount_read(&lock->l_handle.h_ref), + lock->l_readers, lock->l_writers, + ldlm_lockname[lock->l_granted_mode], + ldlm_lockname[lock->l_req_mode], + PLDLMRES(resource), + atomic_read(&resource->lr_refcount), + ldlm_typename[resource->lr_type], + lock->l_policy_data.l_extent.start, + lock->l_policy_data.l_extent.end, + lock->l_req_extent.start, lock->l_req_extent.end, + lock->l_flags, nid, + lock->l_remote_handle.cookie, + exp ? refcount_read(&exp->exp_handle.h_ref) : -99, + lock->l_pid, lock->l_callback_timestamp, + lock->l_lvb_type); break; case LDLM_FLOCK: - libcfs_debug_vmsg2(msgdata, fmt, args, - " ns: %s lock: %p/"LPX64" lrc: %d/%d,%d mode: %s/%s " - "res: "DLDLMRES" rrc: %d type: %s pid: %d " - "["LPU64"->"LPU64"] flags: "LPX64" nid: %s " - "remote: "LPX64" expref: %d pid: %u timeout: %lu\n", - ldlm_lock_to_ns_name(lock), lock, - lock->l_handle.h_cookie, atomic_read(&lock->l_refc), - lock->l_readers, lock->l_writers, - ldlm_lockname[lock->l_granted_mode], - ldlm_lockname[lock->l_req_mode], - PLDLMRES(resource), - atomic_read(&resource->lr_refcount), - ldlm_typename[resource->lr_type], - lock->l_policy_data.l_flock.pid, - lock->l_policy_data.l_flock.start, - lock->l_policy_data.l_flock.end, - lock->l_flags, nid, lock->l_remote_handle.cookie, - exp ? atomic_read(&exp->exp_refcount) : -99, - lock->l_pid, lock->l_callback_timeout); + libcfs_debug_msg(msgdata, + "%pV ns: %s lock: %p/%#llx lrc: %d/%d,%d mode: %s/%s res: " DLDLMRES " rrc: %d type: %s pid: %d [%llu->%llu] flags: %#llx nid: %s remote: %#llx expref: %d pid: %u timeout: %lld\n", + &vaf, + ldlm_lock_to_ns_name(lock), lock, + lock->l_handle.h_cookie, + refcount_read(&lock->l_handle.h_ref), + lock->l_readers, lock->l_writers, + ldlm_lockname[lock->l_granted_mode], + ldlm_lockname[lock->l_req_mode], + PLDLMRES(resource), + atomic_read(&resource->lr_refcount), + ldlm_typename[resource->lr_type], + lock->l_policy_data.l_flock.pid, + lock->l_policy_data.l_flock.start, + lock->l_policy_data.l_flock.end, + lock->l_flags, nid, + lock->l_remote_handle.cookie, + exp ? refcount_read(&exp->exp_handle.h_ref) : -99, + lock->l_pid, lock->l_callback_timestamp); break; case LDLM_IBITS: - libcfs_debug_vmsg2(msgdata, fmt, args, - " ns: %s lock: %p/"LPX64" lrc: %d/%d,%d mode: %s/%s " - "res: "DLDLMRES" bits "LPX64" rrc: %d type: %s " - "flags: "LPX64" nid: %s remote: "LPX64" expref: %d " - "pid: %u timeout: %lu lvb_type: %d\n", - ldlm_lock_to_ns_name(lock), - lock, lock->l_handle.h_cookie, - atomic_read(&lock->l_refc), - lock->l_readers, lock->l_writers, - ldlm_lockname[lock->l_granted_mode], - ldlm_lockname[lock->l_req_mode], - PLDLMRES(resource), - lock->l_policy_data.l_inodebits.bits, - atomic_read(&resource->lr_refcount), - ldlm_typename[resource->lr_type], - lock->l_flags, nid, lock->l_remote_handle.cookie, - exp ? atomic_read(&exp->exp_refcount) : -99, - lock->l_pid, lock->l_callback_timeout, - lock->l_lvb_type); + libcfs_debug_msg(msgdata, + "%pV ns: %s lock: %p/%#llx lrc: %d/%d,%d mode: %s/%s res: " DLDLMRES " bits %#llx/%#llx rrc: %d type: %s flags: %#llx nid: %s remote: %#llx expref: %d pid: %u timeout: %lld lvb_type: %d\n", + &vaf, + ldlm_lock_to_ns_name(lock), + lock, lock->l_handle.h_cookie, + refcount_read(&lock->l_handle.h_ref), + lock->l_readers, lock->l_writers, + ldlm_lockname[lock->l_granted_mode], + ldlm_lockname[lock->l_req_mode], + PLDLMRES(resource), + lock->l_policy_data.l_inodebits.bits, + lock->l_policy_data.l_inodebits.try_bits, + atomic_read(&resource->lr_refcount), + ldlm_typename[resource->lr_type], + lock->l_flags, nid, + lock->l_remote_handle.cookie, + exp ? refcount_read(&exp->exp_handle.h_ref) : -99, + lock->l_pid, lock->l_callback_timestamp, + lock->l_lvb_type); break; default: - libcfs_debug_vmsg2(msgdata, fmt, args, - " ns: %s lock: %p/"LPX64" lrc: %d/%d,%d mode: %s/%s " - "res: "DLDLMRES" rrc: %d type: %s flags: "LPX64" " - "nid: %s remote: "LPX64" expref: %d pid: %u " - "timeout: %lu lvb_type: %d\n", - ldlm_lock_to_ns_name(lock), - lock, lock->l_handle.h_cookie, - atomic_read(&lock->l_refc), - lock->l_readers, lock->l_writers, - ldlm_lockname[lock->l_granted_mode], - ldlm_lockname[lock->l_req_mode], - PLDLMRES(resource), - atomic_read(&resource->lr_refcount), - ldlm_typename[resource->lr_type], - lock->l_flags, nid, lock->l_remote_handle.cookie, - exp ? atomic_read(&exp->exp_refcount) : -99, - lock->l_pid, lock->l_callback_timeout, - lock->l_lvb_type); + libcfs_debug_msg(msgdata, + "%pV ns: %s lock: %p/%#llx lrc: %d/%d,%d mode: %s/%s res: " DLDLMRES " rrc: %d type: %s flags: %#llx nid: %s remote: %#llx expref: %d pid: %u timeout: %lld lvb_type: %d\n", + &vaf, + ldlm_lock_to_ns_name(lock), + lock, lock->l_handle.h_cookie, + refcount_read(&lock->l_handle.h_ref), + lock->l_readers, lock->l_writers, + ldlm_lockname[lock->l_granted_mode], + ldlm_lockname[lock->l_req_mode], + PLDLMRES(resource), + atomic_read(&resource->lr_refcount), + ldlm_typename[resource->lr_type], + lock->l_flags, nid, + lock->l_remote_handle.cookie, + exp ? refcount_read(&exp->exp_handle.h_ref) : -99, + lock->l_pid, lock->l_callback_timestamp, + lock->l_lvb_type); break; } va_end(args); + ldlm_resource_putref(resource); } EXPORT_SYMBOL(_ldlm_lock_debug);