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