branch: b1_6

fix a race condition of early reply release.
b=14274
r=nathan, andreas

lustre/include/lustre_net.h

@@ -284,7 +284,8 @@ struct ptlrpc_request {
                 rq_no_resend:1, rq_waiting:1, rq_receiving_reply:1,
                 rq_no_delay:1, rq_net_err:1, rq_early:1, rq_must_unlink:1,
                 /* server-side flags */
-                rq_packed_final:1;  /* packed final reply */
+                rq_packed_final:1,  /* packed final reply */
+                rq_sent_final:1;    /* stop sending early replies */
         enum rq_phase rq_phase; /* one of RQ_PHASE_* */
         atomic_t rq_refcount;   /* client-side refcount for SENT race,
                                    server-side refcounf for multiple replies */

lustre/ptlrpc/service.c

@@ -378,6 +378,7 @@ static void ptlrpc_server_req_decref(struct ptlrpc_request *req)
         if (!atomic_dec_and_test(&req->rq_refcount))
                 return;
 
+        LASSERT(list_empty(&req->rq_timed_list));
         if (req != &rqbd->rqbd_req) {
                 /* NB request buffers use an embedded
                  * req if the incoming req unlinked the
@@ -412,6 +413,7 @@ static void ptlrpc_server_free_request(struct ptlrpc_request *req)
         if (req->rq_phase != RQ_PHASE_NEW) /* incorrect message magic */
                 DEBUG_REQ(D_INFO, req, "free req");
         spin_lock(&svc->srv_at_lock);
+        req->rq_sent_final = 1;
         list_del_init(&req->rq_timed_list);
         spin_unlock(&svc->srv_at_lock);
 
@@ -614,6 +616,11 @@ static int ptlrpc_at_add_timed(struct ptlrpc_request *req)
         
         spin_lock(&svc->srv_at_lock);
 
+        if (unlikely(req->rq_sent_final)) {
+                spin_unlock(&svc->srv_at_lock);
+                return 0;
+        }
+
         LASSERT(list_empty(&req->rq_timed_list));
         /* Add to sorted list.  Presumably latest rpcs will have the latest
            deadlines, so search backward. */
@@ -709,6 +716,12 @@ static int ptlrpc_at_send_early_reply(struct ptlrpc_request *req,
         reqcopy->rq_reqmsg = reqmsg;
         memcpy(reqmsg, req->rq_reqmsg, req->rq_reqlen);
 
+        if (req->rq_sent_final) {
+                CDEBUG(D_ADAPTTO, "x"LPU64": normal reply already sent out, "
+                       "abort sending early reply\n", req->rq_xid);
+                GOTO(out, rc = 0);
+        }
+
         /* Connection ref */
         reqcopy->rq_export = class_conn2export(
                                      lustre_msg_get_handle(reqcopy->rq_reqmsg));