* GPL HEADER END
*/
/*
- * Copyright 2008 Sun Microsystems, Inc. All rights reserved.
+ * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
*/
/*
* they are not initialized at all. As a temp fix, in this case,
* we still borrow the parent's env to call sublock operations.
*/
- if (!cl_object_same(io->ci_obj, parent->cll_descr.cld_obj)) {
+ if (!io || !cl_object_same(io->ci_obj, parent->cll_descr.cld_obj)) {
subenv->lse_env = env;
subenv->lse_io = io;
subenv->lse_sub = NULL;
lck->lls_sub[idx].sub_lock = lsl;
lck->lls_nr_filled++;
LASSERT(lck->lls_nr_filled <= lck->lls_nr);
- list_add_tail(&link->lll_list, &lsl->lss_parents);
+ cfs_list_add_tail(&link->lll_list, &lsl->lss_parents);
link->lll_idx = idx;
link->lll_super = lck;
cl_lock_get(parent);
int result = 0;
ENTRY;
- LASSERT(list_empty(&closure->clc_list));
+ LASSERT(cfs_list_empty(&closure->clc_list));
sublock = lls->sub_lock;
child = sublock->lss_cl.cls_lock;
LASSERT(cl_lock_is_mutexed(child));
sublock->lss_active = parent;
- if (unlikely(child->cll_state == CLS_FREEING)) {
+ if (unlikely((child->cll_state == CLS_FREEING) ||
+ (child->cll_flags & CLF_CANCELLED))) {
struct lov_lock_link *link;
/*
* we could race with lock deletion which temporarily
nr++;
}
LASSERT(nr > 0);
- OBD_ALLOC(lck->lls_sub, nr * sizeof lck->lls_sub[0]);
+ OBD_ALLOC_LARGE(lck->lls_sub, nr * sizeof lck->lls_sub[0]);
if (lck->lls_sub == NULL)
RETURN(-ENOMEM);
* a reference on its parent.
*/
LASSERT(lck->lls_sub[i].sub_lock == NULL);
- OBD_FREE(lck->lls_sub, lck->lls_nr * sizeof lck->lls_sub[0]);
+ OBD_FREE_LARGE(lck->lls_sub,
+ lck->lls_nr * sizeof lck->lls_sub[0]);
}
OBD_SLAB_FREE_PTR(lck, lov_lock_kmem);
EXIT;
}
+static int lov_lock_enqueue_wait(const struct lu_env *env,
+ struct lov_lock *lck,
+ struct cl_lock *sublock)
+{
+ struct cl_lock *lock = lck->lls_cl.cls_lock;
+ int result;
+ ENTRY;
+
+ LASSERT(cl_lock_is_mutexed(lock));
+
+ cl_lock_mutex_put(env, lock);
+ result = cl_lock_enqueue_wait(env, sublock, 0);
+ cl_lock_mutex_get(env, lock);
+ RETURN(result ?: CLO_REPEAT);
+}
+
/**
* Tries to advance a state machine of a given sub-lock toward enqueuing of
* the top-lock.
subenv->lse_io, enqflags,
i == lck->lls_nr - 1);
minstate = min(minstate, sublock->cll_state);
- /*
- * Don't hold a sub-lock in CLS_CACHED state, see
- * description for lov_lock::lls_sub.
- */
- if (sublock->cll_state > CLS_HELD)
- rc = lov_sublock_release(env, lck, i, 1, rc);
- lov_sublock_unlock(env, sub, closure, subenv);
+ if (rc == CLO_WAIT) {
+ switch (sublock->cll_state) {
+ case CLS_QUEUING:
+ /* take recursive mutex, the lock is
+ * released in lov_lock_enqueue_wait.
+ */
+ cl_lock_mutex_get(env, sublock);
+ lov_sublock_unlock(env, sub, closure,
+ subenv);
+ rc = lov_lock_enqueue_wait(env, lck,
+ sublock);
+ break;
+ case CLS_CACHED:
+ rc = lov_sublock_release(env, lck, i,
+ 1, rc);
+ default:
+ lov_sublock_unlock(env, sub, closure,
+ subenv);
+ break;
+ }
+ } else {
+ LASSERT(sublock->cll_conflict == NULL);
+ lov_sublock_unlock(env, sub, closure, subenv);
+ }
}
result = lov_subresult(result, rc);
if (result != 0)
if (lls->sub_flags & LSF_HELD) {
LASSERT(sublock->cll_state == CLS_HELD);
rc = cl_unuse_try(subenv->lse_env, sublock);
- if (rc != CLO_WAIT)
- rc = lov_sublock_release(env, lck,
- i, 0, rc);
+ rc = lov_sublock_release(env, lck, i, 0, rc);
}
lov_sublock_unlock(env, sub, closure, subenv);
}
result = lov_subresult(result, rc);
- if (result < 0)
- break;
}
if (result == 0 && lck->lls_cancel_race) {
sublock = sub->lss_cl.cls_lock;
rc = lov_sublock_lock(env, lck, lls, closure, &subenv);
if (rc == 0) {
- if (lls->sub_flags & LSF_HELD) {
- if (sublock->cll_state == CLS_HELD) {
- rc = cl_unuse_try(subenv->lse_env,
- sublock);
- lov_sublock_release(env, lck, i, 0, 0);
- } else {
- lov_sublock_release(env, lck, i, 1, 0);
- }
+ if (!(lls->sub_flags & LSF_HELD)) {
+ lov_sublock_unlock(env, sub, closure, subenv);
+ continue;
+ }
+
+ switch(sublock->cll_state) {
+ case CLS_HELD:
+ rc = cl_unuse_try(subenv->lse_env,
+ sublock);
+ lov_sublock_release(env, lck, i, 0, 0);
+ break;
+ case CLS_ENQUEUED:
+ /* TODO: it's not a good idea to cancel this
+ * lock because it's innocent. But it's
+ * acceptable. The better way would be to
+ * define a new lock method to unhold the
+ * dlm lock. */
+ cl_lock_cancel(env, sublock);
+ default:
+ lov_sublock_release(env, lck, i, 1, 0);
+ break;
}
lov_sublock_unlock(env, sub, closure, subenv);
}
+
+ if (rc == CLO_REPEAT) {
+ --i;
+ continue;
+ }
+
result = lov_subresult(result, rc);
- if (result < 0)
- break;
}
- cl_lock_closure_fini(closure);
+ if (result)
+ CL_LOCK_DEBUG(D_ERROR, env, slice->cls_lock,
+ "lov_lock_cancel fails with %d.\n", result);
- return;
+ cl_lock_closure_fini(closure);
}
static int lov_lock_wait(const struct lu_env *env,
cl2lov(slice->cls_obj),
lov->lls_sub[0].sub_stripe,
got, need);
- } else if (io->ci_type != CIT_TRUNC && io->ci_type != CIT_MISC &&
+ } else if (io->ci_type != CIT_SETATTR && io->ci_type != CIT_MISC &&
!cl_io_is_append(io) && need->cld_mode != CLM_PHANTOM)
/*
* Multi-stripe locks are only suitable for `quick' IO and for
* match against original lock extent.
*/
result = cl_lock_ext_match(&lov->lls_orig, need);
- CDEBUG(D_DLMTRACE, DDESCR"/"DDESCR" %i %i/%i: %i\n",
+ CDEBUG(D_DLMTRACE, DDESCR"/"DDESCR" %d %d/%d: %d\n",
PDESCR(&lov->lls_orig), PDESCR(&lov->lls_sub[0].sub_got),
lov->lls_sub[0].sub_stripe, lov->lls_nr, lov_r0(obj)->lo_nr,
result);
LASSERT(cl_lock_is_mutexed(sub->lss_cl.cls_lock));
ENTRY;
- list_del_init(&link->lll_list);
+ cfs_list_del_init(&link->lll_list);
LASSERT(lck->lls_sub[link->lll_idx].sub_lock == sub);
/* yank this sub-lock from parent's array */
lck->lls_sub[link->lll_idx].sub_lock = NULL;
LASSERT(cl_lock_is_mutexed(sub->lss_cl.cls_lock));
ENTRY;
- list_for_each_entry(scan, &sub->lss_parents, lll_list) {
+ cfs_list_for_each_entry(scan, &sub->lss_parents, lll_list) {
if (scan->lll_super == lck)
RETURN(scan);
}
{
struct lov_lock *lck = cl2lov_lock(slice);
struct cl_lock_closure *closure = lov_closure_get(env, slice->cls_lock);
- int i;
+ struct lov_lock_link *link;
+ int rc;
+ int i;
LASSERT(slice->cls_lock->cll_state == CLS_FREEING);
ENTRY;
for (i = 0; i < lck->lls_nr; ++i) {
- struct lov_lock_sub *lls;
- struct lovsub_lock *lsl;
- struct cl_lock *sublock;
- int rc;
+ struct lov_lock_sub *lls = &lck->lls_sub[i];
+ struct lovsub_lock *lsl = lls->sub_lock;
- lls = &lck->lls_sub[i];
- lsl = lls->sub_lock;
- if (lsl == NULL)
+ if (lsl == NULL) /* already removed */
continue;
- sublock = lsl->lss_cl.cls_lock;
rc = lov_sublock_lock(env, lck, lls, closure, NULL);
- if (rc == 0) {
- if (lls->sub_flags & LSF_HELD)
- lov_sublock_release(env, lck, i, 1, 0);
- if (sublock->cll_state < CLS_FREEING) {
- struct lov_lock_link *link;
-
- link = lov_lock_link_find(env, lck, lsl);
- LASSERT(link != NULL);
- lov_lock_unlink(env, link, lsl);
- LASSERT(lck->lls_sub[i].sub_lock == NULL);
- }
- lov_sublock_unlock(env, lsl, closure, NULL);
- } else if (rc == CLO_REPEAT) {
- --i; /* repeat with this lock */
- } else {
- CL_LOCK_DEBUG(D_ERROR, env, sublock,
- "Cannot get sub-lock for delete: %i\n",
- rc);
+ if (rc == CLO_REPEAT) {
+ --i;
+ continue;
}
+
+ LASSERT(rc == 0);
+ LASSERT(lsl->lss_cl.cls_lock->cll_state < CLS_FREEING);
+
+ if (lls->sub_flags & LSF_HELD)
+ lov_sublock_release(env, lck, i, 1, 0);
+
+ link = lov_lock_link_find(env, lck, lsl);
+ LASSERT(link != NULL);
+ lov_lock_unlink(env, link, lsl);
+ LASSERT(lck->lls_sub[i].sub_lock == NULL);
+
+ lov_sublock_unlock(env, lsl, closure, NULL);
}
+
cl_lock_closure_fini(closure);
EXIT;
}
struct cl_lock_closure *closure;
closure = &lov_env_info(env)->lti_closure;
- LASSERT(list_empty(&closure->clc_list));
+ LASSERT(cfs_list_empty(&closure->clc_list));
cl_lock_closure_init(env, closure, parent, 1);
return closure;
}