LU-7079 ptlrpc: imp_peer_committed_transno should increase

[fs/lustre-release.git] / lustre / ptlrpc / client.c

diff --git a/lustre/ptlrpc/client.c b/lustre/ptlrpc/client.c
index d874d70..a791abf 100644 (file)
--- a/lustre/ptlrpc/client.c
+++ b/lustre/ptlrpc/client.c
@@ -526,7 +526,7 @@ EXPORT_SYMBOL(ptlrpc_free_rq_pool);
 /**
  * Allocates, initializes and adds \a num_rq requests to the pool \a pool
  */
-void ptlrpc_add_rqs_to_pool(struct ptlrpc_request_pool *pool, int num_rq)
+int ptlrpc_add_rqs_to_pool(struct ptlrpc_request_pool *pool, int num_rq)
 {
         int i;
         int size = 1;
@@ -548,11 +548,11 @@ void ptlrpc_add_rqs_to_pool(struct ptlrpc_request_pool *pool, int num_rq)
                spin_unlock(&pool->prp_lock);
                req = ptlrpc_request_cache_alloc(GFP_NOFS);
                if (!req)
-                       return;
+                       return i;
                OBD_ALLOC_LARGE(msg, size);
                if (!msg) {
                        ptlrpc_request_cache_free(req);
-                       return;
+                       return i;
                 }
                 req->rq_reqbuf = msg;
                 req->rq_reqbuf_len = size;
@@ -561,7 +561,7 @@ void ptlrpc_add_rqs_to_pool(struct ptlrpc_request_pool *pool, int num_rq)
                list_add_tail(&req->rq_list, &pool->prp_req_list);
        }
        spin_unlock(&pool->prp_lock);
-       return;
+       return num_rq;
 }
 EXPORT_SYMBOL(ptlrpc_add_rqs_to_pool);
 
