} md_iov;
} lnet_libmd_t;
-#define LNET_MD_FLAG_ZOMBIE (1 << 0)
-#define LNET_MD_FLAG_AUTO_UNLINK (1 << 1)
+#define LNET_MD_FLAG_ZOMBIE (1 << 0)
+#define LNET_MD_FLAG_AUTO_UNLINK (1 << 1)
+#define LNET_MD_FLAG_ABORTED (1 << 2)
#ifdef LNET_USE_LIB_FREELIST
typedef struct
/**
* Unlink the memory descriptor from any ME it may be linked to and release
- * the internal resources associated with it.
+ * the internal resources associated with it. As a result, active messages
+ * associated with the MD may get aborted.
*
* This function does not free the memory region associated with the MD;
* i.e., the memory the user allocated for this MD. If the ME associated with
return -ENOENT;
}
- /* If the MD is busy, lnet_md_unlink just marks it for deletion, and
- * when the NAL is done, the completion event flags that the MD was
- * unlinked. Otherwise, we enqueue an event now... */
-
- if (md->md_eq != NULL &&
- md->md_refcount == 0) {
- lnet_build_unlink_event(md, &ev);
+ md->md_flags |= LNET_MD_FLAG_ABORTED;
+ /* If the MD is busy, lnet_md_unlink just marks it for deletion, and
+ * when the LND is done, the completion event flags that the MD was
+ * unlinked. Otherwise, we enqueue an event now... */
+ if (md->md_eq != NULL && md->md_refcount == 0) {
+ lnet_build_unlink_event(md, &ev);
lnet_eq_enqueue_event(md->md_eq, &ev);
- }
+ }
lnet_md_unlink(md);
return -ENOENT;
}
- md = me->me_md;
- if (md != NULL &&
- md->md_eq != NULL &&
- md->md_refcount == 0) {
- lnet_build_unlink_event(md, &ev);
- lnet_eq_enqueue_event(md->md_eq, &ev);
+ md = me->me_md;
+ if (md != NULL) {
+ md->md_flags |= LNET_MD_FLAG_ABORTED;
+ if (md->md_eq != NULL && md->md_refcount == 0) {
+ lnet_build_unlink_event(md, &ev);
+ lnet_eq_enqueue_event(md->md_eq, &ev);
+ }
}
lnet_me_unlink(me);
return 0;
}
-int
+/**
+ * \param msg The message to be sent.
+ * \param do_send True if lnet_ni_send() should be called in this function.
+ * lnet_send() is going to lnet_net_unlock immediately after this, so
+ * it sets do_send FALSE and I don't do the unlock/send/lock bit.
+ *
+ * \retval 0 If \a msg sent or OK to send.
+ * \retval EAGAIN If \a msg blocked for credit.
+ * \retval EHOSTUNREACH If the next hop of the message appears dead.
+ * \retval ECANCELED If the MD of the message has been unlinked.
+ */
+static int
lnet_post_send_locked(lnet_msg_t *msg, int do_send)
{
- /* lnet_send is going to lnet_net_unlock immediately after this,
- * so it sets do_send FALSE and I don't do the unlock/send/lock bit.
- * I return EAGAIN if msg blocked, EHOSTUNREACH if msg_txpeer
- * appears dead, and 0 if sent or OK to send */
- struct lnet_peer *lp = msg->msg_txpeer;
- struct lnet_ni *ni = lp->lp_ni;
- struct lnet_tx_queue *tq;
- int cpt;
+ lnet_peer_t *lp = msg->msg_txpeer;
+ lnet_ni_t *ni = lp->lp_ni;
+ int cpt = msg->msg_tx_cpt;
+ struct lnet_tx_queue *tq = ni->ni_tx_queues[cpt];
/* non-lnet_send() callers have checked before */
LASSERT(!do_send || msg->msg_tx_delayed);
LASSERT(!msg->msg_receiving);
LASSERT(msg->msg_tx_committed);
- cpt = msg->msg_tx_cpt;
- tq = ni->ni_tx_queues[cpt];
-
/* NB 'lp' is always the next hop */
if ((msg->msg_target.pid & LNET_PID_USERFLAG) == 0 &&
lnet_peer_alive_locked(lp) == 0) {
return EHOSTUNREACH;
}
+ if (msg->msg_md != NULL &&
+ (msg->msg_md->md_flags & LNET_MD_FLAG_ABORTED) != 0) {
+ lnet_net_unlock(cpt);
+
+ CNETERR("Aborting message for %s: LNetM[DE]Unlink() already "
+ "called on the MD/ME.\n",
+ libcfs_id2str(msg->msg_target));
+ if (do_send)
+ lnet_finalize(ni, msg, -ECANCELED);
+
+ lnet_net_lock(cpt);
+ return ECANCELED;
+ }
+
if (!msg->msg_peertxcredit) {
LASSERT ((lp->lp_txcredits < 0) ==
!cfs_list_empty(&lp->lp_txq));
rc = lnet_post_send_locked(msg, 0);
lnet_net_unlock(cpt);
- if (rc == EHOSTUNREACH)
- return -EHOSTUNREACH;
+ if (rc == EHOSTUNREACH || rc == ECANCELED)
+ return -rc;
- if (rc == 0)
+ if (rc == 0)
lnet_ni_send(src_ni, msg);
- return 0;
+ return 0; /* rc == 0 or EAGAIN */
}
static void
lnet_res_unlock(cpt);
lnet_msg_free(msg);
-
return -ENOENT;
}
lnet_build_msg_event(msg, LNET_EVENT_SEND);
rc = lnet_send(self, msg, LNET_NID_ANY);
- if (rc < 0) {
- CNETERR( "Error sending GET to %s: %d\n",
- libcfs_id2str(target), rc);
- lnet_finalize (NULL, msg, rc);
- }
+ if (rc < 0) {
+ CNETERR("Error sending GET to %s: %d\n",
+ libcfs_id2str(target), rc);
+ lnet_finalize(NULL, msg, rc);
+ }
/* completion will be signalled by an event */
return 0;
LASSERT(!msg->msg_routing);
msg->msg_md = md;
- if (msg->msg_receiving) { /* commited for receiving */
+ if (msg->msg_receiving) { /* committed for receiving */
msg->msg_offset = offset;
msg->msg_wanted = mlen;
}
* NB: message is committed for sending, we should return
* on success because LND will finalize this message later.
*
- * Also, there is possibility that message is commited for
+ * Also, there is possibility that message is committed for
* sending and also failed before delivering to LND,
* i.e: ENOMEM, in that case we can't fall through either
* because CPT for sending can be different with CPT for
* NB: message is committed for sending, we should return
* on success because LND will finalize this message later.
*
- * Also, there is possibility that message is commited for
+ * Also, there is possibility that message is committed for
* sending and also failed before delivering to LND,
* i.e: ENOMEM, in that case we can't fall through either:
* - The rule is message must decommit for sending first if
again:
rc = 0;
if (!msg->msg_tx_committed && !msg->msg_rx_committed) {
- /* not commited to network yet */
+ /* not committed to network yet */
LASSERT(!msg->msg_onactivelist);
lnet_msg_free(msg);
return;
}
/*
- * NB: routed message can be commited for both receiving and sending,
+ * NB: routed message can be committed for both receiving and sending,
* we should finalize in LIFO order and keep counters correct.
* (finalize sending first then finalize receiving)
*/
git://git.whamcloud.com / fs / lustre-release.git / commitdiff