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