LU-13984 ptlrpc: throttle RPC resend if network error

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

diff --git a/lustre/ptlrpc/client.c b/lustre/ptlrpc/client.c
index 6e16826..368732a 100644 (file)
--- a/lustre/ptlrpc/client.c
+++ b/lustre/ptlrpc/client.c
@@ -169,6 +169,12 @@ struct ptlrpc_bulk_desc *ptlrpc_new_bulk(unsigned int nfrags,
 
        LASSERT(ops->add_kiov_frag != NULL);
 
+       if (max_brw > PTLRPC_BULK_OPS_COUNT)
+               RETURN(NULL);
+
+       if (nfrags > LNET_MAX_IOV * max_brw)
+               RETURN(NULL);
+
        OBD_ALLOC_PTR(desc);
        if (!desc)
                return NULL;
@@ -185,6 +191,7 @@ struct ptlrpc_bulk_desc *ptlrpc_new_bulk(unsigned int nfrags,
        desc->bd_portal = portal;
        desc->bd_type = type;
        desc->bd_md_count = 0;
+       desc->bd_nob_last = LNET_MTU;
        desc->bd_frag_ops = ops;
        LASSERT(max_brw > 0);
        desc->bd_md_max_brw = min(max_brw, PTLRPC_BULK_OPS_COUNT);
@@ -253,6 +260,15 @@ void __ptlrpc_prep_bulk_page(struct ptlrpc_bulk_desc *desc,
 
        kiov = &desc->bd_vec[desc->bd_iov_count];
 
+       if (((desc->bd_iov_count % LNET_MAX_IOV) == 0) ||
+            ((desc->bd_nob_last + len) > LNET_MTU)) {
+               desc->bd_mds_off[desc->bd_md_count] = desc->bd_iov_count;
+               desc->bd_md_count++;
+               desc->bd_nob_last = 0;
+               LASSERT(desc->bd_md_count <= PTLRPC_BULK_OPS_COUNT);
+       }
+
+       desc->bd_nob_last += len;
        desc->bd_nob += len;
 
        if (pin)
@@ -272,7 +288,7 @@ void ptlrpc_free_bulk(struct ptlrpc_bulk_desc *desc)
 
        LASSERT(desc != NULL);
        LASSERT(desc->bd_iov_count != LI_POISON); /* not freed already */
-       LASSERT(desc->bd_md_count == 0);         /* network hands off */
+       LASSERT(desc->bd_refs == 0);         /* network hands off */
        LASSERT((desc->bd_export != NULL) ^ (desc->bd_import != NULL));
        LASSERT(desc->bd_frag_ops != NULL);
 
@@ -829,11 +845,12 @@ int ptlrpc_request_bufs_pack(struct ptlrpc_request *request,
                }
 
                if (fail_t) {
-                       *fail_t = ktime_get_real_seconds() + LONG_UNLINK;
+                       *fail_t = ktime_get_real_seconds() +
+                                 PTLRPC_REQ_LONG_UNLINK;
 
                        if (fail2_t)
                                *fail2_t = ktime_get_real_seconds() +
-                                          LONG_UNLINK;
+                                          PTLRPC_REQ_LONG_UNLINK;
 
                        /*
                         * The RPC is infected, let the test to change the
@@ -888,7 +905,7 @@ struct ptlrpc_request *__ptlrpc_request_alloc(struct obd_import *imp,
        if (request) {
                ptlrpc_cli_req_init(request);
 
-               LASSERTF((unsigned long)imp > 0x1000, "%p", imp);
+               LASSERTF((unsigned long)imp > 0x1000, "%p\n", imp);
                LASSERT(imp != LP_POISON);
                LASSERTF((unsigned long)imp->imp_client > 0x1000, "%p\n",
                         imp->imp_client);
@@ -1984,6 +2001,27 @@ int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
                                        GOTO(interpret, req->rq_status);
                                }
 
+                               /* don't resend too fast in case of network
+                                * errors.
+                                */
+                               if (ktime_get_real_seconds() < (req->rq_sent + 1)
+                                   && req->rq_net_err && req->rq_timedout) {
+
+                                       DEBUG_REQ(D_INFO, req,
+                                                 "throttle request");
+                                       /* Don't try to resend RPC right away
+                                        * as it is likely it will fail again
+                                        * and ptlrpc_check_set() will be
+                                        * called again, keeping this thread
+                                        * busy. Instead, wait for the next
+                                        * timeout. Flag it as resend to
+                                        * ensure we don't wait to long.
+                                        */
+                                       req->rq_resend = 1;
+                                       spin_unlock(&imp->imp_lock);
+                                       continue;
+                               }
+
                                list_move_tail(&req->rq_list,
                                               &imp->imp_sending_list);
 
@@ -2493,10 +2531,10 @@ int ptlrpc_set_wait(const struct lu_env *env, struct ptlrpc_request_set *set)
                         */
                        if (rc == -ETIMEDOUT &&
                            signal_pending(current)) {
-                               sigset_t blocked_sigs;
+                               sigset_t old, new;
 
-                               cfs_block_sigsinv(LUSTRE_FATAL_SIGS,
-                                                 &blocked_sigs);
+                               siginitset(&new, LUSTRE_FATAL_SIGS);
+                               sigprocmask(SIG_BLOCK, &new, &old);
                                /*
                                 * In fact we only interrupt for the
                                 * "fatal" signals like SIGINT or
@@ -2507,7 +2545,7 @@ int ptlrpc_set_wait(const struct lu_env *env, struct ptlrpc_request_set *set)
                                 */
                                if (signal_pending(current))
                                        ptlrpc_interrupted_set(set);
-                               cfs_restore_sigs(&blocked_sigs);
+                               sigprocmask(SIG_SETMASK, &old, NULL);
                        }
                }
 
@@ -2723,7 +2761,7 @@ static int ptlrpc_unregister_reply(struct ptlrpc_request *request, int async)
        if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_REPL_UNLINK) &&
            async && request->rq_reply_deadline == 0 && cfs_fail_val == 0)
                request->rq_reply_deadline = ktime_get_real_seconds() +
-                                            LONG_UNLINK;
+                                            PTLRPC_REQ_LONG_UNLINK;
 
        /*
         * Nothing left to do.
@@ -2757,7 +2795,7 @@ static int ptlrpc_unregister_reply(struct ptlrpc_request *request, int async)
                wait_queue_head_t *wq = (request->rq_set) ?
                                        &request->rq_set->set_waitq :
                                        &request->rq_reply_waitq;
-               int seconds = LONG_UNLINK;
+               int seconds = PTLRPC_REQ_LONG_UNLINK;
                /*
                 * Network access will complete in finite time but the HUGE
                 * timeout lets us CWARN for visibility of sluggish NALs
@@ -3443,9 +3481,7 @@ void ptlrpc_set_bulk_mbits(struct ptlrpc_request *req)
                    || req->rq_mbits == 0) {
                        req->rq_mbits = req->rq_xid;
                } else {
-                       int total_md = (bd->bd_iov_count + LNET_MAX_IOV - 1) /
-                                       LNET_MAX_IOV;
-                       req->rq_mbits -= total_md - 1;
+                       req->rq_mbits -= bd->bd_md_count - 1;
                }
        } else {
                /*
@@ -3460,8 +3496,7 @@ void ptlrpc_set_bulk_mbits(struct ptlrpc_request *req)
         * that server can infer the number of bulks that were prepared,
         * see LU-1431
         */
-       req->rq_mbits += ((bd->bd_iov_count + LNET_MAX_IOV - 1) /
-                         LNET_MAX_IOV) - 1;
+       req->rq_mbits += bd->bd_md_count - 1;
 
        /*
         * Set rq_xid as rq_mbits to indicate the final bulk for the old