*
* You should have received a copy of the GNU General Public License
* version 2 along with this program; If not, see
- * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
- *
- * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
- * CA 95054 USA or visit www.sun.com if you need additional information or
- * have any questions.
+ * http://www.gnu.org/licenses/gpl-2.0.html
*
* GPL HEADER END
*/
* Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
*
- * Copyright (c) 2010, 2015, Intel Corporation.
+ * Copyright (c) 2010, 2017, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
#define DEBUG_SUBSYSTEM S_LDLM
#include <linux/kthread.h>
+#include <linux/list.h>
#include <libcfs/libcfs.h>
+#include <libcfs/linux/linux-mem.h>
+#include <lustre_errno.h>
#include <lustre_dlm.h>
#include <obd_class.h>
-#include <libcfs/list.h>
#include "ldlm_internal.h"
static int ldlm_num_threads;
-CFS_MODULE_PARM(ldlm_num_threads, "i", int, 0444,
- "number of DLM service threads to start");
+module_param(ldlm_num_threads, int, 0444);
+MODULE_PARM_DESC(ldlm_num_threads, "number of DLM service threads to start");
+
+static unsigned int ldlm_cpu_bind = 1;
+module_param(ldlm_cpu_bind, uint, 0444);
+MODULE_PARM_DESC(ldlm_cpu_bind,
+ "bind DLM service threads to particular CPU partitions");
static char *ldlm_cpts;
-CFS_MODULE_PARM(ldlm_cpts, "s", charp, 0444,
- "CPU partitions ldlm threads should run on");
+module_param(ldlm_cpts, charp, 0444);
+MODULE_PARM_DESC(ldlm_cpts, "CPU partitions ldlm threads should run on");
-static struct mutex ldlm_ref_mutex;
+static DEFINE_MUTEX(ldlm_ref_mutex);
static int ldlm_refcount;
-struct ldlm_cb_async_args {
- struct ldlm_cb_set_arg *ca_set_arg;
- struct ldlm_lock *ca_lock;
-};
+struct kobject *ldlm_kobj;
+struct kset *ldlm_ns_kset;
+struct kset *ldlm_svc_kset;
/* LDLM state */
static struct ldlm_state *ldlm_state;
-inline cfs_time_t round_timeout(cfs_time_t timeout)
-{
- return cfs_time_seconds((int)cfs_duration_sec(cfs_time_sub(timeout, 0)) + 1);
-}
-
-/* timeout for initial callback (AST) reply (bz10399) */
-static inline unsigned int ldlm_get_rq_timeout(void)
+/*
+ * timeout for initial callback (AST) reply (bz10399)
+ * Due to having to send a 32 bit time value over the
+ * wire return it as timeout_t instead of time64_t
+ */
+static inline timeout_t ldlm_get_rq_timeout(void)
{
- /* Non-AT value */
- unsigned int timeout = min(ldlm_timeout, obd_timeout / 3);
+ /* Non-AT value */
+ timeout_t timeout = min(ldlm_timeout, obd_timeout / 3);
- return timeout < 1 ? 1 : timeout;
+ return timeout < 1 ? 1 : timeout;
}
-#define ELT_STOPPED 0
-#define ELT_READY 1
-#define ELT_TERMINATE 2
-
struct ldlm_bl_pool {
- spinlock_t blp_lock;
+ spinlock_t blp_lock;
/*
* blp_prio_list is used for callbacks that should be handled
* as a priority. It is used for LDLM_FL_DISCARD_DATA requests.
- * see bug 13843
+ * see b=13843
*/
- struct list_head blp_prio_list;
+ struct list_head blp_prio_list;
/*
* blp_list is used for all other callbacks which are likely
* to take longer to process.
*/
- struct list_head blp_list;
-
- wait_queue_head_t blp_waitq;
- struct completion blp_comp;
- atomic_t blp_num_threads;
- atomic_t blp_busy_threads;
- int blp_min_threads;
- int blp_max_threads;
+ struct list_head blp_list;
+
+ wait_queue_head_t blp_waitq;
+ struct completion blp_comp;
+ atomic_t blp_num_threads;
+ atomic_t blp_busy_threads;
+ int blp_min_threads;
+ int blp_max_threads;
};
struct ldlm_bl_work_item {
/**
* Protects both waiting_locks_list and expired_lock_thread.
*/
-static spinlock_t waiting_locks_spinlock; /* BH lock (timer) */
+static DEFINE_SPINLOCK(waiting_locks_spinlock); /* BH lock (timer) */
/**
* List for contended locks.
*
* All access to it should be under waiting_locks_spinlock.
*/
-static struct list_head waiting_locks_list;
-static struct timer_list waiting_locks_timer;
+static LIST_HEAD(waiting_locks_list);
+static void waiting_locks_callback(TIMER_DATA_TYPE unused);
+static CFS_DEFINE_TIMER(waiting_locks_timer, waiting_locks_callback, 0, 0);
+
+enum elt_state {
+ ELT_STOPPED,
+ ELT_READY,
+ ELT_TERMINATE,
+};
-static struct expired_lock_thread {
- wait_queue_head_t elt_waitq;
- int elt_state;
- int elt_dump;
- struct list_head elt_expired_locks;
-} expired_lock_thread;
+static DECLARE_WAIT_QUEUE_HEAD(expired_lock_wait_queue);
+static enum elt_state expired_lock_thread_state = ELT_STOPPED;
+static int expired_lock_dump;
+static LIST_HEAD(expired_lock_list);
+
+static int ldlm_lock_busy(struct ldlm_lock *lock);
+static int ldlm_add_waiting_lock(struct ldlm_lock *lock, timeout_t timeout);
+static int __ldlm_add_waiting_lock(struct ldlm_lock *lock, timeout_t timeout);
static inline int have_expired_locks(void)
{
ENTRY;
spin_lock_bh(&waiting_locks_spinlock);
- need_to_run = !list_empty(&expired_lock_thread.elt_expired_locks);
+ need_to_run = !list_empty(&expired_lock_list);
spin_unlock_bh(&waiting_locks_spinlock);
RETURN(need_to_run);
*/
static int expired_lock_main(void *arg)
{
- struct list_head *expired = &expired_lock_thread.elt_expired_locks;
- struct l_wait_info lwi = { 0 };
+ struct list_head *expired = &expired_lock_list;
int do_dump;
ENTRY;
- expired_lock_thread.elt_state = ELT_READY;
- wake_up(&expired_lock_thread.elt_waitq);
+ expired_lock_thread_state = ELT_READY;
+ wake_up(&expired_lock_wait_queue);
while (1) {
- l_wait_event(expired_lock_thread.elt_waitq,
- have_expired_locks() ||
- expired_lock_thread.elt_state == ELT_TERMINATE,
- &lwi);
+ wait_event_idle(expired_lock_wait_queue,
+ have_expired_locks() ||
+ expired_lock_thread_state == ELT_TERMINATE);
spin_lock_bh(&waiting_locks_spinlock);
- if (expired_lock_thread.elt_dump) {
- struct libcfs_debug_msg_data msgdata = {
- .msg_file = __FILE__,
- .msg_fn = "waiting_locks_callback",
- .msg_line = expired_lock_thread.elt_dump };
+ if (expired_lock_dump) {
spin_unlock_bh(&waiting_locks_spinlock);
/* from waiting_locks_callback, but not in timer */
libcfs_debug_dumplog();
- libcfs_run_lbug_upcall(&msgdata);
spin_lock_bh(&waiting_locks_spinlock);
- expired_lock_thread.elt_dump = 0;
+ expired_lock_dump = 0;
}
do_dump = 0;
lock = list_entry(expired->next, struct ldlm_lock,
l_pending_chain);
- if ((void *)lock < LP_POISON + PAGE_CACHE_SIZE &&
+ if ((void *)lock < LP_POISON + PAGE_SIZE &&
(void *)lock >= LP_POISON) {
spin_unlock_bh(&waiting_locks_spinlock);
CERROR("free lock on elt list %p\n", lock);
}
list_del_init(&lock->l_pending_chain);
if ((void *)lock->l_export <
- LP_POISON + PAGE_CACHE_SIZE &&
+ LP_POISON + PAGE_SIZE &&
(void *)lock->l_export >= LP_POISON) {
CERROR("lock with free export on elt list %p\n",
lock->l_export);
lock->l_export = NULL;
LDLM_ERROR(lock, "free export");
- /* release extra ref grabbed by
+ /*
+ * release extra ref grabbed by
* ldlm_add_waiting_lock() or
- * ldlm_failed_ast() */
+ * ldlm_failed_ast()
+ */
LDLM_LOCK_RELEASE(lock);
continue;
}
if (ldlm_is_destroyed(lock)) {
- /* release the lock refcount where
- * waiting_locks_callback() founds */
+ /*
+ * release the lock refcount where
+ * waiting_locks_callback() founds
+ */
LDLM_LOCK_RELEASE(lock);
continue;
}
export = class_export_lock_get(lock->l_export, lock);
spin_unlock_bh(&waiting_locks_spinlock);
- spin_lock_bh(&export->exp_bl_list_lock);
- list_del_init(&lock->l_exp_list);
- spin_unlock_bh(&export->exp_bl_list_lock);
-
- do_dump++;
- class_fail_export(export);
+ /* Check if we need to prolong timeout */
+ if (!OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_TIMEOUT) &&
+ lock->l_callback_timestamp != 0 && /* not AST error */
+ ldlm_lock_busy(lock)) {
+ LDLM_DEBUG(lock, "prolong the busy lock");
+ lock_res_and_lock(lock);
+ ldlm_add_waiting_lock(lock,
+ ldlm_bl_timeout(lock) >> 1);
+ unlock_res_and_lock(lock);
+ } else {
+ spin_lock_bh(&export->exp_bl_list_lock);
+ list_del_init(&lock->l_exp_list);
+ spin_unlock_bh(&export->exp_bl_list_lock);
+
+ LDLM_ERROR(lock,
+ "lock callback timer expired after %llds: evicting client at %s ",
+ ktime_get_real_seconds() -
+ lock->l_blast_sent,
+ obd_export_nid2str(export));
+ ldlm_lock_to_ns(lock)->ns_timeouts++;
+ do_dump++;
+ class_fail_export(export);
+ }
class_export_lock_put(export, lock);
-
- /* release extra ref grabbed by ldlm_add_waiting_lock()
- * or ldlm_failed_ast() */
+ /*
+ * release extra ref grabbed by ldlm_add_waiting_lock()
+ * or ldlm_failed_ast()
+ */
LDLM_LOCK_RELEASE(lock);
spin_lock_bh(&waiting_locks_spinlock);
libcfs_debug_dumplog();
}
- if (expired_lock_thread.elt_state == ELT_TERMINATE)
+ if (expired_lock_thread_state == ELT_TERMINATE)
break;
}
- expired_lock_thread.elt_state = ELT_STOPPED;
- wake_up(&expired_lock_thread.elt_waitq);
+ expired_lock_thread_state = ELT_STOPPED;
+ wake_up(&expired_lock_wait_queue);
RETURN(0);
}
-static int ldlm_add_waiting_lock(struct ldlm_lock *lock);
-static int __ldlm_add_waiting_lock(struct ldlm_lock *lock, int seconds);
-
/**
* Check if there is a request in the export request list
* which prevents the lock canceling.
{
struct ptlrpc_request *req;
int match = 0;
+
ENTRY;
if (lock->l_export == NULL)
return 0;
- spin_lock_bh(&lock->l_export->exp_rpc_lock);
+ spin_lock(&lock->l_export->exp_rpc_lock);
list_for_each_entry(req, &lock->l_export->exp_hp_rpcs,
rq_exp_list) {
if (req->rq_ops->hpreq_lock_match) {
break;
}
}
- spin_unlock_bh(&lock->l_export->exp_rpc_lock);
+ spin_unlock(&lock->l_export->exp_rpc_lock);
RETURN(match);
}
/* This is called from within a timer interrupt and cannot schedule */
-static void waiting_locks_callback(unsigned long unused)
+static void waiting_locks_callback(TIMER_DATA_TYPE unused)
{
- struct ldlm_lock *lock;
- int need_dump = 0;
+ struct ldlm_lock *lock;
+ int need_dump = 0;
spin_lock_bh(&waiting_locks_spinlock);
while (!list_empty(&waiting_locks_list)) {
lock = list_entry(waiting_locks_list.next, struct ldlm_lock,
- l_pending_chain);
- if (cfs_time_after(lock->l_callback_timeout,
- cfs_time_current()) ||
- (lock->l_req_mode == LCK_GROUP))
- break;
-
- /* Check if we need to prolong timeout */
- if (!OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_TIMEOUT) &&
- ldlm_lock_busy(lock)) {
- int cont = 1;
-
- if (lock->l_pending_chain.next == &waiting_locks_list)
- cont = 0;
-
- LDLM_LOCK_GET(lock);
-
- spin_unlock_bh(&waiting_locks_spinlock);
- LDLM_DEBUG(lock, "prolong the busy lock");
- ldlm_refresh_waiting_lock(lock,
- ldlm_bl_timeout(lock) >> 1);
- spin_lock_bh(&waiting_locks_spinlock);
+ l_pending_chain);
+ if (lock->l_callback_timestamp > ktime_get_seconds() ||
+ lock->l_req_mode == LCK_GROUP)
+ break;
- if (!cont) {
- LDLM_LOCK_RELEASE(lock);
- break;
- }
-
- LDLM_LOCK_RELEASE(lock);
- continue;
- }
- ldlm_lock_to_ns(lock)->ns_timeouts++;
- LDLM_ERROR(lock, "lock callback timer expired after %lds: "
- "evicting client at %s ",
- cfs_time_current_sec() - lock->l_last_activity,
- libcfs_nid2str(
- lock->l_export->exp_connection->c_peer.nid));
-
- /* no needs to take an extra ref on the lock since it was in
- * the waiting_locks_list and ldlm_add_waiting_lock()
- * already grabbed a ref */
- list_del(&lock->l_pending_chain);
- list_add(&lock->l_pending_chain,
- &expired_lock_thread.elt_expired_locks);
+ /*
+ * no needs to take an extra ref on the lock since it was in
+ * the waiting_locks_list and ldlm_add_waiting_lock()
+ * already grabbed a ref
+ */
+ list_move(&lock->l_pending_chain, &expired_lock_list);
need_dump = 1;
}
- if (!list_empty(&expired_lock_thread.elt_expired_locks)) {
+ if (!list_empty(&expired_lock_list)) {
if (obd_dump_on_timeout && need_dump)
- expired_lock_thread.elt_dump = __LINE__;
+ expired_lock_dump = __LINE__;
- wake_up(&expired_lock_thread.elt_waitq);
+ wake_up(&expired_lock_wait_queue);
}
- /*
- * Make sure the timer will fire again if we have any locks
- * left.
- */
+ /*
+ * Make sure the timer will fire again if we have any locks
+ * left.
+ */
if (!list_empty(&waiting_locks_list)) {
- cfs_time_t timeout_rounded;
+ time64_t now = ktime_get_seconds();
+ timeout_t delta = 0;
+
lock = list_entry(waiting_locks_list.next, struct ldlm_lock,
- l_pending_chain);
- timeout_rounded = (cfs_time_t)round_timeout(lock->l_callback_timeout);
- cfs_timer_arm(&waiting_locks_timer, timeout_rounded);
- }
+ l_pending_chain);
+ if (lock->l_callback_timestamp - now > 0)
+ delta = lock->l_callback_timestamp - now;
+ mod_timer(&waiting_locks_timer,
+ jiffies + cfs_time_seconds(delta));
+ }
spin_unlock_bh(&waiting_locks_spinlock);
}
*
* Called with the namespace lock held.
*/
-static int __ldlm_add_waiting_lock(struct ldlm_lock *lock, int seconds)
+static int __ldlm_add_waiting_lock(struct ldlm_lock *lock, timeout_t delay)
{
- cfs_time_t timeout;
- cfs_time_t timeout_rounded;
+ unsigned long timeout_jiffies = jiffies;
+ time64_t now = ktime_get_seconds();
+ time64_t deadline;
+ timeout_t timeout;
if (!list_empty(&lock->l_pending_chain))
- return 0;
-
- if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_NOTIMEOUT) ||
- OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_TIMEOUT))
- seconds = 1;
-
- timeout = cfs_time_shift(seconds);
- if (likely(cfs_time_after(timeout, lock->l_callback_timeout)))
- lock->l_callback_timeout = timeout;
-
- timeout_rounded = round_timeout(lock->l_callback_timeout);
-
- if (cfs_time_before(timeout_rounded,
- cfs_timer_deadline(&waiting_locks_timer)) ||
- !cfs_timer_is_armed(&waiting_locks_timer)) {
- cfs_timer_arm(&waiting_locks_timer, timeout_rounded);
- }
- /* if the new lock has a shorter timeout than something earlier on
- the list, we'll wait the longer amount of time; no big deal. */
- /* FIFO */
+ return 0;
+
+ if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_NOTIMEOUT) ||
+ OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_HPREQ_TIMEOUT))
+ delay = 1;
+
+ deadline = now + delay;
+ if (likely(deadline > lock->l_callback_timestamp))
+ lock->l_callback_timestamp = deadline;
+
+ timeout = clamp_t(timeout_t, lock->l_callback_timestamp - now,
+ 0, delay);
+ timeout_jiffies += cfs_time_seconds(timeout);
+
+ if (time_before(timeout_jiffies, waiting_locks_timer.expires) ||
+ !timer_pending(&waiting_locks_timer))
+ mod_timer(&waiting_locks_timer, timeout_jiffies);
+
+ /*
+ * if the new lock has a shorter timeout than something earlier on
+ * the list, we'll wait the longer amount of time; no big deal.
+ */
+ /* FIFO */
list_add_tail(&lock->l_pending_chain, &waiting_locks_list);
- return 1;
+ return 1;
}
static void ldlm_add_blocked_lock(struct ldlm_lock *lock)
{
spin_lock_bh(&lock->l_export->exp_bl_list_lock);
if (list_empty(&lock->l_exp_list)) {
- if (lock->l_granted_mode != lock->l_req_mode)
+ if (!ldlm_is_granted(lock))
list_add_tail(&lock->l_exp_list,
&lock->l_export->exp_bl_list);
else
}
spin_unlock_bh(&lock->l_export->exp_bl_list_lock);
- /* A blocked lock is added. Adjust the position in
+ /*
+ * A blocked lock is added. Adjust the position in
* the stale list if the export is in the list.
* If export is stale and not in the list - it is being
* processed and will be placed on the right position
- * on obd_stale_export_put(). */
+ * on obd_stale_export_put().
+ */
if (!list_empty(&lock->l_export->exp_stale_list))
obd_stale_export_adjust(lock->l_export);
}
-static int ldlm_add_waiting_lock(struct ldlm_lock *lock)
+static int ldlm_add_waiting_lock(struct ldlm_lock *lock, timeout_t timeout)
{
int ret;
- int timeout = ldlm_bl_timeout(lock);
/* NB: must be called with hold of lock_res_and_lock() */
LASSERT(ldlm_is_res_locked(lock));
LASSERT(!ldlm_is_cancel_on_block(lock));
- /* Do not put cross-MDT lock in the waiting list, since we
- * will not evict it due to timeout for now */
+ /*
+ * Do not put cross-MDT lock in the waiting list, since we
+ * will not evict it due to timeout for now
+ */
if (lock->l_export != NULL &&
(exp_connect_flags(lock->l_export) & OBD_CONNECT_MDS_MDS))
return 0;
}
if (ldlm_is_destroyed(lock)) {
- static cfs_time_t next;
+ static time64_t next;
spin_unlock_bh(&waiting_locks_spinlock);
- LDLM_ERROR(lock, "not waiting on destroyed lock (bug 5653)");
- if (cfs_time_after(cfs_time_current(), next)) {
- next = cfs_time_shift(14400);
+ LDLM_ERROR(lock, "not waiting on destroyed lock (b=5653)");
+ if (ktime_get_seconds() > next) {
+ next = ktime_get_seconds() + 14400;
libcfs_debug_dumpstack(NULL);
}
return 0;
}
ldlm_set_waited(lock);
- lock->l_last_activity = cfs_time_current_sec();
+ lock->l_blast_sent = ktime_get_real_seconds();
ret = __ldlm_add_waiting_lock(lock, timeout);
if (ret) {
- /* grab ref on the lock if it has been added to the
- * waiting list */
+ /*
+ * grab ref on the lock if it has been added to the
+ * waiting list
+ */
LDLM_LOCK_GET(lock);
}
spin_unlock_bh(&waiting_locks_spinlock);
struct list_head *list_next;
if (list_empty(&lock->l_pending_chain))
- return 0;
-
- list_next = lock->l_pending_chain.next;
- if (lock->l_pending_chain.prev == &waiting_locks_list) {
- /* Removing the head of the list, adjust timer. */
- if (list_next == &waiting_locks_list) {
- /* No more, just cancel. */
- cfs_timer_disarm(&waiting_locks_timer);
- } else {
- struct ldlm_lock *next;
+ return 0;
+
+ list_next = lock->l_pending_chain.next;
+ if (lock->l_pending_chain.prev == &waiting_locks_list) {
+ /* Removing the head of the list, adjust timer. */
+ if (list_next == &waiting_locks_list) {
+ /* No more, just cancel. */
+ del_timer(&waiting_locks_timer);
+ } else {
+ time64_t now = ktime_get_seconds();
+ struct ldlm_lock *next;
+ timeout_t delta = 0;
+
next = list_entry(list_next, struct ldlm_lock,
- l_pending_chain);
- cfs_timer_arm(&waiting_locks_timer,
- round_timeout(next->l_callback_timeout));
- }
- }
+ l_pending_chain);
+ if (next->l_callback_timestamp - now > 0)
+ delta = lock->l_callback_timestamp - now;
+
+ mod_timer(&waiting_locks_timer,
+ jiffies + cfs_time_seconds(delta));
+ }
+ }
list_del_init(&lock->l_pending_chain);
- return 1;
+ return 1;
}
int ldlm_del_waiting_lock(struct ldlm_lock *lock)
{
- int ret;
+ int ret;
- if (lock->l_export == NULL) {
- /* We don't have a "waiting locks list" on clients. */
- CDEBUG(D_DLMTRACE, "Client lock %p : no-op\n", lock);
- return 0;
- }
+ if (lock->l_export == NULL) {
+ /* We don't have a "waiting locks list" on clients. */
+ CDEBUG(D_DLMTRACE, "Client lock %p : no-op\n", lock);
+ return 0;
+ }
spin_lock_bh(&waiting_locks_spinlock);
ret = __ldlm_del_waiting_lock(lock);
list_del_init(&lock->l_exp_list);
spin_unlock_bh(&lock->l_export->exp_bl_list_lock);
- if (ret) {
- /* release lock ref if it has indeed been removed
- * from a list */
- LDLM_LOCK_RELEASE(lock);
- }
+ if (ret) {
+ /*
+ * release lock ref if it has indeed been removed
+ * from a list
+ */
+ LDLM_LOCK_RELEASE(lock);
+ }
- LDLM_DEBUG(lock, "%s", ret == 0 ? "wasn't waiting" : "removed");
- return ret;
+ LDLM_DEBUG(lock, "%s", ret == 0 ? "wasn't waiting" : "removed");
+ return ret;
}
/**
*
* Called with namespace lock held.
*/
-int ldlm_refresh_waiting_lock(struct ldlm_lock *lock, int timeout)
+int ldlm_refresh_waiting_lock(struct ldlm_lock *lock, timeout_t timeout)
{
if (lock->l_export == NULL) {
/* We don't have a "waiting locks list" on clients. */
return 0;
}
- /* we remove/add the lock to the waiting list, so no needs to
- * release/take a lock reference */
+ /*
+ * we remove/add the lock to the waiting list, so no needs to
+ * release/take a lock reference
+ */
__ldlm_del_waiting_lock(lock);
__ldlm_add_waiting_lock(lock, timeout);
spin_unlock_bh(&waiting_locks_spinlock);
int ldlm_del_waiting_lock(struct ldlm_lock *lock)
{
- RETURN(0);
+ RETURN(0);
}
-int ldlm_refresh_waiting_lock(struct ldlm_lock *lock, int timeout)
+int ldlm_refresh_waiting_lock(struct ldlm_lock *lock, timeout_t timeout)
{
- RETURN(0);
+ RETURN(0);
}
#endif /* !HAVE_SERVER_SUPPORT */
*
* \retval timeout in seconds to wait for the client reply
*/
-unsigned int ldlm_bl_timeout(struct ldlm_lock *lock)
+timeout_t ldlm_bl_timeout(struct ldlm_lock *lock)
{
- unsigned int timeout;
+ timeout_t timeout;
if (AT_OFF)
return obd_timeout / 2;
- /* Since these are non-updating timeouts, we should be conservative.
+ /*
+ * Since these are non-updating timeouts, we should be conservative.
* Take more than usually, 150%
* It would be nice to have some kind of "early reply" mechanism for
- * lock callbacks too... */
+ * lock callbacks too...
+ */
timeout = at_get(&lock->l_export->exp_bl_lock_at);
- return max(timeout + (timeout >> 1), ldlm_enqueue_min);
+ return max_t(timeout_t, timeout + (timeout >> 1),
+ (timeout_t)ldlm_enqueue_min);
}
EXPORT_SYMBOL(ldlm_bl_timeout);
* Perform lock cleanup if AST sending failed.
*/
static void ldlm_failed_ast(struct ldlm_lock *lock, int rc,
- const char *ast_type)
+ const char *ast_type)
{
- LCONSOLE_ERROR_MSG(0x138, "%s: A client on nid %s was evicted due "
- "to a lock %s callback time out: rc %d\n",
- lock->l_export->exp_obd->obd_name,
- obd_export_nid2str(lock->l_export), ast_type, rc);
+ LCONSOLE_ERROR_MSG(0x138,
+ "%s: A client on nid %s was evicted due to a lock %s callback time out: rc %d\n",
+ lock->l_export->exp_obd->obd_name,
+ obd_export_nid2str(lock->l_export), ast_type, rc);
- if (obd_dump_on_timeout)
- libcfs_debug_dumplog();
+ if (obd_dump_on_timeout)
+ libcfs_debug_dumplog();
spin_lock_bh(&waiting_locks_spinlock);
if (__ldlm_del_waiting_lock(lock) == 0)
- /* the lock was not in any list, grab an extra ref before adding
- * the lock to the expired list */
+ /*
+ * the lock was not in any list, grab an extra ref before adding
+ * the lock to the expired list
+ */
LDLM_LOCK_GET(lock);
- list_add(&lock->l_pending_chain,
- &expired_lock_thread.elt_expired_locks);
- wake_up(&expired_lock_thread.elt_waitq);
+ /* differentiate it from expired locks */
+ lock->l_callback_timestamp = 0;
+ list_add(&lock->l_pending_chain, &expired_lock_list);
+ wake_up(&expired_lock_wait_queue);
spin_unlock_bh(&waiting_locks_spinlock);
}
* Perform lock cleanup if AST reply came with error.
*/
static int ldlm_handle_ast_error(struct ldlm_lock *lock,
- struct ptlrpc_request *req, int rc,
- const char *ast_type)
+ struct ptlrpc_request *req, int rc,
+ const char *ast_type)
{
- lnet_process_id_t peer = req->rq_import->imp_connection->c_peer;
+ struct lnet_process_id peer = req->rq_import->imp_connection->c_peer;
if (!req->rq_replied || (rc && rc != -EINVAL)) {
- if (lock->l_export && lock->l_export->exp_libclient) {
- LDLM_DEBUG(lock, "%s AST to liblustre client (nid %s)"
- " timeout, just cancelling lock", ast_type,
+ if (ldlm_is_cancel(lock)) {
+ LDLM_DEBUG(lock,
+ "%s AST (req@%p x%llu) timeout from nid %s, but cancel was received (AST reply lost?)",
+ ast_type, req, req->rq_xid,
libcfs_nid2str(peer.nid));
ldlm_lock_cancel(lock);
rc = -ERESTART;
- } else if (ldlm_is_cancel(lock)) {
- LDLM_DEBUG(lock, "%s AST timeout from nid %s, but "
- "cancel was received (AST reply lost?)",
- ast_type, libcfs_nid2str(peer.nid));
- ldlm_lock_cancel(lock);
- rc = -ERESTART;
+ } else if (rc == -ENODEV || rc == -ESHUTDOWN ||
+ (rc == -EIO &&
+ req->rq_import->imp_state == LUSTRE_IMP_CLOSED)) {
+ /*
+ * Upon umount process the AST fails because cannot be
+ * sent. This shouldn't lead to the client eviction.
+ * -ENODEV error is returned by ptl_send_rpc() for
+ * new request in such import.
+ * -SHUTDOWN is returned by ptlrpc_import_delay_req()
+ * if imp_invalid is set or obd_no_recov.
+ * Meanwhile there is also check for LUSTRE_IMP_CLOSED
+ * in ptlrpc_import_delay_req() as well with -EIO code.
+ * In all such cases errors are ignored.
+ */
+ LDLM_DEBUG(lock,
+ "%s AST can't be sent due to a server %s failure or umount process: rc = %d\n",
+ ast_type,
+ req->rq_import->imp_obd->obd_name, rc);
} else {
- LDLM_ERROR(lock, "client (nid %s) %s %s AST "
- "(req status %d rc %d), evict it",
+ LDLM_ERROR(lock,
+ "client (nid %s) %s %s AST (req@%p x%llu status %d rc %d), evict it",
libcfs_nid2str(peer.nid),
req->rq_replied ? "returned error from" :
"failed to reply to",
- ast_type,
+ ast_type, req, req->rq_xid,
(req->rq_repmsg != NULL) ?
lustre_msg_get_status(req->rq_repmsg) : 0,
rc);
if (rc == -EINVAL) {
struct ldlm_resource *res = lock->l_resource;
- LDLM_DEBUG(lock, "client (nid %s) returned %d"
- " from %s AST - normal race",
+ LDLM_DEBUG(lock,
+ "client (nid %s) returned %d from %s AST (req@%p x%llu) - normal race",
libcfs_nid2str(peer.nid),
req->rq_repmsg ?
lustre_msg_get_status(req->rq_repmsg) : -1,
- ast_type);
+ ast_type, req, req->rq_xid);
if (res) {
- /* update lvbo to return proper attributes.
- * see bug 23174 */
+ /*
+ * update lvbo to return proper attributes.
+ * see b=23174
+ */
ldlm_resource_getref(res);
- ldlm_res_lvbo_update(res, NULL, 1);
+ ldlm_lvbo_update(res, lock, NULL, 1);
ldlm_resource_putref(res);
}
ldlm_lock_cancel(lock);
}
static int ldlm_cb_interpret(const struct lu_env *env,
- struct ptlrpc_request *req, void *data, int rc)
+ struct ptlrpc_request *req, void *args, int rc)
{
- struct ldlm_cb_async_args *ca = data;
- struct ldlm_lock *lock = ca->ca_lock;
- struct ldlm_cb_set_arg *arg = ca->ca_set_arg;
- ENTRY;
+ struct ldlm_cb_async_args *ca = args;
+ struct ldlm_lock *lock = ca->ca_lock;
+ struct ldlm_cb_set_arg *arg = ca->ca_set_arg;
- LASSERT(lock != NULL);
+ ENTRY;
+
+ LASSERT(lock != NULL);
switch (arg->type) {
case LDLM_GL_CALLBACK:
- /* Update the LVB from disk if the AST failed
+ /*
+ * Update the LVB from disk if the AST failed
* (this is a legal race)
*
* - Glimpse callback of local lock just returns
* - Glimpse callback of remote lock might return
* -ELDLM_NO_LOCK_DATA when inode is cleared. LU-274
*/
- if (rc == -ELDLM_NO_LOCK_DATA) {
- LDLM_DEBUG(lock, "lost race - client has a lock but no "
- "inode");
- ldlm_res_lvbo_update(lock->l_resource, NULL, 1);
+ if (unlikely(arg->gl_interpret_reply)) {
+ rc = arg->gl_interpret_reply(NULL, req, args, rc);
+ } else if (rc == -ELDLM_NO_LOCK_DATA) {
+ LDLM_DEBUG(lock,
+ "lost race - client has a lock but no inode");
+ ldlm_lvbo_update(lock->l_resource, lock, NULL, 1);
} else if (rc != 0) {
rc = ldlm_handle_ast_error(lock, req, rc, "glimpse");
} else {
- rc = ldlm_res_lvbo_update(lock->l_resource, req, 1);
+ rc = ldlm_lvbo_update(lock->l_resource,
+ lock, req, 1);
}
break;
case LDLM_BL_CALLBACK:
}
/* release extra reference taken in ldlm_ast_fini() */
- LDLM_LOCK_RELEASE(lock);
+ LDLM_LOCK_RELEASE(lock);
if (rc == -ERESTART)
atomic_inc(&arg->restart);
static void ldlm_update_resend(struct ptlrpc_request *req, void *data)
{
- struct ldlm_cb_async_args *ca = data;
- struct ldlm_lock *lock = ca->ca_lock;
+ struct ldlm_cb_async_args *ca = data;
+ struct ldlm_lock *lock = ca->ca_lock;
ldlm_refresh_waiting_lock(lock, ldlm_bl_timeout(lock));
}
int instant_cancel)
{
int rc = 0;
+
ENTRY;
if (unlikely(instant_cancel)) {
static void ldlm_lock_reorder_req(struct ldlm_lock *lock)
{
struct ptlrpc_request *req;
+
ENTRY;
if (lock->l_export == NULL) {
RETURN_EXIT;
}
- spin_lock_bh(&lock->l_export->exp_rpc_lock);
+ spin_lock(&lock->l_export->exp_rpc_lock);
list_for_each_entry(req, &lock->l_export->exp_hp_rpcs,
rq_exp_list) {
- /* Do not process requests that were not yet added to there
+ /*
+ * Do not process requests that were not yet added to there
* incoming queue or were already removed from there for
* processing. We evaluate ptlrpc_nrs_req_can_move() without
* holding svcpt->scp_req_lock, and then redo the check with
req->rq_ops->hpreq_lock_match(req, lock))
ptlrpc_nrs_req_hp_move(req);
}
- spin_unlock_bh(&lock->l_export->exp_rpc_lock);
+ spin_unlock(&lock->l_export->exp_rpc_lock);
EXIT;
}
* to wait for client response.
*/
int ldlm_server_blocking_ast(struct ldlm_lock *lock,
- struct ldlm_lock_desc *desc,
- void *data, int flag)
-{
- struct ldlm_cb_async_args *ca;
- struct ldlm_cb_set_arg *arg = data;
- struct ldlm_request *body;
- struct ptlrpc_request *req;
- int instant_cancel = 0;
- int rc = 0;
- ENTRY;
+ struct ldlm_lock_desc *desc,
+ void *data, int flag)
+{
+ struct ldlm_cb_async_args *ca;
+ struct ldlm_cb_set_arg *arg = data;
+ struct ldlm_request *body;
+ struct ptlrpc_request *req;
+ int instant_cancel = 0;
+ int rc = 0;
+
+ ENTRY;
+
+ if (flag == LDLM_CB_CANCELING)
+ /* Don't need to do anything here. */
+ RETURN(0);
- if (flag == LDLM_CB_CANCELING)
- /* Don't need to do anything here. */
- RETURN(0);
+ if (OBD_FAIL_PRECHECK(OBD_FAIL_LDLM_SRV_BL_AST)) {
+ LDLM_DEBUG(lock, "dropping BL AST");
+ RETURN(0);
+ }
- LASSERT(lock);
- LASSERT(data != NULL);
- if (lock->l_export->exp_obd->obd_recovering != 0)
- LDLM_ERROR(lock, "BUG 6063: lock collide during recovery");
+ LASSERT(lock);
+ LASSERT(data != NULL);
+ if (lock->l_export->exp_obd->obd_recovering != 0)
+ LDLM_ERROR(lock, "BUG 6063: lock collide during recovery");
- ldlm_lock_reorder_req(lock);
+ ldlm_lock_reorder_req(lock);
- req = ptlrpc_request_alloc_pack(lock->l_export->exp_imp_reverse,
- &RQF_LDLM_BL_CALLBACK,
- LUSTRE_DLM_VERSION, LDLM_BL_CALLBACK);
- if (req == NULL)
- RETURN(-ENOMEM);
+ req = ptlrpc_request_alloc_pack(lock->l_export->exp_imp_reverse,
+ &RQF_LDLM_BL_CALLBACK,
+ LUSTRE_DLM_VERSION, LDLM_BL_CALLBACK);
+ if (req == NULL)
+ RETURN(-ENOMEM);
- CLASSERT(sizeof(*ca) <= sizeof(req->rq_async_args));
- ca = ptlrpc_req_async_args(req);
- ca->ca_set_arg = arg;
- ca->ca_lock = lock;
+ ca = ptlrpc_req_async_args(ca, req);
+ ca->ca_set_arg = arg;
+ ca->ca_lock = lock;
- req->rq_interpret_reply = ldlm_cb_interpret;
+ req->rq_interpret_reply = ldlm_cb_interpret;
lock_res_and_lock(lock);
if (ldlm_is_destroyed(lock)) {
RETURN(0);
}
- if (lock->l_granted_mode != lock->l_req_mode) {
- /* this blocking AST will be communicated as part of the
- * completion AST instead */
+ if (!ldlm_is_granted(lock)) {
+ /*
+ * this blocking AST will be communicated as part of the
+ * completion AST instead
+ */
ldlm_add_blocked_lock(lock);
ldlm_set_waited(lock);
unlock_res_and_lock(lock);
}
if (ldlm_is_cancel_on_block(lock))
- instant_cancel = 1;
+ instant_cancel = 1;
- body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
- body->lock_handle[0] = lock->l_remote_handle;
- body->lock_desc = *desc;
+ body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
+ body->lock_handle[0] = lock->l_remote_handle;
+ body->lock_desc = *desc;
body->lock_flags |= ldlm_flags_to_wire(lock->l_flags & LDLM_FL_AST_MASK);
- LDLM_DEBUG(lock, "server preparing blocking AST");
+ LDLM_DEBUG(lock, "server preparing blocking AST");
- ptlrpc_request_set_replen(req);
+ ptlrpc_request_set_replen(req);
ldlm_set_cbpending(lock);
if (instant_cancel) {
unlock_res_and_lock(lock);
req->rq_no_resend = 1;
} else {
- LASSERT(lock->l_granted_mode == lock->l_req_mode);
- ldlm_add_waiting_lock(lock);
+ LASSERT(ldlm_is_granted(lock));
+ ldlm_add_waiting_lock(lock, ldlm_bl_timeout(lock));
unlock_res_and_lock(lock);
/* Do not resend after lock callback timeout */
req->rq_resend_cb = ldlm_update_resend;
}
- req->rq_send_state = LUSTRE_IMP_FULL;
- /* ptlrpc_request_alloc_pack already set timeout */
- if (AT_OFF)
- req->rq_timeout = ldlm_get_rq_timeout();
-
- lock->l_last_activity = cfs_time_current_sec();
+ req->rq_send_state = LUSTRE_IMP_FULL;
+ /* ptlrpc_request_alloc_pack already set timeout */
+ if (AT_OFF)
+ req->rq_timeout = ldlm_get_rq_timeout();
- if (lock->l_export && lock->l_export->exp_nid_stats &&
- lock->l_export->exp_nid_stats->nid_ldlm_stats)
- lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
- LDLM_BL_CALLBACK - LDLM_FIRST_OPC);
+ if (lock->l_export && lock->l_export->exp_nid_stats &&
+ lock->l_export->exp_nid_stats->nid_ldlm_stats)
+ lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
+ LDLM_BL_CALLBACK - LDLM_FIRST_OPC);
rc = ldlm_ast_fini(req, arg, lock, instant_cancel);
- RETURN(rc);
+ RETURN(rc);
}
/**
*/
int ldlm_server_completion_ast(struct ldlm_lock *lock, __u64 flags, void *data)
{
- struct ldlm_cb_set_arg *arg = data;
- struct ldlm_request *body;
- struct ptlrpc_request *req;
- struct ldlm_cb_async_args *ca;
- int instant_cancel = 0;
- int rc = 0;
- int lvb_len;
- ENTRY;
+ struct ldlm_cb_set_arg *arg = data;
+ struct ldlm_request *body;
+ struct ptlrpc_request *req;
+ struct ldlm_cb_async_args *ca;
+ int instant_cancel = 0;
+ int rc = 0;
+ int lvb_len;
+
+ ENTRY;
- LASSERT(lock != NULL);
- LASSERT(data != NULL);
+ LASSERT(lock != NULL);
+ LASSERT(data != NULL);
if (OBD_FAIL_PRECHECK(OBD_FAIL_LDLM_SRV_CP_AST)) {
LDLM_DEBUG(lock, "dropping CP AST");
RETURN(0);
}
- req = ptlrpc_request_alloc(lock->l_export->exp_imp_reverse,
- &RQF_LDLM_CP_CALLBACK);
- if (req == NULL)
- RETURN(-ENOMEM);
+ req = ptlrpc_request_alloc(lock->l_export->exp_imp_reverse,
+ &RQF_LDLM_CP_CALLBACK);
+ if (req == NULL)
+ RETURN(-ENOMEM);
/* server namespace, doesn't need lock */
lvb_len = ldlm_lvbo_size(lock);
- /* LU-3124 & LU-2187: to not return layout in completion AST because
+ /*
+ * LU-3124 & LU-2187: to not return layout in completion AST because
* it may deadlock for LU-2187, or client may not have enough space
* for large layout. The layout will be returned to client with an
- * extra RPC to fetch xattr.lov */
+ * extra RPC to fetch xattr.lov
+ */
if (ldlm_has_layout(lock))
lvb_len = 0;
req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_CLIENT, lvb_len);
- rc = ptlrpc_request_pack(req, LUSTRE_DLM_VERSION, LDLM_CP_CALLBACK);
- if (rc) {
- ptlrpc_request_free(req);
- RETURN(rc);
- }
+ rc = ptlrpc_request_pack(req, LUSTRE_DLM_VERSION, LDLM_CP_CALLBACK);
+ if (rc) {
+ ptlrpc_request_free(req);
+ RETURN(rc);
+ }
- CLASSERT(sizeof(*ca) <= sizeof(req->rq_async_args));
- ca = ptlrpc_req_async_args(req);
- ca->ca_set_arg = arg;
- ca->ca_lock = lock;
+ ca = ptlrpc_req_async_args(ca, req);
+ ca->ca_set_arg = arg;
+ ca->ca_lock = lock;
- req->rq_interpret_reply = ldlm_cb_interpret;
- body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
+ req->rq_interpret_reply = ldlm_cb_interpret;
+ body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
- body->lock_handle[0] = lock->l_remote_handle;
+ body->lock_handle[0] = lock->l_remote_handle;
body->lock_flags = ldlm_flags_to_wire(flags);
- ldlm_lock2desc(lock, &body->lock_desc);
+ ldlm_lock2desc(lock, &body->lock_desc);
if (lvb_len > 0) {
void *lvb = req_capsule_client_get(&req->rq_pill, &RMF_DLM_LVB);
-
- lvb_len = ldlm_lvbo_fill(lock, lvb, lvb_len);
+ lvb_len = ldlm_lvbo_fill(lock, lvb, &lvb_len);
if (lvb_len < 0) {
- /* We still need to send the RPC to wake up the blocked
+ /*
+ * We still need to send the RPC to wake up the blocked
* enqueue thread on the client.
*
* Consider old client, there is no better way to notify
* the failure, just zero-sized the LVB, then the client
- * will fail out as "-EPROTO". */
+ * will fail out as "-EPROTO".
+ */
req_capsule_shrink(&req->rq_pill, &RMF_DLM_LVB, 0,
RCL_CLIENT);
instant_cancel = 1;
req_capsule_shrink(&req->rq_pill, &RMF_DLM_LVB, lvb_len,
RCL_CLIENT);
}
- }
-
- lock->l_last_activity = cfs_time_current_sec();
+ }
LDLM_DEBUG(lock, "server preparing completion AST");
- ptlrpc_request_set_replen(req);
+ ptlrpc_request_set_replen(req);
- req->rq_send_state = LUSTRE_IMP_FULL;
- /* ptlrpc_request_pack already set timeout */
- if (AT_OFF)
- req->rq_timeout = ldlm_get_rq_timeout();
+ req->rq_send_state = LUSTRE_IMP_FULL;
+ /* ptlrpc_request_pack already set timeout */
+ if (AT_OFF)
+ req->rq_timeout = ldlm_get_rq_timeout();
- /* We only send real blocking ASTs after the lock is granted */
- lock_res_and_lock(lock);
+ /* We only send real blocking ASTs after the lock is granted */
+ lock_res_and_lock(lock);
if (ldlm_is_ast_sent(lock)) {
body->lock_flags |= ldlm_flags_to_wire(LDLM_FL_AST_SENT);
/* Copy AST flags like LDLM_FL_DISCARD_DATA. */
body->lock_flags |= ldlm_flags_to_wire(lock->l_flags &
LDLM_FL_AST_MASK);
- /* We might get here prior to ldlm_handle_enqueue setting
- * LDLM_FL_CANCEL_ON_BLOCK flag. Then we will put this lock
- * into waiting list, but this is safe and similar code in
- * ldlm_handle_enqueue will call ldlm_lock_cancel() still,
- * that would not only cancel the lock, but will also remove
- * it from waiting list */
+ /*
+ * We might get here prior to ldlm_handle_enqueue setting
+ * LDLM_FL_CANCEL_ON_BLOCK flag. Then we will put this lock
+ * into waiting list, but this is safe and similar code in
+ * ldlm_handle_enqueue will call ldlm_lock_cancel() still,
+ * that would not only cancel the lock, but will also remove
+ * it from waiting list
+ */
if (ldlm_is_cancel_on_block(lock)) {
unlock_res_and_lock(lock);
ldlm_lock_cancel(lock);
lock_res_and_lock(lock);
} else {
/* start the lock-timeout clock */
- ldlm_add_waiting_lock(lock);
+ ldlm_add_waiting_lock(lock, ldlm_bl_timeout(lock));
/* Do not resend after lock callback timeout */
req->rq_delay_limit = ldlm_bl_timeout(lock);
req->rq_resend_cb = ldlm_update_resend;
}
- }
- unlock_res_and_lock(lock);
+ }
+ unlock_res_and_lock(lock);
- if (lock->l_export && lock->l_export->exp_nid_stats &&
- lock->l_export->exp_nid_stats->nid_ldlm_stats)
- lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
- LDLM_CP_CALLBACK - LDLM_FIRST_OPC);
+ if (lock->l_export && lock->l_export->exp_nid_stats &&
+ lock->l_export->exp_nid_stats->nid_ldlm_stats)
+ lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
+ LDLM_CP_CALLBACK - LDLM_FIRST_OPC);
rc = ldlm_ast_fini(req, arg, lock, instant_cancel);
*/
int ldlm_server_glimpse_ast(struct ldlm_lock *lock, void *data)
{
- struct ldlm_cb_set_arg *arg = data;
- struct ldlm_request *body;
- struct ptlrpc_request *req;
- struct ldlm_cb_async_args *ca;
- int rc;
- struct req_format *req_fmt;
- ENTRY;
+ struct ldlm_cb_set_arg *arg = data;
+ struct ldlm_request *body;
+ struct ptlrpc_request *req;
+ struct ldlm_cb_async_args *ca;
+ int rc;
+ struct req_format *req_fmt;
+
+ ENTRY;
- LASSERT(lock != NULL);
+ LASSERT(lock != NULL);
if (arg->gl_desc != NULL)
/* There is a glimpse descriptor to pack */
- req_fmt = &RQF_LDLM_GL_DESC_CALLBACK;
+ req_fmt = &RQF_LDLM_GL_CALLBACK_DESC;
else
req_fmt = &RQF_LDLM_GL_CALLBACK;
- req = ptlrpc_request_alloc_pack(lock->l_export->exp_imp_reverse,
+ req = ptlrpc_request_alloc_pack(lock->l_export->exp_imp_reverse,
req_fmt, LUSTRE_DLM_VERSION,
LDLM_GL_CALLBACK);
- if (req == NULL)
- RETURN(-ENOMEM);
+ if (req == NULL)
+ RETURN(-ENOMEM);
if (arg->gl_desc != NULL) {
/* copy the GL descriptor */
union ldlm_gl_desc *desc;
+
desc = req_capsule_client_get(&req->rq_pill, &RMF_DLM_GL_DESC);
*desc = *arg->gl_desc;
}
- body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
- body->lock_handle[0] = lock->l_remote_handle;
- ldlm_lock2desc(lock, &body->lock_desc);
+ body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
+ body->lock_handle[0] = lock->l_remote_handle;
+ ldlm_lock2desc(lock, &body->lock_desc);
- CLASSERT(sizeof(*ca) <= sizeof(req->rq_async_args));
- ca = ptlrpc_req_async_args(req);
+ ca = ptlrpc_req_async_args(ca, req);
ca->ca_set_arg = arg;
ca->ca_lock = lock;
- /* server namespace, doesn't need lock */
- req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
- ldlm_lvbo_size(lock));
- ptlrpc_request_set_replen(req);
-
- req->rq_send_state = LUSTRE_IMP_FULL;
- /* ptlrpc_request_alloc_pack already set timeout */
- if (AT_OFF)
- req->rq_timeout = ldlm_get_rq_timeout();
+ /* server namespace, doesn't need lock */
+ req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
+ ldlm_lvbo_size(lock));
+ ptlrpc_request_set_replen(req);
- lock->l_last_activity = cfs_time_current_sec();
+ req->rq_send_state = LUSTRE_IMP_FULL;
+ /* ptlrpc_request_alloc_pack already set timeout */
+ if (AT_OFF)
+ req->rq_timeout = ldlm_get_rq_timeout();
req->rq_interpret_reply = ldlm_cb_interpret;
- if (lock->l_export && lock->l_export->exp_nid_stats &&
- lock->l_export->exp_nid_stats->nid_ldlm_stats)
- lprocfs_counter_incr(lock->l_export->exp_nid_stats->nid_ldlm_stats,
- LDLM_GL_CALLBACK - LDLM_FIRST_OPC);
+ if (lock->l_export && lock->l_export->exp_nid_stats) {
+ struct nid_stat *nid_stats = lock->l_export->exp_nid_stats;
+
+ lprocfs_counter_incr(nid_stats->nid_ldlm_stats,
+ LDLM_GL_CALLBACK - LDLM_FIRST_OPC);
+ }
rc = ldlm_ast_fini(req, arg, lock, 0);
RETURN(rc);
}
+EXPORT_SYMBOL(ldlm_server_glimpse_ast);
int ldlm_glimpse_locks(struct ldlm_resource *res,
struct list_head *gl_work_list)
{
- int rc;
+ int rc;
+
ENTRY;
rc = ldlm_run_ast_work(ldlm_res_to_ns(res), gl_work_list,
LDLM_WORK_GL_AST);
if (rc == -ERESTART)
- ldlm_reprocess_all(res);
+ ldlm_reprocess_all(res, NULL);
RETURN(rc);
}
/* return LDLM lock associated with a lock callback request */
struct ldlm_lock *ldlm_request_lock(struct ptlrpc_request *req)
{
- struct ldlm_cb_async_args *ca;
- struct ldlm_lock *lock;
+ struct ldlm_cb_async_args *ca;
+ struct ldlm_lock *lock;
+
ENTRY;
- ca = ptlrpc_req_async_args(req);
+ ca = ptlrpc_req_async_args(ca, req);
lock = ca->ca_lock;
if (lock == NULL)
RETURN(ERR_PTR(-EFAULT));
}
EXPORT_SYMBOL(ldlm_request_lock);
-static void ldlm_svc_get_eopc(const struct ldlm_request *dlm_req,
- struct lprocfs_stats *srv_stats)
-{
- int lock_type = 0, op = 0;
-
- lock_type = dlm_req->lock_desc.l_resource.lr_type;
-
- switch (lock_type) {
- case LDLM_PLAIN:
- op = PTLRPC_LAST_CNTR + LDLM_PLAIN_ENQUEUE;
- break;
- case LDLM_EXTENT:
- if (dlm_req->lock_flags & LDLM_FL_HAS_INTENT)
- op = PTLRPC_LAST_CNTR + LDLM_GLIMPSE_ENQUEUE;
- else
- op = PTLRPC_LAST_CNTR + LDLM_EXTENT_ENQUEUE;
- break;
- case LDLM_FLOCK:
- op = PTLRPC_LAST_CNTR + LDLM_FLOCK_ENQUEUE;
- break;
- case LDLM_IBITS:
- op = PTLRPC_LAST_CNTR + LDLM_IBITS_ENQUEUE;
- break;
- default:
- op = 0;
- break;
- }
-
- if (op)
- lprocfs_counter_incr(srv_stats, op);
-
- return;
-}
-
/**
* Main server-side entry point into LDLM for enqueue. This is called by ptlrpc
* service threads to carry out client lock enqueueing requests.
void *cookie = NULL;
int rc = 0;
struct ldlm_resource *res = NULL;
+ const struct lu_env *env = req->rq_svc_thread->t_env;
+
ENTRY;
LDLM_DEBUG_NOLOCK("server-side enqueue handler START");
LASSERT(req->rq_export);
- if (ptlrpc_req2svc(req)->srv_stats != NULL)
+ /* for intent enqueue the stat will be updated inside intent policy */
+ if (ptlrpc_req2svc(req)->srv_stats != NULL &&
+ !(dlm_req->lock_flags & LDLM_FL_HAS_INTENT))
ldlm_svc_get_eopc(dlm_req, ptlrpc_req2svc(req)->srv_stats);
- if (req->rq_export && req->rq_export->exp_nid_stats &&
- req->rq_export->exp_nid_stats->nid_ldlm_stats)
- lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
- LDLM_ENQUEUE - LDLM_FIRST_OPC);
-
- if (unlikely(dlm_req->lock_desc.l_resource.lr_type < LDLM_MIN_TYPE ||
- dlm_req->lock_desc.l_resource.lr_type >= LDLM_MAX_TYPE)) {
- DEBUG_REQ(D_ERROR, req, "invalid lock request type %d",
- dlm_req->lock_desc.l_resource.lr_type);
- GOTO(out, rc = -EFAULT);
- }
-
- if (unlikely(dlm_req->lock_desc.l_req_mode <= LCK_MINMODE ||
- dlm_req->lock_desc.l_req_mode >= LCK_MAXMODE ||
- dlm_req->lock_desc.l_req_mode &
- (dlm_req->lock_desc.l_req_mode-1))) {
- DEBUG_REQ(D_ERROR, req, "invalid lock request mode %d",
- dlm_req->lock_desc.l_req_mode);
- GOTO(out, rc = -EFAULT);
- }
-
- if (exp_connect_flags(req->rq_export) & OBD_CONNECT_IBITS) {
- if (unlikely(dlm_req->lock_desc.l_resource.lr_type ==
- LDLM_PLAIN)) {
- DEBUG_REQ(D_ERROR, req,
- "PLAIN lock request from IBITS client?");
- GOTO(out, rc = -EPROTO);
- }
- } else if (unlikely(dlm_req->lock_desc.l_resource.lr_type ==
- LDLM_IBITS)) {
- DEBUG_REQ(D_ERROR, req,
- "IBITS lock request from unaware client?");
- GOTO(out, rc = -EPROTO);
- }
-
-#if 0
- /* FIXME this makes it impossible to use LDLM_PLAIN locks -- check
- against server's _CONNECT_SUPPORTED flags? (I don't want to use
- ibits for mgc/mgs) */
-
- /* INODEBITS_INTEROP: Perform conversion from plain lock to
- * inodebits lock if client does not support them. */
- if (!(exp_connect_flags(req->rq_export) & OBD_CONNECT_IBITS) &&
- (dlm_req->lock_desc.l_resource.lr_type == LDLM_PLAIN)) {
- dlm_req->lock_desc.l_resource.lr_type = LDLM_IBITS;
- dlm_req->lock_desc.l_policy_data.l_inodebits.bits =
- MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE;
- if (dlm_req->lock_desc.l_req_mode == LCK_PR)
- dlm_req->lock_desc.l_req_mode = LCK_CR;
- }
-#endif
+ if (req->rq_export && req->rq_export->exp_nid_stats &&
+ req->rq_export->exp_nid_stats->nid_ldlm_stats)
+ lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
+ LDLM_ENQUEUE - LDLM_FIRST_OPC);
+
+ if (unlikely(dlm_req->lock_desc.l_resource.lr_type < LDLM_MIN_TYPE ||
+ dlm_req->lock_desc.l_resource.lr_type >= LDLM_MAX_TYPE)) {
+ DEBUG_REQ(D_ERROR, req, "invalid lock request type %d",
+ dlm_req->lock_desc.l_resource.lr_type);
+ GOTO(out, rc = -EFAULT);
+ }
+
+ if (unlikely(dlm_req->lock_desc.l_req_mode <= LCK_MINMODE ||
+ dlm_req->lock_desc.l_req_mode >= LCK_MAXMODE ||
+ dlm_req->lock_desc.l_req_mode &
+ (dlm_req->lock_desc.l_req_mode-1))) {
+ DEBUG_REQ(D_ERROR, req, "invalid lock request mode %d",
+ dlm_req->lock_desc.l_req_mode);
+ GOTO(out, rc = -EFAULT);
+ }
if (unlikely((flags & LDLM_FL_REPLAY) ||
(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT))) {
- /* Find an existing lock in the per-export lock hash */
- /* In the function below, .hs_keycmp resolves to
- * ldlm_export_lock_keycmp() */
+ /* Find an existing lock in the per-export lock hash */
+ /*
+ * In the function below, .hs_keycmp resolves to
+ * ldlm_export_lock_keycmp()
+ */
/* coverity[overrun-buffer-val] */
- lock = cfs_hash_lookup(req->rq_export->exp_lock_hash,
- (void *)&dlm_req->lock_handle[0]);
- if (lock != NULL) {
- DEBUG_REQ(D_DLMTRACE, req, "found existing lock cookie "
- LPX64, lock->l_handle.h_cookie);
+ lock = cfs_hash_lookup(req->rq_export->exp_lock_hash,
+ (void *)&dlm_req->lock_handle[0]);
+ if (lock != NULL) {
+ DEBUG_REQ(D_DLMTRACE, req,
+ "found existing lock cookie %#llx",
+ lock->l_handle.h_cookie);
flags |= LDLM_FL_RESENT;
- GOTO(existing_lock, rc = 0);
+ GOTO(existing_lock, rc = 0);
}
} else {
if (ldlm_reclaim_full()) {
- DEBUG_REQ(D_DLMTRACE, req, "Too many granted locks, "
- "reject current enqueue request and let the "
- "client retry later.\n");
+ DEBUG_REQ(D_DLMTRACE, req,
+ "Too many granted locks, reject current enqueue request and let the client retry later");
GOTO(out, rc = -EINPROGRESS);
}
}
GOTO(out, rc);
}
- lock->l_remote_handle = dlm_req->lock_handle[0];
- LDLM_DEBUG(lock, "server-side enqueue handler, new lock created");
+ lock->l_remote_handle = dlm_req->lock_handle[0];
+ LDLM_DEBUG(lock, "server-side enqueue handler, new lock created");
- /* Initialize resource lvb but not for a lock being replayed since
+ /*
+ * Initialize resource lvb but not for a lock being replayed since
* Client already got lvb sent in this case.
* This must occur early since some policy methods assume resource
* lvb is available (lr_lvb_data != NULL).
}
}
- OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_ENQUEUE_BLOCKED, obd_timeout * 2);
- /* Don't enqueue a lock onto the export if it is been disonnected
- * due to eviction (bug 3822) or server umount (bug 24324).
- * Cancel it now instead. */
- if (req->rq_export->exp_disconnected) {
- LDLM_ERROR(lock, "lock on disconnected export %p",
- req->rq_export);
- GOTO(out, rc = -ENOTCONN);
- }
+ OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_ENQUEUE_BLOCKED, obd_timeout * 2);
+ /*
+ * Don't enqueue a lock onto the export if it is been disonnected
+ * due to eviction (b=3822) or server umount (b=24324).
+ * Cancel it now instead.
+ */
+ if (req->rq_export->exp_disconnected) {
+ LDLM_ERROR(lock, "lock on disconnected export %p",
+ req->rq_export);
+ GOTO(out, rc = -ENOTCONN);
+ }
- lock->l_export = class_export_lock_get(req->rq_export, lock);
- if (lock->l_export->exp_lock_hash)
- cfs_hash_add(lock->l_export->exp_lock_hash,
- &lock->l_remote_handle,
- &lock->l_exp_hash);
+ lock->l_export = class_export_lock_get(req->rq_export, lock);
+ if (lock->l_export->exp_lock_hash)
+ cfs_hash_add(lock->l_export->exp_lock_hash,
+ &lock->l_remote_handle,
+ &lock->l_exp_hash);
- /* Inherit the enqueue flags before the operation, because we do not
+ /*
+ * Inherit the enqueue flags before the operation, because we do not
* keep the res lock on return and next operations (BL AST) may proceed
- * without them. */
+ * without them.
+ */
lock->l_flags |= ldlm_flags_from_wire(dlm_req->lock_flags &
LDLM_FL_INHERIT_MASK);
-existing_lock:
- if (flags & LDLM_FL_HAS_INTENT) {
- /* In this case, the reply buffer is allocated deep in
- * local_lock_enqueue by the policy function. */
- cookie = req;
- } else {
- /* based on the assumption that lvb size never changes during
- * resource life time otherwise it need resource->lr_lock's
- * protection */
+ ldlm_convert_policy_to_local(req->rq_export,
+ dlm_req->lock_desc.l_resource.lr_type,
+ &dlm_req->lock_desc.l_policy_data,
+ &lock->l_policy_data);
+ if (dlm_req->lock_desc.l_resource.lr_type == LDLM_EXTENT)
+ lock->l_req_extent = lock->l_policy_data.l_extent;
+ else if (dlm_req->lock_desc.l_resource.lr_type == LDLM_IBITS)
+ lock->l_policy_data.l_inodebits.try_bits =
+ dlm_req->lock_desc.l_policy_data.l_inodebits.try_bits;
+
+existing_lock:
+ cookie = req;
+ if (!(flags & LDLM_FL_HAS_INTENT)) {
+ /* based on the assumption that lvb size never changes during
+ * resource life time otherwise it need resource->lr_lock's
+ * protection */
req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB,
RCL_SERVER, ldlm_lvbo_size(lock));
- if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE_EXTENT_ERR))
- GOTO(out, rc = -ENOMEM);
-
- rc = req_capsule_server_pack(&req->rq_pill);
- if (rc)
- GOTO(out, rc);
- }
+ if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE_EXTENT_ERR))
+ GOTO(out, rc = -ENOMEM);
- if (dlm_req->lock_desc.l_resource.lr_type != LDLM_PLAIN)
- ldlm_convert_policy_to_local(req->rq_export,
- dlm_req->lock_desc.l_resource.lr_type,
- &dlm_req->lock_desc.l_policy_data,
- &lock->l_policy_data);
- if (dlm_req->lock_desc.l_resource.lr_type == LDLM_EXTENT)
- lock->l_req_extent = lock->l_policy_data.l_extent;
+ rc = req_capsule_server_pack(&req->rq_pill);
+ if (rc)
+ GOTO(out, rc);
+ }
- err = ldlm_lock_enqueue(ns, &lock, cookie, &flags);
+ err = ldlm_lock_enqueue(env, ns, &lock, cookie, &flags);
if (err) {
if ((int)err < 0)
rc = (int)err;
GOTO(out, err);
}
- dlm_rep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
+ dlm_rep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
- ldlm_lock2desc(lock, &dlm_rep->lock_desc);
- ldlm_lock2handle(lock, &dlm_rep->lock_handle);
+ ldlm_lock2desc(lock, &dlm_rep->lock_desc);
+ ldlm_lock2handle(lock, &dlm_rep->lock_handle);
if (lock && lock->l_resource->lr_type == LDLM_EXTENT)
OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_BL_EVICT, 6);
- /* We never send a blocking AST until the lock is granted, but
- * we can tell it right now */
- lock_res_and_lock(lock);
+ /*
+ * We never send a blocking AST until the lock is granted, but
+ * we can tell it right now
+ */
+ lock_res_and_lock(lock);
- /* Now take into account flags to be inherited from original lock
- request both in reply to client and in our own lock flags. */
+ /*
+ * Now take into account flags to be inherited from original lock
+ * request both in reply to client and in our own lock flags.
+ */
dlm_rep->lock_flags = ldlm_flags_to_wire(flags);
lock->l_flags |= flags & LDLM_FL_INHERIT_MASK;
- /* Don't move a pending lock onto the export if it has already been
- * disconnected due to eviction (bug 5683) or server umount (bug 24324).
- * Cancel it now instead. */
- if (unlikely(req->rq_export->exp_disconnected ||
- OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE_OLD_EXPORT))) {
- LDLM_ERROR(lock, "lock on destroyed export %p", req->rq_export);
- rc = -ENOTCONN;
+ /*
+ * Don't move a pending lock onto the export if it has already been
+ * disconnected due to eviction (b=5683) or server umount (b=24324).
+ * Cancel it now instead.
+ */
+ if (unlikely(req->rq_export->exp_disconnected ||
+ OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE_OLD_EXPORT))) {
+ LDLM_ERROR(lock, "lock on destroyed export %p", req->rq_export);
+ rc = -ENOTCONN;
} else if (ldlm_is_ast_sent(lock)) {
+ /* fill lock desc for possible lock convert */
+ if (lock->l_blocking_lock &&
+ lock->l_resource->lr_type == LDLM_IBITS) {
+ struct ldlm_lock *bl_lock = lock->l_blocking_lock;
+ struct ldlm_lock_desc *rep_desc = &dlm_rep->lock_desc;
+
+ LDLM_DEBUG(lock,
+ "save blocking bits %llx in granted lock",
+ bl_lock->l_policy_data.l_inodebits.bits);
+ /*
+ * If lock is blocked then save blocking ibits
+ * in returned lock policy for the possible lock
+ * convert on a client.
+ */
+ rep_desc->l_policy_data.l_inodebits.cancel_bits =
+ bl_lock->l_policy_data.l_inodebits.bits;
+ }
dlm_rep->lock_flags |= ldlm_flags_to_wire(LDLM_FL_AST_SENT);
- if (lock->l_granted_mode == lock->l_req_mode) {
- /*
- * Only cancel lock if it was granted, because it would
- * be destroyed immediately and would never be granted
- * in the future, causing timeouts on client. Not
- * granted lock will be cancelled immediately after
- * sending completion AST.
- */
- if (dlm_rep->lock_flags & LDLM_FL_CANCEL_ON_BLOCK) {
- unlock_res_and_lock(lock);
- ldlm_lock_cancel(lock);
- lock_res_and_lock(lock);
- } else
- ldlm_add_waiting_lock(lock);
- }
- }
- /* Make sure we never ever grant usual metadata locks to liblustre
- clients */
- if ((dlm_req->lock_desc.l_resource.lr_type == LDLM_PLAIN ||
- dlm_req->lock_desc.l_resource.lr_type == LDLM_IBITS) &&
- req->rq_export->exp_libclient) {
- if (unlikely(!ldlm_is_cancel_on_block(lock) ||
- !(dlm_rep->lock_flags & LDLM_FL_CANCEL_ON_BLOCK))){
- CERROR("Granting sync lock to libclient. "
- "req fl %d, rep fl %d, lock fl "LPX64"\n",
- dlm_req->lock_flags, dlm_rep->lock_flags,
- lock->l_flags);
- LDLM_ERROR(lock, "sync lock");
- if (dlm_req->lock_flags & LDLM_FL_HAS_INTENT) {
- struct ldlm_intent *it;
-
- it = req_capsule_client_get(&req->rq_pill,
- &RMF_LDLM_INTENT);
- if (it != NULL) {
- CERROR("This is intent %s ("LPU64")\n",
- ldlm_it2str(it->opc), it->opc);
- }
+ if (ldlm_is_granted(lock)) {
+ /*
+ * Only cancel lock if it was granted, because it would
+ * be destroyed immediately and would never be granted
+ * in the future, causing timeouts on client. Not
+ * granted lock will be cancelled immediately after
+ * sending completion AST.
+ */
+ if (ldlm_is_cancel_on_block(lock)) {
+ unlock_res_and_lock(lock);
+ ldlm_lock_cancel(lock);
+ lock_res_and_lock(lock);
+ } else {
+ ldlm_add_waiting_lock(lock,
+ ldlm_bl_timeout(lock));
}
- }
- }
+ }
+ }
+ unlock_res_and_lock(lock);
- unlock_res_and_lock(lock);
+ EXIT;
+out:
+ req->rq_status = rc ?: err; /* return either error - b=11190 */
+ if (!req->rq_packed_final) {
+ err = lustre_pack_reply(req, 1, NULL, NULL);
+ if (rc == 0)
+ rc = err;
+ }
- EXIT;
- out:
- req->rq_status = rc ?: err; /* return either error - bug 11190 */
- if (!req->rq_packed_final) {
- err = lustre_pack_reply(req, 1, NULL, NULL);
- if (rc == 0)
- rc = err;
- }
-
- /* The LOCK_CHANGED code in ldlm_lock_enqueue depends on this
- * ldlm_reprocess_all. If this moves, revisit that code. -phil */
+ /*
+ * The LOCK_CHANGED code in ldlm_lock_enqueue depends on this
+ * ldlm_reprocess_all. If this moves, revisit that code. -phil
+ */
if (lock != NULL) {
- LDLM_DEBUG(lock, "server-side enqueue handler, sending reply"
- "(err=%d, rc=%d)", err, rc);
-
- if (rc == 0) {
- if (req_capsule_has_field(&req->rq_pill, &RMF_DLM_LVB,
- RCL_SERVER) &&
- ldlm_lvbo_size(lock) > 0) {
- void *buf;
- int buflen;
-
- buf = req_capsule_server_get(&req->rq_pill,
- &RMF_DLM_LVB);
- LASSERTF(buf != NULL, "req %p, lock %p\n",
- req, lock);
- buflen = req_capsule_get_size(&req->rq_pill,
- &RMF_DLM_LVB, RCL_SERVER);
- /* non-replayed lock, delayed lvb init may
- * need to be occur now */
- if ((buflen > 0) && !(flags & LDLM_FL_REPLAY)) {
- buflen = ldlm_lvbo_fill(lock, buf,
- buflen);
- if (buflen >= 0)
- req_capsule_shrink(
+ LDLM_DEBUG(lock,
+ "server-side enqueue handler, sending reply (err=%d, rc=%d)",
+ err, rc);
+
+ if (rc == 0 &&
+ req_capsule_has_field(&req->rq_pill, &RMF_DLM_LVB,
+ RCL_SERVER) &&
+ ldlm_lvbo_size(lock) > 0) {
+ void *buf;
+ int buflen;
+
+retry:
+ buf = req_capsule_server_get(&req->rq_pill,
+ &RMF_DLM_LVB);
+ LASSERTF(buf != NULL, "req %p, lock %p\n", req, lock);
+ buflen = req_capsule_get_size(&req->rq_pill,
+ &RMF_DLM_LVB, RCL_SERVER);
+ /*
+ * non-replayed lock, delayed lvb init may
+ * need to be occur now
+ */
+ if ((buflen > 0) && !(flags & LDLM_FL_REPLAY)) {
+ int rc2;
+
+ rc2 = ldlm_lvbo_fill(lock, buf, &buflen);
+ if (rc2 >= 0) {
+ req_capsule_shrink(&req->rq_pill,
+ &RMF_DLM_LVB,
+ rc2, RCL_SERVER);
+ } else if (rc2 == -ERANGE) {
+ rc2 = req_capsule_server_grow(
&req->rq_pill,
- &RMF_DLM_LVB,
- buflen, RCL_SERVER);
- else
- rc = buflen;
- } else if (flags & LDLM_FL_REPLAY) {
- /* no LVB resend upon replay */
- if (buflen > 0)
+ &RMF_DLM_LVB, buflen);
+ if (!rc2) {
+ goto retry;
+ } else {
+ /*
+ * if we can't grow the buffer,
+ * it's ok to return empty lvb
+ * to client.
+ */
req_capsule_shrink(
&req->rq_pill,
- &RMF_DLM_LVB,
- 0, RCL_SERVER);
- else
- rc = buflen;
+ &RMF_DLM_LVB, 0,
+ RCL_SERVER);
+ }
} else {
- rc = buflen;
+ rc = rc2;
}
+ } else if (flags & LDLM_FL_REPLAY) {
+ /* no LVB resend upon replay */
+ if (buflen > 0)
+ req_capsule_shrink(&req->rq_pill,
+ &RMF_DLM_LVB,
+ 0, RCL_SERVER);
+ else
+ rc = buflen;
+ } else {
+ rc = buflen;
}
}
}
}
- if (!err && dlm_req->lock_desc.l_resource.lr_type != LDLM_FLOCK)
- ldlm_reprocess_all(lock->l_resource);
+ if (!err && !ldlm_is_cbpending(lock) &&
+ dlm_req->lock_desc.l_resource.lr_type != LDLM_FLOCK)
+ ldlm_reprocess_all(lock->l_resource, lock);
- LDLM_LOCK_RELEASE(lock);
- }
+ LDLM_LOCK_RELEASE(lock);
+ }
- LDLM_DEBUG_NOLOCK("server-side enqueue handler END (lock %p, rc %d)",
- lock, rc);
+ LDLM_DEBUG_NOLOCK("server-side enqueue handler END (lock %p, rc %d)",
+ lock, rc);
- return rc;
+ return rc;
}
-/**
- * Old-style LDLM main entry point for server code enqueue.
+/*
+ * Clear the blocking lock, the race is possible between ldlm_handle_convert0()
+ * and ldlm_work_bl_ast_lock(), so this is done under lock with check for NULL.
*/
-int ldlm_handle_enqueue(struct ptlrpc_request *req,
- ldlm_completion_callback completion_callback,
- ldlm_blocking_callback blocking_callback,
- ldlm_glimpse_callback glimpse_callback)
-{
- struct ldlm_request *dlm_req;
- struct ldlm_callback_suite cbs = {
- .lcs_completion = completion_callback,
- .lcs_blocking = blocking_callback,
- .lcs_glimpse = glimpse_callback
- };
- int rc;
-
- dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
- if (dlm_req != NULL) {
- rc = ldlm_handle_enqueue0(req->rq_export->exp_obd->obd_namespace,
- req, dlm_req, &cbs);
- } else {
- rc = -EFAULT;
- }
- return rc;
+void ldlm_clear_blocking_lock(struct ldlm_lock *lock)
+{
+ if (lock->l_blocking_lock) {
+ LDLM_LOCK_RELEASE(lock->l_blocking_lock);
+ lock->l_blocking_lock = NULL;
+ }
}
-/**
- * Main LDLM entry point for server code to process lock conversion requests.
+/* A lock can be converted to new ibits or mode and should be considered
+ * as new lock. Clear all states related to a previous blocking AST
+ * processing so new conflicts will cause new blocking ASTs.
+ *
+ * This is used during lock convert below and lock downgrade to COS mode in
+ * ldlm_lock_mode_downgrade().
*/
-int ldlm_handle_convert0(struct ptlrpc_request *req,
- const struct ldlm_request *dlm_req)
-{
- struct ldlm_reply *dlm_rep;
- struct ldlm_lock *lock;
- int rc;
- ENTRY;
-
- if (req->rq_export && req->rq_export->exp_nid_stats &&
- req->rq_export->exp_nid_stats->nid_ldlm_stats)
- lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
- LDLM_CONVERT - LDLM_FIRST_OPC);
-
- rc = req_capsule_server_pack(&req->rq_pill);
- if (rc)
- RETURN(rc);
-
- dlm_rep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
- dlm_rep->lock_flags = dlm_req->lock_flags;
-
- lock = ldlm_handle2lock(&dlm_req->lock_handle[0]);
- if (!lock) {
- req->rq_status = LUSTRE_EINVAL;
- } else {
- void *res = NULL;
-
- LDLM_DEBUG(lock, "server-side convert handler START");
-
- res = ldlm_lock_convert(lock, dlm_req->lock_desc.l_req_mode,
- &dlm_rep->lock_flags);
- if (res) {
- if (ldlm_del_waiting_lock(lock))
- LDLM_DEBUG(lock, "converted waiting lock");
- req->rq_status = 0;
- } else {
- req->rq_status = LUSTRE_EDEADLK;
- }
- }
-
- if (lock) {
- if (!req->rq_status)
- ldlm_reprocess_all(lock->l_resource);
- LDLM_DEBUG(lock, "server-side convert handler END");
- LDLM_LOCK_PUT(lock);
- } else
- LDLM_DEBUG_NOLOCK("server-side convert handler END");
-
- RETURN(0);
+void ldlm_clear_blocking_data(struct ldlm_lock *lock)
+{
+ ldlm_clear_ast_sent(lock);
+ lock->l_bl_ast_run = 0;
+ ldlm_clear_blocking_lock(lock);
}
/**
- * Old-style main LDLM entry point for server code to process lock conversion
- * requests.
+ * Main LDLM entry point for server code to process lock conversion requests.
*/
-int ldlm_handle_convert(struct ptlrpc_request *req)
+int ldlm_handle_convert0(struct ptlrpc_request *req,
+ const struct ldlm_request *dlm_req)
{
- int rc;
- struct ldlm_request *dlm_req;
+ struct obd_export *exp = req->rq_export;
+ struct ldlm_reply *dlm_rep;
+ struct ldlm_lock *lock;
+ __u64 bits;
+ __u64 new_bits;
+ int rc;
- dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
- if (dlm_req != NULL) {
- rc = ldlm_handle_convert0(req, dlm_req);
- } else {
- CERROR ("Can't unpack dlm_req\n");
- rc = -EFAULT;
- }
- return rc;
+ ENTRY;
+
+ if (exp && exp->exp_nid_stats && exp->exp_nid_stats->nid_ldlm_stats)
+ lprocfs_counter_incr(exp->exp_nid_stats->nid_ldlm_stats,
+ LDLM_CONVERT - LDLM_FIRST_OPC);
+
+ rc = req_capsule_server_pack(&req->rq_pill);
+ if (rc)
+ RETURN(rc);
+
+ dlm_rep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
+ dlm_rep->lock_flags = dlm_req->lock_flags;
+
+ lock = ldlm_handle2lock(&dlm_req->lock_handle[0]);
+ if (!lock) {
+ LDLM_DEBUG_NOLOCK("server lock is canceled already");
+ req->rq_status = ELDLM_NO_LOCK_DATA;
+ RETURN(0);
+ }
+
+ LDLM_DEBUG(lock, "server-side convert handler START");
+
+ lock_res_and_lock(lock);
+ bits = lock->l_policy_data.l_inodebits.bits;
+ new_bits = dlm_req->lock_desc.l_policy_data.l_inodebits.bits;
+
+ if (ldlm_is_cancel(lock)) {
+ LDLM_DEBUG(lock, "convert on canceled lock!");
+ unlock_res_and_lock(lock);
+ GOTO(out_put, rc = ELDLM_NO_LOCK_DATA);
+ }
+
+ if (dlm_req->lock_desc.l_req_mode != lock->l_granted_mode) {
+ LDLM_ERROR(lock, "lock mode differs!");
+ unlock_res_and_lock(lock);
+ GOTO(out_put, rc = -EPROTO);
+ }
+
+ if (bits == new_bits) {
+ /*
+ * This can be valid situation if CONVERT RPCs are
+ * re-ordered. Just finish silently
+ */
+ LDLM_DEBUG(lock, "lock is converted already!");
+ unlock_res_and_lock(lock);
+ } else {
+ if (ldlm_is_waited(lock))
+ ldlm_del_waiting_lock(lock);
+
+ ldlm_clear_cbpending(lock);
+ lock->l_policy_data.l_inodebits.cancel_bits = 0;
+ ldlm_inodebits_drop(lock, bits & ~new_bits);
+
+ ldlm_clear_blocking_data(lock);
+ unlock_res_and_lock(lock);
+
+ ldlm_reprocess_all(lock->l_resource, NULL);
+ }
+
+ dlm_rep->lock_handle = lock->l_remote_handle;
+ ldlm_ibits_policy_local_to_wire(&lock->l_policy_data,
+ &dlm_rep->lock_desc.l_policy_data);
+ rc = ELDLM_OK;
+ EXIT;
+out_put:
+ LDLM_DEBUG(lock, "server-side convert handler END, rc = %d", rc);
+ LDLM_LOCK_PUT(lock);
+ req->rq_status = rc;
+ return 0;
}
/**
const struct ldlm_request *dlm_req,
int first, enum lustre_at_flags flags)
{
- struct ldlm_resource *res, *pres = NULL;
- struct ldlm_lock *lock;
- int i, count, done = 0;
- ENTRY;
+ struct ldlm_resource *res, *pres = NULL;
+ struct ldlm_lock *lock;
+ int i, count, done = 0;
+ unsigned int size;
+
+ ENTRY;
- count = dlm_req->lock_count ? dlm_req->lock_count : 1;
- if (first >= count)
- RETURN(0);
+ size = req_capsule_get_size(&req->rq_pill, &RMF_DLM_REQ, RCL_CLIENT);
+ if (size <= offsetof(struct ldlm_request, lock_handle) ||
+ (size - offsetof(struct ldlm_request, lock_handle)) /
+ sizeof(struct lustre_handle) < dlm_req->lock_count)
+ RETURN(0);
+
+ count = dlm_req->lock_count ? dlm_req->lock_count : 1;
+ if (first >= count)
+ RETURN(0);
if (count == 1 && dlm_req->lock_handle[0].cookie == 0)
RETURN(0);
- /* There is no lock on the server at the replay time,
- * skip lock cancelling to make replay tests to pass. */
- if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY)
- RETURN(0);
+ /*
+ * There is no lock on the server at the replay time,
+ * skip lock cancelling to make replay tests to pass.
+ */
+ if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY)
+ RETURN(0);
- LDLM_DEBUG_NOLOCK("server-side cancel handler START: %d locks, "
- "starting at %d", count, first);
+ LDLM_DEBUG_NOLOCK("server-side cancel handler START: %d locks, starting at %d",
+ count, first);
- for (i = first; i < count; i++) {
- lock = ldlm_handle2lock(&dlm_req->lock_handle[i]);
- if (!lock) {
- LDLM_DEBUG_NOLOCK("server-side cancel handler stale "
- "lock (cookie "LPU64")",
- dlm_req->lock_handle[i].cookie);
- continue;
- }
+ for (i = first; i < count; i++) {
+ lock = ldlm_handle2lock(&dlm_req->lock_handle[i]);
+ if (!lock) {
+ /* below message checked in replay-single.sh test_36 */
+ LDLM_DEBUG_NOLOCK("server-side cancel handler stale lock (cookie %llu)",
+ dlm_req->lock_handle[i].cookie);
+ continue;
+ }
- res = lock->l_resource;
- done++;
+ res = lock->l_resource;
+ done++;
- /* This code is an optimization to only attempt lock
+ /*
+ * This code is an optimization to only attempt lock
* granting on the resource (that could be CPU-expensive)
- * after we are done cancelling lock in that resource. */
- if (res != pres) {
- if (pres != NULL) {
- ldlm_reprocess_all(pres);
- LDLM_RESOURCE_DELREF(pres);
- ldlm_resource_putref(pres);
- }
- if (res != NULL) {
- ldlm_resource_getref(res);
- LDLM_RESOURCE_ADDREF(res);
- ldlm_res_lvbo_update(res, NULL, 1);
- }
- pres = res;
- }
-
- if ((flags & LATF_STATS) && ldlm_is_ast_sent(lock)) {
- long delay = cfs_time_sub(cfs_time_current_sec(),
- lock->l_last_activity);
- LDLM_DEBUG(lock, "server cancels blocked lock after "
- CFS_DURATION_T"s", delay);
+ * after we are done cancelling lock in that resource.
+ */
+ if (res != pres) {
+ if (pres != NULL) {
+ ldlm_reprocess_all(pres, NULL);
+ LDLM_RESOURCE_DELREF(pres);
+ ldlm_resource_putref(pres);
+ }
+ if (res != NULL) {
+ ldlm_resource_getref(res);
+ LDLM_RESOURCE_ADDREF(res);
+
+ if (!ldlm_is_discard_data(lock))
+ ldlm_lvbo_update(res, lock,
+ NULL, 1);
+ }
+ pres = res;
+ }
+
+ if ((flags & LATF_STATS) && ldlm_is_ast_sent(lock) &&
+ lock->l_blast_sent != 0) {
+ timeout_t delay = 0;
+
+ if (ktime_get_real_seconds() > lock->l_blast_sent)
+ delay = ktime_get_real_seconds() -
+ lock->l_blast_sent;
+ LDLM_DEBUG(lock,
+ "server cancels blocked lock after %ds",
+ delay);
at_measured(&lock->l_export->exp_bl_lock_at, delay);
}
- ldlm_lock_cancel(lock);
- LDLM_LOCK_PUT(lock);
- }
- if (pres != NULL) {
- ldlm_reprocess_all(pres);
- LDLM_RESOURCE_DELREF(pres);
- ldlm_resource_putref(pres);
- }
- LDLM_DEBUG_NOLOCK("server-side cancel handler END");
- RETURN(done);
+ ldlm_lock_cancel(lock);
+ LDLM_LOCK_PUT(lock);
+ }
+ if (pres != NULL) {
+ ldlm_reprocess_all(pres, NULL);
+ LDLM_RESOURCE_DELREF(pres);
+ ldlm_resource_putref(pres);
+ }
+ LDLM_DEBUG_NOLOCK("server-side cancel handler END");
+ RETURN(done);
}
EXPORT_SYMBOL(ldlm_request_cancel);
*/
int ldlm_handle_cancel(struct ptlrpc_request *req)
{
- struct ldlm_request *dlm_req;
- int rc;
- ENTRY;
+ struct ldlm_request *dlm_req;
+ int rc;
+
+ ENTRY;
- dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
- if (dlm_req == NULL) {
- CDEBUG(D_INFO, "bad request buffer for cancel\n");
- RETURN(-EFAULT);
- }
+ dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
+ if (dlm_req == NULL) {
+ CDEBUG(D_INFO, "bad request buffer for cancel\n");
+ RETURN(-EFAULT);
+ }
+
+ if (req_capsule_get_size(&req->rq_pill, &RMF_DLM_REQ, RCL_CLIENT) <
+ offsetof(struct ldlm_request, lock_handle[1]))
+ RETURN(-EPROTO);
- if (req->rq_export && req->rq_export->exp_nid_stats &&
- req->rq_export->exp_nid_stats->nid_ldlm_stats)
- lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
- LDLM_CANCEL - LDLM_FIRST_OPC);
+ if (req->rq_export && req->rq_export->exp_nid_stats &&
+ req->rq_export->exp_nid_stats->nid_ldlm_stats)
+ lprocfs_counter_incr(req->rq_export->exp_nid_stats->nid_ldlm_stats,
+ LDLM_CANCEL - LDLM_FIRST_OPC);
- rc = req_capsule_server_pack(&req->rq_pill);
- if (rc)
- RETURN(rc);
+ rc = req_capsule_server_pack(&req->rq_pill);
+ if (rc)
+ RETURN(rc);
if (!ldlm_request_cancel(req, dlm_req, 0, LATF_STATS))
req->rq_status = LUSTRE_ESTALE;
- RETURN(ptlrpc_reply(req));
+ RETURN(ptlrpc_reply(req));
}
#endif /* HAVE_SERVER_SUPPORT */
/**
+ * Server may pass additional information about blocking lock.
+ * For IBITS locks it is conflicting bits which can be used for
+ * lock convert instead of cancel.
+ */
+void ldlm_bl_desc2lock(const struct ldlm_lock_desc *ld, struct ldlm_lock *lock)
+{
+ struct ldlm_namespace *ns = ldlm_lock_to_ns(lock);
+
+ check_res_locked(lock->l_resource);
+ if (ns_is_client(ns) && ld &&
+ (lock->l_resource->lr_type == LDLM_IBITS)) {
+ /*
+ * Lock description contains policy of blocking lock,
+ * and its cancel_bits is used to pass conflicting bits.
+ * NOTE: ld can be NULL or can be not NULL but zeroed if
+ * passed from ldlm_bl_thread_blwi(), check below used bits
+ * in ld to make sure it is valid description.
+ *
+ * If server may replace lock resource keeping the same cookie,
+ * never use cancel bits from different resource, full cancel
+ * is to be used.
+ */
+ if (ld->l_policy_data.l_inodebits.cancel_bits &&
+ ldlm_res_eq(&ld->l_resource.lr_name,
+ &lock->l_resource->lr_name) &&
+ !(ldlm_is_cbpending(lock) &&
+ lock->l_policy_data.l_inodebits.cancel_bits == 0)) {
+ /* always combine conflicting ibits */
+ lock->l_policy_data.l_inodebits.cancel_bits |=
+ ld->l_policy_data.l_inodebits.cancel_bits;
+ } else {
+ /* If cancel_bits are not obtained or
+ * if the lock is already CBPENDING and
+ * has no cancel_bits set
+ * - the full lock is to be cancelled
+ */
+ lock->l_policy_data.l_inodebits.cancel_bits = 0;
+ }
+ }
+}
+
+/**
* Callback handler for receiving incoming blocking ASTs.
*
* This can only happen on client side.
*/
void ldlm_handle_bl_callback(struct ldlm_namespace *ns,
- struct ldlm_lock_desc *ld, struct ldlm_lock *lock)
+ struct ldlm_lock_desc *ld, struct ldlm_lock *lock)
{
- int do_ast;
- ENTRY;
+ int do_ast;
- LDLM_DEBUG(lock, "client blocking AST callback handler");
+ ENTRY;
+
+ LDLM_DEBUG(lock, "client blocking AST callback handler");
- lock_res_and_lock(lock);
+ lock_res_and_lock(lock);
+
+ /* get extra information from desc if any */
+ ldlm_bl_desc2lock(ld, lock);
ldlm_set_cbpending(lock);
- if (ldlm_is_cancel_on_block(lock))
- ldlm_set_cancel(lock);
+ do_ast = (!lock->l_readers && !lock->l_writers);
+ unlock_res_and_lock(lock);
- do_ast = (!lock->l_readers && !lock->l_writers);
- unlock_res_and_lock(lock);
+ if (do_ast) {
+ CDEBUG(D_DLMTRACE,
+ "Lock %p already unused, calling callback (%p)\n",
+ lock, lock->l_blocking_ast);
+ if (lock->l_blocking_ast != NULL)
+ lock->l_blocking_ast(lock, ld, lock->l_ast_data,
+ LDLM_CB_BLOCKING);
+ } else {
+ CDEBUG(D_DLMTRACE,
+ "Lock %p is referenced, will be cancelled later\n",
+ lock);
+ }
- if (do_ast) {
- CDEBUG(D_DLMTRACE, "Lock %p already unused, calling callback (%p)\n",
- lock, lock->l_blocking_ast);
- if (lock->l_blocking_ast != NULL)
- lock->l_blocking_ast(lock, ld, lock->l_ast_data,
- LDLM_CB_BLOCKING);
- } else {
- CDEBUG(D_DLMTRACE, "Lock %p is referenced, will be cancelled later\n",
- lock);
- }
+ LDLM_DEBUG(lock, "client blocking callback handler END");
+ LDLM_LOCK_RELEASE(lock);
+ EXIT;
+}
+
+static int ldlm_callback_reply(struct ptlrpc_request *req, int rc)
+{
+ if (req->rq_no_reply)
+ return 0;
- LDLM_DEBUG(lock, "client blocking callback handler END");
- LDLM_LOCK_RELEASE(lock);
- EXIT;
+ req->rq_status = rc;
+ if (!req->rq_packed_final) {
+ rc = lustre_pack_reply(req, 1, NULL, NULL);
+ if (rc)
+ return rc;
+ }
+ return ptlrpc_reply(req);
}
/**
*
* This only can happen on client side.
*/
-static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
+static int ldlm_handle_cp_callback(struct ptlrpc_request *req,
struct ldlm_namespace *ns,
struct ldlm_request *dlm_req,
struct ldlm_lock *lock)
{
- struct list_head ast_list;
+ LIST_HEAD(ast_list);
int lvb_len;
int rc = 0;
+
ENTRY;
LDLM_DEBUG(lock, "client completion callback handler START");
- INIT_LIST_HEAD(&ast_list);
if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_BL_CB_RACE)) {
- int to = cfs_time_seconds(1);
+ long to = cfs_time_seconds(1);
+
+ ldlm_callback_reply(req, 0);
+
while (to > 0) {
- set_current_state(TASK_INTERRUPTIBLE);
- schedule_timeout(to);
- if (lock->l_granted_mode == lock->l_req_mode ||
+ schedule_timeout_interruptible(to);
+ if (ldlm_is_granted(lock) ||
ldlm_is_destroyed(lock))
break;
}
LASSERT(lock->l_lvb_data != NULL);
if (unlikely(lock->l_lvb_len < lvb_len)) {
- LDLM_ERROR(lock, "Replied LVB is larger than "
- "expectation, expected = %d, "
- "replied = %d",
+ LDLM_ERROR(lock,
+ "Replied LVB is larger than expectation, expected = %d, replied = %d",
lock->l_lvb_len, lvb_len);
GOTO(out, rc = -EINVAL);
}
}
lock_res_and_lock(lock);
+
+ if (!ldlm_res_eq(&dlm_req->lock_desc.l_resource.lr_name,
+ &lock->l_resource->lr_name)) {
+ ldlm_resource_unlink_lock(lock);
+ unlock_res_and_lock(lock);
+ rc = ldlm_lock_change_resource(ns, lock,
+ &dlm_req->lock_desc.l_resource.lr_name);
+ if (rc < 0) {
+ LDLM_ERROR(lock, "Failed to allocate resource");
+ GOTO(out, rc);
+ }
+ LDLM_DEBUG(lock, "completion AST, new resource");
+ lock_res_and_lock(lock);
+ }
+
+ if (ldlm_is_failed(lock)) {
+ unlock_res_and_lock(lock);
+ LDLM_LOCK_RELEASE(lock);
+ RETURN(-EINVAL);
+ }
+
if (ldlm_is_destroyed(lock) ||
- lock->l_granted_mode == lock->l_req_mode) {
- /* bug 11300: the lock has already been granted */
+ ldlm_is_granted(lock)) {
+ /* b=11300: the lock has already been granted */
unlock_res_and_lock(lock);
LDLM_DEBUG(lock, "Double grant race happened");
GOTO(out, rc = 0);
}
- /* If we receive the completion AST before the actual enqueue returned,
- * then we might need to switch lock modes, resources, or extents. */
+ /*
+ * If we receive the completion AST before the actual enqueue returned,
+ * then we might need to switch lock modes, resources, or extents.
+ */
if (dlm_req->lock_desc.l_granted_mode != lock->l_req_mode) {
lock->l_req_mode = dlm_req->lock_desc.l_granted_mode;
LDLM_DEBUG(lock, "completion AST, new lock mode");
LDLM_DEBUG(lock, "completion AST, new policy data");
}
- ldlm_resource_unlink_lock(lock);
- if (memcmp(&dlm_req->lock_desc.l_resource.lr_name,
- &lock->l_resource->lr_name,
- sizeof(lock->l_resource->lr_name)) != 0) {
- unlock_res_and_lock(lock);
- rc = ldlm_lock_change_resource(ns, lock,
- &dlm_req->lock_desc.l_resource.lr_name);
- if (rc < 0) {
- LDLM_ERROR(lock, "Failed to allocate resource");
- GOTO(out, rc);
- }
- LDLM_DEBUG(lock, "completion AST, new resource");
- CERROR("change resource!\n");
- lock_res_and_lock(lock);
- }
-
- if (dlm_req->lock_flags & LDLM_FL_AST_SENT) {
- /* BL_AST locks are not needed in LRU.
- * Let ldlm_cancel_lru() be fast. */
- ldlm_lock_remove_from_lru(lock);
+ ldlm_resource_unlink_lock(lock);
+
+ if (dlm_req->lock_flags & LDLM_FL_AST_SENT) {
+ /*
+ * BL_AST locks are not needed in LRU.
+ * Let ldlm_cancel_lru() be fast.
+ */
+ ldlm_lock_remove_from_lru(lock);
+ ldlm_bl_desc2lock(&dlm_req->lock_desc, lock);
lock->l_flags |= LDLM_FL_CBPENDING | LDLM_FL_BL_AST;
- LDLM_DEBUG(lock, "completion AST includes blocking AST");
- }
+ LDLM_DEBUG(lock, "completion AST includes blocking AST");
+ }
if (lock->l_lvb_len > 0) {
rc = ldlm_fill_lvb(lock, &req->rq_pill, RCL_CLIENT,
}
}
- ldlm_grant_lock(lock, &ast_list);
- unlock_res_and_lock(lock);
+ ldlm_grant_lock(lock, &ast_list);
+ unlock_res_and_lock(lock);
- LDLM_DEBUG(lock, "callback handler finished, about to run_ast_work");
+ LDLM_DEBUG(lock, "callback handler finished, about to run_ast_work");
- /* Let Enqueue to call osc_lock_upcall() and initialize
- * l_ast_data */
- OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_ENQ_RACE, 2);
+ /*
+ * Let Enqueue to call osc_lock_upcall() and initialize
+ * l_ast_data
+ */
+ OBD_FAIL_TIMEOUT(OBD_FAIL_OSC_CP_ENQ_RACE, 2);
- ldlm_run_ast_work(ns, &ast_list, LDLM_WORK_CP_AST);
+ ldlm_run_ast_work(ns, &ast_list, LDLM_WORK_CP_AST);
- LDLM_DEBUG_NOLOCK("client completion callback handler END (lock %p)",
- lock);
+ LDLM_DEBUG_NOLOCK("client completion callback handler END (lock %p)",
+ lock);
GOTO(out, rc);
out:
wake_up(&lock->l_waitq);
}
LDLM_LOCK_RELEASE(lock);
+
+ return 0;
}
/**
* long time.
*/
static void ldlm_handle_gl_callback(struct ptlrpc_request *req,
- struct ldlm_namespace *ns,
- struct ldlm_request *dlm_req,
- struct ldlm_lock *lock)
+ struct ldlm_namespace *ns,
+ struct ldlm_request *dlm_req,
+ struct ldlm_lock *lock)
{
- int rc = -ENOSYS;
- ENTRY;
-
- LDLM_DEBUG(lock, "client glimpse AST callback handler");
+ struct ldlm_lock_desc *ld = &dlm_req->lock_desc;
+ int rc = -ENOSYS;
- if (lock->l_glimpse_ast != NULL)
- rc = lock->l_glimpse_ast(lock, req);
+ ENTRY;
- if (req->rq_repmsg != NULL) {
- ptlrpc_reply(req);
- } else {
- req->rq_status = rc;
- ptlrpc_error(req);
- }
+ LDLM_DEBUG(lock, "client glimpse AST callback handler");
- lock_res_and_lock(lock);
- if (lock->l_granted_mode == LCK_PW &&
- !lock->l_readers && !lock->l_writers &&
- cfs_time_after(cfs_time_current(),
- cfs_time_add(lock->l_last_used,
- cfs_time_seconds(10)))) {
- unlock_res_and_lock(lock);
- if (ldlm_bl_to_thread_lock(ns, NULL, lock))
- ldlm_handle_bl_callback(ns, NULL, lock);
+ if (lock->l_glimpse_ast != NULL)
+ rc = lock->l_glimpse_ast(lock, req);
- EXIT;
- return;
- }
- unlock_res_and_lock(lock);
- LDLM_LOCK_RELEASE(lock);
- EXIT;
-}
+ if (req->rq_repmsg != NULL) {
+ ptlrpc_reply(req);
+ } else {
+ req->rq_status = rc;
+ ptlrpc_error(req);
+ }
-static int ldlm_callback_reply(struct ptlrpc_request *req, int rc)
-{
- if (req->rq_no_reply)
- return 0;
+ lock_res_and_lock(lock);
+ if (lock->l_granted_mode == LCK_PW &&
+ !lock->l_readers && !lock->l_writers &&
+ ktime_after(ktime_get(),
+ ktime_add(lock->l_last_used, ns->ns_dirty_age_limit))) {
+ unlock_res_and_lock(lock);
- req->rq_status = rc;
- if (!req->rq_packed_final) {
- rc = lustre_pack_reply(req, 1, NULL, NULL);
- if (rc)
- return rc;
- }
- return ptlrpc_reply(req);
+ /* For MDS glimpse it is always DOM lock, set corresponding
+ * cancel_bits to perform lock convert if needed
+ */
+ if (lock->l_resource->lr_type == LDLM_IBITS)
+ ld->l_policy_data.l_inodebits.cancel_bits =
+ MDS_INODELOCK_DOM;
+ if (ldlm_bl_to_thread_lock(ns, ld, lock))
+ ldlm_handle_bl_callback(ns, ld, lock);
+
+ EXIT;
+ return;
+ }
+ unlock_res_and_lock(lock);
+ LDLM_LOCK_RELEASE(lock);
+ EXIT;
}
static int __ldlm_bl_to_thread(struct ldlm_bl_work_item *blwi,
enum ldlm_cancel_flags cancel_flags)
{
struct ldlm_bl_pool *blp = ldlm_state->ldlm_bl_pool;
+
ENTRY;
spin_lock(&blp->blp_lock);
wake_up(&blp->blp_waitq);
- /* can not check blwi->blwi_flags as blwi could be already freed in
- LCF_ASYNC mode */
+ /*
+ * can not check blwi->blwi_flags as blwi could be already freed in
+ * LCF_ASYNC mode
+ */
if (!(cancel_flags & LCF_ASYNC))
wait_for_completion(&blwi->blwi_comp);
init_completion(&blwi->blwi_comp);
INIT_LIST_HEAD(&blwi->blwi_head);
- if (memory_pressure_get())
- blwi->blwi_mem_pressure = 1;
+ if (current->flags & PF_MEMALLOC)
+ blwi->blwi_mem_pressure = 1;
- blwi->blwi_ns = ns;
+ blwi->blwi_ns = ns;
blwi->blwi_flags = cancel_flags;
- if (ld != NULL)
- blwi->blwi_ld = *ld;
- if (count) {
- list_add(&blwi->blwi_head, cancels);
- list_del_init(cancels);
- blwi->blwi_count = count;
- } else {
- blwi->blwi_lock = lock;
- }
+ if (ld != NULL)
+ blwi->blwi_ld = *ld;
+ if (count) {
+ list_splice_init(cancels, &blwi->blwi_head);
+ blwi->blwi_count = count;
+ } else {
+ blwi->blwi_lock = lock;
+ }
}
/**
RETURN(__ldlm_bl_to_thread(blwi, cancel_flags));
} else {
- /* if it is synchronous call do minimum mem alloc, as it could
+ /*
+ * if it is synchronous call do minimum mem alloc, as it could
* be triggered from kernel shrinker
*/
struct ldlm_bl_work_item blwi;
/* Setinfo coming from Server (eg MDT) to Client (eg MDC)! */
static int ldlm_handle_setinfo(struct ptlrpc_request *req)
{
- struct obd_device *obd = req->rq_export->exp_obd;
- char *key;
- void *val;
- int keylen, vallen;
- int rc = -ENOSYS;
- ENTRY;
-
- DEBUG_REQ(D_HSM, req, "%s: handle setinfo\n", obd->obd_name);
-
- req_capsule_set(&req->rq_pill, &RQF_OBD_SET_INFO);
-
- key = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
- if (key == NULL) {
- DEBUG_REQ(D_IOCTL, req, "no set_info key");
- RETURN(-EFAULT);
- }
- keylen = req_capsule_get_size(&req->rq_pill, &RMF_SETINFO_KEY,
- RCL_CLIENT);
- val = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_VAL);
- if (val == NULL) {
- DEBUG_REQ(D_IOCTL, req, "no set_info val");
- RETURN(-EFAULT);
- }
- vallen = req_capsule_get_size(&req->rq_pill, &RMF_SETINFO_VAL,
- RCL_CLIENT);
-
- /* We are responsible for swabbing contents of val */
-
- if (KEY_IS(KEY_HSM_COPYTOOL_SEND))
- /* Pass it on to mdc (the "export" in this case) */
- rc = obd_set_info_async(req->rq_svc_thread->t_env,
- req->rq_export,
- sizeof(KEY_HSM_COPYTOOL_SEND),
- KEY_HSM_COPYTOOL_SEND,
- vallen, val, NULL);
- else
- DEBUG_REQ(D_WARNING, req, "ignoring unknown key %s", key);
-
- return rc;
+ struct obd_device *obd = req->rq_export->exp_obd;
+ char *key;
+ void *val;
+ int keylen, vallen;
+ int rc = -ENOSYS;
+
+ ENTRY;
+
+ DEBUG_REQ(D_HSM, req, "%s: handle setinfo", obd->obd_name);
+
+ req_capsule_set(&req->rq_pill, &RQF_OBD_SET_INFO);
+
+ key = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
+ if (key == NULL) {
+ DEBUG_REQ(D_IOCTL, req, "no set_info key");
+ RETURN(-EFAULT);
+ }
+ keylen = req_capsule_get_size(&req->rq_pill, &RMF_SETINFO_KEY,
+ RCL_CLIENT);
+ val = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_VAL);
+ if (val == NULL) {
+ DEBUG_REQ(D_IOCTL, req, "no set_info val");
+ RETURN(-EFAULT);
+ }
+ vallen = req_capsule_get_size(&req->rq_pill, &RMF_SETINFO_VAL,
+ RCL_CLIENT);
+
+ /* We are responsible for swabbing contents of val */
+
+ if (KEY_IS(KEY_HSM_COPYTOOL_SEND))
+ /* Pass it on to mdc (the "export" in this case) */
+ rc = obd_set_info_async(req->rq_svc_thread->t_env,
+ req->rq_export,
+ sizeof(KEY_HSM_COPYTOOL_SEND),
+ KEY_HSM_COPYTOOL_SEND,
+ vallen, val, NULL);
+ else
+ DEBUG_REQ(D_WARNING, req, "ignoring unknown key '%s'", key);
+
+ return rc;
}
static inline void ldlm_callback_errmsg(struct ptlrpc_request *req,
- const char *msg, int rc,
- struct lustre_handle *handle)
+ const char *msg, int rc,
+ const struct lustre_handle *handle)
{
- DEBUG_REQ((req->rq_no_reply || rc) ? D_WARNING : D_DLMTRACE, req,
- "%s: [nid %s] [rc %d] [lock "LPX64"]",
- msg, libcfs_id2str(req->rq_peer), rc,
- handle ? handle->cookie : 0);
- if (req->rq_no_reply)
- CWARN("No reply was sent, maybe cause bug 21636.\n");
- else if (rc)
- CWARN("Send reply failed, maybe cause bug 21636.\n");
+ DEBUG_REQ((req->rq_no_reply || rc) ? D_WARNING : D_DLMTRACE, req,
+ "%s, NID=%s lock=%#llx: rc = %d",
+ msg, libcfs_id2str(req->rq_peer),
+ handle ? handle->cookie : 0, rc);
+ if (req->rq_no_reply)
+ CWARN("No reply was sent, maybe cause b=21636.\n");
+ else if (rc)
+ CWARN("Send reply failed, maybe cause b=21636.\n");
}
/* TODO: handle requests in a similar way as MDT: see mdt_handle_common() */
static int ldlm_callback_handler(struct ptlrpc_request *req)
{
- struct ldlm_namespace *ns;
- struct ldlm_request *dlm_req;
- struct ldlm_lock *lock;
- int rc;
- ENTRY;
+ struct ldlm_namespace *ns;
+ struct ldlm_request *dlm_req;
+ struct ldlm_lock *lock;
+ int rc;
- /* Requests arrive in sender's byte order. The ptlrpc service
- * handler has already checked and, if necessary, byte-swapped the
- * incoming request message body, but I am responsible for the
- * message buffers. */
+ ENTRY;
- /* do nothing for sec context finalize */
- if (lustre_msg_get_opc(req->rq_reqmsg) == SEC_CTX_FINI)
- RETURN(0);
+ /*
+ * Requests arrive in sender's byte order. The ptlrpc service
+ * handler has already checked and, if necessary, byte-swapped the
+ * incoming request message body, but I am responsible for the
+ * message buffers.
+ */
- req_capsule_init(&req->rq_pill, req, RCL_SERVER);
+ /* do nothing for sec context finalize */
+ if (lustre_msg_get_opc(req->rq_reqmsg) == SEC_CTX_FINI)
+ RETURN(0);
- if (req->rq_export == NULL) {
- rc = ldlm_callback_reply(req, -ENOTCONN);
- ldlm_callback_errmsg(req, "Operate on unconnected server",
- rc, NULL);
- RETURN(0);
- }
+ req_capsule_init(&req->rq_pill, req, RCL_SERVER);
- LASSERT(req->rq_export != NULL);
- LASSERT(req->rq_export->exp_obd != NULL);
+ if (req->rq_export == NULL) {
+ rc = ldlm_callback_reply(req, -ENOTCONN);
+ ldlm_callback_errmsg(req, "Operate on unconnected server",
+ rc, NULL);
+ RETURN(0);
+ }
+
+ LASSERT(req->rq_export != NULL);
+ LASSERT(req->rq_export->exp_obd != NULL);
switch (lustre_msg_get_opc(req->rq_reqmsg)) {
case LDLM_BL_CALLBACK:
if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_GL_CALLBACK_NET))
RETURN(0);
break;
- case LDLM_SET_INFO:
- rc = ldlm_handle_setinfo(req);
- ldlm_callback_reply(req, rc);
- RETURN(0);
- case LLOG_ORIGIN_HANDLE_CREATE:
- req_capsule_set(&req->rq_pill, &RQF_LLOG_ORIGIN_HANDLE_CREATE);
- if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOGD_NET))
- RETURN(0);
- rc = llog_origin_handle_open(req);
- ldlm_callback_reply(req, rc);
- RETURN(0);
- case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
- req_capsule_set(&req->rq_pill,
- &RQF_LLOG_ORIGIN_HANDLE_NEXT_BLOCK);
- if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOGD_NET))
- RETURN(0);
- rc = llog_origin_handle_next_block(req);
- ldlm_callback_reply(req, rc);
- RETURN(0);
- case LLOG_ORIGIN_HANDLE_READ_HEADER:
- req_capsule_set(&req->rq_pill,
- &RQF_LLOG_ORIGIN_HANDLE_READ_HEADER);
- if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOGD_NET))
- RETURN(0);
- rc = llog_origin_handle_read_header(req);
- ldlm_callback_reply(req, rc);
- RETURN(0);
- case LLOG_ORIGIN_HANDLE_CLOSE:
- if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOGD_NET))
- RETURN(0);
- rc = llog_origin_handle_close(req);
- ldlm_callback_reply(req, rc);
- RETURN(0);
- default:
- CERROR("unknown opcode %u\n",
- lustre_msg_get_opc(req->rq_reqmsg));
- ldlm_callback_reply(req, -EPROTO);
- RETURN(0);
- }
-
- ns = req->rq_export->exp_obd->obd_namespace;
- LASSERT(ns != NULL);
-
- req_capsule_set(&req->rq_pill, &RQF_LDLM_CALLBACK);
-
- dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
- if (dlm_req == NULL) {
- rc = ldlm_callback_reply(req, -EPROTO);
- ldlm_callback_errmsg(req, "Operate without parameter", rc,
- NULL);
- RETURN(0);
- }
-
- /* Force a known safe race, send a cancel to the server for a lock
- * which the server has already started a blocking callback on. */
- if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_BL_CB_RACE) &&
- lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK) {
+ case LDLM_SET_INFO:
+ rc = ldlm_handle_setinfo(req);
+ ldlm_callback_reply(req, rc);
+ RETURN(0);
+ default:
+ CERROR("unknown opcode %u\n",
+ lustre_msg_get_opc(req->rq_reqmsg));
+ ldlm_callback_reply(req, -EPROTO);
+ RETURN(0);
+ }
+
+ ns = req->rq_export->exp_obd->obd_namespace;
+ LASSERT(ns != NULL);
+
+ req_capsule_set(&req->rq_pill, &RQF_LDLM_CALLBACK);
+
+ dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
+ if (dlm_req == NULL) {
+ rc = ldlm_callback_reply(req, -EPROTO);
+ ldlm_callback_errmsg(req, "Operate without parameter", rc,
+ NULL);
+ RETURN(0);
+ }
+
+ /*
+ * Force a known safe race, send a cancel to the server for a lock
+ * which the server has already started a blocking callback on.
+ */
+ if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_BL_CB_RACE) &&
+ lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK) {
rc = ldlm_cli_cancel(&dlm_req->lock_handle[0], 0);
- if (rc < 0)
- CERROR("ldlm_cli_cancel: %d\n", rc);
- }
-
- lock = ldlm_handle2lock_long(&dlm_req->lock_handle[0], 0);
- if (!lock) {
- CDEBUG(D_DLMTRACE, "callback on lock "LPX64" - lock "
- "disappeared\n", dlm_req->lock_handle[0].cookie);
- rc = ldlm_callback_reply(req, -EINVAL);
- ldlm_callback_errmsg(req, "Operate with invalid parameter", rc,
- &dlm_req->lock_handle[0]);
- RETURN(0);
- }
+ if (rc < 0)
+ CERROR("ldlm_cli_cancel: %d\n", rc);
+ }
+
+ lock = ldlm_handle2lock_long(&dlm_req->lock_handle[0], 0);
+ if (!lock) {
+ CDEBUG(D_DLMTRACE,
+ "callback on lock %#llx - lock disappeared\n",
+ dlm_req->lock_handle[0].cookie);
+ rc = ldlm_callback_reply(req, -EINVAL);
+ ldlm_callback_errmsg(req, "Operate with invalid parameter", rc,
+ &dlm_req->lock_handle[0]);
+ RETURN(0);
+ }
if (ldlm_is_fail_loc(lock) &&
- lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK)
- OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE);
+ lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK)
+ OBD_RACE(OBD_FAIL_LDLM_CP_BL_RACE);
- /* Copy hints/flags (e.g. LDLM_FL_DISCARD_DATA) from AST. */
- lock_res_and_lock(lock);
+ /* Copy hints/flags (e.g. LDLM_FL_DISCARD_DATA) from AST. */
+ lock_res_and_lock(lock);
lock->l_flags |= ldlm_flags_from_wire(dlm_req->lock_flags &
LDLM_FL_AST_MASK);
- if (lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK) {
- /* If somebody cancels lock and cache is already dropped,
- * or lock is failed before cp_ast received on client,
- * we can tell the server we have no lock. Otherwise, we
- * should send cancel after dropping the cache. */
+ if (lustre_msg_get_opc(req->rq_reqmsg) == LDLM_BL_CALLBACK) {
+ /*
+ * If somebody cancels lock and cache is already dropped,
+ * or lock is failed before cp_ast received on client,
+ * we can tell the server we have no lock. Otherwise, we
+ * should send cancel after dropping the cache.
+ */
if ((ldlm_is_canceling(lock) && ldlm_is_bl_done(lock)) ||
- ldlm_is_failed(lock)) {
- LDLM_DEBUG(lock, "callback on lock "
- LPX64" - lock disappeared\n",
- dlm_req->lock_handle[0].cookie);
- unlock_res_and_lock(lock);
- LDLM_LOCK_RELEASE(lock);
- rc = ldlm_callback_reply(req, -EINVAL);
- ldlm_callback_errmsg(req, "Operate on stale lock", rc,
- &dlm_req->lock_handle[0]);
- RETURN(0);
- }
- /* BL_AST locks are not needed in LRU.
- * Let ldlm_cancel_lru() be fast. */
- ldlm_lock_remove_from_lru(lock);
+ ldlm_is_failed(lock)) {
+ LDLM_DEBUG(lock,
+ "callback on lock %llx - lock disappeared",
+ dlm_req->lock_handle[0].cookie);
+ unlock_res_and_lock(lock);
+ LDLM_LOCK_RELEASE(lock);
+ rc = ldlm_callback_reply(req, -EINVAL);
+ ldlm_callback_errmsg(req, "Operate on stale lock", rc,
+ &dlm_req->lock_handle[0]);
+ RETURN(0);
+ }
+ /*
+ * BL_AST locks are not needed in LRU.
+ * Let ldlm_cancel_lru() be fast.
+ */
+ ldlm_lock_remove_from_lru(lock);
ldlm_set_bl_ast(lock);
- }
- unlock_res_and_lock(lock);
-
- /* We want the ost thread to get this reply so that it can respond
- * to ost requests (write cache writeback) that might be triggered
- * in the callback.
- *
- * But we'd also like to be able to indicate in the reply that we're
- * cancelling right now, because it's unused, or have an intent result
- * in the reply, so we might have to push the responsibility for sending
- * the reply down into the AST handlers, alas. */
-
- switch (lustre_msg_get_opc(req->rq_reqmsg)) {
- case LDLM_BL_CALLBACK:
- CDEBUG(D_INODE, "blocking ast\n");
- req_capsule_extend(&req->rq_pill, &RQF_LDLM_BL_CALLBACK);
+ }
+ unlock_res_and_lock(lock);
+
+ /*
+ * We want the ost thread to get this reply so that it can respond
+ * to ost requests (write cache writeback) that might be triggered
+ * in the callback.
+ *
+ * But we'd also like to be able to indicate in the reply that we're
+ * cancelling right now, because it's unused, or have an intent result
+ * in the reply, so we might have to push the responsibility for sending
+ * the reply down into the AST handlers, alas.
+ */
+
+ switch (lustre_msg_get_opc(req->rq_reqmsg)) {
+ case LDLM_BL_CALLBACK:
+ CDEBUG(D_INODE, "blocking ast\n");
+ req_capsule_extend(&req->rq_pill, &RQF_LDLM_BL_CALLBACK);
if (!ldlm_is_cancel_on_block(lock)) {
- rc = ldlm_callback_reply(req, 0);
- if (req->rq_no_reply || rc)
- ldlm_callback_errmsg(req, "Normal process", rc,
- &dlm_req->lock_handle[0]);
- }
- if (ldlm_bl_to_thread_lock(ns, &dlm_req->lock_desc, lock))
- ldlm_handle_bl_callback(ns, &dlm_req->lock_desc, lock);
- break;
- case LDLM_CP_CALLBACK:
- CDEBUG(D_INODE, "completion ast\n");
- req_capsule_extend(&req->rq_pill, &RQF_LDLM_CP_CALLBACK);
- ldlm_callback_reply(req, 0);
- ldlm_handle_cp_callback(req, ns, dlm_req, lock);
- break;
- case LDLM_GL_CALLBACK:
- CDEBUG(D_INODE, "glimpse ast\n");
- req_capsule_extend(&req->rq_pill, &RQF_LDLM_GL_CALLBACK);
- ldlm_handle_gl_callback(req, ns, dlm_req, lock);
- break;
- default:
- LBUG(); /* checked above */
- }
-
- RETURN(0);
+ rc = ldlm_callback_reply(req, 0);
+ if (req->rq_no_reply || rc)
+ ldlm_callback_errmsg(req, "Normal process", rc,
+ &dlm_req->lock_handle[0]);
+ }
+ if (ldlm_bl_to_thread_lock(ns, &dlm_req->lock_desc, lock))
+ ldlm_handle_bl_callback(ns, &dlm_req->lock_desc, lock);
+ break;
+ case LDLM_CP_CALLBACK:
+ CDEBUG(D_INODE, "completion ast\n");
+ req_capsule_extend(&req->rq_pill, &RQF_LDLM_CP_CALLBACK);
+ rc = ldlm_handle_cp_callback(req, ns, dlm_req, lock);
+ if (!OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_BL_CB_RACE))
+ ldlm_callback_reply(req, rc);
+ break;
+ case LDLM_GL_CALLBACK:
+ CDEBUG(D_INODE, "glimpse ast\n");
+ req_capsule_extend(&req->rq_pill, &RQF_LDLM_GL_CALLBACK);
+ ldlm_handle_gl_callback(req, ns, dlm_req, lock);
+ break;
+ default:
+ LBUG(); /* checked above */
+ }
+
+ RETURN(0);
}
#ifdef HAVE_SERVER_SUPPORT
*/
static int ldlm_cancel_handler(struct ptlrpc_request *req)
{
- int rc;
- ENTRY;
-
- /* Requests arrive in sender's byte order. The ptlrpc service
- * handler has already checked and, if necessary, byte-swapped the
- * incoming request message body, but I am responsible for the
- * message buffers. */
-
- req_capsule_init(&req->rq_pill, req, RCL_SERVER);
-
- if (req->rq_export == NULL) {
- struct ldlm_request *dlm_req;
-
- CERROR("%s from %s arrived at %lu with bad export cookie "
- LPU64"\n",
- ll_opcode2str(lustre_msg_get_opc(req->rq_reqmsg)),
- libcfs_nid2str(req->rq_peer.nid),
- req->rq_arrival_time.tv_sec,
- lustre_msg_get_handle(req->rq_reqmsg)->cookie);
-
- if (lustre_msg_get_opc(req->rq_reqmsg) == LDLM_CANCEL) {
- req_capsule_set(&req->rq_pill, &RQF_LDLM_CALLBACK);
- dlm_req = req_capsule_client_get(&req->rq_pill,
- &RMF_DLM_REQ);
- if (dlm_req != NULL)
- ldlm_lock_dump_handle(D_ERROR,
- &dlm_req->lock_handle[0]);
- }
- ldlm_callback_reply(req, -ENOTCONN);
- RETURN(0);
- }
-
- switch (lustre_msg_get_opc(req->rq_reqmsg)) {
-
- /* XXX FIXME move this back to mds/handler.c, bug 249 */
- case LDLM_CANCEL:
- req_capsule_set(&req->rq_pill, &RQF_LDLM_CANCEL);
- CDEBUG(D_INODE, "cancel\n");
+ int rc;
+
+ ENTRY;
+
+ /*
+ * Requests arrive in sender's byte order. The ptlrpc service
+ * handler has already checked and, if necessary, byte-swapped the
+ * incoming request message body, but I am responsible for the
+ * message buffers.
+ */
+
+ req_capsule_init(&req->rq_pill, req, RCL_SERVER);
+
+ if (req->rq_export == NULL) {
+ struct ldlm_request *dlm_req;
+
+ CERROR("%s from %s arrived at %llu with bad export cookie %llu\n",
+ ll_opcode2str(lustre_msg_get_opc(req->rq_reqmsg)),
+ libcfs_nid2str(req->rq_peer.nid),
+ (unsigned long long)req->rq_arrival_time.tv_sec,
+ lustre_msg_get_handle(req->rq_reqmsg)->cookie);
+
+ if (lustre_msg_get_opc(req->rq_reqmsg) == LDLM_CANCEL) {
+ req_capsule_set(&req->rq_pill, &RQF_LDLM_CALLBACK);
+ dlm_req = req_capsule_client_get(&req->rq_pill,
+ &RMF_DLM_REQ);
+ if (dlm_req != NULL)
+ ldlm_lock_dump_handle(D_ERROR,
+ &dlm_req->lock_handle[0]);
+ }
+ ldlm_callback_reply(req, -ENOTCONN);
+ RETURN(0);
+ }
+
+ switch (lustre_msg_get_opc(req->rq_reqmsg)) {
+ /* XXX FIXME move this back to mds/handler.c, b=249 */
+ case LDLM_CANCEL:
+ req_capsule_set(&req->rq_pill, &RQF_LDLM_CANCEL);
+ CDEBUG(D_INODE, "cancel\n");
if (CFS_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_NET) ||
CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_CANCEL_RESEND) ||
CFS_FAIL_CHECK(OBD_FAIL_LDLM_BL_EVICT))
RETURN(0);
- rc = ldlm_handle_cancel(req);
- if (rc)
- break;
- RETURN(0);
- default:
- CERROR("invalid opcode %d\n",
- lustre_msg_get_opc(req->rq_reqmsg));
- req_capsule_set(&req->rq_pill, &RQF_LDLM_CALLBACK);
- ldlm_callback_reply(req, -EINVAL);
- }
+ rc = ldlm_handle_cancel(req);
+ break;
+ case LDLM_CONVERT:
+ {
+ struct ldlm_request *dlm_req;
+
+ req_capsule_set(&req->rq_pill, &RQF_LDLM_CONVERT);
+ CDEBUG(D_INODE, "convert\n");
- RETURN(0);
+ dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
+ if (dlm_req == NULL) {
+ CDEBUG(D_INFO, "bad request buffer for cancel\n");
+ rc = ldlm_callback_reply(req, -EPROTO);
+ } else {
+ req->rq_status = ldlm_handle_convert0(req, dlm_req);
+ rc = ptlrpc_reply(req);
+ }
+ break;
+ }
+ default:
+ CERROR("invalid opcode %d\n",
+ lustre_msg_get_opc(req->rq_reqmsg));
+ req_capsule_set(&req->rq_pill, &RQF_LDLM_CALLBACK);
+ rc = ldlm_callback_reply(req, -EINVAL);
+ }
+
+ RETURN(rc);
}
static int ldlm_cancel_hpreq_lock_match(struct ptlrpc_request *req,
- struct ldlm_lock *lock)
+ struct ldlm_lock *lock)
{
- struct ldlm_request *dlm_req;
- struct lustre_handle lockh;
- int rc = 0;
- int i;
- ENTRY;
-
- dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
- if (dlm_req == NULL)
- RETURN(0);
+ struct ldlm_request *dlm_req;
+ struct lustre_handle lockh;
+ int rc = 0;
+ int i;
- ldlm_lock2handle(lock, &lockh);
- for (i = 0; i < dlm_req->lock_count; i++) {
- if (lustre_handle_equal(&dlm_req->lock_handle[i],
- &lockh)) {
- DEBUG_REQ(D_RPCTRACE, req,
- "Prio raised by lock "LPX64".", lockh.cookie);
+ ENTRY;
- rc = 1;
- break;
- }
- }
+ dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
+ if (dlm_req == NULL)
+ RETURN(0);
- RETURN(rc);
+ ldlm_lock2handle(lock, &lockh);
+ for (i = 0; i < dlm_req->lock_count; i++) {
+ if (lustre_handle_equal(&dlm_req->lock_handle[i],
+ &lockh)) {
+ DEBUG_REQ(D_RPCTRACE, req,
+ "Prio raised by lock %#llx", lockh.cookie);
+ rc = 1;
+ break;
+ }
+ }
+ RETURN(rc);
}
static int ldlm_cancel_hpreq_check(struct ptlrpc_request *req)
{
- struct ldlm_request *dlm_req;
- int rc = 0;
- int i;
- ENTRY;
+ struct ldlm_request *dlm_req;
+ int rc = 0;
+ int i;
+ unsigned int size;
+
+ ENTRY;
+
+ /* no prolong in recovery */
+ if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY)
+ RETURN(0);
- /* no prolong in recovery */
- if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY)
- RETURN(0);
+ dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
+ if (dlm_req == NULL)
+ RETURN(-EFAULT);
- dlm_req = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
- if (dlm_req == NULL)
- RETURN(-EFAULT);
+ size = req_capsule_get_size(&req->rq_pill, &RMF_DLM_REQ, RCL_CLIENT);
+ if (size <= offsetof(struct ldlm_request, lock_handle) ||
+ (size - offsetof(struct ldlm_request, lock_handle)) /
+ sizeof(struct lustre_handle) < dlm_req->lock_count)
+ RETURN(-EPROTO);
- for (i = 0; i < dlm_req->lock_count; i++) {
- struct ldlm_lock *lock;
+ for (i = 0; i < dlm_req->lock_count; i++) {
+ struct ldlm_lock *lock;
- lock = ldlm_handle2lock(&dlm_req->lock_handle[i]);
- if (lock == NULL)
- continue;
+ lock = ldlm_handle2lock(&dlm_req->lock_handle[i]);
+ if (lock == NULL)
+ continue;
rc = ldlm_is_ast_sent(lock) ? 1 : 0;
- if (rc)
- LDLM_DEBUG(lock, "hpreq cancel lock");
- LDLM_LOCK_PUT(lock);
+ if (rc)
+ LDLM_DEBUG(lock, "hpreq cancel/convert lock");
+ LDLM_LOCK_PUT(lock);
- if (rc)
- break;
- }
+ if (rc)
+ break;
+ }
- RETURN(rc);
+ RETURN(rc);
}
static struct ptlrpc_hpreq_ops ldlm_cancel_hpreq_ops = {
- .hpreq_lock_match = ldlm_cancel_hpreq_lock_match,
+ .hpreq_lock_match = ldlm_cancel_hpreq_lock_match,
.hpreq_check = ldlm_cancel_hpreq_check,
.hpreq_fini = NULL,
};
static int ldlm_hpreq_handler(struct ptlrpc_request *req)
{
- ENTRY;
+ ENTRY;
- req_capsule_init(&req->rq_pill, req, RCL_SERVER);
+ req_capsule_init(&req->rq_pill, req, RCL_SERVER);
- if (req->rq_export == NULL)
- RETURN(0);
+ if (req->rq_export == NULL)
+ RETURN(0);
- if (LDLM_CANCEL == lustre_msg_get_opc(req->rq_reqmsg)) {
- req_capsule_set(&req->rq_pill, &RQF_LDLM_CANCEL);
- req->rq_ops = &ldlm_cancel_hpreq_ops;
- }
- RETURN(0);
+ if (LDLM_CANCEL == lustre_msg_get_opc(req->rq_reqmsg)) {
+ req_capsule_set(&req->rq_pill, &RQF_LDLM_CANCEL);
+ req->rq_ops = &ldlm_cancel_hpreq_ops;
+ } else if (LDLM_CONVERT == lustre_msg_get_opc(req->rq_reqmsg)) {
+ req_capsule_set(&req->rq_pill, &RQF_LDLM_CONVERT);
+ req->rq_ops = &ldlm_cancel_hpreq_ops;
+ }
+ RETURN(0);
}
static int ldlm_revoke_lock_cb(struct cfs_hash *hs, struct cfs_hash_bd *bd,
struct hlist_node *hnode, void *data)
{
- struct list_head *rpc_list = data;
- struct ldlm_lock *lock = cfs_hash_object(hs, hnode);
+ struct list_head *rpc_list = data;
+ struct ldlm_lock *lock = cfs_hash_object(hs, hnode);
- lock_res_and_lock(lock);
+ lock_res_and_lock(lock);
- if (lock->l_req_mode != lock->l_granted_mode) {
- unlock_res_and_lock(lock);
- return 0;
- }
+ if (!ldlm_is_granted(lock)) {
+ unlock_res_and_lock(lock);
+ return 0;
+ }
- LASSERT(lock->l_resource);
- if (lock->l_resource->lr_type != LDLM_IBITS &&
- lock->l_resource->lr_type != LDLM_PLAIN) {
- unlock_res_and_lock(lock);
- return 0;
- }
+ LASSERT(lock->l_resource);
+ if (lock->l_resource->lr_type != LDLM_IBITS &&
+ lock->l_resource->lr_type != LDLM_PLAIN) {
+ unlock_res_and_lock(lock);
+ return 0;
+ }
if (ldlm_is_ast_sent(lock)) {
- unlock_res_and_lock(lock);
- return 0;
- }
+ unlock_res_and_lock(lock);
+ return 0;
+ }
- LASSERT(lock->l_blocking_ast);
- LASSERT(!lock->l_blocking_lock);
+ LASSERT(lock->l_blocking_ast);
+ LASSERT(!lock->l_blocking_lock);
ldlm_set_ast_sent(lock);
- if (lock->l_export && lock->l_export->exp_lock_hash) {
- /* NB: it's safe to call cfs_hash_del() even lock isn't
- * in exp_lock_hash. */
- /* In the function below, .hs_keycmp resolves to
- * ldlm_export_lock_keycmp() */
+ if (lock->l_export && lock->l_export->exp_lock_hash) {
+ /*
+ * NB: it's safe to call cfs_hash_del() even lock isn't
+ * in exp_lock_hash.
+ */
+ /*
+ * In the function below, .hs_keycmp resolves to
+ * ldlm_export_lock_keycmp()
+ */
/* coverity[overrun-buffer-val] */
cfs_hash_del(lock->l_export->exp_lock_hash,
&lock->l_remote_handle, &lock->l_exp_hash);
}
list_add_tail(&lock->l_rk_ast, rpc_list);
- LDLM_LOCK_GET(lock);
+ LDLM_LOCK_GET(lock);
- unlock_res_and_lock(lock);
- return 0;
+ unlock_res_and_lock(lock);
+ return 0;
}
void ldlm_revoke_export_locks(struct obd_export *exp)
{
- struct list_head rpc_list;
- ENTRY;
+ LIST_HEAD(rpc_list);
- INIT_LIST_HEAD(&rpc_list);
- cfs_hash_for_each_empty(exp->exp_lock_hash,
- ldlm_revoke_lock_cb, &rpc_list);
- ldlm_run_ast_work(exp->exp_obd->obd_namespace, &rpc_list,
- LDLM_WORK_REVOKE_AST);
+ ENTRY;
+
+ cfs_hash_for_each_nolock(exp->exp_lock_hash,
+ ldlm_revoke_lock_cb, &rpc_list, 0);
+ ldlm_run_ast_work(exp->exp_obd->obd_namespace, &rpc_list,
+ LDLM_WORK_REVOKE_AST);
- EXIT;
+ EXIT;
}
EXPORT_SYMBOL(ldlm_revoke_export_locks);
#endif /* HAVE_SERVER_SUPPORT */
struct obd_export **p_exp)
{
struct ldlm_bl_work_item *blwi = NULL;
- static unsigned int num_bl = 0;
+ static unsigned int num_bl;
static unsigned int num_stale;
int num_th = atomic_read(&blp->blp_num_threads);
if (num_th == 1 || ++num_stale < num_th) {
spin_unlock(&blp->blp_lock);
return 1;
- } else {
- num_stale = 0;
}
+ num_stale = 0;
}
/* process a request from the blp_list at least every blp_num_threads */
static int ldlm_bl_thread_blwi(struct ldlm_bl_pool *blp,
struct ldlm_bl_work_item *blwi)
{
+ /* '1' for consistency with code that checks !mpflag to restore */
+ unsigned int mpflags = 1;
+
ENTRY;
if (blwi->blwi_ns == NULL)
RETURN(LDLM_ITER_STOP);
if (blwi->blwi_mem_pressure)
- memory_pressure_set();
+ mpflags = memalloc_noreclaim_save();
OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_PAUSE_CANCEL2, 4);
if (blwi->blwi_count) {
int count;
- /* The special case when we cancel locks in lru
+ /*
+ * The special case when we cancel locks in lru
* asynchronously, we pass the list of locks here.
* Thus locks are marked LDLM_FL_CANCELING, but NOT
- * canceled locally yet. */
+ * canceled locally yet.
+ */
count = ldlm_cli_cancel_list_local(&blwi->blwi_head,
blwi->blwi_count,
LCF_BL_AST);
blwi->blwi_lock);
}
if (blwi->blwi_mem_pressure)
- memory_pressure_clr();
+ memalloc_noreclaim_restore(mpflags);
if (blwi->blwi_flags & LCF_ASYNC)
OBD_FREE(blwi, sizeof(*blwi));
struct obd_export *exp)
{
int num;
+
ENTRY;
OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_BL_EVICT, 4);
*/
static int ldlm_bl_thread_main(void *arg)
{
- struct ldlm_bl_pool *blp;
+ struct lu_env *env;
+ struct ldlm_bl_pool *blp;
struct ldlm_bl_thread_data *bltd = arg;
- ENTRY;
+ int rc;
+
+ ENTRY;
+
+ OBD_ALLOC_PTR(env);
+ if (!env)
+ RETURN(-ENOMEM);
+ rc = lu_env_init(env, LCT_DT_THREAD);
+ if (rc)
+ GOTO(out_env, rc);
+ rc = lu_env_add(env);
+ if (rc)
+ GOTO(out_env_fini, rc);
blp = bltd->bltd_blp;
/* cannot use bltd after this, it is only on caller's stack */
while (1) {
- struct l_wait_info lwi = { 0 };
struct ldlm_bl_work_item *blwi = NULL;
struct obd_export *exp = NULL;
int rc;
rc = ldlm_bl_get_work(blp, &blwi, &exp);
if (rc == 0)
- l_wait_event_exclusive(blp->blp_waitq,
- ldlm_bl_get_work(blp, &blwi,
- &exp),
- &lwi);
+ wait_event_idle_exclusive(blp->blp_waitq,
+ ldlm_bl_get_work(blp, &blwi,
+ &exp));
atomic_inc(&blp->blp_busy_threads);
if (ldlm_bl_thread_need_create(blp, blwi))
if (rc == LDLM_ITER_STOP)
break;
+
+ /*
+ * If there are many namespaces, we will not sleep waiting for
+ * work, and must do a cond_resched to avoid holding the CPU
+ * for too long
+ */
+ cond_resched();
}
atomic_dec(&blp->blp_num_threads);
complete(&blp->blp_comp);
- RETURN(0);
+
+ lu_env_remove(env);
+out_env_fini:
+ lu_env_fini(env);
+out_env:
+ OBD_FREE_PTR(env);
+ RETURN(rc);
}
int ldlm_get_ref(void)
{
- int rc = 0;
- ENTRY;
+ int rc = 0;
+
+ ENTRY;
mutex_lock(&ldlm_ref_mutex);
- if (++ldlm_refcount == 1) {
- rc = ldlm_setup();
- if (rc)
- ldlm_refcount--;
- }
+ if (++ldlm_refcount == 1) {
+ rc = ldlm_setup();
+ if (rc)
+ ldlm_refcount--;
+ }
mutex_unlock(&ldlm_ref_mutex);
- RETURN(rc);
+ RETURN(rc);
}
void ldlm_put_ref(void)
{
- ENTRY;
+ ENTRY;
mutex_lock(&ldlm_ref_mutex);
- if (ldlm_refcount == 1) {
- int rc = ldlm_cleanup();
- if (rc)
- CERROR("ldlm_cleanup failed: %d\n", rc);
- else
- ldlm_refcount--;
- } else {
- ldlm_refcount--;
- }
+ if (ldlm_refcount == 1) {
+ int rc = ldlm_cleanup();
+
+ if (rc)
+ CERROR("ldlm_cleanup failed: %d\n", rc);
+ else
+ ldlm_refcount--;
+ } else {
+ ldlm_refcount--;
+ }
mutex_unlock(&ldlm_ref_mutex);
- EXIT;
+ EXIT;
}
/*
* Export handle<->lock hash operations.
*/
static unsigned
-ldlm_export_lock_hash(struct cfs_hash *hs, const void *key, unsigned mask)
+ldlm_export_lock_hash(struct cfs_hash *hs, const void *key, unsigned int mask)
{
- return cfs_hash_u64_hash(((struct lustre_handle *)key)->cookie, mask);
+ return cfs_hash_u64_hash(((struct lustre_handle *)key)->cookie, mask);
}
static void *
ldlm_export_lock_key(struct hlist_node *hnode)
{
- struct ldlm_lock *lock;
+ struct ldlm_lock *lock;
lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
- return &lock->l_remote_handle;
+ return &lock->l_remote_handle;
}
static void
ldlm_export_lock_keycpy(struct hlist_node *hnode, void *key)
{
- struct ldlm_lock *lock;
+ struct ldlm_lock *lock;
lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
- lock->l_remote_handle = *(struct lustre_handle *)key;
+ lock->l_remote_handle = *(struct lustre_handle *)key;
}
static int
ldlm_export_lock_keycmp(const void *key, struct hlist_node *hnode)
{
- return lustre_handle_equal(ldlm_export_lock_key(hnode), key);
+ return lustre_handle_equal(ldlm_export_lock_key(hnode), key);
}
static void *
static void
ldlm_export_lock_get(struct cfs_hash *hs, struct hlist_node *hnode)
{
- struct ldlm_lock *lock;
+ struct ldlm_lock *lock;
lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
- LDLM_LOCK_GET(lock);
+ LDLM_LOCK_GET(lock);
}
static void
ldlm_export_lock_put(struct cfs_hash *hs, struct hlist_node *hnode)
{
- struct ldlm_lock *lock;
+ struct ldlm_lock *lock;
lock = hlist_entry(hnode, struct ldlm_lock, l_exp_hash);
- LDLM_LOCK_RELEASE(lock);
+ LDLM_LOCK_RELEASE(lock);
}
static struct cfs_hash_ops ldlm_export_lock_ops = {
- .hs_hash = ldlm_export_lock_hash,
- .hs_key = ldlm_export_lock_key,
- .hs_keycmp = ldlm_export_lock_keycmp,
- .hs_keycpy = ldlm_export_lock_keycpy,
- .hs_object = ldlm_export_lock_object,
- .hs_get = ldlm_export_lock_get,
- .hs_put = ldlm_export_lock_put,
- .hs_put_locked = ldlm_export_lock_put,
+ .hs_hash = ldlm_export_lock_hash,
+ .hs_key = ldlm_export_lock_key,
+ .hs_keycmp = ldlm_export_lock_keycmp,
+ .hs_keycpy = ldlm_export_lock_keycpy,
+ .hs_object = ldlm_export_lock_object,
+ .hs_get = ldlm_export_lock_get,
+ .hs_put = ldlm_export_lock_put,
+ .hs_put_locked = ldlm_export_lock_put,
};
int ldlm_init_export(struct obd_export *exp)
{
int rc;
- ENTRY;
- exp->exp_lock_hash =
- cfs_hash_create(obd_uuid2str(&exp->exp_client_uuid),
- HASH_EXP_LOCK_CUR_BITS,
- HASH_EXP_LOCK_MAX_BITS,
- HASH_EXP_LOCK_BKT_BITS, 0,
- CFS_HASH_MIN_THETA, CFS_HASH_MAX_THETA,
- &ldlm_export_lock_ops,
- CFS_HASH_DEFAULT | CFS_HASH_REHASH_KEY |
- CFS_HASH_NBLK_CHANGE);
+ ENTRY;
+
+ exp->exp_lock_hash =
+ cfs_hash_create(obd_uuid2str(&exp->exp_client_uuid),
+ HASH_EXP_LOCK_CUR_BITS,
+ HASH_EXP_LOCK_MAX_BITS,
+ HASH_EXP_LOCK_BKT_BITS, 0,
+ CFS_HASH_MIN_THETA, CFS_HASH_MAX_THETA,
+ &ldlm_export_lock_ops,
+ CFS_HASH_DEFAULT | CFS_HASH_REHASH_KEY |
+ CFS_HASH_NBLK_CHANGE);
- if (!exp->exp_lock_hash)
- RETURN(-ENOMEM);
+ if (!exp->exp_lock_hash)
+ RETURN(-ENOMEM);
rc = ldlm_init_flock_export(exp);
if (rc)
GOTO(err, rc);
- RETURN(0);
+ RETURN(0);
err:
ldlm_destroy_export(exp);
RETURN(rc);
void ldlm_destroy_export(struct obd_export *exp)
{
- ENTRY;
- cfs_hash_putref(exp->exp_lock_hash);
- exp->exp_lock_hash = NULL;
+ ENTRY;
+ cfs_hash_putref(exp->exp_lock_hash);
+ exp->exp_lock_hash = NULL;
ldlm_destroy_flock_export(exp);
- EXIT;
+ EXIT;
}
EXPORT_SYMBOL(ldlm_destroy_export);
+static ssize_t cancel_unused_locks_before_replay_show(struct kobject *kobj,
+ struct attribute *attr,
+ char *buf)
+{
+ return sprintf(buf, "%d\n", ldlm_cancel_unused_locks_before_replay);
+}
+
+static ssize_t cancel_unused_locks_before_replay_store(struct kobject *kobj,
+ struct attribute *attr,
+ const char *buffer,
+ size_t count)
+{
+ int rc;
+ unsigned long val;
+
+ rc = kstrtoul(buffer, 10, &val);
+ if (rc)
+ return rc;
+
+ ldlm_cancel_unused_locks_before_replay = val;
+
+ return count;
+}
+LUSTRE_RW_ATTR(cancel_unused_locks_before_replay);
+
+static struct attribute *ldlm_attrs[] = {
+ &lustre_attr_cancel_unused_locks_before_replay.attr,
+ NULL,
+};
+
+static struct attribute_group ldlm_attr_group = {
+ .attrs = ldlm_attrs,
+};
+
static int ldlm_setup(void)
{
static struct ptlrpc_service_conf conf;
int i;
int rc = 0;
- ENTRY;
+ ENTRY;
+
+ if (ldlm_state != NULL)
+ RETURN(-EALREADY);
- if (ldlm_state != NULL)
- RETURN(-EALREADY);
+ OBD_ALLOC(ldlm_state, sizeof(*ldlm_state));
+ if (ldlm_state == NULL)
+ RETURN(-ENOMEM);
- OBD_ALLOC(ldlm_state, sizeof(*ldlm_state));
- if (ldlm_state == NULL)
- RETURN(-ENOMEM);
+ ldlm_kobj = kobject_create_and_add("ldlm", &lustre_kset->kobj);
+ if (!ldlm_kobj)
+ GOTO(out, -ENOMEM);
-#ifdef CONFIG_PROC_FS
- rc = ldlm_proc_setup();
- if (rc != 0)
+ rc = sysfs_create_group(ldlm_kobj, &ldlm_attr_group);
+ if (rc)
+ GOTO(out, rc);
+
+ ldlm_ns_kset = kset_create_and_add("namespaces", NULL, ldlm_kobj);
+ if (!ldlm_ns_kset)
+ GOTO(out, -ENOMEM);
+
+ ldlm_svc_kset = kset_create_and_add("services", NULL, ldlm_kobj);
+ if (!ldlm_svc_kset)
+ GOTO(out, -ENOMEM);
+
+ rc = ldlm_debugfs_setup();
+ if (rc != 0)
GOTO(out, rc);
-#endif /* CONFIG_PROC_FS */
memset(&conf, 0, sizeof(conf));
conf = (typeof(conf)) {
.tc_nthrs_base = LDLM_NTHRS_BASE,
.tc_nthrs_max = LDLM_NTHRS_MAX,
.tc_nthrs_user = ldlm_num_threads,
- .tc_cpu_affinity = 1,
+ .tc_cpu_bind = ldlm_cpu_bind,
.tc_ctx_tags = LCT_MD_THREAD | LCT_DT_THREAD,
},
.psc_cpt = {
.cc_pattern = ldlm_cpts,
+ .cc_affinity = true,
},
.psc_ops = {
.so_req_handler = ldlm_callback_handler,
},
};
ldlm_state->ldlm_cb_service = \
- ptlrpc_register_service(&conf, ldlm_svc_proc_dir);
+ ptlrpc_register_service(&conf, ldlm_svc_kset,
+ ldlm_svc_debugfs_dir);
if (IS_ERR(ldlm_state->ldlm_cb_service)) {
CERROR("failed to start service\n");
rc = PTR_ERR(ldlm_state->ldlm_cb_service);
.tc_nthrs_base = LDLM_NTHRS_BASE,
.tc_nthrs_max = LDLM_NTHRS_MAX,
.tc_nthrs_user = ldlm_num_threads,
- .tc_cpu_affinity = 1,
+ .tc_cpu_bind = ldlm_cpu_bind,
.tc_ctx_tags = LCT_MD_THREAD | \
LCT_DT_THREAD | \
LCT_CL_THREAD,
},
.psc_cpt = {
.cc_pattern = ldlm_cpts,
+ .cc_affinity = true,
},
.psc_ops = {
.so_req_handler = ldlm_cancel_handler,
},
};
ldlm_state->ldlm_cancel_service = \
- ptlrpc_register_service(&conf, ldlm_svc_proc_dir);
+ ptlrpc_register_service(&conf, ldlm_svc_kset,
+ ldlm_svc_debugfs_dir);
if (IS_ERR(ldlm_state->ldlm_cancel_service)) {
CERROR("failed to start service\n");
rc = PTR_ERR(ldlm_state->ldlm_cancel_service);
}
#ifdef HAVE_SERVER_SUPPORT
- INIT_LIST_HEAD(&expired_lock_thread.elt_expired_locks);
- expired_lock_thread.elt_state = ELT_STOPPED;
- init_waitqueue_head(&expired_lock_thread.elt_waitq);
-
- INIT_LIST_HEAD(&waiting_locks_list);
- spin_lock_init(&waiting_locks_spinlock);
- cfs_timer_init(&waiting_locks_timer, waiting_locks_callback, NULL);
-
task = kthread_run(expired_lock_main, NULL, "ldlm_elt");
if (IS_ERR(task)) {
rc = PTR_ERR(task);
GOTO(out, rc);
}
- wait_event(expired_lock_thread.elt_waitq,
- expired_lock_thread.elt_state == ELT_READY);
+ wait_event(expired_lock_wait_queue,
+ expired_lock_thread_state == ELT_READY);
#endif /* HAVE_SERVER_SUPPORT */
rc = ldlm_pools_init();
static int ldlm_cleanup(void)
{
- ENTRY;
+ ENTRY;
if (!list_empty(ldlm_namespace_list(LDLM_NAMESPACE_SERVER)) ||
!list_empty(ldlm_namespace_list(LDLM_NAMESPACE_CLIENT))) {
- CERROR("ldlm still has namespaces; clean these up first.\n");
- ldlm_dump_all_namespaces(LDLM_NAMESPACE_SERVER, D_DLMTRACE);
- ldlm_dump_all_namespaces(LDLM_NAMESPACE_CLIENT, D_DLMTRACE);
- RETURN(-EBUSY);
- }
+ CERROR("ldlm still has namespaces; clean these up first.\n");
+ ldlm_dump_all_namespaces(LDLM_NAMESPACE_SERVER, D_DLMTRACE);
+ ldlm_dump_all_namespaces(LDLM_NAMESPACE_CLIENT, D_DLMTRACE);
+ RETURN(-EBUSY);
+ }
ldlm_reclaim_cleanup();
ldlm_pools_fini();
ptlrpc_unregister_service(ldlm_state->ldlm_cancel_service);
#endif
- ldlm_proc_cleanup();
+ if (ldlm_ns_kset)
+ kset_unregister(ldlm_ns_kset);
+ if (ldlm_svc_kset)
+ kset_unregister(ldlm_svc_kset);
+ if (ldlm_kobj) {
+ sysfs_remove_group(ldlm_kobj, &ldlm_attr_group);
+ kobject_put(ldlm_kobj);
+ }
+
+ ldlm_debugfs_cleanup();
#ifdef HAVE_SERVER_SUPPORT
- if (expired_lock_thread.elt_state != ELT_STOPPED) {
- expired_lock_thread.elt_state = ELT_TERMINATE;
- wake_up(&expired_lock_thread.elt_waitq);
- wait_event(expired_lock_thread.elt_waitq,
- expired_lock_thread.elt_state == ELT_STOPPED);
+ if (expired_lock_thread_state != ELT_STOPPED) {
+ expired_lock_thread_state = ELT_TERMINATE;
+ wake_up(&expired_lock_wait_queue);
+ wait_event(expired_lock_wait_queue,
+ expired_lock_thread_state == ELT_STOPPED);
}
#endif
- OBD_FREE(ldlm_state, sizeof(*ldlm_state));
- ldlm_state = NULL;
+ OBD_FREE(ldlm_state, sizeof(*ldlm_state));
+ ldlm_state = NULL;
- RETURN(0);
+ RETURN(0);
}
int ldlm_init(void)
{
- mutex_init(&ldlm_ref_mutex);
- mutex_init(ldlm_namespace_lock(LDLM_NAMESPACE_SERVER));
- mutex_init(ldlm_namespace_lock(LDLM_NAMESPACE_CLIENT));
-
- INIT_LIST_HEAD(&ldlm_srv_namespace_list);
- INIT_LIST_HEAD(&ldlm_cli_active_namespace_list);
- INIT_LIST_HEAD(&ldlm_cli_inactive_namespace_list);
-
ldlm_resource_slab = kmem_cache_create("ldlm_resources",
sizeof(struct ldlm_resource), 0,
SLAB_HWCACHE_ALIGN, NULL);
ldlm_lock_slab = kmem_cache_create("ldlm_locks",
sizeof(struct ldlm_lock), 0,
- SLAB_HWCACHE_ALIGN | SLAB_DESTROY_BY_RCU, NULL);
+ SLAB_HWCACHE_ALIGN, NULL);
if (ldlm_lock_slab == NULL)
goto out_resource;
ldlm_interval_slab = kmem_cache_create("interval_node",
- sizeof(struct ldlm_interval),
+ sizeof(struct ldlm_interval),
0, SLAB_HWCACHE_ALIGN, NULL);
if (ldlm_interval_slab == NULL)
goto out_lock;
if (ldlm_interval_tree_slab == NULL)
goto out_interval;
+#ifdef HAVE_SERVER_SUPPORT
+ ldlm_inodebits_slab = kmem_cache_create("ldlm_ibits_node",
+ sizeof(struct ldlm_ibits_node),
+ 0, SLAB_HWCACHE_ALIGN, NULL);
+ if (ldlm_inodebits_slab == NULL)
+ goto out_interval_tree;
+
+ ldlm_glimpse_work_kmem = kmem_cache_create("ldlm_glimpse_work_kmem",
+ sizeof(struct ldlm_glimpse_work),
+ 0, 0, NULL);
+ if (ldlm_glimpse_work_kmem == NULL)
+ goto out_inodebits;
+#endif
+
#if LUSTRE_TRACKS_LOCK_EXP_REFS
class_export_dump_hook = ldlm_dump_export_locks;
#endif
return 0;
-
+#ifdef HAVE_SERVER_SUPPORT
+out_inodebits:
+ kmem_cache_destroy(ldlm_inodebits_slab);
+out_interval_tree:
+ kmem_cache_destroy(ldlm_interval_tree_slab);
+#endif
out_interval:
kmem_cache_destroy(ldlm_interval_slab);
out_lock:
if (ldlm_refcount)
CERROR("ldlm_refcount is %d in ldlm_exit!\n", ldlm_refcount);
kmem_cache_destroy(ldlm_resource_slab);
- /* ldlm_lock_put() use RCU to call ldlm_lock_free, so need call
- * synchronize_rcu() to wait a grace period elapsed, so that
- * ldlm_lock_free() get a chance to be called. */
- synchronize_rcu();
+ /*
+ * ldlm_lock_put() use RCU to call ldlm_lock_free, so need call
+ * rcu_barrier() to wait all outstanding RCU callbacks to complete,
+ * so that ldlm_lock_free() get a chance to be called.
+ */
+ rcu_barrier();
kmem_cache_destroy(ldlm_lock_slab);
kmem_cache_destroy(ldlm_interval_slab);
kmem_cache_destroy(ldlm_interval_tree_slab);
+#ifdef HAVE_SERVER_SUPPORT
+ kmem_cache_destroy(ldlm_inodebits_slab);
+ kmem_cache_destroy(ldlm_glimpse_work_kmem);
+#endif
}
git://git.whamcloud.com / fs / lustre-release.git / blobdiff