LU-5319 ptlrpc: embed highest XID in each request

Atomically assign XIDs and put request and sending list so
we can learn the lowest unreplied XID at any point.

This allows to embed in every resquests the highest XID for
which a reply has been received and does not have an unreplied
lower-numbered XID.

This will be used by the MDT target to release in-memory
reply data corresponding to XIDs of reply received by the client.

Signed-off-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Signed-off-by: Gregoire Pichon <gregoire.pichon@bull.net>
Change-Id: Ic88fb6db704d8e9a78a34fe16f64abb2cdffc4c4
Reviewed-on: http://review.whamcloud.com/14793
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/include/lustre_export.h b/lustre/include/lustre_export.h
index ce40d58..fe41b2b 100644 (file)
--- a/lustre/include/lustre_export.h
+++ b/lustre/include/lustre_export.h
@@ -268,6 +268,11 @@ struct obd_export {
         } u;
 
        struct adaptive_timeout    exp_bl_lock_at;
+
+       /** highest XID received by export client that has no
+        * unreceived lower-numbered XID
+        */
+       __u64                     exp_last_xid;
 };
 
 #define exp_target_data u.eu_target_data

diff --git a/lustre/ptlrpc/client.c b/lustre/ptlrpc/client.c
index 8f075e8..18de641 100644 (file)
--- a/lustre/ptlrpc/client.c
+++ b/lustre/ptlrpc/client.c
@@ -695,7 +695,6 @@ static int __ptlrpc_request_bufs_pack(struct ptlrpc_request *request,
 
         ptlrpc_at_set_req_timeout(request);
 
-       request->rq_xid = ptlrpc_next_xid();
        lustre_msg_set_opc(request->rq_reqmsg, opcode);
 
        RETURN(0);
@@ -1480,6 +1479,8 @@ static int after_reply(struct ptlrpc_request *req)
 static int ptlrpc_send_new_req(struct ptlrpc_request *req)
 {
         struct obd_import     *imp = req->rq_import;
+       struct list_head      *tmp;
+       __u64                  min_xid = ~0ULL;
         int rc;
         ENTRY;
 
@@ -1493,6 +1494,16 @@ static int ptlrpc_send_new_req(struct ptlrpc_request *req)
 
        spin_lock(&imp->imp_lock);
 
+       /* the very first time we assign XID. it's important to assign XID
+        * and put it on the list atomically, so that the lowest assigned
+        * XID is always known. this is vital for multislot last_rcvd */
+       if (req->rq_send_state == LUSTRE_IMP_REPLAY) {
+               LASSERT(req->rq_xid != 0);
+       } else {
+               LASSERT(req->rq_xid == 0);
+               req->rq_xid = ptlrpc_next_xid();
+       }
+
        if (!req->rq_generation_set)
                req->rq_import_generation = imp->imp_generation;
 
@@ -1522,8 +1533,25 @@ static int ptlrpc_send_new_req(struct ptlrpc_request *req)
        LASSERT(list_empty(&req->rq_list));
        list_add_tail(&req->rq_list, &imp->imp_sending_list);
        atomic_inc(&req->rq_import->imp_inflight);
+
+       /* find the lowest unreplied XID */
+       list_for_each(tmp, &imp->imp_delayed_list) {
+               struct ptlrpc_request *r;
+               r = list_entry(tmp, struct ptlrpc_request, rq_list);
+               if (r->rq_xid < min_xid)
+                       min_xid = r->rq_xid;
+       }
+       list_for_each(tmp, &imp->imp_sending_list) {
+               struct ptlrpc_request *r;
+               r = list_entry(tmp, struct ptlrpc_request, rq_list);
+               if (r->rq_xid < min_xid)
+                       min_xid = r->rq_xid;
+       }
        spin_unlock(&imp->imp_lock);
 
+       if (likely(min_xid != ~0ULL))
+               lustre_msg_set_last_xid(req->rq_reqmsg, min_xid - 1);
+
        lustre_msg_set_status(req->rq_reqmsg, current_pid());
 
         rc = sptlrpc_req_refresh_ctx(req, -1);

diff --git a/lustre/target/tgt_handler.c b/lustre/target/tgt_handler.c
index 5be36e2..a8091d5 100644 (file)
--- a/lustre/target/tgt_handler.c
+++ b/lustre/target/tgt_handler.c
@@ -667,6 +667,25 @@ int tgt_request_handle(struct ptlrpc_request *req)
                GOTO(out, rc);
        }
 
+       /* check request's xid is consistent with export's last_xid */
+       if (req->rq_export != NULL) {
+               __u64 last_xid = lustre_msg_get_last_xid(req->rq_reqmsg);
+               if (last_xid != 0)
+                       req->rq_export->exp_last_xid = last_xid;
+               if (req->rq_xid == 0 ||
+                   req->rq_xid <= req->rq_export->exp_last_xid) {
+                       DEBUG_REQ(D_ERROR, req,
+                                 "Unexpected xid %llx vs. last_xid %llx\n",
+                                 req->rq_xid, req->rq_export->exp_last_xid);
+#if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 7, 93, 0)
+                       LBUG();
+#endif
+                       req->rq_status = -EPROTO;
+                       rc = ptlrpc_error(req);
+                       GOTO(out, rc);
+               }
+       }
+
        request_fail_id = tgt->lut_request_fail_id;
        tsi->tsi_reply_fail_id = tgt->lut_reply_fail_id;