/*
* Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
+ *
+ * Copyright (c) 2011, 2012, Whamcloud, Inc.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
* Allocate and initialize new bulk descriptor
* Returns pointer to the descriptor or NULL on error.
*/
-static inline struct ptlrpc_bulk_desc *new_bulk(int npages, int type, int portal)
+struct ptlrpc_bulk_desc *new_bulk(int npages, int type, int portal)
{
struct ptlrpc_bulk_desc *desc;
}
/**
- * Prepare bulk descriptor for specified incoming request \a req that
- * can fit \a npages * pages. \a type is bulk type. \a portal is where
- * the bulk to be sent. Used on server-side after request was already
- * received.
- * Returns pointer to newly allocatrd initialized bulk descriptor or NULL on
- * error.
- */
-struct ptlrpc_bulk_desc *ptlrpc_prep_bulk_exp(struct ptlrpc_request *req,
- int npages, int type, int portal)
-{
- struct obd_export *exp = req->rq_export;
- struct ptlrpc_bulk_desc *desc;
-
- ENTRY;
- LASSERT(type == BULK_PUT_SOURCE || type == BULK_GET_SINK);
-
- desc = new_bulk(npages, type, portal);
- if (desc == NULL)
- RETURN(NULL);
-
- desc->bd_export = class_export_get(exp);
- desc->bd_req = req;
-
- desc->bd_cbid.cbid_fn = server_bulk_callback;
- desc->bd_cbid.cbid_arg = desc;
-
- /* NB we don't assign rq_bulk here; server-side requests are
- * re-used, and the handler frees the bulk desc explicitly. */
-
- return desc;
-}
-
-/**
* Add a page \a page to the bulk descriptor \a desc.
* Data to transfer in the page starts at offset \a pageoffset and
* amount of data to transfer from the page is \a len
*/
void ptlrpc_fakereq_finished(struct ptlrpc_request *req)
{
- /* if we kill request before timeout - need adjust counter */
- if (req->rq_phase == RQ_PHASE_RPC) {
- struct ptlrpc_request_set *set = req->rq_set;
+ struct ptlrpc_request_set *set = req->rq_set;
+ int wakeup = 0;
- if (set)
- cfs_atomic_dec(&set->set_remaining);
- }
+ /* hold ref on the request to prevent others (ptlrpcd) to free it */
+ ptlrpc_request_addref(req);
+ cfs_list_del_init(&req->rq_list);
+
+ /* if we kill request before timeout - need adjust counter */
+ if (req->rq_phase == RQ_PHASE_RPC && set != NULL &&
+ cfs_atomic_dec_and_test(&set->set_remaining))
+ wakeup = 1;
ptlrpc_rqphase_move(req, RQ_PHASE_COMPLETE);
- cfs_list_del_init(&req->rq_list);
+
+ /* Only need to call wakeup once when to be empty. */
+ if (wakeup)
+ cfs_waitq_signal(&set->set_waitq);
+ ptlrpc_req_finished(req);
}
/**
OBD_ALLOC(set, sizeof *set);
if (!set)
RETURN(NULL);
+ cfs_atomic_set(&set->set_refcount, 1);
CFS_INIT_LIST_HEAD(&set->set_requests);
cfs_waitq_init(&set->set_waitq);
+ cfs_atomic_set(&set->set_new_count, 0);
cfs_atomic_set(&set->set_remaining, 0);
cfs_spin_lock_init(&set->set_new_req_lock);
CFS_INIT_LIST_HEAD(&set->set_new_requests);
LASSERT(cfs_atomic_read(&set->set_remaining) == 0);
- OBD_FREE(set, sizeof(*set));
+ ptlrpc_reqset_put(set);
EXIT;
}
void ptlrpc_set_add_req(struct ptlrpc_request_set *set,
struct ptlrpc_request *req)
{
+ LASSERT(cfs_list_empty(&req->rq_set_chain));
+
/* The set takes over the caller's request reference */
cfs_list_add_tail(&req->rq_set_chain, &set->set_requests);
req->rq_set = set;
cfs_atomic_inc(&set->set_remaining);
- req->rq_queued_time = cfs_time_current(); /* Where is the best place to set this? */
+ req->rq_queued_time = cfs_time_current();
}
/**
* Add a request to a request with dedicated server thread
* and wake the thread to make any necessary processing.
* Currently only used for ptlrpcd.
- * Returns 0 if succesful or non zero error code on error.
- * (the only possible error for now is if the dedicated server thread
- * is shutting down)
*/
-int ptlrpc_set_add_new_req(struct ptlrpcd_ctl *pc,
+void ptlrpc_set_add_new_req(struct ptlrpcd_ctl *pc,
struct ptlrpc_request *req)
{
struct ptlrpc_request_set *set = pc->pc_set;
+ int count, i;
- /*
- * Let caller know that we stopped and will not handle this request.
- * It needs to take care itself of request.
- */
- if (cfs_test_bit(LIOD_STOP, &pc->pc_flags))
- return -EALREADY;
+ LASSERT(req->rq_set == NULL);
+ LASSERT(cfs_test_bit(LIOD_STOP, &pc->pc_flags) == 0);
cfs_spin_lock(&set->set_new_req_lock);
/*
* The set takes over the caller's request reference.
*/
- cfs_list_add_tail(&req->rq_set_chain, &set->set_new_requests);
req->rq_set = set;
+ req->rq_queued_time = cfs_time_current();
+ cfs_list_add_tail(&req->rq_set_chain, &set->set_new_requests);
+ count = cfs_atomic_inc_return(&set->set_new_count);
cfs_spin_unlock(&set->set_new_req_lock);
- cfs_waitq_signal(&set->set_waitq);
- return 0;
+ /* Only need to call wakeup once for the first entry. */
+ if (count == 1) {
+ cfs_waitq_signal(&set->set_waitq);
+
+ /* XXX: It maybe unnecessary to wakeup all the partners. But to
+ * guarantee the async RPC can be processed ASAP, we have
+ * no other better choice. It maybe fixed in future. */
+ for (i = 0; i < pc->pc_npartners; i++)
+ cfs_waitq_signal(&pc->pc_partners[i]->pc_set->set_waitq);
+ }
}
/**
*/
static int ptlrpc_console_allow(struct ptlrpc_request *req)
{
- __u32 opc = lustre_msg_get_opc(req->rq_reqmsg);
+ __u32 opc;
int err;
+ /* Fake requests include no rq_reqmsg */
+ if (req->rq_fake)
+ return 0;
+
+ LASSERT(req->rq_reqmsg != NULL);
+ opc = lustre_msg_get_opc(req->rq_reqmsg);
+
/* Suppress particular reconnect errors which are to be expected. No
* errors are suppressed for the initial connection on an import */
if ((lustre_handle_is_used(&req->rq_import->imp_remote_handle)) &&
}
if (lustre_msg_get_opc(req->rq_reqmsg) != OBD_PING)
- OBD_FAIL_TIMEOUT(OBD_FAIL_PTLRPC_PAUSE_REP, obd_fail_val);
+ CFS_FAIL_TIMEOUT(OBD_FAIL_PTLRPC_PAUSE_REP, cfs_fail_val);
ptlrpc_at_adj_service(req, lustre_msg_get_timeout(req->rq_repmsg));
ptlrpc_at_adj_net_latency(req,
lustre_msg_get_service_time(req->rq_repmsg));
* Helper function to send request \a req over the network for the first time
* Also adjusts request phase.
* Returns 0 on success or error code.
- */
+ */
static int ptlrpc_send_new_req(struct ptlrpc_request *req)
{
- struct obd_import *imp;
+ struct obd_import *imp = req->rq_import;
int rc;
ENTRY;
LASSERT(req->rq_phase == RQ_PHASE_NEW);
- if (req->rq_sent && (req->rq_sent > cfs_time_current_sec()))
+ if (req->rq_sent && (req->rq_sent > cfs_time_current_sec()) &&
+ (!req->rq_generation_set ||
+ req->rq_import_generation == imp->imp_generation))
RETURN (0);
ptlrpc_rqphase_move(req, RQ_PHASE_RPC);
- imp = req->rq_import;
cfs_spin_lock(&imp->imp_lock);
- req->rq_import_generation = imp->imp_generation;
+ if (!req->rq_generation_set)
+ req->rq_import_generation = imp->imp_generation;
if (ptlrpc_import_delay_req(imp, req, &rc)) {
cfs_spin_lock(&req->rq_lock);
* process the reply. Similarly if the RPC returned
* an error, and therefore the bulk will never arrive.
*/
- if (req->rq_bulk == NULL || req->rq_status != 0) {
+ if (req->rq_bulk == NULL || req->rq_status < 0) {
ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
GOTO(interpret, req->rq_status);
}
* was good after getting the REPLY for her GET or
* the ACK for her PUT. */
DEBUG_REQ(D_ERROR, req, "bulk transfer failed");
- LBUG();
+ req->rq_status = -EIO;
}
ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
/* This moves to "unregistering" phase we need to wait for
* reply unlink. */
- if (!unregistered && !ptlrpc_unregister_reply(req, 1))
+ if (!unregistered && !ptlrpc_unregister_reply(req, 1)) {
+ /* start async bulk unlink too */
+ ptlrpc_unregister_bulk(req, 1);
continue;
+ }
if (!ptlrpc_unregister_bulk(req, 1))
continue;
req->rq_timedout = 1;
cfs_spin_unlock(&req->rq_lock);
- DEBUG_REQ(req->rq_fake ? D_INFO : D_WARNING, req, "Request x"LPU64
- " sent from %s to NID %s has %s: [sent "CFS_DURATION_T"] "
- "[real_sent "CFS_DURATION_T"] [current "CFS_DURATION_T"] "
- "[deadline "CFS_DURATION_T"s] [delay "CFS_DURATION_T"s]",
- req->rq_xid, imp ? imp->imp_obd->obd_name : "<?>",
- imp ? libcfs_nid2str(imp->imp_connection->c_peer.nid) : "<?>",
+ DEBUG_REQ(req->rq_fake ? D_INFO : D_WARNING, req, "Request "
+ " sent has %s: [sent "CFS_DURATION_T"/"
+ "real "CFS_DURATION_T"]",
req->rq_net_err ? "failed due to network error" :
((req->rq_real_sent == 0 ||
cfs_time_before(req->rq_real_sent, req->rq_sent) ||
cfs_time_aftereq(req->rq_real_sent, req->rq_deadline)) ?
"timed out for sent delay" : "timed out for slow reply"),
- req->rq_sent, req->rq_real_sent, cfs_time_current_sec(),
- cfs_time_sub(req->rq_deadline, req->rq_sent),
- cfs_time_sub(cfs_time_current_sec(), req->rq_deadline));
+ req->rq_sent, req->rq_real_sent);
if (imp != NULL && obd_debug_peer_on_timeout)
LNetCtl(IOC_LIBCFS_DEBUG_PEER, &imp->imp_connection->c_peer);
cfs_list_t *tmp;
LASSERT(set != NULL);
- CERROR("INTERRUPTED SET %p\n", set);
+ CDEBUG(D_RPCTRACE, "INTERRUPTED SET %p\n", set);
cfs_list_for_each(tmp, &set->set_requests) {
struct ptlrpc_request *req =
rc = l_wait_event(set->set_waitq, ptlrpc_check_set(NULL, set), &lwi);
+ /* LU-769 - if we ignored the signal because it was already
+ * pending when we started, we need to handle it now or we risk
+ * it being ignored forever */
+ if (rc == -ETIMEDOUT && !lwi.lwi_allow_intr &&
+ cfs_signal_pending()) {
+ cfs_sigset_t blocked_sigs =
+ cfs_block_sigsinv(LUSTRE_FATAL_SIGS);
+
+ /* In fact we only interrupt for the "fatal" signals
+ * like SIGINT or SIGKILL. We still ignore less
+ * important signals since ptlrpc set is not easily
+ * reentrant from userspace again */
+ if (cfs_signal_pending())
+ ptlrpc_interrupted_set(set);
+ cfs_block_sigs(blocked_sigs);
+ }
+
LASSERT(rc == 0 || rc == -EINTR || rc == -ETIMEDOUT);
/* -EINTR => all requests have been flagged rq_intr so next
LASSERTF(cfs_list_empty(&request->rq_set_chain), "req %p\n", request);
LASSERTF(cfs_list_empty(&request->rq_exp_list), "req %p\n", request);
LASSERTF(!request->rq_replay, "req %p\n", request);
- LASSERT(request->rq_cli_ctx || request->rq_fake);
req_capsule_fini(&request->rq_pill);
imp->imp_vbr_failed = 1;
imp->imp_no_lock_replay = 1;
cfs_spin_unlock(&imp->imp_lock);
+ lustre_msg_set_status(req->rq_repmsg, aa->praa_old_status);
} else {
/** The transno had better not change over replay. */
LASSERTF(lustre_msg_get_transno(req->rq_reqmsg) ==
cfs_spin_unlock(&imp->imp_lock);
LASSERT(imp->imp_last_replay_transno);
+ /* transaction number shouldn't be bigger than the latest replayed */
+ if (req->rq_transno > lustre_msg_get_transno(req->rq_reqmsg)) {
+ DEBUG_REQ(D_ERROR, req,
+ "Reported transno "LPU64" is bigger than the "
+ "replayed one: "LPU64, req->rq_transno,
+ lustre_msg_get_transno(req->rq_reqmsg));
+ GOTO(out, rc = -EINVAL);
+ }
+
DEBUG_REQ(D_HA, req, "got rep");
/* let the callback do fixups, possibly including in the request */
* Errors while replay can set transno to 0, but
* imp_last_replay_transno shouldn't be set to 0 anyway
*/
- if (req->rq_transno > 0) {
- cfs_spin_lock(&imp->imp_lock);
- LASSERT(req->rq_transno <= imp->imp_last_replay_transno);
- imp->imp_last_replay_transno = req->rq_transno;
- cfs_spin_unlock(&imp->imp_lock);
- } else
+ if (req->rq_transno == 0)
CERROR("Transno is 0 during replay!\n");
+
/* continue with recovery */
rc = ptlrpc_import_recovery_state_machine(imp);
out:
if (rc != 0)
/* this replay failed, so restart recovery */
- ptlrpc_connect_import(imp, NULL);
+ ptlrpc_connect_import(imp);
RETURN(rc);
}
/* Readjust the timeout for current conditions */
ptlrpc_at_set_req_timeout(req);
+ /* Tell server the net_latency, so the server can calculate how long
+ * it should wait for next replay */
+ lustre_msg_set_service_time(req->rq_reqmsg,
+ ptlrpc_at_get_net_latency(req));
DEBUG_REQ(D_HA, req, "REPLAY");
cfs_atomic_inc(&req->rq_import->imp_replay_inflight);
ptlrpc_request_addref(req); /* ptlrpcd needs a ref */
- ptlrpcd_add_req(req, PSCOPE_OTHER);
+ ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
RETURN(0);
}
#endif
}
EXPORT_SYMBOL(ptlrpc_sample_next_xid);
+
+/**
+ * Functions for operating ptlrpc workers.
+ *
+ * A ptlrpc work is a function which will be running inside ptlrpc context.
+ * The callback shouldn't sleep otherwise it will block that ptlrpcd thread.
+ *
+ * 1. after a work is created, it can be used many times, that is:
+ * handler = ptlrpcd_alloc_work();
+ * ptlrpcd_queue_work();
+ *
+ * queue it again when necessary:
+ * ptlrpcd_queue_work();
+ * ptlrpcd_destroy_work();
+ * 2. ptlrpcd_queue_work() can be called by multiple processes meanwhile, but
+ * it will only be queued once in any time. Also as its name implies, it may
+ * have delay before it really runs by ptlrpcd thread.
+ */
+struct ptlrpc_work_async_args {
+ __u64 magic;
+ int (*cb)(const struct lu_env *, void *);
+ void *cbdata;
+};
+
+#define PTLRPC_WORK_MAGIC 0x6655436b676f4f44ULL /* magic code */
+
+static int work_interpreter(const struct lu_env *env,
+ struct ptlrpc_request *req, void *data, int rc)
+{
+ struct ptlrpc_work_async_args *arg = data;
+
+ LASSERT(arg->magic == PTLRPC_WORK_MAGIC);
+ LASSERT(arg->cb != NULL);
+
+ return arg->cb(env, arg->cbdata);
+}
+
+/**
+ * Create a work for ptlrpc.
+ */
+void *ptlrpcd_alloc_work(struct obd_import *imp,
+ int (*cb)(const struct lu_env *, void *), void *cbdata)
+{
+ struct ptlrpc_request *req = NULL;
+ struct ptlrpc_work_async_args *args;
+ ENTRY;
+
+ cfs_might_sleep();
+
+ if (cb == NULL)
+ RETURN(ERR_PTR(-EINVAL));
+
+ /* copy some code from deprecated fakereq. */
+ OBD_ALLOC_PTR(req);
+ if (req == NULL) {
+ CERROR("ptlrpc: run out of memory!\n");
+ RETURN(ERR_PTR(-ENOMEM));
+ }
+
+ req->rq_send_state = LUSTRE_IMP_FULL;
+ req->rq_type = PTL_RPC_MSG_REQUEST;
+ req->rq_import = class_import_get(imp);
+ req->rq_export = NULL;
+ req->rq_interpret_reply = work_interpreter;
+ /* don't want reply */
+ req->rq_receiving_reply = 0;
+ req->rq_must_unlink = 0;
+ req->rq_no_delay = req->rq_no_resend = 1;
+
+ cfs_spin_lock_init(&req->rq_lock);
+ CFS_INIT_LIST_HEAD(&req->rq_list);
+ CFS_INIT_LIST_HEAD(&req->rq_replay_list);
+ CFS_INIT_LIST_HEAD(&req->rq_set_chain);
+ CFS_INIT_LIST_HEAD(&req->rq_history_list);
+ CFS_INIT_LIST_HEAD(&req->rq_exp_list);
+ cfs_waitq_init(&req->rq_reply_waitq);
+ cfs_waitq_init(&req->rq_set_waitq);
+ cfs_atomic_set(&req->rq_refcount, 1);
+
+ CLASSERT (sizeof(*args) <= sizeof(req->rq_async_args));
+ args = ptlrpc_req_async_args(req);
+ args->magic = PTLRPC_WORK_MAGIC;
+ args->cb = cb;
+ args->cbdata = cbdata;
+
+ RETURN(req);
+}
+EXPORT_SYMBOL(ptlrpcd_alloc_work);
+
+void ptlrpcd_destroy_work(void *handler)
+{
+ struct ptlrpc_request *req = handler;
+
+ if (req)
+ ptlrpc_req_finished(req);
+}
+EXPORT_SYMBOL(ptlrpcd_destroy_work);
+
+int ptlrpcd_queue_work(void *handler)
+{
+ struct ptlrpc_request *req = handler;
+
+ /*
+ * Check if the req is already being queued.
+ *
+ * Here comes a trick: it lacks a way of checking if a req is being
+ * processed reliably in ptlrpc. Here I have to use refcount of req
+ * for this purpose. This is okay because the caller should use this
+ * req as opaque data. - Jinshan
+ */
+ LASSERT(cfs_atomic_read(&req->rq_refcount) > 0);
+ if (cfs_atomic_read(&req->rq_refcount) > 1)
+ return -EBUSY;
+
+ if (cfs_atomic_inc_return(&req->rq_refcount) > 2) { /* race */
+ cfs_atomic_dec(&req->rq_refcount);
+ return -EBUSY;
+ }
+
+ /* re-initialize the req */
+ req->rq_timeout = obd_timeout;
+ req->rq_sent = cfs_time_current_sec();
+ req->rq_deadline = req->rq_sent + req->rq_timeout;
+ req->rq_reply_deadline = req->rq_deadline;
+ req->rq_phase = RQ_PHASE_INTERPRET;
+ req->rq_next_phase = RQ_PHASE_COMPLETE;
+ req->rq_xid = ptlrpc_next_xid();
+ req->rq_import_generation = req->rq_import->imp_generation;
+
+ ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
+ return 0;
+}
+EXPORT_SYMBOL(ptlrpcd_queue_work);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff