}
/**
+ *
+ * \retval 0 if state-transition can proceed
+ * \retval -ve otherwise.
+ */
+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;
+ struct cl_lock *conflict = sublock->cll_conflict;
+ int result = CLO_REPEAT;
+ ENTRY;
+
+ LASSERT(cl_lock_is_mutexed(lock));
+ LASSERT(cl_lock_is_mutexed(sublock));
+ LASSERT(sublock->cll_state == CLS_QUEUING);
+ LASSERT(conflict != NULL);
+
+ sublock->cll_conflict = NULL;
+ cl_lock_mutex_put(env, lock);
+ cl_lock_mutex_put(env, sublock);
+
+ LASSERT(cl_lock_nr_mutexed(env) == 0);
+
+ cl_lock_mutex_get(env, conflict);
+ cl_lock_cancel(env, conflict);
+ cl_lock_delete(env, conflict);
+ while (conflict->cll_state != CLS_FREEING) {
+ int rc = 0;
+
+ rc = cl_lock_state_wait(env, conflict);
+ if (rc == 0)
+ continue;
+
+ result = lov_subresult(result, rc);
+ break;
+ }
+ cl_lock_mutex_put(env, conflict);
+ lu_ref_del(&conflict->cll_reference, "cancel-wait", sublock);
+ cl_lock_put(env, conflict);
+
+ cl_lock_mutex_get(env, lock);
+ RETURN(result);
+}
+
+/**
* 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)
EXIT;
}
-static int lovsub_lock_state_one(const struct lu_env *env,
- const struct lovsub_lock *lovsub,
- struct lov_lock *lov)
-{
- struct cl_lock *parent;
- struct cl_lock *child;
- int restart = 0;
-
- ENTRY;
- parent = lov->lls_cl.cls_lock;
- child = lovsub->lss_cl.cls_lock;
-
- if (lovsub->lss_active != parent) {
- lovsub_parent_lock(env, lov);
- if (child->cll_error != 0 && parent->cll_error == 0) {
- /*
- * This is a deadlock case:
- * cl_lock_error(for the parent lock)
- * -> cl_lock_delete
- * -> lov_lock_delete
- * -> cl_lock_enclosure
- * -> cl_lock_mutex_try(for the child lock)
- */
- cl_lock_mutex_put(env, child);
- cl_lock_error(env, parent, child->cll_error);
- restart = 1;
- } else {
- cl_lock_signal(env, parent);
- }
- lovsub_parent_unlock(env, lov);
- }
- RETURN(restart);
-}
-
/**
* Implements cl_lock_operations::clo_state() method for lovsub layer, which
* method is called whenever sub-lock state changes. Propagates state change
{
struct lovsub_lock *sub = cl2lovsub_lock(slice);
struct lov_lock_link *scan;
- int restart = 0;
LASSERT(cl_lock_is_mutexed(slice->cls_lock));
ENTRY;
- do {
- restart = 0;
- list_for_each_entry(scan, &sub->lss_parents, lll_list) {
- restart = lovsub_lock_state_one(env, sub,
- scan->lll_super);
- if (restart) {
- cl_lock_mutex_get(env, slice->cls_lock);
- break;
- }
+ list_for_each_entry(scan, &sub->lss_parents, lll_list) {
+ struct lov_lock *lov = scan->lll_super;
+ struct cl_lock *parent = lov->lls_cl.cls_lock;
+
+ if (sub->lss_active != parent) {
+ lovsub_parent_lock(env, lov);
+ cl_lock_signal(env, parent);
+ lovsub_parent_unlock(env, lov);
}
- } while(restart);
+ }
EXIT;
}
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.
*
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));
+}
+
+#ifndef list_for_each_entry_continue
+#define list_for_each_entry_continue(pos, head, member) \
+ for (pos = list_entry(pos->member.next, typeof(*pos), member); \
+ prefetch(pos->member.next), &pos->member != (head); \
+ pos = list_entry(pos->member.next, typeof(*pos), member))
+#endif
+
/**
* Cancel all conflicting locks and wait for them to be destroyed.
*
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;
+ 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. */
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);
+
+ 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;
+ }
RETURN(rc);
}
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;