-/* -*- 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.
* Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
*
- * Copyright (c) 2011 Whamcloud, Inc.
- *
+ * Copyright (c) 2011, 2013, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
*/
#define DEBUG_SUBSYSTEM S_CLASS
-#ifndef EXPORT_SYMTAB
-# define EXPORT_SYMTAB
-#endif
#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 cfs_lock_class_key_t cl_lock_guard_class;
+static struct lock_class_key cl_lock_guard_class;
static cfs_mem_cache_t *cl_lock_kmem;
static struct lu_kmem_descr cl_lock_caches[] = {
}
};
+#ifdef CONFIG_DEBUG_PAGESTATE_TRACKING
+#define CS_LOCK_INC(o, item) \
+ cfs_atomic_inc(&cl_object_site(o)->cs_locks.cs_stats[CS_##item])
+#define CS_LOCK_DEC(o, item) \
+ cfs_atomic_dec(&cl_object_site(o)->cs_locks.cs_stats[CS_##item])
+#define CS_LOCKSTATE_INC(o, state) \
+ cfs_atomic_inc(&cl_object_site(o)->cs_locks_state[state])
+#define CS_LOCKSTATE_DEC(o, state) \
+ cfs_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.
#define RETIP ((unsigned long)__builtin_return_address(0))
#ifdef CONFIG_LOCKDEP
-static cfs_lock_class_key_t cl_lock_key;
+static struct lock_class_key cl_lock_key;
static void cl_lock_lockdep_init(struct cl_lock *lock)
{
cfs_list_del_init(lock->cll_layers.next);
slice->cls_ops->clo_fini(env, slice);
}
- cfs_atomic_dec(&cl_object_site(obj)->cs_locks.cs_total);
- cfs_atomic_dec(&cl_object_site(obj)->cs_locks_state[lock->cll_state]);
+ 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);
- cfs_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_site *site;
LINVRNT(cl_lock_invariant(env, lock));
ENTRY;
obj = lock->cll_descr.cld_obj;
LINVRNT(obj != NULL);
- site = cl_object_site(obj);
CDEBUG(D_TRACE, "releasing reference: %d %p %lu\n",
cfs_atomic_read(&lock->cll_ref), lock, RETIP);
LASSERT(cfs_list_empty(&lock->cll_linkage));
cl_lock_free(env, lock);
}
- cfs_atomic_dec(&site->cs_locks.cs_busy);
+ CS_LOCK_DEC(obj, busy);
}
EXIT;
}
*/
void cl_lock_get_trust(struct cl_lock *lock)
{
- struct cl_site *site = cl_object_site(lock->cll_descr.cld_obj);
-
CDEBUG(D_TRACE, "acquiring trusted reference: %d %p %lu\n",
cfs_atomic_read(&lock->cll_ref), lock, RETIP);
if (cfs_atomic_inc_return(&lock->cll_ref) == 1)
- cfs_atomic_inc(&site->cs_locks.cs_busy);
+ CS_LOCK_INC(lock->cll_descr.cld_obj, busy);
}
EXPORT_SYMBOL(cl_lock_get_trust);
{
struct cl_lock *lock;
struct lu_object_header *head;
- struct cl_site *site = cl_object_site(obj);
ENTRY;
OBD_SLAB_ALLOC_PTR_GFP(lock, cl_lock_kmem, CFS_ALLOC_IO);
CFS_INIT_LIST_HEAD(&lock->cll_inclosure);
lu_ref_init(&lock->cll_reference);
lu_ref_init(&lock->cll_holders);
- cfs_mutex_init(&lock->cll_guard);
- cfs_lockdep_set_class(&lock->cll_guard, &cl_lock_guard_class);
+ 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;
- cfs_atomic_inc(&site->cs_locks_state[CLS_NEW]);
- cfs_atomic_inc(&site->cs_locks.cs_total);
- cfs_atomic_inc(&site->cs_locks.cs_created);
+ CS_LOCKSTATE_INC(obj, CLS_NEW);
+ CS_LOCK_INC(obj, total);
+ CS_LOCK_INC(obj, create);
cl_lock_lockdep_init(lock);
cfs_list_for_each_entry(obj, &head->loh_layers,
co_lu.lo_linkage) {
{
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_LOCKED(&head->coh_lock_guard);
- cfs_atomic_inc(&site->cs_locks.cs_lookup);
+ CS_LOCK_INC(obj, lookup);
cfs_list_for_each_entry(lock, &head->coh_locks, cll_linkage) {
int matched;
matched);
if (matched) {
cl_lock_get_trust(lock);
- cfs_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);
-
- cfs_spin_lock(&head->coh_lock_guard);
- lock = cl_lock_lookup(env, obj, io, need);
- cfs_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;
-
- cfs_spin_lock(&head->coh_lock_guard);
- ghost = cl_lock_lookup(env, obj, io, need);
- if (ghost == NULL) {
- cfs_list_add_tail(&lock->cll_linkage,
- &head->coh_locks);
- cfs_spin_unlock(&head->coh_lock_guard);
- cfs_atomic_inc(&site->cs_locks.cs_busy);
- } else {
- cfs_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) {
+ cfs_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
struct cl_object_header *head;
struct cl_object *obj;
struct cl_lock *lock;
- int ok;
obj = need->cld_obj;
head = cl_object_header(obj);
- cfs_spin_lock(&head->coh_lock_guard);
- lock = cl_lock_lookup(env, obj, io, need);
- cfs_spin_unlock(&head->coh_lock_guard);
-
- if (lock == NULL)
- return NULL;
-
- cl_lock_mutex_get(env, lock);
- if (lock->cll_state == CLS_INTRANSIT)
- cl_lock_state_wait(env, lock); /* Don't care return value. */
- if (lock->cll_state == CLS_CACHED) {
- int result;
- result = cl_use_try(env, lock, 1);
- if (result < 0)
- cl_lock_error(env, lock, result);
- }
- ok = lock->cll_state == CLS_HELD;
- if (ok) {
- cl_lock_hold_add(env, lock, scope, source);
- cl_lock_user_add(env, lock);
- cl_lock_put(env, lock);
- }
- cl_lock_mutex_put(env, lock);
- if (!ok) {
+ 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);
cl_lock_put(env, lock);
lock = NULL;
}
info = cl_env_info(env);
for (i = 0; i < hdr->coh_nesting; ++i)
LASSERT(info->clt_counters[i].ctc_nr_locks_locked == 0);
- cfs_mutex_lock_nested(&lock->cll_guard, hdr->coh_nesting);
+ mutex_lock_nested(&lock->cll_guard, hdr->coh_nesting);
lock->cll_guarder = cfs_current();
LINVRNT(lock->cll_depth == 0);
}
if (lock->cll_guarder == cfs_current()) {
LINVRNT(lock->cll_depth > 0);
cl_lock_mutex_tail(env, lock);
- } else if (cfs_mutex_trylock(&lock->cll_guard)) {
+ } else if (mutex_trylock(&lock->cll_guard)) {
LINVRNT(lock->cll_depth == 0);
lock->cll_guarder = cfs_current();
cl_lock_mutex_tail(env, lock);
counters->ctc_nr_locks_locked--;
if (--lock->cll_depth == 0) {
lock->cll_guarder = NULL;
- cfs_mutex_unlock(&lock->cll_guard);
+ mutex_unlock(&lock->cll_guard);
}
}
EXPORT_SYMBOL(cl_lock_mutex_put);
head = cl_object_header(lock->cll_descr.cld_obj);
- cfs_spin_lock(&head->coh_lock_guard);
- cfs_list_del_init(&lock->cll_linkage);
+ spin_lock(&head->coh_lock_guard);
+ cfs_list_del_init(&lock->cll_linkage);
+ spin_unlock(&head->coh_lock_guard);
- cfs_spin_unlock(&head->coh_lock_guard);
/*
* From now on, no new references to this lock can be acquired
* by cl_lock_lookup().
}
}
-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));
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 ||
- lock->cll_descr.cld_mode == CLM_GROUP)
+ 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 or grouplock when user is
* done with it---destroy the lock.
}
EXIT;
}
-
+EXPORT_SYMBOL(cl_lock_hold_release);
/**
* Waits until lock state is changed.
cl_lock_mutex_put(env, lock);
LASSERT(cl_lock_nr_mutexed(env) == 0);
- cfs_waitq_wait(&waiter, CFS_TASK_INTERRUPTIBLE);
+
+ /* 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))) {
+ cfs_waitq_wait(&waiter, CFS_TASK_INTERRUPTIBLE);
+ if (!cfs_signal_pending())
+ result = 0;
+ }
cl_lock_mutex_get(env, lock);
cfs_set_current_state(CFS_TASK_RUNNING);
cfs_waitq_del(&lock->cll_wq, &waiter);
- result = cfs_signal_pending() ? -EINTR : 0;
/* Restore old blocked signals */
cfs_restore_sigs(blocked);
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 &&
lock->cll_state == CLS_INTRANSIT);
if (lock->cll_state != state) {
- cfs_atomic_dec(&site->cs_locks_state[lock->cll_state]);
- cfs_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;
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_INTRANSIT:
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);
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);
}
break;
} while (1);
- if (result != 0) {
- cl_lock_user_del(env, lock);
- 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.
- * cl_unuse_try is a one-shot operation, so it must NOT return CLO_WAIT.
- *
- * \pre lock->cll_state == CLS_HELD
+ * 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.
*
- * \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
ENTRY;
cl_lock_trace(D_DLMTRACE, env, "unuse lock", lock);
- LASSERT(lock->cll_state == CLS_HELD || lock->cll_state == CLS_ENQUEUED);
if (lock->cll_users > 1) {
cl_lock_user_del(env, lock);
RETURN(0);
}
+ /* 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);
+ }
+
/*
* New lock users (->cll_users) are not protecting unlocking
* from proceeding. From this point, lock eventually reaches
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);
do {
LINVRNT(cl_lock_is_mutexed(lock));
LINVRNT(cl_lock_invariant(env, lock));
- LASSERT(lock->cll_state == CLS_ENQUEUED ||
- lock->cll_state == CLS_HELD ||
- lock->cll_state == CLS_INTRANSIT);
+ 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)) {
cl_lock_state_set(env, lock, CLS_HELD);
}
} while (result == CLO_REPEAT);
- RETURN(result ?: lock->cll_error);
+ RETURN(result);
}
EXPORT_SYMBOL(cl_wait_try);
break;
} while (1);
if (result < 0) {
- cl_lock_user_del(env, lock);
- 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);
* now. If locks were indexed according to their extent and/or mode,
* that index would have to be updated here.
*/
- cfs_spin_lock(&hdr->coh_lock_guard);
- lock->cll_descr = *desc;
- cfs_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_invariant(env, lock));
ENTRY;
- cl_lock_trace(D_DLMTRACE, env, "set lock error", lock);
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);
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;
- cfs_spin_lock(&head->coh_lock_guard);
+ 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 */
cfs_list_for_each_entry(scan, &head->coh_locks, cll_linkage) {
break;
}
}
- cfs_spin_unlock(&head->coh_lock_guard);
- RETURN(lock);
+ spin_unlock(&head->coh_lock_guard);
+ RETURN(lock);
}
-EXPORT_SYMBOL(cl_lock_at_page);
+EXPORT_SYMBOL(cl_lock_at_pgoff);
/**
- * 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.
+ * Calculate the page offset at the layer of @lock.
+ * At the time of this writing, @page is top page and @lock is sub lock.
*/
-void cl_lock_page_list_fixup(const struct lu_env *env,
- struct cl_io *io, struct cl_lock *lock,
- struct cl_page_list *queue)
+static pgoff_t pgoff_at_lock(struct cl_page *page, struct cl_lock *lock)
{
- 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;
+ struct lu_device_type *dtype;
+ const struct cl_page_slice *slice;
- /* No need to fix for WRITE lock because it is exclusive. */
- if (lock->cll_descr.cld_mode >= CLM_WRITE)
- RETURN_EXIT;
+ dtype = lock->cll_descr.cld_obj->co_lu.lo_dev->ld_type;
+ slice = cl_page_at(page, dtype);
+ LASSERT(slice != NULL);
+ return slice->cpl_page->cp_index;
+}
- /* For those pages who are still covered by other PR locks, we should
- * not discard them otherwise a [0, EOF) PR lock will discard all
- * pages.
- */
- 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, 1, 0);
- if (found == NULL)
- continue;
-
- descr = &found->cll_descr;
- cfs_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);
+/**
+ * Check if page @page is covered by an extra lock or discard it.
+ */
+static int check_and_discard_cb(const struct lu_env *env, struct cl_io *io,
+ struct cl_page *page, void *cbdata)
+{
+ struct cl_thread_info *info = cl_env_info(env);
+ struct cl_lock *lock = cbdata;
+ pgoff_t index = pgoff_at_lock(page, lock);
+
+ if (index >= info->clt_fn_index) {
+ struct cl_lock *tmp;
+
+ /* refresh non-overlapped index */
+ tmp = cl_lock_at_pgoff(env, lock->cll_descr.cld_obj, index,
+ lock, 1, 0);
+ if (tmp != NULL) {
+ /* Cache the first-non-overlapped index so as to skip
+ * all pages within [index, clt_fn_index). This
+ * is safe because if tmp lock is canceled, it will
+ * discard these pages. */
+ info->clt_fn_index = tmp->cll_descr.cld_end + 1;
+ if (tmp->cll_descr.cld_end == CL_PAGE_EOF)
+ info->clt_fn_index = CL_PAGE_EOF;
+ cl_lock_put(env, tmp);
+ } else if (cl_page_own(env, io, page) == 0) {
+ /* discard the page */
+ cl_page_unmap(env, io, page);
+ cl_page_discard(env, io, page);
+ cl_page_disown(env, io, page);
+ } else {
+ LASSERT(page->cp_state == CPS_FREEING);
}
- 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;
+ info->clt_next_index = index + 1;
+ return CLP_GANG_OKAY;
+}
+
+static int discard_cb(const struct lu_env *env, struct cl_io *io,
+ struct cl_page *page, void *cbdata)
+{
+ struct cl_thread_info *info = cl_env_info(env);
+ struct cl_lock *lock = cbdata;
+
+ LASSERT(lock->cll_descr.cld_mode >= CLM_WRITE);
+ KLASSERT(ergo(page->cp_type == CPT_CACHEABLE,
+ !PageWriteback(cl_page_vmpage(env, page))));
+ KLASSERT(ergo(page->cp_type == CPT_CACHEABLE,
+ !PageDirty(cl_page_vmpage(env, page))));
+
+ info->clt_next_index = pgoff_at_lock(page, lock) + 1;
+ if (cl_page_own(env, io, page) == 0) {
+ /* discard the page */
+ cl_page_unmap(env, io, page);
+ cl_page_discard(env, io, page);
+ cl_page_disown(env, io, page);
+ } else {
+ LASSERT(page->cp_state == CPS_FREEING);
+ }
+
+ return CLP_GANG_OKAY;
}
-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.
+ * Discard pages protected by the given lock. This function traverses radix
+ * tree to find all covering pages and discard them. If a page is being covered
+ * by other locks, it should remain in cache.
*
* 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)
+int cl_lock_discard_pages(const struct lu_env *env, struct cl_lock *lock)
{
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;
- struct lu_device_type *dtype;
- long page_count;
- pgoff_t next_index;
+ cl_page_gang_cb_t cb;
int res;
int result;
ENTRY;
io->ci_obj = cl_object_top(descr->cld_obj);
+ io->ci_ignore_layout = 1;
result = cl_io_init(env, io, CIT_MISC, io->ci_obj);
if (result != 0)
GOTO(out, result);
- dtype = descr->cld_obj->co_lu.lo_dev->ld_type;
- next_index = descr->cld_start;
+ cb = descr->cld_mode == CLM_READ ? check_and_discard_cb : discard_cb;
+ info->clt_fn_index = info->clt_next_index = descr->cld_start;
do {
- const struct cl_page_slice *slice;
-
- cl_2queue_init(queue);
res = cl_page_gang_lookup(env, descr->cld_obj, io,
- next_index, descr->cld_end,
- &queue->c2_qin);
- page_count = queue->c2_qin.pl_nr;
- if (page_count == 0)
- break;
-
- /* cl_page_gang_lookup() uses subobj and sublock to look for
- * covered pages, but @queue->c2_qin contains the list of top
- * pages. We have to turn the page back to subpage so as to
- * get `correct' next index. -jay */
- slice = cl_page_at(cl_page_list_last(&queue->c2_qin), dtype);
- next_index = slice->cpl_page->cp_index + 1;
-
- result = cl_page_list_unmap(env, io, &queue->c2_qin);
- if (!discard) {
- long timeout = 600; /* 10 minutes. */
- /* for debug purpose, if this request can't be
- * finished in 10 minutes, we hope it can notify us.
- */
- result = cl_io_submit_sync(env, io, CRT_WRITE, queue,
- CRP_CANCEL, timeout);
- if (result)
- CWARN("Writing %lu pages error: %d\n",
- page_count, result);
- }
- 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);
-
- if (next_index > descr->cld_end)
+ info->clt_next_index, descr->cld_end,
+ cb, (void *)lock);
+ if (info->clt_next_index > descr->cld_end)
break;
if (res == CLP_GANG_RESCHED)
cl_io_fini(env, io);
RETURN(result);
}
-EXPORT_SYMBOL(cl_lock_page_out);
+EXPORT_SYMBOL(cl_lock_discard_pages);
/**
* Eliminate all locks for a given object.
LASSERT(ergo(!cancel,
head->coh_tree.rnode == NULL && head->coh_pages == 0));
- cfs_spin_lock(&head->coh_lock_guard);
- while (!cfs_list_empty(&head->coh_locks)) {
- lock = container_of(head->coh_locks.next,
- struct cl_lock, cll_linkage);
- cl_lock_get_trust(lock);
- cfs_spin_unlock(&head->coh_lock_guard);
+ spin_lock(&head->coh_lock_guard);
+ while (!cfs_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());
+
+again:
cl_lock_mutex_get(env, lock);
if (lock->cll_state < CLS_FREEING) {
- LASSERT(lock->cll_holds == 0);
- LASSERT(lock->cll_users == 0);
+ 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", cfs_current());
cl_lock_put(env, lock);
- cfs_spin_lock(&head->coh_lock_guard);
- }
- cfs_spin_unlock(&head->coh_lock_guard);
- EXIT;
+ spin_lock(&head->coh_lock_guard);
+ }
+ spin_unlock(&head->coh_lock_guard);
+ EXIT;
}
EXPORT_SYMBOL(cl_locks_prune);
ENTRY;
do {
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;
}
- cl_unuse_locked(env, lock);
+ rc = 1;
}
- cl_lock_trace(D_DLMTRACE, env, "enqueue failed", 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);
+ 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);
+ }
} while (rc == 0);
RETURN(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));
ENTRY;
cl_lock_used_mod(env, lock, -1);
- RETURN(lock->cll_users == 0);
+ if (lock->cll_users == 0)
+ cfs_waitq_broadcast(&lock->cll_wq);
+ EXIT;
}
EXPORT_SYMBOL(cl_lock_user_del);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff