-/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
- * vim:expandtab:shiftwidth=8:tabstop=8:
- *
+/*
* GPL HEADER START
*
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
/*
* Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
+ *
+ * Copyright (c) 2010, 2011, Whamcloud, Inc.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
RETURN(0);
}
+EXPORT_SYMBOL(ldlm_expired_completion_wait);
/* We use the same basis for both server side and client side functions
from a single node. */
LDLM_DEBUG(lock, "client-side enqueue returned a blocked lock, "
"going forward");
- ldlm_lock_dump(D_OTHER, lock, 0);
ldlm_reprocess_all(lock->l_resource);
RETURN(0);
}
+EXPORT_SYMBOL(ldlm_completion_ast_async);
/**
* Client side LDLM "completion" AST. This is called in several cases:
LDLM_DEBUG(lock, "client-side enqueue returned a blocked lock, "
"sleeping");
- ldlm_lock_dump(D_OTHER, lock, 0);
noreproc:
RETURN(ldlm_completion_tail(lock));
}
+EXPORT_SYMBOL(ldlm_completion_ast);
/**
* A helper to build a blocking ast function
}
RETURN(0);
}
+EXPORT_SYMBOL(ldlm_blocking_ast_nocheck);
/**
* Server blocking AST
lock_res_and_lock(lock);
/* Get this: if ldlm_blocking_ast is racing with intent_policy, such
* that ldlm_blocking_ast is called just before intent_policy method
- * takes the ns_lock, then by the time we get the lock, we might not
+ * takes the lr_lock, then by the time we get the lock, we might not
* be the correct blocking function anymore. So check, and return
* early, if so. */
if (lock->l_blocking_ast != ldlm_blocking_ast) {
}
RETURN(ldlm_blocking_ast_nocheck(lock));
}
+EXPORT_SYMBOL(ldlm_blocking_ast);
/*
* ->l_glimpse_ast() for DLM extent locks acquired on the server-side. See
*/
return -ELDLM_NO_LOCK_DATA;
}
+EXPORT_SYMBOL(ldlm_glimpse_ast);
int ldlm_cli_enqueue_local(struct ldlm_namespace *ns,
const struct ldlm_res_id *res_id,
if (unlikely(!lock))
GOTO(out_nolock, err = -ENOMEM);
- ldlm_lock_addref_internal(lock, mode);
ldlm_lock2handle(lock, lockh);
- lock_res_and_lock(lock);
+
+ /* NB: we don't have any lock now (lock_res_and_lock)
+ * because it's a new lock */
+ ldlm_lock_addref_internal_nolock(lock, mode);
lock->l_flags |= LDLM_FL_LOCAL;
if (*flags & LDLM_FL_ATOMIC_CB)
lock->l_flags |= LDLM_FL_ATOMIC_CB;
- unlock_res_and_lock(lock);
+
if (policy != NULL)
lock->l_policy_data = *policy;
if (client_cookie != NULL)
out_nolock:
return err;
}
+EXPORT_SYMBOL(ldlm_cli_enqueue_local);
static void failed_lock_cleanup(struct ldlm_namespace *ns,
struct ldlm_lock *lock, int mode)
int is_replay = *flags & LDLM_FL_REPLAY;
struct ldlm_lock *lock;
struct ldlm_reply *reply;
+ struct ost_lvb *tmplvb;
int cleanup_phase = 1;
ENTRY;
if (reply == NULL)
rc = -EPROTO;
if (lvb_len) {
- struct ost_lvb *tmplvb;
req_capsule_set_size(&req->rq_pill,
&RMF_DLM_LVB, RCL_SERVER,
lvb_len);
- tmplvb = req_capsule_server_get(&req->rq_pill,
+ tmplvb = req_capsule_server_get(&req->rq_pill,
&RMF_DLM_LVB);
if (tmplvb == NULL)
GOTO(cleanup, rc = -EPROTO);
lock_res_and_lock(lock);
/* Key change rehash lock in per-export hash with new key */
if (exp->exp_lock_hash) {
+ /* In the function below, .hs_keycmp resolves to
+ * ldlm_export_lock_keycmp() */
+ /* coverity[overrun-buffer-val] */
cfs_hash_rehash_key(exp->exp_lock_hash,
&lock->l_remote_handle,
&reply->lock_handle,
if (with_policy)
if (!(type == LDLM_IBITS && !(exp->exp_connect_flags &
OBD_CONNECT_IBITS)))
- lock->l_policy_data =
- reply->lock_desc.l_policy_data;
+ /* We assume lock type cannot change on server*/
+ ldlm_convert_policy_to_local(exp,
+ lock->l_resource->lr_type,
+ &reply->lock_desc.l_policy_data,
+ &lock->l_policy_data);
if (type != LDLM_PLAIN)
LDLM_DEBUG(lock,"client-side enqueue, new policy data");
}
/* If the lock has already been granted by a completion AST, don't
* clobber the LVB with an older one. */
- if (lvb_len && (lock->l_req_mode != lock->l_granted_mode)) {
- void *tmplvb;
+ if (lvb_len) {
+ /* We must lock or a racing completion might update lvb
+ without letting us know and we'll clobber the correct value.
+ Cannot unlock after the check either, a that still leaves
+ a tiny window for completion to get in */
+ lock_res_and_lock(lock);
+ if (lock->l_req_mode != lock->l_granted_mode) {
- req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
- lvb_len);
- tmplvb = req_capsule_server_get(&req->rq_pill,
- &RMF_DLM_LVB);
- if (tmplvb == NULL)
- GOTO(cleanup, rc = -EPROTO);
- memcpy(lock->l_lvb_data, tmplvb, lvb_len);
+ req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB,
+ RCL_SERVER, lvb_len);
+ tmplvb = req_capsule_server_get(&req->rq_pill,
+ &RMF_DLM_LVB);
+ if (tmplvb == NULL) {
+ unlock_res_and_lock(lock);
+ GOTO(cleanup, rc = -EPROTO);
+ }
+ memcpy(lock->l_lvb_data, tmplvb, lvb_len);
+ }
+ unlock_res_and_lock(lock);
}
if (!is_replay) {
LDLM_LOCK_RELEASE(lock);
return rc;
}
+EXPORT_SYMBOL(ldlm_cli_enqueue_fini);
/* PAGE_SIZE-512 is to allow TCP/IP and LNET headers to fit into
* a single page on the send/receive side. XXX: 512 should be changed
struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
struct req_capsule *pill = &req->rq_pill;
struct ldlm_request *dlm = NULL;
- int flags, avail, to_free, bufcount, pack = 0;
+ int flags, avail, to_free, pack = 0;
CFS_LIST_HEAD(head);
int rc;
ENTRY;
cancels = &head;
if (exp_connect_cancelset(exp)) {
/* Estimate the amount of available space in the request. */
- bufcount = req_capsule_filled_sizes(pill, RCL_CLIENT);
+ req_capsule_filled_sizes(pill, RCL_CLIENT);
avail = ldlm_capsule_handles_avail(pill, RCL_CLIENT, canceloff);
flags = ns_connect_lru_resize(ns) ?
}
RETURN(0);
}
+EXPORT_SYMBOL(ldlm_prep_elc_req);
int ldlm_prep_enqueue_req(struct obd_export *exp, struct ptlrpc_request *req,
cfs_list_t *cancels, int count)
return ldlm_prep_elc_req(exp, req, LUSTRE_DLM_VERSION, LDLM_ENQUEUE,
LDLM_ENQUEUE_CANCEL_OFF, cancels, count);
}
+EXPORT_SYMBOL(ldlm_prep_enqueue_req);
/* If a request has some specific initialisation it is passed in @reqp,
* otherwise it is created in ldlm_cli_enqueue.
LDLM_DEBUG(lock, "client-side enqueue START");
}
+ lock->l_conn_export = exp;
+ lock->l_export = NULL;
+ lock->l_blocking_ast = einfo->ei_cb_bl;
+ lock->l_flags |= (*flags & LDLM_FL_NO_LRU);
+
/* lock not sent to server yet */
if (reqp == NULL || *reqp == NULL) {
DLM_LOCKREQ_OFF, len, (int)sizeof(*body));
}
- lock->l_conn_export = exp;
- lock->l_export = NULL;
- lock->l_blocking_ast = einfo->ei_cb_bl;
-
/* Dump lock data into the request buffer */
body = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
ldlm_lock2desc(lock, &body->lock_desc);
RETURN(rc);
}
+EXPORT_SYMBOL(ldlm_cli_enqueue);
static int ldlm_cli_convert_local(struct ldlm_lock *lock, int new_mode,
__u32 *flags)
ptlrpc_req_finished(req);
return rc;
}
+EXPORT_SYMBOL(ldlm_cli_convert);
/* Cancel locks locally.
* Returns:
LASSERT(exp != NULL);
LASSERT(count > 0);
- OBD_FAIL_TIMEOUT(OBD_FAIL_LDLM_PAUSE_CANCEL, obd_fail_val);
+ CFS_FAIL_TIMEOUT(OBD_FAIL_LDLM_PAUSE_CANCEL, cfs_fail_val);
- if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_RACE))
+ if (CFS_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_RACE))
RETURN(count);
free = ldlm_format_handles_avail(class_exp2cliimp(exp),
count = free;
while (1) {
- int bufcount;
-
imp = class_exp2cliimp(exp);
if (imp == NULL || imp->imp_invalid) {
CDEBUG(D_DLMTRACE,
if (req == NULL)
GOTO(out, rc = -ENOMEM);
- bufcount = req_capsule_filled_sizes(&req->rq_pill, RCL_CLIENT);
+ req_capsule_filled_sizes(&req->rq_pill, RCL_CLIENT);
req_capsule_set_size(&req->rq_pill, &RMF_DLM_REQ, RCL_CLIENT,
ldlm_request_bufsize(count, LDLM_CANCEL));
ptlrpc_request_set_replen(req);
if (flags & LCF_ASYNC) {
- ptlrpcd_add_req(req, PSCOPE_OTHER);
+ ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
sent = count;
GOTO(out, 0);
} else {
out:
return sent ? sent : rc;
}
+EXPORT_SYMBOL(ldlm_cli_cancel_req);
static inline struct ldlm_pool *ldlm_imp2pl(struct obd_import *imp)
{
int ldlm_cli_update_pool(struct ptlrpc_request *req)
{
struct obd_device *obd;
- __u64 old_slv, new_slv;
+ __u64 new_slv;
__u32 new_limit;
ENTRY;
if (unlikely(!req->rq_import || !req->rq_import->imp_obd ||
* oops in that time.
*/
cfs_write_lock(&obd->obd_pool_lock);
- old_slv = obd->obd_pool_slv;
obd->obd_pool_slv = new_slv;
obd->obd_pool_limit = new_limit;
cfs_write_unlock(&obd->obd_pool_lock);
ldlm_cli_cancel_list(&cancels, count, NULL, 0);
RETURN(0);
}
+EXPORT_SYMBOL(ldlm_cli_cancel);
/* XXX until we will have compound requests and can cut cancels from generic rpc
* we need send cancels with LDLM_FL_BL_AST flag as separate rpc */
RETURN(count);
}
+EXPORT_SYMBOL(ldlm_cli_cancel_list_local);
/**
* Cancel as many locks as possible w/o sending any rpcs (e.g. to write back
int added = 0, unused, remained;
ENTRY;
- cfs_spin_lock(&ns->ns_unused_lock);
+ cfs_spin_lock(&ns->ns_lock);
unused = ns->ns_nr_unused;
remained = unused;
break;
LDLM_LOCK_GET(lock);
- cfs_spin_unlock(&ns->ns_unused_lock);
+ cfs_spin_unlock(&ns->ns_lock);
lu_ref_add(&lock->l_reference, __FUNCTION__, cfs_current());
/* Pass the lock through the policy filter and see if it
lu_ref_del(&lock->l_reference,
__FUNCTION__, cfs_current());
LDLM_LOCK_RELEASE(lock);
- cfs_spin_lock(&ns->ns_unused_lock);
+ cfs_spin_lock(&ns->ns_lock);
break;
}
if (result == LDLM_POLICY_SKIP_LOCK) {
lu_ref_del(&lock->l_reference,
__FUNCTION__, cfs_current());
LDLM_LOCK_RELEASE(lock);
- cfs_spin_lock(&ns->ns_unused_lock);
+ cfs_spin_lock(&ns->ns_lock);
continue;
}
lu_ref_del(&lock->l_reference,
__FUNCTION__, cfs_current());
LDLM_LOCK_RELEASE(lock);
- cfs_spin_lock(&ns->ns_unused_lock);
+ cfs_spin_lock(&ns->ns_lock);
continue;
}
LASSERT(!lock->l_readers && !lock->l_writers);
/* We can't re-add to l_lru as it confuses the
* refcounting in ldlm_lock_remove_from_lru() if an AST
- * arrives after we drop ns_lock below. We use l_bl_ast
+ * arrives after we drop lr_lock below. We use l_bl_ast
* and can't use l_pending_chain as it is used both on
* server and client nevertheless bug 5666 says it is
* used only on server */
cfs_list_add(&lock->l_bl_ast, cancels);
unlock_res_and_lock(lock);
lu_ref_del(&lock->l_reference, __FUNCTION__, cfs_current());
- cfs_spin_lock(&ns->ns_unused_lock);
+ cfs_spin_lock(&ns->ns_lock);
added++;
unused--;
}
- cfs_spin_unlock(&ns->ns_unused_lock);
+ cfs_spin_unlock(&ns->ns_lock);
RETURN(added);
}
RETURN(ldlm_cli_cancel_list_local(cancels, count, cancel_flags));
}
+EXPORT_SYMBOL(ldlm_cancel_resource_local);
/* If @req is NULL, send CANCEL request to server with handles of locks
* in the @cancels. If EARLY_CANCEL is not supported, send CANCEL requests
LASSERT(count == 0);
RETURN(0);
}
+EXPORT_SYMBOL(ldlm_cli_cancel_list);
int ldlm_cli_cancel_unused_resource(struct ldlm_namespace *ns,
const struct ldlm_res_id *res_id,
ldlm_resource_putref(res);
RETURN(0);
}
+EXPORT_SYMBOL(ldlm_cli_cancel_unused_resource);
-static inline int have_no_nsresource(struct ldlm_namespace *ns)
-{
- int no_resource = 0;
-
- cfs_spin_lock(&ns->ns_hash_lock);
- if (ns->ns_resources == 0)
- no_resource = 1;
- cfs_spin_unlock(&ns->ns_hash_lock);
+struct ldlm_cli_cancel_arg {
+ int lc_flags;
+ void *lc_opaque;
+};
- RETURN(no_resource);
+static int ldlm_cli_hash_cancel_unused(cfs_hash_t *hs, cfs_hash_bd_t *bd,
+ cfs_hlist_node_t *hnode, void *arg)
+{
+ struct ldlm_resource *res = cfs_hash_object(hs, hnode);
+ struct ldlm_cli_cancel_arg *lc = arg;
+ int rc;
+
+ rc = ldlm_cli_cancel_unused_resource(ldlm_res_to_ns(res), &res->lr_name,
+ NULL, LCK_MINMODE,
+ lc->lc_flags, lc->lc_opaque);
+ if (rc != 0) {
+ CERROR("ldlm_cli_cancel_unused ("LPU64"): %d\n",
+ res->lr_name.name[0], rc);
+ }
+ /* must return 0 for hash iteration */
+ return 0;
}
/* Cancel all locks on a namespace (or a specific resource, if given)
const struct ldlm_res_id *res_id,
ldlm_cancel_flags_t flags, void *opaque)
{
- int i;
+ struct ldlm_cli_cancel_arg arg = {
+ .lc_flags = flags,
+ .lc_opaque = opaque,
+ };
+
ENTRY;
if (ns == NULL)
RETURN(ELDLM_OK);
- if (res_id)
+ if (res_id != NULL) {
RETURN(ldlm_cli_cancel_unused_resource(ns, res_id, NULL,
LCK_MINMODE, flags,
opaque));
-
- cfs_spin_lock(&ns->ns_hash_lock);
- for (i = 0; i < RES_HASH_SIZE; i++) {
- cfs_list_t *tmp;
- tmp = ns->ns_hash[i].next;
- while (tmp != &(ns->ns_hash[i])) {
- struct ldlm_resource *res;
- int rc;
-
- res = cfs_list_entry(tmp, struct ldlm_resource,
- lr_hash);
- ldlm_resource_getref(res);
- cfs_spin_unlock(&ns->ns_hash_lock);
-
- LDLM_RESOURCE_ADDREF(res);
- rc = ldlm_cli_cancel_unused_resource(ns, &res->lr_name,
- NULL, LCK_MINMODE,
- flags, opaque);
-
- if (rc)
- CERROR("ldlm_cli_cancel_unused ("LPU64"): %d\n",
- res->lr_name.name[0], rc);
-
- LDLM_RESOURCE_DELREF(res);
- cfs_spin_lock(&ns->ns_hash_lock);
- tmp = tmp->next;
- ldlm_resource_putref_locked(res);
- }
+ } else {
+ cfs_hash_for_each_nolock(ns->ns_rs_hash,
+ ldlm_cli_hash_cancel_unused, &arg);
+ RETURN(ELDLM_OK);
}
- cfs_spin_unlock(&ns->ns_hash_lock);
-
- RETURN(ELDLM_OK);
}
+EXPORT_SYMBOL(ldlm_cli_cancel_unused);
/* Lock iterators. */
unlock_res(res);
RETURN(rc);
}
+EXPORT_SYMBOL(ldlm_resource_foreach);
struct iter_helper_data {
ldlm_iterator_t iter;
return helper->iter(lock, helper->closure);
}
-static int ldlm_res_iter_helper(struct ldlm_resource *res, void *closure)
+static int ldlm_res_iter_helper(cfs_hash_t *hs, cfs_hash_bd_t *bd,
+ cfs_hlist_node_t *hnode, void *arg)
+
{
- return ldlm_resource_foreach(res, ldlm_iter_helper, closure);
+ struct ldlm_resource *res = cfs_hash_object(hs, hnode);
+
+ return ldlm_resource_foreach(res, ldlm_iter_helper, arg) ==
+ LDLM_ITER_STOP;
}
-int ldlm_namespace_foreach(struct ldlm_namespace *ns, ldlm_iterator_t iter,
- void *closure)
+void ldlm_namespace_foreach(struct ldlm_namespace *ns,
+ ldlm_iterator_t iter, void *closure)
+
{
struct iter_helper_data helper = { iter: iter, closure: closure };
- return ldlm_namespace_foreach_res(ns, ldlm_res_iter_helper, &helper);
-}
-int ldlm_namespace_foreach_res(struct ldlm_namespace *ns,
- ldlm_res_iterator_t iter, void *closure)
-{
- int i, rc = LDLM_ITER_CONTINUE;
- struct ldlm_resource *res;
- cfs_list_t *tmp;
+ cfs_hash_for_each_nolock(ns->ns_rs_hash,
+ ldlm_res_iter_helper, &helper);
- ENTRY;
- cfs_spin_lock(&ns->ns_hash_lock);
- for (i = 0; i < RES_HASH_SIZE; i++) {
- tmp = ns->ns_hash[i].next;
- while (tmp != &(ns->ns_hash[i])) {
- res = cfs_list_entry(tmp, struct ldlm_resource,
- lr_hash);
- ldlm_resource_getref(res);
- cfs_spin_unlock(&ns->ns_hash_lock);
- LDLM_RESOURCE_ADDREF(res);
-
- rc = iter(res, closure);
-
- LDLM_RESOURCE_DELREF(res);
- cfs_spin_lock(&ns->ns_hash_lock);
- tmp = tmp->next;
- ldlm_resource_putref_locked(res);
- if (rc == LDLM_ITER_STOP)
- GOTO(out, rc);
- }
- }
- out:
- cfs_spin_unlock(&ns->ns_hash_lock);
- RETURN(rc);
}
+EXPORT_SYMBOL(ldlm_namespace_foreach);
/* non-blocking function to manipulate a lock whose cb_data is being put away.
* return 0: find no resource
ldlm_resource_putref(res);
RETURN(rc);
}
+EXPORT_SYMBOL(ldlm_resource_iterate);
/* Lock replay */
/* Key change rehash lock in per-export hash with new key */
exp = req->rq_export;
if (exp && exp->exp_lock_hash) {
+ /* In the function below, .hs_keycmp resolves to
+ * ldlm_export_lock_keycmp() */
+ /* coverity[overrun-buffer-val] */
cfs_hash_rehash_key(exp->exp_lock_hash,
&lock->l_remote_handle,
&reply->lock_handle,
LDLM_LOCK_PUT(lock);
out:
if (rc != ELDLM_OK)
- ptlrpc_connect_import(req->rq_import, NULL);
+ ptlrpc_connect_import(req->rq_import);
RETURN(rc);
}
aa = ptlrpc_req_async_args(req);
aa->lock_handle = body->lock_handle[0];
req->rq_interpret_reply = (ptlrpc_interpterer_t)replay_lock_interpret;
- ptlrpcd_add_req(req, PSCOPE_OTHER);
+ ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
RETURN(0);
}
CFS_LIST_HEAD(cancels);
CDEBUG(D_DLMTRACE, "Dropping as many unused locks as possible before"
- "replay for namespace %s (%d)\n", ns->ns_name,
- ns->ns_nr_unused);
+ "replay for namespace %s (%d)\n",
+ ldlm_ns_name(ns), ns->ns_nr_unused);
/* We don't need to care whether or not LRU resize is enabled
* because the LDLM_CANCEL_NO_WAIT policy doesn't use the
LCF_LOCAL, LDLM_CANCEL_NO_WAIT);
CDEBUG(D_DLMTRACE, "Canceled %d unused locks from namespace %s\n",
- canceled, ns->ns_name);
+ canceled, ldlm_ns_name(ns));
}
int ldlm_replay_locks(struct obd_import *imp)
if (ldlm_cancel_unused_locks_before_replay)
ldlm_cancel_unused_locks_for_replay(ns);
- (void)ldlm_namespace_foreach(ns, ldlm_chain_lock_for_replay, &list);
+ ldlm_namespace_foreach(ns, ldlm_chain_lock_for_replay, &list);
cfs_list_for_each_entry_safe(lock, next, &list, l_pending_chain) {
cfs_list_del_init(&lock->l_pending_chain);
RETURN(rc);
}
+EXPORT_SYMBOL(ldlm_replay_locks);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff