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