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