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