LU-1496 ptlrpc: prolong rw locks even IO RPCs are finished

Refresh rw lock again after IO RPCs are finished to leave a time
window for clients to cancel covering dlm locks.

This is a part of LU-874 back port.

Signed-off-by: Bobi Jam <bobijam@whamcloud.com>
Change-Id: I5cd185870e601a66bce21b3cc3c91f5f800b4c27
Reviewed-on: http://review.whamcloud.com/3157
Tested-by: Hudson
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Jinshan Xiong <jinshan.xiong@whamcloud.com>
Reviewed-by: Fan Yong <yong.fan@whamcloud.com>
Reviewed-by: Johann Lombardi <johann@whamcloud.com>

diff --git a/lustre/include/lustre_net.h b/lustre/include/lustre_net.h
index 1efff00..c1b9a10 100644 (file)
--- a/lustre/include/lustre_net.h
+++ b/lustre/include/lustre_net.h
@@ -291,6 +291,10 @@ struct ptlrpc_hpreq_ops {
          * Check if the request is a high priority one.
          */
         int  (*hpreq_check)(struct ptlrpc_request *);
+        /**
+         * Called after the request has been handled.
+         */
+        void (*hpreq_fini)(struct ptlrpc_request *);
 };
 
 struct ptlrpc_request {

diff --git a/lustre/ost/ost_handler.c b/lustre/ost/ost_handler.c
index b6f2a77..c1f3b4a 100644 (file)
--- a/lustre/ost/ost_handler.c
+++ b/lustre/ost/ost_handler.c
@@ -1606,6 +1606,11 @@ static int ost_rw_hpreq_check(struct ptlrpc_request *req)
         RETURN(ost_rw_prolong_locks(req, ioo, nb, &body->oa, mode));
 }
 
+static void ost_rw_hpreq_fini(struct ptlrpc_request *req)
+{
+        (void)ost_rw_hpreq_check(req);
+}
+
 static int ost_punch_prolong_locks(struct ptlrpc_request *req, struct obdo *oa)
 {
         struct ldlm_res_id res_id = { .name = { oa->o_id } };
@@ -1668,14 +1673,21 @@ static int ost_punch_hpreq_check(struct ptlrpc_request *req)
         RETURN(ost_punch_prolong_locks(req, &body->oa));
 }
 
+static void ost_punch_hpreq_fini(struct ptlrpc_request *req)
+{
+        (void)ost_punch_hpreq_check(req);
+}
+
 struct ptlrpc_hpreq_ops ost_hpreq_rw = {
         .hpreq_lock_match  = ost_rw_hpreq_lock_match,
         .hpreq_check       = ost_rw_hpreq_check,
+        .hpreq_fini        = ost_rw_hpreq_fini,
 };
 
 struct ptlrpc_hpreq_ops ost_hpreq_punch = {
         .hpreq_lock_match  = ost_punch_hpreq_lock_match,
         .hpreq_check       = ost_punch_hpreq_check,
+        .hpreq_fini        = ost_punch_hpreq_fini,
 };
 
 /** Assign high priority operations to the request if needed. */

