-/* -*- 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.
*/
static void osc_lock_detach(const struct lu_env *env, struct osc_lock *olck)
{
- struct ldlm_lock *dlmlock;
+ struct ldlm_lock *dlmlock;
- cfs_spin_lock(&osc_ast_guard);
- dlmlock = olck->ols_lock;
- if (dlmlock == NULL) {
- cfs_spin_unlock(&osc_ast_guard);
+ spin_lock(&osc_ast_guard);
+ dlmlock = olck->ols_lock;
+ if (dlmlock == NULL) {
+ spin_unlock(&osc_ast_guard);
return;
}
* call to osc_lock_detach() */
dlmlock->l_ast_data = NULL;
olck->ols_handle.cookie = 0ULL;
- cfs_spin_unlock(&osc_ast_guard);
+ spin_unlock(&osc_ast_guard);
lock_res_and_lock(dlmlock);
if (dlmlock->l_granted_mode == dlmlock->l_req_mode) {
LASSERT(!ols->ols_hold);
LASSERT(ols->ols_agl);
return 0;
- case OLS_UPCALL_RECEIVED:
- LASSERT(!ols->ols_hold);
- ols->ols_state = OLS_NEW;
- 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);
policy->l_extent.gid = d->cld_gid;
}
-static int osc_enq2ldlm_flags(__u32 enqflags)
+static __u64 osc_enq2ldlm_flags(__u32 enqflags)
{
- int result = 0;
+ __u64 result = 0;
LASSERT((enqflags & ~CEF_MASK) == 0);
* Global spin-lock protecting consistency of ldlm_lock::l_ast_data
* pointers. Initialized in osc_init().
*/
-cfs_spinlock_t osc_ast_guard;
+spinlock_t osc_ast_guard;
static struct osc_lock *osc_ast_data_get(struct ldlm_lock *dlm_lock)
{
- struct osc_lock *olck;
+ struct osc_lock *olck;
- lock_res_and_lock(dlm_lock);
- cfs_spin_lock(&osc_ast_guard);
+ lock_res_and_lock(dlm_lock);
+ spin_lock(&osc_ast_guard);
olck = dlm_lock->l_ast_data;
if (olck != NULL) {
struct cl_lock *lock = olck->ols_cl.cls_lock;
} else
olck = NULL;
}
- cfs_spin_unlock(&osc_ast_guard);
- unlock_res_and_lock(dlm_lock);
- return olck;
+ spin_unlock(&osc_ast_guard);
+ unlock_res_and_lock(dlm_lock);
+ return olck;
}
static void osc_ast_data_put(const struct lu_env *env, struct osc_lock *olck)
LASSERT(dlmlock != NULL);
lock_res_and_lock(dlmlock);
- cfs_spin_lock(&osc_ast_guard);
- LASSERT(dlmlock->l_ast_data == olck);
- LASSERT(olck->ols_lock == NULL);
- olck->ols_lock = dlmlock;
- cfs_spin_unlock(&osc_ast_guard);
+ 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);
/*
* Lock might be not yet granted. In this case, completion ast
dlmlock = ldlm_handle2lock(&olck->ols_handle);
if (dlmlock != NULL) {
lock_res_and_lock(dlmlock);
- cfs_spin_lock(&osc_ast_guard);
- LASSERT(olck->ols_lock == NULL);
- dlmlock->l_ast_data = NULL;
- olck->ols_handle.cookie = 0ULL;
- cfs_spin_unlock(&osc_ast_guard);
+ 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);
ldlm_lock_fail_match_locked(dlmlock);
unlock_res_and_lock(dlmlock);
LDLM_LOCK_PUT(dlmlock);
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);
}
/**
}
static int osc_ldlm_completion_ast(struct ldlm_lock *dlmlock,
- int flags, void *data)
+ __u64 flags, void *data)
{
struct cl_env_nest nest;
struct lu_env *env;
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);
+ if (!exp_connect_lvb_type(req->rq_export))
+ req_capsule_shrink(&req->rq_pill,
+ &RMF_DLM_LVB,
+ sizeof(struct ost_lvb_v1),
+ RCL_SERVER);
osc_ast_data_put(env, olck);
} else {
/*
struct osc_lock *ols, int force)
{
struct cl_lock_slice *slice = &ols->ols_cl;
- struct cl_lock *lock = slice->cls_lock;
LASSERT(ols->ols_state == OLS_NEW ||
ols->ols_state == OLS_UPCALL_RECEIVED);
if (force) {
ols->ols_locklessable = 1;
- LASSERT(cl_lock_is_mutexed(lock));
slice->cls_ops = &osc_lock_lockless_ops;
} else {
struct osc_io *oio = osc_env_io(env);
if (olck->ols_glimpse)
return 0;
- cfs_spin_lock(&hdr->coh_lock_guard);
+ spin_lock(&hdr->coh_lock_guard);
cfs_list_for_each_entry(scan, &hdr->coh_locks, cll_linkage) {
struct cl_lock_descr *cld = &scan->cll_descr;
const struct osc_lock *scan_ols;
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;
}
- cfs_spin_unlock(&hdr->coh_lock_guard);
+ spin_unlock(&hdr->coh_lock_guard);
if (conflict) {
if (lock->cll_descr.cld_mode == CLM_GROUP) {
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;
- }
- if (ols->ols_flags & LDLM_FL_HAS_INTENT)
- ols->ols_glimpse = 1;
- if (!osc_lock_is_lockless(ols) && !(enqflags & CEF_MUST))
- /* try to convert this lock to a lockless lock */
- osc_lock_to_lockless(env, ols, (enqflags & CEF_NEVER));
+ LASSERTF(ergo(ols->ols_glimpse, lock->cll_descr.cld_mode <= CLM_READ),
+ "lock = %p, ols = %p\n", lock, ols);
result = osc_lock_enqueue_wait(env, ols);
if (result == 0) {
ldlm_policy_data_t *policy = &info->oti_policy;
struct ldlm_enqueue_info *einfo = &ols->ols_einfo;
- 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;
PTLRPCD_SET, 1, ols->ols_agl);
if (result != 0) {
cl_lock_user_del(env, lock);
- lu_ref_del(&lock->cll_reference,
- "upcall", lock);
- cl_lock_put(env, lock);
+ cl_lock_unhold(env, lock, "upcall", lock);
if (unlikely(result == -ECANCELED)) {
ols->ols_state = OLS_NEW;
result = 0;
int rc;
LASSERT(olck->ols_agl);
-
+ olck->ols_agl = 0;
rc = osc_lock_enqueue(env, slice, NULL, CEF_ASYNC | CEF_MUST);
if (rc != 0)
return rc;
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);
+ LDLM_DEBUG(ols->ols_lock,
+ "lock %p: %d pages were %s.\n", lock, result,
+ discard ? "discarded" : "written");
+ 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);
}
/**
int do_cancel;
discard = !!(dlmlock->l_flags & LDLM_FL_DISCARD_DATA);
- result = osc_lock_flush(olck, discard);
+ if (olck->ols_state >= OLS_GRANTED)
+ result = osc_lock_flush(olck, discard);
osc_lock_unhold(olck);
lock_res_and_lock(dlmlock);
lock = olck->ols_cl.cls_lock;
descr = &lock->cll_descr;
- cfs_mutex_lock(&oob->oo_debug_mutex);
+ 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,
cfs_cond_resched();
} while (result != CLP_GANG_OKAY);
cl_io_fini(env, io);
- cfs_mutex_unlock(&oob->oo_debug_mutex);
+ mutex_unlock(&oob->oo_debug_mutex);
cl_env_nested_put(&nest, env);
return (result == CLP_GANG_ABORT);
/*
* XXX print ldlm lock and einfo properly.
*/
- (*p)(env, cookie, "%p %08x "LPX64" %d %p ",
+ (*p)(env, cookie, "%p %#16llx "LPX64" %d %p ",
lock->ols_lock, lock->ols_flags, lock->ols_handle.cookie,
lock->ols_state, lock->ols_owner);
osc_lvb_print(env, cookie, p, &lock->ols_lvb);
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);
OBD_SLAB_ALLOC_PTR_GFP(clk, osc_lock_kmem, CFS_ALLOC_IO);
if (clk != NULL) {
- osc_lock_build_einfo(env, lock, clk, &clk->ols_einfo);
- cfs_atomic_set(&clk->ols_pageref, 0);
- clk->ols_state = OLS_NEW;
- cl_lock_slice_add(lock, &clk->ols_cl, obj, &osc_lock_ops);
+ __u32 enqflags = lock->cll_descr.cld_enq_flags;
+
+ osc_lock_build_einfo(env, lock, clk, &clk->ols_einfo);
+ cfs_atomic_set(&clk->ols_pageref, 0);
+ clk->ols_state = OLS_NEW;
+
+ clk->ols_flags = osc_enq2ldlm_flags(enqflags);
+ clk->ols_agl = !!(enqflags & CEF_AGL);
+ if (clk->ols_agl)
+ clk->ols_flags |= LDLM_FL_BLOCK_NOWAIT;
+ if (clk->ols_flags & LDLM_FL_HAS_INTENT)
+ clk->ols_glimpse = 1;
+
+ cl_lock_slice_add(lock, &clk->ols_cl, obj, &osc_lock_ops);
+
+ if (!(enqflags & CEF_MUST))
+ /* try to convert this lock to a lockless lock */
+ osc_lock_to_lockless(env, clk, (enqflags & CEF_NEVER));
+ if (clk->ols_locklessable && !(enqflags & CEF_DISCARD_DATA))
+ clk->ols_flags |= LDLM_FL_DENY_ON_CONTENTION;
+
+ LDLM_DEBUG_NOLOCK("lock %p, osc lock %p, flags %llx\n",
+ lock, clk, clk->ols_flags);
+
result = 0;
} else
result = -ENOMEM;
int osc_dlm_lock_pageref(struct ldlm_lock *dlm)
{
- struct osc_lock *olock;
- int rc = 0;
+ struct osc_lock *olock;
+ int rc = 0;
- cfs_spin_lock(&osc_ast_guard);
+ spin_lock(&osc_ast_guard);
olock = dlm->l_ast_data;
/*
* there's a very rare race with osc_page_addref_lock(), but that
cfs_atomic_sub(_PAGEREF_MAGIC, &olock->ols_pageref);
rc = 1;
}
- cfs_spin_unlock(&osc_ast_guard);
- return rc;
+ spin_unlock(&osc_ast_guard);
+ return rc;
}
/** @} osc */
git://git.whamcloud.com / fs / lustre-release.git / blobdiff