return cpt;
}
+static int lnet_md_validate(const struct lnet_md *umd);
+
static struct lnet_libmd *
lnet_md_build(const struct lnet_md *umd, int unlink)
{
struct lnet_libmd *lmd;
unsigned int size;
+ if (lnet_md_validate(umd) != 0)
+ return ERR_PTR(-EINVAL);
+
if (umd->options & LNET_MD_KIOV)
niov = umd->length;
else
}
/* must be called with resource lock held */
-static int
+static void
lnet_md_link(struct lnet_libmd *md, lnet_handler_t handler, int cpt)
{
struct lnet_res_container *container = the_lnet.ln_md_containers[cpt];
/* NB we are passed an allocated, but inactive md.
- * if we return success, caller may lnet_md_unlink() it.
- * otherwise caller may only lnet_md_free() it.
+ * Caller may lnet_md_unlink() it, or may lnet_md_free() it.
*/
/* This implementation doesn't know how to create START events or
* disable END events. Best to LASSERT our caller is compliant so
LASSERT(list_empty(&md->md_list));
list_add(&md->md_list, &container->rec_active);
-
- return 0;
}
void lnet_assert_handler_unused(lnet_handler_t handler)
* \param handle On successful returns, a handle to the newly created MD is
* saved here. This handle can be used later in LNetMDUnlink().
*
+ * The ME will either be linked to the new MD, or it will be freed.
+ *
* \retval 0 On success.
* \retval -EINVAL If \a umd is not valid.
* \retval -ENOMEM If new MD cannot be allocated.
- * \retval -ENOENT Either \a meh or \a umd.eq_handle does not point to a
- * valid object. Note that it's OK to supply a NULL \a umd.eq_handle by
- * calling LNetInvalidateHandle() on it.
- * \retval -EBUSY If the ME pointed to by \a meh is already associated with
- * a MD.
*/
int
LNetMDAttach(struct lnet_me *me, const struct lnet_md *umd,
LIST_HEAD(drops);
struct lnet_libmd *md;
int cpt;
- int rc;
LASSERT(the_lnet.ln_refcount > 0);
-
- if (lnet_md_validate(umd) != 0)
- return -EINVAL;
+ LASSERT(!me->me_md);
if ((umd->options & (LNET_MD_OP_GET | LNET_MD_OP_PUT)) == 0) {
CERROR("Invalid option: no MD_OP set\n");
- return -EINVAL;
- }
-
- md = lnet_md_build(umd, unlink);
- if (IS_ERR(md))
- return PTR_ERR(md);
+ md = ERR_PTR(-EINVAL);
+ } else
+ md = lnet_md_build(umd, unlink);
cpt = me->me_cpt;
-
lnet_res_lock(cpt);
- if (me->me_md)
- rc = -EBUSY;
- else
- rc = lnet_md_link(md, umd->handler, cpt);
+ if (IS_ERR(md)) {
+ lnet_me_unlink(me);
+ lnet_res_unlock(cpt);
+ return PTR_ERR(md);
+ }
- if (rc != 0)
- goto out_unlock;
+ lnet_md_link(md, umd->handler, cpt);
/* attach this MD to portal of ME and check if it matches any
* blocked msgs on this portal */
lnet_recv_delayed_msg_list(&matches);
return 0;
-
-out_unlock:
- lnet_res_unlock(cpt);
- lnet_md_free(md);
- return rc;
}
EXPORT_SYMBOL(LNetMDAttach);
* \retval 0 On success.
* \retval -EINVAL If \a umd is not valid.
* \retval -ENOMEM If new MD cannot be allocated.
- * \retval -ENOENT \a umd.eq_handle does not point to a valid EQ. Note that
- * it's OK to supply a NULL \a umd.eq_handle by calling
- * LNetInvalidateHandle() on it.
*/
int
LNetMDBind(const struct lnet_md *umd, enum lnet_unlink unlink,
LASSERT(the_lnet.ln_refcount > 0);
- if (lnet_md_validate(umd) != 0)
- return -EINVAL;
-
if ((umd->options & (LNET_MD_OP_GET | LNET_MD_OP_PUT)) != 0) {
CERROR("Invalid option: GET|PUT illegal on active MDs\n");
return -EINVAL;
cpt = lnet_res_lock_current();
- rc = lnet_md_link(md, umd->handler, cpt);
- if (rc != 0)
- goto out_unlock;
+ lnet_md_link(md, umd->handler, cpt);
lnet_md2handle(handle, md);
lnet_res_unlock(cpt);
return 0;
- out_unlock:
- lnet_res_unlock(cpt);
-
out_free:
lnet_md_free(md);
return rc;
CERROR("%s: LNetMDAttach failed x%llu/%d: rc = %d\n",
desc->bd_import->imp_obd->obd_name, mbits,
posted_md, rc);
- LNetMEUnlink(me);
break;
}
}
LNetInvalidateMDHandle(&bulk_cookie);
- if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_DROP_RPC))
- RETURN(0);
+ if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_DROP_RPC))
+ RETURN(0);
- LASSERT(request->rq_type == PTL_RPC_MSG_REQUEST);
- LASSERT(request->rq_wait_ctx == 0);
+ LASSERT(request->rq_type == PTL_RPC_MSG_REQUEST);
+ LASSERT(request->rq_wait_ctx == 0);
- /* If this is a re-transmit, we're required to have disengaged
- * cleanly from the previous attempt */
- LASSERT(!request->rq_receiving_reply);
+ /* If this is a re-transmit, we're required to have disengaged
+ * cleanly from the previous attempt */
+ LASSERT(!request->rq_receiving_reply);
LASSERT(!((lustre_msg_get_flags(request->rq_reqmsg) & MSG_REPLAY) &&
- (imp->imp_state == LUSTRE_IMP_FULL)));
+ (imp->imp_state == LUSTRE_IMP_FULL)));
if (unlikely(obd != NULL && obd->obd_fail)) {
CDEBUG(D_HA, "muting rpc for failed imp obd %s\n",
- obd->obd_name);
+ obd->obd_name);
/* this prevents us from waiting in ptlrpc_queue_wait */
spin_lock(&request->rq_lock);
request->rq_err = 1;
spin_unlock(&request->rq_lock);
- request->rq_status = -ENODEV;
- RETURN(-ENODEV);
- }
+ request->rq_status = -ENODEV;
+ RETURN(-ENODEV);
+ }
connection = imp->imp_connection;
LASSERT(AT_OFF || imp->imp_state != LUSTRE_IMP_FULL ||
(imp->imp_msghdr_flags & MSGHDR_AT_SUPPORT) ||
!(imp->imp_connect_data.ocd_connect_flags &
- OBD_CONNECT_AT));
+ OBD_CONNECT_AT));
if (request->rq_resend) {
lustre_msg_add_flags(request->rq_reqmsg, MSG_RESENT);
bulk_cookie = request->rq_bulk->bd_mds[0];
}
- if (!noreply) {
- LASSERT (request->rq_replen != 0);
- if (request->rq_repbuf == NULL) {
- LASSERT(request->rq_repdata == NULL);
- LASSERT(request->rq_repmsg == NULL);
- rc = sptlrpc_cli_alloc_repbuf(request,
- request->rq_replen);
- if (rc) {
- /* this prevents us from looping in
- * ptlrpc_queue_wait */
+ if (!noreply) {
+ LASSERT (request->rq_replen != 0);
+ if (request->rq_repbuf == NULL) {
+ LASSERT(request->rq_repdata == NULL);
+ LASSERT(request->rq_repmsg == NULL);
+ rc = sptlrpc_cli_alloc_repbuf(request,
+ request->rq_replen);
+ if (rc) {
+ /* this prevents us from looping in
+ * ptlrpc_queue_wait */
spin_lock(&request->rq_lock);
request->rq_err = 1;
spin_unlock(&request->rq_lock);
request->rq_receiving_reply = !noreply;
/* Clear any flags that may be present from previous sends. */
request->rq_req_unlinked = 0;
- request->rq_replied = 0;
- request->rq_err = 0;
- request->rq_timedout = 0;
- request->rq_net_err = 0;
- request->rq_resend = 0;
- request->rq_restart = 0;
+ request->rq_replied = 0;
+ request->rq_err = 0;
+ request->rq_timedout = 0;
+ request->rq_net_err = 0;
+ request->rq_resend = 0;
+ request->rq_restart = 0;
request->rq_reply_truncated = 0;
spin_unlock(&request->rq_lock);
- if (!noreply) {
- reply_md.start = request->rq_repbuf;
- reply_md.length = request->rq_repbuf_len;
- /* Allow multiple early replies */
- reply_md.threshold = LNET_MD_THRESH_INF;
- /* Manage remote for early replies */
- reply_md.options = PTLRPC_MD_OPTIONS | LNET_MD_OP_PUT |
- LNET_MD_MANAGE_REMOTE |
- LNET_MD_TRUNCATE; /* allow to make EOVERFLOW error */;
+ if (!noreply) {
+ reply_md.start = request->rq_repbuf;
+ reply_md.length = request->rq_repbuf_len;
+ /* Allow multiple early replies */
+ reply_md.threshold = LNET_MD_THRESH_INF;
+ /* Manage remote for early replies */
+ reply_md.options = PTLRPC_MD_OPTIONS | LNET_MD_OP_PUT |
+ LNET_MD_MANAGE_REMOTE |
+ LNET_MD_TRUNCATE; /* allow to make EOVERFLOW error */;
reply_md.user_ptr = &request->rq_reply_cbid;
reply_md.handler = ptlrpc_handler;
/* ...but the MD attach didn't succeed... */
request->rq_receiving_reply = 0;
spin_unlock(&request->rq_lock);
- GOTO(cleanup_me, rc = -ENOMEM);
+ GOTO(cleanup_bulk, rc = -ENOMEM);
}
percpu_ref_get(&ptlrpc_pending);
request->rq_reply_portal);
}
- /* add references on request for request_out_callback */
- ptlrpc_request_addref(request);
+ /* add references on request for request_out_callback */
+ ptlrpc_request_addref(request);
if (obd != NULL && obd->obd_svc_stats != NULL)
lprocfs_counter_add(obd->obd_svc_stats, PTLRPC_REQACTIVE_CNTR,
- atomic_read(&imp->imp_inflight));
+ atomic_read(&imp->imp_inflight));
OBD_FAIL_TIMEOUT(OBD_FAIL_PTLRPC_DELAY_SEND, request->rq_timeout + 5);
request->rq_sent_ns = ktime_get_real();
request->rq_sent = ktime_get_real_seconds();
/* We give the server rq_timeout secs to process the req, and
- add the network latency for our local timeout. */
- request->rq_deadline = request->rq_sent + request->rq_timeout +
- ptlrpc_at_get_net_latency(request);
+ * add the network latency for our local timeout.
+ */
+ request->rq_deadline = request->rq_sent + request->rq_timeout +
+ ptlrpc_at_get_net_latency(request);
ptlrpc_pinger_sending_on_import(imp);
GOTO(out, rc);
request->rq_req_unlinked = 1;
- ptlrpc_req_finished(request);
- if (noreply)
- GOTO(out, rc);
-
- cleanup_me:
- /* MEUnlink is safe; the PUT didn't even get off the ground, and
- * nobody apart from the PUT's target has the right nid+XID to
- * access the reply buffer.
- */
- LNetMEUnlink(reply_me);
+ ptlrpc_req_finished(request);
+ if (noreply)
+ GOTO(out, rc);
+
+ LNetMDUnlink(request->rq_reply_md_h);
+
/* UNLINKED callback called synchronously */
LASSERT(!request->rq_receiving_reply);
CERROR("ptlrpc: LNetMDAttach failed: rc = %d\n", rc);
LASSERT(rc == -ENOMEM);
- LNetMEUnlink(me);
LASSERT(rc == 0);
rqbd->rqbd_refcount = 0;