X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fldlm%2Fldlm_flock.c;h=a765cc7547103d28f2cf5d4374bf2bc509d8b3c9;hp=f61e300e32a810f6512ec3539ecb85a6c1c2d9d0;hb=fef75d4c76745e8cd35df36e4691ed66c9e5902f;hpb=9fb46705ae86aa2c0ac29427f0ff24f923560eb7 diff --git a/lustre/ldlm/ldlm_flock.c b/lustre/ldlm/ldlm_flock.c index f61e300..a765cc7 100644 --- a/lustre/ldlm/ldlm_flock.c +++ b/lustre/ldlm/ldlm_flock.c @@ -31,13 +31,29 @@ * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. * - * Copyright (c) 2010, 2011, Whamcloud, Inc. + * Copyright (c) 2010, 2012, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ * Lustre is a trademark of Sun Microsystems, Inc. */ +/** + * This file implements POSIX lock type for Lustre. + * Its policy properties are start and end of extent and PID. + * + * These locks are only done through MDS due to POSIX semantics requiring + * e.g. that locks could be only partially released and as such split into + * two parts, and also that two adjacent locks from the same process may be + * merged into a single wider lock. + * + * Lock modes are mapped like this: + * PR and PW for READ and WRITE locks + * NL to request a releasing of a portion of the lock + * + * These flock locks never timeout. + */ + #define DEBUG_SUBSYSTEM S_LDLM #ifdef __KERNEL__ @@ -84,20 +100,12 @@ ldlm_flocks_overlap(struct ldlm_lock *lock, struct ldlm_lock *new) lock->l_policy_data.l_flock.start)); } -static inline int ldlm_flock_blocking_link(struct ldlm_lock *req, - struct ldlm_lock *lock) +static inline void ldlm_flock_blocking_link(struct ldlm_lock *req, + struct ldlm_lock *lock) { - int rc = 0; - /* For server only */ if (req->l_export == NULL) - return 0; - - if (unlikely(req->l_export->exp_flock_hash == NULL)) { - rc = ldlm_init_flock_export(req->l_export); - if (rc) - goto error; - } + return; LASSERT(cfs_hlist_unhashed(&req->l_exp_flock_hash)); @@ -110,8 +118,6 @@ static inline int ldlm_flock_blocking_link(struct ldlm_lock *req, cfs_hash_add(req->l_export->exp_flock_hash, &req->l_policy_data.l_flock.owner, &req->l_exp_flock_hash); -error: - return rc; } static inline void ldlm_flock_blocking_unlink(struct ldlm_lock *req) @@ -154,6 +160,15 @@ ldlm_flock_destroy(struct ldlm_lock *lock, ldlm_mode_t mode, __u64 flags) EXIT; } +/** + * POSIX locks deadlock detection code. + * + * Given a new lock \a req and an existing lock \a bl_lock it conflicts + * with, we need to iterate through all blocked POSIX locks for this + * export and see if there is a deadlock condition arising. (i.e. when + * one client holds a lock on something and want a lock on something + * else and at the same time another client has the opposite situation). + */ static int ldlm_flock_deadlock(struct ldlm_lock *req, struct ldlm_lock *bl_lock) { @@ -178,6 +193,7 @@ ldlm_flock_deadlock(struct ldlm_lock *req, struct ldlm_lock *bl_lock) if (lock == NULL) break; + LASSERT(req != lock); flock = &lock->l_policy_data.l_flock; LASSERT(flock->owner == bl_owner); bl_owner = flock->blocking_owner; @@ -197,6 +213,24 @@ ldlm_flock_deadlock(struct ldlm_lock *req, struct ldlm_lock *bl_lock) return 0; } +/** + * Process a granting attempt for flock lock. + * Must be called under ns lock held. + * + * This function looks for any conflicts for \a lock in the granted or + * waiting queues. The lock is granted if no conflicts are found in + * either queue. + * + * It is also responsible for splitting a lock if a portion of the lock + * is released. + * + * If \a first_enq is 0 (ie, called from ldlm_reprocess_queue): + * - blocking ASTs have already been sent + * + * If \a first_enq is 1 (ie, called from ldlm_lock_enqueue): + * - blocking ASTs have not been sent yet, so list of conflicting locks + * would be collected and ASTs sent. + */ int ldlm_process_flock_lock(struct ldlm_lock *req, __u64 *flags, int first_enq, ldlm_error_t *err, cfs_list_t *work_list) @@ -214,7 +248,6 @@ ldlm_process_flock_lock(struct ldlm_lock *req, __u64 *flags, int first_enq, int overlaps = 0; int splitted = 0; const struct ldlm_callback_suite null_cbs = { NULL }; - int rc; ENTRY; CDEBUG(D_DLMTRACE, "flags %#llx owner "LPU64" pid %u mode %u start " @@ -291,18 +324,17 @@ reprocess: RETURN(LDLM_ITER_STOP); } - if (ldlm_flock_deadlock(req, lock)) { - ldlm_flock_destroy(req, mode, *flags); - *err = -EDEADLK; - RETURN(LDLM_ITER_STOP); - } + /* add lock to blocking list before deadlock + * check to prevent race */ + ldlm_flock_blocking_link(req, lock); - rc = ldlm_flock_blocking_link(req, lock); - if (rc) { + if (ldlm_flock_deadlock(req, lock)) { + ldlm_flock_blocking_unlink(req); ldlm_flock_destroy(req, mode, *flags); - *err = rc; + *err = -EDEADLK; RETURN(LDLM_ITER_STOP); } + ldlm_resource_add_lock(res, &res->lr_waiting, req); *flags |= LDLM_FL_BLOCK_GRANTED; RETURN(LDLM_ITER_STOP); @@ -417,9 +449,9 @@ reprocess: * and restart processing this lock. */ if (!new2) { unlock_res_and_lock(req); - new2 = ldlm_lock_create(ns, &res->lr_name, LDLM_FLOCK, - lock->l_granted_mode, &null_cbs, - NULL, 0); + new2 = ldlm_lock_create(ns, &res->lr_name, LDLM_FLOCK, + lock->l_granted_mode, &null_cbs, + NULL, 0, LVB_T_NONE); lock_res_and_lock(req); if (!new2) { ldlm_flock_destroy(req, lock->l_granted_mode, @@ -515,10 +547,10 @@ restart: #endif /* HAVE_SERVER_SUPPORT */ } - /* In case we're reprocessing the requested lock we can't destroy - * it until after calling ldlm_ast_work_item() above so that lawi() - * can bump the reference count on req. Otherwise req could be freed - * before the completion AST can be sent. */ + /* In case we're reprocessing the requested lock we can't destroy + * it until after calling ldlm_add_ast_work_item() above so that laawi() + * can bump the reference count on \a req. Otherwise \a req + * could be freed before the completion AST can be sent. */ if (added) ldlm_flock_destroy(req, mode, *flags); @@ -543,7 +575,7 @@ ldlm_flock_interrupted_wait(void *data) lock_res_and_lock(lock); ldlm_flock_blocking_unlink(lock); - /* client side - set flag to prevent lock from being put on lru list */ + /* client side - set flag to prevent lock from being put on LRU list */ lock->l_flags |= LDLM_FL_CBPENDING; unlock_res_and_lock(lock); @@ -551,7 +583,7 @@ ldlm_flock_interrupted_wait(void *data) } /** - * Flock completion calback function. + * Flock completion callback function. * * \param lock [in,out]: A lock to be handled * \param flags [in]: flags @@ -563,7 +595,7 @@ ldlm_flock_interrupted_wait(void *data) int ldlm_flock_completion_ast(struct ldlm_lock *lock, __u64 flags, void *data) { - cfs_flock_t *getlk = lock->l_ast_data; + struct file_lock *getlk = lock->l_ast_data; struct obd_device *obd; struct obd_import *imp = NULL; struct ldlm_flock_wait_data fwd; @@ -632,10 +664,10 @@ ldlm_flock_completion_ast(struct ldlm_lock *lock, __u64 flags, void *data) granted: OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_CP_CB_WAIT, 10); - if (lock->l_destroyed) { - LDLM_DEBUG(lock, "client-side enqueue waking up: destroyed"); - RETURN(0); - } + if (lock->l_flags & LDLM_FL_DESTROYED) { + LDLM_DEBUG(lock, "client-side enqueue waking up: destroyed"); + RETURN(0); + } if (lock->l_flags & LDLM_FL_FAILED) { LDLM_DEBUG(lock, "client-side enqueue waking up: failed"); @@ -662,33 +694,32 @@ granted: /* fcntl(F_GETLK) request */ /* The old mode was saved in getlk->fl_type so that if the mode * in the lock changes we can decref the appropriate refcount.*/ - ldlm_flock_destroy(lock, cfs_flock_type(getlk), - LDLM_FL_WAIT_NOREPROC); - switch (lock->l_granted_mode) { - case LCK_PR: - cfs_flock_set_type(getlk, F_RDLCK); - break; - case LCK_PW: - cfs_flock_set_type(getlk, F_WRLCK); - break; - default: - cfs_flock_set_type(getlk, F_UNLCK); - } - cfs_flock_set_pid(getlk, - (pid_t)lock->l_policy_data.l_flock.pid); - cfs_flock_set_start(getlk, - (loff_t)lock->l_policy_data.l_flock.start); - cfs_flock_set_end(getlk, - (loff_t)lock->l_policy_data.l_flock.end); - } else { + ldlm_flock_destroy(lock, flock_type(getlk), + LDLM_FL_WAIT_NOREPROC); + switch (lock->l_granted_mode) { + case LCK_PR: + flock_set_type(getlk, F_RDLCK); + break; + case LCK_PW: + flock_set_type(getlk, F_WRLCK); + break; + default: + flock_set_type(getlk, F_UNLCK); + } + flock_set_pid(getlk, (pid_t)lock->l_policy_data.l_flock.pid); + flock_set_start(getlk, + (loff_t)lock->l_policy_data.l_flock.start); + flock_set_end(getlk, + (loff_t)lock->l_policy_data.l_flock.end); + } else { __u64 noreproc = LDLM_FL_WAIT_NOREPROC; - /* We need to reprocess the lock to do merges or splits - * with existing locks owned by this process. */ - ldlm_process_flock_lock(lock, &noreproc, 1, &err, NULL); - } - unlock_res_and_lock(lock); - RETURN(0); + /* We need to reprocess the lock to do merges or splits + * with existing locks owned by this process. */ + ldlm_process_flock_lock(lock, &noreproc, 1, &err, NULL); + } + unlock_res_and_lock(lock); + RETURN(0); } EXPORT_SYMBOL(ldlm_flock_completion_ast); @@ -816,6 +847,9 @@ static cfs_hash_ops_t ldlm_export_flock_ops = { int ldlm_init_flock_export(struct obd_export *exp) { + if( strcmp(exp->exp_obd->obd_type->typ_name, LUSTRE_MDT_NAME) != 0) + RETURN(0); + exp->exp_flock_hash = cfs_hash_create(obd_uuid2str(&exp->exp_client_uuid), HASH_EXP_LOCK_CUR_BITS,