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 (!imp->imp_deactive)
 515                         DEBUG_REQ(D_ERROR, req, "IMP_INVALID");
 516                 *status = -ESHUTDOWN; /* bz 12940 */
 517         } else if (req->rq_import_generation != imp->imp_generation) {
 518                 DEBUG_REQ(D_ERROR, req, "req wrong generation:");
 519                 *status = -EIO;
 520         } else if (req->rq_send_state != imp->imp_state) {
 521                 if (imp->imp_obd->obd_no_recov || imp->imp_dlm_fake ||
 522                     req->rq_no_delay)
 523                         *status = -EWOULDBLOCK;
 524                 else
 525                         delay = 1;
 526         }
 527
 528         RETURN(delay);
 529 }
 530
 531 static int ptlrpc_check_reply(struct ptlrpc_request *req)
 532 {
 533         int rc = 0;
 534         ENTRY;
 535
 536         /* serialise with network callback */
 537         spin_lock(&req->rq_lock);
 538
 539         if (req->rq_replied)
 540                 GOTO(out, rc = 1);
 541
 542         if (req->rq_net_err && !req->rq_timedout) {
 543                 spin_unlock(&req->rq_lock);
 544                 rc = ptlrpc_expire_one_request(req);
 545                 spin_lock(&req->rq_lock);
 546                 GOTO(out, rc);
 547         }
 548
 549         if (req->rq_err)
 550                 GOTO(out, rc = 1);
 551
 552         if (req->rq_resend)
 553                 GOTO(out, rc = 1);
 554
 555         if (req->rq_restart)
 556                 GOTO(out, rc = 1);
 557         EXIT;
 558  out:
 559         spin_unlock(&req->rq_lock);
 560         DEBUG_REQ(D_NET, req, "rc = %d for", rc);
 561         return rc;
 562 }
 563
 564 static int ptlrpc_check_status(struct ptlrpc_request *req)
 565 {
 566         int err;
 567         ENTRY;
 568
 569         err = lustre_msg_get_status(req->rq_repmsg);
 570         if (lustre_msg_get_type(req->rq_repmsg) == PTL_RPC_MSG_ERR) {
 571                 struct obd_export *exp = req->rq_export;
 572                 __u32 opc = lustre_msg_get_opc(req->rq_reqmsg);
 573
 574                 LCONSOLE_ERROR_MSG(0x011, "an error ocurred while communicating"
 575                                    " with %s The %s operation failed with %d\n",
 576                                    exp ? obd_export_nid2str(exp) : "(no nid)",
 577                                    ll_opcode2str(opc), 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", 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", 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)",
 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 (ptlrpc_recoverable_error(rc)) {
 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                         req->rq_status = -EIO;
 909                         req->rq_phase = RQ_PHASE_INTERPRET;
 910                         GOTO(interpret, req->rq_status);
 911                 }
 912
 913                 req->rq_phase = RQ_PHASE_INTERPRET;
 914
 915         interpret:
 916                 LASSERT(req->rq_phase == RQ_PHASE_INTERPRET);
 917                 LASSERT(!req->rq_receiving_reply);
 918
 919                 ptlrpc_unregister_reply(req);
 920                 if (req->rq_bulk != NULL)
 921                         ptlrpc_unregister_bulk (req);
 922
 923                 req->rq_phase = RQ_PHASE_COMPLETE;
 924
 925                 if (req->rq_interpret_reply != NULL) {
 926                         int (*interpreter)(struct ptlrpc_request *,void *,int) =
 927                                 req->rq_interpret_reply;
 928                         req->rq_status = interpreter(req, &req->rq_async_args,
 929                                                      req->rq_status);
 930                 }
 931
 932                 CDEBUG(D_RPCTRACE, "Completed RPC pname:cluuid:pid:xid:nid:"
 933                        "opc %s:%s:%d:"LPU64":%s:%d\n", cfs_curproc_comm(),
 934                        imp->imp_obd->obd_uuid.uuid,
 935                        lustre_msg_get_status(req->rq_reqmsg), req->rq_xid,
 936                        libcfs_nid2str(imp->imp_connection->c_peer.nid),
 937                        lustre_msg_get_opc(req->rq_reqmsg));
 938
 939                 set->set_remaining--;
 940
 941                 atomic_dec(&imp->imp_inflight);
 942                 cfs_waitq_signal(&imp->imp_recovery_waitq);
 943         }
 944
 945         /* If we hit an error, we want to recover promptly. */
 946         RETURN(set->set_remaining == 0 || force_timer_recalc);
 947 }
 948
 949 int ptlrpc_expire_one_request(struct ptlrpc_request *req)
 950 {
 951         struct obd_import *imp = req->rq_import;
 952         int rc = 0;
 953         ENTRY;
 954
 955         DEBUG_REQ(D_ERROR|D_NETERROR, req, "%s (sent at %lu, %lus ago)",
 956                   req->rq_net_err ? "network error" : "timeout",
 957                   (long)req->rq_sent, CURRENT_SECONDS - req->rq_sent);
 958
 959         if (imp != NULL && obd_debug_peer_on_timeout)
 960                 LNetCtl(IOC_LIBCFS_DEBUG_PEER, &imp->imp_connection->c_peer);
 961
 962         spin_lock(&req->rq_lock);
 963         req->rq_timedout = 1;
 964         spin_unlock(&req->rq_lock);
 965
 966         ptlrpc_unregister_reply (req);
 967
 968         if (obd_dump_on_timeout)
 969                 libcfs_debug_dumplog();
 970
 971         if (req->rq_bulk != NULL)
 972                 ptlrpc_unregister_bulk (req);
 973
 974         if (imp == NULL) {
 975                 DEBUG_REQ(D_HA, req, "NULL import: already cleaned up?");
 976                 RETURN(1);
 977         }
 978
 979         /* The DLM server doesn't want recovery run on its imports. */
 980         if (imp->imp_dlm_fake)
 981                 RETURN(1);
 982
 983         /* If this request is for recovery or other primordial tasks,
 984          * then error it out here. */
 985         if (req->rq_send_state != LUSTRE_IMP_FULL ||
 986             imp->imp_obd->obd_no_recov) {
 987                 spin_lock(&req->rq_lock);
 988                 req->rq_status = -ETIMEDOUT;
 989                 req->rq_err = 1;
 990                 spin_unlock(&req->rq_lock);
 991                 RETURN(1);
 992         }
 993
 994         /* if request can't be resend we can't wait answer after timeout */
 995         if (req->rq_no_resend) {
 996                 DEBUG_REQ(D_RPCTRACE, req, "TIMEOUT-NORESEND:");
 997                 rc = 1;
 998         }
 999
1000         ptlrpc_fail_import(imp, lustre_msg_get_conn_cnt(req->rq_reqmsg));
1001
1002         RETURN(rc);
1003 }
1004
1005 int ptlrpc_expired_set(void *data)
1006 {
1007         struct ptlrpc_request_set *set = data;
1008         struct list_head          *tmp;
1009         time_t                     now = CURRENT_SECONDS;
1010         ENTRY;
1011
1012         LASSERT(set != NULL);
1013
1014         /* A timeout expired; see which reqs it applies to... */
1015         list_for_each (tmp, &set->set_requests) {
1016                 struct ptlrpc_request *req =
1017                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
1018
1019                 /* request in-flight? */
1020                 if (!((req->rq_phase == RQ_PHASE_RPC && !req->rq_waiting &&
1021                        !req->rq_resend) ||
1022                       (req->rq_phase == RQ_PHASE_BULK)))
1023                         continue;
1024
1025                 if (req->rq_timedout ||           /* already dealt with */
1026                     req->rq_sent + req->rq_timeout > now) /* not expired */
1027                         continue;
1028
1029                 /* deal with this guy */
1030                 ptlrpc_expire_one_request (req);
1031         }
1032
1033         /* When waiting for a whole set, we always to break out of the
1034          * sleep so we can recalculate the timeout, or enable interrupts
1035          * iff everyone's timed out.
1036          */
1037         RETURN(1);
1038 }
1039
1040 void ptlrpc_mark_interrupted(struct ptlrpc_request *req)
1041 {
1042         spin_lock(&req->rq_lock);
1043         req->rq_intr = 1;
1044         spin_unlock(&req->rq_lock);
1045 }
1046
1047 void ptlrpc_interrupted_set(void *data)
1048 {
1049         struct ptlrpc_request_set *set = data;
1050         struct list_head *tmp;
1051
1052         LASSERT(set != NULL);
1053         CERROR("INTERRUPTED SET %p\n", set);
1054
1055         list_for_each(tmp, &set->set_requests) {
1056                 struct ptlrpc_request *req =
1057                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
1058
1059                 if (req->rq_phase != RQ_PHASE_RPC)
1060                         continue;
1061
1062                 ptlrpc_mark_interrupted(req);
1063         }
1064 }
1065
1066 int ptlrpc_set_next_timeout(struct ptlrpc_request_set *set)
1067 {
1068         struct list_head      *tmp;
1069         time_t                 now = CURRENT_SECONDS;
1070         time_t                 deadline;
1071         int                    timeout = 0;
1072         struct ptlrpc_request *req;
1073         ENTRY;
1074
1075         SIGNAL_MASK_ASSERT(); /* XXX BUG 1511 */
1076
1077         list_for_each(tmp, &set->set_requests) {
1078                 req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
1079
1080                 /* request in-flight? */
1081                 if (!((req->rq_phase == RQ_PHASE_RPC && !req->rq_waiting) ||
1082                       (req->rq_phase == RQ_PHASE_BULK)))
1083                         continue;
1084
1085                 if (req->rq_timedout)   /* already timed out */
1086                         continue;
1087
1088                 deadline = req->rq_sent + req->rq_timeout;
1089                 if (deadline <= now)    /* actually expired already */
1090                         timeout = 1;    /* ASAP */
1091                 else if (timeout == 0 || timeout > deadline - now)
1092                         timeout = deadline - now;
1093         }
1094         RETURN(timeout);
1095 }
1096
1097 int ptlrpc_set_wait(struct ptlrpc_request_set *set)
1098 {
1099         struct list_head      *tmp;
1100         struct ptlrpc_request *req;
1101         struct l_wait_info     lwi;
1102         int                    rc, timeout;
1103         ENTRY;
1104
1105         if (list_empty(&set->set_requests))
1106                 RETURN(0);
1107
1108         list_for_each(tmp, &set->set_requests) {
1109                 req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
1110                 if (req->rq_phase == RQ_PHASE_NEW)
1111                         (void)ptlrpc_send_new_req(req);
1112         }
1113
1114         do {
1115                 timeout = ptlrpc_set_next_timeout(set);
1116
1117                 /* wait until all complete, interrupted, or an in-flight
1118                  * req times out */
1119                 CDEBUG(D_RPCTRACE, "set %p going to sleep for %d seconds\n",
1120                        set, timeout);
1121                 lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(timeout ? timeout : 1),
1122                                        ptlrpc_expired_set,
1123                                        ptlrpc_interrupted_set, set);
1124                 rc = l_wait_event(set->set_waitq, ptlrpc_check_set(set), &lwi);
1125
1126                 LASSERT(rc == 0 || rc == -EINTR || rc == -ETIMEDOUT);
1127
1128                 /* -EINTR => all requests have been flagged rq_intr so next
1129                  * check completes.
1130                  * -ETIMEOUTD => someone timed out.  When all reqs have
1131                  * timed out, signals are enabled allowing completion with
1132                  * EINTR.
1133                  * I don't really care if we go once more round the loop in
1134                  * the error cases -eeb. */
1135         } while (rc != 0 || set->set_remaining != 0);
1136
1137         LASSERT(set->set_remaining == 0);
1138
1139         rc = 0;
1140         list_for_each(tmp, &set->set_requests) {
1141                 req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
1142
1143                 LASSERT(req->rq_phase == RQ_PHASE_COMPLETE);
1144                 if (req->rq_status != 0)
1145                         rc = req->rq_status;
1146         }
1147
1148         if (set->set_interpret != NULL) {
1149                 int (*interpreter)(struct ptlrpc_request_set *set,void *,int) =
1150                         set->set_interpret;
1151                 rc = interpreter (set, set->set_arg, rc);
1152         }
1153
1154         RETURN(rc);
1155 }
1156
1157 static void __ptlrpc_free_req_to_pool(struct ptlrpc_request *request)
1158 {
1159         struct ptlrpc_request_pool *pool = request->rq_pool;
1160
1161         spin_lock(&pool->prp_lock);
1162         list_add_tail(&request->rq_list, &pool->prp_req_list);
1163         spin_unlock(&pool->prp_lock);
1164 }
1165
1166 static void __ptlrpc_free_req(struct ptlrpc_request *request, int locked)
1167 {
1168         ENTRY;
1169         if (request == NULL) {
1170                 EXIT;
1171                 return;
1172         }
1173
1174         LASSERTF(!request->rq_receiving_reply, "req %p\n", request);
1175         LASSERTF(request->rq_rqbd == NULL, "req %p\n",request);/* client-side */
1176         LASSERTF(list_empty(&request->rq_list), "req %p\n", request);
1177         LASSERTF(list_empty(&request->rq_set_chain), "req %p\n", request);
1178
1179         /* We must take it off the imp_replay_list first.  Otherwise, we'll set
1180          * request->rq_reqmsg to NULL while osc_close is dereferencing it. */
1181         if (request->rq_import != NULL) {
1182                 if (!locked)
1183                         spin_lock(&request->rq_import->imp_lock);
1184                 list_del_init(&request->rq_replay_list);
1185                 if (!locked)
1186                         spin_unlock(&request->rq_import->imp_lock);
1187         }
1188         LASSERTF(list_empty(&request->rq_replay_list), "req %p\n", request);
1189
1190         if (atomic_read(&request->rq_refcount) != 0) {
1191                 DEBUG_REQ(D_ERROR, request,
1192                           "freeing request with nonzero refcount");
1193                 LBUG();
1194         }
1195
1196         if (request->rq_repmsg != NULL) {
1197                 OBD_FREE(request->rq_repmsg, request->rq_replen);
1198                 request->rq_repmsg = NULL;
1199         }
1200         if (request->rq_export != NULL) {
1201                 class_export_put(request->rq_export);
1202                 request->rq_export = NULL;
1203         }
1204         if (request->rq_import != NULL) {
1205                 class_import_put(request->rq_import);
1206                 request->rq_import = NULL;
1207         }
1208         if (request->rq_bulk != NULL)
1209                 ptlrpc_free_bulk(request->rq_bulk);
1210
1211         if (request->rq_pool) {
1212                 __ptlrpc_free_req_to_pool(request);
1213         } else {
1214                 if (request->rq_reqmsg != NULL) {
1215                         OBD_FREE(request->rq_reqmsg, request->rq_reqlen);
1216                         request->rq_reqmsg = NULL;
1217                 }
1218                 OBD_FREE(request, sizeof(*request));
1219         }
1220         EXIT;
1221 }
1222
1223 void ptlrpc_free_req(struct ptlrpc_request *request)
1224 {
1225         __ptlrpc_free_req(request, 0);
1226 }
1227
1228 static int __ptlrpc_req_finished(struct ptlrpc_request *request, int locked);
1229 void ptlrpc_req_finished_with_imp_lock(struct ptlrpc_request *request)
1230 {
1231         LASSERT_SPIN_LOCKED(&request->rq_import->imp_lock);
1232         (void)__ptlrpc_req_finished(request, 1);
1233 }
1234
1235 static int __ptlrpc_req_finished(struct ptlrpc_request *request, int locked)
1236 {
1237         ENTRY;
1238         if (request == NULL)
1239                 RETURN(1);
1240
1241         if (request == LP_POISON ||
1242             request->rq_reqmsg == LP_POISON) {
1243                 CERROR("dereferencing freed request (bug 575)\n");
1244                 LBUG();
1245                 RETURN(1);
1246         }
1247
1248         DEBUG_REQ(D_INFO, request, "refcount now %u",
1249                   atomic_read(&request->rq_refcount) - 1);
1250
1251         if (atomic_dec_and_test(&request->rq_refcount)) {
1252                 __ptlrpc_free_req(request, locked);
1253                 RETURN(1);
1254         }
1255
1256         RETURN(0);
1257 }
1258
1259 void ptlrpc_req_finished(struct ptlrpc_request *request)
1260 {
1261         __ptlrpc_req_finished(request, 0);
1262 }
1263
1264 __u64 ptlrpc_req_xid(struct ptlrpc_request *request)
1265 {
1266         return request->rq_xid;
1267 }
1268 EXPORT_SYMBOL(ptlrpc_req_xid);
1269
1270 /* Disengage the client's reply buffer from the network
1271  * NB does _NOT_ unregister any client-side bulk.
1272  * IDEMPOTENT, but _not_ safe against concurrent callers.
1273  * The request owner (i.e. the thread doing the I/O) must call...
1274  */
1275 void ptlrpc_unregister_reply (struct ptlrpc_request *request)
1276 {
1277         int                rc;
1278         cfs_waitq_t       *wq;
1279         struct l_wait_info lwi;
1280         ENTRY;
1281
1282         LASSERT(!in_interrupt ());             /* might sleep */
1283
1284         if (!ptlrpc_client_receiving_reply(request))
1285                 return;
1286
1287         LNetMDUnlink (request->rq_reply_md_h);
1288
1289         /* We have to l_wait_event() whatever the result, to give liblustre
1290          * a chance to run reply_in_callback() */
1291
1292         if (request->rq_set != NULL)
1293                 wq = &request->rq_set->set_waitq;
1294         else
1295                 wq = &request->rq_reply_waitq;
1296
1297         for (;;) {
1298                 /* Network access will complete in finite time but the HUGE
1299                  * timeout lets us CWARN for visibility of sluggish NALs */
1300                 lwi = LWI_TIMEOUT(cfs_time_seconds(300), NULL, NULL);
1301                 rc = l_wait_event (*wq, !ptlrpc_client_receiving_reply(request), &lwi);
1302                 if (rc == 0)
1303                         return;
1304
1305                 LASSERT (rc == -ETIMEDOUT);
1306                 DEBUG_REQ(D_WARNING, request, "Unexpectedly long timeout");
1307         }
1308         EXIT;
1309 }
1310
1311 /* caller must hold imp->imp_lock */
1312 void ptlrpc_free_committed(struct obd_import *imp)
1313 {
1314         struct list_head *tmp, *saved;
1315         struct ptlrpc_request *req;
1316         struct ptlrpc_request *last_req = NULL; /* temporary fire escape */
1317         ENTRY;
1318
1319         LASSERT(imp != NULL);
1320
1321         LASSERT_SPIN_LOCKED(&imp->imp_lock);
1322
1323
1324         if (imp->imp_peer_committed_transno == imp->imp_last_transno_checked &&
1325             imp->imp_generation == imp->imp_last_generation_checked) {
1326                 CDEBUG(D_RPCTRACE, "%s: skip recheck: last_committed "LPU64"\n",
1327                        imp->imp_obd->obd_name, imp->imp_peer_committed_transno);
1328                 return;
1329         }
1330
1331         CDEBUG(D_RPCTRACE, "%s: committing for last_committed "LPU64" gen %d\n",
1332                imp->imp_obd->obd_name, imp->imp_peer_committed_transno,
1333                imp->imp_generation);
1334         imp->imp_last_transno_checked = imp->imp_peer_committed_transno;
1335         imp->imp_last_generation_checked = imp->imp_generation;
1336
1337         list_for_each_safe(tmp, saved, &imp->imp_replay_list) {
1338                 req = list_entry(tmp, struct ptlrpc_request, rq_replay_list);
1339
1340                 /* XXX ok to remove when 1357 resolved - rread 05/29/03  */
1341                 LASSERT(req != last_req);
1342                 last_req = req;
1343
1344                 if (req->rq_import_generation < imp->imp_generation) {
1345                         DEBUG_REQ(D_RPCTRACE, req, "free request with old gen");
1346                         GOTO(free_req, 0);
1347                 }
1348
1349                 if (req->rq_replay) {
1350                         DEBUG_REQ(D_RPCTRACE, req, "keeping (FL_REPLAY)");
1351                         continue;
1352                 }
1353
1354                 /* not yet committed */
1355                 if (req->rq_transno > imp->imp_peer_committed_transno) {
1356                         DEBUG_REQ(D_RPCTRACE, req, "stopping search");
1357                         break;
1358                 }
1359
1360                 DEBUG_REQ(D_RPCTRACE, req, "commit (last_committed "LPU64")",
1361                           imp->imp_peer_committed_transno);
1362 free_req:
1363                 spin_lock(&req->rq_lock);
1364                 req->rq_replay = 0;
1365                 spin_unlock(&req->rq_lock);
1366                 if (req->rq_commit_cb != NULL)
1367                         req->rq_commit_cb(req);
1368                 list_del_init(&req->rq_replay_list);
1369                 __ptlrpc_req_finished(req, 1);
1370         }
1371
1372         EXIT;
1373         return;
1374 }
1375
1376 void ptlrpc_cleanup_client(struct obd_import *imp)
1377 {
1378         ENTRY;
1379         EXIT;
1380         return;
1381 }
1382
1383 void ptlrpc_resend_req(struct ptlrpc_request *req)
1384 {
1385         DEBUG_REQ(D_HA, req, "going to resend");
1386         lustre_msg_set_handle(req->rq_reqmsg, &(struct lustre_handle){ 0 });
1387         req->rq_status = -EAGAIN;
1388
1389         spin_lock(&req->rq_lock);
1390         req->rq_resend = 1;
1391         req->rq_net_err = 0;
1392         req->rq_timedout = 0;
1393         if (req->rq_bulk) {
1394                 __u64 old_xid = req->rq_xid;
1395
1396                 /* ensure previous bulk fails */
1397                 req->rq_xid = ptlrpc_next_xid();
1398                 CDEBUG(D_HA, "resend bulk old x"LPU64" new x"LPU64"\n",
1399                        old_xid, req->rq_xid);
1400         }
1401         ptlrpc_wake_client_req(req);
1402         spin_unlock(&req->rq_lock);
1403 }
1404
1405 /* XXX: this function and rq_status are currently unused */
1406 void ptlrpc_restart_req(struct ptlrpc_request *req)
1407 {
1408         DEBUG_REQ(D_HA, req, "restarting (possibly-)completed request");
1409         req->rq_status = -ERESTARTSYS;
1410
1411         spin_lock(&req->rq_lock);
1412         req->rq_restart = 1;
1413         req->rq_timedout = 0;
1414         ptlrpc_wake_client_req(req);
1415         spin_unlock(&req->rq_lock);
1416 }
1417
1418 static int expired_request(void *data)
1419 {
1420         struct ptlrpc_request *req = data;
1421         ENTRY;
1422
1423         RETURN(ptlrpc_expire_one_request(req));
1424 }
1425
1426 static void interrupted_request(void *data)
1427 {
1428         struct ptlrpc_request *req = data;
1429         DEBUG_REQ(D_HA, req, "request interrupted");
1430         spin_lock(&req->rq_lock);
1431         req->rq_intr = 1;
1432         spin_unlock(&req->rq_lock);
1433 }
1434
1435 struct ptlrpc_request *ptlrpc_request_addref(struct ptlrpc_request *req)
1436 {
1437         ENTRY;
1438         atomic_inc(&req->rq_refcount);
1439         RETURN(req);
1440 }
1441
1442 void ptlrpc_retain_replayable_request(struct ptlrpc_request *req,
1443                                       struct obd_import *imp)
1444 {
1445         struct list_head *tmp;
1446
1447         LASSERT_SPIN_LOCKED(&imp->imp_lock);
1448
1449         /* clear this for new requests that were resent as well
1450            as resent replayed requests. */
1451         lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT);
1452
1453         /* don't re-add requests that have been replayed */
1454         if (!list_empty(&req->rq_replay_list))
1455                 return;
1456
1457         lustre_msg_add_flags(req->rq_reqmsg, MSG_REPLAY);
1458
1459         LASSERT(imp->imp_replayable);
1460         /* Balanced in ptlrpc_free_committed, usually. */
1461         ptlrpc_request_addref(req);
1462         list_for_each_prev(tmp, &imp->imp_replay_list) {
1463                 struct ptlrpc_request *iter =
1464                         list_entry(tmp, struct ptlrpc_request, rq_replay_list);
1465
1466                 /* We may have duplicate transnos if we create and then
1467                  * open a file, or for closes retained if to match creating
1468                  * opens, so use req->rq_xid as a secondary key.
1469                  * (See bugs 684, 685, and 428.)
1470                  * XXX no longer needed, but all opens need transnos!
1471                  */
1472                 if (iter->rq_transno > req->rq_transno)
1473                         continue;
1474
1475                 if (iter->rq_transno == req->rq_transno) {
1476                         LASSERT(iter->rq_xid != req->rq_xid);
1477                         if (iter->rq_xid > req->rq_xid)
1478                                 continue;
1479                 }
1480
1481                 list_add(&req->rq_replay_list, &iter->rq_replay_list);
1482                 return;
1483         }
1484
1485         list_add_tail(&req->rq_replay_list, &imp->imp_replay_list);
1486 }
1487
1488 int ptlrpc_queue_wait(struct ptlrpc_request *req)
1489 {
1490         int rc = 0;
1491         int brc;
1492         struct l_wait_info lwi;
1493         struct obd_import *imp = req->rq_import;
1494         cfs_duration_t timeout = 0;
1495         ENTRY;
1496
1497         LASSERT(req->rq_set == NULL);
1498         LASSERT(!req->rq_receiving_reply);
1499         atomic_inc(&imp->imp_inflight);
1500
1501         /* for distributed debugging */
1502         lustre_msg_set_status(req->rq_reqmsg, cfs_curproc_pid());
1503         LASSERT(imp->imp_obd != NULL);
1504         CDEBUG(D_RPCTRACE, "Sending RPC pname:cluuid:pid:xid:nid:opc "
1505                "%s:%s:%d:"LPU64":%s:%d\n", cfs_curproc_comm(),
1506                imp->imp_obd->obd_uuid.uuid,
1507                lustre_msg_get_status(req->rq_reqmsg), req->rq_xid,
1508                libcfs_nid2str(imp->imp_connection->c_peer.nid),
1509                lustre_msg_get_opc(req->rq_reqmsg));
1510
1511         /* Mark phase here for a little debug help */
1512         req->rq_phase = RQ_PHASE_RPC;
1513
1514         spin_lock(&imp->imp_lock);
1515         req->rq_import_generation = imp->imp_generation;
1516 restart:
1517         if (ptlrpc_import_delay_req(imp, req, &rc)) {
1518                 list_del(&req->rq_list);
1519
1520                 list_add_tail(&req->rq_list, &imp->imp_delayed_list);
1521                 spin_unlock(&imp->imp_lock);
1522
1523                 DEBUG_REQ(D_HA, req, "\"%s\" waiting for recovery: (%s != %s)",
1524                           cfs_curproc_comm(),
1525                           ptlrpc_import_state_name(req->rq_send_state),
1526                           ptlrpc_import_state_name(imp->imp_state));
1527                 lwi = LWI_INTR(interrupted_request, req);
1528                 rc = l_wait_event(req->rq_reply_waitq,
1529                                   (req->rq_send_state == imp->imp_state ||
1530                                    req->rq_err || req->rq_intr),
1531                                   &lwi);
1532                 DEBUG_REQ(D_HA, req, "\"%s\" awake: (%s == %s or %d/%d == 1)",
1533                           cfs_curproc_comm(),
1534                           ptlrpc_import_state_name(imp->imp_state),
1535                           ptlrpc_import_state_name(req->rq_send_state),
1536                           req->rq_err, req->rq_intr);
1537
1538                 spin_lock(&imp->imp_lock);
1539                 list_del_init(&req->rq_list);
1540
1541                 if (req->rq_err) {
1542                         rc = -EIO;
1543                 }
1544                 else if (req->rq_intr) {
1545                         rc = -EINTR;
1546                 }
1547                 else if (req->rq_no_resend) {
1548                         spin_unlock(&imp->imp_lock);
1549                         GOTO(out, rc = -ETIMEDOUT);
1550                 }
1551                 else {
1552                         GOTO(restart, rc);
1553                 }
1554         }
1555
1556         if (rc != 0) {
1557                 list_del_init(&req->rq_list);
1558                 spin_unlock(&imp->imp_lock);
1559                 req->rq_status = rc; // XXX this ok?
1560                 GOTO(out, rc);
1561         }
1562
1563         if (req->rq_resend) {
1564                 lustre_msg_add_flags(req->rq_reqmsg, MSG_RESENT);
1565
1566                 if (req->rq_bulk != NULL) {
1567                         ptlrpc_unregister_bulk (req);
1568
1569                         /* bulk requests are supposed to be
1570                          * idempotent, so we are free to bump the xid
1571                          * here, which we need to do before
1572                          * registering the bulk again (bug 6371).
1573                          * print the old xid first for sanity.
1574                          */
1575                         DEBUG_REQ(D_HA, req, "bumping xid for bulk: ");
1576                         req->rq_xid = ptlrpc_next_xid();
1577                 }
1578
1579                 DEBUG_REQ(D_HA, req, "resending: ");
1580         }
1581
1582         /* XXX this is the same as ptlrpc_set_wait */
1583         LASSERT(list_empty(&req->rq_list));
1584         list_add_tail(&req->rq_list, &imp->imp_sending_list);
1585         spin_unlock(&imp->imp_lock);
1586
1587         rc = ptl_send_rpc(req, 0);
1588         if (rc) {
1589                 DEBUG_REQ(D_HA, req, "send failed (%d); recovering", rc);
1590                 timeout = CFS_TICK;
1591         } else {
1592                 timeout = cfs_timeout_cap(cfs_time_seconds(req->rq_timeout));
1593                 DEBUG_REQ(D_NET, req,
1594                           "-- sleeping for "CFS_DURATION_T" jiffies", timeout);
1595         }
1596         lwi = LWI_TIMEOUT_INTR(timeout, expired_request, interrupted_request,
1597                                req);
1598         l_wait_event(req->rq_reply_waitq, ptlrpc_check_reply(req), &lwi);
1599
1600         CDEBUG(D_RPCTRACE, "Completed RPC pname:cluuid:pid:xid:nid:opc "
1601                "%s:%s:%d:"LPU64":%s:%d\n", cfs_curproc_comm(),
1602                imp->imp_obd->obd_uuid.uuid,
1603                lustre_msg_get_status(req->rq_reqmsg), req->rq_xid,
1604                libcfs_nid2str(imp->imp_connection->c_peer.nid),
1605                lustre_msg_get_opc(req->rq_reqmsg));
1606
1607         spin_lock(&imp->imp_lock);
1608         list_del_init(&req->rq_list);
1609         spin_unlock(&imp->imp_lock);
1610
1611         /* If the reply was received normally, this just grabs the spinlock
1612          * (ensuring the reply callback has returned), sees that
1613          * req->rq_receiving_reply is clear and returns. */
1614         ptlrpc_unregister_reply (req);
1615
1616         if (req->rq_err)
1617                 GOTO(out, rc = -EIO);
1618
1619         /* Resend if we need to, unless we were interrupted. */
1620         if (req->rq_resend && !req->rq_intr) {
1621                 /* ...unless we were specifically told otherwise. */
1622                 if (req->rq_no_resend)
1623                         GOTO(out, rc = -ETIMEDOUT);
1624                 spin_lock(&imp->imp_lock);
1625                 goto restart;
1626         }
1627
1628         if (req->rq_intr) {
1629                 /* Should only be interrupted if we timed out. */
1630                 if (!req->rq_timedout)
1631                         DEBUG_REQ(D_ERROR, req,
1632                                   "rq_intr set but rq_timedout not");
1633                 GOTO(out, rc = -EINTR);
1634         }
1635
1636         if (req->rq_timedout) {                 /* non-recoverable timeout */
1637                 GOTO(out, rc = -ETIMEDOUT);
1638         }
1639
1640         if (!req->rq_replied) {
1641                 /* How can this be? -eeb */
1642                 DEBUG_REQ(D_ERROR, req, "!rq_replied: ");
1643                 LBUG();
1644                 GOTO(out, rc = req->rq_status);
1645         }
1646
1647         rc = after_reply(req);
1648         /* NB may return +ve success rc */
1649         if (req->rq_resend) {
1650                 spin_lock(&imp->imp_lock);
1651                 goto restart;
1652         }
1653
1654  out:
1655         if (req->rq_bulk != NULL) {
1656                 if (rc >= 0) {
1657                         /* success so far.  Note that anything going wrong
1658                          * with bulk now, is EXTREMELY strange, since the
1659                          * server must have believed that the bulk
1660                          * tranferred OK before she replied with success to
1661                          * me. */
1662                         lwi = LWI_TIMEOUT(timeout, NULL, NULL);
1663                         brc = l_wait_event(req->rq_reply_waitq,
1664                                            !ptlrpc_bulk_active(req->rq_bulk),
1665                                            &lwi);
1666                         LASSERT(brc == 0 || brc == -ETIMEDOUT);
1667                         if (brc != 0) {
1668                                 LASSERT(brc == -ETIMEDOUT);
1669                                 DEBUG_REQ(D_ERROR, req, "bulk timed out");
1670                                 rc = brc;
1671                         } else if (!req->rq_bulk->bd_success) {
1672                                 DEBUG_REQ(D_ERROR, req, "bulk transfer failed");
1673                                 rc = -EIO;
1674                         }
1675                 }
1676                 if (rc < 0)
1677                         ptlrpc_unregister_bulk (req);
1678         }
1679
1680         LASSERT(!req->rq_receiving_reply);
1681         req->rq_phase = RQ_PHASE_INTERPRET;
1682
1683         atomic_dec(&imp->imp_inflight);
1684         cfs_waitq_signal(&imp->imp_recovery_waitq);
1685         RETURN(rc);
1686 }
1687
1688 struct ptlrpc_replay_async_args {
1689         int praa_old_state;
1690         int praa_old_status;
1691 };
1692
1693 static int ptlrpc_replay_interpret(struct ptlrpc_request *req,
1694                                     void * data, int rc)
1695 {
1696         struct ptlrpc_replay_async_args *aa = data;
1697         struct obd_import *imp = req->rq_import;
1698
1699         ENTRY;
1700         atomic_dec(&imp->imp_replay_inflight);
1701
1702         if (!req->rq_replied) {
1703                 CERROR("request replay timed out, restarting recovery\n");
1704                 GOTO(out, rc = -ETIMEDOUT);
1705         }
1706
1707         if (lustre_msg_get_type(req->rq_repmsg) == PTL_RPC_MSG_ERR &&
1708             lustre_msg_get_status(req->rq_repmsg) == -ENOTCONN)
1709                 GOTO(out, rc = lustre_msg_get_status(req->rq_repmsg));
1710
1711         /* The transno had better not change over replay. */
1712         LASSERT(lustre_msg_get_transno(req->rq_reqmsg) ==
1713                 lustre_msg_get_transno(req->rq_repmsg));
1714
1715         DEBUG_REQ(D_HA, req, "got rep");
1716
1717         /* let the callback do fixups, possibly including in the request */
1718         if (req->rq_replay_cb)
1719                 req->rq_replay_cb(req);
1720
1721         if (req->rq_replied &&
1722             lustre_msg_get_status(req->rq_repmsg) != aa->praa_old_status) {
1723                 DEBUG_REQ(D_ERROR, req, "status %d, old was %d",
1724                           lustre_msg_get_status(req->rq_repmsg),
1725                           aa->praa_old_status);
1726         } else {
1727                 /* Put it back for re-replay. */
1728                 lustre_msg_set_status(req->rq_repmsg, aa->praa_old_status);
1729         }
1730
1731         spin_lock(&imp->imp_lock);
1732         imp->imp_last_replay_transno = req->rq_transno;
1733         spin_unlock(&imp->imp_lock);
1734
1735         /* continue with recovery */
1736         rc = ptlrpc_import_recovery_state_machine(imp);
1737  out:
1738         req->rq_send_state = aa->praa_old_state;
1739
1740         if (rc != 0)
1741                 /* this replay failed, so restart recovery */
1742                 ptlrpc_connect_import(imp, NULL);
1743
1744         RETURN(rc);
1745 }
1746
1747
1748 int ptlrpc_replay_req(struct ptlrpc_request *req)
1749 {
1750         struct ptlrpc_replay_async_args *aa;
1751         ENTRY;
1752
1753         LASSERT(req->rq_import->imp_state == LUSTRE_IMP_REPLAY);
1754
1755         /* Not handling automatic bulk replay yet (or ever?) */
1756         LASSERT(req->rq_bulk == NULL);
1757
1758         DEBUG_REQ(D_HA, req, "REPLAY");
1759
1760         LASSERT (sizeof (*aa) <= sizeof (req->rq_async_args));
1761         aa = (struct ptlrpc_replay_async_args *)&req->rq_async_args;
1762         memset(aa, 0, sizeof *aa);
1763
1764         /* Prepare request to be resent with ptlrpcd */
1765         aa->praa_old_state = req->rq_send_state;
1766         req->rq_send_state = LUSTRE_IMP_REPLAY;
1767         req->rq_phase = RQ_PHASE_NEW;
1768         aa->praa_old_status = lustre_msg_get_status(req->rq_repmsg);
1769         req->rq_status = 0;
1770
1771         req->rq_interpret_reply = ptlrpc_replay_interpret;
1772         atomic_inc(&req->rq_import->imp_replay_inflight);
1773         ptlrpc_request_addref(req); /* ptlrpcd needs a ref */
1774
1775         ptlrpcd_add_req(req);
1776         RETURN(0);
1777 }
1778
1779 void ptlrpc_abort_inflight(struct obd_import *imp)
1780 {
1781         struct list_head *tmp, *n;
1782         ENTRY;
1783
1784         /* Make sure that no new requests get processed for this import.
1785          * ptlrpc_{queue,set}_wait must (and does) hold imp_lock while testing
1786          * this flag and then putting requests on sending_list or delayed_list.
1787          */
1788         spin_lock(&imp->imp_lock);
1789
1790         /* XXX locking?  Maybe we should remove each request with the list
1791          * locked?  Also, how do we know if the requests on the list are
1792          * being freed at this time?
1793          */
1794         list_for_each_safe(tmp, n, &imp->imp_sending_list) {
1795                 struct ptlrpc_request *req =
1796                         list_entry(tmp, struct ptlrpc_request, rq_list);
1797
1798                 DEBUG_REQ(D_RPCTRACE, req, "inflight");
1799
1800                 spin_lock (&req->rq_lock);
1801                 if (req->rq_import_generation < imp->imp_generation) {
1802                         req->rq_err = 1;
1803                         ptlrpc_wake_client_req(req);
1804                 }
1805                 spin_unlock (&req->rq_lock);
1806         }
1807
1808         list_for_each_safe(tmp, n, &imp->imp_delayed_list) {
1809                 struct ptlrpc_request *req =
1810                         list_entry(tmp, struct ptlrpc_request, rq_list);
1811
1812                 DEBUG_REQ(D_RPCTRACE, req, "aborting waiting req");
1813
1814                 spin_lock (&req->rq_lock);
1815                 if (req->rq_import_generation < imp->imp_generation) {
1816                         req->rq_err = 1;
1817                         ptlrpc_wake_client_req(req);
1818                 }
1819                 spin_unlock (&req->rq_lock);
1820         }
1821
1822         /* Last chance to free reqs left on the replay list, but we
1823          * will still leak reqs that haven't committed.  */
1824         if (imp->imp_replayable)
1825                 ptlrpc_free_committed(imp);
1826
1827         spin_unlock(&imp->imp_lock);
1828
1829         EXIT;
1830 }
1831
1832 static __u64 ptlrpc_last_xid = 0;
1833 spinlock_t ptlrpc_last_xid_lock;
1834
1835 __u64 ptlrpc_next_xid(void)
1836 {
1837         __u64 tmp;
1838         spin_lock(&ptlrpc_last_xid_lock);
1839         tmp = ++ptlrpc_last_xid;
1840         spin_unlock(&ptlrpc_last_xid_lock);
1841         return tmp;
1842 }
1843
1844 __u64 ptlrpc_sample_next_xid(void)
1845 {
1846         __u64 tmp;
1847         spin_lock(&ptlrpc_last_xid_lock);
1848         tmp = ptlrpc_last_xid + 1;
1849         spin_unlock(&ptlrpc_last_xid_lock);
1850         return tmp;
1851 }
1852 EXPORT_SYMBOL(ptlrpc_sample_next_xid);