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