X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fldlm%2Fldlm_lock.c;h=44680dd8d75cb68be6d0976f9e023733113c15f8;hp=e43f5c16b746c78e77f3926a386226bba7fcf304;hb=511ea5850f2553d14891aeda7972d0526c67a3af;hpb=381604c5c45c7f7394185f434bb8ad1dd4ed88af diff --git a/lustre/ldlm/ldlm_lock.c b/lustre/ldlm/ldlm_lock.c index e43f5c1..44680dd 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, 2013, Intel Corporation. + * Copyright (c) 2010, 2017, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -41,16 +37,16 @@ #define DEBUG_SUBSYSTEM S_LDLM -#ifdef __KERNEL__ -# include -# include -#else -# include -#endif +#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] = "--", @@ -71,19 +67,11 @@ char *ldlm_typename[] = { [LDLM_FLOCK] = "FLK", [LDLM_IBITS] = "IBT", }; -EXPORT_SYMBOL(ldlm_typename); - -static ldlm_policy_wire_to_local_t ldlm_policy_wire18_to_local[] = { - [LDLM_PLAIN - LDLM_MIN_TYPE] = ldlm_plain_policy_wire_to_local, - [LDLM_EXTENT - LDLM_MIN_TYPE] = ldlm_extent_policy_wire_to_local, - [LDLM_FLOCK - LDLM_MIN_TYPE] = ldlm_flock_policy_wire18_to_local, - [LDLM_IBITS - LDLM_MIN_TYPE] = ldlm_ibits_policy_wire_to_local, -}; -static ldlm_policy_wire_to_local_t ldlm_policy_wire21_to_local[] = { +static ldlm_policy_wire_to_local_t ldlm_policy_wire_to_local[] = { [LDLM_PLAIN - LDLM_MIN_TYPE] = ldlm_plain_policy_wire_to_local, [LDLM_EXTENT - LDLM_MIN_TYPE] = ldlm_extent_policy_wire_to_local, - [LDLM_FLOCK - LDLM_MIN_TYPE] = ldlm_flock_policy_wire21_to_local, + [LDLM_FLOCK - LDLM_MIN_TYPE] = ldlm_flock_policy_wire_to_local, [LDLM_IBITS - LDLM_MIN_TYPE] = ldlm_ibits_policy_wire_to_local, }; @@ -97,62 +85,54 @@ 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; - int new_client; - /** some badness for 2.0.0 clients, but 2.0.0 isn't supported */ - new_client = (exp_connect_flags(exp) & OBD_CONNECT_FULL20) != 0; - if (new_client) - convert = ldlm_policy_wire21_to_local[type - LDLM_MIN_TYPE]; - else - convert = ldlm_policy_wire18_to_local[type - LDLM_MIN_TYPE]; + convert = ldlm_policy_wire_to_local[type - LDLM_MIN_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); @@ -162,9 +142,7 @@ extern struct kmem_cache *ldlm_lock_slab; static ldlm_processing_policy ldlm_processing_policy_table[] = { [LDLM_PLAIN] = ldlm_process_plain_lock, [LDLM_EXTENT] = ldlm_process_extent_lock, -# ifdef __KERNEL__ [LDLM_FLOCK] = ldlm_process_flock_lock, -# endif [LDLM_IBITS] = ldlm_process_inodebits_lock, }; @@ -220,8 +198,9 @@ void ldlm_lock_put(struct ldlm_lock *lock) res = lock->l_resource; LASSERT(ldlm_is_destroyed(lock)); - LASSERT(cfs_list_empty(&lock->l_res_link)); - LASSERT(cfs_list_empty(&lock->l_pending_chain)); + 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); @@ -251,11 +230,13 @@ EXPORT_SYMBOL(ldlm_lock_put); int ldlm_lock_remove_from_lru_nolock(struct ldlm_lock *lock) { int rc = 0; - if (!cfs_list_empty(&lock->l_lru)) { + if (!list_empty(&lock->l_lru)) { struct ldlm_namespace *ns = ldlm_lock_to_ns(lock); LASSERT(lock->l_resource->lr_type != LDLM_FLOCK); - cfs_list_del_init(&lock->l_lru); + 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--; rc = 1; @@ -265,23 +246,33 @@ 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)) { - LASSERT(cfs_list_empty(&lock->l_lru)); + LASSERT(list_empty(&lock->l_lru)); RETURN(0); } 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); } /** @@ -291,11 +282,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(); - LASSERT(cfs_list_empty(&lock->l_lru)); + lock->l_last_used = ktime_get(); + LASSERT(list_empty(&lock->l_lru)); LASSERT(lock->l_resource->lr_type != LDLM_FLOCK); - cfs_list_add_tail(&lock->l_lru, &ns->ns_unused_list); - ldlm_clear_skipped(lock); + list_add_tail(&lock->l_lru, &ns->ns_unused_list); LASSERT(ns->ns_nr_unused >= 0); ns->ns_nr_unused++; } @@ -325,13 +315,13 @@ void ldlm_lock_touch_in_lru(struct ldlm_lock *lock) ENTRY; if (ldlm_is_ns_srv(lock)) { - LASSERT(cfs_list_empty(&lock->l_lru)); + LASSERT(list_empty(&lock->l_lru)); EXIT; return; } spin_lock(&ns->ns_lock); - if (!cfs_list_empty(&lock->l_lru)) { + if (!list_empty(&lock->l_lru)) { ldlm_lock_remove_from_lru_nolock(lock); ldlm_lock_add_to_lru_nolock(lock); } @@ -358,7 +348,7 @@ void ldlm_lock_touch_in_lru(struct ldlm_lock *lock) * ldlm_lock_destroy, you can never drop your final references on this lock. * Because it's not in the hash table anymore. -phil */ -int ldlm_lock_destroy_internal(struct ldlm_lock *lock) +static int ldlm_lock_destroy_internal(struct ldlm_lock *lock) { ENTRY; @@ -367,13 +357,13 @@ int ldlm_lock_destroy_internal(struct ldlm_lock *lock) LBUG(); } - if (!cfs_list_empty(&lock->l_res_link)) { + if (!list_empty(&lock->l_res_link)) { LDLM_ERROR(lock, "lock still on resource"); LBUG(); } if (ldlm_is_destroyed(lock)) { - LASSERT(cfs_list_empty(&lock->l_lru)); + LASSERT(list_empty(&lock->l_lru)); EXIT; return 0; } @@ -392,16 +382,6 @@ 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; } @@ -453,7 +433,7 @@ static void lock_handle_free(void *lock, int size) OBD_SLAB_FREE(lock, ldlm_lock_slab, size); } -struct portals_handle_ops lock_handle_ops = { +static struct portals_handle_ops lock_handle_ops = { .hop_addref = lock_handle_addref, .hop_free = lock_handle_free, }; @@ -483,34 +463,35 @@ static struct ldlm_lock *ldlm_lock_new(struct ldlm_resource *resource) lu_ref_add(&resource->lr_reference, "lock", lock); atomic_set(&lock->l_refc, 2); - CFS_INIT_LIST_HEAD(&lock->l_res_link); - CFS_INIT_LIST_HEAD(&lock->l_lru); - CFS_INIT_LIST_HEAD(&lock->l_pending_chain); - CFS_INIT_LIST_HEAD(&lock->l_bl_ast); - CFS_INIT_LIST_HEAD(&lock->l_cp_ast); - CFS_INIT_LIST_HEAD(&lock->l_rk_ast); + INIT_LIST_HEAD(&lock->l_res_link); + INIT_LIST_HEAD(&lock->l_lru); + INIT_LIST_HEAD(&lock->l_pending_chain); + INIT_LIST_HEAD(&lock->l_bl_ast); + INIT_LIST_HEAD(&lock->l_cp_ast); + INIT_LIST_HEAD(&lock->l_rk_ast); init_waitqueue_head(&lock->l_waitq); lock->l_blocking_lock = NULL; - CFS_INIT_LIST_HEAD(&lock->l_sl_mode); - CFS_INIT_LIST_HEAD(&lock->l_sl_policy); - CFS_INIT_HLIST_NODE(&lock->l_exp_hash); - CFS_INIT_HLIST_NODE(&lock->l_exp_flock_hash); + INIT_LIST_HEAD(&lock->l_sl_mode); + INIT_LIST_HEAD(&lock->l_sl_policy); + INIT_HLIST_NODE(&lock->l_exp_hash); + INIT_HLIST_NODE(&lock->l_exp_flock_hash); lprocfs_counter_incr(ldlm_res_to_ns(resource)->ns_stats, LDLM_NSS_LOCKS); - CFS_INIT_LIST_HEAD(&lock->l_handle.h_link); + INIT_LIST_HEAD(&lock->l_handle.h_link); class_handle_hash(&lock->l_handle, &lock_handle_ops); lu_ref_init(&lock->l_reference); lu_ref_add(&lock->l_reference, "hash", lock); lock->l_callback_timeout = 0; + lock->l_activity = 0; #if LUSTRE_TRACKS_LOCK_EXP_REFS - CFS_INIT_LIST_HEAD(&lock->l_exp_refs_link); + INIT_LIST_HEAD(&lock->l_exp_refs_link); lock->l_exp_refs_nr = 0; lock->l_exp_refs_target = NULL; #endif - CFS_INIT_LIST_HEAD(&lock->l_exp_list); + INIT_LIST_HEAD(&lock->l_exp_list); RETURN(lock); } @@ -541,14 +522,14 @@ int ldlm_lock_change_resource(struct ldlm_namespace *ns, struct ldlm_lock *lock, LASSERT(new_resid->name[0] != 0); /* This function assumes that the lock isn't on any lists */ - LASSERT(cfs_list_empty(&lock->l_res_link)); + LASSERT(list_empty(&lock->l_res_link)); type = oldres->lr_type; unlock_res_and_lock(lock); - newres = ldlm_resource_get(ns, NULL, new_resid, type, 1); - if (newres == NULL) - RETURN(-ENOMEM); + newres = ldlm_resource_get(ns, NULL, new_resid, type, 1); + if (IS_ERR(newres)) + RETURN(PTR_ERR(newres)); lu_ref_add(&newres->lr_reference, "lock", lock); /* @@ -578,7 +559,6 @@ int ldlm_lock_change_resource(struct ldlm_namespace *ns, struct ldlm_lock *lock, RETURN(0); } -EXPORT_SYMBOL(ldlm_lock_change_resource); /** \defgroup ldlm_handles LDLM HANDLES * Ways to get hold of locks without any addresses. @@ -613,6 +593,13 @@ struct ldlm_lock *__ldlm_handle2lock(const struct lustre_handle *handle, 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)) { @@ -655,61 +642,21 @@ EXPORT_SYMBOL(__ldlm_handle2lock); */ void ldlm_lock2desc(struct ldlm_lock *lock, struct ldlm_lock_desc *desc) { - struct obd_export *exp = lock->l_export ?: lock->l_conn_export; - - /* INODEBITS_INTEROP: If the other side does not support - * inodebits, reply with a plain lock descriptor. */ - if ((lock->l_resource->lr_type == LDLM_IBITS) && - (exp && !(exp_connect_flags(exp) & OBD_CONNECT_IBITS))) { - /* Make sure all the right bits are set in this lock we - are going to pass to client */ - LASSERTF(lock->l_policy_data.l_inodebits.bits == - (MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE | - MDS_INODELOCK_LAYOUT), - "Inappropriate inode lock bits during " - "conversion " LPU64 "\n", - lock->l_policy_data.l_inodebits.bits); - - ldlm_res2desc(lock->l_resource, &desc->l_resource); - desc->l_resource.lr_type = LDLM_PLAIN; - - /* Convert "new" lock mode to something old client can - understand */ - if ((lock->l_req_mode == LCK_CR) || - (lock->l_req_mode == LCK_CW)) - desc->l_req_mode = LCK_PR; - else - desc->l_req_mode = lock->l_req_mode; - if ((lock->l_granted_mode == LCK_CR) || - (lock->l_granted_mode == LCK_CW)) { - desc->l_granted_mode = LCK_PR; - } else { - /* We never grant PW/EX locks to clients */ - LASSERT((lock->l_granted_mode != LCK_PW) && - (lock->l_granted_mode != LCK_EX)); - desc->l_granted_mode = lock->l_granted_mode; - } - - /* We do not copy policy here, because there is no - policy for plain locks */ - } else { - ldlm_res2desc(lock->l_resource, &desc->l_resource); - desc->l_req_mode = lock->l_req_mode; - desc->l_granted_mode = lock->l_granted_mode; - ldlm_convert_policy_to_wire(lock->l_resource->lr_type, - &lock->l_policy_data, - &desc->l_policy_data); - } + ldlm_res2desc(lock->l_resource, &desc->l_resource); + desc->l_req_mode = lock->l_req_mode; + desc->l_granted_mode = lock->l_granted_mode; + ldlm_convert_policy_to_wire(lock->l_resource->lr_type, + &lock->l_policy_data, + &desc->l_policy_data); } -EXPORT_SYMBOL(ldlm_lock2desc); /** * Add a lock to list of conflicting locks to send AST to. * * Only add if we have not sent a blocking AST to the lock yet. */ -void ldlm_add_bl_work_item(struct ldlm_lock *lock, struct ldlm_lock *new, - cfs_list_t *work_list) +static void ldlm_add_bl_work_item(struct ldlm_lock *lock, struct ldlm_lock *new, + struct list_head *work_list) { if (!ldlm_is_ast_sent(lock)) { LDLM_DEBUG(lock, "lock incompatible; sending blocking AST."); @@ -718,8 +665,8 @@ 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(cfs_list_empty(&lock->l_bl_ast)); - cfs_list_add(&lock->l_bl_ast, work_list); + 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); @@ -729,13 +676,14 @@ void ldlm_add_bl_work_item(struct ldlm_lock *lock, struct ldlm_lock *new, /** * Add a lock to list of just granted locks to send completion AST to. */ -void ldlm_add_cp_work_item(struct ldlm_lock *lock, cfs_list_t *work_list) +static void ldlm_add_cp_work_item(struct ldlm_lock *lock, + struct list_head *work_list) { if (!ldlm_is_cp_reqd(lock)) { ldlm_set_cp_reqd(lock); LDLM_DEBUG(lock, "lock granted; sending completion AST."); - LASSERT(cfs_list_empty(&lock->l_cp_ast)); - cfs_list_add(&lock->l_cp_ast, work_list); + LASSERT(list_empty(&lock->l_cp_ast)); + list_add(&lock->l_cp_ast, work_list); LDLM_LOCK_GET(lock); } } @@ -747,7 +695,7 @@ void ldlm_add_cp_work_item(struct ldlm_lock *lock, cfs_list_t *work_list) * Must be called with lr_lock held. */ void ldlm_add_ast_work_item(struct ldlm_lock *lock, struct ldlm_lock *new, - cfs_list_t *work_list) + struct list_head *work_list) { ENTRY; check_res_locked(lock->l_resource); @@ -763,14 +711,14 @@ 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; + struct ldlm_lock *lock; - lock = ldlm_handle2lock(lockh); - LASSERT(lock != NULL); - ldlm_lock_addref_internal(lock, mode); - LDLM_LOCK_PUT(lock); + lock = ldlm_handle2lock(lockh); + LASSERTF(lock != NULL, "Non-existing lock: %#llx\n", lockh->cookie); + ldlm_lock_addref_internal(lock, mode); + LDLM_LOCK_PUT(lock); } EXPORT_SYMBOL(ldlm_lock_addref); @@ -781,7 +729,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)) { @@ -805,7 +754,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; @@ -831,11 +780,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); } /** @@ -845,7 +794,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)) { @@ -871,7 +821,7 @@ 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; @@ -882,18 +832,23 @@ void ldlm_lock_decref_internal(struct ldlm_lock *lock, __u32 mode) ldlm_lock_decref_internal_nolock(lock, mode); - 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"); + 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 */ ldlm_set_cbpending(lock); - } + } - if (!lock->l_readers && !lock->l_writers && - ldlm_is_cbpending(lock)) { - /* If we received a blocked AST and this was the last reference, - * run the callback. */ + if (!lock->l_readers && !lock->l_writers && ldlm_is_cbpending(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) CERROR("FL_CBPENDING set on non-local lock--just a " "warning\n"); @@ -913,7 +868,8 @@ void ldlm_lock_decref_internal(struct ldlm_lock *lock, __u32 mode) } else if (ns_is_client(ns) && !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"); @@ -942,10 +898,10 @@ void ldlm_lock_decref_internal(struct ldlm_lock *lock, __u32 mode) /** * 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); } @@ -956,9 +912,9 @@ EXPORT_SYMBOL(ldlm_lock_decref); * \a lockh and mark it for subsequent cancellation once r/w refcount * drops to zero instead of putting into LRU. * - * Typical usage is for GROUP locks which we cannot allow to be cached. */ -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; @@ -975,9 +931,9 @@ void ldlm_lock_decref_and_cancel(struct lustre_handle *lockh, __u32 mode) EXPORT_SYMBOL(ldlm_lock_decref_and_cancel); struct sl_insert_point { - cfs_list_t *res_link; - cfs_list_t *mode_link; - cfs_list_t *policy_link; + struct list_head *res_link; + struct list_head *mode_link; + struct list_head *policy_link; }; /** @@ -994,18 +950,18 @@ struct sl_insert_point { * NOTE: called by * - ldlm_grant_lock_with_skiplist */ -static void search_granted_lock(cfs_list_t *queue, +static void search_granted_lock(struct list_head *queue, struct ldlm_lock *req, struct sl_insert_point *prev) { - cfs_list_t *tmp; + struct list_head *tmp; struct ldlm_lock *lock, *mode_end, *policy_end; ENTRY; - cfs_list_for_each(tmp, queue) { - lock = cfs_list_entry(tmp, struct ldlm_lock, l_res_link); + list_for_each(tmp, queue) { + lock = list_entry(tmp, struct ldlm_lock, l_res_link); - mode_end = cfs_list_entry(lock->l_sl_mode.prev, + mode_end = list_entry(lock->l_sl_mode.prev, struct ldlm_lock, l_sl_mode); if (lock->l_req_mode != req->l_req_mode) { @@ -1025,7 +981,7 @@ static void search_granted_lock(cfs_list_t *queue, } else if (lock->l_resource->lr_type == LDLM_IBITS) { for (;;) { policy_end = - cfs_list_entry(lock->l_sl_policy.prev, + list_entry(lock->l_sl_policy.prev, struct ldlm_lock, l_sl_policy); @@ -1049,7 +1005,7 @@ static void search_granted_lock(cfs_list_t *queue, /* go to next policy group within mode group */ tmp = policy_end->l_res_link.next; - lock = cfs_list_entry(tmp, struct ldlm_lock, + lock = list_entry(tmp, struct ldlm_lock, l_res_link); } /* loop over policy groups within the mode group */ @@ -1095,20 +1051,20 @@ static void ldlm_granted_list_add_lock(struct ldlm_lock *lock, return; } - LASSERT(cfs_list_empty(&lock->l_res_link)); - LASSERT(cfs_list_empty(&lock->l_sl_mode)); - LASSERT(cfs_list_empty(&lock->l_sl_policy)); + LASSERT(list_empty(&lock->l_res_link)); + LASSERT(list_empty(&lock->l_sl_mode)); + LASSERT(list_empty(&lock->l_sl_policy)); /* * lock->link == prev->link means lock is first starting the group. * Don't re-add to itself to suppress kernel warnings. */ if (&lock->l_res_link != prev->res_link) - cfs_list_add(&lock->l_res_link, prev->res_link); + list_add(&lock->l_res_link, prev->res_link); if (&lock->l_sl_mode != prev->mode_link) - cfs_list_add(&lock->l_sl_mode, prev->mode_link); + list_add(&lock->l_sl_mode, prev->mode_link); if (&lock->l_sl_policy != prev->policy_link) - cfs_list_add(&lock->l_sl_policy, prev->policy_link); + list_add(&lock->l_sl_policy, prev->policy_link); EXIT; } @@ -1117,16 +1073,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(lock->l_req_mode == lock->l_granted_mode); - 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); } /** @@ -1136,11 +1090,10 @@ 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 */ -void ldlm_grant_lock(struct ldlm_lock *lock, cfs_list_t *work_list) +void ldlm_grant_lock(struct ldlm_lock *lock, struct list_head *work_list) { struct ldlm_resource *res = lock->l_resource; ENTRY; @@ -1152,113 +1105,192 @@ void ldlm_grant_lock(struct ldlm_lock *lock, cfs_list_t *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. + * Describe the overlap between two locks. itree_overlap_cb data. + */ +struct lock_match_data { + struct ldlm_lock *lmd_old; + struct ldlm_lock *lmd_lock; + enum ldlm_mode *lmd_mode; + union ldlm_policy_data *lmd_policy; + __u64 lmd_flags; + int lmd_unref; +}; + +/** + * 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(cfs_list_t *queue, - ldlm_mode_t *mode, - ldlm_policy_data_t *policy, - struct ldlm_lock *old_lock, - __u64 flags, int unref) -{ - struct ldlm_lock *lock; - cfs_list_t *tmp; +static int lock_matches(struct ldlm_lock *lock, struct lock_match_data *data) +{ + union ldlm_policy_data *lpol = &lock->l_policy_data; + enum ldlm_mode match; + + 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_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; + match = lock->l_req_mode; + + 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; + + 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: + ; + } - cfs_list_for_each(tmp, queue) { - ldlm_mode_t match; + /* We match if we have existing lock with same or wider set + of bits. */ + if (!data->lmd_unref && LDLM_HAVE_MASK(lock, GONE)) + return INTERVAL_ITER_CONT; - lock = cfs_list_entry(tmp, struct ldlm_lock, l_res_link); + if (!equi(data->lmd_flags & LDLM_FL_LOCAL_ONLY, ldlm_is_local(lock))) + return INTERVAL_ITER_CONT; - if (lock == old_lock) - break; + 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); + } - /* Check if this lock can be matched. - * Used by LU-2919(exclusive open) for open lease lock */ - if (ldlm_is_excl(lock)) - continue; + *data->lmd_mode = match; + data->lmd_lock = lock; - /* 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; + return INTERVAL_ITER_STOP; +} - if (!(lock->l_req_mode & *mode)) - continue; - match = lock->l_req_mode; +static unsigned int itree_overlap_cb(struct interval_node *in, void *args) +{ + struct ldlm_interval *node = to_ldlm_interval(in); + struct lock_match_data *data = args; + struct ldlm_lock *lock; + int rc; - 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; + 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; +} - if (unlikely(match == LCK_GROUP) && - lock->l_resource->lr_type == LDLM_EXTENT && - lock->l_policy_data.l_extent.gid != policy->l_extent.gid) - continue; +/** + * 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. + */ +static struct ldlm_lock *search_itree(struct ldlm_resource *res, + struct lock_match_data *data) +{ + struct interval_node_extent ext = { + .start = data->lmd_policy->l_extent.start, + .end = data->lmd_policy->l_extent.end + }; + int idx; - /* 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; + for (idx = 0; idx < LCK_MODE_NUM; idx++) { + struct ldlm_interval_tree *tree = &res->lr_itree[idx]; - if (!unref && LDLM_HAVE_MASK(lock, GONE)) - continue; + if (tree->lit_root == NULL) + continue; - if ((flags & LDLM_FL_LOCAL_ONLY) && - !ldlm_is_local(lock)) - continue; + if (!(tree->lit_mode & *data->lmd_mode)) + continue; - 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; - } + interval_search(tree->lit_root, &ext, + itree_overlap_cb, data); + } + return data->lmd_lock; +} + + +/** + * 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 lock_match_data *data) +{ + struct ldlm_lock *lock; + int rc; - return 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) @@ -1276,7 +1308,6 @@ void ldlm_lock_fail_match(struct ldlm_lock *lock) ldlm_lock_fail_match_locked(lock); unlock_res_and_lock(lock); } -EXPORT_SYMBOL(ldlm_lock_fail_match); /** * Mark lock as "matchable" by OST. @@ -1334,49 +1365,56 @@ 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(struct ldlm_namespace *ns, __u64 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, int unref) +{ + struct lock_match_data data = { + .lmd_old = NULL, + .lmd_lock = NULL, + .lmd_mode = &mode, + .lmd_policy = policy, + .lmd_flags = flags, + .lmd_unref = unref, + }; + struct ldlm_resource *res; + struct ldlm_lock *lock; + int rc = 0; + ENTRY; - if (ns == NULL) { - old_lock = ldlm_handle2lock(lockh); - LASSERT(old_lock); + if (ns == NULL) { + data.lmd_old = ldlm_handle2lock(lockh); + LASSERT(data.lmd_old != NULL); - 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; - } + 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 (res == NULL) { - LASSERT(old_lock == NULL); - RETURN(0); - } + res = ldlm_resource_get(ns, NULL, res_id, type, 0); + if (IS_ERR(res)) { + 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); + 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 != NULL) + GOTO(out, rc = 1); + if (flags & LDLM_FL_BLOCK_GRANTED) + GOTO(out, rc = 0); + lock = search_queue(&res->lr_waiting, &data); + if (lock != NULL) + GOTO(out, rc = 1); EXIT; out: @@ -1424,7 +1462,7 @@ ldlm_mode_t ldlm_lock_match(struct ldlm_namespace *ns, __u64 flags, } out2: if (rc) { - LDLM_DEBUG(lock, "matched ("LPU64" "LPU64")", + 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) ? @@ -1444,32 +1482,32 @@ ldlm_mode_t ldlm_lock_match(struct ldlm_namespace *ns, __u64 flags, } 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, + "%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 (old_lock) - LDLM_LOCK_PUT(old_lock); + if (data.lmd_old != NULL) + LDLM_LOCK_PUT(data.lmd_old); - return rc ? mode : 0; + return rc ? mode : 0; } EXPORT_SYMBOL(ldlm_lock_match); -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) @@ -1583,7 +1621,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); } @@ -1596,63 +1634,89 @@ int ldlm_fill_lvb(struct ldlm_lock *lock, struct req_capsule *pill, * Returns a referenced lock */ 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, - const struct ldlm_callback_suite *cbs, + const struct ldlm_res_id *res_id, + enum ldlm_type type, + enum ldlm_mode mode, + const struct ldlm_callback_suite *cbs, void *data, __u32 lvb_len, enum lvb_type lvb_type) { - struct ldlm_lock *lock; - struct ldlm_resource *res; - ENTRY; - - res = ldlm_resource_get(ns, NULL, res_id, type, 1); - if (res == NULL) - RETURN(NULL); + struct ldlm_lock *lock; + struct ldlm_resource *res; + int rc; + ENTRY; - lock = ldlm_lock_new(res); + res = ldlm_resource_get(ns, NULL, res_id, type, 1); + if (IS_ERR(res)) + RETURN(ERR_CAST(res)); - if (lock == NULL) - RETURN(NULL); + lock = ldlm_lock_new(res); + if (lock == NULL) + RETURN(ERR_PTR(-ENOMEM)); lock->l_req_mode = mode; lock->l_ast_data = data; lock->l_pid = current_pid(); if (ns_is_server(ns)) ldlm_set_ns_srv(lock); - if (cbs) { - lock->l_blocking_ast = cbs->lcs_blocking; - lock->l_completion_ast = cbs->lcs_completion; - 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, 0); - } + if (cbs) { + lock->l_blocking_ast = cbs->lcs_blocking; + lock->l_completion_ast = cbs->lcs_completion; + lock->l_glimpse_ast = cbs->lcs_glimpse; + } - if (lvb_len) { - lock->l_lvb_len = lvb_len; - OBD_ALLOC_LARGE(lock->l_lvb_data, lvb_len); - if (lock->l_lvb_data == NULL) - GOTO(out, 0); - } + 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); + + if (lvb_len) { + lock->l_lvb_len = lvb_len; + OBD_ALLOC_LARGE(lock->l_lvb_data, lvb_len); + if (lock->l_lvb_data == NULL) + GOTO(out, rc = -ENOMEM); + } lock->l_lvb_type = lvb_type; - if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_NEW_LOCK)) - GOTO(out, 0); + if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_NEW_LOCK)) + GOTO(out, rc = -ENOENT); - RETURN(lock); + RETURN(lock); out: - ldlm_lock_destroy(lock); - LDLM_LOCK_RELEASE(lock); - return NULL; + ldlm_lock_destroy(lock); + LDLM_LOCK_RELEASE(lock); + 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; + struct list_head rpc_list = LIST_HEAD_INIT(rpc_list); + ldlm_processing_policy policy; + ENTRY; + + policy = ldlm_processing_policy_table[res->lr_type]; +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. * @@ -1663,21 +1727,17 @@ 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) -{ - struct ldlm_lock *lock = *lockp; - struct ldlm_resource *res = lock->l_resource; - int local = ns_is_client(ldlm_res_to_ns(res)); -#ifdef HAVE_SERVER_SUPPORT - ldlm_processing_policy policy; -#endif - ldlm_error_t rc = ELDLM_OK; - struct ldlm_interval *node = NULL; - ENTRY; +enum ldlm_error ldlm_lock_enqueue(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)); + enum ldlm_error rc = ELDLM_OK; + struct ldlm_interval *node = NULL; + ENTRY; - lock->l_last_activity = cfs_time_current_sec(); /* 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) { @@ -1694,6 +1754,16 @@ ldlm_error_t ldlm_lock_enqueue(struct ldlm_namespace *ns, } *flags |= LDLM_FL_LOCK_CHANGED; RETURN(0); + } else if (rc != ELDLM_OK && + lock->l_req_mode == lock->l_granted_mode) { + 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); @@ -1701,6 +1771,19 @@ ldlm_error_t ldlm_lock_enqueue(struct ldlm_namespace *ns, } } + if (*flags & LDLM_FL_RESENT) { + /* Reconstruct LDLM_FL_SRV_ENQ_MASK @flags for reply. + * Set LOCK_CHANGED always. + * Check if the lock is granted for BLOCK_GRANTED. + * 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) + *flags |= LDLM_FL_BLOCK_GRANTED; + *flags |= lock->l_flags & LDLM_FL_NO_TIMEOUT; + RETURN(ELDLM_OK); + } + /* 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 @@ -1713,8 +1796,7 @@ ldlm_error_t ldlm_lock_enqueue(struct ldlm_namespace *ns, /* 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. */ - *flags &= ~(LDLM_FL_BLOCK_GRANTED | - LDLM_FL_BLOCK_CONV | LDLM_FL_BLOCK_WAIT); + *flags &= ~LDLM_FL_BLOCKED_MASK; GOTO(out, rc = ELDLM_OK); } @@ -1725,7 +1807,7 @@ ldlm_error_t ldlm_lock_enqueue(struct ldlm_namespace *ns, GOTO(out, rc = -ENOMEM); } - CFS_INIT_LIST_HEAD(&node->li_group); + INIT_LIST_HEAD(&node->li_group); ldlm_interval_attach(node, lock); node = NULL; } @@ -1736,6 +1818,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 @@ -1747,33 +1833,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); - 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_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_GRANTED) { - ldlm_grant_lock(lock, NULL); + } 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 " @@ -1796,35 +1876,164 @@ out: * * Must be called with resource lock held. */ -int ldlm_reprocess_queue(struct ldlm_resource *res, cfs_list_t *queue, - cfs_list_t *work_list) +int ldlm_reprocess_queue(struct ldlm_resource *res, struct list_head *queue, + struct list_head *work_list, + enum ldlm_process_intention intention) { - cfs_list_t *tmp, *pos; - ldlm_processing_policy policy; + struct list_head *tmp, *pos; + 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; + struct list_head bl_ast_list = LIST_HEAD_INIT(bl_ast_list); + ENTRY; - check_res_locked(res); + check_res_locked(res); + + policy = ldlm_processing_policy_table[res->lr_type]; + LASSERT(policy); + LASSERT(intention == LDLM_PROCESS_RESCAN || + intention == LDLM_PROCESS_RECOVERY); - policy = ldlm_processing_policy_table[res->lr_type]; - LASSERT(policy); +restart: + list_for_each_safe(tmp, pos, queue) { + struct ldlm_lock *pending; + struct list_head rpc_list = LIST_HEAD_INIT(rpc_list); - cfs_list_for_each_safe(tmp, pos, queue) { - struct ldlm_lock *pending; - pending = cfs_list_entry(tmp, struct ldlm_lock, l_res_link); + 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); +} + +/** + * 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. + */ +int ldlm_handle_conflict_lock(struct ldlm_lock *lock, __u64 *flags, + struct list_head *rpc_list) +{ + struct ldlm_resource *res = lock->l_resource; + int rc; + ENTRY; + + check_res_locked(res); + + /* 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 (lock->l_granted_mode == lock->l_req_mode) { + /* 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); + } + + RETURN(rc); + } + *flags |= LDLM_FL_BLOCK_GRANTED; + + RETURN(0); } + +/** + * 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 list_head *tmp, *pos; + ENTRY; + + list_for_each_safe(pos, tmp, bl_list) { + struct ldlm_lock *lock = + list_entry(pos, struct ldlm_lock, l_bl_ast); + + list_del_init(&lock->l_bl_ast); + LASSERT(ldlm_is_ast_sent(lock)); + ldlm_clear_ast_sent(lock); + LASSERT(lock->l_bl_ast_run == 0); + LASSERT(lock->l_blocking_lock); + LDLM_LOCK_RELEASE(lock->l_blocking_lock); + lock->l_blocking_lock = NULL; + LDLM_LOCK_RELEASE(lock); + } + EXIT; +} + #endif /** @@ -1839,14 +2048,14 @@ ldlm_work_bl_ast_lock(struct ptlrpc_request_set *rqset, void *opaq) struct ldlm_lock *lock; ENTRY; - if (cfs_list_empty(arg->list)) + if (list_empty(arg->list)) RETURN(-ENOENT); - lock = cfs_list_entry(arg->list->next, struct ldlm_lock, l_bl_ast); + lock = list_entry(arg->list->next, struct ldlm_lock, l_bl_ast); /* nobody should touch l_bl_ast */ lock_res_and_lock(lock); - cfs_list_del_init(&lock->l_bl_ast); + list_del_init(&lock->l_bl_ast); LASSERT(ldlm_is_ast_sent(lock)); LASSERT(lock->l_bl_ast_run == 0); @@ -1855,6 +2064,13 @@ ldlm_work_bl_ast_lock(struct ptlrpc_request_set *rqset, void *opaq) unlock_res_and_lock(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; rc = lock->l_blocking_ast(lock, &d, (void *)arg, LDLM_CB_BLOCKING); LDLM_LOCK_RELEASE(lock->l_blocking_lock); @@ -1876,10 +2092,10 @@ ldlm_work_cp_ast_lock(struct ptlrpc_request_set *rqset, void *opaq) ldlm_completion_callback completion_callback; ENTRY; - if (cfs_list_empty(arg->list)) + if (list_empty(arg->list)) RETURN(-ENOENT); - lock = cfs_list_entry(arg->list->next, struct ldlm_lock, l_cp_ast); + 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 @@ -1894,7 +2110,7 @@ ldlm_work_cp_ast_lock(struct ptlrpc_request_set *rqset, void *opaq) /* nobody should touch l_cp_ast */ lock_res_and_lock(lock); - cfs_list_del_init(&lock->l_cp_ast); + 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 */ @@ -1921,11 +2137,11 @@ ldlm_work_revoke_ast_lock(struct ptlrpc_request_set *rqset, void *opaq) struct ldlm_lock *lock; ENTRY; - if (cfs_list_empty(arg->list)) + if (list_empty(arg->list)) RETURN(-ENOENT); - lock = cfs_list_entry(arg->list->next, struct ldlm_lock, l_rk_ast); - cfs_list_del_init(&lock->l_rk_ast); + lock = list_entry(arg->list->next, struct ldlm_lock, l_rk_ast); + list_del_init(&lock->l_rk_ast); /* the desc just pretend to exclusive */ ldlm_lock2desc(lock, &desc); @@ -1949,25 +2165,28 @@ int ldlm_work_gl_ast_lock(struct ptlrpc_request_set *rqset, void *opaq) int rc = 0; ENTRY; - if (cfs_list_empty(arg->list)) + if (list_empty(arg->list)) RETURN(-ENOENT); - gl_work = cfs_list_entry(arg->list->next, struct ldlm_glimpse_work, + gl_work = list_entry(arg->list->next, struct ldlm_glimpse_work, gl_list); - cfs_list_del_init(&gl_work->gl_list); + list_del_init(&gl_work->gl_list); lock = gl_work->gl_lock; /* 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; 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); @@ -1979,14 +2198,14 @@ int ldlm_work_gl_ast_lock(struct ptlrpc_request_set *rqset, void *opaq) * Used on server to send multiple ASTs together instead of sending one by * one. */ -int ldlm_run_ast_work(struct ldlm_namespace *ns, cfs_list_t *rpc_list, +int ldlm_run_ast_work(struct ldlm_namespace *ns, struct list_head *rpc_list, ldlm_desc_ast_t ast_type) { struct ldlm_cb_set_arg *arg; set_producer_func work_ast_lock; int rc; - if (cfs_list_empty(rpc_list)) + if (list_empty(rpc_list)) RETURN(0); OBD_ALLOC_PTR(arg); @@ -2036,75 +2255,52 @@ 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, - cfs_hlist_node_t *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; -} -EXPORT_SYMBOL(ldlm_reprocess_all_ns); - /** * 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) { - CFS_LIST_HEAD(rpc_list); - + struct list_head rpc_list; #ifdef HAVE_SERVER_SUPPORT + struct obd_device *obd; int rc; ENTRY; + + INIT_LIST_HEAD(&rpc_list); /* 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) { - LASSERT(cfs_list_empty(&rpc_list)); - goto restart; - } + lock_res(res); + ldlm_reprocess_queue(res, &res->lr_waiting, &rpc_list, intention); + 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; + } #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"); @@ -2113,8 +2309,38 @@ restart: #endif EXIT; } + +void ldlm_reprocess_all(struct ldlm_resource *res) +{ + __ldlm_reprocess_all(res, LDLM_PROCESS_RESCAN); +} 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); + 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. @@ -2132,8 +2358,19 @@ void ldlm_cancel_callback(struct ldlm_lock *lock) } else { LDLM_DEBUG(lock, "no blocking ast"); } - } - ldlm_set_bl_done(lock); + + /* 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)) { + struct l_wait_info lwi = { 0 }; + + /* The lock is guaranteed to have been canceled once + * returning from this function. */ + unlock_res_and_lock(lock); + l_wait_event(lock->l_waitq, is_bl_done(lock), &lwi); + lock_res_and_lock(lock); + } } /** @@ -2145,8 +2382,8 @@ void ldlm_unlink_lock_skiplist(struct ldlm_lock *req) req->l_resource->lr_type != LDLM_IBITS) return; - cfs_list_del_init(&req->l_sl_policy); - cfs_list_del_init(&req->l_sl_mode); + list_del_init(&req->l_sl_policy); + list_del_init(&req->l_sl_mode); } /** @@ -2167,6 +2404,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(); } @@ -2199,7 +2437,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; @@ -2221,225 +2459,165 @@ struct export_cl_data { 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); + 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. */ -int ldlm_cancel_locks_for_export_cb(cfs_hash_t *hs, cfs_hash_bd_t *bd, - cfs_hlist_node_t *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 export_cl_data ecl = { .ecl_exp = exp, .ecl_loop = 0, }; - cfs_hash_for_each_empty(exp->exp_lock_hash, - ldlm_cancel_locks_for_export_cb, &ecl); + 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); + } + + 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 = { + .ecl_exp = exp, + .ecl_loop = 0, + }; - LASSERT(lock->l_granted_mode & (LCK_PW | LCK_EX)); - LASSERT(new_mode == LCK_COS); + cfs_hash_for_each_empty(exp->exp_lock_hash, + ldlm_cancel_locks_for_export_cb, &ecl); - 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); + 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)); - lock->l_req_mode = new_mode; - ldlm_grant_lock(lock, NULL); - unlock_res_and_lock(lock); - ldlm_reprocess_all(lock->l_resource); + 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); - EXIT; + 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 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 only for now. * - * While lock conversion is not currently used, future client-side - * optimizations could take advantage of it to avoid discarding cached - * pages on a file. + * \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) { - CFS_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; - /* Just return if mode is unchanged. */ - if (new_mode == lock->l_granted_mode) { - *flags |= LDLM_FL_BLOCK_GRANTED; - RETURN(lock->l_resource); - } - - /* 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); - - LASSERTF((new_mode == LCK_PW && lock->l_granted_mode == LCK_PR), - "new_mode %u, granted %u\n", new_mode, lock->l_granted_mode); + LASSERT(new_mode == LCK_COS); lock_res_and_lock(lock); - res = lock->l_resource; - ns = ldlm_res_to_ns(res); + if (!(lock->l_granted_mode & (LCK_PW | LCK_EX))) { + unlock_res_and_lock(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 */ - CFS_INIT_LIST_HEAD(&node->li_group); - ldlm_interval_attach(node, lock); - node = NULL; - } - } + LASSERT(lock->l_granted_mode == LCK_MINMODE); + LDLM_DEBUG(lock, "lock was canceled before downgrade"); + RETURN_EXIT; + } - /* - * Remove old lock from the pool before adding the lock with new - * mode below in ->policy() - */ - ldlm_pool_del(&ns->ns_pool, 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); + lock->l_req_mode = new_mode; + ldlm_grant_lock(lock, NULL); - /* 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(); + unlock_res_and_lock(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(); - } -#endif - unlock_res_and_lock(lock); + ldlm_reprocess_all(lock->l_resource); - 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); + EXIT; } -EXPORT_SYMBOL(ldlm_lock_convert); +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; @@ -2466,23 +2644,33 @@ 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; 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); 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 " + " 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", lock, lock->l_handle.h_cookie, atomic_read(&lock->l_refc), @@ -2499,10 +2687,10 @@ void _ldlm_lock_debug(struct ldlm_lock *lock, 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", + " 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", ldlm_lock_to_ns_name(lock), lock, lock->l_handle.h_cookie, atomic_read(&lock->l_refc), lock->l_readers, lock->l_writers, @@ -2522,10 +2710,10 @@ void _ldlm_lock_debug(struct ldlm_lock *lock, case LDLM_FLOCK: libcfs_debug_vmsg2(msgdata, fmt, args, - " ns: %s lock: %p/"LPX64" lrc: %d/%d,%d mode: %s/%s " + " ns: %s lock: %p/%#llx 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", + "[%llu->%llu] flags: %#llx nid: %s " + "remote: %#llx expref: %d pid: %u timeout: %lld\n", ldlm_lock_to_ns_name(lock), lock, lock->l_handle.h_cookie, atomic_read(&lock->l_refc), lock->l_readers, lock->l_writers, @@ -2544,10 +2732,10 @@ void _ldlm_lock_debug(struct ldlm_lock *lock, 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", + " 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", ldlm_lock_to_ns_name(lock), lock, lock->l_handle.h_cookie, atomic_read(&lock->l_refc), @@ -2556,6 +2744,7 @@ void _ldlm_lock_debug(struct ldlm_lock *lock, 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, @@ -2566,10 +2755,10 @@ void _ldlm_lock_debug(struct ldlm_lock *lock, 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", + " 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", ldlm_lock_to_ns_name(lock), lock, lock->l_handle.h_cookie, atomic_read(&lock->l_refc), @@ -2586,5 +2775,6 @@ void _ldlm_lock_debug(struct ldlm_lock *lock, break; } va_end(args); + ldlm_resource_putref(resource); } EXPORT_SYMBOL(_ldlm_lock_debug);