lustre/ptlrpc/client.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (c) 2002, 2003 Cluster File Systems, Inc.
   5  *
   6  *   This file is part of the Lustre file system, http://www.lustre.org
   7  *   Lustre is a trademark of Cluster File Systems, Inc.
   8  *
   9  *   You may have signed or agreed to another license before downloading
  10  *   this software.  If so, you are bound by the terms and conditions
  11  *   of that agreement, and the following does not apply to you.  See the
  12  *   LICENSE file included with this distribution for more information.
  13  *
  14  *   If you did not agree to a different license, then this copy of Lustre
  15  *   is open source software; you can redistribute it and/or modify it
  16  *   under the terms of version 2 of the GNU General Public License as
  17  *   published by the Free Software Foundation.
  18  *
  19  *   In either case, Lustre is distributed in the hope that it will be
  20  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  21  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  22  *   license text for more details.
  23  *
  24  */
  25
  26 #define DEBUG_SUBSYSTEM S_RPC
  27 #ifndef __KERNEL__
  28 #include <errno.h>
  29 #include <signal.h>
  30 #include <liblustre.h>
  31 #endif
  32
  33 #include <obd_support.h>
  34 #include <obd_class.h>
  35 #include <lustre_lib.h>
  36 #include <lustre_ha.h>
  37 #include <lustre_import.h>
  38 #include <lustre_req_layout.h>
  39
  40 #include "ptlrpc_internal.h"
  41
  42 void ptlrpc_init_client(int req_portal, int rep_portal, char *name,
  43                         struct ptlrpc_client *cl)
  44 {
  45         cl->cli_request_portal = req_portal;
  46         cl->cli_reply_portal   = rep_portal;
  47         cl->cli_name           = name;
  48 }
  49
  50 struct ptlrpc_connection *ptlrpc_uuid_to_connection(struct obd_uuid *uuid)
  51 {
  52         struct ptlrpc_connection *c;
  53         lnet_nid_t                self;
  54         lnet_process_id_t         peer;
  55         int                       err;
  56
  57         err = ptlrpc_uuid_to_peer(uuid, &peer, &self);
  58         if (err != 0) {
  59                 CERROR("cannot find peer %s!\n", uuid->uuid);
  60                 return NULL;
  61         }
  62
  63         c = ptlrpc_get_connection(peer, self, uuid);
  64         if (c) {
  65                 memcpy(c->c_remote_uuid.uuid,
  66                        uuid->uuid, sizeof(c->c_remote_uuid.uuid));
  67         }
  68
  69         CDEBUG(D_INFO, "%s -> %p\n", uuid->uuid, c);
  70
  71         return c;
  72 }
  73
  74 void ptlrpc_readdress_connection(struct ptlrpc_connection *conn,
  75                                  struct obd_uuid *uuid)
  76 {
  77         lnet_nid_t        self;
  78         lnet_process_id_t peer;
  79         int               err;
  80
  81         err = ptlrpc_uuid_to_peer(uuid, &peer, &self);
  82         if (err != 0) {
  83                 CERROR("cannot find peer %s!\n", uuid->uuid);
  84                 return;
  85         }
  86
  87         conn->c_peer = peer;
  88         conn->c_self = self;
  89         return;
  90 }
  91
  92 static inline struct ptlrpc_bulk_desc *new_bulk(int npages, int type, int portal)
  93 {
  94         struct ptlrpc_bulk_desc *desc;
  95
  96         OBD_ALLOC(desc, offsetof (struct ptlrpc_bulk_desc, bd_iov[npages]));
  97         if (!desc)
  98                 return NULL;
  99
 100         spin_lock_init(&desc->bd_lock);
 101         cfs_waitq_init(&desc->bd_waitq);
 102         desc->bd_max_iov = npages;
 103         desc->bd_iov_count = 0;
 104         desc->bd_md_h = LNET_INVALID_HANDLE;
 105         desc->bd_portal = portal;
 106         desc->bd_type = type;
 107
 108         return desc;
 109 }
 110
 111 struct ptlrpc_bulk_desc *ptlrpc_prep_bulk_imp (struct ptlrpc_request *req,
 112                                                int npages, int type, int portal)
 113 {
 114         struct obd_import *imp = req->rq_import;
 115         struct ptlrpc_bulk_desc *desc;
 116
 117         ENTRY;
 118         LASSERT(type == BULK_PUT_SINK || type == BULK_GET_SOURCE);
 119         desc = new_bulk(npages, type, portal);
 120         if (desc == NULL)
 121                 RETURN(NULL);
 122
 123         desc->bd_import_generation = req->rq_import_generation;
 124         desc->bd_import = class_import_get(imp);
 125         desc->bd_req = req;
 126
 127         desc->bd_cbid.cbid_fn  = client_bulk_callback;
 128         desc->bd_cbid.cbid_arg = desc;
 129
 130         /* This makes req own desc, and free it when she frees herself */
 131         req->rq_bulk = desc;
 132
 133         return desc;
 134 }
 135
 136 struct ptlrpc_bulk_desc *ptlrpc_prep_bulk_exp(struct ptlrpc_request *req,
 137                                               int npages, int type, int portal)
 138 {
 139         struct obd_export *exp = req->rq_export;
 140         struct ptlrpc_bulk_desc *desc;
 141
 142         ENTRY;
 143         LASSERT(type == BULK_PUT_SOURCE || type == BULK_GET_SINK);
 144
 145         desc = new_bulk(npages, type, portal);
 146         if (desc == NULL)
 147                 RETURN(NULL);
 148
 149         desc->bd_export = class_export_get(exp);
 150         desc->bd_req = req;
 151
 152         desc->bd_cbid.cbid_fn  = server_bulk_callback;
 153         desc->bd_cbid.cbid_arg = desc;
 154
 155         /* NB we don't assign rq_bulk here; server-side requests are
 156          * re-used, and the handler frees the bulk desc explicitly. */
 157
 158         return desc;
 159 }
 160
 161 void ptlrpc_prep_bulk_page(struct ptlrpc_bulk_desc *desc,
 162                            cfs_page_t *page, int pageoffset, int len)
 163 {
 164         LASSERT(desc->bd_iov_count < desc->bd_max_iov);
 165         LASSERT(page != NULL);
 166         LASSERT(pageoffset >= 0);
 167         LASSERT(len > 0);
 168         LASSERT(pageoffset + len <= CFS_PAGE_SIZE);
 169
 170         desc->bd_nob += len;
 171
 172         ptlrpc_add_bulk_page(desc, page, pageoffset, len);
 173 }
 174
 175 void ptlrpc_free_bulk(struct ptlrpc_bulk_desc *desc)
 176 {
 177         ENTRY;
 178
 179         LASSERT(desc != NULL);
 180         LASSERT(desc->bd_iov_count != LI_POISON); /* not freed already */
 181         LASSERT(!desc->bd_network_rw);         /* network hands off or */
 182         LASSERT((desc->bd_export != NULL) ^ (desc->bd_import != NULL));
 183
 184         sptlrpc_enc_pool_put_pages(desc);
 185
 186         if (desc->bd_export)
 187                 class_export_put(desc->bd_export);
 188         else
 189                 class_import_put(desc->bd_import);
 190
 191         OBD_FREE(desc, offsetof(struct ptlrpc_bulk_desc,
 192                                 bd_iov[desc->bd_max_iov]));
 193         EXIT;
 194 }
 195
 196 void ptlrpc_free_rq_pool(struct ptlrpc_request_pool *pool)
 197 {
 198         struct list_head *l, *tmp;
 199         struct ptlrpc_request *req;
 200
 201         if (!pool)
 202                 return;
 203
 204         list_for_each_safe(l, tmp, &pool->prp_req_list) {
 205                 req = list_entry(l, struct ptlrpc_request, rq_list);
 206                 list_del(&req->rq_list);
 207                 LASSERT(req->rq_reqbuf);
 208                 LASSERT(req->rq_reqbuf_len == pool->prp_rq_size);
 209                 OBD_FREE(req->rq_reqbuf, pool->prp_rq_size);
 210                 OBD_FREE(req, sizeof(*req));
 211         }
 212         OBD_FREE(pool, sizeof(*pool));
 213 }
 214
 215 void ptlrpc_add_rqs_to_pool(struct ptlrpc_request_pool *pool, int num_rq)
 216 {
 217         int i;
 218         int size = 1;
 219
 220         while (size < pool->prp_rq_size + SPTLRPC_MAX_PAYLOAD)
 221                 size <<= 1;
 222
 223         LASSERTF(list_empty(&pool->prp_req_list) || size == pool->prp_rq_size,
 224                  "Trying to change pool size with nonempty pool "
 225                  "from %d to %d bytes\n", pool->prp_rq_size, size);
 226
 227         spin_lock(&pool->prp_lock);
 228         pool->prp_rq_size = size;
 229         for (i = 0; i < num_rq; i++) {
 230                 struct ptlrpc_request *req;
 231                 struct lustre_msg *msg;
 232
 233                 spin_unlock(&pool->prp_lock);
 234                 OBD_ALLOC(req, sizeof(struct ptlrpc_request));
 235                 if (!req)
 236                         return;
 237                 OBD_ALLOC_GFP(msg, size, CFS_ALLOC_STD);
 238                 if (!msg) {
 239                         OBD_FREE(req, sizeof(struct ptlrpc_request));
 240                         return;
 241                 }
 242                 req->rq_reqbuf = msg;
 243                 req->rq_reqbuf_len = size;
 244                 req->rq_pool = pool;
 245                 spin_lock(&pool->prp_lock);
 246                 list_add_tail(&req->rq_list, &pool->prp_req_list);
 247         }
 248         spin_unlock(&pool->prp_lock);
 249         return;
 250 }
 251
 252 struct ptlrpc_request_pool *ptlrpc_init_rq_pool(int num_rq, int msgsize,
 253                                                 void (*populate_pool)(struct ptlrpc_request_pool *, int))
 254 {
 255         struct ptlrpc_request_pool *pool;
 256
 257         OBD_ALLOC(pool, sizeof (struct ptlrpc_request_pool));
 258         if (!pool)
 259                 return NULL;
 260
 261         /* Request next power of two for the allocation, because internally
 262            kernel would do exactly this */
 263
 264         spin_lock_init(&pool->prp_lock);
 265         CFS_INIT_LIST_HEAD(&pool->prp_req_list);
 266         pool->prp_rq_size = msgsize;
 267         pool->prp_populate = populate_pool;
 268
 269         populate_pool(pool, num_rq);
 270
 271         if (list_empty(&pool->prp_req_list)) {
 272                 /* have not allocated a single request for the pool */
 273                 OBD_FREE(pool, sizeof (struct ptlrpc_request_pool));
 274                 pool = NULL;
 275         }
 276         return pool;
 277 }
 278
 279 static struct ptlrpc_request *ptlrpc_prep_req_from_pool(struct ptlrpc_request_pool *pool)
 280 {
 281         struct ptlrpc_request *request;
 282         struct lustre_msg *reqbuf;
 283
 284         if (!pool)
 285                 return NULL;
 286
 287         spin_lock(&pool->prp_lock);
 288
 289         /* See if we have anything in a pool, and bail out if nothing,
 290          * in writeout path, where this matters, this is safe to do, because
 291          * nothing is lost in this case, and when some in-flight requests
 292          * complete, this code will be called again. */
 293         if (unlikely(list_empty(&pool->prp_req_list))) {
 294                 spin_unlock(&pool->prp_lock);
 295                 return NULL;
 296         }
 297
 298         request = list_entry(pool->prp_req_list.next, struct ptlrpc_request,
 299                              rq_list);
 300         list_del(&request->rq_list);
 301         spin_unlock(&pool->prp_lock);
 302
 303         LASSERT(request->rq_reqbuf);
 304         LASSERT(request->rq_pool);
 305
 306         reqbuf = request->rq_reqbuf;
 307         memset(request, 0, sizeof(*request));
 308         request->rq_reqbuf = reqbuf;
 309         request->rq_reqbuf_len = pool->prp_rq_size;
 310         request->rq_pool = pool;
 311         return request;
 312 }
 313
 314 static void __ptlrpc_free_req_to_pool(struct ptlrpc_request *request)
 315 {
 316         struct ptlrpc_request_pool *pool = request->rq_pool;
 317
 318         spin_lock(&pool->prp_lock);
 319         LASSERT(list_empty(&request->rq_list));
 320         list_add_tail(&request->rq_list, &pool->prp_req_list);
 321         spin_unlock(&pool->prp_lock);
 322 }
 323
 324 static int __ptlrpc_request_bufs_pack(struct ptlrpc_request *request,
 325                                       __u32 version, int opcode,
 326                                       int count, int *lengths, char **bufs,
 327                                       struct ptlrpc_cli_ctx *ctx)
 328 {
 329         struct obd_import  *imp = request->rq_import;
 330         int                 rc;
 331         ENTRY;
 332
 333         if (unlikely(ctx))
 334                 request->rq_cli_ctx = sptlrpc_cli_ctx_get(ctx);
 335         else {
 336                 rc = sptlrpc_req_get_ctx(request);
 337                 if (rc)
 338                         GOTO(out_free, rc);
 339         }
 340
 341         sptlrpc_req_set_flavor(request, opcode);
 342
 343         rc = lustre_pack_request(request, imp->imp_msg_magic, count,
 344                                  lengths, bufs);
 345         if (rc) {
 346                 LASSERT(!request->rq_pool);
 347                 GOTO(out_ctx, rc);
 348         }
 349
 350         lustre_msg_add_version(request->rq_reqmsg, version);
 351
 352         if (imp->imp_server_timeout)
 353                 request->rq_timeout = obd_timeout / 2;
 354         else
 355                 request->rq_timeout = obd_timeout;
 356         request->rq_send_state = LUSTRE_IMP_FULL;
 357         request->rq_type = PTL_RPC_MSG_REQUEST;
 358         request->rq_export = NULL;
 359
 360         request->rq_req_cbid.cbid_fn  = request_out_callback;
 361         request->rq_req_cbid.cbid_arg = request;
 362
 363         request->rq_reply_cbid.cbid_fn  = reply_in_callback;
 364         request->rq_reply_cbid.cbid_arg = request;
 365
 366         request->rq_phase = RQ_PHASE_NEW;
 367
 368         /* XXX FIXME bug 249 */
 369         request->rq_request_portal = imp->imp_client->cli_request_portal;
 370         request->rq_reply_portal = imp->imp_client->cli_reply_portal;
 371
 372         spin_lock_init(&request->rq_lock);
 373         CFS_INIT_LIST_HEAD(&request->rq_list);
 374         CFS_INIT_LIST_HEAD(&request->rq_replay_list);
 375         CFS_INIT_LIST_HEAD(&request->rq_mod_list);
 376         CFS_INIT_LIST_HEAD(&request->rq_ctx_chain);
 377         CFS_INIT_LIST_HEAD(&request->rq_set_chain);
 378         cfs_waitq_init(&request->rq_reply_waitq);
 379         request->rq_xid = ptlrpc_next_xid();
 380         atomic_set(&request->rq_refcount, 1);
 381
 382         lustre_msg_set_opc(request->rq_reqmsg, opcode);
 383         lustre_msg_set_flags(request->rq_reqmsg, 0);
 384
 385         RETURN(0);
 386 out_ctx:
 387         sptlrpc_cli_ctx_put(request->rq_cli_ctx, 1);
 388 out_free:
 389         class_import_put(imp);
 390         return rc;
 391 }
 392
 393 int ptlrpc_request_bufs_pack(struct ptlrpc_request *request,
 394                              __u32 version, int opcode, char **bufs,
 395                              struct ptlrpc_cli_ctx *ctx)
 396 {
 397         int count;
 398
 399         count = req_capsule_filled_sizes(&request->rq_pill, RCL_CLIENT);
 400         return __ptlrpc_request_bufs_pack(request, version, opcode, count,
 401                                           request->rq_pill.rc_area[RCL_CLIENT],
 402                                           bufs, ctx);
 403 }
 404
 405 int ptlrpc_request_pack(struct ptlrpc_request *request,
 406                         __u32 version, int opcode)
 407 {
 408         return ptlrpc_request_bufs_pack(request, version, opcode, NULL, NULL);
 409 }
 410
 411 static inline
 412 struct ptlrpc_request *__ptlrpc_request_alloc(struct obd_import *imp,
 413                                               struct ptlrpc_request_pool *pool)
 414 {
 415         struct ptlrpc_request *request = NULL;
 416
 417         if (pool)
 418                 request = ptlrpc_prep_req_from_pool(pool);
 419
 420         if (!request)
 421                 OBD_ALLOC_PTR(request);
 422
 423         if (request) {
 424                 LASSERT((unsigned long)imp > 0x1000);
 425                 LASSERT(imp != LP_POISON);
 426                 LASSERT((unsigned long)imp->imp_client > 0x1000);
 427                 LASSERT(imp->imp_client != LP_POISON);
 428
 429                 request->rq_import = class_import_get(imp);
 430         } else {
 431                 CERROR("request allocation out of memory\n");
 432         }
 433
 434         return request;
 435 }
 436
 437 static struct ptlrpc_request *
 438 ptlrpc_request_alloc_internal(struct obd_import *imp,
 439                               struct ptlrpc_request_pool * pool,
 440                               const struct req_format *format)
 441 {
 442         struct ptlrpc_request *request;
 443
 444         request = __ptlrpc_request_alloc(imp, pool);
 445         if (request == NULL)
 446                 return NULL;
 447
 448         req_capsule_init(&request->rq_pill, request, RCL_CLIENT);
 449         req_capsule_set(&request->rq_pill, format);
 450         return request;
 451 }
 452
 453 struct ptlrpc_request *ptlrpc_request_alloc(struct obd_import *imp,
 454                                             const struct req_format *format)
 455 {
 456         return ptlrpc_request_alloc_internal(imp, NULL, format);
 457 }
 458
 459 struct ptlrpc_request *ptlrpc_request_alloc_pool(struct obd_import *imp,
 460                                             struct ptlrpc_request_pool * pool,
 461                                             const struct req_format *format)
 462 {
 463         return ptlrpc_request_alloc_internal(imp, pool, format);
 464 }
 465
 466 void ptlrpc_request_free(struct ptlrpc_request *request)
 467 {
 468         if (request->rq_pool)
 469                 __ptlrpc_free_req_to_pool(request);
 470         else
 471                 OBD_FREE_PTR(request);
 472 }
 473
 474 struct ptlrpc_request *ptlrpc_request_alloc_pack(struct obd_import *imp,
 475                                                 const struct req_format *format,
 476                                                 __u32 version, int opcode)
 477 {
 478         struct ptlrpc_request *req = ptlrpc_request_alloc(imp, format);
 479         int                    rc;
 480
 481         if (req) {
 482                 rc = ptlrpc_request_pack(req, version, opcode);
 483                 if (rc) {
 484                         ptlrpc_request_free(req);
 485                         req = NULL;
 486                 }
 487         }
 488         return req;
 489 }
 490
 491 struct ptlrpc_request *
 492 ptlrpc_prep_req_pool(struct obd_import *imp,
 493                      __u32 version, int opcode,
 494                      int count, int *lengths, char **bufs,
 495                      struct ptlrpc_request_pool *pool)
 496 {
 497         struct ptlrpc_request *request;
 498         int                    rc;
 499
 500         request = __ptlrpc_request_alloc(imp, pool);
 501         if (!request)
 502                 return NULL;
 503
 504         rc = __ptlrpc_request_bufs_pack(request, version, opcode, count,
 505                                         lengths, bufs, NULL);
 506         if (rc) {
 507                 ptlrpc_request_free(request);
 508                 request = NULL;
 509         }
 510         return request;
 511 }
 512
 513 struct ptlrpc_request *
 514 ptlrpc_prep_req(struct obd_import *imp, __u32 version, int opcode, int count,
 515                 int *lengths, char **bufs)
 516 {
 517         return ptlrpc_prep_req_pool(imp, version, opcode, count, lengths, bufs,
 518                                     NULL);
 519 }
 520
 521 struct ptlrpc_request_set *ptlrpc_prep_set(void)
 522 {
 523         struct ptlrpc_request_set *set;
 524
 525         ENTRY;
 526         OBD_ALLOC(set, sizeof *set);
 527         if (!set)
 528                 RETURN(NULL);
 529         CFS_INIT_LIST_HEAD(&set->set_requests);
 530         cfs_waitq_init(&set->set_waitq);
 531         set->set_remaining = 0;
 532         spin_lock_init(&set->set_new_req_lock);
 533         CFS_INIT_LIST_HEAD(&set->set_new_requests);
 534         CFS_INIT_LIST_HEAD(&set->set_cblist);
 535
 536         RETURN(set);
 537 }
 538
 539 /* Finish with this set; opposite of prep_set. */
 540 void ptlrpc_set_destroy(struct ptlrpc_request_set *set)
 541 {
 542         struct list_head *tmp;
 543         struct list_head *next;
 544         int               expected_phase;
 545         int               n = 0;
 546         ENTRY;
 547
 548         /* Requests on the set should either all be completed, or all be new */
 549         expected_phase = (set->set_remaining == 0) ?
 550                          RQ_PHASE_COMPLETE : RQ_PHASE_NEW;
 551         list_for_each (tmp, &set->set_requests) {
 552                 struct ptlrpc_request *req =
 553                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 554
 555                 LASSERT(req->rq_phase == expected_phase);
 556                 n++;
 557         }
 558
 559         LASSERT(set->set_remaining == 0 || set->set_remaining == n);
 560
 561         list_for_each_safe(tmp, next, &set->set_requests) {
 562                 struct ptlrpc_request *req =
 563                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 564                 list_del_init(&req->rq_set_chain);
 565
 566                 LASSERT(req->rq_phase == expected_phase);
 567
 568                 if (req->rq_phase == RQ_PHASE_NEW) {
 569
 570                         if (req->rq_interpret_reply != NULL) {
 571                                 int (*interpreter)(struct ptlrpc_request *,
 572                                                    void *, int) =
 573                                         req->rq_interpret_reply;
 574
 575                                 /* higher level (i.e. LOV) failed;
 576                                  * let the sub reqs clean up */
 577                                 req->rq_status = -EBADR;
 578                                 interpreter(req, &req->rq_async_args,
 579                                             req->rq_status);
 580                         }
 581                         set->set_remaining--;
 582                 }
 583
 584                 req->rq_set = NULL;
 585                 ptlrpc_req_finished (req);
 586         }
 587
 588         LASSERT(set->set_remaining == 0);
 589
 590         OBD_FREE(set, sizeof(*set));
 591         EXIT;
 592 }
 593
 594 int ptlrpc_set_add_cb(struct ptlrpc_request_set *set,
 595                       set_interpreter_func fn, void *data)
 596 {
 597         struct ptlrpc_set_cbdata *cbdata;
 598
 599         OBD_ALLOC_PTR(cbdata);
 600         if (cbdata == NULL)
 601                 RETURN(-ENOMEM);
 602
 603         cbdata->psc_interpret = fn;
 604         cbdata->psc_data = data;
 605         list_add_tail(&cbdata->psc_item, &set->set_cblist);
 606
 607         RETURN(0);
 608 }
 609
 610 void ptlrpc_set_add_req(struct ptlrpc_request_set *set,
 611                         struct ptlrpc_request *req)
 612 {
 613         /* The set takes over the caller's request reference */
 614         list_add_tail(&req->rq_set_chain, &set->set_requests);
 615         req->rq_set = set;
 616         set->set_remaining++;
 617
 618         atomic_inc(&req->rq_import->imp_inflight);
 619 }
 620
 621 /* lock so many callers can add things, the context that owns the set
 622  * is supposed to notice these and move them into the set proper. */
 623 void ptlrpc_set_add_new_req(struct ptlrpc_request_set *set,
 624                             struct ptlrpc_request *req)
 625 {
 626         spin_lock(&set->set_new_req_lock);
 627         /* The set takes over the caller's request reference */
 628         list_add_tail(&req->rq_set_chain, &set->set_new_requests);
 629         req->rq_set = set;
 630         spin_unlock(&set->set_new_req_lock);
 631 }
 632
 633 /*
 634  * Based on the current state of the import, determine if the request
 635  * can be sent, is an error, or should be delayed.
 636  *
 637  * Returns true if this request should be delayed. If false, and
 638  * *status is set, then the request can not be sent and *status is the
 639  * error code.  If false and status is 0, then request can be sent.
 640  *
 641  * The imp->imp_lock must be held.
 642  */
 643 static int ptlrpc_import_delay_req(struct obd_import *imp,
 644                                    struct ptlrpc_request *req, int *status)
 645 {
 646         int delay = 0;
 647         ENTRY;
 648
 649         LASSERT (status != NULL);
 650         *status = 0;
 651
 652         if (req->rq_ctx_init || req->rq_ctx_fini) {
 653                 /* always allow ctx init/fini rpc go through */
 654         } else if (imp->imp_state == LUSTRE_IMP_NEW) {
 655                 DEBUG_REQ(D_ERROR, req, "Uninitialized import.");
 656                 *status = -EIO;
 657                 LBUG();
 658         } else if (imp->imp_state == LUSTRE_IMP_CLOSED) {
 659                 DEBUG_REQ(D_ERROR, req, "IMP_CLOSED ");
 660                 *status = -EIO;
 661         } else if (req->rq_send_state == LUSTRE_IMP_CONNECTING &&
 662                  imp->imp_state == LUSTRE_IMP_CONNECTING) {
 663                 /* allow CONNECT even if import is invalid */ ;
 664         } else if (imp->imp_invalid) {
 665                 /* if it is mgc, wait for recovry. b=13464 */
 666                 if (imp->imp_recon_bk && !imp->imp_obd->obd_no_recov)
 667                         delay = 1;
 668                 /* If the import has been invalidated (such as by an OST
 669                  * failure) the request must fail with -ESHUTDOWN.  This
 670                  * indicates the requests should be discarded; an -EIO
 671                  * may result in a resend of the request. */
 672                 if (!imp->imp_deactive)
 673                         DEBUG_REQ(D_ERROR, req, "IMP_INVALID");
 674                 *status = -ESHUTDOWN; /* bz 12940 */
 675         } else if (req->rq_import_generation != imp->imp_generation) {
 676                 DEBUG_REQ(D_ERROR, req, "req wrong generation:");
 677                 *status = -EIO;
 678         } else if (req->rq_send_state != imp->imp_state) {
 679                 if (imp->imp_obd->obd_no_recov || imp->imp_dlm_fake ||
 680                     req->rq_no_delay)
 681                         *status = -EWOULDBLOCK;
 682                 else
 683                         delay = 1;
 684         }
 685
 686         RETURN(delay);
 687 }
 688
 689 static int ptlrpc_check_reply(struct ptlrpc_request *req)
 690 {
 691         int rc = 0;
 692         ENTRY;
 693
 694         /* serialise with network callback */
 695         spin_lock(&req->rq_lock);
 696
 697         if (req->rq_replied)
 698                 GOTO(out, rc = 1);
 699
 700         if (req->rq_net_err && !req->rq_timedout) {
 701                 spin_unlock(&req->rq_lock);
 702                 rc = ptlrpc_expire_one_request(req);
 703                 spin_lock(&req->rq_lock);
 704                 GOTO(out, rc);
 705         }
 706
 707         if (req->rq_err)
 708                 GOTO(out, rc = 1);
 709
 710         if (req->rq_resend)
 711                 GOTO(out, rc = 1);
 712
 713         if (req->rq_restart)
 714                 GOTO(out, rc = 1);
 715         EXIT;
 716  out:
 717         spin_unlock(&req->rq_lock);
 718         DEBUG_REQ(D_NET, req, "rc = %d for", rc);
 719         return rc;
 720 }
 721
 722 static int ptlrpc_check_status(struct ptlrpc_request *req)
 723 {
 724         int err;
 725         ENTRY;
 726
 727         err = lustre_msg_get_status(req->rq_repmsg);
 728         if (lustre_msg_get_type(req->rq_repmsg) == PTL_RPC_MSG_ERR) {
 729                 struct obd_import *imp = req->rq_import;
 730                 __u32 opc = lustre_msg_get_opc(req->rq_reqmsg);
 731                 LCONSOLE_ERROR_MSG(0x011,"an error occurred while communicating"
 732                                 " with %s. The %s operation failed with %d\n",
 733                                 libcfs_nid2str(imp->imp_connection->c_peer.nid),
 734                                 ll_opcode2str(opc), err);
 735                 RETURN(err < 0 ? err : -EINVAL);
 736         }
 737
 738         if (err < 0) {
 739                 DEBUG_REQ(D_INFO, req, "status is %d", err);
 740         } else if (err > 0) {
 741                 /* XXX: translate this error from net to host */
 742                 DEBUG_REQ(D_INFO, req, "status is %d", err);
 743         }
 744
 745         RETURN(err);
 746 }
 747
 748 static int after_reply(struct ptlrpc_request *req)
 749 {
 750         struct obd_import *imp = req->rq_import;
 751         struct obd_device *obd = req->rq_import->imp_obd;
 752         int rc;
 753         struct timeval work_start;
 754         long timediff;
 755         ENTRY;
 756
 757         LASSERT(!req->rq_receiving_reply);
 758         LASSERT(obd);
 759         LASSERT(req->rq_nob_received <= req->rq_repbuf_len);
 760
 761         /* NB Until this point, the whole of the incoming message,
 762          * including buflens, status etc is in the sender's byte order. */
 763
 764         /* Clear reply swab mask; this is a new reply in sender's byte order */
 765         req->rq_rep_swab_mask = 0;
 766
 767         rc = sptlrpc_cli_unwrap_reply(req);
 768         if (rc) {
 769                 DEBUG_REQ(D_ERROR, req, "unwrap reply failed (%d):", rc);
 770                 RETURN(rc);
 771         }
 772
 773         /* security layer unwrap might ask resend this request */
 774         if (req->rq_resend)
 775                 RETURN(0);
 776
 777         rc = lustre_unpack_msg(req->rq_repmsg, req->rq_replen);
 778         if (rc) {
 779                 DEBUG_REQ(D_ERROR, req, "unpack_rep failed: %d", rc);
 780                 RETURN(-EPROTO);
 781         }
 782
 783         rc = lustre_unpack_rep_ptlrpc_body(req, MSG_PTLRPC_BODY_OFF);
 784         if (rc) {
 785                 DEBUG_REQ(D_ERROR, req, "unpack ptlrpc body failed: %d", rc);
 786                 RETURN(-EPROTO);
 787         }
 788
 789         do_gettimeofday(&work_start);
 790         timediff = cfs_timeval_sub(&work_start, &req->rq_arrival_time, NULL);
 791         if (obd->obd_svc_stats != NULL)
 792                 lprocfs_counter_add(obd->obd_svc_stats, PTLRPC_REQWAIT_CNTR,
 793                                     timediff);
 794
 795         if (lustre_msg_get_type(req->rq_repmsg) != PTL_RPC_MSG_REPLY &&
 796             lustre_msg_get_type(req->rq_repmsg) != PTL_RPC_MSG_ERR) {
 797                 DEBUG_REQ(D_ERROR, req, "invalid packet received (type=%u)",
 798                           lustre_msg_get_type(req->rq_repmsg));
 799                 RETURN(-EPROTO);
 800         }
 801
 802         rc = ptlrpc_check_status(req);
 803         imp->imp_connect_error = rc;
 804
 805         if (rc) {
 806                 /* Either we've been evicted, or the server has failed for
 807                  * some reason. Try to reconnect, and if that fails, punt to
 808                  * the upcall. */
 809                 if (ll_rpc_recoverable_error(rc)) {
 810                         if (req->rq_send_state != LUSTRE_IMP_FULL ||
 811                             imp->imp_obd->obd_no_recov || imp->imp_dlm_fake) {
 812                                 RETURN(rc);
 813                         }
 814                         ptlrpc_request_handle_notconn(req);
 815                         RETURN(rc);
 816                 }
 817         } else {
 818                 /* Let's look if server send slv. Do it only for RPC with
 819                  * rc == 0. */
 820                 if (imp->imp_obd->obd_namespace) {
 821                         /* Disconnect rpc is sent when namespace is already
 822                          * destroyed. Let's check this and will not try update
 823                          * pool. */
 824                         ldlm_cli_update_pool(req);
 825                 }
 826         }
 827
 828         /* Store transno in reqmsg for replay. */
 829         req->rq_transno = lustre_msg_get_transno(req->rq_repmsg);
 830         lustre_msg_set_transno(req->rq_reqmsg, req->rq_transno);
 831
 832         if (req->rq_import->imp_replayable) {
 833                 spin_lock(&imp->imp_lock);
 834                 /* no point in adding already-committed requests to the replay
 835                  * list, we will just remove them immediately. b=9829 */
 836                 if (req->rq_transno != 0 &&
 837                     (req->rq_transno >
 838                      lustre_msg_get_last_committed(req->rq_repmsg) ||
 839                      req->rq_replay))
 840                         ptlrpc_retain_replayable_request(req, imp);
 841                 else if (req->rq_commit_cb != NULL) {
 842                         spin_unlock(&imp->imp_lock);
 843                         req->rq_commit_cb(req);
 844                         spin_lock(&imp->imp_lock);
 845                 }
 846
 847                 /* Replay-enabled imports return commit-status information. */
 848                 if (lustre_msg_get_last_committed(req->rq_repmsg)) {
 849                         imp->imp_peer_committed_transno =
 850                                 lustre_msg_get_last_committed(req->rq_repmsg);
 851                 }
 852                 ptlrpc_free_committed(imp);
 853                 spin_unlock(&imp->imp_lock);
 854         }
 855
 856         RETURN(rc);
 857 }
 858
 859 static int ptlrpc_send_new_req(struct ptlrpc_request *req)
 860 {
 861         struct obd_import     *imp;
 862         int rc;
 863         ENTRY;
 864
 865         LASSERT(req->rq_phase == RQ_PHASE_NEW);
 866         if (req->rq_sent && (req->rq_sent > CURRENT_SECONDS))
 867                 RETURN (0);
 868
 869         req->rq_phase = RQ_PHASE_RPC;
 870
 871         imp = req->rq_import;
 872         spin_lock(&imp->imp_lock);
 873
 874         req->rq_import_generation = imp->imp_generation;
 875
 876         if (ptlrpc_import_delay_req(imp, req, &rc)) {
 877                 spin_lock (&req->rq_lock);
 878                 req->rq_waiting = 1;
 879                 spin_unlock (&req->rq_lock);
 880
 881                 DEBUG_REQ(D_HA, req, "req from PID %d waiting for recovery: "
 882                           "(%s != %s)",
 883                           lustre_msg_get_status(req->rq_reqmsg) ,
 884                           ptlrpc_import_state_name(req->rq_send_state),
 885                           ptlrpc_import_state_name(imp->imp_state));
 886                 LASSERT(list_empty (&req->rq_list));
 887
 888                 list_add_tail(&req->rq_list, &imp->imp_delayed_list);
 889                 spin_unlock(&imp->imp_lock);
 890                 RETURN(0);
 891         }
 892
 893         if (rc != 0) {
 894                 spin_unlock(&imp->imp_lock);
 895                 req->rq_status = rc;
 896                 req->rq_phase = RQ_PHASE_INTERPRET;
 897                 RETURN(rc);
 898         }
 899
 900         /* XXX this is the same as ptlrpc_queue_wait */
 901         LASSERT(list_empty(&req->rq_list));
 902         list_add_tail(&req->rq_list, &imp->imp_sending_list);
 903         spin_unlock(&imp->imp_lock);
 904
 905         lustre_msg_set_status(req->rq_reqmsg, cfs_curproc_pid());
 906
 907         rc = sptlrpc_req_refresh_ctx(req, -1);
 908         if (rc) {
 909                 if (req->rq_err) {
 910                         req->rq_status = rc;
 911                         RETURN(1);
 912                 } else {
 913                         /* here begins timeout counting */
 914                         req->rq_sent = CURRENT_SECONDS;
 915                         req->rq_wait_ctx = 1;
 916                         RETURN(0);
 917                 }
 918         }
 919
 920         CDEBUG(D_RPCTRACE, "Sending RPC pname:cluuid:pid:xid:nid:opc"
 921                " %s:%s:%d:"LPU64":%s:%d\n", cfs_curproc_comm(),
 922                imp->imp_obd->obd_uuid.uuid,
 923                lustre_msg_get_status(req->rq_reqmsg), req->rq_xid,
 924                libcfs_nid2str(imp->imp_connection->c_peer.nid),
 925                lustre_msg_get_opc(req->rq_reqmsg));
 926
 927         rc = ptl_send_rpc(req, 0);
 928         if (rc) {
 929                 DEBUG_REQ(D_HA, req, "send failed (%d); expect timeout", rc);
 930                 req->rq_net_err = 1;
 931                 RETURN(rc);
 932         }
 933         RETURN(0);
 934 }
 935
 936 /* this sends any unsent RPCs in @set and returns TRUE if all are sent */
 937 int ptlrpc_check_set(struct ptlrpc_request_set *set)
 938 {
 939         struct list_head *tmp;
 940         int force_timer_recalc = 0;
 941         ENTRY;
 942
 943         if (set->set_remaining == 0)
 944                 RETURN(1);
 945
 946         list_for_each(tmp, &set->set_requests) {
 947                 struct ptlrpc_request *req =
 948                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 949                 struct obd_import *imp = req->rq_import;
 950                 int rc = 0;
 951
 952                 if (req->rq_phase == RQ_PHASE_NEW &&
 953                     ptlrpc_send_new_req(req)) {
 954                         force_timer_recalc = 1;
 955                 }
 956                 /* delayed send - skip */
 957                 if (req->rq_phase == RQ_PHASE_NEW && req->rq_sent)
 958                         continue;
 959
 960                 if (!(req->rq_phase == RQ_PHASE_RPC ||
 961                       req->rq_phase == RQ_PHASE_BULK ||
 962                       req->rq_phase == RQ_PHASE_INTERPRET ||
 963                       req->rq_phase == RQ_PHASE_COMPLETE)) {
 964                         DEBUG_REQ(D_ERROR, req, "bad phase %x", req->rq_phase);
 965                         LBUG();
 966                 }
 967
 968                 if (req->rq_phase == RQ_PHASE_COMPLETE)
 969                         continue;
 970
 971                 if (req->rq_phase == RQ_PHASE_INTERPRET)
 972                         GOTO(interpret, req->rq_status);
 973
 974                 if (req->rq_net_err && !req->rq_timedout)
 975                         ptlrpc_expire_one_request(req);
 976
 977                 if (req->rq_err) {
 978                         ptlrpc_unregister_reply(req);
 979                         req->rq_replied = 0;
 980                         if (req->rq_status == 0)
 981                                 req->rq_status = -EIO;
 982                         req->rq_phase = RQ_PHASE_INTERPRET;
 983
 984                         spin_lock(&imp->imp_lock);
 985                         list_del_init(&req->rq_list);
 986                         spin_unlock(&imp->imp_lock);
 987
 988                         GOTO(interpret, req->rq_status);
 989                 }
 990
 991                 /* ptlrpc_queue_wait->l_wait_event guarantees that rq_intr
 992                  * will only be set after rq_timedout, but the oig waiting
 993                  * path sets rq_intr irrespective of whether ptlrpcd has
 994                  * seen a timeout.  our policy is to only interpret
 995                  * interrupted rpcs after they have timed out */
 996                 if (req->rq_intr && (req->rq_timedout || req->rq_waiting ||
 997                                      req->rq_wait_ctx)) {
 998                         /* NB could be on delayed list */
 999                         ptlrpc_unregister_reply(req);
1000                         req->rq_status = -EINTR;
1001                         req->rq_phase = RQ_PHASE_INTERPRET;
1002
1003                         spin_lock(&imp->imp_lock);
1004                         list_del_init(&req->rq_list);
1005                         spin_unlock(&imp->imp_lock);
1006
1007                         GOTO(interpret, req->rq_status);
1008                 }
1009
1010                 if (req->rq_phase == RQ_PHASE_RPC) {
1011                         if (req->rq_timedout || req->rq_resend ||
1012                             req->rq_waiting || req->rq_wait_ctx) {
1013                                 int status;
1014
1015                                 /* rq_wait_ctx is only touched in ptlrpcd,
1016                                  * no lock needed here.
1017                                  */
1018                                 if (req->rq_wait_ctx)
1019                                         goto check_ctx;
1020
1021                                 ptlrpc_unregister_reply(req);
1022
1023                                 spin_lock(&imp->imp_lock);
1024
1025                                 if (ptlrpc_import_delay_req(imp, req, &status)){
1026                                         spin_unlock(&imp->imp_lock);
1027                                         continue;
1028                                 }
1029
1030                                 list_del_init(&req->rq_list);
1031                                 if (status != 0)  {
1032                                         req->rq_status = status;
1033                                         req->rq_phase = RQ_PHASE_INTERPRET;
1034                                         spin_unlock(&imp->imp_lock);
1035                                         GOTO(interpret, req->rq_status);
1036                                 }
1037                                 if (req->rq_no_resend && !req->rq_wait_ctx) {
1038                                         req->rq_status = -ENOTCONN;
1039                                         req->rq_phase = RQ_PHASE_INTERPRET;
1040                                         spin_unlock(&imp->imp_lock);
1041                                         GOTO(interpret, req->rq_status);
1042                                 }
1043                                 list_add_tail(&req->rq_list,
1044                                               &imp->imp_sending_list);
1045
1046                                 spin_unlock(&imp->imp_lock);
1047
1048                                 req->rq_waiting = 0;
1049                                 if (req->rq_resend) {
1050                                         lustre_msg_add_flags(req->rq_reqmsg,
1051                                                              MSG_RESENT);
1052                                         if (req->rq_bulk) {
1053                                                 __u64 old_xid = req->rq_xid;
1054
1055                                                 ptlrpc_unregister_bulk (req);
1056
1057                                                 /* ensure previous bulk fails */
1058                                                 req->rq_xid = ptlrpc_next_xid();
1059                                                 CDEBUG(D_HA, "resend bulk "
1060                                                        "old x"LPU64
1061                                                        " new x"LPU64"\n",
1062                                                        old_xid, req->rq_xid);
1063                                         }
1064                                 }
1065 check_ctx:
1066                                 status = sptlrpc_req_refresh_ctx(req, -1);
1067                                 if (status) {
1068                                         if (req->rq_err) {
1069                                                 req->rq_status = status;
1070                                                 force_timer_recalc = 1;
1071                                         }
1072                                         if (!req->rq_wait_ctx) {
1073                                                 /* begins timeout counting */
1074                                                 req->rq_sent = CURRENT_SECONDS;
1075                                                 req->rq_wait_ctx = 1;
1076                                         }
1077                                         continue;
1078                                 } else {
1079                                         req->rq_sent = 0;
1080                                         req->rq_wait_ctx = 0;
1081                                 }
1082
1083                                 rc = ptl_send_rpc(req, 0);
1084                                 if (rc) {
1085                                         DEBUG_REQ(D_HA, req, "send failed (%d)",
1086                                                   rc);
1087                                         force_timer_recalc = 1;
1088                                         req->rq_net_err = 1;
1089                                 }
1090                                 /* need to reset the timeout */
1091                                 force_timer_recalc = 1;
1092                         }
1093
1094                         /* Still waiting for a reply? */
1095                         if (ptlrpc_client_receiving_reply(req))
1096                                 continue;
1097
1098                         /* Did we actually receive a reply? */
1099                         if (!ptlrpc_client_replied(req))
1100                                 continue;
1101
1102                         spin_lock(&imp->imp_lock);
1103                         list_del_init(&req->rq_list);
1104                         spin_unlock(&imp->imp_lock);
1105
1106                         req->rq_status = after_reply(req);
1107                         if (req->rq_resend) {
1108                                 /* Add this req to the delayed list so
1109                                    it can be errored if the import is
1110                                    evicted after recovery. */
1111                                 spin_lock(&imp->imp_lock);
1112                                 list_add_tail(&req->rq_list,
1113                                               &imp->imp_delayed_list);
1114                                 spin_unlock(&imp->imp_lock);
1115                                 continue;
1116                         }
1117
1118                         /* If there is no bulk associated with this request,
1119                          * then we're done and should let the interpreter
1120                          * process the reply.  Similarly if the RPC returned
1121                          * an error, and therefore the bulk will never arrive.
1122                          */
1123                         if (req->rq_bulk == NULL || req->rq_status != 0) {
1124                                 req->rq_phase = RQ_PHASE_INTERPRET;
1125                                 GOTO(interpret, req->rq_status);
1126                         }
1127
1128                         req->rq_phase = RQ_PHASE_BULK;
1129                 }
1130
1131                 LASSERT(req->rq_phase == RQ_PHASE_BULK);
1132                 if (ptlrpc_bulk_active(req->rq_bulk))
1133                         continue;
1134
1135                 if (!req->rq_bulk->bd_success) {
1136                         /* The RPC reply arrived OK, but the bulk screwed
1137                          * up!  Dead wierd since the server told us the RPC
1138                          * was good after getting the REPLY for her GET or
1139                          * the ACK for her PUT. */
1140                         DEBUG_REQ(D_ERROR, req, "bulk transfer failed");
1141                         LBUG();
1142                 }
1143
1144                 req->rq_phase = RQ_PHASE_INTERPRET;
1145
1146         interpret:
1147                 LASSERT(req->rq_phase == RQ_PHASE_INTERPRET);
1148                 LASSERT(!req->rq_receiving_reply);
1149
1150                 ptlrpc_unregister_reply(req);
1151                 if (req->rq_bulk != NULL)
1152                         ptlrpc_unregister_bulk (req);
1153
1154                 if (req->rq_interpret_reply != NULL) {
1155                         int (*interpreter)(struct ptlrpc_request *,void *,int) =
1156                                 req->rq_interpret_reply;
1157                         req->rq_status = interpreter(req, &req->rq_async_args,
1158                                                      req->rq_status);
1159                 }
1160                 req->rq_phase = RQ_PHASE_COMPLETE;
1161
1162                 CDEBUG(D_RPCTRACE, "Completed RPC pname:cluuid:pid:xid:nid:"
1163                        "opc %s:%s:%d:"LPU64":%s:%d\n", cfs_curproc_comm(),
1164                        imp->imp_obd->obd_uuid.uuid,
1165                        lustre_msg_get_status(req->rq_reqmsg), req->rq_xid,
1166                        libcfs_nid2str(imp->imp_connection->c_peer.nid),
1167                        lustre_msg_get_opc(req->rq_reqmsg));
1168
1169                 set->set_remaining--;
1170
1171                 atomic_dec(&imp->imp_inflight);
1172                 cfs_waitq_signal(&imp->imp_recovery_waitq);
1173         }
1174
1175         /* If we hit an error, we want to recover promptly. */
1176         RETURN(set->set_remaining == 0 || force_timer_recalc);
1177 }
1178
1179 int ptlrpc_expire_one_request(struct ptlrpc_request *req)
1180 {
1181         struct obd_import *imp = req->rq_import;
1182         int rc = 0;
1183         ENTRY;
1184
1185         DEBUG_REQ(D_ERROR|D_NETERROR, req, "%s (sent at %lu, %lus ago)",
1186                   req->rq_net_err ? "network error" : "timeout",
1187                   (long)req->rq_sent, CURRENT_SECONDS - req->rq_sent);
1188
1189         if (imp != NULL && obd_debug_peer_on_timeout)
1190                 LNetCtl(IOC_LIBCFS_DEBUG_PEER, &imp->imp_connection->c_peer);
1191
1192         spin_lock(&req->rq_lock);
1193         req->rq_timedout = 1;
1194         req->rq_wait_ctx = 0;
1195         spin_unlock(&req->rq_lock);
1196
1197         ptlrpc_unregister_reply (req);
1198
1199         if (obd_dump_on_timeout)
1200                 libcfs_debug_dumplog();
1201
1202         if (req->rq_bulk != NULL)
1203                 ptlrpc_unregister_bulk (req);
1204
1205         if (imp == NULL) {
1206                 DEBUG_REQ(D_HA, req, "NULL import: already cleaned up?");
1207                 RETURN(1);
1208         }
1209
1210         /* The DLM server doesn't want recovery run on its imports. */
1211         if (imp->imp_dlm_fake)
1212                 RETURN(1);
1213
1214         /* If this request is for recovery or other primordial tasks,
1215          * then error it out here. */
1216         if (req->rq_ctx_init || req->rq_ctx_fini ||
1217             req->rq_send_state != LUSTRE_IMP_FULL ||
1218             imp->imp_obd->obd_no_recov) {
1219                 spin_lock(&req->rq_lock);
1220                 req->rq_status = -ETIMEDOUT;
1221                 req->rq_err = 1;
1222                 spin_unlock(&req->rq_lock);
1223                 RETURN(1);
1224         }
1225
1226         /* if request can't be resend we can't wait answer after timeout */
1227         if (req->rq_no_resend) {
1228                 DEBUG_REQ(D_RPCTRACE, req, "TIMEOUT-NORESEND:");
1229                 rc = 1;
1230         }
1231
1232         ptlrpc_fail_import(imp, lustre_msg_get_conn_cnt(req->rq_reqmsg));
1233
1234         RETURN(rc);
1235 }
1236
1237 int ptlrpc_expired_set(void *data)
1238 {
1239         struct ptlrpc_request_set *set = data;
1240         struct list_head          *tmp;
1241         time_t                     now = CURRENT_SECONDS;
1242         ENTRY;
1243
1244         LASSERT(set != NULL);
1245
1246         /* A timeout expired; see which reqs it applies to... */
1247         list_for_each (tmp, &set->set_requests) {
1248                 struct ptlrpc_request *req =
1249                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
1250
1251                 /* request in-flight? */
1252                 if (!((req->rq_phase == RQ_PHASE_RPC && !req->rq_waiting &&
1253                        !req->rq_resend) ||
1254                       (req->rq_phase == RQ_PHASE_BULK)))
1255                         continue;
1256
1257                 if (req->rq_timedout ||           /* already dealt with */
1258                     req->rq_sent + req->rq_timeout > now) /* not expired */
1259                         continue;
1260
1261                 /* deal with this guy */
1262                 ptlrpc_expire_one_request (req);
1263         }
1264
1265         /* When waiting for a whole set, we always to break out of the
1266          * sleep so we can recalculate the timeout, or enable interrupts
1267          * iff everyone's timed out.
1268          */
1269         RETURN(1);
1270 }
1271
1272 void ptlrpc_mark_interrupted(struct ptlrpc_request *req)
1273 {
1274         spin_lock(&req->rq_lock);
1275         req->rq_intr = 1;
1276         spin_unlock(&req->rq_lock);
1277 }
1278
1279 void ptlrpc_interrupted_set(void *data)
1280 {
1281         struct ptlrpc_request_set *set = data;
1282         struct list_head *tmp;
1283
1284         LASSERT(set != NULL);
1285         CERROR("INTERRUPTED SET %p\n", set);
1286
1287         list_for_each(tmp, &set->set_requests) {
1288                 struct ptlrpc_request *req =
1289                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
1290
1291                 if (req->rq_phase != RQ_PHASE_RPC)
1292                         continue;
1293
1294                 ptlrpc_mark_interrupted(req);
1295         }
1296 }
1297
1298 int ptlrpc_set_next_timeout(struct ptlrpc_request_set *set)
1299 {
1300         struct list_head      *tmp;
1301         time_t                 now = CURRENT_SECONDS;
1302         time_t                 deadline;
1303         int                    timeout = 0;
1304         struct ptlrpc_request *req;
1305         ENTRY;
1306
1307         SIGNAL_MASK_ASSERT(); /* XXX BUG 1511 */
1308
1309         list_for_each(tmp, &set->set_requests) {
1310                 req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
1311
1312                 /* request in-flight? */
1313                 if (!((req->rq_phase == RQ_PHASE_RPC && !req->rq_waiting) ||
1314                       (req->rq_phase == RQ_PHASE_BULK) ||
1315                       (req->rq_phase == RQ_PHASE_NEW)))
1316                         continue;
1317
1318                 if (req->rq_timedout)   /* already timed out */
1319                         continue;
1320
1321                 if (req->rq_phase == RQ_PHASE_NEW)
1322                         deadline = req->rq_sent;
1323                 else
1324                         deadline = req->rq_sent + req->rq_timeout;
1325
1326                 if (deadline <= now)    /* actually expired already */
1327                         timeout = 1;    /* ASAP */
1328                 else if (timeout == 0 || timeout > deadline - now)
1329                         timeout = deadline - now;
1330         }
1331         RETURN(timeout);
1332 }
1333
1334 int ptlrpc_set_wait(struct ptlrpc_request_set *set)
1335 {
1336         struct list_head      *tmp;
1337         struct ptlrpc_request *req;
1338         struct l_wait_info     lwi;
1339         int                    rc, timeout;
1340         ENTRY;
1341
1342         if (list_empty(&set->set_requests))
1343                 RETURN(0);
1344
1345         list_for_each(tmp, &set->set_requests) {
1346                 req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
1347                 if (req->rq_phase == RQ_PHASE_NEW)
1348                         (void)ptlrpc_send_new_req(req);
1349         }
1350
1351         do {
1352                 timeout = ptlrpc_set_next_timeout(set);
1353
1354                 /* wait until all complete, interrupted, or an in-flight
1355                  * req times out */
1356                 CDEBUG(D_RPCTRACE, "set %p going to sleep for %d seconds\n",
1357                        set, timeout);
1358                 lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(timeout ? timeout : 1),
1359                                        ptlrpc_expired_set,
1360                                        ptlrpc_interrupted_set, set);
1361                 rc = l_wait_event(set->set_waitq, ptlrpc_check_set(set), &lwi);
1362
1363                 LASSERT(rc == 0 || rc == -EINTR || rc == -ETIMEDOUT);
1364
1365                 /* -EINTR => all requests have been flagged rq_intr so next
1366                  * check completes.
1367                  * -ETIMEOUTD => someone timed out.  When all reqs have
1368                  * timed out, signals are enabled allowing completion with
1369                  * EINTR.
1370                  * I don't really care if we go once more round the loop in
1371                  * the error cases -eeb. */
1372         } while (rc != 0 || set->set_remaining != 0);
1373
1374         LASSERT(set->set_remaining == 0);
1375
1376         rc = 0;
1377         list_for_each(tmp, &set->set_requests) {
1378                 req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
1379
1380                 LASSERT(req->rq_phase == RQ_PHASE_COMPLETE);
1381                 if (req->rq_status != 0)
1382                         rc = req->rq_status;
1383         }
1384
1385         if (set->set_interpret != NULL) {
1386                 int (*interpreter)(struct ptlrpc_request_set *set,void *,int) =
1387                         set->set_interpret;
1388                 rc = interpreter (set, set->set_arg, rc);
1389         } else {
1390                 struct ptlrpc_set_cbdata *cbdata, *n;
1391                 int err;
1392
1393                 list_for_each_entry_safe(cbdata, n,
1394                                          &set->set_cblist, psc_item) {
1395                         list_del_init(&cbdata->psc_item);
1396                         err = cbdata->psc_interpret(set, cbdata->psc_data, rc);
1397                         if (err && !rc)
1398                                 rc = err;
1399                         OBD_FREE_PTR(cbdata);
1400                 }
1401         }
1402
1403         RETURN(rc);
1404 }
1405
1406 static void __ptlrpc_free_req(struct ptlrpc_request *request, int locked)
1407 {
1408         ENTRY;
1409         if (request == NULL) {
1410                 EXIT;
1411                 return;
1412         }
1413
1414         LASSERTF(!request->rq_receiving_reply, "req %p\n", request);
1415         LASSERTF(request->rq_rqbd == NULL, "req %p\n",request);/* client-side */
1416         LASSERTF(list_empty(&request->rq_list), "req %p\n", request);
1417         LASSERTF(list_empty(&request->rq_set_chain), "req %p\n", request);
1418         LASSERT(request->rq_cli_ctx);
1419
1420         req_capsule_fini(&request->rq_pill);
1421
1422         /* We must take it off the imp_replay_list first.  Otherwise, we'll set
1423          * request->rq_reqmsg to NULL while osc_close is dereferencing it. */
1424         if (request->rq_import != NULL) {
1425                 if (!locked)
1426                         spin_lock(&request->rq_import->imp_lock);
1427                 list_del_init(&request->rq_mod_list);
1428                 list_del_init(&request->rq_replay_list);
1429                 if (!locked)
1430                         spin_unlock(&request->rq_import->imp_lock);
1431         }
1432         LASSERTF(list_empty(&request->rq_replay_list), "req %p\n", request);
1433
1434         if (atomic_read(&request->rq_refcount) != 0) {
1435                 DEBUG_REQ(D_ERROR, request,
1436                           "freeing request with nonzero refcount");
1437                 LBUG();
1438         }
1439
1440         if (request->rq_repbuf != NULL)
1441                 sptlrpc_cli_free_repbuf(request);
1442         if (request->rq_export != NULL) {
1443                 class_export_put(request->rq_export);
1444                 request->rq_export = NULL;
1445         }
1446         if (request->rq_import != NULL) {
1447                 class_import_put(request->rq_import);
1448                 request->rq_import = NULL;
1449         }
1450         if (request->rq_bulk != NULL)
1451                 ptlrpc_free_bulk(request->rq_bulk);
1452
1453         if (request->rq_reqbuf != NULL || request->rq_clrbuf != NULL)
1454                 sptlrpc_cli_free_reqbuf(request);
1455
1456         sptlrpc_req_put_ctx(request, !locked);
1457
1458         if (request->rq_pool)
1459                 __ptlrpc_free_req_to_pool(request);
1460         else
1461                 OBD_FREE(request, sizeof(*request));
1462         EXIT;
1463 }
1464
1465 void ptlrpc_free_req(struct ptlrpc_request *request)
1466 {
1467         __ptlrpc_free_req(request, 0);
1468 }
1469
1470 static int __ptlrpc_req_finished(struct ptlrpc_request *request, int locked);
1471 void ptlrpc_req_finished_with_imp_lock(struct ptlrpc_request *request)
1472 {
1473         LASSERT_SPIN_LOCKED(&request->rq_import->imp_lock);
1474         (void)__ptlrpc_req_finished(request, 1);
1475 }
1476
1477 static int __ptlrpc_req_finished(struct ptlrpc_request *request, int locked)
1478 {
1479         ENTRY;
1480         if (request == NULL)
1481                 RETURN(1);
1482
1483         if (request == LP_POISON ||
1484             request->rq_reqmsg == LP_POISON) {
1485                 CERROR("dereferencing freed request (bug 575)\n");
1486                 LBUG();
1487                 RETURN(1);
1488         }
1489
1490         DEBUG_REQ(D_INFO, request, "refcount now %u",
1491                   atomic_read(&request->rq_refcount) - 1);
1492
1493         if (atomic_dec_and_test(&request->rq_refcount)) {
1494                 __ptlrpc_free_req(request, locked);
1495                 RETURN(1);
1496         }
1497
1498         RETURN(0);
1499 }
1500
1501 void ptlrpc_req_finished(struct ptlrpc_request *request)
1502 {
1503         __ptlrpc_req_finished(request, 0);
1504 }
1505
1506 __u64 ptlrpc_req_xid(struct ptlrpc_request *request)
1507 {
1508         return request->rq_xid;
1509 }
1510 EXPORT_SYMBOL(ptlrpc_req_xid);
1511
1512 /* Disengage the client's reply buffer from the network
1513  * NB does _NOT_ unregister any client-side bulk.
1514  * IDEMPOTENT, but _not_ safe against concurrent callers.
1515  * The request owner (i.e. the thread doing the I/O) must call...
1516  */
1517 void ptlrpc_unregister_reply (struct ptlrpc_request *request)
1518 {
1519         int                rc;
1520         cfs_waitq_t       *wq;
1521         struct l_wait_info lwi;
1522
1523         LASSERT(!in_interrupt ());             /* might sleep */
1524
1525         if (!ptlrpc_client_receiving_reply(request))
1526                 return;
1527
1528         LNetMDUnlink (request->rq_reply_md_h);
1529
1530         /* We have to l_wait_event() whatever the result, to give liblustre
1531          * a chance to run reply_in_callback() */
1532
1533         if (request->rq_set != NULL)
1534                 wq = &request->rq_set->set_waitq;
1535         else
1536                 wq = &request->rq_reply_waitq;
1537
1538         for (;;) {
1539                 /* Network access will complete in finite time but the HUGE
1540                  * timeout lets us CWARN for visibility of sluggish NALs */
1541                 lwi = LWI_TIMEOUT(cfs_time_seconds(300), NULL, NULL);
1542                 rc = l_wait_event (*wq, !ptlrpc_client_receiving_reply(request), &lwi);
1543                 if (rc == 0)
1544                         return;
1545
1546                 LASSERT (rc == -ETIMEDOUT);
1547                 DEBUG_REQ(D_WARNING, request, "Unexpectedly long timeout");
1548         }
1549 }
1550
1551 /* caller must hold imp->imp_lock */
1552 void ptlrpc_free_committed(struct obd_import *imp)
1553 {
1554         struct list_head *tmp, *saved;
1555         struct ptlrpc_request *req;
1556         struct ptlrpc_request *last_req = NULL; /* temporary fire escape */
1557         ENTRY;
1558
1559         LASSERT(imp != NULL);
1560
1561         LASSERT_SPIN_LOCKED(&imp->imp_lock);
1562
1563
1564         if (imp->imp_peer_committed_transno == imp->imp_last_transno_checked &&
1565             imp->imp_generation == imp->imp_last_generation_checked) {
1566                 CDEBUG(D_RPCTRACE, "%s: skip recheck: last_committed "LPU64"\n",
1567                        imp->imp_obd->obd_name, imp->imp_peer_committed_transno);
1568                 return;
1569         }
1570
1571         CDEBUG(D_RPCTRACE, "%s: committing for last_committed "LPU64" gen %d\n",
1572                imp->imp_obd->obd_name, imp->imp_peer_committed_transno,
1573                imp->imp_generation);
1574         imp->imp_last_transno_checked = imp->imp_peer_committed_transno;
1575         imp->imp_last_generation_checked = imp->imp_generation;
1576
1577         list_for_each_safe(tmp, saved, &imp->imp_replay_list) {
1578                 req = list_entry(tmp, struct ptlrpc_request, rq_replay_list);
1579
1580                 /* XXX ok to remove when 1357 resolved - rread 05/29/03  */
1581                 LASSERT(req != last_req);
1582                 last_req = req;
1583
1584                 if (req->rq_import_generation < imp->imp_generation) {
1585                         DEBUG_REQ(D_RPCTRACE, req, "free request with old gen");
1586                         GOTO(free_req, 0);
1587                 }
1588
1589                 if (req->rq_replay) {
1590                         DEBUG_REQ(D_RPCTRACE, req, "keeping (FL_REPLAY)");
1591                         continue;
1592                 }
1593
1594                 /* not yet committed */
1595                 if (req->rq_transno > imp->imp_peer_committed_transno) {
1596                         DEBUG_REQ(D_RPCTRACE, req, "stopping search");
1597                         break;
1598                 }
1599
1600                 DEBUG_REQ(D_RPCTRACE, req, "commit (last_committed "LPU64")",
1601                           imp->imp_peer_committed_transno);
1602 free_req:
1603                 spin_lock(&req->rq_lock);
1604                 req->rq_replay = 0;
1605                 spin_unlock(&req->rq_lock);
1606                 if (req->rq_commit_cb != NULL)
1607                         req->rq_commit_cb(req);
1608                 list_del_init(&req->rq_replay_list);
1609                 __ptlrpc_req_finished(req, 1);
1610         }
1611
1612         EXIT;
1613         return;
1614 }
1615
1616 void ptlrpc_cleanup_client(struct obd_import *imp)
1617 {
1618         ENTRY;
1619         EXIT;
1620         return;
1621 }
1622
1623 void ptlrpc_resend_req(struct ptlrpc_request *req)
1624 {
1625         DEBUG_REQ(D_HA, req, "going to resend");
1626         lustre_msg_set_handle(req->rq_reqmsg, &(struct lustre_handle){ 0 });
1627         req->rq_status = -EAGAIN;
1628
1629         spin_lock(&req->rq_lock);
1630         req->rq_resend = 1;
1631         req->rq_net_err = 0;
1632         req->rq_timedout = 0;
1633         if (req->rq_bulk) {
1634                 __u64 old_xid = req->rq_xid;
1635
1636                 /* ensure previous bulk fails */
1637                 req->rq_xid = ptlrpc_next_xid();
1638                 CDEBUG(D_HA, "resend bulk old x"LPU64" new x"LPU64"\n",
1639                        old_xid, req->rq_xid);
1640         }
1641         ptlrpc_wake_client_req(req);
1642         spin_unlock(&req->rq_lock);
1643 }
1644
1645 /* XXX: this function and rq_status are currently unused */
1646 void ptlrpc_restart_req(struct ptlrpc_request *req)
1647 {
1648         DEBUG_REQ(D_HA, req, "restarting (possibly-)completed request");
1649         req->rq_status = -ERESTARTSYS;
1650
1651         spin_lock(&req->rq_lock);
1652         req->rq_restart = 1;
1653         req->rq_timedout = 0;
1654         ptlrpc_wake_client_req(req);
1655         spin_unlock(&req->rq_lock);
1656 }
1657
1658 static int expired_request(void *data)
1659 {
1660         struct ptlrpc_request *req = data;
1661         ENTRY;
1662
1663         /* some failure can suspend regular timeouts */
1664         if (ptlrpc_check_suspend())
1665                 RETURN(1);
1666
1667         RETURN(ptlrpc_expire_one_request(req));
1668 }
1669
1670 static void interrupted_request(void *data)
1671 {
1672         struct ptlrpc_request *req = data;
1673         DEBUG_REQ(D_HA, req, "request interrupted");
1674         spin_lock(&req->rq_lock);
1675         req->rq_intr = 1;
1676         spin_unlock(&req->rq_lock);
1677 }
1678
1679 struct ptlrpc_request *ptlrpc_request_addref(struct ptlrpc_request *req)
1680 {
1681         ENTRY;
1682         atomic_inc(&req->rq_refcount);
1683         RETURN(req);
1684 }
1685
1686 void ptlrpc_retain_replayable_request(struct ptlrpc_request *req,
1687                                       struct obd_import *imp)
1688 {
1689         struct list_head *tmp;
1690
1691         LASSERT_SPIN_LOCKED(&imp->imp_lock);
1692
1693         /* clear this for new requests that were resent as well
1694            as resent replayed requests. */
1695         lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT);
1696
1697         /* don't re-add requests that have been replayed */
1698         if (!list_empty(&req->rq_replay_list))
1699                 return;
1700
1701         lustre_msg_add_flags(req->rq_reqmsg, MSG_REPLAY);
1702
1703         LASSERT(imp->imp_replayable);
1704         /* Balanced in ptlrpc_free_committed, usually. */
1705         ptlrpc_request_addref(req);
1706         list_for_each_prev(tmp, &imp->imp_replay_list) {
1707                 struct ptlrpc_request *iter =
1708                         list_entry(tmp, struct ptlrpc_request, rq_replay_list);
1709
1710                 /* We may have duplicate transnos if we create and then
1711                  * open a file, or for closes retained if to match creating
1712                  * opens, so use req->rq_xid as a secondary key.
1713                  * (See bugs 684, 685, and 428.)
1714                  * XXX no longer needed, but all opens need transnos!
1715                  */
1716                 if (iter->rq_transno > req->rq_transno)
1717                         continue;
1718
1719                 if (iter->rq_transno == req->rq_transno) {
1720                         LASSERT(iter->rq_xid != req->rq_xid);
1721                         if (iter->rq_xid > req->rq_xid)
1722                                 continue;
1723                 }
1724
1725                 list_add(&req->rq_replay_list, &iter->rq_replay_list);
1726                 return;
1727         }
1728
1729         list_add_tail(&req->rq_replay_list, &imp->imp_replay_list);
1730 }
1731
1732 int ptlrpc_queue_wait(struct ptlrpc_request *req)
1733 {
1734         int rc = 0;
1735         int brc;
1736         struct l_wait_info lwi;
1737         struct obd_import *imp = req->rq_import;
1738         cfs_duration_t timeout = 0;
1739         ENTRY;
1740
1741         LASSERT(req->rq_set == NULL);
1742         LASSERT(!req->rq_receiving_reply);
1743         atomic_inc(&imp->imp_inflight);
1744
1745         /* for distributed debugging */
1746         lustre_msg_set_status(req->rq_reqmsg, cfs_curproc_pid());
1747         LASSERT(imp->imp_obd != NULL);
1748         CDEBUG(D_RPCTRACE, "Sending RPC pname:cluuid:pid:xid:nid:opc "
1749                "%s:%s:%d:"LPU64":%s:%d\n", cfs_curproc_comm(),
1750                imp->imp_obd->obd_uuid.uuid,
1751                lustre_msg_get_status(req->rq_reqmsg), req->rq_xid,
1752                libcfs_nid2str(imp->imp_connection->c_peer.nid),
1753                lustre_msg_get_opc(req->rq_reqmsg));
1754
1755         /* Mark phase here for a little debug help */
1756         req->rq_phase = RQ_PHASE_RPC;
1757
1758         spin_lock(&imp->imp_lock);
1759 restart:
1760         req->rq_import_generation = imp->imp_generation;
1761         if (ptlrpc_import_delay_req(imp, req, &rc)) {
1762                 list_del(&req->rq_list);
1763
1764                 list_add_tail(&req->rq_list, &imp->imp_delayed_list);
1765                 spin_unlock(&imp->imp_lock);
1766
1767                 DEBUG_REQ(D_HA, req, "\"%s\" waiting for recovery: (%s != %s)",
1768                           cfs_curproc_comm(),
1769                           ptlrpc_import_state_name(req->rq_send_state),
1770                           ptlrpc_import_state_name(imp->imp_state));
1771                 lwi = LWI_INTR(interrupted_request, req);
1772                 rc = l_wait_event(req->rq_reply_waitq,
1773                                   (req->rq_send_state == imp->imp_state ||
1774                                    req->rq_err || req->rq_intr),
1775                                   &lwi);
1776                 DEBUG_REQ(D_HA, req, "\"%s\" awake: (%s == %s or %d/%d == 1)",
1777                           cfs_curproc_comm(),
1778                           ptlrpc_import_state_name(imp->imp_state),
1779                           ptlrpc_import_state_name(req->rq_send_state),
1780                           req->rq_err, req->rq_intr);
1781
1782                 spin_lock(&imp->imp_lock);
1783                 list_del_init(&req->rq_list);
1784
1785                 if (req->rq_err) {
1786                         rc = -EIO;
1787                 }
1788                 else if (req->rq_intr) {
1789                         rc = -EINTR;
1790                 }
1791                 else if (req->rq_no_resend) {
1792                         spin_unlock(&imp->imp_lock);
1793                         GOTO(out, rc = -ETIMEDOUT);
1794                 }
1795                 else {
1796                         GOTO(restart, rc);
1797                 }
1798         }
1799
1800         if (rc != 0) {
1801                 list_del_init(&req->rq_list);
1802                 spin_unlock(&imp->imp_lock);
1803                 req->rq_status = rc; // XXX this ok?
1804                 GOTO(out, rc);
1805         }
1806
1807         if (req->rq_resend) {
1808                 lustre_msg_add_flags(req->rq_reqmsg, MSG_RESENT);
1809
1810                 if (req->rq_bulk != NULL) {
1811                         ptlrpc_unregister_bulk (req);
1812
1813                         /* bulk requests are supposed to be
1814                          * idempotent, so we are free to bump the xid
1815                          * here, which we need to do before
1816                          * registering the bulk again (bug 6371).
1817                          * print the old xid first for sanity.
1818                          */
1819                         DEBUG_REQ(D_HA, req, "bumping xid for bulk: ");
1820                         req->rq_xid = ptlrpc_next_xid();
1821                 }
1822
1823                 DEBUG_REQ(D_HA, req, "resending: ");
1824         }
1825
1826         /* XXX this is the same as ptlrpc_set_wait */
1827         LASSERT(list_empty(&req->rq_list));
1828         list_add_tail(&req->rq_list, &imp->imp_sending_list);
1829         spin_unlock(&imp->imp_lock);
1830
1831         rc = sptlrpc_req_refresh_ctx(req, 0);
1832         if (rc) {
1833                 if (req->rq_err) {
1834                         /* we got fatal ctx refresh error, directly jump out
1835                          * thus we can pass back the actual error code.
1836                          */
1837                         spin_lock(&imp->imp_lock);
1838                         list_del_init(&req->rq_list);
1839                         spin_unlock(&imp->imp_lock);
1840
1841                         CERROR("Failed to refresh ctx of req %p: %d\n", req, rc);
1842                         GOTO(out, rc);
1843                 }
1844                 /* simulating we got error during send rpc */
1845                 goto after_send;
1846         }
1847
1848         rc = ptl_send_rpc(req, 0);
1849         if (rc) {
1850                 DEBUG_REQ(D_HA, req, "send failed (%d); recovering", rc);
1851                 timeout = CFS_TICK;
1852         } else {
1853                 timeout = cfs_timeout_cap(cfs_time_seconds(req->rq_timeout));
1854                 DEBUG_REQ(D_NET, req,
1855                           "-- sleeping for "CFS_DURATION_T" jiffies", timeout);
1856         }
1857 repeat:
1858         lwi = LWI_TIMEOUT_INTR(timeout, expired_request, interrupted_request,
1859                                req);
1860         rc = l_wait_event(req->rq_reply_waitq, ptlrpc_check_reply(req), &lwi);
1861         if (rc == -ETIMEDOUT && ptlrpc_check_and_wait_suspend(req))
1862                 goto repeat;
1863
1864 after_send:
1865         CDEBUG(D_RPCTRACE, "Completed RPC pname:cluuid:pid:xid:nid:opc "
1866                "%s:%s:%d:"LPU64":%s:%d\n", cfs_curproc_comm(),
1867                imp->imp_obd->obd_uuid.uuid,
1868                lustre_msg_get_status(req->rq_reqmsg), req->rq_xid,
1869                libcfs_nid2str(imp->imp_connection->c_peer.nid),
1870                lustre_msg_get_opc(req->rq_reqmsg));
1871
1872         spin_lock(&imp->imp_lock);
1873         list_del_init(&req->rq_list);
1874         spin_unlock(&imp->imp_lock);
1875
1876         /* If the reply was received normally, this just grabs the spinlock
1877          * (ensuring the reply callback has returned), sees that
1878          * req->rq_receiving_reply is clear and returns. */
1879         ptlrpc_unregister_reply (req);
1880
1881         if (req->rq_err)
1882                 GOTO(out, rc = -EIO);
1883
1884         /* Resend if we need to, unless we were interrupted. */
1885         if (req->rq_resend && !req->rq_intr) {
1886                 /* ...unless we were specifically told otherwise. */
1887                 if (req->rq_no_resend)
1888                         GOTO(out, rc = -ETIMEDOUT);
1889                 spin_lock(&imp->imp_lock);
1890                 goto restart;
1891         }
1892
1893         if (req->rq_intr) {
1894                 /* Should only be interrupted if we timed out. */
1895                 if (!req->rq_timedout)
1896                         DEBUG_REQ(D_ERROR, req,
1897                                   "rq_intr set but rq_timedout not");
1898                 GOTO(out, rc = -EINTR);
1899         }
1900
1901         if (req->rq_timedout) {                 /* non-recoverable timeout */
1902                 GOTO(out, rc = -ETIMEDOUT);
1903         }
1904
1905         if (!req->rq_replied) {
1906                 /* How can this be? -eeb */
1907                 DEBUG_REQ(D_ERROR, req, "!rq_replied: ");
1908                 LBUG();
1909                 GOTO(out, rc = req->rq_status);
1910         }
1911
1912         rc = after_reply(req);
1913         /* NB may return +ve success rc */
1914         if (req->rq_resend) {
1915                 spin_lock(&imp->imp_lock);
1916                 goto restart;
1917         }
1918
1919  out:
1920         if (req->rq_bulk != NULL) {
1921                 if (rc >= 0) {
1922                         /* success so far.  Note that anything going wrong
1923                          * with bulk now, is EXTREMELY strange, since the
1924                          * server must have believed that the bulk
1925                          * tranferred OK before she replied with success to
1926                          * me. */
1927                         lwi = LWI_TIMEOUT(timeout, NULL, NULL);
1928                         brc = l_wait_event(req->rq_reply_waitq,
1929                                            !ptlrpc_bulk_active(req->rq_bulk),
1930                                            &lwi);
1931                         LASSERT(brc == 0 || brc == -ETIMEDOUT);
1932                         if (brc != 0) {
1933                                 LASSERT(brc == -ETIMEDOUT);
1934                                 DEBUG_REQ(D_ERROR, req, "bulk timed out");
1935                                 rc = brc;
1936                         } else if (!req->rq_bulk->bd_success) {
1937                                 DEBUG_REQ(D_ERROR, req, "bulk transfer failed");
1938                                 rc = -EIO;
1939                         }
1940                 }
1941                 if (rc < 0)
1942                         ptlrpc_unregister_bulk (req);
1943         }
1944
1945         LASSERT(!req->rq_receiving_reply);
1946         req->rq_phase = RQ_PHASE_INTERPRET;
1947
1948         atomic_dec(&imp->imp_inflight);
1949         cfs_waitq_signal(&imp->imp_recovery_waitq);
1950         RETURN(rc);
1951 }
1952
1953 struct ptlrpc_replay_async_args {
1954         int praa_old_state;
1955         int praa_old_status;
1956 };
1957
1958 static int ptlrpc_replay_interpret(struct ptlrpc_request *req,
1959                                     void * data, int rc)
1960 {
1961         struct ptlrpc_replay_async_args *aa = data;
1962         struct obd_import *imp = req->rq_import;
1963
1964         ENTRY;
1965         atomic_dec(&imp->imp_replay_inflight);
1966
1967         if (!req->rq_replied) {
1968                 CERROR("request replay timed out, restarting recovery\n");
1969                 GOTO(out, rc = -ETIMEDOUT);
1970         }
1971
1972         if (lustre_msg_get_type(req->rq_repmsg) == PTL_RPC_MSG_ERR &&
1973             (lustre_msg_get_status(req->rq_repmsg) == -ENOTCONN ||
1974              lustre_msg_get_status(req->rq_repmsg) == -ENODEV))
1975                 GOTO(out, rc = lustre_msg_get_status(req->rq_repmsg));
1976
1977         /* The transno had better not change over replay. */
1978         LASSERT(lustre_msg_get_transno(req->rq_reqmsg) ==
1979                 lustre_msg_get_transno(req->rq_repmsg));
1980
1981         DEBUG_REQ(D_HA, req, "got rep");
1982
1983         /* let the callback do fixups, possibly including in the request */
1984         if (req->rq_replay_cb)
1985                 req->rq_replay_cb(req);
1986
1987         if (req->rq_replied &&
1988             lustre_msg_get_status(req->rq_repmsg) != aa->praa_old_status) {
1989                 DEBUG_REQ(D_ERROR, req, "status %d, old was %d",
1990                           lustre_msg_get_status(req->rq_repmsg),
1991                           aa->praa_old_status);
1992         } else {
1993                 /* Put it back for re-replay. */
1994                 lustre_msg_set_status(req->rq_repmsg, aa->praa_old_status);
1995         }
1996
1997         /*
1998          * Errors while replay can set transno to 0, but
1999          * imp_last_replay_transno shouldn't be set to 0 anyway
2000          */
2001         if (req->rq_transno > 0) {
2002                 spin_lock(&imp->imp_lock);
2003                 LASSERT(req->rq_transno <= imp->imp_last_replay_transno);
2004                 imp->imp_last_replay_transno = req->rq_transno;
2005                 spin_unlock(&imp->imp_lock);
2006         } else
2007                 CERROR("Transno is 0 during replay!\n");
2008         /* continue with recovery */
2009         rc = ptlrpc_import_recovery_state_machine(imp);
2010  out:
2011         req->rq_send_state = aa->praa_old_state;
2012
2013         if (rc != 0)
2014                 /* this replay failed, so restart recovery */
2015                 ptlrpc_connect_import(imp, NULL);
2016
2017         RETURN(rc);
2018 }
2019
2020 int ptlrpc_replay_req(struct ptlrpc_request *req)
2021 {
2022         struct ptlrpc_replay_async_args *aa;
2023         ENTRY;
2024
2025         LASSERT(req->rq_import->imp_state == LUSTRE_IMP_REPLAY);
2026
2027         /* Not handling automatic bulk replay yet (or ever?) */
2028         LASSERT(req->rq_bulk == NULL);
2029
2030         DEBUG_REQ(D_HA, req, "REPLAY");
2031
2032         LASSERT (sizeof (*aa) <= sizeof (req->rq_async_args));
2033         aa = (struct ptlrpc_replay_async_args *)&req->rq_async_args;
2034         memset(aa, 0, sizeof *aa);
2035
2036         /* Prepare request to be resent with ptlrpcd */
2037         aa->praa_old_state = req->rq_send_state;
2038         req->rq_send_state = LUSTRE_IMP_REPLAY;
2039         req->rq_phase = RQ_PHASE_NEW;
2040         aa->praa_old_status = lustre_msg_get_status(req->rq_repmsg);
2041         req->rq_status = 0;
2042
2043         req->rq_interpret_reply = ptlrpc_replay_interpret;
2044         atomic_inc(&req->rq_import->imp_replay_inflight);
2045         ptlrpc_request_addref(req); /* ptlrpcd needs a ref */
2046
2047         ptlrpcd_add_req(req);
2048         RETURN(0);
2049 }
2050
2051 void ptlrpc_abort_inflight(struct obd_import *imp)
2052 {
2053         struct list_head *tmp, *n;
2054         ENTRY;
2055
2056         /* Make sure that no new requests get processed for this import.
2057          * ptlrpc_{queue,set}_wait must (and does) hold imp_lock while testing
2058          * this flag and then putting requests on sending_list or delayed_list.
2059          */
2060         spin_lock(&imp->imp_lock);
2061
2062         /* XXX locking?  Maybe we should remove each request with the list
2063          * locked?  Also, how do we know if the requests on the list are
2064          * being freed at this time?
2065          */
2066         list_for_each_safe(tmp, n, &imp->imp_sending_list) {
2067                 struct ptlrpc_request *req =
2068                         list_entry(tmp, struct ptlrpc_request, rq_list);
2069
2070                 DEBUG_REQ(D_RPCTRACE, req, "inflight");
2071
2072                 spin_lock (&req->rq_lock);
2073                 if (req->rq_import_generation < imp->imp_generation) {
2074                         req->rq_err = 1;
2075                         ptlrpc_wake_client_req(req);
2076                 }
2077                 spin_unlock (&req->rq_lock);
2078         }
2079
2080         list_for_each_safe(tmp, n, &imp->imp_delayed_list) {
2081                 struct ptlrpc_request *req =
2082                         list_entry(tmp, struct ptlrpc_request, rq_list);
2083
2084                 DEBUG_REQ(D_RPCTRACE, req, "aborting waiting req");
2085
2086                 spin_lock (&req->rq_lock);
2087                 if (req->rq_import_generation < imp->imp_generation) {
2088                         req->rq_err = 1;
2089                         ptlrpc_wake_client_req(req);
2090                 }
2091                 spin_unlock (&req->rq_lock);
2092         }
2093
2094         /* Last chance to free reqs left on the replay list, but we
2095          * will still leak reqs that haven't committed.  */
2096         if (imp->imp_replayable)
2097                 ptlrpc_free_committed(imp);
2098
2099         spin_unlock(&imp->imp_lock);
2100
2101         EXIT;
2102 }
2103
2104 static __u64 ptlrpc_last_xid = 0;
2105 spinlock_t ptlrpc_last_xid_lock;
2106
2107 __u64 ptlrpc_next_xid(void)
2108 {
2109         __u64 tmp;
2110         spin_lock(&ptlrpc_last_xid_lock);
2111         tmp = ++ptlrpc_last_xid;
2112         spin_unlock(&ptlrpc_last_xid_lock);
2113         return tmp;
2114 }
2115
2116 __u64 ptlrpc_sample_next_xid(void)
2117 {
2118         __u64 tmp;
2119         spin_lock(&ptlrpc_last_xid_lock);
2120         tmp = ptlrpc_last_xid + 1;
2121         spin_unlock(&ptlrpc_last_xid_lock);
2122         return tmp;
2123 }
2124 EXPORT_SYMBOL(ptlrpc_sample_next_xid);