-/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
- * vim:expandtab:shiftwidth=8:tabstop=8:
- *
+/*
* GPL HEADER START
*
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
/*
* Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
+ *
+ * Copyright (c) 2011, 2012, Whamcloud, Inc.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
{
struct osc_lock *ols = cl2osc_lock(slice);
- LASSERT(ols->ols_state == OLS_GRANTED ||
- ols->ols_state == OLS_UPCALL_RECEIVED);
LINVRNT(osc_lock_invariant(ols));
- if (ols->ols_glimpse) {
- LASSERT(ols->ols_hold == 0);
+ switch (ols->ols_state) {
+ case OLS_NEW:
+ LASSERT(!ols->ols_hold);
+ LASSERT(ols->ols_agl);
return 0;
+ case OLS_UPCALL_RECEIVED:
+ osc_lock_unhold(ols);
+ case OLS_ENQUEUED:
+ LASSERT(!ols->ols_hold);
+ osc_lock_detach(env, ols);
+ ols->ols_state = OLS_NEW;
+ return 0;
+ case OLS_GRANTED:
+ LASSERT(!ols->ols_glimpse);
+ LASSERT(ols->ols_hold);
+ /*
+ * Move lock into OLS_RELEASED state before calling
+ * osc_cancel_base() so that possible synchronous cancellation
+ * (that always happens e.g., for liblustre) sees that lock is
+ * released.
+ */
+ ols->ols_state = OLS_RELEASED;
+ return osc_lock_unhold(ols);
+ default:
+ CERROR("Impossible state: %d\n", ols->ols_state);
+ LBUG();
}
- LASSERT(ols->ols_hold);
-
- /*
- * Move lock into OLS_RELEASED state before calling osc_cancel_base()
- * so that possible synchronous cancellation (that always happens
- * e.g., for liblustre) sees that lock is released.
- */
- ols->ols_state = OLS_RELEASED;
- return osc_lock_unhold(ols);
}
static void osc_lock_fini(const struct lu_env *env,
ENTRY;
- if (!(olck->ols_flags & LDLM_FL_LVB_READY)) {
- EXIT;
- return;
- }
+ if (!(olck->ols_flags & LDLM_FL_LVB_READY))
+ RETURN_EXIT;
lvb = &olck->ols_lvb;
obj = olck->ols_cl.cls_obj;
dlmlock->l_ast_data = NULL;
olck->ols_handle.cookie = 0ULL;
cfs_spin_unlock(&osc_ast_guard);
+ ldlm_lock_fail_match_locked(dlmlock);
unlock_res_and_lock(dlmlock);
LDLM_LOCK_PUT(dlmlock);
}
rc = 0;
}
- if (rc == 0)
- /* on error, lock was signaled by cl_lock_error() */
+ if (rc == 0) {
+ /* For AGL case, the RPC sponsor may exits the cl_lock
+ * processing without wait() called before related OSC
+ * lock upcall(). So update the lock status according
+ * to the enqueue result inside AGL upcall(). */
+ if (olck->ols_agl) {
+ lock->cll_flags |= CLF_FROM_UPCALL;
+ cl_wait_try(env, lock);
+ lock->cll_flags &= ~CLF_FROM_UPCALL;
+ if (!olck->ols_glimpse)
+ olck->ols_agl = 0;
+ }
cl_lock_signal(env, lock);
- else
+ /* del user for lock upcall cookie */
+ cl_unuse_try(env, lock);
+ } else {
+ /* del user for lock upcall cookie */
+ cl_lock_user_del(env, lock);
cl_lock_error(env, lock, rc);
+ }
- cl_lock_mutex_put(env, lock);
-
- /* release cookie reference, acquired by osc_lock_enqueue() */
- lu_ref_del(&lock->cll_reference, "upcall", lock);
- cl_lock_put(env, lock);
- cl_env_nested_put(&nest, env);
- } else
- /* should never happen, similar to osc_ldlm_blocking_ast(). */
- LBUG();
- RETURN(errcode);
+ /* release cookie reference, acquired by osc_lock_enqueue() */
+ cl_lock_hold_release(env, lock, "upcall", lock);
+ cl_lock_mutex_put(env, lock);
+
+ lu_ref_del(&lock->cll_reference, "upcall", lock);
+ /* This maybe the last reference, so must be called after
+ * cl_lock_mutex_put(). */
+ cl_lock_put(env, lock);
+
+ cl_env_nested_put(&nest, env);
+ } else {
+ /* should never happen, similar to osc_ldlm_blocking_ast(). */
+ LBUG();
+ }
+ RETURN(errcode);
}
/**
env = cl_env_nested_get(&nest);
if (!IS_ERR(env)) {
- /*
- * osc_ast_data_get() has to go after environment is
+ /* osc_ast_data_get() has to go after environment is
* allocated, because osc_ast_data() acquires a
* reference to a lock, and it can only be released in
* environment.
olck = osc_ast_data_get(dlmlock);
if (olck != NULL) {
lock = olck->ols_cl.cls_lock;
- cl_lock_mutex_get(env, lock);
+ /* Do not grab the mutex of cl_lock for glimpse.
+ * See LU-1274 for details.
+ * BTW, it's okay for cl_lock to be cancelled during
+ * this period because server can handle this race.
+ * See ldlm_server_glimpse_ast() for details.
+ * cl_lock_mutex_get(env, lock); */
cap = &req->rq_pill;
req_capsule_extend(cap, &RQF_LDLM_GL_CALLBACK);
req_capsule_set_size(cap, &RMF_DLM_LVB, RCL_SERVER,
obj = lock->cll_descr.cld_obj;
result = cl_object_glimpse(env, obj, lvb);
}
- cl_lock_mutex_put(env, lock);
osc_ast_data_put(env, olck);
} else {
/*
ENTRY;
LASSERT(cl_lock_is_mutexed(lock));
- LASSERT(lock->cll_state == CLS_QUEUING);
/* make it enqueue anyway for glimpse lock, because we actually
* don't need to cancel any conflicting locks. */
if (!lockless && osc_lock_compatible(olck, scan_ols))
continue;
- /* Now @scan is conflicting with @lock, this means current
- * thread have to sleep for @scan being destroyed. */
- if (scan_ols->ols_owner == osc_env_io(env)) {
- CERROR("DEADLOCK POSSIBLE!\n");
- CL_LOCK_DEBUG(D_ERROR, env, scan, "queued.\n");
- CL_LOCK_DEBUG(D_ERROR, env, lock, "queuing.\n");
- libcfs_debug_dumpstack(NULL);
- }
cl_lock_get_trust(scan);
conflict = scan;
break;
ENTRY;
LASSERT(cl_lock_is_mutexed(lock));
- LASSERT(lock->cll_state == CLS_QUEUING);
- LASSERT(ols->ols_state == OLS_NEW);
+ LASSERTF(ols->ols_state == OLS_NEW,
+ "Impossible state: %d\n", ols->ols_state);
ols->ols_flags = osc_enq2ldlm_flags(enqflags);
+ if (enqflags & CEF_AGL) {
+ ols->ols_flags |= LDLM_FL_BLOCK_NOWAIT;
+ ols->ols_agl = 1;
+ } else {
+ ols->ols_agl = 0;
+ }
if (ols->ols_flags & LDLM_FL_HAS_INTENT)
ols->ols_glimpse = 1;
if (!osc_lock_is_lockless(ols) && !(enqflags & CEF_MUST))
if (ols->ols_locklessable)
ols->ols_flags |= LDLM_FL_DENY_ON_CONTENTION;
- /* a reference for lock, passed as an upcall cookie */
- cl_lock_get(lock);
- lu_ref_add(&lock->cll_reference, "upcall", lock);
+ /* lock will be passed as upcall cookie,
+ * hold ref to prevent to be released. */
+ cl_lock_hold_add(env, lock, "upcall", lock);
+ /* a user for lock also */
+ cl_lock_user_add(env, lock);
ols->ols_state = OLS_ENQUEUED;
/*
obj->oo_oinfo->loi_kms_valid,
osc_lock_upcall,
ols, einfo, &ols->ols_handle,
- PTLRPCD_SET, 1);
+ PTLRPCD_SET, 1, ols->ols_agl);
if (result != 0) {
- lu_ref_del(&lock->cll_reference,
- "upcall", lock);
- cl_lock_put(env, lock);
+ cl_lock_user_del(env, lock);
+ cl_lock_unhold(env, lock, "upcall", lock);
+ if (unlikely(result == -ECANCELED)) {
+ ols->ols_state = OLS_NEW;
+ result = 0;
+ }
}
} else {
ols->ols_state = OLS_GRANTED;
struct cl_lock *lock = olck->ols_cl.cls_lock;
LINVRNT(osc_lock_invariant(olck));
- if (olck->ols_glimpse && olck->ols_state >= OLS_UPCALL_RECEIVED)
- return 0;
+
+ if (olck->ols_glimpse && olck->ols_state >= OLS_UPCALL_RECEIVED) {
+ if (olck->ols_flags & LDLM_FL_LVB_READY) {
+ return 0;
+ } else if (olck->ols_agl) {
+ if (lock->cll_flags & CLF_FROM_UPCALL)
+ /* It is from enqueue RPC reply upcall for
+ * updating state. Do not re-enqueue. */
+ return -ENAVAIL;
+ else
+ olck->ols_state = OLS_NEW;
+ } else {
+ LASSERT(lock->cll_error);
+ return lock->cll_error;
+ }
+ }
+
+ if (olck->ols_state == OLS_NEW) {
+ int rc;
+
+ LASSERT(olck->ols_agl);
+
+ rc = osc_lock_enqueue(env, slice, NULL, CEF_ASYNC | CEF_MUST);
+ if (rc != 0)
+ return rc;
+ else
+ return CLO_REENQUEUED;
+ }
LASSERT(equi(olck->ols_state >= OLS_UPCALL_RECEIVED &&
lock->cll_error == 0, olck->ols_lock != NULL));
static int osc_lock_flush(struct osc_lock *ols, int discard)
{
- struct cl_lock *lock = ols->ols_cl.cls_lock;
- struct cl_env_nest nest;
- struct lu_env *env;
- int result = 0;
+ struct cl_lock *lock = ols->ols_cl.cls_lock;
+ struct cl_env_nest nest;
+ struct lu_env *env;
+ int result = 0;
+ ENTRY;
+
+ env = cl_env_nested_get(&nest);
+ if (!IS_ERR(env)) {
+ struct osc_object *obj = cl2osc(ols->ols_cl.cls_obj);
+ struct cl_lock_descr *descr = &lock->cll_descr;
+ int rc = 0;
+
+ if (descr->cld_mode >= CLM_WRITE) {
+ result = osc_cache_writeback_range(env, obj,
+ descr->cld_start, descr->cld_end,
+ 1, discard);
+ CDEBUG(D_DLMTRACE, "write out %d pages for lock %p.\n",
+ result, lock);
+ if (result > 0)
+ result = 0;
+ }
+
+ rc = cl_lock_discard_pages(env, lock);
+ if (result == 0 && rc < 0)
+ result = rc;
- env = cl_env_nested_get(&nest);
- if (!IS_ERR(env)) {
- result = cl_lock_page_out(env, lock, discard);
cl_env_nested_put(&nest, env);
} else
result = PTR_ERR(env);
ols->ols_flush = 1;
LINVRNT(!osc_lock_has_pages(ols));
}
- return result;
+ RETURN(result);
}
/**
if (dlmlock != NULL) {
int do_cancel;
- discard = dlmlock->l_flags & LDLM_FL_DISCARD_DATA;
+ discard = !!(dlmlock->l_flags & LDLM_FL_DISCARD_DATA);
result = osc_lock_flush(olck, discard);
osc_lock_unhold(olck);
lock, result);
}
olck->ols_state = OLS_CANCELLED;
+ olck->ols_flags &= ~LDLM_FL_LVB_READY;
osc_lock_detach(env, olck);
}
-void cl_lock_page_list_fixup(const struct lu_env *env,
- struct cl_io *io, struct cl_lock *lock,
- struct cl_page_list *queue);
-
#ifdef INVARIANT_CHECK
+static int check_cb(const struct lu_env *env, struct cl_io *io,
+ struct cl_page *page, void *cbdata)
+{
+ struct cl_lock *lock = cbdata;
+
+ if (lock->cll_descr.cld_mode == CLM_READ) {
+ struct cl_lock *tmp;
+ tmp = cl_lock_at_page(env, lock->cll_descr.cld_obj,
+ page, lock, 1, 0);
+ if (tmp != NULL) {
+ cl_lock_put(env, tmp);
+ return CLP_GANG_OKAY;
+ }
+ }
+
+ CL_LOCK_DEBUG(D_ERROR, env, lock, "still has pages\n");
+ CL_PAGE_DEBUG(D_ERROR, env, page, "\n");
+ return CLP_GANG_ABORT;
+}
+
/**
* Returns true iff there are pages under \a olck not protected by other
* locks.
struct cl_lock_descr *descr;
struct cl_object *obj;
struct osc_object *oob;
- struct cl_page_list *plist;
- struct cl_page *page;
struct cl_env_nest nest;
struct cl_io *io;
struct lu_env *env;
int result;
env = cl_env_nested_get(&nest);
- if (!IS_ERR(env)) {
- obj = olck->ols_cl.cls_obj;
- oob = cl2osc(obj);
- io = &oob->oo_debug_io;
- lock = olck->ols_cl.cls_lock;
- descr = &lock->cll_descr;
- plist = &osc_env_info(env)->oti_plist;
- cl_page_list_init(plist);
-
- cfs_mutex_lock(&oob->oo_debug_mutex);
-
- io->ci_obj = cl_object_top(obj);
- cl_io_init(env, io, CIT_MISC, io->ci_obj);
- cl_page_gang_lookup(env, obj, io,
- descr->cld_start, descr->cld_end, plist, 0,
- NULL);
- cl_lock_page_list_fixup(env, io, lock, plist);
- if (plist->pl_nr > 0) {
- CL_LOCK_DEBUG(D_ERROR, env, lock, "still has pages\n");
- cl_page_list_for_each(page, plist)
- CL_PAGE_DEBUG(D_ERROR, env, page, "\n");
- }
- result = plist->pl_nr > 0;
- cl_page_list_disown(env, io, plist);
- cl_page_list_fini(env, plist);
- cl_io_fini(env, io);
- cfs_mutex_unlock(&oob->oo_debug_mutex);
- cl_env_nested_put(&nest, env);
- } else
- result = 0;
- return result;
+ if (IS_ERR(env))
+ return 0;
+
+ obj = olck->ols_cl.cls_obj;
+ oob = cl2osc(obj);
+ io = &oob->oo_debug_io;
+ lock = olck->ols_cl.cls_lock;
+ descr = &lock->cll_descr;
+
+ cfs_mutex_lock(&oob->oo_debug_mutex);
+
+ io->ci_obj = cl_object_top(obj);
+ io->ci_ignore_layout = 1;
+ cl_io_init(env, io, CIT_MISC, io->ci_obj);
+ do {
+ result = cl_page_gang_lookup(env, obj, io,
+ descr->cld_start, descr->cld_end,
+ check_cb, (void *)lock);
+ if (result == CLP_GANG_ABORT)
+ break;
+ if (result == CLP_GANG_RESCHED)
+ cfs_cond_resched();
+ } while (result != CLP_GANG_OKAY);
+ cl_io_fini(env, io);
+ cfs_mutex_unlock(&oob->oo_debug_mutex);
+ cl_env_nested_put(&nest, env);
+
+ return (result == CLP_GANG_ABORT);
}
#else
static int osc_lock_has_pages(struct osc_lock *olck)
if (need->cld_enq_flags & CEF_NEVER)
return 0;
+ if (ols->ols_state >= OLS_CANCELLED)
+ return 0;
+
if (need->cld_mode == CLM_PHANTOM) {
+ if (ols->ols_agl)
+ return !(ols->ols_state > OLS_RELEASED);
+
/*
* Note: the QUEUED lock can't be matched here, otherwise
* it might cause the deadlocks.
git://git.whamcloud.com / fs / lustre-release.git / blobdiff