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