@@ -575,7 +575,7 @@ EXPORT_SYMBOL(ptlrpc_add_rqs_to_pool);
  */
 struct ptlrpc_request_pool *
 ptlrpc_init_rq_pool(int num_rq, int msgsize,
-                   void (*populate_pool)(struct ptlrpc_request_pool *, int))
+                   int (*populate_pool)(struct ptlrpc_request_pool *, int))
 {
        struct ptlrpc_request_pool *pool;
 
@@ -593,11 +593,6 @@ ptlrpc_init_rq_pool(int num_rq, int msgsize,
 
        populate_pool(pool, num_rq);
 
-       if (list_empty(&pool->prp_req_list)) {
-               /* have not allocated a single request for the pool */
-               OBD_FREE(pool, sizeof(struct ptlrpc_request_pool));
-               pool = NULL;
-       }
        return pool;
 }
 EXPORT_SYMBOL(ptlrpc_init_rq_pool);
@@ -770,11 +765,10 @@ struct ptlrpc_request *__ptlrpc_request_alloc(struct obd_import *imp,
 {
        struct ptlrpc_request *request = NULL;
 
-       if (pool)
-               request = ptlrpc_prep_req_from_pool(pool);
+       request = ptlrpc_request_cache_alloc(GFP_NOFS);
 
-       if (!request)
-               request = ptlrpc_request_cache_alloc(GFP_NOFS);
+       if (!request && pool)
+               request = ptlrpc_prep_req_from_pool(pool);
 
        if (request) {
                ptlrpc_cli_req_init(request);
@@ -917,15 +911,17 @@ ptlrpc_prep_req(struct obd_import *imp, __u32 version, int opcode, int count,
 }
 
 /**
- * Allocate and initialize new request set structure.
+ * Allocate and initialize new request set structure on the current CPT.
  * Returns a pointer to the newly allocated set structure or NULL on error.
  */
 struct ptlrpc_request_set *ptlrpc_prep_set(void)
 {
-       struct ptlrpc_request_set *set;
+       struct ptlrpc_request_set       *set;
+       int                             cpt;
 
        ENTRY;
-       OBD_ALLOC(set, sizeof *set);
+       cpt = cfs_cpt_current(cfs_cpt_table, 0);
+       OBD_CPT_ALLOC(set, cfs_cpt_table, cpt, sizeof *set);
        if (!set)
                RETURN(NULL);
        atomic_set(&set->set_refcount, 1);
@@ -1287,6 +1283,7 @@ static int after_reply(struct ptlrpc_request *req)
         struct obd_device *obd = req->rq_import->imp_obd;
         int rc;
         struct timeval work_start;
+       __u64 committed;
         long timediff;
         ENTRY;
 
@@ -1441,10 +1438,9 @@ static int after_reply(struct ptlrpc_request *req)
                 /*
                  * Replay-enabled imports return commit-status information.
                  */
-                if (lustre_msg_get_last_committed(req->rq_repmsg)) {
-                        imp->imp_peer_committed_transno =
-                                lustre_msg_get_last_committed(req->rq_repmsg);
-                }
+               committed = lustre_msg_get_last_committed(req->rq_repmsg);
+               if (likely(committed > imp->imp_peer_committed_transno))
+                       imp->imp_peer_committed_transno = committed;
 
                ptlrpc_free_committed(imp);
 
@@ -1482,6 +1478,13 @@ static int ptlrpc_send_new_req(struct ptlrpc_request *req)
         ENTRY;
 
         LASSERT(req->rq_phase == RQ_PHASE_NEW);
+
+       /* do not try to go further if there is not enough memory in enc_pool */
+       if (req->rq_sent && req->rq_bulk != NULL)
+               if (req->rq_bulk->bd_iov_count > get_free_pages_in_pool() &&
+                   pool_is_at_full_capacity())
+                       RETURN(-ENOMEM);
+
         if (req->rq_sent && (req->rq_sent > cfs_time_current_sec()) &&
             (!req->rq_generation_set ||
              req->rq_import_generation == imp->imp_generation))
@@ -1572,6 +1575,16 @@ static int ptlrpc_send_new_req(struct ptlrpc_request *req)
               lustre_msg_get_opc(req->rq_reqmsg));
 
         rc = ptl_send_rpc(req, 0);
+       if (rc == -ENOMEM) {
+               spin_lock(&imp->imp_lock);
+               if (!list_empty(&req->rq_list)) {
+                       list_del_init(&req->rq_list);
+                       atomic_dec(&req->rq_import->imp_inflight);
+               }
+               spin_unlock(&imp->imp_lock);
+               ptlrpc_rqphase_move(req, RQ_PHASE_NEW);
+               RETURN(rc);
+       }
         if (rc) {
                 DEBUG_REQ(D_HA, req, "send failed (%d); expect timeout", rc);
                spin_lock(&req->rq_lock);
@@ -1841,6 +1854,14 @@ int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
                                }
 
                                rc = ptl_send_rpc(req, 0);
+                               if (rc == -ENOMEM) {
+                                       spin_lock(&imp->imp_lock);
+                                       if (!list_empty(&req->rq_list))
+                                               list_del_init(&req->rq_list);
+                                       spin_unlock(&imp->imp_lock);
+                                       ptlrpc_rqphase_move(req, RQ_PHASE_NEW);
+                                       continue;
+                               }
                                if (rc) {
                                        DEBUG_REQ(D_HA, req,
                                                  "send failed: rc = %d", rc);
@@ -2733,6 +2754,10 @@ void ptlrpc_retain_replayable_request(struct ptlrpc_request *req,
 
        lustre_msg_add_flags(req->rq_reqmsg, MSG_REPLAY);
 
+       spin_lock(&req->rq_lock);
+       req->rq_resend = 0;
+       spin_unlock(&req->rq_lock);
+
        LASSERT(imp->imp_replayable);
        /* Balanced in ptlrpc_free_committed, usually. */
        ptlrpc_request_addref(req);
@@ -2810,10 +2835,15 @@ static int ptlrpc_replay_interpret(const struct lu_env *env,
        ENTRY;
        atomic_dec(&imp->imp_replay_inflight);
 
-        if (!ptlrpc_client_replied(req)) {
-                CERROR("request replay timed out, restarting recovery\n");
-                GOTO(out, rc = -ETIMEDOUT);
-        }
+       /* Note: if it is bulk replay (MDS-MDS replay), then even if
+        * server got the request, but bulk transfer timeout, let's
+        * replay the bulk req again */
+       if (!ptlrpc_client_replied(req) ||
+           (req->rq_bulk != NULL &&
+            lustre_msg_get_status(req->rq_repmsg) == -ETIMEDOUT)) {
+               DEBUG_REQ(D_ERROR, req, "request replay timed out.\n");
+               GOTO(out, rc = -ETIMEDOUT);
+       }
 
         if (lustre_msg_get_type(req->rq_repmsg) == PTL_RPC_MSG_ERR &&
             (lustre_msg_get_status(req->rq_repmsg) == -ENOTCONN ||
@@ -2928,7 +2958,7 @@ int ptlrpc_replay_req(struct ptlrpc_request *req)
        atomic_inc(&req->rq_import->imp_replay_inflight);
        ptlrpc_request_addref(req);     /* ptlrpcd needs a ref */
 
-       ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
+       ptlrpcd_add_req(req);
        RETURN(0);
 }
 
@@ -3172,7 +3202,7 @@ static void ptlrpcd_add_work_req(struct ptlrpc_request *req)
        req->rq_xid             = ptlrpc_next_xid();
        req->rq_import_generation = req->rq_import->imp_generation;
 
-       ptlrpcd_add_req(req, PDL_POLICY_ROUND, -1);
+       ptlrpcd_add_req(req);
 }
 
 static int work_interpreter(const struct lu_env *env,