Whamcloud - gitweb
LU-13984 ptlrpc: throttle RPC resend if network error
[fs/lustre-release.git] / lustre / ptlrpc / client.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2011, 2017, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  * Lustre is a trademark of Sun Microsystems, Inc.
31  */
32
33 /** Implementation of client-side PortalRPC interfaces */
34
35 #define DEBUG_SUBSYSTEM S_RPC
36
37 #include <linux/delay.h>
38 #include <linux/random.h>
39
40 #include <lnet/lib-lnet.h>
41 #include <obd_support.h>
42 #include <obd_class.h>
43 #include <lustre_lib.h>
44 #include <lustre_ha.h>
45 #include <lustre_import.h>
46 #include <lustre_req_layout.h>
47
48 #include "ptlrpc_internal.h"
49
50 static void ptlrpc_prep_bulk_page_pin(struct ptlrpc_bulk_desc *desc,
51                                       struct page *page, int pageoffset,
52                                       int len)
53 {
54         __ptlrpc_prep_bulk_page(desc, page, pageoffset, len, 1);
55 }
56
57 static void ptlrpc_prep_bulk_page_nopin(struct ptlrpc_bulk_desc *desc,
58                                         struct page *page, int pageoffset,
59                                         int len)
60 {
61         __ptlrpc_prep_bulk_page(desc, page, pageoffset, len, 0);
62 }
63
64 static void ptlrpc_release_bulk_page_pin(struct ptlrpc_bulk_desc *desc)
65 {
66         int i;
67
68         for (i = 0; i < desc->bd_iov_count ; i++)
69                 put_page(desc->bd_vec[i].bv_page);
70 }
71
72 static int ptlrpc_prep_bulk_frag_pages(struct ptlrpc_bulk_desc *desc,
73                                        void *frag, int len)
74 {
75         unsigned int offset = (unsigned long)frag & ~PAGE_MASK;
76
77         ENTRY;
78         while (len > 0) {
79                 int page_len = min_t(unsigned int, PAGE_SIZE - offset,
80                                      len);
81                 unsigned long vaddr = (unsigned long)frag;
82
83                 ptlrpc_prep_bulk_page_nopin(desc,
84                                             lnet_kvaddr_to_page(vaddr),
85                                             offset, page_len);
86                 offset = 0;
87                 len -= page_len;
88                 frag += page_len;
89         }
90
91         RETURN(desc->bd_nob);
92 }
93
94 const struct ptlrpc_bulk_frag_ops ptlrpc_bulk_kiov_pin_ops = {
95         .add_kiov_frag  = ptlrpc_prep_bulk_page_pin,
96         .release_frags  = ptlrpc_release_bulk_page_pin,
97 };
98 EXPORT_SYMBOL(ptlrpc_bulk_kiov_pin_ops);
99
100 const struct ptlrpc_bulk_frag_ops ptlrpc_bulk_kiov_nopin_ops = {
101         .add_kiov_frag  = ptlrpc_prep_bulk_page_nopin,
102         .release_frags  = ptlrpc_release_bulk_noop,
103         .add_iov_frag   = ptlrpc_prep_bulk_frag_pages,
104 };
105 EXPORT_SYMBOL(ptlrpc_bulk_kiov_nopin_ops);
106
107 static int ptlrpc_send_new_req(struct ptlrpc_request *req);
108 static int ptlrpcd_check_work(struct ptlrpc_request *req);
109 static int ptlrpc_unregister_reply(struct ptlrpc_request *request, int async);
110
111 /**
112  * Initialize passed in client structure \a cl.
113  */
114 void ptlrpc_init_client(int req_portal, int rep_portal, const char *name,
115                         struct ptlrpc_client *cl)
116 {
117         cl->cli_request_portal = req_portal;
118         cl->cli_reply_portal   = rep_portal;
119         cl->cli_name           = name;
120 }
121 EXPORT_SYMBOL(ptlrpc_init_client);
122
123 /**
124  * Return PortalRPC connection for remore uud \a uuid
125  */
126 struct ptlrpc_connection *ptlrpc_uuid_to_connection(struct obd_uuid *uuid,
127                                                     lnet_nid_t nid4refnet)
128 {
129         struct ptlrpc_connection *c;
130         lnet_nid_t self;
131         struct lnet_process_id peer;
132         int err;
133
134         /*
135          * ptlrpc_uuid_to_peer() initializes its 2nd parameter
136          * before accessing its values.
137          */
138         /* coverity[uninit_use_in_call] */
139         peer.nid = nid4refnet;
140         err = ptlrpc_uuid_to_peer(uuid, &peer, &self);
141         if (err != 0) {
142                 CNETERR("cannot find peer %s!\n", uuid->uuid);
143                 return NULL;
144         }
145
146         c = ptlrpc_connection_get(peer, self, uuid);
147         if (c) {
148                 memcpy(c->c_remote_uuid.uuid,
149                        uuid->uuid, sizeof(c->c_remote_uuid.uuid));
150         }
151
152         CDEBUG(D_INFO, "%s -> %p\n", uuid->uuid, c);
153
154         return c;
155 }
156
157 /**
158  * Allocate and initialize new bulk descriptor on the sender.
159  * Returns pointer to the descriptor or NULL on error.
160  */
161 struct ptlrpc_bulk_desc *ptlrpc_new_bulk(unsigned int nfrags,
162                                          unsigned int max_brw,
163                                          enum ptlrpc_bulk_op_type type,
164                                          unsigned int portal,
165                                          const struct ptlrpc_bulk_frag_ops *ops)
166 {
167         struct ptlrpc_bulk_desc *desc;
168         int i;
169
170         LASSERT(ops->add_kiov_frag != NULL);
171
172         if (max_brw > PTLRPC_BULK_OPS_COUNT)
173                 RETURN(NULL);
174
175         if (nfrags > LNET_MAX_IOV * max_brw)
176                 RETURN(NULL);
177
178         OBD_ALLOC_PTR(desc);
179         if (!desc)
180                 return NULL;
181
182         OBD_ALLOC_LARGE(desc->bd_vec,
183                         nfrags * sizeof(*desc->bd_vec));
184         if (!desc->bd_vec)
185                 goto out;
186
187         spin_lock_init(&desc->bd_lock);
188         init_waitqueue_head(&desc->bd_waitq);
189         desc->bd_max_iov = nfrags;
190         desc->bd_iov_count = 0;
191         desc->bd_portal = portal;
192         desc->bd_type = type;
193         desc->bd_md_count = 0;
194         desc->bd_nob_last = LNET_MTU;
195         desc->bd_frag_ops = ops;
196         LASSERT(max_brw > 0);
197         desc->bd_md_max_brw = min(max_brw, PTLRPC_BULK_OPS_COUNT);
198         /*
199          * PTLRPC_BULK_OPS_COUNT is the compile-time transfer limit for this
200          * node. Negotiated ocd_brw_size will always be <= this number.
201          */
202         for (i = 0; i < PTLRPC_BULK_OPS_COUNT; i++)
203                 LNetInvalidateMDHandle(&desc->bd_mds[i]);
204
205         return desc;
206 out:
207         OBD_FREE_PTR(desc);
208         return NULL;
209 }
210
211 /**
212  * Prepare bulk descriptor for specified outgoing request \a req that
213  * can fit \a nfrags * pages. \a type is bulk type. \a portal is where
214  * the bulk to be sent. Used on client-side.
215  * Returns pointer to newly allocatrd initialized bulk descriptor or NULL on
216  * error.
217  */
218 struct ptlrpc_bulk_desc *ptlrpc_prep_bulk_imp(struct ptlrpc_request *req,
219                                               unsigned int nfrags,
220                                               unsigned int max_brw,
221                                               unsigned int type,
222                                               unsigned int portal,
223                                               const struct ptlrpc_bulk_frag_ops
224                                                 *ops)
225 {
226         struct obd_import *imp = req->rq_import;
227         struct ptlrpc_bulk_desc *desc;
228
229         ENTRY;
230         LASSERT(ptlrpc_is_bulk_op_passive(type));
231
232         desc = ptlrpc_new_bulk(nfrags, max_brw, type, portal, ops);
233         if (!desc)
234                 RETURN(NULL);
235
236         desc->bd_import = class_import_get(imp);
237         desc->bd_req = req;
238
239         desc->bd_cbid.cbid_fn  = client_bulk_callback;
240         desc->bd_cbid.cbid_arg = desc;
241
242         /* This makes req own desc, and free it when she frees herself */
243         req->rq_bulk = desc;
244
245         return desc;
246 }
247 EXPORT_SYMBOL(ptlrpc_prep_bulk_imp);
248
249 void __ptlrpc_prep_bulk_page(struct ptlrpc_bulk_desc *desc,
250                              struct page *page, int pageoffset, int len,
251                              int pin)
252 {
253         struct bio_vec *kiov;
254
255         LASSERT(desc->bd_iov_count < desc->bd_max_iov);
256         LASSERT(page != NULL);
257         LASSERT(pageoffset >= 0);
258         LASSERT(len > 0);
259         LASSERT(pageoffset + len <= PAGE_SIZE);
260
261         kiov = &desc->bd_vec[desc->bd_iov_count];
262
263         if (((desc->bd_iov_count % LNET_MAX_IOV) == 0) ||
264              ((desc->bd_nob_last + len) > LNET_MTU)) {
265                 desc->bd_mds_off[desc->bd_md_count] = desc->bd_iov_count;
266                 desc->bd_md_count++;
267                 desc->bd_nob_last = 0;
268                 LASSERT(desc->bd_md_count <= PTLRPC_BULK_OPS_COUNT);
269         }
270
271         desc->bd_nob_last += len;
272         desc->bd_nob += len;
273
274         if (pin)
275                 get_page(page);
276
277         kiov->bv_page = page;
278         kiov->bv_offset = pageoffset;
279         kiov->bv_len = len;
280
281         desc->bd_iov_count++;
282 }
283 EXPORT_SYMBOL(__ptlrpc_prep_bulk_page);
284
285 void ptlrpc_free_bulk(struct ptlrpc_bulk_desc *desc)
286 {
287         ENTRY;
288
289         LASSERT(desc != NULL);
290         LASSERT(desc->bd_iov_count != LI_POISON); /* not freed already */
291         LASSERT(desc->bd_refs == 0);         /* network hands off */
292         LASSERT((desc->bd_export != NULL) ^ (desc->bd_import != NULL));
293         LASSERT(desc->bd_frag_ops != NULL);
294
295         sptlrpc_enc_pool_put_pages(desc);
296
297         if (desc->bd_export)
298                 class_export_put(desc->bd_export);
299         else
300                 class_import_put(desc->bd_import);
301
302         if (desc->bd_frag_ops->release_frags != NULL)
303                 desc->bd_frag_ops->release_frags(desc);
304
305         OBD_FREE_LARGE(desc->bd_vec,
306                        desc->bd_max_iov * sizeof(*desc->bd_vec));
307         OBD_FREE_PTR(desc);
308         EXIT;
309 }
310 EXPORT_SYMBOL(ptlrpc_free_bulk);
311
312 /**
313  * Set server timelimit for this req, i.e. how long are we willing to wait
314  * for reply before timing out this request.
315  */
316 void ptlrpc_at_set_req_timeout(struct ptlrpc_request *req)
317 {
318         LASSERT(req->rq_import);
319
320         if (AT_OFF) {
321                 /* non-AT settings */
322                 /**
323                  * \a imp_server_timeout means this is reverse import and
324                  * we send (currently only) ASTs to the client and cannot afford
325                  * to wait too long for the reply, otherwise the other client
326                  * (because of which we are sending this request) would
327                  * timeout waiting for us
328                  */
329                 req->rq_timeout = req->rq_import->imp_server_timeout ?
330                                   obd_timeout / 2 : obd_timeout;
331         } else {
332                 struct imp_at *at = &req->rq_import->imp_at;
333                 timeout_t serv_est;
334                 int idx;
335
336                 idx = import_at_get_index(req->rq_import,
337                                           req->rq_request_portal);
338                 serv_est = at_get(&at->iat_service_estimate[idx]);
339                 /*
340                  * Currently a 32 bit value is sent over the
341                  * wire for rq_timeout so please don't change this
342                  * to time64_t. The work for LU-1158 will in time
343                  * replace rq_timeout with a 64 bit nanosecond value
344                  */
345                 req->rq_timeout = at_est2timeout(serv_est);
346         }
347         /*
348          * We could get even fancier here, using history to predict increased
349          * loading...
350          *
351          * Let the server know what this RPC timeout is by putting it in the
352          * reqmsg
353          */
354         lustre_msg_set_timeout(req->rq_reqmsg, req->rq_timeout);
355 }
356 EXPORT_SYMBOL(ptlrpc_at_set_req_timeout);
357
358 /* Adjust max service estimate based on server value */
359 static void ptlrpc_at_adj_service(struct ptlrpc_request *req,
360                                   timeout_t serv_est)
361 {
362         int idx;
363         timeout_t oldse;
364         struct imp_at *at;
365
366         LASSERT(req->rq_import);
367         at = &req->rq_import->imp_at;
368
369         idx = import_at_get_index(req->rq_import, req->rq_request_portal);
370         /*
371          * max service estimates are tracked on the server side,
372          * so just keep minimal history here
373          */
374         oldse = at_measured(&at->iat_service_estimate[idx], serv_est);
375         if (oldse != 0)
376                 CDEBUG(D_ADAPTTO,
377                        "The RPC service estimate for %s ptl %d has changed from %d to %d\n",
378                        req->rq_import->imp_obd->obd_name,
379                        req->rq_request_portal,
380                        oldse, at_get(&at->iat_service_estimate[idx]));
381 }
382
383 /* Expected network latency per remote node (secs) */
384 int ptlrpc_at_get_net_latency(struct ptlrpc_request *req)
385 {
386         return AT_OFF ? 0 : at_get(&req->rq_import->imp_at.iat_net_latency);
387 }
388
389 /* Adjust expected network latency */
390 void ptlrpc_at_adj_net_latency(struct ptlrpc_request *req,
391                                timeout_t service_timeout)
392 {
393         time64_t now = ktime_get_real_seconds();
394         struct imp_at *at;
395         timeout_t oldnl;
396         timeout_t nl;
397
398         LASSERT(req->rq_import);
399
400         if (service_timeout > now - req->rq_sent + 3) {
401                 /*
402                  * b=16408, however, this can also happen if early reply
403                  * is lost and client RPC is expired and resent, early reply
404                  * or reply of original RPC can still be fit in reply buffer
405                  * of resent RPC, now client is measuring time from the
406                  * resent time, but server sent back service time of original
407                  * RPC.
408                  */
409                 CDEBUG((lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT) ?
410                        D_ADAPTTO : D_WARNING,
411                        "Reported service time %u > total measured time %lld\n",
412                        service_timeout, now - req->rq_sent);
413                 return;
414         }
415
416         /* Network latency is total time less server processing time,
417          * st rounding
418          */
419         nl = max_t(timeout_t, now - req->rq_sent - service_timeout, 0) + 1;
420         at = &req->rq_import->imp_at;
421
422         oldnl = at_measured(&at->iat_net_latency, nl);
423         if (oldnl != 0)
424                 CDEBUG(D_ADAPTTO,
425                        "The network latency for %s (nid %s) has changed from %d to %d\n",
426                        req->rq_import->imp_obd->obd_name,
427                        obd_uuid2str(&req->rq_import->imp_connection->c_remote_uuid),
428                        oldnl, at_get(&at->iat_net_latency));
429 }
430
431 static int unpack_reply(struct ptlrpc_request *req)
432 {
433         int rc;
434
435         if (SPTLRPC_FLVR_POLICY(req->rq_flvr.sf_rpc) != SPTLRPC_POLICY_NULL) {
436                 rc = ptlrpc_unpack_rep_msg(req, req->rq_replen);
437                 if (rc) {
438                         DEBUG_REQ(D_ERROR, req, "unpack_rep failed: rc = %d",
439                                   rc);
440                         return -EPROTO;
441                 }
442         }
443
444         rc = lustre_unpack_rep_ptlrpc_body(req, MSG_PTLRPC_BODY_OFF);
445         if (rc) {
446                 DEBUG_REQ(D_ERROR, req, "unpack ptlrpc body failed: rc = %d",
447                           rc);
448                 return -EPROTO;
449         }
450         return 0;
451 }
452
453 /**
454  * Handle an early reply message, called with the rq_lock held.
455  * If anything goes wrong just ignore it - same as if it never happened
456  */
457 static int ptlrpc_at_recv_early_reply(struct ptlrpc_request *req)
458 __must_hold(&req->rq_lock)
459 {
460         struct ptlrpc_request *early_req;
461         timeout_t service_timeout;
462         time64_t olddl;
463         int rc;
464
465         ENTRY;
466         req->rq_early = 0;
467         spin_unlock(&req->rq_lock);
468
469         rc = sptlrpc_cli_unwrap_early_reply(req, &early_req);
470         if (rc) {
471                 spin_lock(&req->rq_lock);
472                 RETURN(rc);
473         }
474
475         rc = unpack_reply(early_req);
476         if (rc != 0) {
477                 sptlrpc_cli_finish_early_reply(early_req);
478                 spin_lock(&req->rq_lock);
479                 RETURN(rc);
480         }
481
482         /*
483          * Use new timeout value just to adjust the local value for this
484          * request, don't include it into at_history. It is unclear yet why
485          * service time increased and should it be counted or skipped, e.g.
486          * that can be recovery case or some error or server, the real reply
487          * will add all new data if it is worth to add.
488          */
489         req->rq_timeout = lustre_msg_get_timeout(early_req->rq_repmsg);
490         lustre_msg_set_timeout(req->rq_reqmsg, req->rq_timeout);
491
492         /* Network latency can be adjusted, it is pure network delays */
493         service_timeout = lustre_msg_get_service_timeout(early_req->rq_repmsg);
494         ptlrpc_at_adj_net_latency(req, service_timeout);
495
496         sptlrpc_cli_finish_early_reply(early_req);
497
498         spin_lock(&req->rq_lock);
499         olddl = req->rq_deadline;
500         /*
501          * server assumes it now has rq_timeout from when the request
502          * arrived, so the client should give it at least that long.
503          * since we don't know the arrival time we'll use the original
504          * sent time
505          */
506         req->rq_deadline = req->rq_sent + req->rq_timeout +
507                            ptlrpc_at_get_net_latency(req);
508
509         /* The below message is checked in replay-single.sh test_65{a,b} */
510         /* The below message is checked in sanity-{gss,krb5} test_8 */
511         DEBUG_REQ(D_ADAPTTO, req,
512                   "Early reply #%d, new deadline in %llds (%llds)",
513                   req->rq_early_count,
514                   req->rq_deadline - ktime_get_real_seconds(),
515                   req->rq_deadline - olddl);
516
517         RETURN(rc);
518 }
519
520 static struct kmem_cache *request_cache;
521
522 int ptlrpc_request_cache_init(void)
523 {
524         request_cache = kmem_cache_create("ptlrpc_cache",
525                                           sizeof(struct ptlrpc_request),
526                                           0, SLAB_HWCACHE_ALIGN, NULL);
527         return request_cache ? 0 : -ENOMEM;
528 }
529
530 void ptlrpc_request_cache_fini(void)
531 {
532         kmem_cache_destroy(request_cache);
533 }
534
535 struct ptlrpc_request *ptlrpc_request_cache_alloc(gfp_t flags)
536 {
537         struct ptlrpc_request *req;
538
539         OBD_SLAB_ALLOC_PTR_GFP(req, request_cache, flags);
540         return req;
541 }
542
543 void ptlrpc_request_cache_free(struct ptlrpc_request *req)
544 {
545         OBD_SLAB_FREE_PTR(req, request_cache);
546 }
547
548 /**
549  * Wind down request pool \a pool.
550  * Frees all requests from the pool too
551  */
552 void ptlrpc_free_rq_pool(struct ptlrpc_request_pool *pool)
553 {
554         struct list_head *l, *tmp;
555         struct ptlrpc_request *req;
556
557         LASSERT(pool != NULL);
558
559         spin_lock(&pool->prp_lock);
560         list_for_each_safe(l, tmp, &pool->prp_req_list) {
561                 req = list_entry(l, struct ptlrpc_request, rq_list);
562                 list_del(&req->rq_list);
563                 LASSERT(req->rq_reqbuf);
564                 LASSERT(req->rq_reqbuf_len == pool->prp_rq_size);
565                 OBD_FREE_LARGE(req->rq_reqbuf, pool->prp_rq_size);
566                 ptlrpc_request_cache_free(req);
567         }
568         spin_unlock(&pool->prp_lock);
569         OBD_FREE(pool, sizeof(*pool));
570 }
571 EXPORT_SYMBOL(ptlrpc_free_rq_pool);
572
573 /**
574  * Allocates, initializes and adds \a num_rq requests to the pool \a pool
575  */
576 int ptlrpc_add_rqs_to_pool(struct ptlrpc_request_pool *pool, int num_rq)
577 {
578         int i;
579         int size = 1;
580
581         while (size < pool->prp_rq_size)
582                 size <<= 1;
583
584         LASSERTF(list_empty(&pool->prp_req_list) ||
585                  size == pool->prp_rq_size,
586                  "Trying to change pool size with nonempty pool from %d to %d bytes\n",
587                  pool->prp_rq_size, size);
588
589         pool->prp_rq_size = size;
590         for (i = 0; i < num_rq; i++) {
591                 struct ptlrpc_request *req;
592                 struct lustre_msg *msg;
593
594                 req = ptlrpc_request_cache_alloc(GFP_NOFS);
595                 if (!req)
596                         return i;
597                 OBD_ALLOC_LARGE(msg, size);
598                 if (!msg) {
599                         ptlrpc_request_cache_free(req);
600                         return i;
601                 }
602                 req->rq_reqbuf = msg;
603                 req->rq_reqbuf_len = size;
604                 req->rq_pool = pool;
605                 spin_lock(&pool->prp_lock);
606                 list_add_tail(&req->rq_list, &pool->prp_req_list);
607                 spin_unlock(&pool->prp_lock);
608         }
609         return num_rq;
610 }
611 EXPORT_SYMBOL(ptlrpc_add_rqs_to_pool);
612
613 /**
614  * Create and initialize new request pool with given attributes:
615  * \a num_rq - initial number of requests to create for the pool
616  * \a msgsize - maximum message size possible for requests in thid pool
617  * \a populate_pool - function to be called when more requests need to be added
618  *                    to the pool
619  * Returns pointer to newly created pool or NULL on error.
620  */
621 struct ptlrpc_request_pool *
622 ptlrpc_init_rq_pool(int num_rq, int msgsize,
623                     int (*populate_pool)(struct ptlrpc_request_pool *, int))
624 {
625         struct ptlrpc_request_pool *pool;
626
627         OBD_ALLOC_PTR(pool);
628         if (!pool)
629                 return NULL;
630
631         /*
632          * Request next power of two for the allocation, because internally
633          * kernel would do exactly this
634          */
635         spin_lock_init(&pool->prp_lock);
636         INIT_LIST_HEAD(&pool->prp_req_list);
637         pool->prp_rq_size = msgsize + SPTLRPC_MAX_PAYLOAD;
638         pool->prp_populate = populate_pool;
639
640         populate_pool(pool, num_rq);
641
642         return pool;
643 }
644 EXPORT_SYMBOL(ptlrpc_init_rq_pool);
645
646 /**
647  * Fetches one request from pool \a pool
648  */
649 static struct ptlrpc_request *
650 ptlrpc_prep_req_from_pool(struct ptlrpc_request_pool *pool)
651 {
652         struct ptlrpc_request *request;
653         struct lustre_msg *reqbuf;
654
655         if (!pool)
656                 return NULL;
657
658         spin_lock(&pool->prp_lock);
659
660         /*
661          * See if we have anything in a pool, and bail out if nothing,
662          * in writeout path, where this matters, this is safe to do, because
663          * nothing is lost in this case, and when some in-flight requests
664          * complete, this code will be called again.
665          */
666         if (unlikely(list_empty(&pool->prp_req_list))) {
667                 spin_unlock(&pool->prp_lock);
668                 return NULL;
669         }
670
671         request = list_entry(pool->prp_req_list.next, struct ptlrpc_request,
672                              rq_list);
673         list_del_init(&request->rq_list);
674         spin_unlock(&pool->prp_lock);
675
676         LASSERT(request->rq_reqbuf);
677         LASSERT(request->rq_pool);
678
679         reqbuf = request->rq_reqbuf;
680         memset(request, 0, sizeof(*request));
681         request->rq_reqbuf = reqbuf;
682         request->rq_reqbuf_len = pool->prp_rq_size;
683         request->rq_pool = pool;
684
685         return request;
686 }
687
688 /**
689  * Returns freed \a request to pool.
690  */
691 static void __ptlrpc_free_req_to_pool(struct ptlrpc_request *request)
692 {
693         struct ptlrpc_request_pool *pool = request->rq_pool;
694
695         spin_lock(&pool->prp_lock);
696         LASSERT(list_empty(&request->rq_list));
697         LASSERT(!request->rq_receiving_reply);
698         list_add_tail(&request->rq_list, &pool->prp_req_list);
699         spin_unlock(&pool->prp_lock);
700 }
701
702 void ptlrpc_add_unreplied(struct ptlrpc_request *req)
703 {
704         struct obd_import *imp = req->rq_import;
705         struct list_head *tmp;
706         struct ptlrpc_request *iter;
707
708         assert_spin_locked(&imp->imp_lock);
709         LASSERT(list_empty(&req->rq_unreplied_list));
710
711         /* unreplied list is sorted by xid in ascending order */
712         list_for_each_prev(tmp, &imp->imp_unreplied_list) {
713                 iter = list_entry(tmp, struct ptlrpc_request,
714                                   rq_unreplied_list);
715
716                 LASSERT(req->rq_xid != iter->rq_xid);
717                 if (req->rq_xid < iter->rq_xid)
718                         continue;
719                 list_add(&req->rq_unreplied_list, &iter->rq_unreplied_list);
720                 return;
721         }
722         list_add(&req->rq_unreplied_list, &imp->imp_unreplied_list);
723 }
724
725 void ptlrpc_assign_next_xid_nolock(struct ptlrpc_request *req)
726 {
727         req->rq_xid = ptlrpc_next_xid();
728         ptlrpc_add_unreplied(req);
729 }
730
731 static inline void ptlrpc_assign_next_xid(struct ptlrpc_request *req)
732 {
733         spin_lock(&req->rq_import->imp_lock);
734         ptlrpc_assign_next_xid_nolock(req);
735         spin_unlock(&req->rq_import->imp_lock);
736 }
737
738 static atomic64_t ptlrpc_last_xid;
739
740 static void ptlrpc_reassign_next_xid(struct ptlrpc_request *req)
741 {
742         spin_lock(&req->rq_import->imp_lock);
743         list_del_init(&req->rq_unreplied_list);
744         ptlrpc_assign_next_xid_nolock(req);
745         spin_unlock(&req->rq_import->imp_lock);
746         DEBUG_REQ(D_RPCTRACE, req, "reassign xid");
747 }
748
749 void ptlrpc_get_mod_rpc_slot(struct ptlrpc_request *req)
750 {
751         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
752         __u32 opc;
753         __u16 tag;
754
755         opc = lustre_msg_get_opc(req->rq_reqmsg);
756         tag = obd_get_mod_rpc_slot(cli, opc);
757         lustre_msg_set_tag(req->rq_reqmsg, tag);
758         ptlrpc_reassign_next_xid(req);
759 }
760 EXPORT_SYMBOL(ptlrpc_get_mod_rpc_slot);
761
762 void ptlrpc_put_mod_rpc_slot(struct ptlrpc_request *req)
763 {
764         __u16 tag = lustre_msg_get_tag(req->rq_reqmsg);
765
766         if (tag != 0) {
767                 struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
768                 __u32 opc = lustre_msg_get_opc(req->rq_reqmsg);
769
770                 obd_put_mod_rpc_slot(cli, opc, tag);
771         }
772 }
773 EXPORT_SYMBOL(ptlrpc_put_mod_rpc_slot);
774
775 int ptlrpc_request_bufs_pack(struct ptlrpc_request *request,
776                              __u32 version, int opcode, char **bufs,
777                              struct ptlrpc_cli_ctx *ctx)
778 {
779         int count;
780         struct obd_import *imp;
781         __u32 *lengths;
782         int rc;
783
784         ENTRY;
785
786         count = req_capsule_filled_sizes(&request->rq_pill, RCL_CLIENT);
787         imp = request->rq_import;
788         lengths = request->rq_pill.rc_area[RCL_CLIENT];
789
790         if (ctx) {
791                 request->rq_cli_ctx = sptlrpc_cli_ctx_get(ctx);
792         } else {
793                 rc = sptlrpc_req_get_ctx(request);
794                 if (rc)
795                         GOTO(out_free, rc);
796         }
797         sptlrpc_req_set_flavor(request, opcode);
798
799         rc = lustre_pack_request(request, imp->imp_msg_magic, count,
800                                  lengths, bufs);
801         if (rc)
802                 GOTO(out_ctx, rc);
803
804         lustre_msg_add_version(request->rq_reqmsg, version);
805         request->rq_send_state = LUSTRE_IMP_FULL;
806         request->rq_type = PTL_RPC_MSG_REQUEST;
807
808         request->rq_req_cbid.cbid_fn  = request_out_callback;
809         request->rq_req_cbid.cbid_arg = request;
810
811         request->rq_reply_cbid.cbid_fn  = reply_in_callback;
812         request->rq_reply_cbid.cbid_arg = request;
813
814         request->rq_reply_deadline = 0;
815         request->rq_bulk_deadline = 0;
816         request->rq_req_deadline = 0;
817         request->rq_phase = RQ_PHASE_NEW;
818         request->rq_next_phase = RQ_PHASE_UNDEFINED;
819
820         request->rq_request_portal = imp->imp_client->cli_request_portal;
821         request->rq_reply_portal = imp->imp_client->cli_reply_portal;
822
823         ptlrpc_at_set_req_timeout(request);
824
825         lustre_msg_set_opc(request->rq_reqmsg, opcode);
826
827         /* Let's setup deadline for req/reply/bulk unlink for opcode. */
828         if (cfs_fail_val == opcode) {
829                 time64_t *fail_t = NULL, *fail2_t = NULL;
830
831                 if (CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_BULK_UNLINK)) {
832                         fail_t = &request->rq_bulk_deadline;
833                 } else if (CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_REPL_UNLINK)) {
834                         fail_t = &request->rq_reply_deadline;
835                 } else if (CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_REQ_UNLINK)) {
836                         fail_t = &request->rq_req_deadline;
837                 } else if (CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_BOTH_UNLINK)) {
838                         fail_t = &request->rq_reply_deadline;
839                         fail2_t = &request->rq_bulk_deadline;
840                 } else if (CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_ROUND_XID)) {
841                         time64_t now = ktime_get_real_seconds();
842                         u64 xid = ((u64)now >> 4) << 24;
843
844                         atomic64_set(&ptlrpc_last_xid, xid);
845                 }
846
847                 if (fail_t) {
848                         *fail_t = ktime_get_real_seconds() +
849                                   PTLRPC_REQ_LONG_UNLINK;
850
851                         if (fail2_t)
852                                 *fail2_t = ktime_get_real_seconds() +
853                                            PTLRPC_REQ_LONG_UNLINK;
854
855                         /*
856                          * The RPC is infected, let the test to change the
857                          * fail_loc
858                          */
859                         msleep(4 * MSEC_PER_SEC);
860                 }
861         }
862         ptlrpc_assign_next_xid(request);
863
864         RETURN(0);
865
866 out_ctx:
867         LASSERT(!request->rq_pool);
868         sptlrpc_cli_ctx_put(request->rq_cli_ctx, 1);
869 out_free:
870         atomic_dec(&imp->imp_reqs);
871         class_import_put(imp);
872
873         return rc;
874 }
875 EXPORT_SYMBOL(ptlrpc_request_bufs_pack);
876
877 /**
878  * Pack request buffers for network transfer, performing necessary encryption
879  * steps if necessary.
880  */
881 int ptlrpc_request_pack(struct ptlrpc_request *request,
882                         __u32 version, int opcode)
883 {
884         return ptlrpc_request_bufs_pack(request, version, opcode, NULL, NULL);
885 }
886 EXPORT_SYMBOL(ptlrpc_request_pack);
887
888 /**
889  * Helper function to allocate new request on import \a imp
890  * and possibly using existing request from pool \a pool if provided.
891  * Returns allocated request structure with import field filled or
892  * NULL on error.
893  */
894 static inline
895 struct ptlrpc_request *__ptlrpc_request_alloc(struct obd_import *imp,
896                                               struct ptlrpc_request_pool *pool)
897 {
898         struct ptlrpc_request *request = NULL;
899
900         request = ptlrpc_request_cache_alloc(GFP_NOFS);
901
902         if (!request && pool)
903                 request = ptlrpc_prep_req_from_pool(pool);
904
905         if (request) {
906                 ptlrpc_cli_req_init(request);
907
908                 LASSERTF((unsigned long)imp > 0x1000, "%p\n", imp);
909                 LASSERT(imp != LP_POISON);
910                 LASSERTF((unsigned long)imp->imp_client > 0x1000, "%p\n",
911                          imp->imp_client);
912                 LASSERT(imp->imp_client != LP_POISON);
913
914                 request->rq_import = class_import_get(imp);
915                 atomic_inc(&imp->imp_reqs);
916         } else {
917                 CERROR("request allocation out of memory\n");
918         }
919
920         return request;
921 }
922
923 static int ptlrpc_reconnect_if_idle(struct obd_import *imp)
924 {
925         int rc;
926
927         /*
928          * initiate connection if needed when the import has been
929          * referenced by the new request to avoid races with disconnect.
930          * serialize this check against conditional state=IDLE
931          * in ptlrpc_disconnect_idle_interpret()
932          */
933         spin_lock(&imp->imp_lock);
934         if (imp->imp_state == LUSTRE_IMP_IDLE) {
935                 imp->imp_generation++;
936                 imp->imp_initiated_at = imp->imp_generation;
937                 imp->imp_state = LUSTRE_IMP_NEW;
938
939                 /* connect_import_locked releases imp_lock */
940                 rc = ptlrpc_connect_import_locked(imp);
941                 if (rc)
942                         return rc;
943                 ptlrpc_pinger_add_import(imp);
944         } else {
945                 spin_unlock(&imp->imp_lock);
946         }
947         return 0;
948 }
949
950 /**
951  * Helper function for creating a request.
952  * Calls __ptlrpc_request_alloc to allocate new request sturcture and inits
953  * buffer structures according to capsule template \a format.
954  * Returns allocated request structure pointer or NULL on error.
955  */
956 static struct ptlrpc_request *
957 ptlrpc_request_alloc_internal(struct obd_import *imp,
958                               struct ptlrpc_request_pool *pool,
959                               const struct req_format *format)
960 {
961         struct ptlrpc_request *request;
962
963         request = __ptlrpc_request_alloc(imp, pool);
964         if (!request)
965                 return NULL;
966
967         /* don't make expensive check for idling connection
968          * if it's already connected */
969         if (unlikely(imp->imp_state != LUSTRE_IMP_FULL)) {
970                 if (ptlrpc_reconnect_if_idle(imp) < 0) {
971                         atomic_dec(&imp->imp_reqs);
972                         ptlrpc_request_free(request);
973                         return NULL;
974                 }
975         }
976
977         req_capsule_init(&request->rq_pill, request, RCL_CLIENT);
978         req_capsule_set(&request->rq_pill, format);
979         return request;
980 }
981
982 /**
983  * Allocate new request structure for import \a imp and initialize its
984  * buffer structure according to capsule template \a format.
985  */
986 struct ptlrpc_request *ptlrpc_request_alloc(struct obd_import *imp,
987                                             const struct req_format *format)
988 {
989         return ptlrpc_request_alloc_internal(imp, NULL, format);
990 }
991 EXPORT_SYMBOL(ptlrpc_request_alloc);
992
993 /**
994  * Allocate new request structure for import \a imp from pool \a pool and
995  * initialize its buffer structure according to capsule template \a format.
996  */
997 struct ptlrpc_request *
998 ptlrpc_request_alloc_pool(struct obd_import *imp,
999                           struct ptlrpc_request_pool *pool,
1000                           const struct req_format *format)
1001 {
1002         return ptlrpc_request_alloc_internal(imp, pool, format);
1003 }
1004 EXPORT_SYMBOL(ptlrpc_request_alloc_pool);
1005
1006 /**
1007  * For requests not from pool, free memory of the request structure.
1008  * For requests obtained from a pool earlier, return request back to pool.
1009  */
1010 void ptlrpc_request_free(struct ptlrpc_request *request)
1011 {
1012         if (request->rq_pool)
1013                 __ptlrpc_free_req_to_pool(request);
1014         else
1015                 ptlrpc_request_cache_free(request);
1016 }
1017 EXPORT_SYMBOL(ptlrpc_request_free);
1018
1019 /**
1020  * Allocate new request for operatione \a opcode and immediatelly pack it for
1021  * network transfer.
1022  * Only used for simple requests like OBD_PING where the only important
1023  * part of the request is operation itself.
1024  * Returns allocated request or NULL on error.
1025  */
1026 struct ptlrpc_request *ptlrpc_request_alloc_pack(struct obd_import *imp,
1027                                                  const struct req_format *format,
1028                                                  __u32 version, int opcode)
1029 {
1030         struct ptlrpc_request *req = ptlrpc_request_alloc(imp, format);
1031         int rc;
1032
1033         if (req) {
1034                 rc = ptlrpc_request_pack(req, version, opcode);
1035                 if (rc) {
1036                         ptlrpc_request_free(req);
1037                         req = NULL;
1038                 }
1039         }
1040         return req;
1041 }
1042 EXPORT_SYMBOL(ptlrpc_request_alloc_pack);
1043
1044 /**
1045  * Allocate and initialize new request set structure on the current CPT.
1046  * Returns a pointer to the newly allocated set structure or NULL on error.
1047  */
1048 struct ptlrpc_request_set *ptlrpc_prep_set(void)
1049 {
1050         struct ptlrpc_request_set *set;
1051         int cpt;
1052
1053         ENTRY;
1054         cpt = cfs_cpt_current(cfs_cpt_tab, 0);
1055         OBD_CPT_ALLOC(set, cfs_cpt_tab, cpt, sizeof(*set));
1056         if (!set)
1057                 RETURN(NULL);
1058         atomic_set(&set->set_refcount, 1);
1059         INIT_LIST_HEAD(&set->set_requests);
1060         init_waitqueue_head(&set->set_waitq);
1061         atomic_set(&set->set_new_count, 0);
1062         atomic_set(&set->set_remaining, 0);
1063         spin_lock_init(&set->set_new_req_lock);
1064         INIT_LIST_HEAD(&set->set_new_requests);
1065         set->set_max_inflight = UINT_MAX;
1066         set->set_producer     = NULL;
1067         set->set_producer_arg = NULL;
1068         set->set_rc           = 0;
1069
1070         RETURN(set);
1071 }
1072 EXPORT_SYMBOL(ptlrpc_prep_set);
1073
1074 /**
1075  * Allocate and initialize new request set structure with flow control
1076  * extension. This extension allows to control the number of requests in-flight
1077  * for the whole set. A callback function to generate requests must be provided
1078  * and the request set will keep the number of requests sent over the wire to
1079  * @max_inflight.
1080  * Returns a pointer to the newly allocated set structure or NULL on error.
1081  */
1082 struct ptlrpc_request_set *ptlrpc_prep_fcset(int max, set_producer_func func,
1083                                              void *arg)
1084
1085 {
1086         struct ptlrpc_request_set *set;
1087
1088         set = ptlrpc_prep_set();
1089         if (!set)
1090                 RETURN(NULL);
1091
1092         set->set_max_inflight  = max;
1093         set->set_producer      = func;
1094         set->set_producer_arg  = arg;
1095
1096         RETURN(set);
1097 }
1098
1099 /**
1100  * Wind down and free request set structure previously allocated with
1101  * ptlrpc_prep_set.
1102  * Ensures that all requests on the set have completed and removes
1103  * all requests from the request list in a set.
1104  * If any unsent request happen to be on the list, pretends that they got
1105  * an error in flight and calls their completion handler.
1106  */
1107 void ptlrpc_set_destroy(struct ptlrpc_request_set *set)
1108 {
1109         struct list_head *tmp;
1110         struct list_head *next;
1111         int expected_phase;
1112         int n = 0;
1113
1114         ENTRY;
1115
1116         /* Requests on the set should either all be completed, or all be new */
1117         expected_phase = (atomic_read(&set->set_remaining) == 0) ?
1118                          RQ_PHASE_COMPLETE : RQ_PHASE_NEW;
1119         list_for_each(tmp, &set->set_requests) {
1120                 struct ptlrpc_request *req =
1121                         list_entry(tmp, struct ptlrpc_request,
1122                                    rq_set_chain);
1123
1124                 LASSERT(req->rq_phase == expected_phase);
1125                 n++;
1126         }
1127
1128         LASSERTF(atomic_read(&set->set_remaining) == 0 ||
1129                  atomic_read(&set->set_remaining) == n, "%d / %d\n",
1130                  atomic_read(&set->set_remaining), n);
1131
1132         list_for_each_safe(tmp, next, &set->set_requests) {
1133                 struct ptlrpc_request *req =
1134                         list_entry(tmp, struct ptlrpc_request,
1135                                    rq_set_chain);
1136                 list_del_init(&req->rq_set_chain);
1137
1138                 LASSERT(req->rq_phase == expected_phase);
1139
1140                 if (req->rq_phase == RQ_PHASE_NEW) {
1141                         ptlrpc_req_interpret(NULL, req, -EBADR);
1142                         atomic_dec(&set->set_remaining);
1143                 }
1144
1145                 spin_lock(&req->rq_lock);
1146                 req->rq_set = NULL;
1147                 req->rq_invalid_rqset = 0;
1148                 spin_unlock(&req->rq_lock);
1149
1150                 ptlrpc_req_finished(req);
1151         }
1152
1153         LASSERT(atomic_read(&set->set_remaining) == 0);
1154
1155         ptlrpc_reqset_put(set);
1156         EXIT;
1157 }
1158 EXPORT_SYMBOL(ptlrpc_set_destroy);
1159
1160 /**
1161  * Add a new request to the general purpose request set.
1162  * Assumes request reference from the caller.
1163  */
1164 void ptlrpc_set_add_req(struct ptlrpc_request_set *set,
1165                         struct ptlrpc_request *req)
1166 {
1167         if (set == PTLRPCD_SET) {
1168                 ptlrpcd_add_req(req);
1169                 return;
1170         }
1171
1172         LASSERT(req->rq_import->imp_state != LUSTRE_IMP_IDLE);
1173         LASSERT(list_empty(&req->rq_set_chain));
1174
1175         if (req->rq_allow_intr)
1176                 set->set_allow_intr = 1;
1177
1178         /* The set takes over the caller's request reference */
1179         list_add_tail(&req->rq_set_chain, &set->set_requests);
1180         req->rq_set = set;
1181         atomic_inc(&set->set_remaining);
1182         req->rq_queued_time = ktime_get_seconds();
1183
1184         if (req->rq_reqmsg)
1185                 lustre_msg_set_jobid(req->rq_reqmsg, NULL);
1186
1187         if (set->set_producer)
1188                 /*
1189                  * If the request set has a producer callback, the RPC must be
1190                  * sent straight away
1191                  */
1192                 ptlrpc_send_new_req(req);
1193 }
1194 EXPORT_SYMBOL(ptlrpc_set_add_req);
1195
1196 /**
1197  * Add a request to a request with dedicated server thread
1198  * and wake the thread to make any necessary processing.
1199  * Currently only used for ptlrpcd.
1200  */
1201 void ptlrpc_set_add_new_req(struct ptlrpcd_ctl *pc,
1202                             struct ptlrpc_request *req)
1203 {
1204         struct ptlrpc_request_set *set = pc->pc_set;
1205         int count, i;
1206
1207         LASSERT(req->rq_set == NULL);
1208         LASSERT(test_bit(LIOD_STOP, &pc->pc_flags) == 0);
1209
1210         spin_lock(&set->set_new_req_lock);
1211         /*
1212          * The set takes over the caller's request reference.
1213          */
1214         req->rq_set = set;
1215         req->rq_queued_time = ktime_get_seconds();
1216         list_add_tail(&req->rq_set_chain, &set->set_new_requests);
1217         count = atomic_inc_return(&set->set_new_count);
1218         spin_unlock(&set->set_new_req_lock);
1219
1220         /* Only need to call wakeup once for the first entry. */
1221         if (count == 1) {
1222                 wake_up(&set->set_waitq);
1223
1224                 /*
1225                  * XXX: It maybe unnecessary to wakeup all the partners. But to
1226                  *      guarantee the async RPC can be processed ASAP, we have
1227                  *      no other better choice. It maybe fixed in future.
1228                  */
1229                 for (i = 0; i < pc->pc_npartners; i++)
1230                         wake_up(&pc->pc_partners[i]->pc_set->set_waitq);
1231         }
1232 }
1233
1234 /**
1235  * Based on the current state of the import, determine if the request
1236  * can be sent, is an error, or should be delayed.
1237  *
1238  * Returns true if this request should be delayed. If false, and
1239  * *status is set, then the request can not be sent and *status is the
1240  * error code.  If false and status is 0, then request can be sent.
1241  *
1242  * The imp->imp_lock must be held.
1243  */
1244 static int ptlrpc_import_delay_req(struct obd_import *imp,
1245                                    struct ptlrpc_request *req, int *status)
1246 {
1247         int delay = 0;
1248
1249         ENTRY;
1250         LASSERT(status);
1251         *status = 0;
1252
1253         if (req->rq_ctx_init || req->rq_ctx_fini) {
1254                 /* always allow ctx init/fini rpc go through */
1255         } else if (imp->imp_state == LUSTRE_IMP_NEW) {
1256                 DEBUG_REQ(D_ERROR, req, "Uninitialized import");
1257                 *status = -EIO;
1258         } else if (imp->imp_state == LUSTRE_IMP_CLOSED) {
1259                 unsigned int opc = lustre_msg_get_opc(req->rq_reqmsg);
1260
1261                 /*
1262                  * pings or MDS-equivalent STATFS may safely
1263                  * race with umount
1264                  */
1265                 DEBUG_REQ((opc == OBD_PING || opc == OST_STATFS) ?
1266                           D_HA : D_ERROR, req, "IMP_CLOSED");
1267                 *status = -EIO;
1268         } else if (ptlrpc_send_limit_expired(req)) {
1269                 /* probably doesn't need to be a D_ERROR afterinitial testing */
1270                 DEBUG_REQ(D_HA, req, "send limit expired");
1271                 *status = -ETIMEDOUT;
1272         } else if (req->rq_send_state == LUSTRE_IMP_CONNECTING &&
1273                    imp->imp_state == LUSTRE_IMP_CONNECTING) {
1274                 ;/* allow CONNECT even if import is invalid */
1275                 if (atomic_read(&imp->imp_inval_count) != 0) {
1276                         DEBUG_REQ(D_ERROR, req, "invalidate in flight");
1277                         *status = -EIO;
1278                 }
1279         } else if (imp->imp_invalid || imp->imp_obd->obd_no_recov) {
1280                 if (!imp->imp_deactive)
1281                         DEBUG_REQ(D_NET, req, "IMP_INVALID");
1282                 *status = -ESHUTDOWN; /* b=12940 */
1283         } else if (req->rq_import_generation != imp->imp_generation) {
1284                 DEBUG_REQ(D_ERROR, req, "req wrong generation:");
1285                 *status = -EIO;
1286         } else if (req->rq_send_state != imp->imp_state) {
1287                 /* invalidate in progress - any requests should be drop */
1288                 if (atomic_read(&imp->imp_inval_count) != 0) {
1289                         DEBUG_REQ(D_ERROR, req, "invalidate in flight");
1290                         *status = -EIO;
1291                 } else if (req->rq_no_delay &&
1292                            imp->imp_generation != imp->imp_initiated_at) {
1293                         /* ignore nodelay for requests initiating connections */
1294                         *status = -EWOULDBLOCK;
1295                 } else if (req->rq_allow_replay &&
1296                            (imp->imp_state == LUSTRE_IMP_REPLAY ||
1297                             imp->imp_state == LUSTRE_IMP_REPLAY_LOCKS ||
1298                             imp->imp_state == LUSTRE_IMP_REPLAY_WAIT ||
1299                             imp->imp_state == LUSTRE_IMP_RECOVER)) {
1300                         DEBUG_REQ(D_HA, req, "allow during recovery");
1301                 } else {
1302                         delay = 1;
1303                 }
1304         }
1305
1306         RETURN(delay);
1307 }
1308
1309 /**
1310  * Decide if the error message should be printed to the console or not.
1311  * Makes its decision based on request type, status, and failure frequency.
1312  *
1313  * \param[in] req  request that failed and may need a console message
1314  *
1315  * \retval false if no message should be printed
1316  * \retval true  if console message should be printed
1317  */
1318 static bool ptlrpc_console_allow(struct ptlrpc_request *req, __u32 opc, int err)
1319 {
1320         LASSERT(req->rq_reqmsg != NULL);
1321
1322         /* Suppress particular reconnect errors which are to be expected. */
1323         if (opc == OST_CONNECT || opc == MDS_CONNECT || opc == MGS_CONNECT) {
1324                 /* Suppress timed out reconnect requests */
1325                 if (lustre_handle_is_used(&req->rq_import->imp_remote_handle) ||
1326                     req->rq_timedout)
1327                         return false;
1328
1329                 /*
1330                  * Suppress most unavailable/again reconnect requests, but
1331                  * print occasionally so it is clear client is trying to
1332                  * connect to a server where no target is running.
1333                  */
1334                 if ((err == -ENODEV || err == -EAGAIN) &&
1335                     req->rq_import->imp_conn_cnt % 30 != 20)
1336                         return false;
1337         }
1338
1339         if (opc == LDLM_ENQUEUE && err == -EAGAIN)
1340                 /* -EAGAIN is normal when using POSIX flocks */
1341                 return false;
1342
1343         if (opc == OBD_PING && (err == -ENODEV || err == -ENOTCONN) &&
1344             (req->rq_xid & 0xf) != 10)
1345                 /* Suppress most ping requests, they may fail occasionally */
1346                 return false;
1347
1348         return true;
1349 }
1350
1351 /**
1352  * Check request processing status.
1353  * Returns the status.
1354  */
1355 static int ptlrpc_check_status(struct ptlrpc_request *req)
1356 {
1357         int rc;
1358
1359         ENTRY;
1360         rc = lustre_msg_get_status(req->rq_repmsg);
1361         if (lustre_msg_get_type(req->rq_repmsg) == PTL_RPC_MSG_ERR) {
1362                 struct obd_import *imp = req->rq_import;
1363                 lnet_nid_t nid = imp->imp_connection->c_peer.nid;
1364                 __u32 opc = lustre_msg_get_opc(req->rq_reqmsg);
1365
1366                 if (ptlrpc_console_allow(req, opc, rc))
1367                         LCONSOLE_ERROR_MSG(0x11,
1368                                            "%s: operation %s to node %s failed: rc = %d\n",
1369                                            imp->imp_obd->obd_name,
1370                                            ll_opcode2str(opc),
1371                                            libcfs_nid2str(nid), rc);
1372                 RETURN(rc < 0 ? rc : -EINVAL);
1373         }
1374
1375         if (rc)
1376                 DEBUG_REQ(D_INFO, req, "check status: rc = %d", rc);
1377
1378         RETURN(rc);
1379 }
1380
1381 /**
1382  * save pre-versions of objects into request for replay.
1383  * Versions are obtained from server reply.
1384  * used for VBR.
1385  */
1386 static void ptlrpc_save_versions(struct ptlrpc_request *req)
1387 {
1388         struct lustre_msg *repmsg = req->rq_repmsg;
1389         struct lustre_msg *reqmsg = req->rq_reqmsg;
1390         __u64 *versions = lustre_msg_get_versions(repmsg);
1391
1392         ENTRY;
1393         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY)
1394                 return;
1395
1396         LASSERT(versions);
1397         lustre_msg_set_versions(reqmsg, versions);
1398         CDEBUG(D_INFO, "Client save versions [%#llx/%#llx]\n",
1399                versions[0], versions[1]);
1400
1401         EXIT;
1402 }
1403
1404 __u64 ptlrpc_known_replied_xid(struct obd_import *imp)
1405 {
1406         struct ptlrpc_request *req;
1407
1408         assert_spin_locked(&imp->imp_lock);
1409         if (list_empty(&imp->imp_unreplied_list))
1410                 return 0;
1411
1412         req = list_entry(imp->imp_unreplied_list.next, struct ptlrpc_request,
1413                          rq_unreplied_list);
1414         LASSERTF(req->rq_xid >= 1, "XID:%llu\n", req->rq_xid);
1415
1416         if (imp->imp_known_replied_xid < req->rq_xid - 1)
1417                 imp->imp_known_replied_xid = req->rq_xid - 1;
1418
1419         return req->rq_xid - 1;
1420 }
1421
1422 /**
1423  * Callback function called when client receives RPC reply for \a req.
1424  * Returns 0 on success or error code.
1425  * The return alue would be assigned to req->rq_status by the caller
1426  * as request processing status.
1427  * This function also decides if the request needs to be saved for later replay.
1428  */
1429 static int after_reply(struct ptlrpc_request *req)
1430 {
1431         struct obd_import *imp = req->rq_import;
1432         struct obd_device *obd = req->rq_import->imp_obd;
1433         ktime_t work_start;
1434         u64 committed;
1435         s64 timediff;
1436         int rc;
1437
1438         ENTRY;
1439         LASSERT(obd != NULL);
1440         /* repbuf must be unlinked */
1441         LASSERT(!req->rq_receiving_reply && req->rq_reply_unlinked);
1442
1443         if (req->rq_reply_truncated) {
1444                 if (ptlrpc_no_resend(req)) {
1445                         DEBUG_REQ(D_ERROR, req,
1446                                   "reply buffer overflow, expected=%d, actual size=%d",
1447                                   req->rq_nob_received, req->rq_repbuf_len);
1448                         RETURN(-EOVERFLOW);
1449                 }
1450
1451                 sptlrpc_cli_free_repbuf(req);
1452                 /*
1453                  * Pass the required reply buffer size (include
1454                  * space for early reply).
1455                  * NB: no need to roundup because alloc_repbuf
1456                  * will roundup it
1457                  */
1458                 req->rq_replen = req->rq_nob_received;
1459                 req->rq_nob_received = 0;
1460                 spin_lock(&req->rq_lock);
1461                 req->rq_resend       = 1;
1462                 spin_unlock(&req->rq_lock);
1463                 RETURN(0);
1464         }
1465
1466         work_start = ktime_get_real();
1467         timediff = ktime_us_delta(work_start, req->rq_sent_ns);
1468
1469         /*
1470          * NB Until this point, the whole of the incoming message,
1471          * including buflens, status etc is in the sender's byte order.
1472          */
1473         rc = sptlrpc_cli_unwrap_reply(req);
1474         if (rc) {
1475                 DEBUG_REQ(D_ERROR, req, "unwrap reply failed: rc = %d", rc);
1476                 RETURN(rc);
1477         }
1478
1479         /*
1480          * Security layer unwrap might ask resend this request.
1481          */
1482         if (req->rq_resend)
1483                 RETURN(0);
1484
1485         rc = unpack_reply(req);
1486         if (rc)
1487                 RETURN(rc);
1488
1489         /* retry indefinitely on EINPROGRESS */
1490         if (lustre_msg_get_status(req->rq_repmsg) == -EINPROGRESS &&
1491             ptlrpc_no_resend(req) == 0 && !req->rq_no_retry_einprogress) {
1492                 time64_t now = ktime_get_real_seconds();
1493
1494                 DEBUG_REQ((req->rq_nr_resend % 8 == 1 ? D_WARNING : 0) |
1495                           D_RPCTRACE, req, "resending request on EINPROGRESS");
1496                 spin_lock(&req->rq_lock);
1497                 req->rq_resend = 1;
1498                 spin_unlock(&req->rq_lock);
1499                 req->rq_nr_resend++;
1500
1501                 /* Readjust the timeout for current conditions */
1502                 ptlrpc_at_set_req_timeout(req);
1503                 /*
1504                  * delay resend to give a chance to the server to get ready.
1505                  * The delay is increased by 1s on every resend and is capped to
1506                  * the current request timeout (i.e. obd_timeout if AT is off,
1507                  * or AT service time x 125% + 5s, see at_est2timeout)
1508                  */
1509                 if (req->rq_nr_resend > req->rq_timeout)
1510                         req->rq_sent = now + req->rq_timeout;
1511                 else
1512                         req->rq_sent = now + req->rq_nr_resend;
1513
1514                 /* Resend for EINPROGRESS will use a new XID */
1515                 spin_lock(&imp->imp_lock);
1516                 list_del_init(&req->rq_unreplied_list);
1517                 spin_unlock(&imp->imp_lock);
1518
1519                 RETURN(0);
1520         }
1521
1522         if (obd->obd_svc_stats) {
1523                 lprocfs_counter_add(obd->obd_svc_stats, PTLRPC_REQWAIT_CNTR,
1524                                     timediff);
1525                 ptlrpc_lprocfs_rpc_sent(req, timediff);
1526         }
1527
1528         if (lustre_msg_get_type(req->rq_repmsg) != PTL_RPC_MSG_REPLY &&
1529             lustre_msg_get_type(req->rq_repmsg) != PTL_RPC_MSG_ERR) {
1530                 DEBUG_REQ(D_ERROR, req, "invalid packet received (type=%u)",
1531                           lustre_msg_get_type(req->rq_repmsg));
1532                 RETURN(-EPROTO);
1533         }
1534
1535         if (lustre_msg_get_opc(req->rq_reqmsg) != OBD_PING)
1536                 CFS_FAIL_TIMEOUT(OBD_FAIL_PTLRPC_PAUSE_REP, cfs_fail_val);
1537         ptlrpc_at_adj_service(req, lustre_msg_get_timeout(req->rq_repmsg));
1538         ptlrpc_at_adj_net_latency(req,
1539                                   lustre_msg_get_service_timeout(req->rq_repmsg));
1540
1541         rc = ptlrpc_check_status(req);
1542
1543         if (rc) {
1544                 /*
1545                  * Either we've been evicted, or the server has failed for
1546                  * some reason. Try to reconnect, and if that fails, punt to
1547                  * the upcall.
1548                  */
1549                 if (ptlrpc_recoverable_error(rc)) {
1550                         if (req->rq_send_state != LUSTRE_IMP_FULL ||
1551                             imp->imp_obd->obd_no_recov || imp->imp_dlm_fake) {
1552                                 RETURN(rc);
1553                         }
1554                         ptlrpc_request_handle_notconn(req);
1555                         RETURN(rc);
1556                 }
1557         } else {
1558                 /*
1559                  * Let's look if server sent slv. Do it only for RPC with
1560                  * rc == 0.
1561                  */
1562                 ldlm_cli_update_pool(req);
1563         }
1564
1565         /*
1566          * Store transno in reqmsg for replay.
1567          */
1568         if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY)) {
1569                 req->rq_transno = lustre_msg_get_transno(req->rq_repmsg);
1570                 lustre_msg_set_transno(req->rq_reqmsg, req->rq_transno);
1571         }
1572
1573         if (imp->imp_replayable) {
1574                 spin_lock(&imp->imp_lock);
1575                 /*
1576                  * No point in adding already-committed requests to the replay
1577                  * list, we will just remove them immediately. b=9829
1578                  */
1579                 if (req->rq_transno != 0 &&
1580                     (req->rq_transno >
1581                      lustre_msg_get_last_committed(req->rq_repmsg) ||
1582                      req->rq_replay)) {
1583                         /** version recovery */
1584                         ptlrpc_save_versions(req);
1585                         ptlrpc_retain_replayable_request(req, imp);
1586                 } else if (req->rq_commit_cb &&
1587                            list_empty(&req->rq_replay_list)) {
1588                         /*
1589                          * NB: don't call rq_commit_cb if it's already on
1590                          * rq_replay_list, ptlrpc_free_committed() will call
1591                          * it later, see LU-3618 for details
1592                          */
1593                         spin_unlock(&imp->imp_lock);
1594                         req->rq_commit_cb(req);
1595                         spin_lock(&imp->imp_lock);
1596                 }
1597
1598                 /*
1599                  * Replay-enabled imports return commit-status information.
1600                  */
1601                 committed = lustre_msg_get_last_committed(req->rq_repmsg);
1602                 if (likely(committed > imp->imp_peer_committed_transno))
1603                         imp->imp_peer_committed_transno = committed;
1604
1605                 ptlrpc_free_committed(imp);
1606
1607                 if (!list_empty(&imp->imp_replay_list)) {
1608                         struct ptlrpc_request *last;
1609
1610                         last = list_entry(imp->imp_replay_list.prev,
1611                                           struct ptlrpc_request,
1612                                           rq_replay_list);
1613                         /*
1614                          * Requests with rq_replay stay on the list even if no
1615                          * commit is expected.
1616                          */
1617                         if (last->rq_transno > imp->imp_peer_committed_transno)
1618                                 ptlrpc_pinger_commit_expected(imp);
1619                 }
1620
1621                 spin_unlock(&imp->imp_lock);
1622         }
1623
1624         RETURN(rc);
1625 }
1626
1627 /**
1628  * Helper function to send request \a req over the network for the first time
1629  * Also adjusts request phase.
1630  * Returns 0 on success or error code.
1631  */
1632 static int ptlrpc_send_new_req(struct ptlrpc_request *req)
1633 {
1634         struct obd_import *imp = req->rq_import;
1635         __u64 min_xid = 0;
1636         int rc;
1637
1638         ENTRY;
1639         LASSERT(req->rq_phase == RQ_PHASE_NEW);
1640
1641         /* do not try to go further if there is not enough memory in enc_pool */
1642         if (req->rq_sent && req->rq_bulk)
1643                 if (req->rq_bulk->bd_iov_count > get_free_pages_in_pool() &&
1644                     pool_is_at_full_capacity())
1645                         RETURN(-ENOMEM);
1646
1647         if (req->rq_sent && (req->rq_sent > ktime_get_real_seconds()) &&
1648             (!req->rq_generation_set ||
1649              req->rq_import_generation == imp->imp_generation))
1650                 RETURN(0);
1651
1652         ptlrpc_rqphase_move(req, RQ_PHASE_RPC);
1653
1654         spin_lock(&imp->imp_lock);
1655
1656         LASSERT(req->rq_xid != 0);
1657         LASSERT(!list_empty(&req->rq_unreplied_list));
1658
1659         if (!req->rq_generation_set)
1660                 req->rq_import_generation = imp->imp_generation;
1661
1662         if (ptlrpc_import_delay_req(imp, req, &rc)) {
1663                 spin_lock(&req->rq_lock);
1664                 req->rq_waiting = 1;
1665                 spin_unlock(&req->rq_lock);
1666
1667                 DEBUG_REQ(D_HA, req, "req waiting for recovery: (%s != %s)",
1668                           ptlrpc_import_state_name(req->rq_send_state),
1669                           ptlrpc_import_state_name(imp->imp_state));
1670                 LASSERT(list_empty(&req->rq_list));
1671                 list_add_tail(&req->rq_list, &imp->imp_delayed_list);
1672                 atomic_inc(&req->rq_import->imp_inflight);
1673                 spin_unlock(&imp->imp_lock);
1674                 RETURN(0);
1675         }
1676
1677         if (rc != 0) {
1678                 spin_unlock(&imp->imp_lock);
1679                 req->rq_status = rc;
1680                 ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
1681                 RETURN(rc);
1682         }
1683
1684         LASSERT(list_empty(&req->rq_list));
1685         list_add_tail(&req->rq_list, &imp->imp_sending_list);
1686         atomic_inc(&req->rq_import->imp_inflight);
1687
1688         /*
1689          * find the known replied XID from the unreplied list, CONNECT
1690          * and DISCONNECT requests are skipped to make the sanity check
1691          * on server side happy. see process_req_last_xid().
1692          *
1693          * For CONNECT: Because replay requests have lower XID, it'll
1694          * break the sanity check if CONNECT bump the exp_last_xid on
1695          * server.
1696          *
1697          * For DISCONNECT: Since client will abort inflight RPC before
1698          * sending DISCONNECT, DISCONNECT may carry an XID which higher
1699          * than the inflight RPC.
1700          */
1701         if (!ptlrpc_req_is_connect(req) && !ptlrpc_req_is_disconnect(req))
1702                 min_xid = ptlrpc_known_replied_xid(imp);
1703         spin_unlock(&imp->imp_lock);
1704
1705         lustre_msg_set_last_xid(req->rq_reqmsg, min_xid);
1706
1707         lustre_msg_set_status(req->rq_reqmsg, current->pid);
1708
1709         rc = sptlrpc_req_refresh_ctx(req, 0);
1710         if (rc) {
1711                 if (req->rq_err) {
1712                         req->rq_status = rc;
1713                         RETURN(1);
1714                 } else {
1715                         spin_lock(&req->rq_lock);
1716                         req->rq_wait_ctx = 1;
1717                         spin_unlock(&req->rq_lock);
1718                         RETURN(0);
1719                 }
1720         }
1721
1722         CDEBUG(D_RPCTRACE,
1723                "Sending RPC req@%p pname:cluuid:pid:xid:nid:opc:job %s:%s:%d:%llu:%s:%d:%s\n",
1724                req, current->comm,
1725                imp->imp_obd->obd_uuid.uuid,
1726                lustre_msg_get_status(req->rq_reqmsg), req->rq_xid,
1727                obd_import_nid2str(imp), lustre_msg_get_opc(req->rq_reqmsg),
1728                lustre_msg_get_jobid(req->rq_reqmsg) ?: "");
1729
1730         rc = ptl_send_rpc(req, 0);
1731         if (rc == -ENOMEM) {
1732                 spin_lock(&imp->imp_lock);
1733                 if (!list_empty(&req->rq_list)) {
1734                         list_del_init(&req->rq_list);
1735                         if (atomic_dec_and_test(&req->rq_import->imp_inflight))
1736                                 wake_up(&req->rq_import->imp_recovery_waitq);
1737                 }
1738                 spin_unlock(&imp->imp_lock);
1739                 ptlrpc_rqphase_move(req, RQ_PHASE_NEW);
1740                 RETURN(rc);
1741         }
1742         if (rc) {
1743                 DEBUG_REQ(D_HA, req, "send failed, expect timeout: rc = %d",
1744                           rc);
1745                 spin_lock(&req->rq_lock);
1746                 req->rq_net_err = 1;
1747                 spin_unlock(&req->rq_lock);
1748                 RETURN(rc);
1749         }
1750         RETURN(0);
1751 }
1752
1753 static inline int ptlrpc_set_producer(struct ptlrpc_request_set *set)
1754 {
1755         int remaining, rc;
1756
1757         ENTRY;
1758         LASSERT(set->set_producer != NULL);
1759
1760         remaining = atomic_read(&set->set_remaining);
1761
1762         /*
1763          * populate the ->set_requests list with requests until we
1764          * reach the maximum number of RPCs in flight for this set
1765          */
1766         while (atomic_read(&set->set_remaining) < set->set_max_inflight) {
1767                 rc = set->set_producer(set, set->set_producer_arg);
1768                 if (rc == -ENOENT) {
1769                         /* no more RPC to produce */
1770                         set->set_producer     = NULL;
1771                         set->set_producer_arg = NULL;
1772                         RETURN(0);
1773                 }
1774         }
1775
1776         RETURN((atomic_read(&set->set_remaining) - remaining));
1777 }
1778
1779 /**
1780  * this sends any unsent RPCs in \a set and returns 1 if all are sent
1781  * and no more replies are expected.
1782  * (it is possible to get less replies than requests sent e.g. due to timed out
1783  * requests or requests that we had trouble to send out)
1784  *
1785  * NOTE: This function contains a potential schedule point (cond_resched()).
1786  */
1787 int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
1788 {
1789         struct list_head *tmp, *next;
1790         LIST_HEAD(comp_reqs);
1791         int force_timer_recalc = 0;
1792
1793         ENTRY;
1794         if (atomic_read(&set->set_remaining) == 0)
1795                 RETURN(1);
1796
1797         list_for_each_safe(tmp, next, &set->set_requests) {
1798                 struct ptlrpc_request *req =
1799                         list_entry(tmp, struct ptlrpc_request,
1800                                    rq_set_chain);
1801                 struct obd_import *imp = req->rq_import;
1802                 int unregistered = 0;
1803                 int async = 1;
1804                 int rc = 0;
1805
1806                 if (req->rq_phase == RQ_PHASE_COMPLETE) {
1807                         list_move_tail(&req->rq_set_chain, &comp_reqs);
1808                         continue;
1809                 }
1810
1811                 /*
1812                  * This schedule point is mainly for the ptlrpcd caller of this
1813                  * function.  Most ptlrpc sets are not long-lived and unbounded
1814                  * in length, but at the least the set used by the ptlrpcd is.
1815                  * Since the processing time is unbounded, we need to insert an
1816                  * explicit schedule point to make the thread well-behaved.
1817                  */
1818                 cond_resched();
1819
1820                 /*
1821                  * If the caller requires to allow to be interpreted by force
1822                  * and it has really been interpreted, then move the request
1823                  * to RQ_PHASE_INTERPRET phase in spite of what the current
1824                  * phase is.
1825                  */
1826                 if (unlikely(req->rq_allow_intr && req->rq_intr)) {
1827                         req->rq_status = -EINTR;
1828                         ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
1829
1830                         /*
1831                          * Since it is interpreted and we have to wait for
1832                          * the reply to be unlinked, then use sync mode.
1833                          */
1834                         async = 0;
1835
1836                         GOTO(interpret, req->rq_status);
1837                 }
1838
1839                 if (req->rq_phase == RQ_PHASE_NEW && ptlrpc_send_new_req(req))
1840                         force_timer_recalc = 1;
1841
1842                 /* delayed send - skip */
1843                 if (req->rq_phase == RQ_PHASE_NEW && req->rq_sent)
1844                         continue;
1845
1846                 /* delayed resend - skip */
1847                 if (req->rq_phase == RQ_PHASE_RPC && req->rq_resend &&
1848                     req->rq_sent > ktime_get_real_seconds())
1849                         continue;
1850
1851                 if (!(req->rq_phase == RQ_PHASE_RPC ||
1852                       req->rq_phase == RQ_PHASE_BULK ||
1853                       req->rq_phase == RQ_PHASE_INTERPRET ||
1854                       req->rq_phase == RQ_PHASE_UNREG_RPC ||
1855                       req->rq_phase == RQ_PHASE_UNREG_BULK)) {
1856                         DEBUG_REQ(D_ERROR, req, "bad phase %x", req->rq_phase);
1857                         LBUG();
1858                 }
1859
1860                 if (req->rq_phase == RQ_PHASE_UNREG_RPC ||
1861                     req->rq_phase == RQ_PHASE_UNREG_BULK) {
1862                         LASSERT(req->rq_next_phase != req->rq_phase);
1863                         LASSERT(req->rq_next_phase != RQ_PHASE_UNDEFINED);
1864
1865                         if (req->rq_req_deadline &&
1866                             !OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_REQ_UNLINK))
1867                                 req->rq_req_deadline = 0;
1868                         if (req->rq_reply_deadline &&
1869                             !OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_REPL_UNLINK))
1870                                 req->rq_reply_deadline = 0;
1871                         if (req->rq_bulk_deadline &&
1872                             !OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_BULK_UNLINK))
1873                                 req->rq_bulk_deadline = 0;
1874
1875                         /*
1876                          * Skip processing until reply is unlinked. We
1877                          * can't return to pool before that and we can't
1878                          * call interpret before that. We need to make
1879                          * sure that all rdma transfers finished and will
1880                          * not corrupt any data.
1881                          */
1882                         if (req->rq_phase == RQ_PHASE_UNREG_RPC &&
1883                             ptlrpc_client_recv_or_unlink(req))
1884                                 continue;
1885                         if (req->rq_phase == RQ_PHASE_UNREG_BULK &&
1886                             ptlrpc_client_bulk_active(req))
1887                                 continue;
1888
1889                         /*
1890                          * Turn fail_loc off to prevent it from looping
1891                          * forever.
1892                          */
1893                         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_REPL_UNLINK)) {
1894                                 OBD_FAIL_CHECK_ORSET(OBD_FAIL_PTLRPC_LONG_REPL_UNLINK,
1895                                                      OBD_FAIL_ONCE);
1896                         }
1897                         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_BULK_UNLINK)) {
1898                                 OBD_FAIL_CHECK_ORSET(OBD_FAIL_PTLRPC_LONG_BULK_UNLINK,
1899                                                      OBD_FAIL_ONCE);
1900                         }
1901
1902                         /*
1903                          * Move to next phase if reply was successfully
1904                          * unlinked.
1905                          */
1906                         ptlrpc_rqphase_move(req, req->rq_next_phase);
1907                 }
1908
1909                 if (req->rq_phase == RQ_PHASE_INTERPRET)
1910                         GOTO(interpret, req->rq_status);
1911
1912                 /*
1913                  * Note that this also will start async reply unlink.
1914                  */
1915                 if (req->rq_net_err && !req->rq_timedout) {
1916                         ptlrpc_expire_one_request(req, 1);
1917
1918                         /*
1919                          * Check if we still need to wait for unlink.
1920                          */
1921                         if (ptlrpc_client_recv_or_unlink(req) ||
1922                             ptlrpc_client_bulk_active(req))
1923                                 continue;
1924                         /* If there is no need to resend, fail it now. */
1925                         if (req->rq_no_resend) {
1926                                 if (req->rq_status == 0)
1927                                         req->rq_status = -EIO;
1928                                 ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
1929                                 GOTO(interpret, req->rq_status);
1930                         } else {
1931                                 continue;
1932                         }
1933                 }
1934
1935                 if (req->rq_err) {
1936                         spin_lock(&req->rq_lock);
1937                         req->rq_replied = 0;
1938                         spin_unlock(&req->rq_lock);
1939                         if (req->rq_status == 0)
1940                                 req->rq_status = -EIO;
1941                         ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
1942                         GOTO(interpret, req->rq_status);
1943                 }
1944
1945                 /*
1946                  * ptlrpc_set_wait uses l_wait_event_abortable_timeout()
1947                  * so it sets rq_intr regardless of individual rpc
1948                  * timeouts. The synchronous IO waiting path sets
1949                  * rq_intr irrespective of whether ptlrpcd
1950                  * has seen a timeout.  Our policy is to only interpret
1951                  * interrupted rpcs after they have timed out, so we
1952                  * need to enforce that here.
1953                  */
1954
1955                 if (req->rq_intr && (req->rq_timedout || req->rq_waiting ||
1956                                      req->rq_wait_ctx)) {
1957                         req->rq_status = -EINTR;
1958                         ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
1959                         GOTO(interpret, req->rq_status);
1960                 }
1961
1962                 if (req->rq_phase == RQ_PHASE_RPC) {
1963                         if (req->rq_timedout || req->rq_resend ||
1964                             req->rq_waiting || req->rq_wait_ctx) {
1965                                 int status;
1966
1967                                 if (!ptlrpc_unregister_reply(req, 1)) {
1968                                         ptlrpc_unregister_bulk(req, 1);
1969                                         continue;
1970                                 }
1971
1972                                 spin_lock(&imp->imp_lock);
1973                                 if (ptlrpc_import_delay_req(imp, req,
1974                                                             &status)) {
1975                                         /*
1976                                          * put on delay list - only if we wait
1977                                          * recovery finished - before send
1978                                          */
1979                                         list_move_tail(&req->rq_list,
1980                                                        &imp->imp_delayed_list);
1981                                         spin_unlock(&imp->imp_lock);
1982                                         continue;
1983                                 }
1984
1985                                 if (status != 0)  {
1986                                         req->rq_status = status;
1987                                         ptlrpc_rqphase_move(req,
1988                                                             RQ_PHASE_INTERPRET);
1989                                         spin_unlock(&imp->imp_lock);
1990                                         GOTO(interpret, req->rq_status);
1991                                 }
1992                                 /* ignore on just initiated connections */
1993                                 if (ptlrpc_no_resend(req) &&
1994                                     !req->rq_wait_ctx &&
1995                                     imp->imp_generation !=
1996                                     imp->imp_initiated_at) {
1997                                         req->rq_status = -ENOTCONN;
1998                                         ptlrpc_rqphase_move(req,
1999                                                             RQ_PHASE_INTERPRET);
2000                                         spin_unlock(&imp->imp_lock);
2001                                         GOTO(interpret, req->rq_status);
2002                                 }
2003
2004                                 /* don't resend too fast in case of network
2005                                  * errors.
2006                                  */
2007                                 if (ktime_get_real_seconds() < (req->rq_sent + 1)
2008                                     && req->rq_net_err && req->rq_timedout) {
2009
2010                                         DEBUG_REQ(D_INFO, req,
2011                                                   "throttle request");
2012                                         /* Don't try to resend RPC right away
2013                                          * as it is likely it will fail again
2014                                          * and ptlrpc_check_set() will be
2015                                          * called again, keeping this thread
2016                                          * busy. Instead, wait for the next
2017                                          * timeout. Flag it as resend to
2018                                          * ensure we don't wait to long.
2019                                          */
2020                                         req->rq_resend = 1;
2021                                         spin_unlock(&imp->imp_lock);
2022                                         continue;
2023                                 }
2024
2025                                 list_move_tail(&req->rq_list,
2026                                                &imp->imp_sending_list);
2027
2028                                 spin_unlock(&imp->imp_lock);
2029
2030                                 spin_lock(&req->rq_lock);
2031                                 req->rq_waiting = 0;
2032                                 spin_unlock(&req->rq_lock);
2033
2034                                 if (req->rq_timedout || req->rq_resend) {
2035                                         /*
2036                                          * This is re-sending anyways,
2037                                          * let's mark req as resend.
2038                                          */
2039                                         spin_lock(&req->rq_lock);
2040                                         req->rq_resend = 1;
2041                                         spin_unlock(&req->rq_lock);
2042                                 }
2043                                 /*
2044                                  * rq_wait_ctx is only touched by ptlrpcd,
2045                                  * so no lock is needed here.
2046                                  */
2047                                 status = sptlrpc_req_refresh_ctx(req, 0);
2048                                 if (status) {
2049                                         if (req->rq_err) {
2050                                                 req->rq_status = status;
2051                                                 spin_lock(&req->rq_lock);
2052                                                 req->rq_wait_ctx = 0;
2053                                                 spin_unlock(&req->rq_lock);
2054                                                 force_timer_recalc = 1;
2055                                         } else {
2056                                                 spin_lock(&req->rq_lock);
2057                                                 req->rq_wait_ctx = 1;
2058                                                 spin_unlock(&req->rq_lock);
2059                                         }
2060
2061                                         continue;
2062                                 } else {
2063                                         spin_lock(&req->rq_lock);
2064                                         req->rq_wait_ctx = 0;
2065                                         spin_unlock(&req->rq_lock);
2066                                 }
2067
2068                                 /*
2069                                  * In any case, the previous bulk should be
2070                                  * cleaned up to prepare for the new sending
2071                                  */
2072                                 if (req->rq_bulk &&
2073                                     !ptlrpc_unregister_bulk(req, 1))
2074                                         continue;
2075
2076                                 rc = ptl_send_rpc(req, 0);
2077                                 if (rc == -ENOMEM) {
2078                                         spin_lock(&imp->imp_lock);
2079                                         if (!list_empty(&req->rq_list))
2080                                                 list_del_init(&req->rq_list);
2081                                         spin_unlock(&imp->imp_lock);
2082                                         ptlrpc_rqphase_move(req, RQ_PHASE_NEW);
2083                                         continue;
2084                                 }
2085                                 if (rc) {
2086                                         DEBUG_REQ(D_HA, req,
2087                                                   "send failed: rc = %d", rc);
2088                                         force_timer_recalc = 1;
2089                                         spin_lock(&req->rq_lock);
2090                                         req->rq_net_err = 1;
2091                                         spin_unlock(&req->rq_lock);
2092                                         continue;
2093                                 }
2094                                 /* need to reset the timeout */
2095                                 force_timer_recalc = 1;
2096                         }
2097
2098                         spin_lock(&req->rq_lock);
2099
2100                         if (ptlrpc_client_early(req)) {
2101                                 ptlrpc_at_recv_early_reply(req);
2102                                 spin_unlock(&req->rq_lock);
2103                                 continue;
2104                         }
2105
2106                         /* Still waiting for a reply? */
2107                         if (ptlrpc_client_recv(req)) {
2108                                 spin_unlock(&req->rq_lock);
2109                                 continue;
2110                         }
2111
2112                         /* Did we actually receive a reply? */
2113                         if (!ptlrpc_client_replied(req)) {
2114                                 spin_unlock(&req->rq_lock);
2115                                 continue;
2116                         }
2117
2118                         spin_unlock(&req->rq_lock);
2119
2120                         /*
2121                          * unlink from net because we are going to
2122                          * swab in-place of reply buffer
2123                          */
2124                         unregistered = ptlrpc_unregister_reply(req, 1);
2125                         if (!unregistered)
2126                                 continue;
2127
2128                         req->rq_status = after_reply(req);
2129                         if (req->rq_resend)
2130                                 continue;
2131
2132                         /*
2133                          * If there is no bulk associated with this request,
2134                          * then we're done and should let the interpreter
2135                          * process the reply. Similarly if the RPC returned
2136                          * an error, and therefore the bulk will never arrive.
2137                          */
2138                         if (!req->rq_bulk || req->rq_status < 0) {
2139                                 ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
2140                                 GOTO(interpret, req->rq_status);
2141                         }
2142
2143                         ptlrpc_rqphase_move(req, RQ_PHASE_BULK);
2144                 }
2145
2146                 LASSERT(req->rq_phase == RQ_PHASE_BULK);
2147                 if (ptlrpc_client_bulk_active(req))
2148                         continue;
2149
2150                 if (req->rq_bulk->bd_failure) {
2151                         /*
2152                          * The RPC reply arrived OK, but the bulk screwed
2153                          * up!  Dead weird since the server told us the RPC
2154                          * was good after getting the REPLY for her GET or
2155                          * the ACK for her PUT.
2156                          */
2157                         DEBUG_REQ(D_ERROR, req, "bulk transfer failed");
2158                         req->rq_status = -EIO;
2159                 }
2160
2161                 ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
2162
2163 interpret:
2164                 LASSERT(req->rq_phase == RQ_PHASE_INTERPRET);
2165
2166                 /*
2167                  * This moves to "unregistering" phase we need to wait for
2168                  * reply unlink.
2169                  */
2170                 if (!unregistered && !ptlrpc_unregister_reply(req, async)) {
2171                         /* start async bulk unlink too */
2172                         ptlrpc_unregister_bulk(req, 1);
2173                         continue;
2174                 }
2175
2176                 if (!ptlrpc_unregister_bulk(req, async))
2177                         continue;
2178
2179                 /*
2180                  * When calling interpret receiving already should be
2181                  * finished.
2182                  */
2183                 LASSERT(!req->rq_receiving_reply);
2184
2185                 ptlrpc_req_interpret(env, req, req->rq_status);
2186
2187                 if (ptlrpcd_check_work(req)) {
2188                         atomic_dec(&set->set_remaining);
2189                         continue;
2190                 }
2191                 ptlrpc_rqphase_move(req, RQ_PHASE_COMPLETE);
2192
2193                 if (req->rq_reqmsg)
2194                         CDEBUG(D_RPCTRACE,
2195                                "Completed RPC req@%p pname:cluuid:pid:xid:nid:opc:job %s:%s:%d:%llu:%s:%d:%s\n",
2196                                req, current->comm,
2197                                imp->imp_obd->obd_uuid.uuid,
2198                                lustre_msg_get_status(req->rq_reqmsg),
2199                                req->rq_xid,
2200                                obd_import_nid2str(imp),
2201                                lustre_msg_get_opc(req->rq_reqmsg),
2202                                lustre_msg_get_jobid(req->rq_reqmsg) ?: "");
2203
2204                 spin_lock(&imp->imp_lock);
2205                 /*
2206                  * Request already may be not on sending or delaying list. This
2207                  * may happen in the case of marking it erroneous for the case
2208                  * ptlrpc_import_delay_req(req, status) find it impossible to
2209                  * allow sending this rpc and returns *status != 0.
2210                  */
2211                 if (!list_empty(&req->rq_list)) {
2212                         list_del_init(&req->rq_list);
2213                         if (atomic_dec_and_test(&imp->imp_inflight))
2214                                 wake_up(&imp->imp_recovery_waitq);
2215                 }
2216                 list_del_init(&req->rq_unreplied_list);
2217                 spin_unlock(&imp->imp_lock);
2218
2219                 atomic_dec(&set->set_remaining);
2220                 wake_up(&imp->imp_recovery_waitq);
2221
2222                 if (set->set_producer) {
2223                         /* produce a new request if possible */
2224                         if (ptlrpc_set_producer(set) > 0)
2225                                 force_timer_recalc = 1;
2226
2227                         /*
2228                          * free the request that has just been completed
2229                          * in order not to pollute set->set_requests
2230                          */
2231                         list_del_init(&req->rq_set_chain);
2232                         spin_lock(&req->rq_lock);
2233                         req->rq_set = NULL;
2234                         req->rq_invalid_rqset = 0;
2235                         spin_unlock(&req->rq_lock);
2236
2237                         /* record rq_status to compute the final status later */
2238                         if (req->rq_status != 0)
2239                                 set->set_rc = req->rq_status;
2240                         ptlrpc_req_finished(req);
2241                 } else {
2242                         list_move_tail(&req->rq_set_chain, &comp_reqs);
2243                 }
2244         }
2245
2246         /*
2247          * move completed request at the head of list so it's easier for
2248          * caller to find them
2249          */
2250         list_splice(&comp_reqs, &set->set_requests);
2251
2252         /* If we hit an error, we want to recover promptly. */
2253         RETURN(atomic_read(&set->set_remaining) == 0 || force_timer_recalc);
2254 }
2255 EXPORT_SYMBOL(ptlrpc_check_set);
2256
2257 /**
2258  * Time out request \a req. is \a async_unlink is set, that means do not wait
2259  * until LNet actually confirms network buffer unlinking.
2260  * Return 1 if we should give up further retrying attempts or 0 otherwise.
2261  */
2262 int ptlrpc_expire_one_request(struct ptlrpc_request *req, int async_unlink)
2263 {
2264         struct obd_import *imp = req->rq_import;
2265         unsigned int debug_mask = D_RPCTRACE;
2266         int rc = 0;
2267
2268         ENTRY;
2269         spin_lock(&req->rq_lock);
2270         req->rq_timedout = 1;
2271         spin_unlock(&req->rq_lock);
2272
2273         if (ptlrpc_console_allow(req, lustre_msg_get_opc(req->rq_reqmsg),
2274                                  lustre_msg_get_status(req->rq_reqmsg)))
2275                 debug_mask = D_WARNING;
2276         DEBUG_REQ(debug_mask, req, "Request sent has %s: [sent %lld/real %lld]",
2277                   req->rq_net_err ? "failed due to network error" :
2278                      ((req->rq_real_sent == 0 ||
2279                        req->rq_real_sent < req->rq_sent ||
2280                        req->rq_real_sent >= req->rq_deadline) ?
2281                       "timed out for sent delay" : "timed out for slow reply"),
2282                   req->rq_sent, req->rq_real_sent);
2283
2284         if (imp && obd_debug_peer_on_timeout)
2285                 LNetDebugPeer(imp->imp_connection->c_peer);
2286
2287         ptlrpc_unregister_reply(req, async_unlink);
2288         ptlrpc_unregister_bulk(req, async_unlink);
2289
2290         if (obd_dump_on_timeout)
2291                 libcfs_debug_dumplog();
2292
2293         if (!imp) {
2294                 DEBUG_REQ(D_HA, req, "NULL import: already cleaned up?");
2295                 RETURN(1);
2296         }
2297
2298         atomic_inc(&imp->imp_timeouts);
2299
2300         /* The DLM server doesn't want recovery run on its imports. */
2301         if (imp->imp_dlm_fake)
2302                 RETURN(1);
2303
2304         /*
2305          * If this request is for recovery or other primordial tasks,
2306          * then error it out here.
2307          */
2308         if (req->rq_ctx_init || req->rq_ctx_fini ||
2309             req->rq_send_state != LUSTRE_IMP_FULL ||
2310             imp->imp_obd->obd_no_recov) {
2311                 DEBUG_REQ(D_RPCTRACE, req, "err -110, sent_state=%s (now=%s)",
2312                           ptlrpc_import_state_name(req->rq_send_state),
2313                           ptlrpc_import_state_name(imp->imp_state));
2314                 spin_lock(&req->rq_lock);
2315                 req->rq_status = -ETIMEDOUT;
2316                 req->rq_err = 1;
2317                 spin_unlock(&req->rq_lock);
2318                 RETURN(1);
2319         }
2320
2321         /*
2322          * if a request can't be resent we can't wait for an answer after
2323          * the timeout
2324          */
2325         if (ptlrpc_no_resend(req)) {
2326                 DEBUG_REQ(D_RPCTRACE, req, "TIMEOUT-NORESEND:");
2327                 rc = 1;
2328         }
2329
2330         ptlrpc_fail_import(imp, lustre_msg_get_conn_cnt(req->rq_reqmsg));
2331
2332         RETURN(rc);
2333 }
2334
2335 /**
2336  * Time out all uncompleted requests in request set pointed by \a data
2337  * This is called when a wait times out.
2338  */
2339 void ptlrpc_expired_set(struct ptlrpc_request_set *set)
2340 {
2341         struct list_head *tmp;
2342         time64_t now = ktime_get_real_seconds();
2343
2344         ENTRY;
2345         LASSERT(set != NULL);
2346
2347         /*
2348          * A timeout expired. See which reqs it applies to...
2349          */
2350         list_for_each(tmp, &set->set_requests) {
2351                 struct ptlrpc_request *req =
2352                         list_entry(tmp, struct ptlrpc_request,
2353                                    rq_set_chain);
2354
2355                 /* don't expire request waiting for context */
2356                 if (req->rq_wait_ctx)
2357                         continue;
2358
2359                 /* Request in-flight? */
2360                 if (!((req->rq_phase == RQ_PHASE_RPC &&
2361                        !req->rq_waiting && !req->rq_resend) ||
2362                       (req->rq_phase == RQ_PHASE_BULK)))
2363                         continue;
2364
2365                 if (req->rq_timedout ||     /* already dealt with */
2366                     req->rq_deadline > now) /* not expired */
2367                         continue;
2368
2369                 /*
2370                  * Deal with this guy. Do it asynchronously to not block
2371                  * ptlrpcd thread.
2372                  */
2373                 ptlrpc_expire_one_request(req, 1);
2374         }
2375 }
2376
2377 /**
2378  * Interrupts (sets interrupted flag) all uncompleted requests in
2379  * a set \a data. This is called when a wait_event is interrupted
2380  * by a signal.
2381  */
2382 static void ptlrpc_interrupted_set(struct ptlrpc_request_set *set)
2383 {
2384         struct list_head *tmp;
2385
2386         LASSERT(set != NULL);
2387         CDEBUG(D_RPCTRACE, "INTERRUPTED SET %p\n", set);
2388
2389         list_for_each(tmp, &set->set_requests) {
2390                 struct ptlrpc_request *req =
2391                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
2392
2393                 if (req->rq_intr)
2394                         continue;
2395
2396                 if (req->rq_phase != RQ_PHASE_RPC &&
2397                     req->rq_phase != RQ_PHASE_UNREG_RPC &&
2398                     !req->rq_allow_intr)
2399                         continue;
2400
2401                 spin_lock(&req->rq_lock);
2402                 req->rq_intr = 1;
2403                 spin_unlock(&req->rq_lock);
2404         }
2405 }
2406
2407 /**
2408  * Get the smallest timeout in the set; this does NOT set a timeout.
2409  */
2410 time64_t ptlrpc_set_next_timeout(struct ptlrpc_request_set *set)
2411 {
2412         struct list_head *tmp;
2413         time64_t now = ktime_get_real_seconds();
2414         int timeout = 0;
2415         struct ptlrpc_request *req;
2416         time64_t deadline;
2417
2418         ENTRY;
2419         list_for_each(tmp, &set->set_requests) {
2420                 req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
2421
2422                 /* Request in-flight? */
2423                 if (!(((req->rq_phase == RQ_PHASE_RPC) && !req->rq_waiting) ||
2424                       (req->rq_phase == RQ_PHASE_BULK) ||
2425                       (req->rq_phase == RQ_PHASE_NEW)))
2426                         continue;
2427
2428                 /* Already timed out. */
2429                 if (req->rq_timedout)
2430                         continue;
2431
2432                 /* Waiting for ctx. */
2433                 if (req->rq_wait_ctx)
2434                         continue;
2435
2436                 if (req->rq_phase == RQ_PHASE_NEW)
2437                         deadline = req->rq_sent;
2438                 else if (req->rq_phase == RQ_PHASE_RPC && req->rq_resend)
2439                         deadline = req->rq_sent;
2440                 else
2441                         deadline = req->rq_sent + req->rq_timeout;
2442
2443                 if (deadline <= now)    /* actually expired already */
2444                         timeout = 1;    /* ASAP */
2445                 else if (timeout == 0 || timeout > deadline - now)
2446                         timeout = deadline - now;
2447         }
2448         RETURN(timeout);
2449 }
2450
2451 /**
2452  * Send all unset request from the set and then wait untill all
2453  * requests in the set complete (either get a reply, timeout, get an
2454  * error or otherwise be interrupted).
2455  * Returns 0 on success or error code otherwise.
2456  */
2457 int ptlrpc_set_wait(const struct lu_env *env, struct ptlrpc_request_set *set)
2458 {
2459         struct list_head *tmp;
2460         struct ptlrpc_request *req;
2461         time64_t timeout;
2462         int rc;
2463
2464         ENTRY;
2465         if (set->set_producer)
2466                 (void)ptlrpc_set_producer(set);
2467         else
2468                 list_for_each(tmp, &set->set_requests) {
2469                         req = list_entry(tmp, struct ptlrpc_request,
2470                                          rq_set_chain);
2471                         if (req->rq_phase == RQ_PHASE_NEW)
2472                                 (void)ptlrpc_send_new_req(req);
2473                 }
2474
2475         if (list_empty(&set->set_requests))
2476                 RETURN(0);
2477
2478         do {
2479                 timeout = ptlrpc_set_next_timeout(set);
2480
2481                 /*
2482                  * wait until all complete, interrupted, or an in-flight
2483                  * req times out
2484                  */
2485                 CDEBUG(D_RPCTRACE, "set %p going to sleep for %lld seconds\n",
2486                        set, timeout);
2487
2488                 if ((timeout == 0 && !signal_pending(current)) ||
2489                     set->set_allow_intr) {
2490                         /*
2491                          * No requests are in-flight (ether timed out
2492                          * or delayed), so we can allow interrupts.
2493                          * We still want to block for a limited time,
2494                          * so we allow interrupts during the timeout.
2495                          */
2496                         rc = l_wait_event_abortable_timeout(
2497                                 set->set_waitq,
2498                                 ptlrpc_check_set(NULL, set),
2499                                 cfs_time_seconds(timeout ? timeout : 1));
2500                         if (rc == 0) {
2501                                 rc = -ETIMEDOUT;
2502                                 ptlrpc_expired_set(set);
2503                         } else if (rc < 0) {
2504                                 rc = -EINTR;
2505                                 ptlrpc_interrupted_set(set);
2506                         } else {
2507                                 rc = 0;
2508                         }
2509                 } else {
2510                         /*
2511                          * At least one request is in flight, so no
2512                          * interrupts are allowed. Wait until all
2513                          * complete, or an in-flight req times out.
2514                          */
2515                         rc = wait_event_idle_timeout(
2516                                 set->set_waitq,
2517                                 ptlrpc_check_set(NULL, set),
2518                                 cfs_time_seconds(timeout ? timeout : 1));
2519                         if (rc == 0) {
2520                                 ptlrpc_expired_set(set);
2521                                 rc = -ETIMEDOUT;
2522                         } else {
2523                                 rc = 0;
2524                         }
2525
2526                         /*
2527                          * LU-769 - if we ignored the signal because
2528                          * it was already pending when we started, we
2529                          * need to handle it now or we risk it being
2530                          * ignored forever
2531                          */
2532                         if (rc == -ETIMEDOUT &&
2533                             signal_pending(current)) {
2534                                 sigset_t old, new;
2535
2536                                 siginitset(&new, LUSTRE_FATAL_SIGS);
2537                                 sigprocmask(SIG_BLOCK, &new, &old);
2538                                 /*
2539                                  * In fact we only interrupt for the
2540                                  * "fatal" signals like SIGINT or
2541                                  * SIGKILL. We still ignore less
2542                                  * important signals since ptlrpc set
2543                                  * is not easily reentrant from
2544                                  * userspace again
2545                                  */
2546                                 if (signal_pending(current))
2547                                         ptlrpc_interrupted_set(set);
2548                                 sigprocmask(SIG_SETMASK, &old, NULL);
2549                         }
2550                 }
2551
2552                 LASSERT(rc == 0 || rc == -EINTR || rc == -ETIMEDOUT);
2553
2554                 /*
2555                  * -EINTR => all requests have been flagged rq_intr so next
2556                  * check completes.
2557                  * -ETIMEDOUT => someone timed out.  When all reqs have
2558                  * timed out, signals are enabled allowing completion with
2559                  * EINTR.
2560                  * I don't really care if we go once more round the loop in
2561                  * the error cases -eeb.
2562                  */
2563                 if (rc == 0 && atomic_read(&set->set_remaining) == 0) {
2564                         list_for_each(tmp, &set->set_requests) {
2565                                 req = list_entry(tmp, struct ptlrpc_request,
2566                                                  rq_set_chain);
2567                                 spin_lock(&req->rq_lock);
2568                                 req->rq_invalid_rqset = 1;
2569                                 spin_unlock(&req->rq_lock);
2570                         }
2571                 }
2572         } while (rc != 0 || atomic_read(&set->set_remaining) != 0);
2573
2574         LASSERT(atomic_read(&set->set_remaining) == 0);
2575
2576         rc = set->set_rc; /* rq_status of already freed requests if any */
2577         list_for_each(tmp, &set->set_requests) {
2578                 req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
2579
2580                 LASSERT(req->rq_phase == RQ_PHASE_COMPLETE);
2581                 if (req->rq_status != 0)
2582                         rc = req->rq_status;
2583         }
2584
2585         RETURN(rc);
2586 }
2587 EXPORT_SYMBOL(ptlrpc_set_wait);
2588
2589 /**
2590  * Helper fuction for request freeing.
2591  * Called when request count reached zero and request needs to be freed.
2592  * Removes request from all sorts of sending/replay lists it might be on,
2593  * frees network buffers if any are present.
2594  * If \a locked is set, that means caller is already holding import imp_lock
2595  * and so we no longer need to reobtain it (for certain lists manipulations)
2596  */
2597 static void __ptlrpc_free_req(struct ptlrpc_request *request, int locked)
2598 {
2599         ENTRY;
2600
2601         if (!request)
2602                 RETURN_EXIT;
2603
2604         LASSERT(!request->rq_srv_req);
2605         LASSERT(request->rq_export == NULL);
2606         LASSERTF(!request->rq_receiving_reply, "req %p\n", request);
2607         LASSERTF(list_empty(&request->rq_list), "req %p\n", request);
2608         LASSERTF(list_empty(&request->rq_set_chain), "req %p\n", request);
2609         LASSERTF(!request->rq_replay, "req %p\n", request);
2610
2611         req_capsule_fini(&request->rq_pill);
2612
2613         /*
2614          * We must take it off the imp_replay_list first.  Otherwise, we'll set
2615          * request->rq_reqmsg to NULL while osc_close is dereferencing it.
2616          */
2617         if (request->rq_import) {
2618                 if (!locked)
2619                         spin_lock(&request->rq_import->imp_lock);
2620                 list_del_init(&request->rq_replay_list);
2621                 list_del_init(&request->rq_unreplied_list);
2622                 if (!locked)
2623                         spin_unlock(&request->rq_import->imp_lock);
2624         }
2625         LASSERTF(list_empty(&request->rq_replay_list), "req %p\n", request);
2626
2627         if (atomic_read(&request->rq_refcount) != 0) {
2628                 DEBUG_REQ(D_ERROR, request,
2629                           "freeing request with nonzero refcount");
2630                 LBUG();
2631         }
2632
2633         if (request->rq_repbuf)
2634                 sptlrpc_cli_free_repbuf(request);
2635
2636         if (request->rq_import) {
2637                 if (!ptlrpcd_check_work(request)) {
2638                         LASSERT(atomic_read(&request->rq_import->imp_reqs) > 0);
2639                         atomic_dec(&request->rq_import->imp_reqs);
2640                 }
2641                 class_import_put(request->rq_import);
2642                 request->rq_import = NULL;
2643         }
2644         if (request->rq_bulk)
2645                 ptlrpc_free_bulk(request->rq_bulk);
2646
2647         if (request->rq_reqbuf || request->rq_clrbuf)
2648                 sptlrpc_cli_free_reqbuf(request);
2649
2650         if (request->rq_cli_ctx)
2651                 sptlrpc_req_put_ctx(request, !locked);
2652
2653         if (request->rq_pool)
2654                 __ptlrpc_free_req_to_pool(request);
2655         else
2656                 ptlrpc_request_cache_free(request);
2657         EXIT;
2658 }
2659
2660 static int __ptlrpc_req_finished(struct ptlrpc_request *request, int locked);
2661 /**
2662  * Drop one request reference. Must be called with import imp_lock held.
2663  * When reference count drops to zero, request is freed.
2664  */
2665 void ptlrpc_req_finished_with_imp_lock(struct ptlrpc_request *request)
2666 {
2667         assert_spin_locked(&request->rq_import->imp_lock);
2668         (void)__ptlrpc_req_finished(request, 1);
2669 }
2670
2671 /**
2672  * Helper function
2673  * Drops one reference count for request \a request.
2674  * \a locked set indicates that caller holds import imp_lock.
2675  * Frees the request whe reference count reaches zero.
2676  *
2677  * \retval 1    the request is freed
2678  * \retval 0    some others still hold references on the request
2679  */
2680 static int __ptlrpc_req_finished(struct ptlrpc_request *request, int locked)
2681 {
2682         int count;
2683
2684         ENTRY;
2685         if (!request)
2686                 RETURN(1);
2687
2688         LASSERT(request != LP_POISON);
2689         LASSERT(request->rq_reqmsg != LP_POISON);
2690
2691         DEBUG_REQ(D_INFO, request, "refcount now %u",
2692                   atomic_read(&request->rq_refcount) - 1);
2693
2694         spin_lock(&request->rq_lock);
2695         count = atomic_dec_return(&request->rq_refcount);
2696         LASSERTF(count >= 0, "Invalid ref count %d\n", count);
2697
2698         /*
2699          * For open RPC, the client does not know the EA size (LOV, ACL, and
2700          * so on) before replied, then the client has to reserve very large
2701          * reply buffer. Such buffer will not be released until the RPC freed.
2702          * Since The open RPC is replayable, we need to keep it in the replay
2703          * list until close. If there are a lot of files opened concurrently,
2704          * then the client may be OOM.
2705          *
2706          * If fact, it is unnecessary to keep reply buffer for open replay,
2707          * related EAs have already been saved via mdc_save_lovea() before
2708          * coming here. So it is safe to free the reply buffer some earlier
2709          * before releasing the RPC to avoid client OOM. LU-9514
2710          */
2711         if (count == 1 && request->rq_early_free_repbuf && request->rq_repbuf) {
2712                 spin_lock(&request->rq_early_free_lock);
2713                 sptlrpc_cli_free_repbuf(request);
2714                 request->rq_repbuf = NULL;
2715                 request->rq_repbuf_len = 0;
2716                 request->rq_repdata = NULL;
2717                 request->rq_reqdata_len = 0;
2718                 spin_unlock(&request->rq_early_free_lock);
2719         }
2720         spin_unlock(&request->rq_lock);
2721
2722         if (!count)
2723                 __ptlrpc_free_req(request, locked);
2724
2725         RETURN(!count);
2726 }
2727
2728 /**
2729  * Drops one reference count for a request.
2730  */
2731 void ptlrpc_req_finished(struct ptlrpc_request *request)
2732 {
2733         __ptlrpc_req_finished(request, 0);
2734 }
2735 EXPORT_SYMBOL(ptlrpc_req_finished);
2736
2737 /**
2738  * Returns xid of a \a request
2739  */
2740 __u64 ptlrpc_req_xid(struct ptlrpc_request *request)
2741 {
2742         return request->rq_xid;
2743 }
2744 EXPORT_SYMBOL(ptlrpc_req_xid);
2745
2746 /**
2747  * Disengage the client's reply buffer from the network
2748  * NB does _NOT_ unregister any client-side bulk.
2749  * IDEMPOTENT, but _not_ safe against concurrent callers.
2750  * The request owner (i.e. the thread doing the I/O) must call...
2751  * Returns 0 on success or 1 if unregistering cannot be made.
2752  */
2753 static int ptlrpc_unregister_reply(struct ptlrpc_request *request, int async)
2754 {
2755         /*
2756          * Might sleep.
2757          */
2758         LASSERT(!in_interrupt());
2759
2760         /* Let's setup deadline for reply unlink. */
2761         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_REPL_UNLINK) &&
2762             async && request->rq_reply_deadline == 0 && cfs_fail_val == 0)
2763                 request->rq_reply_deadline = ktime_get_real_seconds() +
2764                                              PTLRPC_REQ_LONG_UNLINK;
2765
2766         /*
2767          * Nothing left to do.
2768          */
2769         if (!ptlrpc_client_recv_or_unlink(request))
2770                 RETURN(1);
2771
2772         LNetMDUnlink(request->rq_reply_md_h);
2773
2774         /*
2775          * Let's check it once again.
2776          */
2777         if (!ptlrpc_client_recv_or_unlink(request))
2778                 RETURN(1);
2779
2780         /* Move to "Unregistering" phase as reply was not unlinked yet. */
2781         ptlrpc_rqphase_move(request, RQ_PHASE_UNREG_RPC);
2782
2783         /*
2784          * Do not wait for unlink to finish.
2785          */
2786         if (async)
2787                 RETURN(0);
2788
2789         /*
2790          * We have to wait_event_idle_timeout() whatever the result, to get
2791          * a chance to run reply_in_callback(), and to make sure we've
2792          * unlinked before returning a req to the pool.
2793          */
2794         for (;;) {
2795                 wait_queue_head_t *wq = (request->rq_set) ?
2796                                         &request->rq_set->set_waitq :
2797                                         &request->rq_reply_waitq;
2798                 int seconds = PTLRPC_REQ_LONG_UNLINK;
2799                 /*
2800                  * Network access will complete in finite time but the HUGE
2801                  * timeout lets us CWARN for visibility of sluggish NALs
2802                  */
2803                 while (seconds > 0 &&
2804                        wait_event_idle_timeout(
2805                                *wq,
2806                                !ptlrpc_client_recv_or_unlink(request),
2807                                cfs_time_seconds(1)) == 0)
2808                         seconds -= 1;
2809                 if (seconds > 0) {
2810                         ptlrpc_rqphase_move(request, request->rq_next_phase);
2811                         RETURN(1);
2812                 }
2813
2814                 DEBUG_REQ(D_WARNING, request,
2815                           "Unexpectedly long timeout receiving_reply=%d req_ulinked=%d reply_unlinked=%d",
2816                           request->rq_receiving_reply,
2817                           request->rq_req_unlinked,
2818                           request->rq_reply_unlinked);
2819         }
2820         RETURN(0);
2821 }
2822
2823 static void ptlrpc_free_request(struct ptlrpc_request *req)
2824 {
2825         spin_lock(&req->rq_lock);
2826         req->rq_replay = 0;
2827         spin_unlock(&req->rq_lock);
2828
2829         if (req->rq_commit_cb)
2830                 req->rq_commit_cb(req);
2831         list_del_init(&req->rq_replay_list);
2832
2833         __ptlrpc_req_finished(req, 1);
2834 }
2835
2836 /**
2837  * the request is committed and dropped from the replay list of its import
2838  */
2839 void ptlrpc_request_committed(struct ptlrpc_request *req, int force)
2840 {
2841         struct obd_import *imp = req->rq_import;
2842
2843         spin_lock(&imp->imp_lock);
2844         if (list_empty(&req->rq_replay_list)) {
2845                 spin_unlock(&imp->imp_lock);
2846                 return;
2847         }
2848
2849         if (force || req->rq_transno <= imp->imp_peer_committed_transno) {
2850                 if (imp->imp_replay_cursor == &req->rq_replay_list)
2851                         imp->imp_replay_cursor = req->rq_replay_list.next;
2852                 ptlrpc_free_request(req);
2853         }
2854
2855         spin_unlock(&imp->imp_lock);
2856 }
2857 EXPORT_SYMBOL(ptlrpc_request_committed);
2858
2859 /**
2860  * Iterates through replay_list on import and prunes
2861  * all requests have transno smaller than last_committed for the
2862  * import and don't have rq_replay set.
2863  * Since requests are sorted in transno order, stops when meetign first
2864  * transno bigger than last_committed.
2865  * caller must hold imp->imp_lock
2866  */
2867 void ptlrpc_free_committed(struct obd_import *imp)
2868 {
2869         struct ptlrpc_request *req, *saved;
2870         struct ptlrpc_request *last_req = NULL; /* temporary fire escape */
2871         bool skip_committed_list = true;
2872
2873         ENTRY;
2874         LASSERT(imp != NULL);
2875         assert_spin_locked(&imp->imp_lock);
2876
2877         if (imp->imp_peer_committed_transno == imp->imp_last_transno_checked &&
2878             imp->imp_generation == imp->imp_last_generation_checked) {
2879                 CDEBUG(D_INFO, "%s: skip recheck: last_committed %llu\n",
2880                        imp->imp_obd->obd_name, imp->imp_peer_committed_transno);
2881                 RETURN_EXIT;
2882         }
2883         CDEBUG(D_RPCTRACE, "%s: committing for last_committed %llu gen %d\n",
2884                imp->imp_obd->obd_name, imp->imp_peer_committed_transno,
2885                imp->imp_generation);
2886
2887         if (imp->imp_generation != imp->imp_last_generation_checked ||
2888             imp->imp_last_transno_checked == 0)
2889                 skip_committed_list = false;
2890
2891         imp->imp_last_transno_checked = imp->imp_peer_committed_transno;
2892         imp->imp_last_generation_checked = imp->imp_generation;
2893
2894         list_for_each_entry_safe(req, saved, &imp->imp_replay_list,
2895                                  rq_replay_list) {
2896                 /* XXX ok to remove when 1357 resolved - rread 05/29/03  */
2897                 LASSERT(req != last_req);
2898                 last_req = req;
2899
2900                 if (req->rq_transno == 0) {
2901                         DEBUG_REQ(D_EMERG, req, "zero transno during replay");
2902                         LBUG();
2903                 }
2904                 if (req->rq_import_generation < imp->imp_generation) {
2905                         DEBUG_REQ(D_RPCTRACE, req, "free request with old gen");
2906                         GOTO(free_req, 0);
2907                 }
2908
2909                 /* not yet committed */
2910                 if (req->rq_transno > imp->imp_peer_committed_transno) {
2911                         DEBUG_REQ(D_RPCTRACE, req, "stopping search");
2912                         break;
2913                 }
2914
2915                 if (req->rq_replay) {
2916                         DEBUG_REQ(D_RPCTRACE, req, "keeping (FL_REPLAY)");
2917                         list_move_tail(&req->rq_replay_list,
2918                                        &imp->imp_committed_list);
2919                         continue;
2920                 }
2921
2922                 DEBUG_REQ(D_INFO, req, "commit (last_committed %llu)",
2923                           imp->imp_peer_committed_transno);
2924 free_req:
2925                 ptlrpc_free_request(req);
2926         }
2927
2928         if (skip_committed_list)
2929                 GOTO(out, 0);
2930
2931         list_for_each_entry_safe(req, saved, &imp->imp_committed_list,
2932                                  rq_replay_list) {
2933                 LASSERT(req->rq_transno != 0);
2934                 if (req->rq_import_generation < imp->imp_generation ||
2935                     !req->rq_replay) {
2936                         DEBUG_REQ(D_RPCTRACE, req, "free %s open request",
2937                                   req->rq_import_generation <
2938                                   imp->imp_generation ? "stale" : "closed");
2939
2940                         if (imp->imp_replay_cursor == &req->rq_replay_list)
2941                                 imp->imp_replay_cursor =
2942                                         req->rq_replay_list.next;
2943
2944                         ptlrpc_free_request(req);
2945                 }
2946         }
2947 out:
2948         EXIT;
2949 }
2950
2951 void ptlrpc_cleanup_client(struct obd_import *imp)
2952 {
2953         ENTRY;
2954         EXIT;
2955 }
2956
2957 /**
2958  * Schedule previously sent request for resend.
2959  * For bulk requests we assign new xid (to avoid problems with
2960  * lost replies and therefore several transfers landing into same buffer
2961  * from different sending attempts).
2962  */
2963 void ptlrpc_resend_req(struct ptlrpc_request *req)
2964 {
2965         DEBUG_REQ(D_HA, req, "going to resend");
2966         spin_lock(&req->rq_lock);
2967
2968         /*
2969          * Request got reply but linked to the import list still.
2970          * Let ptlrpc_check_set() process it.
2971          */
2972         if (ptlrpc_client_replied(req)) {
2973                 spin_unlock(&req->rq_lock);
2974                 DEBUG_REQ(D_HA, req, "it has reply, so skip it");
2975                 return;
2976         }
2977
2978         req->rq_status = -EAGAIN;
2979
2980         req->rq_resend = 1;
2981         req->rq_net_err = 0;
2982         req->rq_timedout = 0;
2983
2984         ptlrpc_client_wake_req(req);
2985         spin_unlock(&req->rq_lock);
2986 }
2987
2988 /* XXX: this function and rq_status are currently unused */
2989 void ptlrpc_restart_req(struct ptlrpc_request *req)
2990 {
2991         DEBUG_REQ(D_HA, req, "restarting (possibly-)completed request");
2992         req->rq_status = -ERESTARTSYS;
2993
2994         spin_lock(&req->rq_lock);
2995         req->rq_restart = 1;
2996         req->rq_timedout = 0;
2997         ptlrpc_client_wake_req(req);
2998         spin_unlock(&req->rq_lock);
2999 }
3000
3001 /**
3002  * Grab additional reference on a request \a req
3003  */
3004 struct ptlrpc_request *ptlrpc_request_addref(struct ptlrpc_request *req)
3005 {
3006         ENTRY;
3007         atomic_inc(&req->rq_refcount);
3008         RETURN(req);
3009 }
3010 EXPORT_SYMBOL(ptlrpc_request_addref);
3011
3012 /**
3013  * Add a request to import replay_list.
3014  * Must be called under imp_lock
3015  */
3016 void ptlrpc_retain_replayable_request(struct ptlrpc_request *req,
3017                                       struct obd_import *imp)
3018 {
3019         struct list_head *tmp;
3020
3021         assert_spin_locked(&imp->imp_lock);
3022
3023         if (req->rq_transno == 0) {
3024                 DEBUG_REQ(D_EMERG, req, "saving request with zero transno");
3025                 LBUG();
3026         }
3027
3028         /*
3029          * clear this for new requests that were resent as well
3030          * as resent replayed requests.
3031          */
3032         lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT);
3033
3034         /* don't re-add requests that have been replayed */
3035         if (!list_empty(&req->rq_replay_list))
3036                 return;
3037
3038         lustre_msg_add_flags(req->rq_reqmsg, MSG_REPLAY);
3039
3040         spin_lock(&req->rq_lock);
3041         req->rq_resend = 0;
3042         spin_unlock(&req->rq_lock);
3043
3044         LASSERT(imp->imp_replayable);
3045         /* Balanced in ptlrpc_free_committed, usually. */
3046         ptlrpc_request_addref(req);
3047         list_for_each_prev(tmp, &imp->imp_replay_list) {
3048                 struct ptlrpc_request *iter = list_entry(tmp,
3049                                                          struct ptlrpc_request,
3050                                                          rq_replay_list);
3051
3052                 /*
3053                  * We may have duplicate transnos if we create and then
3054                  * open a file, or for closes retained if to match creating
3055                  * opens, so use req->rq_xid as a secondary key.
3056                  * (See bugs 684, 685, and 428.)
3057                  * XXX no longer needed, but all opens need transnos!
3058                  */
3059                 if (iter->rq_transno > req->rq_transno)
3060                         continue;
3061
3062                 if (iter->rq_transno == req->rq_transno) {
3063                         LASSERT(iter->rq_xid != req->rq_xid);
3064                         if (iter->rq_xid > req->rq_xid)
3065                                 continue;
3066                 }
3067
3068                 list_add(&req->rq_replay_list, &iter->rq_replay_list);
3069                 return;
3070         }
3071
3072         list_add(&req->rq_replay_list, &imp->imp_replay_list);
3073 }
3074
3075 /**
3076  * Send request and wait until it completes.
3077  * Returns request processing status.
3078  */
3079 int ptlrpc_queue_wait(struct ptlrpc_request *req)
3080 {
3081         struct ptlrpc_request_set *set;
3082         int rc;
3083
3084         ENTRY;
3085         LASSERT(req->rq_set == NULL);
3086         LASSERT(!req->rq_receiving_reply);
3087
3088         set = ptlrpc_prep_set();
3089         if (!set) {
3090                 CERROR("cannot allocate ptlrpc set: rc = %d\n", -ENOMEM);
3091                 RETURN(-ENOMEM);
3092         }
3093
3094         /* for distributed debugging */
3095         lustre_msg_set_status(req->rq_reqmsg, current->pid);
3096
3097         /* add a ref for the set (see comment in ptlrpc_set_add_req) */
3098         ptlrpc_request_addref(req);
3099         ptlrpc_set_add_req(set, req);
3100         rc = ptlrpc_set_wait(NULL, set);
3101         ptlrpc_set_destroy(set);
3102
3103         RETURN(rc);
3104 }
3105 EXPORT_SYMBOL(ptlrpc_queue_wait);
3106
3107 /**
3108  * Callback used for replayed requests reply processing.
3109  * In case of successful reply calls registered request replay callback.
3110  * In case of error restart replay process.
3111  */
3112 static int ptlrpc_replay_interpret(const struct lu_env *env,
3113                                    struct ptlrpc_request *req,
3114                                    void *args, int rc)
3115 {
3116         struct ptlrpc_replay_async_args *aa = args;
3117         struct obd_import *imp = req->rq_import;
3118
3119         ENTRY;
3120         atomic_dec(&imp->imp_replay_inflight);
3121
3122         /*
3123          * Note: if it is bulk replay (MDS-MDS replay), then even if
3124          * server got the request, but bulk transfer timeout, let's
3125          * replay the bulk req again
3126          */
3127         if (!ptlrpc_client_replied(req) ||
3128             (req->rq_bulk &&
3129              lustre_msg_get_status(req->rq_repmsg) == -ETIMEDOUT)) {
3130                 DEBUG_REQ(D_ERROR, req, "request replay timed out");
3131                 GOTO(out, rc = -ETIMEDOUT);
3132         }
3133
3134         if (lustre_msg_get_type(req->rq_repmsg) == PTL_RPC_MSG_ERR &&
3135             (lustre_msg_get_status(req->rq_repmsg) == -ENOTCONN ||
3136             lustre_msg_get_status(req->rq_repmsg) == -ENODEV))
3137                 GOTO(out, rc = lustre_msg_get_status(req->rq_repmsg));
3138
3139         /** VBR: check version failure */
3140         if (lustre_msg_get_status(req->rq_repmsg) == -EOVERFLOW) {
3141                 /** replay was failed due to version mismatch */
3142                 DEBUG_REQ(D_WARNING, req, "Version mismatch during replay");
3143                 spin_lock(&imp->imp_lock);
3144                 imp->imp_vbr_failed = 1;
3145                 spin_unlock(&imp->imp_lock);
3146                 lustre_msg_set_status(req->rq_repmsg, aa->praa_old_status);
3147         } else {
3148                 /** The transno had better not change over replay. */
3149                 LASSERTF(lustre_msg_get_transno(req->rq_reqmsg) ==
3150                          lustre_msg_get_transno(req->rq_repmsg) ||
3151                          lustre_msg_get_transno(req->rq_repmsg) == 0,
3152                          "%#llx/%#llx\n",
3153                          lustre_msg_get_transno(req->rq_reqmsg),
3154                          lustre_msg_get_transno(req->rq_repmsg));
3155         }
3156
3157         spin_lock(&imp->imp_lock);
3158         imp->imp_last_replay_transno = lustre_msg_get_transno(req->rq_reqmsg);
3159         spin_unlock(&imp->imp_lock);
3160         LASSERT(imp->imp_last_replay_transno);
3161
3162         /* transaction number shouldn't be bigger than the latest replayed */
3163         if (req->rq_transno > lustre_msg_get_transno(req->rq_reqmsg)) {
3164                 DEBUG_REQ(D_ERROR, req,
3165                           "Reported transno=%llu is bigger than replayed=%llu",
3166                           req->rq_transno,
3167                           lustre_msg_get_transno(req->rq_reqmsg));
3168                 GOTO(out, rc = -EINVAL);
3169         }
3170
3171         DEBUG_REQ(D_HA, req, "got reply");
3172
3173         /* let the callback do fixups, possibly including in the request */
3174         if (req->rq_replay_cb)
3175                 req->rq_replay_cb(req);
3176
3177         if (ptlrpc_client_replied(req) &&
3178             lustre_msg_get_status(req->rq_repmsg) != aa->praa_old_status) {
3179                 DEBUG_REQ(D_ERROR, req, "status %d, old was %d",
3180                           lustre_msg_get_status(req->rq_repmsg),
3181                           aa->praa_old_status);
3182
3183                 /*
3184                  * Note: If the replay fails for MDT-MDT recovery, let's
3185                  * abort all of the following requests in the replay
3186                  * and sending list, because MDT-MDT update requests
3187                  * are dependent on each other, see LU-7039
3188                  */
3189                 if (imp->imp_connect_flags_orig & OBD_CONNECT_MDS_MDS) {
3190                         struct ptlrpc_request *free_req;
3191                         struct ptlrpc_request *tmp;
3192
3193                         spin_lock(&imp->imp_lock);
3194                         list_for_each_entry_safe(free_req, tmp,
3195                                                  &imp->imp_replay_list,
3196                                                  rq_replay_list) {
3197                                 ptlrpc_free_request(free_req);
3198                         }
3199
3200                         list_for_each_entry_safe(free_req, tmp,
3201     &nb