-/* -*- 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.
* 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.
+ *
+ * Copyright (c) 2011, 2013, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
* Client Extent Lock.
*
* Author: Nikita Danilov <nikita.danilov@sun.com>
+ * Author: Jinshan Xiong <jinshan.xiong@intel.com>
*/
#define DEBUG_SUBSYSTEM S_CLASS
-#ifndef EXPORT_SYMTAB
-# define EXPORT_SYMTAB
-#endif
+#include <libcfs/libcfs.h>
#include <obd_class.h>
#include <obd_support.h>
#include <lustre_fid.h>
#include <libcfs/list.h>
-/* lu_time_global_{init,fini}() */
-#include <lu_time.h>
-
#include <cl_object.h>
#include "cl_internal.h"
/** Lock class of cl_lock::cll_guard */
static struct lock_class_key cl_lock_guard_class;
-static cfs_mem_cache_t *cl_lock_kmem;
+static struct kmem_cache *cl_lock_kmem;
static struct lu_kmem_descr cl_lock_caches[] = {
{
}
};
+#ifdef CONFIG_DEBUG_PAGESTATE_TRACKING
+#define CS_LOCK_INC(o, item) \
+ atomic_inc(&cl_object_site(o)->cs_locks.cs_stats[CS_##item])
+#define CS_LOCK_DEC(o, item) \
+ atomic_dec(&cl_object_site(o)->cs_locks.cs_stats[CS_##item])
+#define CS_LOCKSTATE_INC(o, state) \
+ atomic_inc(&cl_object_site(o)->cs_locks_state[state])
+#define CS_LOCKSTATE_DEC(o, state) \
+ atomic_dec(&cl_object_site(o)->cs_locks_state[state])
+#else
+#define CS_LOCK_INC(o, item)
+#define CS_LOCK_DEC(o, item)
+#define CS_LOCKSTATE_INC(o, state)
+#define CS_LOCKSTATE_DEC(o, state)
+#endif
+
/**
* Basic lock invariant that is maintained at all times. Caller either has a
* reference to \a lock, or somehow assures that \a lock cannot be freed.
static int cl_lock_invariant_trusted(const struct lu_env *env,
const struct cl_lock *lock)
{
- return
- cl_is_lock(lock) &&
- ergo(lock->cll_state == CLS_FREEING, lock->cll_holds == 0) &&
- atomic_read(&lock->cll_ref) >= lock->cll_holds &&
+ return ergo(lock->cll_state == CLS_FREEING, lock->cll_holds == 0) &&
+ atomic_read(&lock->cll_ref) >= lock->cll_holds &&
lock->cll_holds >= lock->cll_users &&
lock->cll_holds >= 0 &&
lock->cll_users >= 0 &&
{
int result;
- result = atomic_read(&lock->cll_ref) > 0 &&
+ result = atomic_read(&lock->cll_ref) > 0 &&
cl_lock_invariant_trusted(env, lock);
- if (!result && env != NULL)
- CL_LOCK_DEBUG(D_ERROR, env, lock, "invariant broken");
- return result;
+ if (!result && env != NULL)
+ CL_LOCK_DEBUG(D_ERROR, env, lock, "invariant broken\n");
+ return result;
+}
+
+/**
+ * Returns lock "nesting": 0 for a top-lock and 1 for a sub-lock.
+ */
+static enum clt_nesting_level cl_lock_nesting(const struct cl_lock *lock)
+{
+ return cl_object_header(lock->cll_descr.cld_obj)->coh_nesting;
}
+/**
+ * Returns a set of counters for this lock, depending on a lock nesting.
+ */
+static struct cl_thread_counters *cl_lock_counters(const struct lu_env *env,
+ const struct cl_lock *lock)
+{
+ struct cl_thread_info *info;
+ enum clt_nesting_level nesting;
+
+ info = cl_env_info(env);
+ nesting = cl_lock_nesting(lock);
+ LASSERT(nesting < ARRAY_SIZE(info->clt_counters));
+ return &info->clt_counters[nesting];
+}
+
+static void cl_lock_trace0(int level, const struct lu_env *env,
+ const char *prefix, const struct cl_lock *lock,
+ const char *func, const int line)
+{
+ struct cl_object_header *h = cl_object_header(lock->cll_descr.cld_obj);
+ CDEBUG(level, "%s: %p@(%d %p %d %d %d %d %d %lx)"
+ "(%p/%d/%d) at %s():%d\n",
+ prefix, lock, atomic_read(&lock->cll_ref),
+ lock->cll_guarder, lock->cll_depth,
+ lock->cll_state, lock->cll_error, lock->cll_holds,
+ lock->cll_users, lock->cll_flags,
+ env, h->coh_nesting, cl_lock_nr_mutexed(env),
+ func, line);
+}
+#define cl_lock_trace(level, env, prefix, lock) \
+ cl_lock_trace0(level, env, prefix, lock, __FUNCTION__, __LINE__)
+
#define RETIP ((unsigned long)__builtin_return_address(0))
#ifdef CONFIG_LOCKDEP
static void cl_lock_lockdep_acquire(const struct lu_env *env,
struct cl_lock *lock, __u32 enqflags)
{
- cl_env_info(env)->clt_nr_locks_acquired++;
- lock_acquire(&lock->dep_map, !!(enqflags & CEF_ASYNC),
- /* try: */ 0, lock->cll_descr.cld_mode <= CLM_READ,
- /* check: */ 2, RETIP);
+ cl_lock_counters(env, lock)->ctc_nr_locks_acquired++;
+ lock_map_acquire(&lock->dep_map);
}
static void cl_lock_lockdep_release(const struct lu_env *env,
struct cl_lock *lock)
{
- cl_env_info(env)->clt_nr_locks_acquired--;
- lock_release(&lock->dep_map, 0, RETIP);
+ cl_lock_counters(env, lock)->ctc_nr_locks_acquired--;
+ lock_map_release(&lock->dep_map);
}
#else /* !CONFIG_LOCKDEP */
struct cl_object *obj,
const struct cl_lock_operations *ops)
{
- ENTRY;
- slice->cls_lock = lock;
- list_add_tail(&slice->cls_linkage, &lock->cll_layers);
- slice->cls_obj = obj;
- slice->cls_ops = ops;
- EXIT;
+ ENTRY;
+ slice->cls_lock = lock;
+ list_add_tail(&slice->cls_linkage, &lock->cll_layers);
+ slice->cls_obj = obj;
+ slice->cls_ops = ops;
+ EXIT;
}
EXPORT_SYMBOL(cl_lock_slice_add);
*/
int cl_lock_mode_match(enum cl_lock_mode has, enum cl_lock_mode need)
{
- LINVRNT(need == CLM_READ || need == CLM_WRITE || need == CLM_PHANTOM);
- LINVRNT(has == CLM_READ || has == CLM_WRITE || has == CLM_PHANTOM);
+ LINVRNT(need == CLM_READ || need == CLM_WRITE ||
+ need == CLM_PHANTOM || need == CLM_GROUP);
+ LINVRNT(has == CLM_READ || has == CLM_WRITE ||
+ has == CLM_PHANTOM || has == CLM_GROUP);
CLASSERT(CLM_PHANTOM < CLM_READ);
CLASSERT(CLM_READ < CLM_WRITE);
+ CLASSERT(CLM_WRITE < CLM_GROUP);
- return need <= has;
+ if (has != CLM_GROUP)
+ return need <= has;
+ else
+ return need == has;
}
EXPORT_SYMBOL(cl_lock_mode_match);
return
has->cld_start <= need->cld_start &&
has->cld_end >= need->cld_end &&
- cl_lock_mode_match(has->cld_mode, need->cld_mode);
+ cl_lock_mode_match(has->cld_mode, need->cld_mode) &&
+ (has->cld_mode != CLM_GROUP || has->cld_gid == need->cld_gid);
}
EXPORT_SYMBOL(cl_lock_ext_match);
{
struct cl_object *obj = lock->cll_descr.cld_obj;
- LASSERT(cl_is_lock(lock));
LINVRNT(!cl_lock_is_mutexed(lock));
- ENTRY;
- might_sleep();
- while (!list_empty(&lock->cll_layers)) {
- struct cl_lock_slice *slice;
-
- slice = list_entry(lock->cll_layers.next, struct cl_lock_slice,
- cls_linkage);
- list_del_init(lock->cll_layers.next);
- slice->cls_ops->clo_fini(env, slice);
- }
- atomic_dec(&cl_object_site(obj)->cs_locks.cs_total);
- atomic_dec(&cl_object_site(obj)->cs_locks_state[lock->cll_state]);
- lu_object_ref_del_at(&obj->co_lu, lock->cll_obj_ref, "cl_lock", lock);
+ ENTRY;
+ cl_lock_trace(D_DLMTRACE, env, "free lock", lock);
+ while (!list_empty(&lock->cll_layers)) {
+ struct cl_lock_slice *slice;
+
+ slice = list_entry(lock->cll_layers.next,
+ struct cl_lock_slice, cls_linkage);
+ list_del_init(lock->cll_layers.next);
+ slice->cls_ops->clo_fini(env, slice);
+ }
+ CS_LOCK_DEC(obj, total);
+ CS_LOCKSTATE_DEC(obj, lock->cll_state);
+ lu_object_ref_del_at(&obj->co_lu, &lock->cll_obj_ref, "cl_lock", lock);
cl_object_put(env, obj);
lu_ref_fini(&lock->cll_reference);
lu_ref_fini(&lock->cll_holders);
- mutex_destroy(&lock->cll_guard);
+ mutex_destroy(&lock->cll_guard);
OBD_SLAB_FREE_PTR(lock, cl_lock_kmem);
EXIT;
}
void cl_lock_put(const struct lu_env *env, struct cl_lock *lock)
{
struct cl_object *obj;
- struct cl_object_header *head;
- struct cl_site *site;
LINVRNT(cl_lock_invariant(env, lock));
ENTRY;
obj = lock->cll_descr.cld_obj;
LINVRNT(obj != NULL);
- head = cl_object_header(obj);
- site = cl_object_site(obj);
- CDEBUG(D_DLMTRACE, "releasing reference: %d %p %lu\n",
- atomic_read(&lock->cll_ref), lock, RETIP);
+ CDEBUG(D_TRACE, "releasing reference: %d %p %lu\n",
+ atomic_read(&lock->cll_ref), lock, RETIP);
- if (atomic_dec_and_test(&lock->cll_ref)) {
+ if (atomic_dec_and_test(&lock->cll_ref)) {
if (lock->cll_state == CLS_FREEING) {
- LASSERT(list_empty(&lock->cll_linkage));
+ LASSERT(list_empty(&lock->cll_linkage));
cl_lock_free(env, lock);
}
- atomic_dec(&site->cs_locks.cs_busy);
+ CS_LOCK_DEC(obj, busy);
}
EXIT;
}
void cl_lock_get(struct cl_lock *lock)
{
LINVRNT(cl_lock_invariant(NULL, lock));
- CDEBUG(D_DLMTRACE|D_TRACE, "acquiring reference: %d %p %lu\n",
- atomic_read(&lock->cll_ref), lock, RETIP);
- atomic_inc(&lock->cll_ref);
+ CDEBUG(D_TRACE, "acquiring reference: %d %p %lu\n",
+ atomic_read(&lock->cll_ref), lock, RETIP);
+ atomic_inc(&lock->cll_ref);
}
EXPORT_SYMBOL(cl_lock_get);
*/
void cl_lock_get_trust(struct cl_lock *lock)
{
- struct cl_site *site = cl_object_site(lock->cll_descr.cld_obj);
-
- LASSERT(cl_is_lock(lock));
- CDEBUG(D_DLMTRACE|D_TRACE, "acquiring trusted reference: %d %p %lu\n",
- atomic_read(&lock->cll_ref), lock, RETIP);
- if (atomic_inc_return(&lock->cll_ref) == 1)
- atomic_inc(&site->cs_locks.cs_busy);
+ CDEBUG(D_TRACE, "acquiring trusted reference: %d %p %lu\n",
+ atomic_read(&lock->cll_ref), lock, RETIP);
+ if (atomic_inc_return(&lock->cll_ref) == 1)
+ CS_LOCK_INC(lock->cll_descr.cld_obj, busy);
}
EXPORT_SYMBOL(cl_lock_get_trust);
static void cl_lock_finish(const struct lu_env *env, struct cl_lock *lock)
{
cl_lock_mutex_get(env, lock);
+ cl_lock_cancel(env, lock);
cl_lock_delete(env, lock);
cl_lock_mutex_put(env, lock);
cl_lock_put(env, lock);
}
static struct cl_lock *cl_lock_alloc(const struct lu_env *env,
- struct cl_object *obj,
- const struct cl_io *io,
- const struct cl_lock_descr *descr)
+ struct cl_object *obj,
+ const struct cl_io *io,
+ const struct cl_lock_descr *descr)
+{
+ struct cl_lock *lock;
+ struct lu_object_header *head;
+
+ ENTRY;
+ OBD_SLAB_ALLOC_PTR_GFP(lock, cl_lock_kmem, GFP_NOFS);
+ if (lock != NULL) {
+ atomic_set(&lock->cll_ref, 1);
+ lock->cll_descr = *descr;
+ lock->cll_state = CLS_NEW;
+ cl_object_get(obj);
+ lu_object_ref_add_at(&obj->co_lu, &lock->cll_obj_ref, "cl_lock",
+ lock);
+ INIT_LIST_HEAD(&lock->cll_layers);
+ INIT_LIST_HEAD(&lock->cll_linkage);
+ INIT_LIST_HEAD(&lock->cll_inclosure);
+ lu_ref_init(&lock->cll_reference);
+ lu_ref_init(&lock->cll_holders);
+ mutex_init(&lock->cll_guard);
+ lockdep_set_class(&lock->cll_guard, &cl_lock_guard_class);
+ init_waitqueue_head(&lock->cll_wq);
+ head = obj->co_lu.lo_header;
+ CS_LOCKSTATE_INC(obj, CLS_NEW);
+ CS_LOCK_INC(obj, total);
+ CS_LOCK_INC(obj, create);
+ cl_lock_lockdep_init(lock);
+ list_for_each_entry(obj, &head->loh_layers, co_lu.lo_linkage) {
+ int err;
+
+ err = obj->co_ops->coo_lock_init(env, obj, lock, io);
+ if (err != 0) {
+ cl_lock_finish(env, lock);
+ lock = ERR_PTR(err);
+ break;
+ }
+ }
+ } else
+ lock = ERR_PTR(-ENOMEM);
+ RETURN(lock);
+}
+
+/**
+ * Transfer the lock into INTRANSIT state and return the original state.
+ *
+ * \pre state: CLS_CACHED, CLS_HELD or CLS_ENQUEUED
+ * \post state: CLS_INTRANSIT
+ * \see CLS_INTRANSIT
+ */
+enum cl_lock_state cl_lock_intransit(const struct lu_env *env,
+ struct cl_lock *lock)
{
- struct cl_lock *lock;
- struct lu_object_header *head;
- struct cl_site *site = cl_object_site(obj);
+ enum cl_lock_state state = lock->cll_state;
- ENTRY;
- OBD_SLAB_ALLOC_PTR(lock, cl_lock_kmem);
- if (lock != NULL) {
- atomic_set(&lock->cll_ref, 1);
- lock->cll_descr = *descr;
- lock->cll_state = CLS_NEW;
- cl_object_get(obj);
- lock->cll_obj_ref = lu_object_ref_add(&obj->co_lu,
- "cl_lock", lock);
- CFS_INIT_LIST_HEAD(&lock->cll_layers);
- CFS_INIT_LIST_HEAD(&lock->cll_linkage);
- CFS_INIT_LIST_HEAD(&lock->cll_inclosure);
- lu_ref_init(&lock->cll_reference);
- lu_ref_init(&lock->cll_holders);
- mutex_init(&lock->cll_guard);
- lockdep_set_class(&lock->cll_guard, &cl_lock_guard_class);
- cfs_waitq_init(&lock->cll_wq);
- head = obj->co_lu.lo_header;
- atomic_inc(&site->cs_locks_state[CLS_NEW]);
- atomic_inc(&site->cs_locks.cs_total);
- atomic_inc(&site->cs_locks.cs_created);
- cl_lock_lockdep_init(lock);
- list_for_each_entry(obj, &head->loh_layers, co_lu.lo_linkage) {
- int err;
-
- err = obj->co_ops->coo_lock_init(env, obj, lock, io);
- if (err != 0) {
- cl_lock_finish(env, lock);
- lock = ERR_PTR(err);
- break;
- }
- }
- } else
- lock = ERR_PTR(-ENOMEM);
- RETURN(lock);
+ LASSERT(cl_lock_is_mutexed(lock));
+ LASSERT(state != CLS_INTRANSIT);
+ LASSERTF(state >= CLS_ENQUEUED && state <= CLS_CACHED,
+ "Malformed lock state %d.\n", state);
+
+ cl_lock_state_set(env, lock, CLS_INTRANSIT);
+ lock->cll_intransit_owner = current;
+ cl_lock_hold_add(env, lock, "intransit", current);
+ return state;
+}
+EXPORT_SYMBOL(cl_lock_intransit);
+
+/**
+ * Exit the intransit state and restore the lock state to the original state
+ */
+void cl_lock_extransit(const struct lu_env *env, struct cl_lock *lock,
+ enum cl_lock_state state)
+{
+ LASSERT(cl_lock_is_mutexed(lock));
+ LASSERT(lock->cll_state == CLS_INTRANSIT);
+ LASSERT(state != CLS_INTRANSIT);
+ LASSERT(lock->cll_intransit_owner == current);
+
+ lock->cll_intransit_owner = NULL;
+ cl_lock_state_set(env, lock, state);
+ cl_lock_unhold(env, lock, "intransit", current);
}
+EXPORT_SYMBOL(cl_lock_extransit);
/**
+ * Checking whether the lock is intransit state
+ */
+int cl_lock_is_intransit(struct cl_lock *lock)
+{
+ LASSERT(cl_lock_is_mutexed(lock));
+ return lock->cll_state == CLS_INTRANSIT &&
+ lock->cll_intransit_owner != current;
+}
+EXPORT_SYMBOL(cl_lock_is_intransit);
+/**
* Returns true iff lock is "suitable" for given io. E.g., locks acquired by
* truncate and O_APPEND cannot be reused for read/non-append-write, as they
* cover multiple stripes and can trigger cascading timeouts.
LINVRNT(cl_lock_invariant_trusted(env, lock));
ENTRY;
- list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
+ list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
if (slice->cls_ops->clo_fits_into != NULL &&
!slice->cls_ops->clo_fits_into(env, slice, need, io))
RETURN(0);
{
struct cl_lock *lock;
struct cl_object_header *head;
- struct cl_site *site;
ENTRY;
- head = cl_object_header(obj);
- site = cl_object_site(obj);
- LINVRNT(spin_is_locked(&head->coh_lock_guard));
- atomic_inc(&site->cs_locks.cs_lookup);
- list_for_each_entry(lock, &head->coh_locks, cll_linkage) {
- int matched;
+ head = cl_object_header(obj);
+ assert_spin_locked(&head->coh_lock_guard);
+ CS_LOCK_INC(obj, lookup);
+ list_for_each_entry(lock, &head->coh_locks, cll_linkage) {
+ int matched;
- LASSERT(cl_is_lock(lock));
matched = cl_lock_ext_match(&lock->cll_descr, need) &&
- lock->cll_state < CLS_FREEING &&
- !(lock->cll_flags & CLF_CANCELLED) &&
- cl_lock_fits_into(env, lock, need, io);
- CDEBUG(D_DLMTRACE, "has: "DDESCR"(%i) need: "DDESCR": %d\n",
+ lock->cll_state < CLS_FREEING &&
+ lock->cll_error == 0 &&
+ !(lock->cll_flags & CLF_CANCELLED) &&
+ cl_lock_fits_into(env, lock, need, io);
+ CDEBUG(D_DLMTRACE, "has: "DDESCR"(%d) need: "DDESCR": %d\n",
PDESCR(&lock->cll_descr), lock->cll_state, PDESCR(need),
matched);
if (matched) {
cl_lock_get_trust(lock);
- /* move the lock to the LRU head */
- list_move(&lock->cll_linkage, &head->coh_locks);
- atomic_inc(&cl_object_site(obj)->cs_locks.cs_hit);
+ CS_LOCK_INC(obj, hit);
RETURN(lock);
}
}
struct cl_object_header *head;
struct cl_object *obj;
struct cl_lock *lock;
- struct cl_site *site;
ENTRY;
obj = need->cld_obj;
head = cl_object_header(obj);
- site = cl_object_site(obj);
-
- spin_lock(&head->coh_lock_guard);
- lock = cl_lock_lookup(env, obj, io, need);
- spin_unlock(&head->coh_lock_guard);
-
- if (lock == NULL) {
- lock = cl_lock_alloc(env, obj, io, need);
- if (!IS_ERR(lock)) {
- struct cl_lock *ghost;
-
- spin_lock(&head->coh_lock_guard);
- ghost = cl_lock_lookup(env, obj, io, need);
- if (ghost == NULL) {
- list_add(&lock->cll_linkage, &head->coh_locks);
- spin_unlock(&head->coh_lock_guard);
- atomic_inc(&site->cs_locks.cs_busy);
- } else {
- spin_unlock(&head->coh_lock_guard);
+
+ spin_lock(&head->coh_lock_guard);
+ lock = cl_lock_lookup(env, obj, io, need);
+ spin_unlock(&head->coh_lock_guard);
+
+ if (lock == NULL) {
+ lock = cl_lock_alloc(env, obj, io, need);
+ if (!IS_ERR(lock)) {
+ struct cl_lock *ghost;
+
+ spin_lock(&head->coh_lock_guard);
+ ghost = cl_lock_lookup(env, obj, io, need);
+ if (ghost == NULL) {
+ cl_lock_get_trust(lock);
+ list_add_tail(&lock->cll_linkage,
+ &head->coh_locks);
+ spin_unlock(&head->coh_lock_guard);
+ CS_LOCK_INC(obj, busy);
+ } else {
+ spin_unlock(&head->coh_lock_guard);
/*
* Other threads can acquire references to the
* top-lock through its sub-locks. Hence, it
obj = need->cld_obj;
head = cl_object_header(obj);
- spin_lock(&head->coh_lock_guard);
- lock = cl_lock_lookup(env, obj, io, need);
- spin_unlock(&head->coh_lock_guard);
-
- if (lock != NULL) {
- int ok;
-
- cl_lock_mutex_get(env, lock);
- if (lock->cll_state == CLS_CACHED)
- cl_use_try(env, lock);
- ok = lock->cll_state == CLS_HELD;
- if (ok) {
- cl_lock_hold_add(env, lock, scope, source);
- cl_lock_user_add(env, lock);
- }
+ do {
+ spin_lock(&head->coh_lock_guard);
+ lock = cl_lock_lookup(env, obj, io, need);
+ spin_unlock(&head->coh_lock_guard);
+ if (lock == NULL)
+ return NULL;
+
+ cl_lock_mutex_get(env, lock);
+ if (lock->cll_state == CLS_INTRANSIT)
+ /* Don't care return value. */
+ cl_lock_state_wait(env, lock);
+ if (lock->cll_state == CLS_FREEING) {
+ cl_lock_mutex_put(env, lock);
+ cl_lock_put(env, lock);
+ lock = NULL;
+ }
+ } while (lock == NULL);
+
+ cl_lock_hold_add(env, lock, scope, source);
+ cl_lock_user_add(env, lock);
+ if (lock->cll_state == CLS_CACHED)
+ cl_use_try(env, lock, 1);
+ if (lock->cll_state == CLS_HELD) {
+ cl_lock_mutex_put(env, lock);
+ cl_lock_lockdep_acquire(env, lock, 0);
+ cl_lock_put(env, lock);
+ } else {
+ cl_unuse_try(env, lock);
+ cl_lock_unhold(env, lock, scope, source);
cl_lock_mutex_put(env, lock);
- if (!ok) {
- cl_lock_put(env, lock);
- lock = NULL;
- }
+ cl_lock_put(env, lock);
+ lock = NULL;
}
+
return lock;
}
EXPORT_SYMBOL(cl_lock_peek);
LINVRNT(cl_lock_invariant_trusted(NULL, lock));
ENTRY;
- list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
+ list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
if (slice->cls_obj->co_lu.lo_dev->ld_type == dtype)
RETURN(slice);
}
}
EXPORT_SYMBOL(cl_lock_at);
-static void cl_lock_trace(struct cl_thread_info *info,
- const char *prefix, const struct cl_lock *lock)
-{
- CDEBUG(D_DLMTRACE|D_TRACE, "%s: %i@%p %p %i %i\n", prefix,
- atomic_read(&lock->cll_ref), lock, lock->cll_guarder,
- lock->cll_depth, info->clt_nr_locks_locked);
-}
-
static void cl_lock_mutex_tail(const struct lu_env *env, struct cl_lock *lock)
{
- struct cl_thread_info *info;
+ struct cl_thread_counters *counters;
- info = cl_env_info(env);
+ counters = cl_lock_counters(env, lock);
lock->cll_depth++;
- info->clt_nr_locks_locked++;
- lu_ref_add(&info->clt_locks_locked, "cll_guard", lock);
- cl_lock_trace(info, "got mutex", lock);
+ counters->ctc_nr_locks_locked++;
+ lu_ref_add(&counters->ctc_locks_locked, "cll_guard", lock);
+ cl_lock_trace(D_TRACE, env, "got mutex", lock);
}
/**
*/
void cl_lock_mutex_get(const struct lu_env *env, struct cl_lock *lock)
{
- LINVRNT(cl_lock_invariant(env, lock));
-
- if (lock->cll_guarder == cfs_current()) {
- LINVRNT(cl_lock_is_mutexed(lock));
- LINVRNT(lock->cll_depth > 0);
- } else {
- struct cl_object_header *hdr;
-
- LINVRNT(lock->cll_guarder != cfs_current());
- hdr = cl_object_header(lock->cll_descr.cld_obj);
- mutex_lock_nested(&lock->cll_guard, hdr->coh_nesting);
- lock->cll_guarder = cfs_current();
- LINVRNT(lock->cll_depth == 0);
- }
- cl_lock_mutex_tail(env, lock);
+ LINVRNT(cl_lock_invariant(env, lock));
+
+ if (lock->cll_guarder == current) {
+ LINVRNT(cl_lock_is_mutexed(lock));
+ LINVRNT(lock->cll_depth > 0);
+ } else {
+ struct cl_object_header *hdr;
+ struct cl_thread_info *info;
+ int i;
+
+ LINVRNT(lock->cll_guarder != current);
+ hdr = cl_object_header(lock->cll_descr.cld_obj);
+ /*
+ * Check that mutices are taken in the bottom-to-top order.
+ */
+ info = cl_env_info(env);
+ for (i = 0; i < hdr->coh_nesting; ++i)
+ LASSERT(info->clt_counters[i].ctc_nr_locks_locked == 0);
+ mutex_lock_nested(&lock->cll_guard, hdr->coh_nesting);
+ lock->cll_guarder = current;
+ LINVRNT(lock->cll_depth == 0);
+ }
+ cl_lock_mutex_tail(env, lock);
}
EXPORT_SYMBOL(cl_lock_mutex_get);
*/
int cl_lock_mutex_try(const struct lu_env *env, struct cl_lock *lock)
{
- int result;
+ int result;
- LINVRNT(cl_lock_invariant_trusted(env, lock));
- ENTRY;
+ LINVRNT(cl_lock_invariant_trusted(env, lock));
+ ENTRY;
- result = 0;
- if (lock->cll_guarder == cfs_current()) {
- LINVRNT(lock->cll_depth > 0);
- cl_lock_mutex_tail(env, lock);
- } else if (mutex_trylock(&lock->cll_guard)) {
- LINVRNT(lock->cll_depth == 0);
- lock->cll_guarder = cfs_current();
- cl_lock_mutex_tail(env, lock);
- } else
- result = -EBUSY;
- RETURN(result);
+ result = 0;
+ if (lock->cll_guarder == current) {
+ LINVRNT(lock->cll_depth > 0);
+ cl_lock_mutex_tail(env, lock);
+ } else if (mutex_trylock(&lock->cll_guard)) {
+ LINVRNT(lock->cll_depth == 0);
+ lock->cll_guarder = current;
+ cl_lock_mutex_tail(env, lock);
+ } else
+ result = -EBUSY;
+ RETURN(result);
}
EXPORT_SYMBOL(cl_lock_mutex_try);
/**
- * Unlocks cl_lock object.
+ {* Unlocks cl_lock object.
*
* \pre cl_lock_is_mutexed(lock)
*
*/
void cl_lock_mutex_put(const struct lu_env *env, struct cl_lock *lock)
{
- struct cl_thread_info *info;
+ struct cl_thread_counters *counters;
- LINVRNT(cl_lock_invariant(env, lock));
- LINVRNT(cl_lock_is_mutexed(lock));
- LINVRNT(lock->cll_guarder == cfs_current());
- LINVRNT(lock->cll_depth > 0);
+ LINVRNT(cl_lock_invariant(env, lock));
+ LINVRNT(cl_lock_is_mutexed(lock));
+ LINVRNT(lock->cll_guarder == current);
+ LINVRNT(lock->cll_depth > 0);
- info = cl_env_info(env);
- LINVRNT(info->clt_nr_locks_locked > 0);
-
- cl_lock_trace(info, "put mutex", lock);
- lu_ref_del(&info->clt_locks_locked, "cll_guard", lock);
- info->clt_nr_locks_locked--;
- if (--lock->cll_depth == 0) {
- lock->cll_guarder = NULL;
- mutex_unlock(&lock->cll_guard);
- }
+ counters = cl_lock_counters(env, lock);
+ LINVRNT(counters->ctc_nr_locks_locked > 0);
+
+ cl_lock_trace(D_TRACE, env, "put mutex", lock);
+ lu_ref_del(&counters->ctc_locks_locked, "cll_guard", lock);
+ counters->ctc_nr_locks_locked--;
+ if (--lock->cll_depth == 0) {
+ lock->cll_guarder = NULL;
+ mutex_unlock(&lock->cll_guard);
+ }
}
EXPORT_SYMBOL(cl_lock_mutex_put);
*/
int cl_lock_is_mutexed(struct cl_lock *lock)
{
- return lock->cll_guarder == cfs_current();
+ return lock->cll_guarder == current;
}
EXPORT_SYMBOL(cl_lock_is_mutexed);
*/
int cl_lock_nr_mutexed(const struct lu_env *env)
{
- return cl_env_info(env)->clt_nr_locks_locked;
+ struct cl_thread_info *info;
+ int i;
+ int locked;
+
+ /*
+ * NOTE: if summation across all nesting levels (currently 2) proves
+ * too expensive, a summary counter can be added to
+ * struct cl_thread_info.
+ */
+ info = cl_env_info(env);
+ for (i = 0, locked = 0; i < ARRAY_SIZE(info->clt_counters); ++i)
+ locked += info->clt_counters[i].ctc_nr_locks_locked;
+ return locked;
}
EXPORT_SYMBOL(cl_lock_nr_mutexed);
const struct cl_lock_slice *slice;
lock->cll_flags |= CLF_CANCELLED;
- list_for_each_entry_reverse(slice, &lock->cll_layers,
- cls_linkage) {
+ list_for_each_entry_reverse(slice, &lock->cll_layers,
+ cls_linkage) {
if (slice->cls_ops->clo_cancel != NULL)
slice->cls_ops->clo_cancel(env, slice);
}
ENTRY;
if (lock->cll_state < CLS_FREEING) {
+ bool in_cache;
+
+ LASSERT(lock->cll_state != CLS_INTRANSIT);
cl_lock_state_set(env, lock, CLS_FREEING);
head = cl_object_header(lock->cll_descr.cld_obj);
- spin_lock(&head->coh_lock_guard);
- list_del_init(&lock->cll_linkage);
- /*
- * No locks, no pages. This is only valid for bottom sub-locks
- * and head->coh_nesting == 1 check assumes two level top-sub
- * hierarchy.
- */
- LASSERT(ergo(head->coh_nesting == 1 &&
- list_empty(&head->coh_locks), !head->coh_pages));
- spin_unlock(&head->coh_lock_guard);
- /*
- * From now on, no new references to this lock can be acquired
- * by cl_lock_lookup().
- */
- list_for_each_entry_reverse(slice, &lock->cll_layers,
- cls_linkage) {
- if (slice->cls_ops->clo_delete != NULL)
- slice->cls_ops->clo_delete(env, slice);
- }
- /*
- * From now on, no new references to this lock can be acquired
- * by layer-specific means (like a pointer from struct
- * ldlm_lock in osc, or a pointer from top-lock to sub-lock in
- * lov).
- *
- * Lock will be finally freed in cl_lock_put() when last of
- * existing references goes away.
- */
- }
- EXIT;
+ spin_lock(&head->coh_lock_guard);
+ in_cache = !list_empty(&lock->cll_linkage);
+ if (in_cache)
+ list_del_init(&lock->cll_linkage);
+ spin_unlock(&head->coh_lock_guard);
+
+ if (in_cache) /* coh_locks cache holds a refcount. */
+ cl_lock_put(env, lock);
+
+ /*
+ * From now on, no new references to this lock can be acquired
+ * by cl_lock_lookup().
+ */
+ list_for_each_entry_reverse(slice, &lock->cll_layers,
+ cls_linkage) {
+ if (slice->cls_ops->clo_delete != NULL)
+ slice->cls_ops->clo_delete(env, slice);
+ }
+ /*
+ * From now on, no new references to this lock can be acquired
+ * by layer-specific means (like a pointer from struct
+ * ldlm_lock in osc, or a pointer from top-lock to sub-lock in
+ * lov).
+ *
+ * Lock will be finally freed in cl_lock_put() when last of
+ * existing references goes away.
+ */
+ }
+ EXIT;
}
+/**
+ * Mod(ifie)s cl_lock::cll_holds counter for a given lock. Also, for a
+ * top-lock (nesting == 0) accounts for this modification in the per-thread
+ * debugging counters. Sub-lock holds can be released by a thread different
+ * from one that acquired it.
+ */
static void cl_lock_hold_mod(const struct lu_env *env, struct cl_lock *lock,
int delta)
{
- struct cl_thread_info *cti;
- struct cl_object_header *hdr;
+ struct cl_thread_counters *counters;
+ enum clt_nesting_level nesting;
- cti = cl_env_info(env);
- hdr = cl_object_header(lock->cll_descr.cld_obj);
lock->cll_holds += delta;
- if (hdr->coh_nesting == 0) {
- cti->clt_nr_held += delta;
- LASSERT(cti->clt_nr_held >= 0);
+ nesting = cl_lock_nesting(lock);
+ if (nesting == CNL_TOP) {
+ counters = &cl_env_info(env)->clt_counters[CNL_TOP];
+ counters->ctc_nr_held += delta;
+ LASSERT(counters->ctc_nr_held >= 0);
}
}
+/**
+ * Mod(ifie)s cl_lock::cll_users counter for a given lock. See
+ * cl_lock_hold_mod() for the explanation of the debugging code.
+ */
static void cl_lock_used_mod(const struct lu_env *env, struct cl_lock *lock,
int delta)
{
- struct cl_thread_info *cti;
- struct cl_object_header *hdr;
+ struct cl_thread_counters *counters;
+ enum clt_nesting_level nesting;
- cti = cl_env_info(env);
- hdr = cl_object_header(lock->cll_descr.cld_obj);
lock->cll_users += delta;
- if (hdr->coh_nesting == 0) {
- cti->clt_nr_used += delta;
- LASSERT(cti->clt_nr_used >= 0);
+ nesting = cl_lock_nesting(lock);
+ if (nesting == CNL_TOP) {
+ counters = &cl_env_info(env)->clt_counters[CNL_TOP];
+ counters->ctc_nr_used += delta;
+ LASSERT(counters->ctc_nr_used >= 0);
}
}
-static void cl_lock_hold_release(const struct lu_env *env, struct cl_lock *lock,
- const char *scope, const void *source)
+void cl_lock_hold_release(const struct lu_env *env, struct cl_lock *lock,
+ const char *scope, const void *source)
{
LINVRNT(cl_lock_is_mutexed(lock));
LINVRNT(cl_lock_invariant(env, lock));
LASSERT(lock->cll_holds > 0);
ENTRY;
+ cl_lock_trace(D_DLMTRACE, env, "hold release lock", lock);
lu_ref_del(&lock->cll_holders, scope, source);
cl_lock_hold_mod(env, lock, -1);
if (lock->cll_holds == 0) {
- if (lock->cll_descr.cld_mode == CLM_PHANTOM)
+ CL_LOCK_ASSERT(lock->cll_state != CLS_HELD, env, lock);
+ if (lock->cll_descr.cld_mode == CLM_PHANTOM ||
+ lock->cll_descr.cld_mode == CLM_GROUP ||
+ lock->cll_state != CLS_CACHED)
/*
- * If lock is still phantom when user is done with
- * it---destroy the lock.
+ * If lock is still phantom or grouplock when user is
+ * done with it---destroy the lock.
*/
lock->cll_flags |= CLF_CANCELPEND|CLF_DOOMED;
if (lock->cll_flags & CLF_CANCELPEND) {
}
EXIT;
}
-
+EXPORT_SYMBOL(cl_lock_hold_release);
/**
* Waits until lock state is changed.
*/
int cl_lock_state_wait(const struct lu_env *env, struct cl_lock *lock)
{
- cfs_waitlink_t waiter;
- int result;
-
- ENTRY;
- LINVRNT(cl_lock_is_mutexed(lock));
- LINVRNT(cl_lock_invariant(env, lock));
- LASSERT(lock->cll_depth == 1);
- LASSERT(lock->cll_state != CLS_FREEING); /* too late to wait */
-
- result = lock->cll_error;
- if (result == 0 && !(lock->cll_flags & CLF_STATE)) {
- cfs_waitlink_init(&waiter);
- cfs_waitq_add(&lock->cll_wq, &waiter);
- set_current_state(CFS_TASK_INTERRUPTIBLE);
- cl_lock_mutex_put(env, lock);
-
- LASSERT(cl_lock_nr_mutexed(env) == 0);
- cfs_waitq_wait(&waiter, CFS_TASK_INTERRUPTIBLE);
-
- cl_lock_mutex_get(env, lock);
- set_current_state(CFS_TASK_RUNNING);
- cfs_waitq_del(&lock->cll_wq, &waiter);
- result = cfs_signal_pending() ? -EINTR : 0;
- }
- lock->cll_flags &= ~CLF_STATE;
- RETURN(result);
+ wait_queue_t waiter;
+ sigset_t blocked;
+ int result;
+
+ ENTRY;
+ LINVRNT(cl_lock_is_mutexed(lock));
+ LINVRNT(cl_lock_invariant(env, lock));
+ LASSERT(lock->cll_depth == 1);
+ LASSERT(lock->cll_state != CLS_FREEING); /* too late to wait */
+
+ cl_lock_trace(D_DLMTRACE, env, "state wait lock", lock);
+ result = lock->cll_error;
+ if (result == 0) {
+ /* To avoid being interrupted by the 'non-fatal' signals
+ * (SIGCHLD, for instance), we'd block them temporarily.
+ * LU-305 */
+ blocked = cfs_block_sigsinv(LUSTRE_FATAL_SIGS);
+
+ init_waitqueue_entry_current(&waiter);
+ add_wait_queue(&lock->cll_wq, &waiter);
+ set_current_state(TASK_INTERRUPTIBLE);
+ cl_lock_mutex_put(env, lock);
+
+ LASSERT(cl_lock_nr_mutexed(env) == 0);
+
+ /* Returning ERESTARTSYS instead of EINTR so syscalls
+ * can be restarted if signals are pending here */
+ result = -ERESTARTSYS;
+ if (likely(!OBD_FAIL_CHECK(OBD_FAIL_LOCK_STATE_WAIT_INTR))) {
+ waitq_wait(&waiter, TASK_INTERRUPTIBLE);
+ if (!cfs_signal_pending())
+ result = 0;
+ }
+
+ cl_lock_mutex_get(env, lock);
+ set_current_state(TASK_RUNNING);
+ remove_wait_queue(&lock->cll_wq, &waiter);
+
+ /* Restore old blocked signals */
+ cfs_restore_sigs(blocked);
+ }
+ RETURN(result);
}
EXPORT_SYMBOL(cl_lock_state_wait);
static void cl_lock_state_signal(const struct lu_env *env, struct cl_lock *lock,
- enum cl_lock_state state)
+ enum cl_lock_state state)
{
- const struct cl_lock_slice *slice;
+ const struct cl_lock_slice *slice;
- ENTRY;
- LINVRNT(cl_lock_is_mutexed(lock));
- LINVRNT(cl_lock_invariant(env, lock));
+ ENTRY;
+ LINVRNT(cl_lock_is_mutexed(lock));
+ LINVRNT(cl_lock_invariant(env, lock));
- list_for_each_entry(slice, &lock->cll_layers, cls_linkage)
- if (slice->cls_ops->clo_state != NULL)
- slice->cls_ops->clo_state(env, slice, state);
- lock->cll_flags |= CLF_STATE;
- cfs_waitq_broadcast(&lock->cll_wq);
- EXIT;
+ list_for_each_entry(slice, &lock->cll_layers, cls_linkage)
+ if (slice->cls_ops->clo_state != NULL)
+ slice->cls_ops->clo_state(env, slice, state);
+ wake_up_all(&lock->cll_wq);
+ EXIT;
}
/**
void cl_lock_signal(const struct lu_env *env, struct cl_lock *lock)
{
ENTRY;
+ cl_lock_trace(D_DLMTRACE, env, "state signal lock", lock);
cl_lock_state_signal(env, lock, lock->cll_state);
EXIT;
}
void cl_lock_state_set(const struct lu_env *env, struct cl_lock *lock,
enum cl_lock_state state)
{
- struct cl_site *site = cl_object_site(lock->cll_descr.cld_obj);
-
ENTRY;
LASSERT(lock->cll_state <= state ||
(lock->cll_state == CLS_CACHED &&
(state == CLS_HELD || /* lock found in cache */
- state == CLS_NEW /* sub-lock canceled */)) ||
- /* sub-lock canceled during unlocking */
- (lock->cll_state == CLS_UNLOCKING && state == CLS_NEW));
+ state == CLS_NEW || /* sub-lock canceled */
+ state == CLS_INTRANSIT)) ||
+ /* lock is in transit state */
+ lock->cll_state == CLS_INTRANSIT);
if (lock->cll_state != state) {
- atomic_dec(&site->cs_locks_state[lock->cll_state]);
- atomic_inc(&site->cs_locks_state[state]);
+ CS_LOCKSTATE_DEC(lock->cll_descr.cld_obj, lock->cll_state);
+ CS_LOCKSTATE_INC(lock->cll_descr.cld_obj, state);
cl_lock_state_signal(env, lock, state);
lock->cll_state = state;
}
EXPORT_SYMBOL(cl_lock_state_set);
+static int cl_unuse_try_internal(const struct lu_env *env, struct cl_lock *lock)
+{
+ const struct cl_lock_slice *slice;
+ int result;
+
+ do {
+ result = 0;
+
+ LINVRNT(cl_lock_is_mutexed(lock));
+ LINVRNT(cl_lock_invariant(env, lock));
+ LASSERT(lock->cll_state == CLS_INTRANSIT);
+
+ result = -ENOSYS;
+ list_for_each_entry_reverse(slice, &lock->cll_layers,
+ cls_linkage) {
+ if (slice->cls_ops->clo_unuse != NULL) {
+ result = slice->cls_ops->clo_unuse(env, slice);
+ if (result != 0)
+ break;
+ }
+ }
+ LASSERT(result != -ENOSYS);
+ } while (result == CLO_REPEAT);
+
+ return result;
+}
+
/**
* Yanks lock from the cache (cl_lock_state::CLS_CACHED state) by calling
* cl_lock_operations::clo_use() top-to-bottom to notify layers.
+ * @atomic = 1, it must unuse the lock to recovery the lock to keep the
+ * use process atomic
*/
-int cl_use_try(const struct lu_env *env, struct cl_lock *lock)
+int cl_use_try(const struct lu_env *env, struct cl_lock *lock, int atomic)
{
- int result;
const struct cl_lock_slice *slice;
+ int result;
+ enum cl_lock_state state;
ENTRY;
+ cl_lock_trace(D_DLMTRACE, env, "use lock", lock);
+
+ LASSERT(lock->cll_state == CLS_CACHED);
+ if (lock->cll_error)
+ RETURN(lock->cll_error);
+
result = -ENOSYS;
- list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
+ state = cl_lock_intransit(env, lock);
+ list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
if (slice->cls_ops->clo_use != NULL) {
result = slice->cls_ops->clo_use(env, slice);
if (result != 0)
}
}
LASSERT(result != -ENOSYS);
- if (result == 0)
- cl_lock_state_set(env, lock, CLS_HELD);
+
+ LASSERTF(lock->cll_state == CLS_INTRANSIT, "Wrong state %d.\n",
+ lock->cll_state);
+
+ if (result == 0) {
+ state = CLS_HELD;
+ } else {
+ if (result == -ESTALE) {
+ /*
+ * ESTALE means sublock being cancelled
+ * at this time, and set lock state to
+ * be NEW here and ask the caller to repeat.
+ */
+ state = CLS_NEW;
+ result = CLO_REPEAT;
+ }
+
+ /* @atomic means back-off-on-failure. */
+ if (atomic) {
+ int rc;
+ rc = cl_unuse_try_internal(env, lock);
+ /* Vet the results. */
+ if (rc < 0 && result > 0)
+ result = rc;
+ }
+
+ }
+ cl_lock_extransit(env, lock, state);
RETURN(result);
}
EXPORT_SYMBOL(cl_use_try);
ENTRY;
result = -ENOSYS;
- list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
+ list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
if (slice->cls_ops->clo_enqueue != NULL) {
result = slice->cls_ops->clo_enqueue(env,
slice, io, flags);
int result;
ENTRY;
+ cl_lock_trace(D_DLMTRACE, env, "enqueue lock", lock);
do {
- result = 0;
-
LINVRNT(cl_lock_is_mutexed(lock));
- if (lock->cll_error != 0)
+ result = lock->cll_error;
+ if (result != 0)
break;
+
switch (lock->cll_state) {
case CLS_NEW:
cl_lock_state_set(env, lock, CLS_QUEUING);
case CLS_QUEUING:
/* kick layers. */
result = cl_enqueue_kick(env, lock, io, flags);
- if (result == 0)
+ /* For AGL case, the cl_lock::cll_state may
+ * become CLS_HELD already. */
+ if (result == 0 && lock->cll_state == CLS_QUEUING)
cl_lock_state_set(env, lock, CLS_ENQUEUED);
break;
- case CLS_UNLOCKING:
- /* wait until unlocking finishes, and enqueue lock
- * afresh. */
+ case CLS_INTRANSIT:
+ LASSERT(cl_lock_is_intransit(lock));
result = CLO_WAIT;
break;
case CLS_CACHED:
/* yank lock from the cache. */
- result = cl_use_try(env, lock);
+ result = cl_use_try(env, lock, 0);
break;
case CLS_ENQUEUED:
case CLS_HELD:
LBUG();
}
} while (result == CLO_REPEAT);
- if (result < 0)
- cl_lock_error(env, lock, result);
- RETURN(result ?: lock->cll_error);
+ RETURN(result);
}
EXPORT_SYMBOL(cl_enqueue_try);
+/**
+ * Cancel the conflicting lock found during previous enqueue.
+ *
+ * \retval 0 conflicting lock has been canceled.
+ * \retval -ve error code.
+ */
+int cl_lock_enqueue_wait(const struct lu_env *env,
+ struct cl_lock *lock,
+ int keep_mutex)
+{
+ struct cl_lock *conflict;
+ int rc = 0;
+ ENTRY;
+
+ LASSERT(cl_lock_is_mutexed(lock));
+ LASSERT(lock->cll_state == CLS_QUEUING);
+ LASSERT(lock->cll_conflict != NULL);
+
+ conflict = lock->cll_conflict;
+ lock->cll_conflict = NULL;
+
+ cl_lock_mutex_put(env, lock);
+ LASSERT(cl_lock_nr_mutexed(env) == 0);
+
+ cl_lock_mutex_get(env, conflict);
+ cl_lock_trace(D_DLMTRACE, env, "enqueue wait", conflict);
+ cl_lock_cancel(env, conflict);
+ cl_lock_delete(env, conflict);
+
+ while (conflict->cll_state != CLS_FREEING) {
+ rc = cl_lock_state_wait(env, conflict);
+ if (rc != 0)
+ break;
+ }
+ cl_lock_mutex_put(env, conflict);
+ lu_ref_del(&conflict->cll_reference, "cancel-wait", lock);
+ cl_lock_put(env, conflict);
+
+ if (keep_mutex)
+ cl_lock_mutex_get(env, lock);
+
+ LASSERT(rc <= 0);
+ RETURN(rc);
+}
+EXPORT_SYMBOL(cl_lock_enqueue_wait);
+
static int cl_enqueue_locked(const struct lu_env *env, struct cl_lock *lock,
struct cl_io *io, __u32 enqflags)
{
do {
result = cl_enqueue_try(env, lock, io, enqflags);
if (result == CLO_WAIT) {
- result = cl_lock_state_wait(env, lock);
+ if (lock->cll_conflict != NULL)
+ result = cl_lock_enqueue_wait(env, lock, 1);
+ else
+ result = cl_lock_state_wait(env, lock);
if (result == 0)
continue;
}
break;
} while (1);
- if (result != 0) {
- cl_lock_user_del(env, lock);
- if (result != -EINTR)
- cl_lock_error(env, lock, result);
- }
- LASSERT(ergo(result == 0, lock->cll_state == CLS_ENQUEUED ||
+ if (result != 0)
+ cl_unuse_try(env, lock);
+ LASSERT(ergo(result == 0 && !(enqflags & CEF_AGL),
+ lock->cll_state == CLS_ENQUEUED ||
lock->cll_state == CLS_HELD));
RETURN(result);
}
/**
* Tries to unlock a lock.
*
- * This function is called repeatedly by cl_unuse() until either lock is
- * unlocked, or error occurs.
+ * This function is called to release underlying resource:
+ * 1. for top lock, the resource is sublocks it held;
+ * 2. for sublock, the resource is the reference to dlmlock.
*
- * \ppre lock->cll_state <= CLS_HELD || lock->cll_state == CLS_UNLOCKING
- *
- * \post ergo(result == 0, lock->cll_state == CLS_CACHED)
+ * cl_unuse_try is a one-shot operation, so it must NOT return CLO_WAIT.
*
* \see cl_unuse() cl_lock_operations::clo_unuse()
* \see cl_lock_state::CLS_CACHED
*/
int cl_unuse_try(const struct lu_env *env, struct cl_lock *lock)
{
- const struct cl_lock_slice *slice;
int result;
+ enum cl_lock_state state = CLS_NEW;
ENTRY;
- if (lock->cll_state != CLS_UNLOCKING) {
- if (lock->cll_users > 1) {
- cl_lock_user_del(env, lock);
- RETURN(0);
- }
- /*
- * New lock users (->cll_users) are not protecting unlocking
- * from proceeding. From this point, lock eventually reaches
- * CLS_CACHED, is reinitialized to CLS_NEW or fails into
- * CLS_FREEING.
- */
- cl_lock_state_set(env, lock, CLS_UNLOCKING);
+ cl_lock_trace(D_DLMTRACE, env, "unuse lock", lock);
+
+ if (lock->cll_users > 1) {
+ cl_lock_user_del(env, lock);
+ RETURN(0);
}
- do {
- result = 0;
- if (lock->cll_error != 0)
- break;
+ /* Only if the lock is in CLS_HELD or CLS_ENQUEUED state, it can hold
+ * underlying resources. */
+ if (!(lock->cll_state == CLS_HELD || lock->cll_state == CLS_ENQUEUED)) {
+ cl_lock_user_del(env, lock);
+ RETURN(0);
+ }
- LINVRNT(cl_lock_is_mutexed(lock));
- LINVRNT(cl_lock_invariant(env, lock));
- LASSERT(lock->cll_state == CLS_UNLOCKING);
- LASSERT(lock->cll_users > 0);
- LASSERT(lock->cll_holds > 0);
+ /*
+ * New lock users (->cll_users) are not protecting unlocking
+ * from proceeding. From this point, lock eventually reaches
+ * CLS_CACHED, is reinitialized to CLS_NEW or fails into
+ * CLS_FREEING.
+ */
+ state = cl_lock_intransit(env, lock);
- result = -ENOSYS;
- list_for_each_entry_reverse(slice, &lock->cll_layers,
- cls_linkage) {
- if (slice->cls_ops->clo_unuse != NULL) {
- result = slice->cls_ops->clo_unuse(env, slice);
- if (result != 0)
- break;
- }
- }
- LASSERT(result != -ENOSYS);
- } while (result == CLO_REPEAT);
- if (result != CLO_WAIT)
- /*
- * Once there is no more need to iterate ->clo_unuse() calls,
- * remove lock user. This is done even if unrecoverable error
- * happened during unlocking, because nothing else can be
- * done.
- */
- cl_lock_user_del(env, lock);
+ result = cl_unuse_try_internal(env, lock);
+ LASSERT(lock->cll_state == CLS_INTRANSIT);
+ LASSERT(result != CLO_WAIT);
+ cl_lock_user_del(env, lock);
if (result == 0 || result == -ESTALE) {
- enum cl_lock_state state;
-
/*
* Return lock back to the cache. This is the only
* place where lock is moved into CLS_CACHED state.
* re-initialized. This happens e.g., when a sub-lock was
* canceled while unlocking was in progress.
*/
- state = result == 0 ? CLS_CACHED : CLS_NEW;
- cl_lock_state_set(env, lock, state);
+ if (state == CLS_HELD && result == 0)
+ state = CLS_CACHED;
+ else
+ state = CLS_NEW;
+ cl_lock_extransit(env, lock, state);
/*
* Hide -ESTALE error.
* pages won't be written to OSTs. -jay
*/
result = 0;
+ } else {
+ CERROR("result = %d, this is unlikely!\n", result);
+ state = CLS_NEW;
+ cl_lock_extransit(env, lock, state);
}
- result = result ?: lock->cll_error;
- if (result < 0)
- cl_lock_error(env, lock, result);
- RETURN(result);
+ RETURN(result ?: lock->cll_error);
}
EXPORT_SYMBOL(cl_unuse_try);
static void cl_unuse_locked(const struct lu_env *env, struct cl_lock *lock)
{
+ int result;
ENTRY;
- LASSERT(lock->cll_state <= CLS_HELD);
- do {
- int result;
- result = cl_unuse_try(env, lock);
- if (result == CLO_WAIT) {
- result = cl_lock_state_wait(env, lock);
- if (result == 0)
- continue;
- }
- break;
- } while (1);
+ result = cl_unuse_try(env, lock);
+ if (result)
+ CL_LOCK_DEBUG(D_ERROR, env, lock, "unuse return %d\n", result);
+
EXIT;
}
int result;
ENTRY;
+ cl_lock_trace(D_DLMTRACE, env, "wait lock try", lock);
do {
LINVRNT(cl_lock_is_mutexed(lock));
LINVRNT(cl_lock_invariant(env, lock));
- LASSERT(lock->cll_state == CLS_ENQUEUED ||
- lock->cll_state == CLS_HELD);
+ LASSERTF(lock->cll_state == CLS_QUEUING ||
+ lock->cll_state == CLS_ENQUEUED ||
+ lock->cll_state == CLS_HELD ||
+ lock->cll_state == CLS_INTRANSIT,
+ "lock state: %d\n", lock->cll_state);
LASSERT(lock->cll_users > 0);
LASSERT(lock->cll_holds > 0);
- result = 0;
- if (lock->cll_error != 0)
+ result = lock->cll_error;
+ if (result != 0)
break;
+
+ if (cl_lock_is_intransit(lock)) {
+ result = CLO_WAIT;
+ break;
+ }
+
if (lock->cll_state == CLS_HELD)
/* nothing to do */
break;
result = -ENOSYS;
- list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
+ list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
if (slice->cls_ops->clo_wait != NULL) {
result = slice->cls_ops->clo_wait(env, slice);
if (result != 0)
}
}
LASSERT(result != -ENOSYS);
- if (result == 0)
+ if (result == 0) {
+ LASSERT(lock->cll_state != CLS_INTRANSIT);
cl_lock_state_set(env, lock, CLS_HELD);
+ }
} while (result == CLO_REPEAT);
- RETURN(result ?: lock->cll_error);
+ RETURN(result);
}
EXPORT_SYMBOL(cl_wait_try);
cl_lock_mutex_get(env, lock);
LINVRNT(cl_lock_invariant(env, lock));
- LASSERT(lock->cll_state == CLS_ENQUEUED || lock->cll_state == CLS_HELD);
+ LASSERTF(lock->cll_state == CLS_ENQUEUED || lock->cll_state == CLS_HELD,
+ "Wrong state %d \n", lock->cll_state);
LASSERT(lock->cll_holds > 0);
do {
break;
} while (1);
if (result < 0) {
- cl_lock_user_del(env, lock);
- if (result != -EINTR)
- cl_lock_error(env, lock, result);
+ cl_unuse_try(env, lock);
cl_lock_lockdep_release(env, lock);
}
+ cl_lock_trace(D_DLMTRACE, env, "wait lock", lock);
cl_lock_mutex_put(env, lock);
LASSERT(ergo(result == 0, lock->cll_state == CLS_HELD));
RETURN(result);
LINVRNT(cl_lock_invariant(env, lock));
pound = 0;
- list_for_each_entry_reverse(slice, &lock->cll_layers, cls_linkage) {
+ list_for_each_entry_reverse(slice, &lock->cll_layers, cls_linkage) {
if (slice->cls_ops->clo_weigh != NULL) {
ounce = slice->cls_ops->clo_weigh(env, slice);
pound += ounce;
int result;
ENTRY;
+ cl_lock_trace(D_DLMTRACE, env, "modify lock", lock);
/* don't allow object to change */
LASSERT(obj == desc->cld_obj);
LINVRNT(cl_lock_is_mutexed(lock));
LINVRNT(cl_lock_invariant(env, lock));
- list_for_each_entry_reverse(slice, &lock->cll_layers, cls_linkage) {
+ list_for_each_entry_reverse(slice, &lock->cll_layers, cls_linkage) {
if (slice->cls_ops->clo_modify != NULL) {
result = slice->cls_ops->clo_modify(env, slice, desc);
if (result != 0)
* now. If locks were indexed according to their extent and/or mode,
* that index would have to be updated here.
*/
- spin_lock(&hdr->coh_lock_guard);
- lock->cll_descr = *desc;
- spin_unlock(&hdr->coh_lock_guard);
- RETURN(0);
+ spin_lock(&hdr->coh_lock_guard);
+ lock->cll_descr = *desc;
+ spin_unlock(&hdr->coh_lock_guard);
+ RETURN(0);
}
EXPORT_SYMBOL(cl_lock_modify);
LINVRNT(cl_lock_is_mutexed(origin));
LINVRNT(cl_lock_invariant(env, origin));
- CFS_INIT_LIST_HEAD(&closure->clc_list);
+ INIT_LIST_HEAD(&closure->clc_list);
closure->clc_origin = origin;
closure->clc_wait = wait;
closure->clc_nr = 0;
result = cl_lock_enclosure(env, lock, closure);
if (result == 0) {
- list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
+ list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
if (slice->cls_ops->clo_closure != NULL) {
result = slice->cls_ops->clo_closure(env, slice,
closure);
int cl_lock_enclosure(const struct lu_env *env, struct cl_lock *lock,
struct cl_lock_closure *closure)
{
- int result;
+ int result = 0;
ENTRY;
+ cl_lock_trace(D_DLMTRACE, env, "enclosure lock", lock);
if (!cl_lock_mutex_try(env, lock)) {
/*
* If lock->cll_inclosure is not empty, lock is already in
* this closure.
*/
- if (list_empty(&lock->cll_inclosure)) {
+ if (list_empty(&lock->cll_inclosure)) {
cl_lock_get_trust(lock);
lu_ref_add(&lock->cll_reference, "closure", closure);
- list_add(&lock->cll_inclosure, &closure->clc_list);
+ list_add(&lock->cll_inclosure, &closure->clc_list);
closure->clc_nr++;
} else
cl_lock_mutex_put(env, lock);
void cl_lock_disclosure(const struct lu_env *env,
struct cl_lock_closure *closure)
{
- struct cl_lock *scan;
- struct cl_lock *temp;
+ struct cl_lock *scan;
+ struct cl_lock *temp;
- list_for_each_entry_safe(scan, temp, &closure->clc_list, cll_inclosure){
- list_del_init(&scan->cll_inclosure);
- cl_lock_mutex_put(env, scan);
- lu_ref_del(&scan->cll_reference, "closure", closure);
- cl_lock_put(env, scan);
- closure->clc_nr--;
- }
- LASSERT(closure->clc_nr == 0);
+ cl_lock_trace(D_DLMTRACE, env, "disclosure lock", closure->clc_origin);
+ list_for_each_entry_safe(scan, temp, &closure->clc_list,
+ cll_inclosure){
+ list_del_init(&scan->cll_inclosure);
+ cl_lock_mutex_put(env, scan);
+ lu_ref_del(&scan->cll_reference, "closure", closure);
+ cl_lock_put(env, scan);
+ closure->clc_nr--;
+ }
+ LASSERT(closure->clc_nr == 0);
}
EXPORT_SYMBOL(cl_lock_disclosure);
void cl_lock_closure_fini(struct cl_lock_closure *closure)
{
LASSERT(closure->clc_nr == 0);
- LASSERT(list_empty(&closure->clc_list));
+ LASSERT(list_empty(&closure->clc_list));
}
EXPORT_SYMBOL(cl_lock_closure_fini);
* cl_lock_put() to finish it.
*
* \pre atomic_read(&lock->cll_ref) > 0
+ * \pre ergo(cl_lock_nesting(lock) == CNL_TOP,
+ * cl_lock_nr_mutexed(env) == 1)
+ * [i.e., if a top-lock is deleted, mutices of no other locks can be
+ * held, as deletion of sub-locks might require releasing a top-lock
+ * mutex]
*
* \see cl_lock_operations::clo_delete()
* \see cl_lock::cll_holds
{
LINVRNT(cl_lock_is_mutexed(lock));
LINVRNT(cl_lock_invariant(env, lock));
+ LASSERT(ergo(cl_lock_nesting(lock) == CNL_TOP,
+ cl_lock_nr_mutexed(env) == 1));
ENTRY;
+ cl_lock_trace(D_DLMTRACE, env, "delete lock", lock);
if (lock->cll_holds == 0)
cl_lock_delete0(env, lock);
else
ENTRY;
if (lock->cll_error == 0 && error != 0) {
+ cl_lock_trace(D_DLMTRACE, env, "set lock error", lock);
lock->cll_error = error;
cl_lock_signal(env, lock);
cl_lock_cancel(env, lock);
{
LINVRNT(cl_lock_is_mutexed(lock));
LINVRNT(cl_lock_invariant(env, lock));
+
ENTRY;
+ cl_lock_trace(D_DLMTRACE, env, "cancel lock", lock);
if (lock->cll_holds == 0)
cl_lock_cancel0(env, lock);
else
EXPORT_SYMBOL(cl_lock_cancel);
/**
- * Finds an existing lock covering given page and optionally different from a
+ * Finds an existing lock covering given index and optionally different from a
* given \a except lock.
*/
-struct cl_lock *cl_lock_at_page(const struct lu_env *env, struct cl_object *obj,
- struct cl_page *page, struct cl_lock *except,
- int pending, int canceld)
+struct cl_lock *cl_lock_at_pgoff(const struct lu_env *env,
+ struct cl_object *obj, pgoff_t index,
+ struct cl_lock *except,
+ int pending, int canceld)
{
struct cl_object_header *head;
struct cl_lock *scan;
need->cld_mode = CLM_READ; /* CLM_READ matches both READ & WRITE, but
* not PHANTOM */
- need->cld_start = need->cld_end = page->cp_index;
+ need->cld_start = need->cld_end = index;
+ need->cld_enq_flags = 0;
- spin_lock(&head->coh_lock_guard);
- list_for_each_entry(scan, &head->coh_locks, cll_linkage) {
+ spin_lock(&head->coh_lock_guard);
+ /* It is fine to match any group lock since there could be only one
+ * with a uniq gid and it conflicts with all other lock modes too */
+ list_for_each_entry(scan, &head->coh_locks, cll_linkage) {
if (scan != except &&
- cl_lock_ext_match(&scan->cll_descr, need) &&
+ (scan->cll_descr.cld_mode == CLM_GROUP ||
+ cl_lock_ext_match(&scan->cll_descr, need)) &&
+ scan->cll_state >= CLS_HELD &&
scan->cll_state < CLS_FREEING &&
/*
* This check is racy as the lock can be canceled right
break;
}
}
- spin_unlock(&head->coh_lock_guard);
- RETURN(lock);
+ spin_unlock(&head->coh_lock_guard);
+ RETURN(lock);
}
-EXPORT_SYMBOL(cl_lock_at_page);
-
-/**
- * Returns a list of pages protected (only) by a given lock.
- *
- * Scans an extent of page radix tree, corresponding to the \a lock and queues
- * all pages that are not protected by locks other than \a lock into \a queue.
- */
-void cl_lock_page_list_fixup(const struct lu_env *env,
- struct cl_io *io, struct cl_lock *lock,
- struct cl_page_list *queue)
-{
- struct cl_page *page;
- struct cl_page *temp;
- struct cl_page_list *plist = &cl_env_info(env)->clt_list;
-
- LINVRNT(cl_lock_invariant(env, lock));
- ENTRY;
-
- /* Now, we have a list of cl_pages under the \a lock, we need
- * to check if some of pages are covered by other ldlm lock.
- * If this is the case, they aren't needed to be written out this time.
- *
- * For example, we have A:[0,200] & B:[100,300] PW locks on client, now
- * the latter is to be canceled, this means other client is
- * reading/writing [200,300] since A won't canceled. Actually
- * we just need to write the pages covered by [200,300]. This is safe,
- * since [100,200] is also protected lock A.
- */
-
- cl_page_list_init(plist);
- cl_page_list_for_each_safe(page, temp, queue) {
- pgoff_t idx = page->cp_index;
- struct cl_lock *found;
- struct cl_lock_descr *descr;
-
- /* The algorithm counts on the index-ascending page index. */
- LASSERT(ergo(&temp->cp_batch != &queue->pl_pages,
- page->cp_index < temp->cp_index));
-
- found = cl_lock_at_page(env, lock->cll_descr.cld_obj,
- page, lock, 0, 0);
- if (found == NULL)
- continue;
-
- descr = &found->cll_descr;
- list_for_each_entry_safe_from(page, temp, &queue->pl_pages,
- cp_batch) {
- idx = page->cp_index;
- if (descr->cld_start > idx || descr->cld_end < idx)
- break;
- cl_page_list_move(plist, queue, page);
- }
- cl_lock_put(env, found);
- }
-
- /* The pages in plist are covered by other locks, don't handle them
- * this time.
- */
- if (io != NULL)
- cl_page_list_disown(env, io, plist);
- cl_page_list_fini(env, plist);
- EXIT;
-}
-EXPORT_SYMBOL(cl_lock_page_list_fixup);
-
-/**
- * Invalidate pages protected by the given lock, sending them out to the
- * server first, if necessary.
- *
- * This function does the following:
- *
- * - collects a list of pages to be invalidated,
- *
- * - unmaps them from the user virtual memory,
- *
- * - sends dirty pages to the server,
- *
- * - waits for transfer completion,
- *
- * - discards pages, and throws them out of memory.
- *
- * If \a discard is set, pages are discarded without sending them to the
- * server.
- *
- * If error happens on any step, the process continues anyway (the reasoning
- * behind this being that lock cancellation cannot be delayed indefinitely).
- */
-int cl_lock_page_out(const struct lu_env *env, struct cl_lock *lock,
- int discard)
-{
- struct cl_thread_info *info = cl_env_info(env);
- struct cl_io *io = &info->clt_io;
- struct cl_2queue *queue = &info->clt_queue;
- struct cl_lock_descr *descr = &lock->cll_descr;
- int result;
- int rc0;
- int rc1;
-
- LINVRNT(cl_lock_invariant(env, lock));
- ENTRY;
-
- io->ci_obj = cl_object_top(descr->cld_obj);
- result = cl_io_init(env, io, CIT_MISC, io->ci_obj);
- if (result == 0) {
-
- cl_2queue_init(queue);
- cl_page_gang_lookup(env, descr->cld_obj, io, descr->cld_start,
- descr->cld_end, &queue->c2_qin);
- if (queue->c2_qin.pl_nr > 0) {
- result = cl_page_list_unmap(env, io, &queue->c2_qin);
- if (!discard) {
- rc0 = cl_io_submit_rw(env, io,
- CRT_WRITE, queue);
- rc1 = cl_page_list_own(env, io,
- &queue->c2_qout);
- result = result ?: rc0 ?: rc1;
- }
- cl_lock_page_list_fixup(env, io, lock, &queue->c2_qout);
- cl_2queue_discard(env, io, queue);
- cl_2queue_disown(env, io, queue);
- }
- cl_2queue_fini(env, queue);
- }
- cl_io_fini(env, io);
- RETURN(result);
-}
-EXPORT_SYMBOL(cl_lock_page_out);
+EXPORT_SYMBOL(cl_lock_at_pgoff);
/**
* Eliminate all locks for a given object.
*/
void cl_locks_prune(const struct lu_env *env, struct cl_object *obj, int cancel)
{
- struct cl_object_header *head;
- struct cl_lock *lock;
-
- ENTRY;
- head = cl_object_header(obj);
- /*
- * If locks are destroyed without cancellation, all pages must be
- * already destroyed (as otherwise they will be left unprotected).
- */
- LASSERT(ergo(!cancel,
- head->coh_tree.rnode == NULL && head->coh_pages == 0));
-
- spin_lock(&head->coh_lock_guard);
- while (!list_empty(&head->coh_locks)) {
- lock = container_of(head->coh_locks.next,
- struct cl_lock, cll_linkage);
- cl_lock_get_trust(lock);
- spin_unlock(&head->coh_lock_guard);
- lu_ref_add(&lock->cll_reference, "prune", cfs_current());
- cl_lock_mutex_get(env, lock);
- if (lock->cll_state < CLS_FREEING) {
- LASSERT(lock->cll_holds == 0);
- LASSERT(lock->cll_users == 0);
- if (cancel)
- cl_lock_cancel(env, lock);
- cl_lock_delete(env, lock);
- }
- cl_lock_mutex_put(env, lock);
- lu_ref_del(&lock->cll_reference, "prune", cfs_current());
- cl_lock_put(env, lock);
- spin_lock(&head->coh_lock_guard);
- }
- spin_unlock(&head->coh_lock_guard);
- EXIT;
+ struct cl_object_header *head;
+ struct cl_lock *lock;
+
+ ENTRY;
+ head = cl_object_header(obj);
+
+ spin_lock(&head->coh_lock_guard);
+ while (!list_empty(&head->coh_locks)) {
+ lock = container_of(head->coh_locks.next,
+ struct cl_lock, cll_linkage);
+ cl_lock_get_trust(lock);
+ spin_unlock(&head->coh_lock_guard);
+ lu_ref_add(&lock->cll_reference, "prune", current);
+
+again:
+ cl_lock_mutex_get(env, lock);
+ if (lock->cll_state < CLS_FREEING) {
+ LASSERT(lock->cll_users <= 1);
+ if (unlikely(lock->cll_users == 1)) {
+ struct l_wait_info lwi = { 0 };
+
+ cl_lock_mutex_put(env, lock);
+ l_wait_event(lock->cll_wq,
+ lock->cll_users == 0,
+ &lwi);
+ goto again;
+ }
+
+ if (cancel)
+ cl_lock_cancel(env, lock);
+ cl_lock_delete(env, lock);
+ }
+ cl_lock_mutex_put(env, lock);
+ lu_ref_del(&lock->cll_reference, "prune", current);
+ cl_lock_put(env, lock);
+ spin_lock(&head->coh_lock_guard);
+ }
+ spin_unlock(&head->coh_lock_guard);
+ EXIT;
}
EXPORT_SYMBOL(cl_locks_prune);
-/**
- * Returns true if \a addr is an address of an allocated cl_lock. Used in
- * assertions. This check is optimistically imprecise, i.e., it occasionally
- * returns true for the incorrect addresses, but if it returns false, then the
- * address is guaranteed to be incorrect. (Should be named cl_lockp().)
- *
- * \see cl_is_page()
- */
-int cl_is_lock(const void *addr)
-{
- return cfs_mem_is_in_cache(addr, cl_lock_kmem);
-}
-EXPORT_SYMBOL(cl_is_lock);
-
static struct cl_lock *cl_lock_hold_mutex(const struct lu_env *env,
const struct cl_io *io,
const struct cl_lock_descr *need,
if (IS_ERR(lock))
break;
cl_lock_mutex_get(env, lock);
- if (lock->cll_state < CLS_FREEING) {
+ if (lock->cll_state < CLS_FREEING &&
+ !(lock->cll_flags & CLF_CANCELLED)) {
cl_lock_hold_mod(env, lock, +1);
lu_ref_add(&lock->cll_holders, scope, source);
lu_ref_add(&lock->cll_reference, scope, source);
*/
struct cl_lock *cl_lock_request(const struct lu_env *env, struct cl_io *io,
const struct cl_lock_descr *need,
- __u32 enqflags,
const char *scope, const void *source)
{
struct cl_lock *lock;
- const struct lu_fid *fid;
int rc;
- int iter;
- int warn;
+ __u32 enqflags = need->cld_enq_flags;
ENTRY;
- fid = lu_object_fid(&io->ci_obj->co_lu);
- iter = 0;
do {
- warn = iter >= 16 && IS_PO2(iter);
- CDEBUG(warn ? D_WARNING : D_DLMTRACE,
- DDESCR"@"DFID" %i %08x `%s'\n",
- PDESCR(need), PFID(fid), iter, enqflags, scope);
lock = cl_lock_hold_mutex(env, io, need, scope, source);
- if (!IS_ERR(lock)) {
- rc = cl_enqueue_locked(env, lock, io, enqflags);
- if (rc == 0) {
- if (cl_lock_fits_into(env, lock, need, io)) {
+ if (IS_ERR(lock))
+ break;
+
+ rc = cl_enqueue_locked(env, lock, io, enqflags);
+ if (rc == 0) {
+ if (cl_lock_fits_into(env, lock, need, io)) {
+ if (!(enqflags & CEF_AGL)) {
cl_lock_mutex_put(env, lock);
- cl_lock_lockdep_acquire(env,
- lock, enqflags);
+ cl_lock_lockdep_acquire(env, lock,
+ enqflags);
break;
- } else if (warn)
- CL_LOCK_DEBUG(D_WARNING, env, lock,
- "got\n");
- cl_unuse_locked(env, lock);
+ }
+ rc = 1;
}
- cl_lock_hold_release(env, lock, scope, source);
- cl_lock_mutex_put(env, lock);
- lu_ref_del(&lock->cll_reference, scope, source);
- cl_lock_put(env, lock);
+ cl_unuse_locked(env, lock);
+ }
+ cl_lock_trace(D_DLMTRACE, env,
+ rc <= 0 ? "enqueue failed" : "agl succeed", lock);
+ cl_lock_hold_release(env, lock, scope, source);
+ cl_lock_mutex_put(env, lock);
+ lu_ref_del(&lock->cll_reference, scope, source);
+ cl_lock_put(env, lock);
+ if (rc > 0) {
+ LASSERT(enqflags & CEF_AGL);
+ lock = NULL;
+ } else if (rc != 0) {
lock = ERR_PTR(rc);
- } else
- rc = PTR_ERR(lock);
- iter++;
+ }
} while (rc == 0);
RETURN(lock);
}
LASSERT(lock->cll_state != CLS_FREEING);
ENTRY;
- cl_lock_hold_mod(env, lock, +1);
cl_lock_get(lock);
+ cl_lock_hold_mod(env, lock, +1);
lu_ref_add(&lock->cll_holders, scope, source);
lu_ref_add(&lock->cll_reference, scope, source);
EXIT;
{
LINVRNT(cl_lock_invariant(env, lock));
ENTRY;
+ cl_lock_trace(D_DLMTRACE, env, "release lock", lock);
cl_lock_mutex_get(env, lock);
cl_lock_hold_release(env, lock, scope, source);
cl_lock_mutex_put(env, lock);
}
EXPORT_SYMBOL(cl_lock_user_add);
-int cl_lock_user_del(const struct lu_env *env, struct cl_lock *lock)
+void cl_lock_user_del(const struct lu_env *env, struct cl_lock *lock)
{
- LINVRNT(cl_lock_is_mutexed(lock));
- LINVRNT(cl_lock_invariant(env, lock));
- LASSERT(lock->cll_users > 0);
+ LINVRNT(cl_lock_is_mutexed(lock));
+ LINVRNT(cl_lock_invariant(env, lock));
+ LASSERT(lock->cll_users > 0);
- ENTRY;
- cl_lock_used_mod(env, lock, -1);
- RETURN(lock->cll_users == 0);
+ ENTRY;
+ cl_lock_used_mod(env, lock, -1);
+ if (lock->cll_users == 0)
+ wake_up_all(&lock->cll_wq);
+ EXIT;
}
EXPORT_SYMBOL(cl_lock_user_del);
-/**
- * Check if two lock's mode are compatible.
- *
- * This returns true iff en-queuing \a lock2 won't cause cancellation of \a
- * lock1 even when these locks overlap.
- */
-int cl_lock_compatible(const struct cl_lock *lock1, const struct cl_lock *lock2)
-{
- enum cl_lock_mode mode1;
- enum cl_lock_mode mode2;
-
- ENTRY;
- mode1 = lock1->cll_descr.cld_mode;
- mode2 = lock2->cll_descr.cld_mode;
- RETURN(mode2 == CLM_PHANTOM ||
- (mode1 == CLM_READ && mode2 == CLM_READ));
-}
-EXPORT_SYMBOL(cl_lock_compatible);
-
const char *cl_lock_mode_name(const enum cl_lock_mode mode)
{
static const char *names[] = {
- [CLM_PHANTOM] = "PHANTOM",
- [CLM_READ] = "READ",
- [CLM_WRITE] = "WRITE"
+ [CLM_PHANTOM] = "P",
+ [CLM_READ] = "R",
+ [CLM_WRITE] = "W",
+ [CLM_GROUP] = "G"
};
- if (0 <= mode && mode < ARRAY_SIZE(names))
- return names[mode];
- else
- return "UNKNW";
+ CLASSERT(CLM_MAX == ARRAY_SIZE(names));
+ return names[mode];
}
EXPORT_SYMBOL(cl_lock_mode_name);
{
const struct cl_lock_slice *slice;
(*printer)(env, cookie, "lock@%p[%d %d %d %d %d %08lx] ",
- lock, atomic_read(&lock->cll_ref),
+ lock, atomic_read(&lock->cll_ref),
lock->cll_state, lock->cll_error, lock->cll_holds,
lock->cll_users, lock->cll_flags);
cl_lock_descr_print(env, cookie, printer, &lock->cll_descr);
(*printer)(env, cookie, " {\n");
- list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
+ list_for_each_entry(slice, &lock->cll_layers, cls_linkage) {
(*printer)(env, cookie, " %s@%p: ",
slice->cls_obj->co_lu.lo_dev->ld_type->ldt_name,
slice);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff