Whamcloud - gitweb
git://git.whamcloud.com
/
fs
/
lustre-release.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
| inline |
side by side
Revert "b=19427 correct lmm_object_id and reserve fids for fid-on-OST."
[fs/lustre-release.git]
/
lustre
/
osc
/
osc_lock.c
diff --git
a/lustre/osc/osc_lock.c
b/lustre/osc/osc_lock.c
index
67baec8
..
26a2d2d
100644
(file)
--- a/
lustre/osc/osc_lock.c
+++ b/
lustre/osc/osc_lock.c
@@
-135,10
+135,10
@@
static void osc_lock_detach(const struct lu_env *env, struct osc_lock *olck)
{
struct ldlm_lock *dlmlock;
- spin_lock(&osc_ast_guard);
+
cfs_
spin_lock(&osc_ast_guard);
dlmlock = olck->ols_lock;
if (dlmlock == NULL) {
- spin_unlock(&osc_ast_guard);
+
cfs_
spin_unlock(&osc_ast_guard);
return;
}
@@
-147,7
+147,7
@@
static void osc_lock_detach(const struct lu_env *env, struct osc_lock *olck)
* call to osc_lock_detach() */
dlmlock->l_ast_data = NULL;
olck->ols_handle.cookie = 0ULL;
- spin_unlock(&osc_ast_guard);
+
cfs_
spin_unlock(&osc_ast_guard);
lock_res_and_lock(dlmlock);
if (dlmlock->l_granted_mode == dlmlock->l_req_mode) {
@@
-276,14
+276,14
@@
static int osc_enq2ldlm_flags(__u32 enqflags)
* Global spin-lock protecting consistency of ldlm_lock::l_ast_data
* pointers. Initialized in osc_init().
*/
-spinlock_t osc_ast_guard;
+
cfs_
spinlock_t osc_ast_guard;
static struct osc_lock *osc_ast_data_get(struct ldlm_lock *dlm_lock)
{
struct osc_lock *olck;
lock_res_and_lock(dlm_lock);
- spin_lock(&osc_ast_guard);
+
cfs_
spin_lock(&osc_ast_guard);
olck = dlm_lock->l_ast_data;
if (olck != NULL) {
struct cl_lock *lock = olck->ols_cl.cls_lock;
@@
-303,7
+303,7
@@
static struct osc_lock *osc_ast_data_get(struct ldlm_lock *dlm_lock)
} else
olck = NULL;
}
- spin_unlock(&osc_ast_guard);
+
cfs_
spin_unlock(&osc_ast_guard);
unlock_res_and_lock(dlm_lock);
return olck;
}
@@
-451,11
+451,11
@@
static void osc_lock_upcall0(const struct lu_env *env, struct osc_lock *olck)
LASSERT(dlmlock != NULL);
lock_res_and_lock(dlmlock);
- spin_lock(&osc_ast_guard);
+
cfs_
spin_lock(&osc_ast_guard);
LASSERT(dlmlock->l_ast_data == olck);
LASSERT(olck->ols_lock == NULL);
olck->ols_lock = dlmlock;
- spin_unlock(&osc_ast_guard);
+
cfs_
spin_unlock(&osc_ast_guard);
/*
* Lock might be not yet granted. In this case, completion ast
@@
-515,11
+515,11
@@
static int osc_lock_upcall(void *cookie, int errcode)
dlmlock = ldlm_handle2lock(&olck->ols_handle);
if (dlmlock != NULL) {
lock_res_and_lock(dlmlock);
- spin_lock(&osc_ast_guard);
+
cfs_
spin_lock(&osc_ast_guard);
LASSERT(olck->ols_lock == NULL);
dlmlock->l_ast_data = NULL;
olck->ols_handle.cookie = 0ULL;
- spin_unlock(&osc_ast_guard);
+
cfs_
spin_unlock(&osc_ast_guard);
unlock_res_and_lock(dlmlock);
LDLM_LOCK_PUT(dlmlock);
}
@@
-890,7
+890,7
@@
static unsigned long osc_ldlm_weigh_ast(struct ldlm_lock *dlmlock)
unsigned long weight;
ENTRY;
- might_sleep();
+
cfs_
might_sleep();
/*
* osc_ldlm_weigh_ast has a complex context since it might be called
* because of lock canceling, or from user's input. We have to make
@@
-950,86
+950,6
@@
static void osc_lock_build_einfo(const struct lu_env *env,
einfo->ei_cbdata = lock; /* value to be put into ->l_ast_data */
}
-static int osc_lock_delete0(struct cl_lock *conflict)
-{
- struct cl_env_nest nest;
- struct lu_env *env;
- int rc = 0;
-
- env = cl_env_nested_get(&nest);
- if (!IS_ERR(env)) {
- cl_lock_delete(env, conflict);
- cl_env_nested_put(&nest, env);
- } else
- rc = PTR_ERR(env);
- return rc;
-}
-/**
- * Cancels \a conflict lock and waits until it reached CLS_FREEING state. This
- * is called as a part of enqueuing to cancel conflicting locks early.
- *
- * \retval 0: success, \a conflict was cancelled and destroyed.
- *
- * \retval CLO_REPEAT: \a conflict was cancelled, but \a lock mutex was
- * released in the process. Repeat enqueing.
- *
- * \retval -EWOULDBLOCK: \a conflict cannot be cancelled immediately, and
- * either \a lock is non-blocking, or current thread
- * holds other locks, that prevent it from waiting
- * for cancel to complete.
- *
- * \retval -ve: other error, including -EINTR.
- *
- */
-static int osc_lock_cancel_wait(const struct lu_env *env, struct cl_lock *lock,
- struct cl_lock *conflict, int canwait)
-{
- int rc;
-
- LASSERT(cl_lock_is_mutexed(lock));
- LASSERT(cl_lock_is_mutexed(conflict));
-
- rc = 0;
- if (conflict->cll_state != CLS_FREEING) {
- cl_lock_cancel(env, conflict);
- rc = osc_lock_delete0(conflict);
- if (rc)
- return rc;
- if (conflict->cll_flags & (CLF_CANCELPEND|CLF_DOOMED)) {
- rc = -EWOULDBLOCK;
- if (cl_lock_nr_mutexed(env) > 2)
- /*
- * If mutices of locks other than @lock and
- * @scan are held by the current thread, it
- * cannot wait on @scan state change in a
- * dead-lock safe matter, so simply skip early
- * cancellation in this case.
- *
- * This means that early cancellation doesn't
- * work when there is even slight mutex
- * contention, as top-lock's mutex is usually
- * held at this time.
- */
- ;
- else if (canwait) {
- /* Waiting for @scan to be destroyed */
- cl_lock_mutex_put(env, lock);
- do {
- rc = cl_lock_state_wait(env, conflict);
- } while (!rc &&
- conflict->cll_state < CLS_FREEING);
- /* mutex was released, repeat enqueue. */
- rc = rc ?: CLO_REPEAT;
- cl_lock_mutex_get(env, lock);
- }
- }
- LASSERT(ergo(!rc, conflict->cll_state == CLS_FREEING));
- CDEBUG(D_INFO, "lock %p was %s freed now, rc (%d)\n",
- conflict, rc ? "not":"", rc);
- }
- return rc;
-}
-
/**
* Determine if the lock should be converted into a lockless lock.
*
@@
-1068,14
+988,14
@@
static void osc_lock_to_lockless(const struct lu_env *env,
io->ci_lockreq == CILR_NEVER);
ocd = &class_exp2cliimp(osc_export(oob))->imp_connect_data;
- ols->ols_locklessable = (io->ci_type != CIT_
TRUNC
) &&
+ ols->ols_locklessable = (io->ci_type != CIT_
SETATTR
) &&
(io->ci_lockreq == CILR_MAYBE) &&
(ocd->ocd_connect_flags & OBD_CONNECT_SRVLOCK);
if (io->ci_lockreq == CILR_NEVER ||
/* lockless IO */
(ols->ols_locklessable && osc_object_is_contended(oob)) ||
/* lockless truncate */
- (
io->ci_type == CIT_TRUNC
&&
+ (
cl_io_is_trunc(io)
&&
(ocd->ocd_connect_flags & OBD_CONNECT_TRUNCLOCK) &&
osd->od_lockless_truncate)) {
ols->ols_locklessable = 1;
@@
-1085,6
+1005,21
@@
static void osc_lock_to_lockless(const struct lu_env *env,
LASSERT(ergo(ols->ols_glimpse, !osc_lock_is_lockless(ols)));
}
+static int osc_lock_compatible(const struct osc_lock *qing,
+ const struct osc_lock *qed)
+{
+ enum cl_lock_mode qing_mode;
+ enum cl_lock_mode qed_mode;
+
+ qing_mode = qing->ols_cl.cls_lock->cll_descr.cld_mode;
+ if (qed->ols_glimpse &&
+ (qed->ols_state >= OLS_UPCALL_RECEIVED || qing_mode == CLM_READ))
+ return 1;
+
+ qed_mode = qed->ols_cl.cls_lock->cll_descr.cld_mode;
+ return ((qing_mode == CLM_READ) && (qed_mode == CLM_READ));
+}
+
/**
* Cancel all conflicting locks and wait for them to be destroyed.
*
@@
-1102,36
+1037,29
@@
static int osc_lock_enqueue_wait(const struct lu_env *env,
struct cl_lock *lock = olck->ols_cl.cls_lock;
struct cl_lock_descr *descr = &lock->cll_descr;
struct cl_object_header *hdr = cl_object_header(descr->cld_obj);
- struct cl_lock_closure *closure = &osc_env_info(env)->oti_closure;
- struct cl_lock *scan;
- struct cl_lock *temp;
+ struct cl_lock *scan = lock;
+ struct cl_lock *conflict= NULL;
int lockless = osc_lock_is_lockless(olck);
int rc = 0;
- int canwait;
- int stop;
ENTRY;
LASSERT(cl_lock_is_mutexed(lock));
LASSERT(lock->cll_state == CLS_QUEUING);
- /*
- *
XXX This function could be sped up if we had asynchronous
- * cancellation.
- */
+ /*
make it enqueue anyway for glimpse lock, because we actually
+ *
don't need to cancel any conflicting locks. */
+ if (olck->ols_glimpse)
+ return 0;
- canwait =
- !(olck->ols_flags & LDLM_FL_BLOCK_NOWAIT) &&
- cl_lock_nr_mutexed(env) == 1;
- cl_lock_closure_init(env, closure, lock, canwait);
- spin_lock(&hdr->coh_lock_guard);
- list_for_each_entry_safe(scan, temp, &hdr->coh_locks, cll_linkage) {
- if (scan == lock)
- continue;
+ cfs_spin_lock(&hdr->coh_lock_guard);
+ cfs_list_for_each_entry_continue(scan, &hdr->coh_locks, cll_linkage) {
+ struct cl_lock_descr *cld = &scan->cll_descr;
+ const struct osc_lock *scan_ols;
if (scan->cll_state < CLS_QUEUING ||
scan->cll_state == CLS_FREEING ||
-
scan->cll_descr.
cld_start > descr->cld_end ||
-
scan->cll_descr.
cld_end < descr->cld_start)
+
cld->
cld_start > descr->cld_end ||
+
cld->
cld_end < descr->cld_start)
continue;
/* overlapped and living locks. */
@@
-1143,118
+1071,46
@@
static int osc_lock_enqueue_wait(const struct lu_env *env,
continue;
}
- /* A tricky case for lockless pages:
- * We need to cancel the compatible locks if we're enqueuing
+ scan_ols = osc_lock_at(scan);
+
+ /* We need to cancel the compatible locks if we're enqueuing
* a lockless lock, for example:
* imagine that client has PR lock on [0, 1000], and thread T0
* is doing lockless IO in [500, 1500] region. Concurrent
* thread T1 can see lockless data in [500, 1000], which is
- * wrong, because these data are possibly stale.
- */
- if (!lockless && cl_lock_compatible(scan, lock))
+ * wrong, because these data are possibly stale. */
+ 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. */
- cl_lock_get_trust(scan);
- if (&temp->cll_linkage != &hdr->coh_locks)
- cl_lock_get_trust(temp);
- spin_unlock(&hdr->coh_lock_guard);
- lu_ref_add(&scan->cll_reference, "cancel-wait", lock);
-
- LASSERT(list_empty(&closure->clc_list));
- rc = cl_lock_closure_build(env, scan, closure);
- if (rc == 0) {
- rc = osc_lock_cancel_wait(env, lock, scan, canwait);
- cl_lock_disclosure(env, closure);
- if (rc == -EWOULDBLOCK)
- rc = 0;
+ 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);
}
- if (rc == CLO_REPEAT && !canwait)
- /* cannot wait... no early cancellation. */
- rc = 0;
-
- lu_ref_del(&scan->cll_reference, "cancel-wait", lock);
- cl_lock_put(env, scan);
- spin_lock(&hdr->coh_lock_guard);
- /*
- * Lock list could have been modified, while spin-lock was
- * released. Check that it is safe to continue.
- */
- stop = list_empty(&temp->cll_linkage);
- if (&temp->cll_linkage != &hdr->coh_locks)
- cl_lock_put(env, temp);
- if (stop || rc != 0)
- break;
+ cl_lock_get_trust(scan);
+ conflict = scan;
+ break;
}
- spin_unlock(&hdr->coh_lock_guard);
- cl_lock_closure_fini(closure);
- RETURN(rc);
-}
-
-/**
- * Deadlock avoidance for osc_lock_enqueue(). Consider following scenario:
- *
- * - Thread0: obtains PR:[0, 10]. Lock is busy.
- *
- * - Thread1: enqueues PW:[5, 50]. Blocking ast is sent to
- * PR:[0, 10], but cancellation of busy lock is postponed.
- *
- * - Thread0: enqueue PR:[30, 40]. Lock is locally matched to
- * PW:[5, 50], and thread0 waits for the lock completion never
- * releasing PR:[0, 10]---deadlock.
- *
- * The second PR lock can be glimpse (it is to deal with that situation that
- * ll_glimpse_size() has second argument, preventing local match of
- * not-yet-granted locks, see bug 10295). Similar situation is possible in the
- * case of memory mapped user level buffer.
- *
- * To prevent this we can detect a situation when current "thread" or "io"
- * already holds a lock on this object and either add LDLM_FL_BLOCK_GRANTED to
- * the ols->ols_flags, or prevent local match with PW locks.
- */
-static int osc_deadlock_is_possible(const struct lu_env *env,
- struct cl_lock *lock)
-{
- struct cl_object *obj;
- struct cl_object_header *head;
- struct cl_lock *scan;
- struct osc_io *oio;
-
- int result;
-
- ENTRY;
-
- LASSERT(cl_lock_is_mutexed(lock));
-
- oio = osc_env_io(env);
- obj = lock->cll_descr.cld_obj;
- head = cl_object_header(obj);
-
- result = 0;
- spin_lock(&head->coh_lock_guard);
- list_for_each_entry(scan, &head->coh_locks, cll_linkage) {
- if (scan != lock) {
- struct osc_lock *oscan;
-
- oscan = osc_lock_at(scan);
- LASSERT(oscan != NULL);
- if (oscan->ols_owner == oio) {
- result = 1;
- break;
- }
- }
+ cfs_spin_unlock(&hdr->coh_lock_guard);
+
+ if (conflict) {
+ CDEBUG(D_DLMTRACE, "lock %p is confliced with %p, will wait\n",
+ lock, conflict);
+ lu_ref_add(&conflict->cll_reference, "cancel-wait", lock);
+ LASSERT(lock->cll_conflict == NULL);
+ lock->cll_conflict = conflict;
+ rc = CLO_WAIT;
}
- spin_unlock(&head->coh_lock_guard);
- RETURN(result);
+ RETURN(rc);
}
/**
* Implementation of cl_lock_operations::clo_enqueue() method for osc
* layer. This initiates ldlm enqueue:
*
- * - checks for possible dead-lock conditions (osc_deadlock_is_possible());
- *
* - cancels conflicting locks early (osc_lock_enqueue_wait());
*
* - calls osc_enqueue_base() to do actual enqueue.
@@
-1286,16
+1142,14
@@
static int osc_lock_enqueue(const struct lu_env *env,
osc_lock_build_res(env, obj, resname);
osc_lock_build_policy(env, lock, policy);
ols->ols_flags = osc_enq2ldlm_flags(enqflags);
- if (osc_deadlock_is_possible(env, lock))
- ols->ols_flags |= LDLM_FL_BLOCK_GRANTED;
if (ols->ols_flags & LDLM_FL_HAS_INTENT)
ols->ols_glimpse = 1;
+ if (!(enqflags & CEF_MUST))
+ /* try to convert this lock to a lockless lock */
+ osc_lock_to_lockless(env, ols, (enqflags & CEF_NEVER));
result = osc_lock_enqueue_wait(env, ols);
if (result == 0) {
- if (!(enqflags & CEF_MUST))
- /* try to convert this lock to a lockless lock */
- osc_lock_to_lockless(env, ols, (enqflags & CEF_NEVER));
if (!osc_lock_is_lockless(ols)) {
if (ols->ols_locklessable)
ols->ols_flags |= LDLM_FL_DENY_ON_CONTENTION;
@@
-1324,6
+1178,7
@@
static int osc_lock_enqueue(const struct lu_env *env,
}
} else {
ols->ols_state = OLS_GRANTED;
+ ols->ols_owner = osc_env_io(env);
}
}
LASSERT(ergo(ols->ols_glimpse, !osc_lock_is_lockless(ols)));
@@
-1491,12
+1346,13
@@
static int osc_lock_has_pages(struct osc_lock *olck)
plist = &osc_env_info(env)->oti_plist;
cl_page_list_init(plist);
- mutex_lock(&oob->oo_debug_mutex);
+
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);
+ 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");
@@
-1507,7
+1363,7
@@
static int osc_lock_has_pages(struct osc_lock *olck)
cl_page_list_disown(env, io, plist);
cl_page_list_fini(env, plist);
cl_io_fini(env, io);
- mutex_unlock(&oob->oo_debug_mutex);
+
cfs_
mutex_unlock(&oob->oo_debug_mutex);
cl_env_nested_put(&nest, env);
} else
result = 0;
@@
-1554,13
+1410,14
@@
static void osc_lock_state(const struct lu_env *env,
enum cl_lock_state state)
{
struct osc_lock *lock = cl2osc_lock(slice);
- struct osc_io *oio = osc_env_io(env);
/*
* XXX multiple io contexts can use the lock at the same time.
*/
LINVRNT(osc_lock_invariant(lock));
if (state == CLS_HELD && slice->cls_lock->cll_state != CLS_HELD) {
+ struct osc_io *oio = osc_env_io(env);
+
LASSERT(lock->ols_owner == NULL);
lock->ols_owner = oio;
} else if (state != CLS_HELD)
@@
-1607,20
+1464,17
@@
static int osc_lock_fits_into(const struct lu_env *env,
* will not release sublock1. Bang!
*/
if (ols->ols_state < OLS_GRANTED ||
-
ols->ols_state > OLS_RELEASED)
+ ols->ols_state > OLS_RELEASED)
return 0;
} else if (need->cld_enq_flags & CEF_MUST) {
-
/*
+ /*
* If the lock hasn't ever enqueued, it can't be matched
* because enqueue process brings in many information
* which can be used to determine things such as lockless,
* CEF_MUST, etc.
*/
- if (ols->ols_state < OLS_GRANTED ||
- ols->ols_state > OLS_RELEASED)
- return 0;
if (ols->ols_state < OLS_UPCALL_RECEIVED &&
-
ols->ols_locklessable)
+ ols->ols_locklessable)
return 0;
}
return 1;
@@
-1692,19
+1546,19
@@
static void osc_lock_lockless_state(const struct lu_env *env,
enum cl_lock_state state)
{
struct osc_lock *lock = cl2osc_lock(slice);
- struct osc_io *oio = osc_env_io(env);
LINVRNT(osc_lock_invariant(lock));
if (state == CLS_HELD) {
- LASSERT(lock->ols_owner == NULL);
+ struct osc_io *oio = osc_env_io(env);
+
+ LASSERT(ergo(lock->ols_owner, lock->ols_owner == oio));
lock->ols_owner = oio;
/* set the io to be lockless if this lock is for io's
* host object */
if (cl_object_same(oio->oi_cl.cis_obj, slice->cls_obj))
oio->oi_lockless = 1;
- } else
- lock->ols_owner = NULL;
+ }
}
static int osc_lock_lockless_fits_into(const struct lu_env *env,
@@
-1712,7
+1566,13
@@
static int osc_lock_lockless_fits_into(const struct lu_env *env,
const struct cl_lock_descr *need,
const struct cl_io *io)
{
- return 0;
+ struct osc_lock *lock = cl2osc_lock(slice);
+
+ if (!(need->cld_enq_flags & CEF_NEVER))
+ return 0;
+
+ /* lockless lock should only be used by its owning io. b22147 */
+ return (lock->ols_owner == osc_env_io(env));
}
static const struct cl_lock_operations osc_lock_lockless_ops = {