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