From 0bc249f65c5b20b5bb02af7cfa844f265363185d Mon Sep 17 00:00:00 2001 From: bobijam Date: Tue, 23 Dec 2008 07:03:29 +0000 Subject: [PATCH] Branch HEAD b=17046 i=johann i=oleg.drokin (green) Reconstruct flock completion ast to make it in line with ldlm_completion_ast(). --- lustre/ldlm/ldlm_flock.c | 104 +++++++++++++++++++++++++++----------------- lustre/ldlm/ldlm_internal.h | 14 ++++++ lustre/ldlm/ldlm_request.c | 30 ++++--------- 3 files changed, 86 insertions(+), 62 deletions(-) diff --git a/lustre/ldlm/ldlm_flock.c b/lustre/ldlm/ldlm_flock.c index 1089022..439a4f4 100644 --- a/lustre/ldlm/ldlm_flock.c +++ b/lustre/ldlm/ldlm_flock.c @@ -108,7 +108,8 @@ ldlm_flock_destroy(struct ldlm_lock *lock, ldlm_mode_t mode, int flags) LASSERT(list_empty(&lock->l_flock_waitq)); list_del_init(&lock->l_res_link); - if (flags == LDLM_FL_WAIT_NOREPROC) { + if (flags == LDLM_FL_WAIT_NOREPROC && + !(lock->l_flags & LDLM_FL_FAILED)) { /* client side - set a flag to prevent sending a CANCEL */ lock->l_flags |= LDLM_FL_LOCAL_ONLY | LDLM_FL_CBPENDING; @@ -378,7 +379,8 @@ reprocess: NULL, 0); lock_res_and_lock(req); if (!new2) { - ldlm_flock_destroy(req, lock->l_granted_mode, *flags); + ldlm_flock_destroy(req, lock->l_granted_mode, + *flags); *err = -ENOLCK; RETURN(LDLM_ITER_STOP); } @@ -399,14 +401,15 @@ reprocess: new2->l_conn_export = lock->l_conn_export; if (lock->l_export != NULL) { new2->l_export = class_export_get(lock->l_export); - if (new2->l_export->exp_lock_hash && + if (new2->l_export->exp_lock_hash && hlist_unhashed(&new2->l_exp_hash)) lustre_hash_add(new2->l_export->exp_lock_hash, &new2->l_remote_handle, &new2->l_exp_hash); } if (*flags == LDLM_FL_WAIT_NOREPROC) - ldlm_lock_addref_internal_nolock(new2, lock->l_granted_mode); + ldlm_lock_addref_internal_nolock(new2, + lock->l_granted_mode); /* insert new2 at lock */ ldlm_resource_add_lock(res, ownlocks, new2); @@ -502,17 +505,26 @@ ldlm_flock_interrupted_wait(void *data) EXIT; } +/** + * Flock completion calback function. + * + * \param lock [in,out]: A lock to be handled + * \param flags [in]: flags + * \param *data [in]: ldlm_work_cp_ast_lock() will use ldlm_cb_set_arg + * + * \retval 0 : success + * \retval <0 : failure + */ int ldlm_flock_completion_ast(struct ldlm_lock *lock, int flags, void *data) { - struct ldlm_namespace *ns; - cfs_flock_t *getlk = lock->l_ast_data; - struct ldlm_flock_wait_data fwd; - struct obd_device *obd; - struct obd_import *imp = NULL; - ldlm_error_t err; - int rc = 0; - struct l_wait_info lwi; + cfs_flock_t *getlk = lock->l_ast_data; + struct obd_device *obd; + struct obd_import *imp = NULL; + struct ldlm_flock_wait_data fwd; + struct l_wait_info lwi; + ldlm_error_t err; + int rc = 0; ENTRY; CDEBUG(D_DLMTRACE, "flags: 0x%x data: %p getlk: %p\n", @@ -527,7 +539,8 @@ ldlm_flock_completion_ast(struct ldlm_lock *lock, int flags, void *data) (LDLM_FL_FAILED|LDLM_FL_LOCAL_ONLY)) { unlock_res_and_lock(lock); if (lock->l_req_mode == lock->l_granted_mode && - lock->l_granted_mode != LCK_NL) + lock->l_granted_mode != LCK_NL && + NULL == data) ldlm_lock_decref_internal(lock, lock->l_req_mode); RETURN(0); } @@ -536,20 +549,25 @@ ldlm_flock_completion_ast(struct ldlm_lock *lock, int flags, void *data) LASSERT(flags != LDLM_FL_WAIT_NOREPROC); if (!(flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED | - LDLM_FL_BLOCK_CONV))) - goto granted; + LDLM_FL_BLOCK_CONV))) { + if (NULL == data) + /* mds granted the lock in the reply */ + goto granted; + /* CP AST RPC: lock get granted, wake it up */ + cfs_waitq_signal(&lock->l_waitq); + RETURN(0); + } LDLM_DEBUG(lock, "client-side enqueue returned a blocked lock, " "sleeping"); - fwd.fwd_lock = lock; obd = class_exp2obd(lock->l_conn_export); - /* if this is a local lock, then there is no import */ - if (obd != NULL) + /* if this is a local lock, there is no import */ + if (NULL != obd) imp = obd->u.cli.cl_import; - if (imp != NULL) { + if (NULL != imp) { spin_lock(&imp->imp_lock); fwd.fwd_generation = imp->imp_generation; spin_unlock(&imp->imp_lock); @@ -558,27 +576,31 @@ ldlm_flock_completion_ast(struct ldlm_lock *lock, int flags, void *data) lwi = LWI_TIMEOUT_INTR(0, NULL, ldlm_flock_interrupted_wait, &fwd); /* Go to sleep until the lock is granted. */ - rc = l_wait_event(lock->l_waitq, - ((lock->l_req_mode == lock->l_granted_mode) || - lock->l_destroyed), &lwi); + rc = l_wait_event(lock->l_waitq, is_granted_or_cancelled(lock), &lwi); - LDLM_DEBUG(lock, "client-side enqueue waking up: rc = %d", rc); - RETURN(rc); + if (rc) { + LDLM_DEBUG(lock, "client-side enqueue waking up: failed (%d)", + rc); + RETURN(rc); + } granted: OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_CP_CB_WAIT, 10); - LDLM_DEBUG(lock, "client-side enqueue granted"); - ns = lock->l_resource->lr_namespace; - lock_res_and_lock(lock); - /* before flock's complete ast gets here, the flock - * can possibly be freed by another thread - */ - if (lock->l_destroyed) { - LDLM_DEBUG(lock, "already destroyed by another thread"); + lock_res_and_lock(lock); + if (lock->l_destroyed || lock->l_flags & LDLM_FL_FAILED) { + LDLM_DEBUG(lock, "client-side enqueue waking up: destroyed"); unlock_res(lock->l_resource); - RETURN(0); + RETURN(-EIO); } + if (rc) { + LDLM_DEBUG(lock, "client-side enqueue waking up: failed (%d)", + rc); + unlock_res_and_lock(lock); + RETURN(rc); + } + + LDLM_DEBUG(lock, "client-side enqueue granted"); /* take lock off the deadlock detection waitq. */ spin_lock(&ldlm_flock_waitq_lock); @@ -591,8 +613,9 @@ granted: if (flags & LDLM_FL_TEST_LOCK) { /* 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 approprate refcount. */ - ldlm_flock_destroy(lock, cfs_flock_type(getlk), LDLM_FL_WAIT_NOREPROC); + * 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); @@ -603,17 +626,18 @@ granted: 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); + 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 { int 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); - if (flags == 0) - cfs_waitq_signal(&lock->l_waitq); } unlock_res_and_lock(lock); RETURN(0); diff --git a/lustre/ldlm/ldlm_internal.h b/lustre/ldlm/ldlm_internal.h index 5332b2f..5ff07e7 100644 --- a/lustre/ldlm/ldlm_internal.h +++ b/lustre/ldlm/ldlm_internal.h @@ -239,3 +239,17 @@ typedef enum ldlm_policy_res ldlm_policy_res_t; return rc; \ } \ struct __##var##__dummy_write {;} /* semicolon catcher */ + +static inline int is_granted_or_cancelled(struct ldlm_lock *lock) +{ + int ret = 0; + + lock_res_and_lock(lock); + if (((lock->l_req_mode == lock->l_granted_mode) && + !(lock->l_flags & LDLM_FL_CP_REQD)) || + (lock->l_flags & LDLM_FL_FAILED)) + ret = 1; + unlock_res_and_lock(lock); + + return ret; +} diff --git a/lustre/ldlm/ldlm_request.c b/lustre/ldlm/ldlm_request.c index 7b84a27..46e7607 100644 --- a/lustre/ldlm/ldlm_request.c +++ b/lustre/ldlm/ldlm_request.c @@ -122,20 +122,6 @@ int ldlm_get_enq_timeout(struct ldlm_lock *lock) } EXPORT_SYMBOL(ldlm_get_enq_timeout); -static int is_granted_or_cancelled(struct ldlm_lock *lock) -{ - int ret = 0; - - lock_res_and_lock(lock); - if (((lock->l_req_mode == lock->l_granted_mode) && - !(lock->l_flags & LDLM_FL_CP_REQD)) || - (lock->l_flags & LDLM_FL_FAILED)) - ret = 1; - unlock_res_and_lock(lock); - - return ret; -} - /** * Helper function for ldlm_completion_ast(), updating timings when lock is * actually granted. @@ -266,14 +252,14 @@ noreproc: spin_unlock(&imp->imp_lock); } - if (ns_is_client(lock->l_resource->lr_namespace) && + if (ns_is_client(lock->l_resource->lr_namespace) && OBD_FAIL_CHECK_RESET(OBD_FAIL_LDLM_INTR_CP_AST, OBD_FAIL_LDLM_CP_BL_RACE | OBD_FAIL_ONCE)) { lock->l_flags |= LDLM_FL_FAIL_LOC; rc = -EINTR; } else { /* Go to sleep until the lock is granted or cancelled. */ - rc = l_wait_event(lock->l_waitq, + rc = l_wait_event(lock->l_waitq, is_granted_or_cancelled(lock), &lwi); } @@ -460,20 +446,20 @@ static void failed_lock_cleanup(struct ldlm_namespace *ns, /* Set a flag to prevent us from sending a CANCEL (bug 407) */ lock_res_and_lock(lock); /* Check that lock is not granted or failed, we might race. */ - if ((lock->l_req_mode != lock->l_granted_mode) && + if ((lock->l_req_mode != lock->l_granted_mode) && !(lock->l_flags & LDLM_FL_FAILED)) { /* Make sure that this lock will not be found by raced - * bl_ast and -EINVAL reply is sent to server anyways. + * bl_ast and -EINVAL reply is sent to server anyways. * bug 17645 */ - lock->l_flags |= LDLM_FL_LOCAL_ONLY | LDLM_FL_FAILED | + lock->l_flags |= LDLM_FL_LOCAL_ONLY | LDLM_FL_FAILED | LDLM_FL_ATOMIC_CB; need_cancel = 1; } unlock_res_and_lock(lock); - + if (need_cancel) { - LDLM_DEBUG(lock, - "setting FL_LOCAL_ONLY | LDLM_FL_FAILED | " + LDLM_DEBUG(lock, + "setting FL_LOCAL_ONLY | LDLM_FL_FAILED | " "LDLM_FL_ATOMIC_CB"); ldlm_lock_decref_and_cancel(lockh, mode); } else { -- 1.8.3.1