diff --git a/lustre/ptlrpc/service.c b/lustre/ptlrpc/service.c
index 50256bc..673e029 100644 (file)
--- a/lustre/ptlrpc/service.c
+++ b/lustre/ptlrpc/service.c
@@ -64,6 +64,7 @@ CFS_MODULE_PARM(at_extra, "i", int, 0644,
 
 /* forward ref */
 static int ptlrpc_server_post_idle_rqbds (struct ptlrpc_service *svc);
+static void ptlrpc_hpreq_fini(struct ptlrpc_request *req);
 
 static CFS_LIST_HEAD (ptlrpc_all_services);
 spinlock_t ptlrpc_all_services_lock;
@@ -558,6 +559,7 @@ void ptlrpc_server_drop_request(struct ptlrpc_request *req)
  */
 static void ptlrpc_server_finish_request(struct ptlrpc_request *req)
 {
+        ptlrpc_hpreq_fini(req);
         ptlrpc_server_drop_request(req);
 }
 
@@ -991,7 +993,7 @@ static int ptlrpc_at_check_timed(struct ptlrpc_service *svc)
 static int ptlrpc_hpreq_init(struct ptlrpc_service *svc,
                              struct ptlrpc_request *req)
 {
-        int rc;
+        int rc = 0;
         ENTRY;
 
         if (svc->srv_hpreq_handler) {
@@ -1000,12 +1002,18 @@ static int ptlrpc_hpreq_init(struct ptlrpc_service *svc,
                         RETURN(rc);
         }
         if (req->rq_export && req->rq_ops) {
+                /* Perform request specific check. We should do this check
+                 * before the request is added into exp_queued_rpc list
+                 * otherwise it may hit swab race at LU-1044. */
+                if (req->rq_ops->hpreq_check)
+                        rc = req->rq_ops->hpreq_check(req);
+
                 spin_lock(&req->rq_export->exp_lock);
                 list_add(&req->rq_exp_list, &req->rq_export->exp_queued_rpc);
                 spin_unlock(&req->rq_export->exp_lock);
         }
 
-        RETURN(0);
+        RETURN(rc);
 }
 
 /** Remove the request from the export list. */
@@ -1013,6 +1021,11 @@ static void ptlrpc_hpreq_fini(struct ptlrpc_request *req)
 {
         ENTRY;
         if (req->rq_export && req->rq_ops) {
+                /* refresh lock timeout again so that client has more
+                 * room to send lock cancel RPC. */
+                if (req->rq_ops->hpreq_fini)
+                        req->rq_ops->hpreq_fini(req);
+
                 spin_lock(&req->rq_export->exp_lock);
                 list_del_init(&req->rq_exp_list);
                 spin_unlock(&req->rq_export->exp_lock);
@@ -1043,7 +1056,7 @@ static void ptlrpc_hpreq_reorder_nolock(struct ptlrpc_service *svc,
                 list_move_tail(&req->rq_list, &svc->srv_request_hpq);
                 req->rq_hp = 1;
                 if (opc != OBD_PING)
-                        DEBUG_REQ(D_NET, req, "high priority req");
+                        DEBUG_REQ(D_RPCTRACE, req, "high priority req");
         }
         spin_unlock(&req->rq_lock);
         EXIT;
@@ -1064,20 +1077,16 @@ void ptlrpc_hpreq_reorder(struct ptlrpc_request *req)
 }
 
 /** Check if the request if a high priority one. */
-static int ptlrpc_server_hpreq_check(struct ptlrpc_request *req)
+static int ptlrpc_server_hpreq_check(struct ptlrpc_service *svc,
+                                     struct ptlrpc_request *req)
 {
-        int opc, rc = 0;
         ENTRY;
 
         /* Check by request opc. */
-        opc = lustre_msg_get_opc(req->rq_reqmsg);
-        if (opc == OBD_PING)
+        if (OBD_PING == lustre_msg_get_opc(req->rq_reqmsg))
                 RETURN(1);
 
-        /* Perform request specific check. */
-        if (req->rq_ops && req->rq_ops->hpreq_check)
-                rc = req->rq_ops->hpreq_check(req);
-        RETURN(rc);
+        RETURN(ptlrpc_hpreq_init(svc, req));
 }
 
 /** Check if a request is a high priority one. */
@@ -1087,7 +1096,7 @@ static int ptlrpc_server_request_add(struct ptlrpc_service *svc,
         int rc;
         ENTRY;
 
-        rc = ptlrpc_server_hpreq_check(req);
+        rc = ptlrpc_server_hpreq_check(svc, req);
         if (rc < 0)
                 RETURN(rc);
 
@@ -1201,7 +1210,7 @@ ptlrpc_server_handle_req_in(struct ptlrpc_service *svc)
                 goto err_req;
         }
 
-        CDEBUG(D_NET, "got req "LPU64"\n", req->rq_xid);
+        CDEBUG(D_RPCTRACE, "got req "LPU64"\n", req->rq_xid);
 
         req->rq_export = class_conn2export(
                 lustre_msg_get_handle(req->rq_reqmsg));
@@ -1229,9 +1238,6 @@ ptlrpc_server_handle_req_in(struct ptlrpc_service *svc)
         }
 
         ptlrpc_at_add_timed(req);
-        rc = ptlrpc_hpreq_init(svc, req);
-        if (rc)
-                GOTO(err_req, rc);
 
         /* Move it over to the request processing queue */
         rc = ptlrpc_server_request_add(svc, req);
@@ -1372,8 +1378,6 @@ ptlrpc_server_handle_request(struct ptlrpc_service *svc,
         ptlrpc_rqphase_move(request, RQ_PHASE_INTERPRET);
         spin_unlock(&svc->srv_lock);
 
-        ptlrpc_hpreq_fini(request);
-
         if(OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_DUMP_LOG))
                 libcfs_debug_dumplog();
 
@@ -2027,7 +2031,6 @@ int ptlrpc_unregister_service(struct ptlrpc_service *service)
                 list_del(&req->rq_list);
                 service->srv_n_queued_reqs--;
                 service->srv_n_active_reqs++;
-                ptlrpc_hpreq_fini(req);
                 ptlrpc_server_finish_request(req);
         }
         LASSERT(service->srv_n_queued_reqs == 0);