/**
* Converts lock policy from local format to on the wire lock_desc format
*/
-void ldlm_convert_policy_to_wire(ldlm_type_t type,
- const ldlm_policy_data_t *lpolicy,
- ldlm_wire_policy_data_t *wpolicy)
+void ldlm_convert_policy_to_wire(enum ldlm_type type,
+ const union ldlm_policy_data *lpolicy,
+ union ldlm_wire_policy_data *wpolicy)
{
- ldlm_policy_local_to_wire_t convert;
+ ldlm_policy_local_to_wire_t convert;
- convert = ldlm_policy_local_to_wire[type - LDLM_MIN_TYPE];
+ convert = ldlm_policy_local_to_wire[type - LDLM_MIN_TYPE];
- convert(lpolicy, wpolicy);
+ convert(lpolicy, wpolicy);
}
/**
* Converts lock policy from on the wire lock_desc format to local format
*/
-void ldlm_convert_policy_to_local(struct obd_export *exp, ldlm_type_t type,
- const ldlm_wire_policy_data_t *wpolicy,
- ldlm_policy_data_t *lpolicy)
+void ldlm_convert_policy_to_local(struct obd_export *exp, enum ldlm_type type,
+ const union ldlm_wire_policy_data *wpolicy,
+ union ldlm_policy_data *lpolicy)
{
ldlm_policy_wire_to_local_t convert;
res = lock->l_resource;
LASSERT(ldlm_is_destroyed(lock));
+ LASSERT(list_empty(&lock->l_exp_list));
LASSERT(list_empty(&lock->l_res_link));
LASSERT(list_empty(&lock->l_pending_chain));
/**
* Removes LDLM lock \a lock from LRU. Obtains the LRU lock first.
+ *
+ * If \a last_use is non-zero, it will remove the lock from LRU only if
+ * it matches lock's l_last_used.
+ *
+ * \retval 0 if \a last_use is set, the lock is not in LRU list or \a last_use
+ * doesn't match lock's l_last_used;
+ * otherwise, the lock hasn't been in the LRU list.
+ * \retval 1 the lock was in LRU list and removed.
*/
-int ldlm_lock_remove_from_lru(struct ldlm_lock *lock)
+int ldlm_lock_remove_from_lru_check(struct ldlm_lock *lock, cfs_time_t last_use)
{
struct ldlm_namespace *ns = ldlm_lock_to_ns(lock);
- int rc;
+ int rc = 0;
ENTRY;
if (ldlm_is_ns_srv(lock)) {
}
spin_lock(&ns->ns_lock);
- rc = ldlm_lock_remove_from_lru_nolock(lock);
+ if (last_use == 0 || last_use == lock->l_last_used)
+ rc = ldlm_lock_remove_from_lru_nolock(lock);
spin_unlock(&ns->ns_lock);
- EXIT;
- return rc;
+
+ RETURN(rc);
}
/**
* r/w reference type is determined by \a mode
* Calls ldlm_lock_addref_internal.
*/
-void ldlm_lock_addref(struct lustre_handle *lockh, __u32 mode)
+void ldlm_lock_addref(struct lustre_handle *lockh, enum ldlm_mode mode)
{
struct ldlm_lock *lock;
* Removes lock from LRU if it is there.
* Assumes the LDLM lock is already locked.
*/
-void ldlm_lock_addref_internal_nolock(struct ldlm_lock *lock, __u32 mode)
+void ldlm_lock_addref_internal_nolock(struct ldlm_lock *lock,
+ enum ldlm_mode mode)
{
ldlm_lock_remove_from_lru(lock);
if (mode & (LCK_NL | LCK_CR | LCK_PR)) {
*
* \retval -EAGAIN lock is being canceled.
*/
-int ldlm_lock_addref_try(struct lustre_handle *lockh, __u32 mode)
+int ldlm_lock_addref_try(struct lustre_handle *lockh, enum ldlm_mode mode)
{
struct ldlm_lock *lock;
int result;
* Locks LDLM lock and calls ldlm_lock_addref_internal_nolock to do the work.
* Only called for local locks.
*/
-void ldlm_lock_addref_internal(struct ldlm_lock *lock, __u32 mode)
+void ldlm_lock_addref_internal(struct ldlm_lock *lock, enum ldlm_mode mode)
{
- lock_res_and_lock(lock);
- ldlm_lock_addref_internal_nolock(lock, mode);
- unlock_res_and_lock(lock);
+ lock_res_and_lock(lock);
+ ldlm_lock_addref_internal_nolock(lock, mode);
+ unlock_res_and_lock(lock);
}
/**
* Does NOT add lock to LRU if no r/w references left to accomodate flock locks
* that cannot be placed in LRU.
*/
-void ldlm_lock_decref_internal_nolock(struct ldlm_lock *lock, __u32 mode)
+void ldlm_lock_decref_internal_nolock(struct ldlm_lock *lock,
+ enum ldlm_mode mode)
{
LDLM_DEBUG(lock, "ldlm_lock_decref(%s)", ldlm_lockname[mode]);
if (mode & (LCK_NL | LCK_CR | LCK_PR)) {
* on the namespace.
* For blocked LDLM locks if r/w count drops to zero, blocking_ast is called.
*/
-void ldlm_lock_decref_internal(struct ldlm_lock *lock, __u32 mode)
+void ldlm_lock_decref_internal(struct ldlm_lock *lock, enum ldlm_mode mode)
{
struct ldlm_namespace *ns;
ENTRY;
/**
* Decrease reader/writer refcount for LDLM lock with handle \a lockh
*/
-void ldlm_lock_decref(struct lustre_handle *lockh, __u32 mode)
+void ldlm_lock_decref(struct lustre_handle *lockh, enum ldlm_mode mode)
{
struct ldlm_lock *lock = __ldlm_handle2lock(lockh, 0);
LASSERTF(lock != NULL, "Non-existing lock: "LPX64"\n", lockh->cookie);
* drops to zero instead of putting into LRU.
*
*/
-void ldlm_lock_decref_and_cancel(struct lustre_handle *lockh, __u32 mode)
+void ldlm_lock_decref_and_cancel(struct lustre_handle *lockh,
+ enum ldlm_mode mode)
{
struct ldlm_lock *lock = __ldlm_handle2lock(lockh, 0);
ENTRY;
else
ldlm_resource_add_lock(res, &res->lr_granted, lock);
- if (lock->l_granted_mode < res->lr_most_restr)
- res->lr_most_restr = lock->l_granted_mode;
-
ldlm_pool_add(&ldlm_res_to_ns(res)->ns_pool, lock);
EXIT;
}
* Describe the overlap between two locks. itree_overlap_cb data.
*/
struct lock_match_data {
- struct ldlm_lock *lmd_old;
- struct ldlm_lock *lmd_lock;
- ldlm_mode_t *lmd_mode;
- ldlm_policy_data_t *lmd_policy;
- __u64 lmd_flags;
- int lmd_unref;
+ struct ldlm_lock *lmd_old;
+ struct ldlm_lock *lmd_lock;
+ enum ldlm_mode *lmd_mode;
+ union ldlm_policy_data *lmd_policy;
+ __u64 lmd_flags;
+ int lmd_unref;
};
/**
*/
static int lock_matches(struct ldlm_lock *lock, struct lock_match_data *data)
{
- ldlm_policy_data_t *lpol = &lock->l_policy_data;
- ldlm_mode_t match;
+ union ldlm_policy_data *lpol = &lock->l_policy_data;
+ enum ldlm_mode match;
if (lock == data->lmd_old)
return INTERVAL_ITER_STOP;
if (!data->lmd_unref && LDLM_HAVE_MASK(lock, GONE))
return INTERVAL_ITER_CONT;
- if ((data->lmd_flags & LDLM_FL_LOCAL_ONLY) &&
- !ldlm_is_local(lock))
+ if (!equi(data->lmd_flags & LDLM_FL_LOCAL_ONLY, ldlm_is_local(lock)))
return INTERVAL_ITER_CONT;
if (data->lmd_flags & LDLM_FL_TEST_LOCK) {
* keep caller code unchanged), the context failure will be discovered by
* caller sometime later.
*/
-ldlm_mode_t ldlm_lock_match(struct ldlm_namespace *ns, __u64 flags,
- const struct ldlm_res_id *res_id, ldlm_type_t type,
- ldlm_policy_data_t *policy, ldlm_mode_t mode,
- struct lustre_handle *lockh, int unref)
+enum ldlm_mode ldlm_lock_match(struct ldlm_namespace *ns, __u64 flags,
+ const struct ldlm_res_id *res_id,
+ enum ldlm_type type,
+ union ldlm_policy_data *policy,
+ enum ldlm_mode mode,
+ struct lustre_handle *lockh, int unref)
{
struct lock_match_data data = {
.lmd_old = NULL,
}
EXPORT_SYMBOL(ldlm_lock_match);
-ldlm_mode_t ldlm_revalidate_lock_handle(struct lustre_handle *lockh,
- __u64 *bits)
+enum ldlm_mode ldlm_revalidate_lock_handle(struct lustre_handle *lockh,
+ __u64 *bits)
{
- struct ldlm_lock *lock;
- ldlm_mode_t mode = 0;
- ENTRY;
+ struct ldlm_lock *lock;
+ enum ldlm_mode mode = 0;
+ ENTRY;
- lock = ldlm_handle2lock(lockh);
- if (lock != NULL) {
- lock_res_and_lock(lock);
+ lock = ldlm_handle2lock(lockh);
+ if (lock != NULL) {
+ lock_res_and_lock(lock);
if (LDLM_HAVE_MASK(lock, GONE))
- GOTO(out, mode);
+ GOTO(out, mode);
if (ldlm_is_cbpending(lock) &&
lock->l_readers == 0 && lock->l_writers == 0)
*/
struct ldlm_lock *ldlm_lock_create(struct ldlm_namespace *ns,
const struct ldlm_res_id *res_id,
- ldlm_type_t type,
- ldlm_mode_t mode,
+ enum ldlm_type type,
+ enum ldlm_mode mode,
const struct ldlm_callback_suite *cbs,
void *data, __u32 lvb_len,
enum lvb_type lvb_type)
* set, skip all the enqueueing and delegate lock processing to intent policy
* function.
*/
-ldlm_error_t ldlm_lock_enqueue(struct ldlm_namespace *ns,
- struct ldlm_lock **lockp,
- void *cookie, __u64 *flags)
+enum ldlm_error ldlm_lock_enqueue(struct ldlm_namespace *ns,
+ struct ldlm_lock **lockp,
+ void *cookie, __u64 *flags)
{
- struct ldlm_lock *lock = *lockp;
- struct ldlm_resource *res = lock->l_resource;
- int local = ns_is_client(ldlm_res_to_ns(res));
+ struct ldlm_lock *lock = *lockp;
+ struct ldlm_resource *res = lock->l_resource;
+ int local = ns_is_client(ldlm_res_to_ns(res));
#ifdef HAVE_SERVER_SUPPORT
- ldlm_processing_policy policy;
+ ldlm_processing_policy policy;
#endif
- ldlm_error_t rc = ELDLM_OK;
- struct ldlm_interval *node = NULL;
- ENTRY;
+ enum ldlm_error rc = ELDLM_OK;
+ struct ldlm_interval *node = NULL;
+ ENTRY;
/* policies are not executed on the client or during replay */
if ((*flags & (LDLM_FL_HAS_INTENT|LDLM_FL_REPLAY)) == LDLM_FL_HAS_INTENT
struct list_head *work_list)
{
struct list_head *tmp, *pos;
- ldlm_processing_policy policy;
+ ldlm_processing_policy policy;
__u64 flags;
- int rc = LDLM_ITER_CONTINUE;
- ldlm_error_t err;
- ENTRY;
+ int rc = LDLM_ITER_CONTINUE;
+ enum ldlm_error err;
+ ENTRY;
- check_res_locked(res);
+ check_res_locked(res);
- policy = ldlm_processing_policy_table[res->lr_type];
- LASSERT(policy);
+ policy = ldlm_processing_policy_table[res->lr_type];
+ LASSERT(policy);
list_for_each_safe(tmp, pos, queue) {
- struct ldlm_lock *pending;
+ struct ldlm_lock *pending;
+
pending = list_entry(tmp, struct ldlm_lock, l_res_link);
CDEBUG(D_INFO, "Reprocessing lock %p\n", pending);
*/
void ldlm_reprocess_all_ns(struct ldlm_namespace *ns)
{
- ENTRY;
+ ENTRY;
- if (ns != NULL) {
- cfs_hash_for_each_nolock(ns->ns_rs_hash,
- ldlm_reprocess_res, NULL);
- }
- EXIT;
+ if (ns != NULL) {
+ cfs_hash_for_each_nolock(ns->ns_rs_hash,
+ ldlm_reprocess_res, NULL, 0);
+ }
+ EXIT;
}
/**
int ecl_loop;
};
+static void ldlm_cancel_lock_for_export(struct obd_export *exp,
+ struct ldlm_lock *lock,
+ struct export_cl_data *ecl)
+{
+ struct ldlm_resource *res;
+
+ res = ldlm_resource_getref(lock->l_resource);
+
+ ldlm_res_lvbo_update(res, NULL, 1);
+ ldlm_lock_cancel(lock);
+ if (!exp->exp_obd->obd_stopping)
+ ldlm_reprocess_all(res);
+ ldlm_resource_putref(res);
+
+ ecl->ecl_loop++;
+ if ((ecl->ecl_loop & -ecl->ecl_loop) == ecl->ecl_loop) {
+ CDEBUG(D_INFO, "Export %p, %d locks cancelled.\n",
+ exp, ecl->ecl_loop);
+ }
+}
+
/**
- * Iterator function for ldlm_cancel_locks_for_export.
+ * Iterator function for ldlm_export_cancel_locks.
* Cancels passed locks.
*/
static int
{
struct export_cl_data *ecl = (struct export_cl_data *)data;
struct obd_export *exp = ecl->ecl_exp;
- struct ldlm_lock *lock = cfs_hash_object(hs, hnode);
- struct ldlm_resource *res;
+ struct ldlm_lock *lock = cfs_hash_object(hs, hnode);
- res = ldlm_resource_getref(lock->l_resource);
- LDLM_LOCK_GET(lock);
+ LDLM_LOCK_GET(lock);
+ ldlm_cancel_lock_for_export(exp, lock, ecl);
+ LDLM_LOCK_RELEASE(lock);
- LDLM_DEBUG(lock, "export %p", exp);
- ldlm_res_lvbo_update(res, NULL, 1);
- ldlm_lock_cancel(lock);
- ldlm_reprocess_all(res);
- ldlm_resource_putref(res);
- LDLM_LOCK_RELEASE(lock);
+ return 0;
+}
- ecl->ecl_loop++;
- if ((ecl->ecl_loop & -ecl->ecl_loop) == ecl->ecl_loop) {
- CDEBUG(D_INFO,
- "Cancel lock %p for export %p (loop %d), still have "
- "%d locks left on hash table.\n",
- lock, exp, ecl->ecl_loop,
- atomic_read(&hs->hs_count));
+/**
+ * Cancel all blocked locks for given export.
+ *
+ * Typically called on client disconnection/eviction
+ */
+int ldlm_export_cancel_blocked_locks(struct obd_export *exp)
+{
+ struct export_cl_data ecl = {
+ .ecl_exp = exp,
+ .ecl_loop = 0,
+ };
+
+ while (!list_empty(&exp->exp_bl_list)) {
+ struct ldlm_lock *lock;
+
+ spin_lock_bh(&exp->exp_bl_list_lock);
+ if (!list_empty(&exp->exp_bl_list)) {
+ lock = list_entry(exp->exp_bl_list.next,
+ struct ldlm_lock, l_exp_list);
+ LDLM_LOCK_GET(lock);
+ list_del_init(&lock->l_exp_list);
+ } else {
+ lock = NULL;
+ }
+ spin_unlock_bh(&exp->exp_bl_list_lock);
+
+ if (lock == NULL)
+ break;
+
+ ldlm_cancel_lock_for_export(exp, lock, &ecl);
+ LDLM_LOCK_RELEASE(lock);
}
- return 0;
+ CDEBUG(D_DLMTRACE, "Export %p, canceled %d locks, "
+ "left on hash table %d.\n", exp, ecl.ecl_loop,
+ atomic_read(&exp->exp_lock_hash->hs_count));
+
+ return ecl.ecl_loop;
}
/**
* Cancel all locks for given export.
*
- * Typically called on client disconnection/eviction
+ * Typically called after client disconnection/eviction
*/
-void ldlm_cancel_locks_for_export(struct obd_export *exp)
+int ldlm_export_cancel_locks(struct obd_export *exp)
{
struct export_cl_data ecl = {
.ecl_exp = exp,
cfs_hash_for_each_empty(exp->exp_lock_hash,
ldlm_cancel_locks_for_export_cb, &ecl);
+
+ CDEBUG(D_DLMTRACE, "Export %p, canceled %d locks, "
+ "left on hash table %d.\n", exp, ecl.ecl_loop,
+ atomic_read(&exp->exp_lock_hash->hs_count));
+
+ return ecl.ecl_loop;
}
/**
* \param lock A lock to convert
* \param new_mode new lock mode
*/
-void ldlm_lock_downgrade(struct ldlm_lock *lock, int new_mode)
+void ldlm_lock_downgrade(struct ldlm_lock *lock, enum ldlm_mode new_mode)
{
ENTRY;
* optimizations could take advantage of it to avoid discarding cached
* pages on a file.
*/
-struct ldlm_resource *ldlm_lock_convert(struct ldlm_lock *lock, int new_mode,
- __u32 *flags)
+struct ldlm_resource *ldlm_lock_convert(struct ldlm_lock *lock,
+ enum ldlm_mode new_mode, __u32 *flags)
{
struct list_head rpc_list;
struct ldlm_resource *res;
lock->l_completion_ast(lock, 0, NULL);
}
#ifdef HAVE_SERVER_SUPPORT
- } else {
- int rc;
- ldlm_error_t err;
+ } else {
+ int rc;
+ enum ldlm_error err;
__u64 pflags = 0;
- ldlm_processing_policy policy;
+ ldlm_processing_policy policy;
+
policy = ldlm_processing_policy_table[res->lr_type];
rc = policy(lock, &pflags, 0, &err, &rpc_list);
if (rc == LDLM_ITER_STOP) {
git://git.whamcloud.com / fs / lustre-release.git / blobdiff