@@ -492,7 +492,8 @@ struct ptlrpc_reply_state {
unsigned long rs_scheduled:1; /* being handled? */
unsigned long rs_scheduled_ever:1; /* any schedule attempts? */
unsigned long rs_handled:1; /* been handled yet? */
- unsigned long rs_on_net:1; /* reply_out_callback pending? */
+ unsigned long rs_sent:1; /* Got LNET_EVENT_SEND? */
+ unsigned long rs_unlinked:1; /* Reply MD unlinked? */
unsigned long rs_prealloc:1; /* rs from prealloc list */
unsigned long rs_committed:1; /* the transaction was committed
* and the rs was dispatched
@@ -813,7 +813,8 @@ void target_send_reply(struct ptlrpc_request *req, int rc, int fail_id)
LASSERT(!rs->rs_scheduled);
LASSERT(!rs->rs_scheduled_ever);
LASSERT(!rs->rs_handled);
- LASSERT(!rs->rs_on_net);
+ LASSERT(!rs->rs_sent);
+ LASSERT(!rs->rs_unlinked);
LASSERT(!rs->rs_export);
LASSERT(list_empty(&rs->rs_obd_list));
LASSERT(list_empty(&rs->rs_exp_list));
@@ -822,7 +823,8 @@ void target_send_reply(struct ptlrpc_request *req, int rc, int fail_id)
/* disable reply scheduling while I'm setting up */
rs->rs_scheduled = 1;
- rs->rs_on_net = 1;
+ rs->rs_sent = 0;
+ rs->rs_unlinked = 0;
rs->rs_xid = req->rq_xid;
rs->rs_transno = req->rq_transno;
rs->rs_export = exp;
@@ -856,13 +858,14 @@ void target_send_reply(struct ptlrpc_request *req, int rc, int fail_id)
* would have been +1 ref for the net, which
* reply_out_callback leaves alone)
*/
- rs->rs_on_net = 0;
+ rs->rs_sent = 1;
+ rs->rs_unlinked = 1;
ptlrpc_rs_addref(rs);
}
spin_lock(&rs->rs_lock);
if (rs->rs_transno <= exp->exp_last_committed ||
- (!rs->rs_on_net && !rs->rs_no_ack) ||
+ (rs->rs_unlinked && !rs->rs_no_ack) ||
list_empty(&rs->rs_exp_list) || /* completed already */
list_empty(&rs->rs_obd_list)) {
CDEBUG(D_HA, "Schedule reply immediately\n");
@@ -401,6 +401,7 @@ void reply_out_callback(struct lnet_event *ev)
struct ptlrpc_cb_id *cbid = ev->md_user_ptr;
struct ptlrpc_reply_state *rs = cbid->cbid_arg;
struct ptlrpc_service_part *svcpt = rs->rs_svcpt;
+ bool need_schedule = false;
LASSERT(ev->type == LNET_EVENT_SEND ||
ev->type == LNET_EVENT_ACK ||
@@ -415,16 +416,28 @@ void reply_out_callback(struct lnet_event *ev)
return;
}
- LASSERT(rs->rs_on_net);
+ if (ev->type == LNET_EVENT_SEND) {
+ spin_lock(&rs->rs_lock);
+ rs->rs_sent = 1;
+ /* If transaction was committed before the SEND, and the ACK
+ * is lost, then we need to schedule so ptlrpc_hr can unlink
+ * the MD.
+ */
+ if (rs->rs_handled)
+ need_schedule = true;
+ spin_unlock(&rs->rs_lock);
+ }
+
+ if (ev->unlinked || need_schedule) {
+ LASSERT(rs->rs_sent);
- if (ev->unlinked) {
/* Last network callback. The net's ref on 'rs' stays put
* until ptlrpc_handle_rs() is done with it
*/
spin_lock(&svcpt->scp_rep_lock);
spin_lock(&rs->rs_lock);
- rs->rs_on_net = 0;
+ rs->rs_unlinked = ev->unlinked;
if (!rs->rs_no_ack ||
rs->rs_transno <=
rs->rs_export->exp_obd->obd_last_committed ||
@@ -458,7 +458,7 @@ void lustre_free_reply_state(struct ptlrpc_reply_state *rs)
LASSERT(atomic_read(&rs->rs_refcount) == 0);
LASSERT(!rs->rs_difficult || rs->rs_handled);
- LASSERT(!rs->rs_on_net);
+ LASSERT(!rs->rs_difficult || rs->rs_unlinked);
LASSERT(!rs->rs_scheduled);
LASSERT(!rs->rs_export);
LASSERT(rs->rs_nlocks == 0);
@@ -1940,10 +1940,14 @@ static int ptlrpc_handle_rs(struct ptlrpc_reply_state *rs)
libcfs_nid2str(exp->exp_connection->c_peer.nid));
}
- if ((!been_handled && rs->rs_on_net) || nlocks > 0) {
+ if ((rs->rs_sent && !rs->rs_unlinked) || nlocks > 0) {
spin_unlock(&rs->rs_lock);
- if (!been_handled && rs->rs_on_net) {
+ /* We can unlink if the LNET_EVENT_SEND has occurred.
+ * If rs_unlinked is set then MD is already unlinked and no
+ * need to do so here.
+ */
+ if ((rs->rs_sent && !rs->rs_unlinked)) {
LNetMDUnlink(rs->rs_md_h);
/* Ignore return code; we're racing with completion */
}
@@ -1957,7 +1961,7 @@ static int ptlrpc_handle_rs(struct ptlrpc_reply_state *rs)
rs->rs_scheduled = 0;
- if (!rs->rs_on_net) {
+ if (rs->rs_unlinked) {
/* Off the net */
spin_unlock(&rs->rs_lock);