Whamcloud - gitweb
LU-6142 lustre: ptlrpc: don't use list_for_each_entry_safe unnecessarily.
[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 ptlrpc_request *req;
555
556         LASSERT(pool != NULL);
557
558         spin_lock(&pool->prp_lock);
559         while ((req = list_first_entry_or_null(&pool->prp_req_list,
560                                                struct ptlrpc_request,
561                                                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 ptlrpc_request *iter;
706
707         assert_spin_locked(&imp->imp_lock);
708         LASSERT(list_empty(&req->rq_unreplied_list));
709
710         /* unreplied list is sorted by xid in ascending order */
711         list_for_each_entry_reverse(iter, &imp->imp_unreplied_list,
712                                     rq_unreplied_list) {
713                 LASSERT(req->rq_xid != iter->rq_xid);
714                 if (req->rq_xid < iter->rq_xid)
715                         continue;
716                 list_add(&req->rq_unreplied_list, &iter->rq_unreplied_list);
717                 return;
718         }
719         list_add(&req->rq_unreplied_list, &imp->imp_unreplied_list);
720 }
721
722 void ptlrpc_assign_next_xid_nolock(struct ptlrpc_request *req)
723 {
724         req->rq_xid = ptlrpc_next_xid();
725         ptlrpc_add_unreplied(req);
726 }
727
728 static inline void ptlrpc_assign_next_xid(struct ptlrpc_request *req)
729 {
730         spin_lock(&req->rq_import->imp_lock);
731         ptlrpc_assign_next_xid_nolock(req);
732         spin_unlock(&req->rq_import->imp_lock);
733 }
734
735 static atomic64_t ptlrpc_last_xid;
736
737 static void ptlrpc_reassign_next_xid(struct ptlrpc_request *req)
738 {
739         spin_lock(&req->rq_import->imp_lock);
740         list_del_init(&req->rq_unreplied_list);
741         ptlrpc_assign_next_xid_nolock(req);
742         spin_unlock(&req->rq_import->imp_lock);
743         DEBUG_REQ(D_RPCTRACE, req, "reassign xid");
744 }
745
746 void ptlrpc_get_mod_rpc_slot(struct ptlrpc_request *req)
747 {
748         struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
749         __u32 opc;
750         __u16 tag;
751
752         opc = lustre_msg_get_opc(req->rq_reqmsg);
753         tag = obd_get_mod_rpc_slot(cli, opc);
754         lustre_msg_set_tag(req->rq_reqmsg, tag);
755         ptlrpc_reassign_next_xid(req);
756 }
757 EXPORT_SYMBOL(ptlrpc_get_mod_rpc_slot);
758
759 void ptlrpc_put_mod_rpc_slot(struct ptlrpc_request *req)
760 {
761         __u16 tag = lustre_msg_get_tag(req->rq_reqmsg);
762
763         if (tag != 0) {
764                 struct client_obd *cli = &req->rq_import->imp_obd->u.cli;
765                 __u32 opc = lustre_msg_get_opc(req->rq_reqmsg);
766
767                 obd_put_mod_rpc_slot(cli, opc, tag);
768         }
769 }
770 EXPORT_SYMBOL(ptlrpc_put_mod_rpc_slot);
771
772 int ptlrpc_request_bufs_pack(struct ptlrpc_request *request,
773                              __u32 version, int opcode, char **bufs,
774                              struct ptlrpc_cli_ctx *ctx)
775 {
776         int count;
777         struct obd_import *imp;
778         __u32 *lengths;
779         int rc;
780
781         ENTRY;
782
783         count = req_capsule_filled_sizes(&request->rq_pill, RCL_CLIENT);
784         imp = request->rq_import;
785         lengths = request->rq_pill.rc_area[RCL_CLIENT];
786
787         if (ctx) {
788                 request->rq_cli_ctx = sptlrpc_cli_ctx_get(ctx);
789         } else {
790                 rc = sptlrpc_req_get_ctx(request);
791                 if (rc)
792                         GOTO(out_free, rc);
793         }
794         sptlrpc_req_set_flavor(request, opcode);
795
796         rc = lustre_pack_request(request, imp->imp_msg_magic, count,
797                                  lengths, bufs);
798         if (rc)
799                 GOTO(out_ctx, rc);
800
801         lustre_msg_add_version(request->rq_reqmsg, version);
802         request->rq_send_state = LUSTRE_IMP_FULL;
803         request->rq_type = PTL_RPC_MSG_REQUEST;
804
805         request->rq_req_cbid.cbid_fn  = request_out_callback;
806         request->rq_req_cbid.cbid_arg = request;
807
808         request->rq_reply_cbid.cbid_fn  = reply_in_callback;
809         request->rq_reply_cbid.cbid_arg = request;
810
811         request->rq_reply_deadline = 0;
812         request->rq_bulk_deadline = 0;
813         request->rq_req_deadline = 0;
814         request->rq_phase = RQ_PHASE_NEW;
815         request->rq_next_phase = RQ_PHASE_UNDEFINED;
816
817         request->rq_request_portal = imp->imp_client->cli_request_portal;
818         request->rq_reply_portal = imp->imp_client->cli_reply_portal;
819
820         ptlrpc_at_set_req_timeout(request);
821
822         lustre_msg_set_opc(request->rq_reqmsg, opcode);
823
824         /* Let's setup deadline for req/reply/bulk unlink for opcode. */
825         if (cfs_fail_val == opcode) {
826                 time64_t *fail_t = NULL, *fail2_t = NULL;
827
828                 if (CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_BULK_UNLINK)) {
829                         fail_t = &request->rq_bulk_deadline;
830                 } else if (CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_REPL_UNLINK)) {
831                         fail_t = &request->rq_reply_deadline;
832                 } else if (CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_REQ_UNLINK)) {
833                         fail_t = &request->rq_req_deadline;
834                 } else if (CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_BOTH_UNLINK)) {
835                         fail_t = &request->rq_reply_deadline;
836                         fail2_t = &request->rq_bulk_deadline;
837                 } else if (CFS_FAIL_CHECK(OBD_FAIL_PTLRPC_ROUND_XID)) {
838                         time64_t now = ktime_get_real_seconds();
839                         u64 xid = ((u64)now >> 4) << 24;
840
841                         atomic64_set(&ptlrpc_last_xid, xid);
842                 }
843
844                 if (fail_t) {
845                         *fail_t = ktime_get_real_seconds() +
846                                   PTLRPC_REQ_LONG_UNLINK;
847
848                         if (fail2_t)
849                                 *fail2_t = ktime_get_real_seconds() +
850                                            PTLRPC_REQ_LONG_UNLINK;
851
852                         /*
853                          * The RPC is infected, let the test to change the
854                          * fail_loc
855                          */
856                         msleep(4 * MSEC_PER_SEC);
857                 }
858         }
859         ptlrpc_assign_next_xid(request);
860
861         RETURN(0);
862
863 out_ctx:
864         LASSERT(!request->rq_pool);
865         sptlrpc_cli_ctx_put(request->rq_cli_ctx, 1);
866 out_free:
867         atomic_dec(&imp->imp_reqs);
868         class_import_put(imp);
869
870         return rc;
871 }
872 EXPORT_SYMBOL(ptlrpc_request_bufs_pack);
873
874 /**
875  * Pack request buffers for network transfer, performing necessary encryption
876  * steps if necessary.
877  */
878 int ptlrpc_request_pack(struct ptlrpc_request *request,
879                         __u32 version, int opcode)
880 {
881         return ptlrpc_request_bufs_pack(request, version, opcode, NULL, NULL);
882 }
883 EXPORT_SYMBOL(ptlrpc_request_pack);
884
885 /**
886  * Helper function to allocate new request on import \a imp
887  * and possibly using existing request from pool \a pool if provided.
888  * Returns allocated request structure with import field filled or
889  * NULL on error.
890  */
891 static inline
892 struct ptlrpc_request *__ptlrpc_request_alloc(struct obd_import *imp,
893                                               struct ptlrpc_request_pool *pool)
894 {
895         struct ptlrpc_request *request = NULL;
896
897         request = ptlrpc_request_cache_alloc(GFP_NOFS);
898
899         if (!request && pool)
900                 request = ptlrpc_prep_req_from_pool(pool);
901
902         if (request) {
903                 ptlrpc_cli_req_init(request);
904
905                 LASSERTF((unsigned long)imp > 0x1000, "%p\n", imp);
906                 LASSERT(imp != LP_POISON);
907                 LASSERTF((unsigned long)imp->imp_client > 0x1000, "%p\n",
908                          imp->imp_client);
909                 LASSERT(imp->imp_client != LP_POISON);
910
911                 request->rq_import = class_import_get(imp);
912                 atomic_inc(&imp->imp_reqs);
913         } else {
914                 CERROR("request allocation out of memory\n");
915         }
916
917         return request;
918 }
919
920 static int ptlrpc_reconnect_if_idle(struct obd_import *imp)
921 {
922         int rc;
923
924         /*
925          * initiate connection if needed when the import has been
926          * referenced by the new request to avoid races with disconnect.
927          * serialize this check against conditional state=IDLE
928          * in ptlrpc_disconnect_idle_interpret()
929          */
930         spin_lock(&imp->imp_lock);
931         if (imp->imp_state == LUSTRE_IMP_IDLE) {
932                 imp->imp_generation++;
933                 imp->imp_initiated_at = imp->imp_generation;
934                 imp->imp_state = LUSTRE_IMP_NEW;
935
936                 /* connect_import_locked releases imp_lock */
937                 rc = ptlrpc_connect_import_locked(imp);
938                 if (rc)
939                         return rc;
940                 ptlrpc_pinger_add_import(imp);
941         } else {
942                 spin_unlock(&imp->imp_lock);
943         }
944         return 0;
945 }
946
947 /**
948  * Helper function for creating a request.
949  * Calls __ptlrpc_request_alloc to allocate new request sturcture and inits
950  * buffer structures according to capsule template \a format.
951  * Returns allocated request structure pointer or NULL on error.
952  */
953 static struct ptlrpc_request *
954 ptlrpc_request_alloc_internal(struct obd_import *imp,
955                               struct ptlrpc_request_pool *pool,
956                               const struct req_format *format)
957 {
958         struct ptlrpc_request *request;
959
960         request = __ptlrpc_request_alloc(imp, pool);
961         if (!request)
962                 return NULL;
963
964         /* don't make expensive check for idling connection
965          * if it's already connected */
966         if (unlikely(imp->imp_state != LUSTRE_IMP_FULL)) {
967                 if (ptlrpc_reconnect_if_idle(imp) < 0) {
968                         atomic_dec(&imp->imp_reqs);
969                         ptlrpc_request_free(request);
970                         return NULL;
971                 }
972         }
973
974         req_capsule_init(&request->rq_pill, request, RCL_CLIENT);
975         req_capsule_set(&request->rq_pill, format);
976         return request;
977 }
978
979 /**
980  * Allocate new request structure for import \a imp and initialize its
981  * buffer structure according to capsule template \a format.
982  */
983 struct ptlrpc_request *ptlrpc_request_alloc(struct obd_import *imp,
984                                             const struct req_format *format)
985 {
986         return ptlrpc_request_alloc_internal(imp, NULL, format);
987 }
988 EXPORT_SYMBOL(ptlrpc_request_alloc);
989
990 /**
991  * Allocate new request structure for import \a imp from pool \a pool and
992  * initialize its buffer structure according to capsule template \a format.
993  */
994 struct ptlrpc_request *
995 ptlrpc_request_alloc_pool(struct obd_import *imp,
996                           struct ptlrpc_request_pool *pool,
997                           const struct req_format *format)
998 {
999         return ptlrpc_request_alloc_internal(imp, pool, format);
1000 }
1001 EXPORT_SYMBOL(ptlrpc_request_alloc_pool);
1002
1003 /**
1004  * For requests not from pool, free memory of the request structure.
1005  * For requests obtained from a pool earlier, return request back to pool.
1006  */
1007 void ptlrpc_request_free(struct ptlrpc_request *request)
1008 {
1009         if (request->rq_pool)
1010                 __ptlrpc_free_req_to_pool(request);
1011         else
1012                 ptlrpc_request_cache_free(request);
1013 }
1014 EXPORT_SYMBOL(ptlrpc_request_free);
1015
1016 /**
1017  * Allocate new request for operatione \a opcode and immediatelly pack it for
1018  * network transfer.
1019  * Only used for simple requests like OBD_PING where the only important
1020  * part of the request is operation itself.
1021  * Returns allocated request or NULL on error.
1022  */
1023 struct ptlrpc_request *ptlrpc_request_alloc_pack(struct obd_import *imp,
1024                                                  const struct req_format *format,
1025                                                  __u32 version, int opcode)
1026 {
1027         struct ptlrpc_request *req = ptlrpc_request_alloc(imp, format);
1028         int rc;
1029
1030         if (req) {
1031                 rc = ptlrpc_request_pack(req, version, opcode);
1032                 if (rc) {
1033                         ptlrpc_request_free(req);
1034                         req = NULL;
1035                 }
1036         }
1037         return req;
1038 }
1039 EXPORT_SYMBOL(ptlrpc_request_alloc_pack);
1040
1041 /**
1042  * Allocate and initialize new request set structure on the current CPT.
1043  * Returns a pointer to the newly allocated set structure or NULL on error.
1044  */
1045 struct ptlrpc_request_set *ptlrpc_prep_set(void)
1046 {
1047         struct ptlrpc_request_set *set;
1048         int cpt;
1049
1050         ENTRY;
1051         cpt = cfs_cpt_current(cfs_cpt_tab, 0);
1052         OBD_CPT_ALLOC(set, cfs_cpt_tab, cpt, sizeof(*set));
1053         if (!set)
1054                 RETURN(NULL);
1055         atomic_set(&set->set_refcount, 1);
1056         INIT_LIST_HEAD(&set->set_requests);
1057         init_waitqueue_head(&set->set_waitq);
1058         atomic_set(&set->set_new_count, 0);
1059         atomic_set(&set->set_remaining, 0);
1060         spin_lock_init(&set->set_new_req_lock);
1061         INIT_LIST_HEAD(&set->set_new_requests);
1062         set->set_max_inflight = UINT_MAX;
1063         set->set_producer     = NULL;
1064         set->set_producer_arg = NULL;
1065         set->set_rc           = 0;
1066
1067         RETURN(set);
1068 }
1069 EXPORT_SYMBOL(ptlrpc_prep_set);
1070
1071 /**
1072  * Allocate and initialize new request set structure with flow control
1073  * extension. This extension allows to control the number of requests in-flight
1074  * for the whole set. A callback function to generate requests must be provided
1075  * and the request set will keep the number of requests sent over the wire to
1076  * @max_inflight.
1077  * Returns a pointer to the newly allocated set structure or NULL on error.
1078  */
1079 struct ptlrpc_request_set *ptlrpc_prep_fcset(int max, set_producer_func func,
1080                                              void *arg)
1081
1082 {
1083         struct ptlrpc_request_set *set;
1084
1085         set = ptlrpc_prep_set();
1086         if (!set)
1087                 RETURN(NULL);
1088
1089         set->set_max_inflight  = max;
1090         set->set_producer      = func;
1091         set->set_producer_arg  = arg;
1092
1093         RETURN(set);
1094 }
1095
1096 /**
1097  * Wind down and free request set structure previously allocated with
1098  * ptlrpc_prep_set.
1099  * Ensures that all requests on the set have completed and removes
1100  * all requests from the request list in a set.
1101  * If any unsent request happen to be on the list, pretends that they got
1102  * an error in flight and calls their completion handler.
1103  */
1104 void ptlrpc_set_destroy(struct ptlrpc_request_set *set)
1105 {
1106         struct ptlrpc_request *req;
1107         int expected_phase;
1108         int n = 0;
1109
1110         ENTRY;
1111
1112         /* Requests on the set should either all be completed, or all be new */
1113         expected_phase = (atomic_read(&set->set_remaining) == 0) ?
1114                          RQ_PHASE_COMPLETE : RQ_PHASE_NEW;
1115         list_for_each_entry(req, &set->set_requests, rq_set_chain) {
1116                 LASSERT(req->rq_phase == expected_phase);
1117                 n++;
1118         }
1119
1120         LASSERTF(atomic_read(&set->set_remaining) == 0 ||
1121                  atomic_read(&set->set_remaining) == n, "%d / %d\n",
1122                  atomic_read(&set->set_remaining), n);
1123
1124         while ((req = list_first_entry_or_null(&set->set_requests,
1125                                                struct ptlrpc_request,
1126                                                rq_set_chain))) {
1127                 list_del_init(&req->rq_set_chain);
1128
1129                 LASSERT(req->rq_phase == expected_phase);
1130
1131                 if (req->rq_phase == RQ_PHASE_NEW) {
1132                         ptlrpc_req_interpret(NULL, req, -EBADR);
1133                         atomic_dec(&set->set_remaining);
1134                 }
1135
1136                 spin_lock(&req->rq_lock);
1137                 req->rq_set = NULL;
1138                 req->rq_invalid_rqset = 0;
1139                 spin_unlock(&req->rq_lock);
1140
1141                 ptlrpc_req_finished(req);
1142         }
1143
1144         LASSERT(atomic_read(&set->set_remaining) == 0);
1145
1146         ptlrpc_reqset_put(set);
1147         EXIT;
1148 }
1149 EXPORT_SYMBOL(ptlrpc_set_destroy);
1150
1151 /**
1152  * Add a new request to the general purpose request set.
1153  * Assumes request reference from the caller.
1154  */
1155 void ptlrpc_set_add_req(struct ptlrpc_request_set *set,
1156                         struct ptlrpc_request *req)
1157 {
1158         if (set == PTLRPCD_SET) {
1159                 ptlrpcd_add_req(req);
1160                 return;
1161         }
1162
1163         LASSERT(req->rq_import->imp_state != LUSTRE_IMP_IDLE);
1164         LASSERT(list_empty(&req->rq_set_chain));
1165
1166         if (req->rq_allow_intr)
1167                 set->set_allow_intr = 1;
1168
1169         /* The set takes over the caller's request reference */
1170         list_add_tail(&req->rq_set_chain, &set->set_requests);
1171         req->rq_set = set;
1172         atomic_inc(&set->set_remaining);
1173         req->rq_queued_time = ktime_get_seconds();
1174
1175         if (req->rq_reqmsg)
1176                 lustre_msg_set_jobid(req->rq_reqmsg, NULL);
1177
1178         if (set->set_producer)
1179                 /*
1180                  * If the request set has a producer callback, the RPC must be
1181                  * sent straight away
1182                  */
1183                 ptlrpc_send_new_req(req);
1184 }
1185 EXPORT_SYMBOL(ptlrpc_set_add_req);
1186
1187 /**
1188  * Add a request to a request with dedicated server thread
1189  * and wake the thread to make any necessary processing.
1190  * Currently only used for ptlrpcd.
1191  */
1192 void ptlrpc_set_add_new_req(struct ptlrpcd_ctl *pc,
1193                             struct ptlrpc_request *req)
1194 {
1195         struct ptlrpc_request_set *set = pc->pc_set;
1196         int count, i;
1197
1198         LASSERT(req->rq_set == NULL);
1199         LASSERT(test_bit(LIOD_STOP, &pc->pc_flags) == 0);
1200
1201         spin_lock(&set->set_new_req_lock);
1202         /*
1203          * The set takes over the caller's request reference.
1204          */
1205         req->rq_set = set;
1206         req->rq_queued_time = ktime_get_seconds();
1207         list_add_tail(&req->rq_set_chain, &set->set_new_requests);
1208         count = atomic_inc_return(&set->set_new_count);
1209         spin_unlock(&set->set_new_req_lock);
1210
1211         /* Only need to call wakeup once for the first entry. */
1212         if (count == 1) {
1213                 wake_up(&set->set_waitq);
1214
1215                 /*
1216                  * XXX: It maybe unnecessary to wakeup all the partners. But to
1217                  *      guarantee the async RPC can be processed ASAP, we have
1218                  *      no other better choice. It maybe fixed in future.
1219                  */
1220                 for (i = 0; i < pc->pc_npartners; i++)
1221                         wake_up(&pc->pc_partners[i]->pc_set->set_waitq);
1222         }
1223 }
1224
1225 /**
1226  * Based on the current state of the import, determine if the request
1227  * can be sent, is an error, or should be delayed.
1228  *
1229  * Returns true if this request should be delayed. If false, and
1230  * *status is set, then the request can not be sent and *status is the
1231  * error code.  If false and status is 0, then request can be sent.
1232  *
1233  * The imp->imp_lock must be held.
1234  */
1235 static int ptlrpc_import_delay_req(struct obd_import *imp,
1236                                    struct ptlrpc_request *req, int *status)
1237 {
1238         int delay = 0;
1239
1240         ENTRY;
1241         LASSERT(status);
1242         *status = 0;
1243
1244         if (req->rq_ctx_init || req->rq_ctx_fini) {
1245                 /* always allow ctx init/fini rpc go through */
1246         } else if (imp->imp_state == LUSTRE_IMP_NEW) {
1247                 DEBUG_REQ(D_ERROR, req, "Uninitialized import");
1248                 *status = -EIO;
1249         } else if (imp->imp_state == LUSTRE_IMP_CLOSED) {
1250                 unsigned int opc = lustre_msg_get_opc(req->rq_reqmsg);
1251
1252                 /*
1253                  * pings or MDS-equivalent STATFS may safely
1254                  * race with umount
1255                  */
1256                 DEBUG_REQ((opc == OBD_PING || opc == OST_STATFS) ?
1257                           D_HA : D_ERROR, req, "IMP_CLOSED");
1258                 *status = -EIO;
1259         } else if (ptlrpc_send_limit_expired(req)) {
1260                 /* probably doesn't need to be a D_ERROR afterinitial testing */
1261                 DEBUG_REQ(D_HA, req, "send limit expired");
1262                 *status = -ETIMEDOUT;
1263         } else if (req->rq_send_state == LUSTRE_IMP_CONNECTING &&
1264                    imp->imp_state == LUSTRE_IMP_CONNECTING) {
1265                 ;/* allow CONNECT even if import is invalid */
1266                 if (atomic_read(&imp->imp_inval_count) != 0) {
1267                         DEBUG_REQ(D_ERROR, req, "invalidate in flight");
1268                         *status = -EIO;
1269                 }
1270         } else if (imp->imp_invalid || imp->imp_obd->obd_no_recov) {
1271                 if (!imp->imp_deactive)
1272                         DEBUG_REQ(D_NET, req, "IMP_INVALID");
1273                 *status = -ESHUTDOWN; /* b=12940 */
1274         } else if (req->rq_import_generation != imp->imp_generation) {
1275                 DEBUG_REQ(D_ERROR, req, "req wrong generation:");
1276                 *status = -EIO;
1277         } else if (req->rq_send_state != imp->imp_state) {
1278                 /* invalidate in progress - any requests should be drop */
1279                 if (atomic_read(&imp->imp_inval_count) != 0) {
1280                         DEBUG_REQ(D_ERROR, req, "invalidate in flight");
1281                         *status = -EIO;
1282                 } else if (req->rq_no_delay &&
1283                            imp->imp_generation != imp->imp_initiated_at) {
1284                         /* ignore nodelay for requests initiating connections */
1285                         *status = -EAGAIN;
1286                 } else if (req->rq_allow_replay &&
1287                            (imp->imp_state == LUSTRE_IMP_REPLAY ||
1288                             imp->imp_state == LUSTRE_IMP_REPLAY_LOCKS ||
1289                             imp->imp_state == LUSTRE_IMP_REPLAY_WAIT ||
1290                             imp->imp_state == LUSTRE_IMP_RECOVER)) {
1291                         DEBUG_REQ(D_HA, req, "allow during recovery");
1292                 } else {
1293                         delay = 1;
1294                 }
1295         }
1296
1297         RETURN(delay);
1298 }
1299
1300 /**
1301  * Decide if the error message should be printed to the console or not.
1302  * Makes its decision based on request type, status, and failure frequency.
1303  *
1304  * \param[in] req  request that failed and may need a console message
1305  *
1306  * \retval false if no message should be printed
1307  * \retval true  if console message should be printed
1308  */
1309 static bool ptlrpc_console_allow(struct ptlrpc_request *req, __u32 opc, int err)
1310 {
1311         LASSERT(req->rq_reqmsg != NULL);
1312
1313         /* Suppress particular reconnect errors which are to be expected. */
1314         if (opc == OST_CONNECT || opc == MDS_CONNECT || opc == MGS_CONNECT) {
1315                 /* Suppress timed out reconnect requests */
1316                 if (lustre_handle_is_used(&req->rq_import->imp_remote_handle) ||
1317                     req->rq_timedout)
1318                         return false;
1319
1320                 /*
1321                  * Suppress most unavailable/again reconnect requests, but
1322                  * print occasionally so it is clear client is trying to
1323                  * connect to a server where no target is running.
1324                  */
1325                 if ((err == -ENODEV || err == -EAGAIN) &&
1326                     req->rq_import->imp_conn_cnt % 30 != 20)
1327                         return false;
1328         }
1329
1330         if (opc == LDLM_ENQUEUE && err == -EAGAIN)
1331                 /* -EAGAIN is normal when using POSIX flocks */
1332                 return false;
1333
1334         if (opc == OBD_PING && (err == -ENODEV || err == -ENOTCONN) &&
1335             (req->rq_xid & 0xf) != 10)
1336                 /* Suppress most ping requests, they may fail occasionally */
1337                 return false;
1338
1339         return true;
1340 }
1341
1342 /**
1343  * Check request processing status.
1344  * Returns the status.
1345  */
1346 static int ptlrpc_check_status(struct ptlrpc_request *req)
1347 {
1348         int rc;
1349
1350         ENTRY;
1351         rc = lustre_msg_get_status(req->rq_repmsg);
1352         if (lustre_msg_get_type(req->rq_repmsg) == PTL_RPC_MSG_ERR) {
1353                 struct obd_import *imp = req->rq_import;
1354                 lnet_nid_t nid = imp->imp_connection->c_peer.nid;
1355                 __u32 opc = lustre_msg_get_opc(req->rq_reqmsg);
1356
1357                 if (ptlrpc_console_allow(req, opc, rc))
1358                         LCONSOLE_ERROR_MSG(0x11,
1359                                            "%s: operation %s to node %s failed: rc = %d\n",
1360                                            imp->imp_obd->obd_name,
1361                                            ll_opcode2str(opc),
1362                                            libcfs_nid2str(nid), rc);
1363                 RETURN(rc < 0 ? rc : -EINVAL);
1364         }
1365
1366         if (rc)
1367                 DEBUG_REQ(D_INFO, req, "check status: rc = %d", rc);
1368
1369         RETURN(rc);
1370 }
1371
1372 /**
1373  * save pre-versions of objects into request for replay.
1374  * Versions are obtained from server reply.
1375  * used for VBR.
1376  */
1377 static void ptlrpc_save_versions(struct ptlrpc_request *req)
1378 {
1379         struct lustre_msg *repmsg = req->rq_repmsg;
1380         struct lustre_msg *reqmsg = req->rq_reqmsg;
1381         __u64 *versions = lustre_msg_get_versions(repmsg);
1382
1383         ENTRY;
1384         if (lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY)
1385                 return;
1386
1387         LASSERT(versions);
1388         lustre_msg_set_versions(reqmsg, versions);
1389         CDEBUG(D_INFO, "Client save versions [%#llx/%#llx]\n",
1390                versions[0], versions[1]);
1391
1392         EXIT;
1393 }
1394
1395 __u64 ptlrpc_known_replied_xid(struct obd_import *imp)
1396 {
1397         struct ptlrpc_request *req;
1398
1399         assert_spin_locked(&imp->imp_lock);
1400         if (list_empty(&imp->imp_unreplied_list))
1401                 return 0;
1402
1403         req = list_entry(imp->imp_unreplied_list.next, struct ptlrpc_request,
1404                          rq_unreplied_list);
1405         LASSERTF(req->rq_xid >= 1, "XID:%llu\n", req->rq_xid);
1406
1407         if (imp->imp_known_replied_xid < req->rq_xid - 1)
1408                 imp->imp_known_replied_xid = req->rq_xid - 1;
1409
1410         return req->rq_xid - 1;
1411 }
1412
1413 /**
1414  * Callback function called when client receives RPC reply for \a req.
1415  * Returns 0 on success or error code.
1416  * The return alue would be assigned to req->rq_status by the caller
1417  * as request processing status.
1418  * This function also decides if the request needs to be saved for later replay.
1419  */
1420 static int after_reply(struct ptlrpc_request *req)
1421 {
1422         struct obd_import *imp = req->rq_import;
1423         struct obd_device *obd = req->rq_import->imp_obd;
1424         ktime_t work_start;
1425         u64 committed;
1426         s64 timediff;
1427         int rc;
1428
1429         ENTRY;
1430         LASSERT(obd != NULL);
1431         /* repbuf must be unlinked */
1432         LASSERT(!req->rq_receiving_reply && req->rq_reply_unlinked);
1433
1434         if (req->rq_reply_truncated) {
1435                 if (ptlrpc_no_resend(req)) {
1436                         DEBUG_REQ(D_ERROR, req,
1437                                   "reply buffer overflow, expected=%d, actual size=%d",
1438                                   req->rq_nob_received, req->rq_repbuf_len);
1439                         RETURN(-EOVERFLOW);
1440                 }
1441
1442                 sptlrpc_cli_free_repbuf(req);
1443                 /*
1444                  * Pass the required reply buffer size (include
1445                  * space for early reply).
1446                  * NB: no need to roundup because alloc_repbuf
1447                  * will roundup it
1448                  */
1449                 req->rq_replen = req->rq_nob_received;
1450                 req->rq_nob_received = 0;
1451                 spin_lock(&req->rq_lock);
1452                 req->rq_resend       = 1;
1453                 spin_unlock(&req->rq_lock);
1454                 RETURN(0);
1455         }
1456
1457         work_start = ktime_get_real();
1458         timediff = ktime_us_delta(work_start, req->rq_sent_ns);
1459
1460         /*
1461          * NB Until this point, the whole of the incoming message,
1462          * including buflens, status etc is in the sender's byte order.
1463          */
1464         rc = sptlrpc_cli_unwrap_reply(req);
1465         if (rc) {
1466                 DEBUG_REQ(D_ERROR, req, "unwrap reply failed: rc = %d", rc);
1467                 RETURN(rc);
1468         }
1469
1470         /*
1471          * Security layer unwrap might ask resend this request.
1472          */
1473         if (req->rq_resend)
1474                 RETURN(0);
1475
1476         rc = unpack_reply(req);
1477         if (rc)
1478                 RETURN(rc);
1479
1480         /* retry indefinitely on EINPROGRESS */
1481         if (lustre_msg_get_status(req->rq_repmsg) == -EINPROGRESS &&
1482             ptlrpc_no_resend(req) == 0 && !req->rq_no_retry_einprogress) {
1483                 time64_t now = ktime_get_real_seconds();
1484
1485                 DEBUG_REQ((req->rq_nr_resend % 8 == 1 ? D_WARNING : 0) |
1486                           D_RPCTRACE, req, "resending request on EINPROGRESS");
1487                 spin_lock(&req->rq_lock);
1488                 req->rq_resend = 1;
1489                 spin_unlock(&req->rq_lock);
1490                 req->rq_nr_resend++;
1491
1492                 /* Readjust the timeout for current conditions */
1493                 ptlrpc_at_set_req_timeout(req);
1494                 /*
1495                  * delay resend to give a chance to the server to get ready.
1496                  * The delay is increased by 1s on every resend and is capped to
1497                  * the current request timeout (i.e. obd_timeout if AT is off,
1498                  * or AT service time x 125% + 5s, see at_est2timeout)
1499                  */
1500                 if (req->rq_nr_resend > req->rq_timeout)
1501                         req->rq_sent = now + req->rq_timeout;
1502                 else
1503                         req->rq_sent = now + req->rq_nr_resend;
1504
1505                 /* Resend for EINPROGRESS will use a new XID */
1506                 spin_lock(&imp->imp_lock);
1507                 list_del_init(&req->rq_unreplied_list);
1508                 spin_unlock(&imp->imp_lock);
1509
1510                 RETURN(0);
1511         }
1512
1513         if (obd->obd_svc_stats) {
1514                 lprocfs_counter_add(obd->obd_svc_stats, PTLRPC_REQWAIT_CNTR,
1515                                     timediff);
1516                 ptlrpc_lprocfs_rpc_sent(req, timediff);
1517         }
1518
1519         if (lustre_msg_get_type(req->rq_repmsg) != PTL_RPC_MSG_REPLY &&
1520             lustre_msg_get_type(req->rq_repmsg) != PTL_RPC_MSG_ERR) {
1521                 DEBUG_REQ(D_ERROR, req, "invalid packet received (type=%u)",
1522                           lustre_msg_get_type(req->rq_repmsg));
1523                 RETURN(-EPROTO);
1524         }
1525
1526         if (lustre_msg_get_opc(req->rq_reqmsg) != OBD_PING)
1527                 CFS_FAIL_TIMEOUT(OBD_FAIL_PTLRPC_PAUSE_REP, cfs_fail_val);
1528         ptlrpc_at_adj_service(req, lustre_msg_get_timeout(req->rq_repmsg));
1529         ptlrpc_at_adj_net_latency(req,
1530                                   lustre_msg_get_service_timeout(req->rq_repmsg));
1531
1532         rc = ptlrpc_check_status(req);
1533
1534         if (rc) {
1535                 /*
1536                  * Either we've been evicted, or the server has failed for
1537                  * some reason. Try to reconnect, and if that fails, punt to
1538                  * the upcall.
1539                  */
1540                 if (ptlrpc_recoverable_error(rc)) {
1541                         if (req->rq_send_state != LUSTRE_IMP_FULL ||
1542                             imp->imp_obd->obd_no_recov || imp->imp_dlm_fake) {
1543                                 RETURN(rc);
1544                         }
1545                         ptlrpc_request_handle_notconn(req);
1546                         RETURN(rc);
1547                 }
1548         } else {
1549                 /*
1550                  * Let's look if server sent slv. Do it only for RPC with
1551                  * rc == 0.
1552                  */
1553                 ldlm_cli_update_pool(req);
1554         }
1555
1556         /*
1557          * Store transno in reqmsg for replay.
1558          */
1559         if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY)) {
1560                 req->rq_transno = lustre_msg_get_transno(req->rq_repmsg);
1561                 lustre_msg_set_transno(req->rq_reqmsg, req->rq_transno);
1562         }
1563
1564         if (imp->imp_replayable) {
1565                 spin_lock(&imp->imp_lock);
1566                 /*
1567                  * No point in adding already-committed requests to the replay
1568                  * list, we will just remove them immediately. b=9829
1569                  */
1570                 if (req->rq_transno != 0 &&
1571                     (req->rq_transno >
1572                      lustre_msg_get_last_committed(req->rq_repmsg) ||
1573                      req->rq_replay)) {
1574                         /** version recovery */
1575                         ptlrpc_save_versions(req);
1576                         ptlrpc_retain_replayable_request(req, imp);
1577                 } else if (req->rq_commit_cb &&
1578                            list_empty(&req->rq_replay_list)) {
1579                         /*
1580                          * NB: don't call rq_commit_cb if it's already on
1581                          * rq_replay_list, ptlrpc_free_committed() will call
1582                          * it later, see LU-3618 for details
1583                          */
1584                         spin_unlock(&imp->imp_lock);
1585                         req->rq_commit_cb(req);
1586                         spin_lock(&imp->imp_lock);
1587                 }
1588
1589                 /*
1590                  * Replay-enabled imports return commit-status information.
1591                  */
1592                 committed = lustre_msg_get_last_committed(req->rq_repmsg);
1593                 if (likely(committed > imp->imp_peer_committed_transno))
1594                         imp->imp_peer_committed_transno = committed;
1595
1596                 ptlrpc_free_committed(imp);
1597
1598                 if (!list_empty(&imp->imp_replay_list)) {
1599                         struct ptlrpc_request *last;
1600
1601                         last = list_entry(imp->imp_replay_list.prev,
1602                                           struct ptlrpc_request,
1603                                           rq_replay_list);
1604                         /*
1605                          * Requests with rq_replay stay on the list even if no
1606                          * commit is expected.
1607                          */
1608                         if (last->rq_transno > imp->imp_peer_committed_transno)
1609                                 ptlrpc_pinger_commit_expected(imp);
1610                 }
1611
1612                 spin_unlock(&imp->imp_lock);
1613         }
1614
1615         RETURN(rc);
1616 }
1617
1618 /**
1619  * Helper function to send request \a req over the network for the first time
1620  * Also adjusts request phase.
1621  * Returns 0 on success or error code.
1622  */
1623 static int ptlrpc_send_new_req(struct ptlrpc_request *req)
1624 {
1625         struct obd_import *imp = req->rq_import;
1626         __u64 min_xid = 0;
1627         int rc;
1628
1629         ENTRY;
1630         LASSERT(req->rq_phase == RQ_PHASE_NEW);
1631
1632         /* do not try to go further if there is not enough memory in enc_pool */
1633         if (req->rq_sent && req->rq_bulk)
1634                 if (req->rq_bulk->bd_iov_count > get_free_pages_in_pool() &&
1635                     pool_is_at_full_capacity())
1636                         RETURN(-ENOMEM);
1637
1638         if (req->rq_sent && (req->rq_sent > ktime_get_real_seconds()) &&
1639             (!req->rq_generation_set ||
1640              req->rq_import_generation == imp->imp_generation))
1641                 RETURN(0);
1642
1643         ptlrpc_rqphase_move(req, RQ_PHASE_RPC);
1644
1645         spin_lock(&imp->imp_lock);
1646
1647         LASSERT(req->rq_xid != 0);
1648         LASSERT(!list_empty(&req->rq_unreplied_list));
1649
1650         if (!req->rq_generation_set)
1651                 req->rq_import_generation = imp->imp_generation;
1652
1653         if (ptlrpc_import_delay_req(imp, req, &rc)) {
1654                 spin_lock(&req->rq_lock);
1655                 req->rq_waiting = 1;
1656                 spin_unlock(&req->rq_lock);
1657
1658                 DEBUG_REQ(D_HA, req, "req waiting for recovery: (%s != %s)",
1659                           ptlrpc_import_state_name(req->rq_send_state),
1660                           ptlrpc_import_state_name(imp->imp_state));
1661                 LASSERT(list_empty(&req->rq_list));
1662                 list_add_tail(&req->rq_list, &imp->imp_delayed_list);
1663                 atomic_inc(&req->rq_import->imp_inflight);
1664                 spin_unlock(&imp->imp_lock);
1665                 RETURN(0);
1666         }
1667
1668         if (rc != 0) {
1669                 spin_unlock(&imp->imp_lock);
1670                 req->rq_status = rc;
1671                 ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
1672                 RETURN(rc);
1673         }
1674
1675         LASSERT(list_empty(&req->rq_list));
1676         list_add_tail(&req->rq_list, &imp->imp_sending_list);
1677         atomic_inc(&req->rq_import->imp_inflight);
1678
1679         /*
1680          * find the known replied XID from the unreplied list, CONNECT
1681          * and DISCONNECT requests are skipped to make the sanity check
1682          * on server side happy. see process_req_last_xid().
1683          *
1684          * For CONNECT: Because replay requests have lower XID, it'll
1685          * break the sanity check if CONNECT bump the exp_last_xid on
1686          * server.
1687          *
1688          * For DISCONNECT: Since client will abort inflight RPC before
1689          * sending DISCONNECT, DISCONNECT may carry an XID which higher
1690          * than the inflight RPC.
1691          */
1692         if (!ptlrpc_req_is_connect(req) && !ptlrpc_req_is_disconnect(req))
1693                 min_xid = ptlrpc_known_replied_xid(imp);
1694         spin_unlock(&imp->imp_lock);
1695
1696         lustre_msg_set_last_xid(req->rq_reqmsg, min_xid);
1697
1698         lustre_msg_set_status(req->rq_reqmsg, current->pid);
1699
1700         rc = sptlrpc_req_refresh_ctx(req, 0);
1701         if (rc) {
1702                 if (req->rq_err) {
1703                         req->rq_status = rc;
1704                         RETURN(1);
1705                 } else {
1706                         spin_lock(&req->rq_lock);
1707                         req->rq_wait_ctx = 1;
1708                         spin_unlock(&req->rq_lock);
1709                         RETURN(0);
1710                 }
1711         }
1712
1713         CDEBUG(D_RPCTRACE,
1714                "Sending RPC req@%p pname:cluuid:pid:xid:nid:opc:job %s:%s:%d:%llu:%s:%d:%s\n",
1715                req, current->comm,
1716                imp->imp_obd->obd_uuid.uuid,
1717                lustre_msg_get_status(req->rq_reqmsg), req->rq_xid,
1718                obd_import_nid2str(imp), lustre_msg_get_opc(req->rq_reqmsg),
1719                lustre_msg_get_jobid(req->rq_reqmsg) ?: "");
1720
1721         rc = ptl_send_rpc(req, 0);
1722         if (rc == -ENOMEM) {
1723                 spin_lock(&imp->imp_lock);
1724                 if (!list_empty(&req->rq_list)) {
1725                         list_del_init(&req->rq_list);
1726                         if (atomic_dec_and_test(&req->rq_import->imp_inflight))
1727                                 wake_up(&req->rq_import->imp_recovery_waitq);
1728                 }
1729                 spin_unlock(&imp->imp_lock);
1730                 ptlrpc_rqphase_move(req, RQ_PHASE_NEW);
1731                 RETURN(rc);
1732         }
1733         if (rc) {
1734                 DEBUG_REQ(D_HA, req, "send failed, expect timeout: rc = %d",
1735                           rc);
1736                 spin_lock(&req->rq_lock);
1737                 req->rq_net_err = 1;
1738                 spin_unlock(&req->rq_lock);
1739                 RETURN(rc);
1740         }
1741         RETURN(0);
1742 }
1743
1744 static inline int ptlrpc_set_producer(struct ptlrpc_request_set *set)
1745 {
1746         int remaining, rc;
1747
1748         ENTRY;
1749         LASSERT(set->set_producer != NULL);
1750
1751         remaining = atomic_read(&set->set_remaining);
1752
1753         /*
1754          * populate the ->set_requests list with requests until we
1755          * reach the maximum number of RPCs in flight for this set
1756          */
1757         while (atomic_read(&set->set_remaining) < set->set_max_inflight) {
1758                 rc = set->set_producer(set, set->set_producer_arg);
1759                 if (rc == -ENOENT) {
1760                         /* no more RPC to produce */
1761                         set->set_producer     = NULL;
1762                         set->set_producer_arg = NULL;
1763                         RETURN(0);
1764                 }
1765         }
1766
1767         RETURN((atomic_read(&set->set_remaining) - remaining));
1768 }
1769
1770 /**
1771  * this sends any unsent RPCs in \a set and returns 1 if all are sent
1772  * and no more replies are expected.
1773  * (it is possible to get less replies than requests sent e.g. due to timed out
1774  * requests or requests that we had trouble to send out)
1775  *
1776  * NOTE: This function contains a potential schedule point (cond_resched()).
1777  */
1778 int ptlrpc_check_set(const struct lu_env *env, struct ptlrpc_request_set *set)
1779 {
1780         struct ptlrpc_request *req, *next;
1781         LIST_HEAD(comp_reqs);
1782         int force_timer_recalc = 0;
1783
1784         ENTRY;
1785         if (atomic_read(&set->set_remaining) == 0)
1786                 RETURN(1);
1787
1788         list_for_each_entry_safe(req, next, &set->set_requests,
1789                                  rq_set_chain) {
1790                 struct obd_import *imp = req->rq_import;
1791                 int unregistered = 0;
1792                 int async = 1;
1793                 int rc = 0;
1794
1795                 if (req->rq_phase == RQ_PHASE_COMPLETE) {
1796                         list_move_tail(&req->rq_set_chain, &comp_reqs);
1797                         continue;
1798                 }
1799
1800                 /*
1801                  * This schedule point is mainly for the ptlrpcd caller of this
1802                  * function.  Most ptlrpc sets are not long-lived and unbounded
1803                  * in length, but at the least the set used by the ptlrpcd is.
1804                  * Since the processing time is unbounded, we need to insert an
1805                  * explicit schedule point to make the thread well-behaved.
1806                  */
1807                 cond_resched();
1808
1809                 /*
1810                  * If the caller requires to allow to be interpreted by force
1811                  * and it has really been interpreted, then move the request
1812                  * to RQ_PHASE_INTERPRET phase in spite of what the current
1813                  * phase is.
1814                  */
1815                 if (unlikely(req->rq_allow_intr && req->rq_intr)) {
1816                         req->rq_status = -EINTR;
1817                         ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
1818
1819                         /*
1820                          * Since it is interpreted and we have to wait for
1821                          * the reply to be unlinked, then use sync mode.
1822                          */
1823                         async = 0;
1824
1825                         GOTO(interpret, req->rq_status);
1826                 }
1827
1828                 if (req->rq_phase == RQ_PHASE_NEW && ptlrpc_send_new_req(req))
1829                         force_timer_recalc = 1;
1830
1831                 /* delayed send - skip */
1832                 if (req->rq_phase == RQ_PHASE_NEW && req->rq_sent)
1833                         continue;
1834
1835                 /* delayed resend - skip */
1836                 if (req->rq_phase == RQ_PHASE_RPC && req->rq_resend &&
1837                     req->rq_sent > ktime_get_real_seconds())
1838                         continue;
1839
1840                 if (!(req->rq_phase == RQ_PHASE_RPC ||
1841                       req->rq_phase == RQ_PHASE_BULK ||
1842                       req->rq_phase == RQ_PHASE_INTERPRET ||
1843                       req->rq_phase == RQ_PHASE_UNREG_RPC ||
1844                       req->rq_phase == RQ_PHASE_UNREG_BULK)) {
1845                         DEBUG_REQ(D_ERROR, req, "bad phase %x", req->rq_phase);
1846                         LBUG();
1847                 }
1848
1849                 if (req->rq_phase == RQ_PHASE_UNREG_RPC ||
1850                     req->rq_phase == RQ_PHASE_UNREG_BULK) {
1851                         LASSERT(req->rq_next_phase != req->rq_phase);
1852                         LASSERT(req->rq_next_phase != RQ_PHASE_UNDEFINED);
1853
1854                         if (req->rq_req_deadline &&
1855                             !OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_REQ_UNLINK))
1856                                 req->rq_req_deadline = 0;
1857                         if (req->rq_reply_deadline &&
1858                             !OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_REPL_UNLINK))
1859                                 req->rq_reply_deadline = 0;
1860                         if (req->rq_bulk_deadline &&
1861                             !OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_BULK_UNLINK))
1862                                 req->rq_bulk_deadline = 0;
1863
1864                         /*
1865                          * Skip processing until reply is unlinked. We
1866                          * can't return to pool before that and we can't
1867                          * call interpret before that. We need to make
1868                          * sure that all rdma transfers finished and will
1869                          * not corrupt any data.
1870                          */
1871                         if (req->rq_phase == RQ_PHASE_UNREG_RPC &&
1872                             ptlrpc_cli_wait_unlink(req))
1873                                 continue;
1874                         if (req->rq_phase == RQ_PHASE_UNREG_BULK &&
1875                             ptlrpc_client_bulk_active(req))
1876                                 continue;
1877
1878                         /*
1879                          * Turn fail_loc off to prevent it from looping
1880                          * forever.
1881                          */
1882                         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_REPL_UNLINK)) {
1883                                 OBD_FAIL_CHECK_ORSET(OBD_FAIL_PTLRPC_LONG_REPL_UNLINK,
1884                                                      OBD_FAIL_ONCE);
1885                         }
1886                         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_BULK_UNLINK)) {
1887                                 OBD_FAIL_CHECK_ORSET(OBD_FAIL_PTLRPC_LONG_BULK_UNLINK,
1888                                                      OBD_FAIL_ONCE);
1889                         }
1890
1891                         /*
1892                          * Move to next phase if reply was successfully
1893                          * unlinked.
1894                          */
1895                         ptlrpc_rqphase_move(req, req->rq_next_phase);
1896                 }
1897
1898                 if (req->rq_phase == RQ_PHASE_INTERPRET)
1899                         GOTO(interpret, req->rq_status);
1900
1901                 /*
1902                  * Note that this also will start async reply unlink.
1903                  */
1904                 if (req->rq_net_err && !req->rq_timedout) {
1905                         ptlrpc_expire_one_request(req, 1);
1906
1907                         /*
1908                          * Check if we still need to wait for unlink.
1909                          */
1910                         if (ptlrpc_cli_wait_unlink(req) ||
1911                             ptlrpc_client_bulk_active(req))
1912                                 continue;
1913                         /* If there is no need to resend, fail it now. */
1914                         if (req->rq_no_resend) {
1915                                 if (req->rq_status == 0)
1916                                         req->rq_status = -EIO;
1917                                 ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
1918                                 GOTO(interpret, req->rq_status);
1919                         } else {
1920                                 continue;
1921                         }
1922                 }
1923
1924                 if (req->rq_err) {
1925                         spin_lock(&req->rq_lock);
1926                         req->rq_replied = 0;
1927                         spin_unlock(&req->rq_lock);
1928                         if (req->rq_status == 0)
1929                                 req->rq_status = -EIO;
1930                         ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
1931                         GOTO(interpret, req->rq_status);
1932                 }
1933
1934                 /*
1935                  * ptlrpc_set_wait uses l_wait_event_abortable_timeout()
1936                  * so it sets rq_intr regardless of individual rpc
1937                  * timeouts. The synchronous IO waiting path sets
1938                  * rq_intr irrespective of whether ptlrpcd
1939                  * has seen a timeout.  Our policy is to only interpret
1940                  * interrupted rpcs after they have timed out, so we
1941                  * need to enforce that here.
1942                  */
1943
1944                 if (req->rq_intr && (req->rq_timedout || req->rq_waiting ||
1945                                      req->rq_wait_ctx)) {
1946                         req->rq_status = -EINTR;
1947                         ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
1948                         GOTO(interpret, req->rq_status);
1949                 }
1950
1951                 if (req->rq_phase == RQ_PHASE_RPC) {
1952                         if (req->rq_timedout || req->rq_resend ||
1953                             req->rq_waiting || req->rq_wait_ctx) {
1954                                 int status;
1955
1956                                 if (!ptlrpc_unregister_reply(req, 1)) {
1957                                         ptlrpc_unregister_bulk(req, 1);
1958                                         continue;
1959                                 }
1960
1961                                 spin_lock(&imp->imp_lock);
1962                                 if (ptlrpc_import_delay_req(imp, req,
1963                                                             &status)) {
1964                                         /*
1965                                          * put on delay list - only if we wait
1966                                          * recovery finished - before send
1967                                          */
1968                                         list_move_tail(&req->rq_list,
1969                                                        &imp->imp_delayed_list);
1970                                         spin_unlock(&imp->imp_lock);
1971                                         continue;
1972                                 }
1973
1974                                 if (status != 0)  {
1975                                         req->rq_status = status;
1976                                         ptlrpc_rqphase_move(req,
1977                                                             RQ_PHASE_INTERPRET);
1978                                         spin_unlock(&imp->imp_lock);
1979                                         GOTO(interpret, req->rq_status);
1980                                 }
1981                                 /* ignore on just initiated connections */
1982                                 if (ptlrpc_no_resend(req) &&
1983                                     !req->rq_wait_ctx &&
1984                                     imp->imp_generation !=
1985                                     imp->imp_initiated_at) {
1986                                         req->rq_status = -ENOTCONN;
1987                                         ptlrpc_rqphase_move(req,
1988                                                             RQ_PHASE_INTERPRET);
1989                                         spin_unlock(&imp->imp_lock);
1990                                         GOTO(interpret, req->rq_status);
1991                                 }
1992
1993                                 /* don't resend too fast in case of network
1994                                  * errors.
1995                                  */
1996                                 if (ktime_get_real_seconds() < (req->rq_sent + 1)
1997                                     && req->rq_net_err && req->rq_timedout) {
1998
1999                                         DEBUG_REQ(D_INFO, req,
2000                                                   "throttle request");
2001                                         /* Don't try to resend RPC right away
2002                                          * as it is likely it will fail again
2003                                          * and ptlrpc_check_set() will be
2004                                          * called again, keeping this thread
2005                                          * busy. Instead, wait for the next
2006                                          * timeout. Flag it as resend to
2007                                          * ensure we don't wait to long.
2008                                          */
2009                                         req->rq_resend = 1;
2010                                         spin_unlock(&imp->imp_lock);
2011                                         continue;
2012                                 }
2013
2014                                 list_move_tail(&req->rq_list,
2015                                                &imp->imp_sending_list);
2016
2017                                 spin_unlock(&imp->imp_lock);
2018
2019                                 spin_lock(&req->rq_lock);
2020                                 req->rq_waiting = 0;
2021                                 spin_unlock(&req->rq_lock);
2022
2023                                 if (req->rq_timedout || req->rq_resend) {
2024                                         /*
2025                                          * This is re-sending anyways,
2026                                          * let's mark req as resend.
2027                                          */
2028                                         spin_lock(&req->rq_lock);
2029                                         req->rq_resend = 1;
2030                                         spin_unlock(&req->rq_lock);
2031                                 }
2032                                 /*
2033                                  * rq_wait_ctx is only touched by ptlrpcd,
2034                                  * so no lock is needed here.
2035                                  */
2036                                 status = sptlrpc_req_refresh_ctx(req, 0);
2037                                 if (status) {
2038                                         if (req->rq_err) {
2039                                                 req->rq_status = status;
2040                                                 spin_lock(&req->rq_lock);
2041                                                 req->rq_wait_ctx = 0;
2042                                                 spin_unlock(&req->rq_lock);
2043                                                 force_timer_recalc = 1;
2044                                         } else {
2045                                                 spin_lock(&req->rq_lock);
2046                                                 req->rq_wait_ctx = 1;
2047                                                 spin_unlock(&req->rq_lock);
2048                                         }
2049
2050                                         continue;
2051                                 } else {
2052                                         spin_lock(&req->rq_lock);
2053                                         req->rq_wait_ctx = 0;
2054                                         spin_unlock(&req->rq_lock);
2055                                 }
2056
2057                                 /*
2058                                  * In any case, the previous bulk should be
2059                                  * cleaned up to prepare for the new sending
2060                                  */
2061                                 if (req->rq_bulk &&
2062                                     !ptlrpc_unregister_bulk(req, 1))
2063                                         continue;
2064
2065                                 rc = ptl_send_rpc(req, 0);
2066                                 if (rc == -ENOMEM) {
2067                                         spin_lock(&imp->imp_lock);
2068                                         if (!list_empty(&req->rq_list))
2069                                                 list_del_init(&req->rq_list);
2070                                         spin_unlock(&imp->imp_lock);
2071                                         ptlrpc_rqphase_move(req, RQ_PHASE_NEW);
2072                                         continue;
2073                                 }
2074                                 if (rc) {
2075                                         DEBUG_REQ(D_HA, req,
2076                                                   "send failed: rc = %d", rc);
2077                                         force_timer_recalc = 1;
2078                                         spin_lock(&req->rq_lock);
2079                                         req->rq_net_err = 1;
2080                                         spin_unlock(&req->rq_lock);
2081                                         continue;
2082                                 }
2083                                 /* need to reset the timeout */
2084                                 force_timer_recalc = 1;
2085                         }
2086
2087                         spin_lock(&req->rq_lock);
2088
2089                         if (ptlrpc_client_early(req)) {
2090                                 ptlrpc_at_recv_early_reply(req);
2091                                 spin_unlock(&req->rq_lock);
2092                                 continue;
2093                         }
2094
2095                         /* Still waiting for a reply? */
2096                         if (ptlrpc_client_recv(req)) {
2097                                 spin_unlock(&req->rq_lock);
2098                                 continue;
2099                         }
2100
2101                         /* Did we actually receive a reply? */
2102                         if (!ptlrpc_client_replied(req)) {
2103                                 spin_unlock(&req->rq_lock);
2104                                 continue;
2105                         }
2106
2107                         spin_unlock(&req->rq_lock);
2108
2109                         /*
2110                          * unlink from net because we are going to
2111                          * swab in-place of reply buffer
2112                          */
2113                         unregistered = ptlrpc_unregister_reply(req, 1);
2114                         if (!unregistered)
2115                                 continue;
2116
2117                         req->rq_status = after_reply(req);
2118                         if (req->rq_resend)
2119                                 continue;
2120
2121                         /*
2122                          * If there is no bulk associated with this request,
2123                          * then we're done and should let the interpreter
2124                          * process the reply. Similarly if the RPC returned
2125                          * an error, and therefore the bulk will never arrive.
2126                          */
2127                         if (!req->rq_bulk || req->rq_status < 0) {
2128                                 ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
2129                                 GOTO(interpret, req->rq_status);
2130                         }
2131
2132                         ptlrpc_rqphase_move(req, RQ_PHASE_BULK);
2133                 }
2134
2135                 LASSERT(req->rq_phase == RQ_PHASE_BULK);
2136                 if (ptlrpc_client_bulk_active(req))
2137                         continue;
2138
2139                 if (req->rq_bulk->bd_failure) {
2140                         /*
2141                          * The RPC reply arrived OK, but the bulk screwed
2142                          * up!  Dead weird since the server told us the RPC
2143                          * was good after getting the REPLY for her GET or
2144                          * the ACK for her PUT.
2145                          */
2146                         DEBUG_REQ(D_ERROR, req, "bulk transfer failed");
2147                         req->rq_status = -EIO;
2148                 }
2149
2150                 ptlrpc_rqphase_move(req, RQ_PHASE_INTERPRET);
2151
2152 interpret:
2153                 LASSERT(req->rq_phase == RQ_PHASE_INTERPRET);
2154
2155                 /*
2156                  * This moves to "unregistering" phase we need to wait for
2157                  * reply unlink.
2158                  */
2159                 if (!unregistered && !ptlrpc_unregister_reply(req, async)) {
2160                         /* start async bulk unlink too */
2161                         ptlrpc_unregister_bulk(req, 1);
2162                         continue;
2163                 }
2164
2165                 if (!ptlrpc_unregister_bulk(req, async))
2166                         continue;
2167
2168                 /*
2169                  * When calling interpret receiving already should be
2170                  * finished.
2171                  */
2172                 LASSERT(!req->rq_receiving_reply);
2173
2174                 ptlrpc_req_interpret(env, req, req->rq_status);
2175
2176                 if (ptlrpcd_check_work(req)) {
2177                         atomic_dec(&set->set_remaining);
2178                         continue;
2179                 }
2180                 ptlrpc_rqphase_move(req, RQ_PHASE_COMPLETE);
2181
2182                 if (req->rq_reqmsg)
2183                         CDEBUG(D_RPCTRACE,
2184                                "Completed RPC req@%p pname:cluuid:pid:xid:nid:opc:job %s:%s:%d:%llu:%s:%d:%s\n",
2185                                req, current->comm,
2186                                imp->imp_obd->obd_uuid.uuid,
2187                                lustre_msg_get_status(req->rq_reqmsg),
2188                                req->rq_xid,
2189                                obd_import_nid2str(imp),
2190                                lustre_msg_get_opc(req->rq_reqmsg),
2191                                lustre_msg_get_jobid(req->rq_reqmsg) ?: "");
2192
2193                 spin_lock(&imp->imp_lock);
2194                 /*
2195                  * Request already may be not on sending or delaying list. This
2196                  * may happen in the case of marking it erroneous for the case
2197                  * ptlrpc_import_delay_req(req, status) find it impossible to
2198                  * allow sending this rpc and returns *status != 0.
2199                  */
2200                 if (!list_empty(&req->rq_list)) {
2201                         list_del_init(&req->rq_list);
2202                         if (atomic_dec_and_test(&imp->imp_inflight))
2203                                 wake_up(&imp->imp_recovery_waitq);
2204                 }
2205                 list_del_init(&req->rq_unreplied_list);
2206                 spin_unlock(&imp->imp_lock);
2207
2208                 atomic_dec(&set->set_remaining);
2209                 wake_up(&imp->imp_recovery_waitq);
2210
2211                 if (set->set_producer) {
2212                         /* produce a new request if possible */
2213                         if (ptlrpc_set_producer(set) > 0)
2214                                 force_timer_recalc = 1;
2215
2216                         /*
2217                          * free the request that has just been completed
2218                          * in order not to pollute set->set_requests
2219                          */
2220                         list_del_init(&req->rq_set_chain);
2221                         spin_lock(&req->rq_lock);
2222                         req->rq_set = NULL;
2223                         req->rq_invalid_rqset = 0;
2224                         spin_unlock(&req->rq_lock);
2225
2226                         /* record rq_status to compute the final status later */
2227                         if (req->rq_status != 0)
2228                                 set->set_rc = req->rq_status;
2229                         ptlrpc_req_finished(req);
2230                 } else {
2231                         list_move_tail(&req->rq_set_chain, &comp_reqs);
2232                 }
2233         }
2234
2235         /*
2236          * move completed request at the head of list so it's easier for
2237          * caller to find them
2238          */
2239         list_splice(&comp_reqs, &set->set_requests);
2240
2241         /* If we hit an error, we want to recover promptly. */
2242         RETURN(atomic_read(&set->set_remaining) == 0 || force_timer_recalc);
2243 }
2244 EXPORT_SYMBOL(ptlrpc_check_set);
2245
2246 /**
2247  * Time out request \a req. is \a async_unlink is set, that means do not wait
2248  * until LNet actually confirms network buffer unlinking.
2249  * Return 1 if we should give up further retrying attempts or 0 otherwise.
2250  */
2251 int ptlrpc_expire_one_request(struct ptlrpc_request *req, int async_unlink)
2252 {
2253         struct obd_import *imp = req->rq_import;
2254         unsigned int debug_mask = D_RPCTRACE;
2255         int rc = 0;
2256
2257         ENTRY;
2258         spin_lock(&req->rq_lock);
2259         req->rq_timedout = 1;
2260         spin_unlock(&req->rq_lock);
2261
2262         if (ptlrpc_console_allow(req, lustre_msg_get_opc(req->rq_reqmsg),
2263                                  lustre_msg_get_status(req->rq_reqmsg)))
2264                 debug_mask = D_WARNING;
2265         DEBUG_REQ(debug_mask, req, "Request sent has %s: [sent %lld/real %lld]",
2266                   req->rq_net_err ? "failed due to network error" :
2267                      ((req->rq_real_sent == 0 ||
2268                        req->rq_real_sent < req->rq_sent ||
2269                        req->rq_real_sent >= req->rq_deadline) ?
2270                       "timed out for sent delay" : "timed out for slow reply"),
2271                   req->rq_sent, req->rq_real_sent);
2272
2273         if (imp && obd_debug_peer_on_timeout)
2274                 LNetDebugPeer(imp->imp_connection->c_peer);
2275
2276         ptlrpc_unregister_reply(req, async_unlink);
2277         ptlrpc_unregister_bulk(req, async_unlink);
2278
2279         if (obd_dump_on_timeout)
2280                 libcfs_debug_dumplog();
2281
2282         if (!imp) {
2283                 DEBUG_REQ(D_HA, req, "NULL import: already cleaned up?");
2284                 RETURN(1);
2285         }
2286
2287         atomic_inc(&imp->imp_timeouts);
2288
2289         /* The DLM server doesn't want recovery run on its imports. */
2290         if (imp->imp_dlm_fake)
2291                 RETURN(1);
2292
2293         /*
2294          * If this request is for recovery or other primordial tasks,
2295          * then error it out here.
2296          */
2297         if (req->rq_ctx_init || req->rq_ctx_fini ||
2298             req->rq_send_state != LUSTRE_IMP_FULL ||
2299             imp->imp_obd->obd_no_recov) {
2300                 DEBUG_REQ(D_RPCTRACE, req, "err -110, sent_state=%s (now=%s)",
2301                           ptlrpc_import_state_name(req->rq_send_state),
2302                           ptlrpc_import_state_name(imp->imp_state));
2303                 spin_lock(&req->rq_lock);
2304                 req->rq_status = -ETIMEDOUT;
2305                 req->rq_err = 1;
2306                 spin_unlock(&req->rq_lock);
2307                 RETURN(1);
2308         }
2309
2310         /*
2311          * if a request can't be resent we can't wait for an answer after
2312          * the timeout
2313          */
2314         if (ptlrpc_no_resend(req)) {
2315                 DEBUG_REQ(D_RPCTRACE, req, "TIMEOUT-NORESEND:");
2316                 rc = 1;
2317         }
2318
2319         ptlrpc_fail_import(imp, lustre_msg_get_conn_cnt(req->rq_reqmsg));
2320
2321         RETURN(rc);
2322 }
2323
2324 /**
2325  * Time out all uncompleted requests in request set pointed by \a data
2326  * This is called when a wait times out.
2327  */
2328 void ptlrpc_expired_set(struct ptlrpc_request_set *set)
2329 {
2330         struct ptlrpc_request *req;
2331         time64_t now = ktime_get_real_seconds();
2332
2333         ENTRY;
2334         LASSERT(set != NULL);
2335
2336         /*
2337          * A timeout expired. See which reqs it applies to...
2338          */
2339         list_for_each_entry(req, &set->set_requests, rq_set_chain) {
2340                 /* don't expire request waiting for context */
2341                 if (req->rq_wait_ctx)
2342                         continue;
2343
2344                 /* Request in-flight? */
2345                 if (!((req->rq_phase == RQ_PHASE_RPC &&
2346                        !req->rq_waiting && !req->rq_resend) ||
2347                       (req->rq_phase == RQ_PHASE_BULK)))
2348                         continue;
2349
2350                 if (req->rq_timedout ||     /* already dealt with */
2351                     req->rq_deadline > now) /* not expired */
2352                         continue;
2353
2354                 /*
2355                  * Deal with this guy. Do it asynchronously to not block
2356                  * ptlrpcd thread.
2357                  */
2358                 ptlrpc_expire_one_request(req, 1);
2359                 /*
2360                  * Loops require that we resched once in a while to avoid
2361                  * RCU stalls and a few other problems.
2362                  */
2363                 cond_resched();
2364
2365         }
2366 }
2367
2368 /**
2369  * Interrupts (sets interrupted flag) all uncompleted requests in
2370  * a set \a data. This is called when a wait_event is interrupted
2371  * by a signal.
2372  */
2373 static void ptlrpc_interrupted_set(struct ptlrpc_request_set *set)
2374 {
2375         struct ptlrpc_request *req;
2376
2377         LASSERT(set != NULL);
2378         CDEBUG(D_RPCTRACE, "INTERRUPTED SET %p\n", set);
2379
2380         list_for_each_entry(req, &set->set_requests, rq_set_chain) {
2381                 if (req->rq_intr)
2382                         continue;
2383
2384                 if (req->rq_phase != RQ_PHASE_RPC &&
2385                     req->rq_phase != RQ_PHASE_UNREG_RPC &&
2386                     !req->rq_allow_intr)
2387                         continue;
2388
2389                 spin_lock(&req->rq_lock);
2390                 req->rq_intr = 1;
2391                 spin_unlock(&req->rq_lock);
2392         }
2393 }
2394
2395 /**
2396  * Get the smallest timeout in the set; this does NOT set a timeout.
2397  */
2398 time64_t ptlrpc_set_next_timeout(struct ptlrpc_request_set *set)
2399 {
2400         time64_t now = ktime_get_real_seconds();
2401         int timeout = 0;
2402         struct ptlrpc_request *req;
2403         time64_t deadline;
2404
2405         ENTRY;
2406         list_for_each_entry(req, &set->set_requests, rq_set_chain) {
2407                 /* Request in-flight? */
2408                 if (!(((req->rq_phase == RQ_PHASE_RPC) && !req->rq_waiting) ||
2409                       (req->rq_phase == RQ_PHASE_BULK) ||
2410                       (req->rq_phase == RQ_PHASE_NEW)))
2411                         continue;
2412
2413                 /* Already timed out. */
2414                 if (req->rq_timedout)
2415                         continue;
2416
2417                 /* Waiting for ctx. */
2418                 if (req->rq_wait_ctx)
2419                         continue;
2420
2421                 if (req->rq_phase == RQ_PHASE_NEW)
2422                         deadline = req->rq_sent;
2423                 else if (req->rq_phase == RQ_PHASE_RPC && req->rq_resend)
2424                         deadline = req->rq_sent;
2425                 else
2426                         deadline = req->rq_sent + req->rq_timeout;
2427
2428                 if (deadline <= now)    /* actually expired already */
2429                         timeout = 1;    /* ASAP */
2430                 else if (timeout == 0 || timeout > deadline - now)
2431                         timeout = deadline - now;
2432         }
2433         RETURN(timeout);
2434 }
2435
2436 /**
2437  * Send all unset request from the set and then wait untill all
2438  * requests in the set complete (either get a reply, timeout, get an
2439  * error or otherwise be interrupted).
2440  * Returns 0 on success or error code otherwise.
2441  */
2442 int ptlrpc_set_wait(const struct lu_env *env, struct ptlrpc_request_set *set)
2443 {
2444         struct ptlrpc_request *req;
2445         time64_t timeout;
2446         int rc;
2447
2448         ENTRY;
2449         if (set->set_producer)
2450                 (void)ptlrpc_set_producer(set);
2451         else
2452                 list_for_each_entry(req, &set->set_requests, rq_set_chain) {
2453                         if (req->rq_phase == RQ_PHASE_NEW)
2454                                 (void)ptlrpc_send_new_req(req);
2455                 }
2456
2457         if (list_empty(&set->set_requests))
2458                 RETURN(0);
2459
2460         do {
2461                 timeout = ptlrpc_set_next_timeout(set);
2462
2463                 /*
2464                  * wait until all complete, interrupted, or an in-flight
2465                  * req times out
2466                  */
2467                 CDEBUG(D_RPCTRACE, "set %p going to sleep for %lld seconds\n",
2468                        set, timeout);
2469
2470                 if ((timeout == 0 && !signal_pending(current)) ||
2471                     set->set_allow_intr) {
2472                         /*
2473                          * No requests are in-flight (ether timed out
2474                          * or delayed), so we can allow interrupts.
2475                          * We still want to block for a limited time,
2476                          * so we allow interrupts during the timeout.
2477                          */
2478                         rc = l_wait_event_abortable_timeout(
2479                                 set->set_waitq,
2480                                 ptlrpc_check_set(NULL, set),
2481                                 cfs_time_seconds(timeout ? timeout : 1));
2482                         if (rc == 0) {
2483                                 rc = -ETIMEDOUT;
2484                                 ptlrpc_expired_set(set);
2485                         } else if (rc < 0) {
2486                                 rc = -EINTR;
2487                                 ptlrpc_interrupted_set(set);
2488                         } else {
2489                                 rc = 0;
2490                         }
2491                 } else {
2492                         /*
2493                          * At least one request is in flight, so no
2494                          * interrupts are allowed. Wait until all
2495                          * complete, or an in-flight req times out.
2496                          */
2497                         rc = wait_event_idle_timeout(
2498                                 set->set_waitq,
2499                                 ptlrpc_check_set(NULL, set),
2500                                 cfs_time_seconds(timeout ? timeout : 1));
2501                         if (rc == 0) {
2502                                 ptlrpc_expired_set(set);
2503                                 rc = -ETIMEDOUT;
2504                         } else {
2505                                 rc = 0;
2506                         }
2507
2508                         /*
2509                          * LU-769 - if we ignored the signal because
2510                          * it was already pending when we started, we
2511                          * need to handle it now or we risk it being
2512                          * ignored forever
2513                          */
2514                         if (rc == -ETIMEDOUT &&
2515                             signal_pending(current)) {
2516                                 sigset_t old, new;
2517
2518                                 siginitset(&new, LUSTRE_FATAL_SIGS);
2519                                 sigprocmask(SIG_BLOCK, &new, &old);
2520                                 /*
2521                                  * In fact we only interrupt for the
2522                                  * "fatal" signals like SIGINT or
2523                                  * SIGKILL. We still ignore less
2524                                  * important signals since ptlrpc set
2525                                  * is not easily reentrant from
2526                                  * userspace again
2527                                  */
2528                                 if (signal_pending(current))
2529                                         ptlrpc_interrupted_set(set);
2530                                 sigprocmask(SIG_SETMASK, &old, NULL);
2531                         }
2532                 }
2533
2534                 LASSERT(rc == 0 || rc == -EINTR || rc == -ETIMEDOUT);
2535
2536                 /*
2537                  * -EINTR => all requests have been flagged rq_intr so next
2538                  * check completes.
2539                  * -ETIMEDOUT => someone timed out.  When all reqs have
2540                  * timed out, signals are enabled allowing completion with
2541                  * EINTR.
2542                  * I don't really care if we go once more round the loop in
2543                  * the error cases -eeb.
2544                  */
2545                 if (rc == 0 && atomic_read(&set->set_remaining) == 0) {
2546                         list_for_each_entry(req, &set->set_requests,
2547                                             rq_set_chain) {
2548                                 spin_lock(&req->rq_lock);
2549                                 req->rq_invalid_rqset = 1;
2550                                 spin_unlock(&req->rq_lock);
2551                         }
2552                 }
2553         } while (rc != 0 || atomic_read(&set->set_remaining) != 0);
2554
2555         LASSERT(atomic_read(&set->set_remaining) == 0);
2556
2557         rc = set->set_rc; /* rq_status of already freed requests if any */
2558         list_for_each_entry(req, &set->set_requests, rq_set_chain) {
2559                 LASSERT(req->rq_phase == RQ_PHASE_COMPLETE);
2560                 if (req->rq_status != 0)
2561                         rc = req->rq_status;
2562         }
2563
2564         RETURN(rc);
2565 }
2566 EXPORT_SYMBOL(ptlrpc_set_wait);
2567
2568 /**
2569  * Helper fuction for request freeing.
2570  * Called when request count reached zero and request needs to be freed.
2571  * Removes request from all sorts of sending/replay lists it might be on,
2572  * frees network buffers if any are present.
2573  * If \a locked is set, that means caller is already holding import imp_lock
2574  * and so we no longer need to reobtain it (for certain lists manipulations)
2575  */
2576 static void __ptlrpc_free_req(struct ptlrpc_request *request, int locked)
2577 {
2578         ENTRY;
2579
2580         if (!request)
2581                 RETURN_EXIT;
2582
2583         LASSERT(!request->rq_srv_req);
2584         LASSERT(request->rq_export == NULL);
2585         LASSERTF(!request->rq_receiving_reply, "req %p\n", request);
2586         LASSERTF(list_empty(&request->rq_list), "req %p\n", request);
2587         LASSERTF(list_empty(&request->rq_set_chain), "req %p\n", request);
2588         LASSERTF(!request->rq_replay, "req %p\n", request);
2589
2590         req_capsule_fini(&request->rq_pill);
2591
2592         /*
2593          * We must take it off the imp_replay_list first.  Otherwise, we'll set
2594          * request->rq_reqmsg to NULL while osc_close is dereferencing it.
2595          */
2596         if (request->rq_import) {
2597                 if (!locked)
2598                         spin_lock(&request->rq_import->imp_lock);
2599                 list_del_init(&request->rq_replay_list);
2600                 list_del_init(&request->rq_unreplied_list);
2601                 if (!locked)
2602                         spin_unlock(&request->rq_import->imp_lock);
2603         }
2604         LASSERTF(list_empty(&request->rq_replay_list), "req %p\n", request);
2605
2606         if (atomic_read(&request->rq_refcount) != 0) {
2607                 DEBUG_REQ(D_ERROR, request,
2608                           "freeing request with nonzero refcount");
2609                 LBUG();
2610         }
2611
2612         if (request->rq_repbuf)
2613                 sptlrpc_cli_free_repbuf(request);
2614
2615         if (request->rq_import) {
2616                 if (!ptlrpcd_check_work(request)) {
2617                         LASSERT(atomic_read(&request->rq_import->imp_reqs) > 0);
2618                         atomic_dec(&request->rq_import->imp_reqs);
2619                 }
2620                 class_import_put(request->rq_import);
2621                 request->rq_import = NULL;
2622         }
2623         if (request->rq_bulk)
2624                 ptlrpc_free_bulk(request->rq_bulk);
2625
2626         if (request->rq_reqbuf || request->rq_clrbuf)
2627                 sptlrpc_cli_free_reqbuf(request);
2628
2629         if (request->rq_cli_ctx)
2630                 sptlrpc_req_put_ctx(request, !locked);
2631
2632         if (request->rq_pool)
2633                 __ptlrpc_free_req_to_pool(request);
2634         else
2635                 ptlrpc_request_cache_free(request);
2636         EXIT;
2637 }
2638
2639 static int __ptlrpc_req_finished(struct ptlrpc_request *request, int locked);
2640 /**
2641  * Drop one request reference. Must be called with import imp_lock held.
2642  * When reference count drops to zero, request is freed.
2643  */
2644 void ptlrpc_req_finished_with_imp_lock(struct ptlrpc_request *request)
2645 {
2646         assert_spin_locked(&request->rq_import->imp_lock);
2647         (void)__ptlrpc_req_finished(request, 1);
2648 }
2649
2650 /**
2651  * Helper function
2652  * Drops one reference count for request \a request.
2653  * \a locked set indicates that caller holds import imp_lock.
2654  * Frees the request whe reference count reaches zero.
2655  *
2656  * \retval 1    the request is freed
2657  * \retval 0    some others still hold references on the request
2658  */
2659 static int __ptlrpc_req_finished(struct ptlrpc_request *request, int locked)
2660 {
2661         int count;
2662
2663         ENTRY;
2664         if (!request)
2665                 RETURN(1);
2666
2667         LASSERT(request != LP_POISON);
2668         LASSERT(request->rq_reqmsg != LP_POISON);
2669
2670         DEBUG_REQ(D_INFO, request, "refcount now %u",
2671                   atomic_read(&request->rq_refcount) - 1);
2672
2673         spin_lock(&request->rq_lock);
2674         count = atomic_dec_return(&request->rq_refcount);
2675         LASSERTF(count >= 0, "Invalid ref count %d\n", count);
2676
2677         /*
2678          * For open RPC, the client does not know the EA size (LOV, ACL, and
2679          * so on) before replied, then the client has to reserve very large
2680          * reply buffer. Such buffer will not be released until the RPC freed.
2681          * Since The open RPC is replayable, we need to keep it in the replay
2682          * list until close. If there are a lot of files opened concurrently,
2683          * then the client may be OOM.
2684          *
2685          * If fact, it is unnecessary to keep reply buffer for open replay,
2686          * related EAs have already been saved via mdc_save_lovea() before
2687          * coming here. So it is safe to free the reply buffer some earlier
2688          * before releasing the RPC to avoid client OOM. LU-9514
2689          */
2690         if (count == 1 && request->rq_early_free_repbuf && request->rq_repbuf) {
2691                 spin_lock(&request->rq_early_free_lock);
2692                 sptlrpc_cli_free_repbuf(request);
2693                 request->rq_repbuf = NULL;
2694                 request->rq_repbuf_len = 0;
2695                 request->rq_repdata = NULL;
2696                 request->rq_reqdata_len = 0;
2697                 spin_unlock(&request->rq_early_free_lock);
2698         }
2699         spin_unlock(&request->rq_lock);
2700
2701         if (!count)
2702                 __ptlrpc_free_req(request, locked);
2703
2704         RETURN(!count);
2705 }
2706
2707 /**
2708  * Drops one reference count for a request.
2709  */
2710 void ptlrpc_req_finished(struct ptlrpc_request *request)
2711 {
2712         __ptlrpc_req_finished(request, 0);
2713 }
2714 EXPORT_SYMBOL(ptlrpc_req_finished);
2715
2716 /**
2717  * Returns xid of a \a request
2718  */
2719 __u64 ptlrpc_req_xid(struct ptlrpc_request *request)
2720 {
2721         return request->rq_xid;
2722 }
2723 EXPORT_SYMBOL(ptlrpc_req_xid);
2724
2725 /**
2726  * Disengage the client's reply buffer from the network
2727  * NB does _NOT_ unregister any client-side bulk.
2728  * IDEMPOTENT, but _not_ safe against concurrent callers.
2729  * The request owner (i.e. the thread doing the I/O) must call...
2730  * Returns 0 on success or 1 if unregistering cannot be made.
2731  */
2732 static int ptlrpc_unregister_reply(struct ptlrpc_request *request, int async)
2733 {
2734         bool discard = false;
2735         /*
2736          * Might sleep.
2737          */
2738         LASSERT(!in_interrupt());
2739
2740         /* Let's setup deadline for reply unlink. */
2741         if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_REPL_UNLINK) &&
2742             async && request->rq_reply_deadline == 0 && cfs_fail_val == 0)
2743                 request->rq_reply_deadline = ktime_get_real_seconds() +
2744                                              PTLRPC_REQ_LONG_UNLINK;
2745
2746         /*
2747          * Nothing left to do.
2748          */
2749         if (!__ptlrpc_cli_wait_unlink(request, &discard))
2750                 RETURN(1);
2751
2752         LNetMDUnlink(request->rq_reply_md_h);
2753
2754         if (discard) /* Discard the request-out callback */
2755                 __LNetMDUnlink(request->rq_req_md_h, discard);
2756
2757         /*
2758          * Let's check it once again.
2759          */
2760         if (!ptlrpc_cli_wait_unlink(request))
2761                 RETURN(1);
2762
2763         /* Move to "Unregistering" phase as reply was not unlinked yet. */
2764         ptlrpc_rqphase_move(request, RQ_PHASE_UNREG_RPC);
2765
2766         /*
2767          * Do not wait for unlink to finish.
2768          */
2769         if (async)
2770                 RETURN(0);
2771
2772         /*
2773          * We have to wait_event_idle_timeout() whatever the result, to get
2774          * a chance to run reply_in_callback(), and to make sure we've
2775          * unlinked before returning a req to the pool.
2776          */
2777         for (;;) {
2778                 wait_queue_head_t *wq = (request->rq_set) ?
2779                                         &request->rq_set->set_waitq :
2780                                         &request->rq_reply_waitq;
2781                 int seconds = PTLRPC_REQ_LONG_UNLINK;
2782                 /*
2783                  * Network access will complete in finite time but the HUGE
2784                  * timeout lets us CWARN for visibility of sluggish NALs
2785                  */
2786                 while (seconds > 0 &&
2787                        wait_event_idle_timeout(
2788                                *wq,
2789                                !ptlrpc_cli_wait_unlink(request),
2790                                cfs_time_seconds(1)) == 0)
2791                         seconds -= 1;
2792                 if (seconds > 0) {
2793                         ptlrpc_rqphase_move(request, request->rq_next_phase);
2794                         RETURN(1);
2795                 }
2796
2797                 DEBUG_REQ(D_WARNING, request,
2798                           "Unexpectedly long timeout receiving_reply=%d req_ulinked=%d reply_unlinked=%d",
2799                           request->rq_receiving_reply,
2800                           request->rq_req_unlinked,
2801                           request->rq_reply_unlinked);
2802         }
2803         RETURN(0);
2804 }
2805
2806 static void ptlrpc_free_request(struct ptlrpc_request *req)
2807 {
2808         spin_lock(&req->rq_lock);
2809         req->rq_replay = 0;
2810         spin_unlock(&req->rq_lock);
2811
2812         if (req->rq_commit_cb)
2813                 req->rq_commit_cb(req);
2814         list_del_init(&req->rq_replay_list);
2815
2816         __ptlrpc_req_finished(req, 1);
2817 }
2818
2819 /**
2820  * the request is committed and dropped from the replay list of its import
2821  */
2822 void ptlrpc_request_committed(struct ptlrpc_request *req, int force)
2823 {
2824         struct obd_import *imp = req->rq_import;
2825
2826         spin_lock(&imp->imp_lock);
2827         if (list_empty(&req->rq_replay_list)) {
2828                 spin_unlock(&imp->imp_lock);
2829                 return;
2830         }
2831
2832         if (force || req->rq_transno <= imp->imp_peer_committed_transno) {
2833                 if (imp->imp_replay_cursor == &req->rq_replay_list)
2834                         imp->imp_replay_cursor = req->rq_replay_list.next;
2835                 ptlrpc_free_request(req);
2836         }
2837
2838         spin_unlock(&imp->imp_lock);
2839 }
2840 EXPORT_SYMBOL(ptlrpc_request_committed);
2841
2842 /**
2843  * Iterates through replay_list on import and prunes
2844  * all requests have transno smaller than last_committed for the
2845  * import and don't have rq_replay set.
2846  * Since requests are sorted in transno order, stops when meetign first
2847  * transno bigger than last_committed.
2848  * caller must hold imp->imp_lock
2849  */
2850 void ptlrpc_free_committed(struct obd_import *imp)
2851 {
2852         struct ptlrpc_request *req, *saved;
2853         struct ptlrpc_request *last_req = NULL; /* temporary fire escape */
2854         bool skip_committed_list = true;
2855
2856         ENTRY;
2857         LASSERT(imp != NULL);
2858         assert_spin_locked(&imp->imp_lock);
2859
2860         if (imp->imp_peer_committed_transno == imp->imp_last_transno_checked &&
2861             imp->imp_generation == imp->imp_last_generation_checked) {
2862                 CDEBUG(D_INFO, "%s: skip recheck: last_committed %llu\n",
2863                        imp->imp_obd->obd_name, imp->imp_peer_committed_transno);
2864                 RETURN_EXIT;
2865         }
2866         CDEBUG(D_RPCTRACE, "%s: committing for last_committed %llu gen %d\n",
2867                imp->imp_obd->obd_name, imp->imp_peer_committed_transno,
2868                imp->imp_generation);
2869
2870         if (imp->imp_generation != imp->imp_last_generation_checked ||
2871             imp->imp_last_transno_checked == 0)
2872                 skip_committed_list = false;
2873
2874         imp->imp_last_transno_checked = imp->imp_peer_committed_transno;
2875         imp->imp_last_generation_checked = imp->imp_generation;
2876
2877         list_for_each_entry_safe(req, saved, &imp->imp_replay_list,
2878                                  rq_replay_list) {
2879                 /* XXX ok to remove when 1357 resolved - rread 05/29/03  */
2880                 LASSERT(req != last_req);
2881                 last_req = req;
2882
2883                 if (req->rq_transno == 0) {
2884                         DEBUG_REQ(D_EMERG, req, "zero transno during replay");
2885                         LBUG();
2886                 }
2887                 if (req->rq_import_generation < imp->imp_generation) {
2888                         DEBUG_REQ(D_RPCTRACE, req, "free request with old gen");
2889                         GOTO(free_req, 0);
2890                 }
2891
2892                 /* not yet committed */
2893                 if (req->rq_transno > imp->imp_peer_committed_transno) {
2894                         DEBUG_REQ(D_RPCTRACE, req, "stopping search");
2895                         break;
2896                 }
2897
2898                 if (req->rq_replay) {
2899                         DEBUG_REQ(D_RPCTRACE, req, "keeping (FL_REPLAY)");
2900                         list_move_tail(&req->rq_replay_list,
2901                                        &imp->imp_committed_list);
2902                         continue;
2903                 }
2904
2905                 DEBUG_REQ(D_INFO, req, "commit (last_committed %llu)",
2906                           imp->imp_peer_committed_transno);
2907 free_req:
2908                 ptlrpc_free_request(req);
2909         }
2910
2911         if (skip_committed_list)
2912                 GOTO(out, 0);
2913
2914         list_for_each_entry_safe(req, saved, &imp->imp_committed_list,
2915                                  rq_replay_list) {
2916                 LASSERT(req->rq_transno != 0);
2917                 if (req->rq_import_generation < imp->imp_generation ||
2918                     !req->rq_replay) {
2919                         DEBUG_REQ(D_RPCTRACE, req, "free %s open request",
2920                                   req->rq_import_generation <
2921                                   imp->imp_generation ? "stale" : "closed");
2922
2923                         if (imp->imp_replay_cursor == &req->rq_replay_list)
2924                                 imp->imp_replay_cursor =
2925                                         req->rq_replay_list.next;
2926
2927                         ptlrpc_free_request(req);
2928                 }
2929         }
2930 out:
2931         EXIT;
2932 }
2933
2934 void ptlrpc_cleanup_client(struct obd_import *imp)
2935 {
2936         ENTRY;
2937         EXIT;
2938 }
2939
2940 /**
2941  * Schedule previously sent request for resend.
2942  * For bulk requests we assign new xid (to avoid problems with
2943  * lost replies and therefore several transfers landing into same buffer
2944  * from different sending attempts).
2945  */
2946 void ptlrpc_resend_req(struct ptlrpc_request *req)
2947 {
2948         DEBUG_REQ(D_HA, req, "going to resend");
2949         spin_lock(&req->rq_lock);
2950
2951         /*
2952          * Request got reply but linked to the import list still.
2953          * Let ptlrpc_check_set() process it.
2954          */
2955         if (ptlrpc_client_replied(req)) {
2956                 spin_unlock(&req->rq_lock);
2957                 DEBUG_REQ(D_HA, req, "it has reply, so skip it");
2958                 return;
2959         }
2960
2961         req->rq_status = -EAGAIN;
2962
2963         req->rq_resend = 1;
2964         req->rq_net_err = 0;
2965         req->rq_timedout = 0;
2966
2967         ptlrpc_client_wake_req(req);
2968         spin_unlock(&req->rq_lock);
2969 }
2970
2971 /* XXX: this function and rq_status are currently unused */
2972 void ptlrpc_restart_req(struct ptlrpc_request *req)
2973 {
2974         DEBUG_REQ(D_HA, req, "restarting (possibly-)completed request");
2975         req->rq_status = -ERESTARTSYS;
2976
2977         spin_lock(&req->rq_lock);
2978         req->rq_restart = 1;
2979         req->rq_timedout = 0;
2980         ptlrpc_client_wake_req(req);
2981         spin_unlock(&req->rq_lock);
2982 }
2983
2984 /**
2985  * Grab additional reference on a request \a req
2986  */
2987 struct ptlrpc_request *ptlrpc_request_addref(struct ptlrpc_request *req)
2988 {
2989         ENTRY;
2990         atomic_inc(&req->rq_refcount);
2991         RETURN(req);
2992 }
2993 EXPORT_SYMBOL(ptlrpc_request_addref);
2994
2995 /**
2996  * Add a request to import replay_list.
2997  * Must be called under imp_lock
2998  */
2999 void ptlrpc_retain_replayable_request(struct ptlrpc_request *req,
3000                                       struct obd_import *imp)
3001 {
3002         struct ptlrpc_request *iter;
3003
3004         assert_spin_locked(&imp->imp_lock);
3005
3006         if (req->rq_transno == 0) {
3007                 DEBUG_REQ(D_EMERG, req, "saving request with zero transno");
3008                 LBUG();
3009         }
3010
3011         /*
3012          * clear this for new requests that were resent as well
3013          * as resent replayed requests.
3014          */
3015         lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT);
3016
3017         /* don't re-add requests that have been replayed */
3018         if (!list_empty(&req->rq_replay_list))
3019                 return;
3020
3021         lustre_msg_add_flags(req->rq_reqmsg, MSG_REPLAY);
3022
3023         spin_lock(&req->rq_lock);
3024         req->rq_resend = 0;
3025         spin_unlock(&req->rq_lock);
3026
3027         LASSERT(imp->imp_replayable);
3028         /* Balanced in ptlrpc_free_committed, usually. */
3029         ptlrpc_request_addref(req);
3030         list_for_each_entry_reverse(iter, &imp->imp_replay_list,
3031                                     rq_replay_list) {
3032                 /*
3033                  * We may have duplicate transnos if we create and then
3034                  * open a file, or for closes retained if to match creating
3035                  * opens, so use req->rq_xid as a secondary key.
3036                  * (See bugs 684, 685, and 428.)
3037                  * XXX no longer needed, but all opens need transnos!
3038                  */
3039                 if (iter->rq_transno > req->rq_transno)
3040                         continue;
3041
3042                 if (iter->rq_transno == req->rq_transno) {
3043                         LASSERT(iter->rq_xid != req->rq_xid);
3044                         if (iter->rq_xid > req->rq_xid)
3045                                 continue;
3046                 }
3047
3048                 list_add(&req->rq_replay_list, &iter->rq_replay_list);
3049                 return;
3050         }
3051
3052         list_add(&req->rq_replay_list, &imp->imp_replay_list);
3053 }
3054
3055 /**
3056  * Send request and wait until it completes.
3057  * Returns request processing status.
3058  */
3059 int ptlrpc_queue_wait(struct ptlrpc_request *req)
3060 {
3061         struct ptlrpc_request_set *set;
3062         int rc;
3063
3064         ENTRY;
3065         LASSERT(req->rq_set == NULL);
3066         LASSERT(!req->rq_receiving_reply);
3067
3068         set = ptlrpc_prep_set();
3069         if (!set) {
3070                 CERROR("cannot allocate ptlrpc set: rc = %d\n", -ENOMEM);
3071                 RETURN(-ENOMEM);
3072         }
3073
3074         /* for distributed debugging */
3075         lustre_msg_set_status(req->rq_reqmsg, current->pid);
3076
3077         /* add a ref for the set (see comment in ptlrpc_set_add_req) */
3078         ptlrpc_request_addref(req);
3079         ptlrpc_set_add_req(set, req);
3080         rc = ptlrpc_set_wait(NULL, set);
3081         ptlrpc_set_destroy(set);
3082
3083         RETURN(rc);
3084 }
3085 EXPORT_SYMBOL(ptlrpc_queue_wait);
3086
3087 /**
3088  * Callback used for replayed requests reply processing.
3089  * In case of successful reply calls registered request replay callback.
3090  * In case of error restart replay process.
3091  */
3092 static int ptlrpc_replay_interpret(const struct lu_env *env,
3093                                    struct ptlrpc_request *req,
3094                                    void *args, int rc)
3095 {
3096         struct ptlrpc_replay_async_args *aa = args;
3097         struct obd_import *imp = req->rq_import;
3098
3099         ENTRY;
3100         atomic_dec(&imp->imp_replay_inflight);
3101
3102         /*
3103          * Note: if it is bulk replay (MDS-MDS replay), then even if
3104          * server got the request, but bulk transfer timeout, let's
3105          * replay the bulk req again
3106          */
3107         if (!ptlrpc_client_replied(req) ||
3108             (req->rq_bulk &&
3109              lustre_msg_get_status(req->rq_repmsg) == -ETIMEDOUT)) {
3110                 DEBUG_REQ(D_ERROR, req, "request replay timed out");
3111                 GOTO(out, rc = -ETIMEDOUT);
3112         }
3113
3114         if (lustre_msg_get_type(req->rq_repmsg) == PTL_RPC_MSG_ERR &&
3115             (lustre_msg_get_status(req->rq_repmsg) == -ENOTCONN ||
3116             lustre_msg_get_status(req->rq_repmsg) == -ENODEV))
3117                 GOTO(out, rc = lustre_msg_get_status(req->rq_repmsg));
3118
3119         /** VBR: check version failure */
3120         if (lustre_msg_get_status(req->rq_repmsg) == -EOVERFLOW) {
3121                 /** replay was failed due to version mismatch */
3122                 DEBUG_REQ(D_WARNING, req, "Version mismatch during replay");
3123                 spin_lock(&imp->imp_lock);
3124                 imp->imp_vbr_failed = 1;
3125                 spin_unlock(&imp->imp_lock);
3126                 lustre_msg_set_status(req->rq_repmsg, aa->praa_old_status);
3127         } else {
3128                 /** The transno had better not change over replay. */
3129                 LASSERTF(lustre_msg_get_transno(req->rq_reqmsg) ==
3130                          lustre_msg_get_transno(req->rq_repmsg) ||
3131                          lustre_msg_get_transno(req->rq_repmsg) == 0,
3132                          "%#llx/%#llx\n",
3133                          lustre_msg_get_transno(req->rq_reqmsg),
3134                          lustre_msg_get_transno(req->rq_repmsg));
3135         }
3136
3137         spin_lock(&imp->imp_lock);
3138         imp->imp_last_replay_transno = lustre_msg_get_transno(req->rq_reqmsg);
3139         spin_unlock(&imp->imp_lock);
3140         LASSERT(imp->imp_last_replay_transno);
3141
3142         /* transaction number shouldn't be bigger than the latest replayed */
3143         if (req->rq_transno > lustre_msg_get_transno(req->rq_reqmsg)) {
3144                 DEBUG_REQ(D_ERROR, req,
3145                           "Reported transno=%llu is bigger than replayed=%llu",
3146                           req->rq_transno,
3147                           lustre_msg_get_transno(req->rq_reqmsg));
3148                 GOTO(out, rc = -EINVAL);
3149         }
3150
3151         DEBUG_REQ(D_HA, req, "got reply");
3152
3153         /* let the callback do fixups, possibly including in the request */
3154         if (req->rq_replay_cb)
3155                 req->rq_replay_cb(req);
3156
3157         if (ptlrpc_client_replied(req) &&
3158             lustre_msg_get_status(req->rq_repmsg) != aa->praa_old_status) {
3159                 DEBUG_REQ(D_ERROR, req, "status %d, old was %d",
3160                           lustre_msg_get_status(req->rq_repmsg),
3161                           aa->praa_old_status);
3162
3163                 /*
3164                  * Note: If the replay fails for MDT-MDT recovery, let's
3165                  * abort all of the following requests in the replay
3166                  * and sending list, because MDT-MDT update requests
3167                  * are dependent on each other, see LU-7039
3168                  */
3169                 if (imp->imp_connect_flags_orig & OBD_CONNECT_MDS_MDS) {
3170                         struct ptlrpc_request *free_req;
3171                         struct ptlrpc_request *tmp;
3172
3173                         spin_lock(&imp->imp_lock);
3174                         list_for_each_entry_safe(free_req, tmp,
3175                                                  &imp->imp_replay_list,
3176                                                  rq_replay_list) {
3177                                 ptlrpc_free_request(free_req);
3178                         }
3179
3180                         list_for_each_entry_safe(free_req, tmp,
3181                                                  &imp->imp_committed_list,
3182                                                  rq_replay_list) {
3183                                 ptlrpc_free_request(free_req);
3184                         }
3185
3186                         list_for_each_entry_safe(free_req, tmp,
3187                                                  &imp->imp_delayed_list,
3188                                                  rq_list) {
3189                                 spin_lock(&free_req->rq_lock);
3190                                 free_req->rq_err = 1;
3191                                 free_req->rq_status = -EIO;
3192                                 ptlrpc_client_wake_req(free_req);
3193                                 spin_unlock(&free_req->rq_lock);
3194                         }
3195
3196                         list_for_each_entry_safe(free_req, tmp,