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                 struct obd_export *exp = req->rq_export;
 574                 __u32 opc = lustre_msg_get_opc(req->rq_reqmsg);
 575
 576                 LCONSOLE_ERROR_MSG(0x011, "an error ocurred while communicating"
 577                                    " with %s The %s operation failed with %d",
 578                                    exp ? obd_export_nid2str(exp) : "(no nid)",
 579                                    ll_opcode2str(opc), err);
 580                 RETURN(err < 0 ? err : -EINVAL);
 581         }
 582
 583         if (err < 0) {
 584                 DEBUG_REQ(D_INFO, req, "status is %d", err);
 585         } else if (err > 0) {
 586                 /* XXX: translate this error from net to host */
 587                 DEBUG_REQ(D_INFO, req, "status is %d", err);
 588         }
 589
 590         RETURN(err);
 591 }
 592
 593 static int after_reply(struct ptlrpc_request *req)
 594 {
 595         struct obd_import *imp = req->rq_import;
 596         int rc;
 597         ENTRY;
 598
 599         LASSERT(!req->rq_receiving_reply);
 600
 601         /* NB Until this point, the whole of the incoming message,
 602          * including buflens, status etc is in the sender's byte order. */
 603
 604 #if SWAB_PARANOIA
 605         /* Clear reply swab mask; this is a new reply in sender's byte order */
 606         req->rq_rep_swab_mask = 0;
 607 #endif
 608         LASSERT (req->rq_nob_received <= req->rq_replen);
 609         rc = lustre_unpack_msg(req->rq_repmsg, req->rq_nob_received);
 610         if (rc) {
 611                 DEBUG_REQ(D_ERROR, req, "unpack_rep failed: %d\n", rc);
 612                 RETURN(-EPROTO);
 613         }
 614
 615         rc = lustre_unpack_rep_ptlrpc_body(req, MSG_PTLRPC_BODY_OFF);
 616         if (rc) {
 617                 DEBUG_REQ(D_ERROR, req, "unpack ptlrpc body failed: %d\n", rc);
 618                 RETURN(-EPROTO);
 619         }
 620
 621         if (lustre_msg_get_type(req->rq_repmsg) != PTL_RPC_MSG_REPLY &&
 622             lustre_msg_get_type(req->rq_repmsg) != PTL_RPC_MSG_ERR) {
 623                 DEBUG_REQ(D_ERROR, req, "invalid packet received (type=%u)\n",
 624                           lustre_msg_get_type(req->rq_repmsg));
 625                 RETURN(-EPROTO);
 626         }
 627
 628         rc = ptlrpc_check_status(req);
 629
 630         /* Either we've been evicted, or the server has failed for
 631          * some reason. Try to reconnect, and if that fails, punt to the
 632          * upcall. */
 633         if ((rc == -ENOTCONN) || (rc == -ENODEV)) {
 634                 if (req->rq_send_state != LUSTRE_IMP_FULL ||
 635                     imp->imp_obd->obd_no_recov || imp->imp_dlm_fake) {
 636                         RETURN(-ENOTCONN);
 637                 }
 638
 639                 ptlrpc_request_handle_notconn(req);
 640
 641                 RETURN(rc);
 642         }
 643
 644         /* Store transno in reqmsg for replay. */
 645         req->rq_transno = lustre_msg_get_transno(req->rq_repmsg);
 646         lustre_msg_set_transno(req->rq_reqmsg, req->rq_transno);
 647
 648         if (req->rq_import->imp_replayable) {
 649                 spin_lock(&imp->imp_lock);
 650                 /* no point in adding already-committed requests to the replay
 651                  * list, we will just remove them immediately. b=9829 */
 652                 if (req->rq_transno != 0 &&
 653                     (req->rq_transno >
 654                      lustre_msg_get_last_committed(req->rq_repmsg) ||
 655                      req->rq_replay))
 656                         ptlrpc_retain_replayable_request(req, imp);
 657                 else if (req->rq_commit_cb != NULL) {
 658                         spin_unlock(&imp->imp_lock);
 659                         req->rq_commit_cb(req);
 660                         spin_lock(&imp->imp_lock);
 661                 }
 662
 663                 /* Replay-enabled imports return commit-status information. */
 664                 if (lustre_msg_get_last_committed(req->rq_repmsg))
 665                         imp->imp_peer_committed_transno =
 666                                 lustre_msg_get_last_committed(req->rq_repmsg);
 667                 ptlrpc_free_committed(imp);
 668                 spin_unlock(&imp->imp_lock);
 669         }
 670
 671         RETURN(rc);
 672 }
 673
 674 static int ptlrpc_send_new_req(struct ptlrpc_request *req)
 675 {
 676         struct obd_import     *imp;
 677         int rc;
 678         ENTRY;
 679
 680         LASSERT(req->rq_phase == RQ_PHASE_NEW);
 681         req->rq_phase = RQ_PHASE_RPC;
 682
 683         imp = req->rq_import;
 684         spin_lock(&imp->imp_lock);
 685
 686         req->rq_import_generation = imp->imp_generation;
 687
 688         if (ptlrpc_import_delay_req(imp, req, &rc)) {
 689                 spin_lock (&req->rq_lock);
 690                 req->rq_waiting = 1;
 691                 spin_unlock (&req->rq_lock);
 692
 693                 DEBUG_REQ(D_HA, req, "req from PID %d waiting for recovery: "
 694                           "(%s != %s)",
 695                           lustre_msg_get_status(req->rq_reqmsg) ,
 696                           ptlrpc_import_state_name(req->rq_send_state),
 697                           ptlrpc_import_state_name(imp->imp_state));
 698                 LASSERT(list_empty (&req->rq_list));
 699
 700                 list_add_tail(&req->rq_list, &imp->imp_delayed_list);
 701                 spin_unlock(&imp->imp_lock);
 702                 RETURN(0);
 703         }
 704
 705         if (rc != 0) {
 706                 spin_unlock(&imp->imp_lock);
 707                 req->rq_status = rc;
 708                 req->rq_phase = RQ_PHASE_INTERPRET;
 709                 RETURN(rc);
 710         }
 711
 712         /* XXX this is the same as ptlrpc_queue_wait */
 713         LASSERT(list_empty(&req->rq_list));
 714         list_add_tail(&req->rq_list, &imp->imp_sending_list);
 715         spin_unlock(&imp->imp_lock);
 716
 717         lustre_msg_set_status(req->rq_reqmsg, cfs_curproc_pid());
 718         CDEBUG(D_RPCTRACE, "Sending RPC pname:cluuid:pid:xid:nid:opc"
 719                " %s:%s:%d:"LPU64":%s:%d\n", cfs_curproc_comm(),
 720                imp->imp_obd->obd_uuid.uuid,
 721                lustre_msg_get_status(req->rq_reqmsg), req->rq_xid,
 722                libcfs_nid2str(imp->imp_connection->c_peer.nid),
 723                lustre_msg_get_opc(req->rq_reqmsg));
 724
 725         rc = ptl_send_rpc(req, 0);
 726         if (rc) {
 727                 DEBUG_REQ(D_HA, req, "send failed (%d); expect timeout", rc);
 728                 req->rq_net_err = 1;
 729                 RETURN(rc);
 730         }
 731         RETURN(0);
 732 }
 733
 734 /* this sends any unsent RPCs in @set and returns TRUE if all are sent */
 735 int ptlrpc_check_set(struct ptlrpc_request_set *set)
 736 {
 737         struct list_head *tmp;
 738         int force_timer_recalc = 0;
 739         ENTRY;
 740
 741         if (set->set_remaining == 0)
 742                 RETURN(1);
 743
 744         list_for_each(tmp, &set->set_requests) {
 745                 struct ptlrpc_request *req =
 746                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 747                 struct obd_import *imp = req->rq_import;
 748                 int rc = 0;
 749
 750                 if (req->rq_phase == RQ_PHASE_NEW &&
 751                     ptlrpc_send_new_req(req)) {
 752                         force_timer_recalc = 1;
 753                 }
 754
 755                 if (!(req->rq_phase == RQ_PHASE_RPC ||
 756                       req->rq_phase == RQ_PHASE_BULK ||
 757                       req->rq_phase == RQ_PHASE_INTERPRET ||
 758                       req->rq_phase == RQ_PHASE_COMPLETE)) {
 759                         DEBUG_REQ(D_ERROR, req, "bad phase %x", req->rq_phase);
 760                         LBUG();
 761                 }
 762
 763                 if (req->rq_phase == RQ_PHASE_COMPLETE)
 764                         continue;
 765
 766                 if (req->rq_phase == RQ_PHASE_INTERPRET)
 767                         GOTO(interpret, req->rq_status);
 768
 769                 if (req->rq_net_err && !req->rq_timedout)
 770                         ptlrpc_expire_one_request(req);
 771
 772                 if (req->rq_err) {
 773                         ptlrpc_unregister_reply(req);
 774                         if (req->rq_status == 0)
 775                                 req->rq_status = -EIO;
 776                         req->rq_phase = RQ_PHASE_INTERPRET;
 777
 778                         spin_lock(&imp->imp_lock);
 779                         list_del_init(&req->rq_list);
 780                         spin_unlock(&imp->imp_lock);
 781
 782                         GOTO(interpret, req->rq_status);
 783                 }
 784
 785                 /* ptlrpc_queue_wait->l_wait_event guarantees that rq_intr
 786                  * will only be set after rq_timedout, but the oig waiting
 787                  * path sets rq_intr irrespective of whether ptlrpcd has
 788                  * seen a timeout.  our policy is to only interpret
 789                  * interrupted rpcs after they have timed out */
 790                 if (req->rq_intr && (req->rq_timedout || req->rq_waiting)) {
 791                         /* NB could be on delayed list */
 792                         ptlrpc_unregister_reply(req);
 793                         req->rq_status = -EINTR;
 794                         req->rq_phase = RQ_PHASE_INTERPRET;
 795
 796                         spin_lock(&imp->imp_lock);
 797                         list_del_init(&req->rq_list);
 798                         spin_unlock(&imp->imp_lock);
 799
 800                         GOTO(interpret, req->rq_status);
 801                 }
 802
 803                 if (req->rq_phase == RQ_PHASE_RPC) {
 804                         if (req->rq_timedout||req->rq_waiting||req->rq_resend) {
 805                                 int status;
 806
 807                                 ptlrpc_unregister_reply(req);
 808
 809                                 spin_lock(&imp->imp_lock);
 810
 811                                 if (ptlrpc_import_delay_req(imp, req, &status)){
 812                                         spin_unlock(&imp->imp_lock);
 813                                         continue;
 814                                 }
 815
 816                                 list_del_init(&req->rq_list);
 817                                 if (status != 0)  {
 818                                         req->rq_status = status;
 819                                         req->rq_phase = RQ_PHASE_INTERPRET;
 820                                         spin_unlock(&imp->imp_lock);
 821                                         GOTO(interpret, req->rq_status);
 822                                 }
 823                                 if (req->rq_no_resend) {
 824                                         req->rq_status = -ENOTCONN;
 825                                         req->rq_phase = RQ_PHASE_INTERPRET;
 826                                         spin_unlock(&imp->imp_lock);
 827                                         GOTO(interpret, req->rq_status);
 828                                 }
 829                                 list_add_tail(&req->rq_list,
 830                                               &imp->imp_sending_list);
 831
 832                                 spin_unlock(&imp->imp_lock);
 833
 834                                 req->rq_waiting = 0;
 835                                 if (req->rq_resend) {
 836                                         lustre_msg_add_flags(req->rq_reqmsg,
 837                                                              MSG_RESENT);
 838                                         if (req->rq_bulk) {
 839                                                 __u64 old_xid = req->rq_xid;
 840
 841                                                 ptlrpc_unregister_bulk (req);
 842
 843                                                 /* ensure previous bulk fails */
 844                                                 req->rq_xid = ptlrpc_next_xid();
 845                                                 CDEBUG(D_HA, "resend bulk "
 846                                                        "old x"LPU64
 847                                                        " new x"LPU64"\n",
 848                                                        old_xid, req->rq_xid);
 849                                         }
 850                                 }
 851
 852                                 rc = ptl_send_rpc(req, 0);
 853                                 if (rc) {
 854                                         DEBUG_REQ(D_HA, req, "send failed (%d)",
 855                                                   rc);
 856                                         force_timer_recalc = 1;
 857                                         req->rq_net_err = 1;
 858                                 }
 859                                 /* need to reset the timeout */
 860                                 force_timer_recalc = 1;
 861                         }
 862
 863                         /* Still waiting for a reply? */
 864                         if (ptlrpc_client_receiving_reply(req))
 865                                 continue;
 866
 867                         /* Did we actually receive a reply? */
 868                         if (!ptlrpc_client_replied(req))
 869                                 continue;
 870
 871                         spin_lock(&imp->imp_lock);
 872                         list_del_init(&req->rq_list);
 873                         spin_unlock(&imp->imp_lock);
 874
 875                         req->rq_status = after_reply(req);
 876                         if (req->rq_resend) {
 877                                 /* Add this req to the delayed list so
 878                                    it can be errored if the import is
 879                                    evicted after recovery. */
 880                                 spin_lock(&imp->imp_lock);
 881                                 list_add_tail(&req->rq_list,
 882                                               &imp->imp_delayed_list);
 883                                 spin_unlock(&imp->imp_lock);
 884                                 continue;
 885                         }
 886
 887                         /* If there is no bulk associated with this request,
 888                          * then we're done and should let the interpreter
 889                          * process the reply.  Similarly if the RPC returned
 890                          * an error, and therefore the bulk will never arrive.
 891                          */
 892                         if (req->rq_bulk == NULL || req->rq_status != 0) {
 893                                 req->rq_phase = RQ_PHASE_INTERPRET;
 894                                 GOTO(interpret, req->rq_status);
 895                         }
 896
 897                         req->rq_phase = RQ_PHASE_BULK;
 898                 }
 899
 900                 LASSERT(req->rq_phase == RQ_PHASE_BULK);
 901                 if (ptlrpc_bulk_active(req->rq_bulk))
 902                         continue;
 903
 904                 if (!req->rq_bulk->bd_success) {
 905                         /* The RPC reply arrived OK, but the bulk screwed
 906                          * up!  Dead wierd since the server told us the RPC
 907                          * was good after getting the REPLY for her GET or
 908                          * the ACK for her PUT. */
 909                         DEBUG_REQ(D_ERROR, req, "bulk transfer failed");
 910                         LBUG();
 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, "timeout (sent at %lu, %lus ago)",
 956                   (long)req->rq_sent, CURRENT_SECONDS - req->rq_sent);
 957
 958         if (imp != NULL && obd_debug_peer_on_timeout)
 959                 LNetCtl(IOC_LIBCFS_DEBUG_PEER, &imp->imp_connection->c_peer);
 960
 961         spin_lock(&req->rq_lock);
 962         req->rq_timedout = 1;
 963         spin_unlock(&req->rq_lock);
 964
 965         ptlrpc_unregister_reply (req);
 966
 967         if (obd_dump_on_timeout)
 968                 libcfs_debug_dumplog();
 969
 970         if (req->rq_bulk != NULL)
 971                 ptlrpc_unregister_bulk (req);
 972
 973         if (imp == NULL) {
 974                 DEBUG_REQ(D_HA, req, "NULL import: already cleaned up?");
 975                 RETURN(1);
 976         }
 977
 978         /* The DLM server doesn't want recovery run on its imports. */
 979         if (imp->imp_dlm_fake)
 980                 RETURN(1);
 981
 982         /* If this request is for recovery or other primordial tasks,
 983          * then error it out here. */
 984         if (req->rq_send_state != LUSTRE_IMP_FULL ||
 985             imp->imp_obd->obd_no_recov) {
 986                 spin_lock(&req->rq_lock);
 987                 req->rq_status = -ETIMEDOUT;
 988                 req->rq_err = 1;
 989                 spin_unlock(&req->rq_lock);
 990                 RETURN(1);
 991         }
 992
 993         /* if request can't be resend we can't wait answer after timeout */
 994         if (req->rq_no_resend) {
 995                 DEBUG_REQ(D_RPCTRACE, req, "TIMEOUT-NORESEND:");
 996                 rc = 1;
 997         }
 998
 999         ptlrpc_fail_import(imp, lustre_msg_get_conn_cnt(req->rq_reqmsg));
1000
1001         RETURN(rc);
1002 }
1003
1004 int ptlrpc_expired_set(void *data)
1005 {
1006         struct ptlrpc_request_set *set = data;
1007         struct list_head          *tmp;
1008         time_t                     now = CURRENT_SECONDS;
1009         ENTRY;
1010
1011         LASSERT(set != NULL);
1012
1013         /* A timeout expired; see which reqs it applies to... */
1014         list_for_each (tmp, &set->set_requests) {
1015                 struct ptlrpc_request *req =
1016                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
1017
1018                 /* request in-flight? */
1019                 if (!((req->rq_phase == RQ_PHASE_RPC && !req->rq_waiting &&
1020                        !req->rq_resend) ||
1021                       (req->rq_phase == RQ_PHASE_BULK)))
1022                         continue;
1023
1024                 if (req->rq_timedout ||           /* already dealt with */
1025                     req->rq_sent + req->rq_timeout > now) /* not expired */
1026                         continue;
1027
1028                 /* deal with this guy */
1029                 ptlrpc_expire_one_request (req);
1030         }
1031
1032         /* When waiting for a whole set, we always to break out of the
1033          * sleep so we can recalculate the timeout, or enable interrupts
1034          * iff everyone's timed out.
1035          */
1036         RETURN(1);
1037 }
1038
1039 void ptlrpc_mark_interrupted(struct ptlrpc_request *req)
1040 {
1041         spin_lock(&req->rq_lock);
1042         req->rq_intr = 1;
1043         spin_unlock(&req->rq_lock);
1044 }
1045
1046 void ptlrpc_interrupted_set(void *data)
1047 {
1048         struct ptlrpc_request_set *set = data;
1049         struct list_head *tmp;
1050
1051         LASSERT(set != NULL);
1052         CERROR("INTERRUPTED SET %p\n", set);
1053
1054         list_for_each(tmp, &set->set_requests) {
1055                 struct ptlrpc_request *req =
1056                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
1057
1058                 if (req->rq_phase != RQ_PHASE_RPC)
1059                         continue;
1060
1061                 ptlrpc_mark_interrupted(req);
1062         }
1063 }
1064
1065 int ptlrpc_set_next_timeout(struct ptlrpc_request_set *set)
1066 {
1067         struct list_head      *tmp;
1068         time_t                 now = CURRENT_SECONDS;
1069         time_t                 deadline;
1070         int                    timeout = 0;
1071         struct ptlrpc_request *req;
1072         ENTRY;
1073
1074         SIGNAL_MASK_ASSERT(); /* XXX BUG 1511 */
1075
1076         list_for_each(tmp, &set->set_requests) {
1077                 req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
1078
1079                 /* request in-flight? */
1080                 if (!((req->rq_phase == RQ_PHASE_RPC && !req->rq_waiting) ||
1081                       (req->rq_phase == RQ_PHASE_BULK)))
1082                         continue;
1083
1084                 if (req->rq_timedout)   /* already timed out */
1085                         continue;
1086
1087                 deadline = req->rq_sent + req->rq_timeout;
1088                 if (deadline <= now)    /* actually expired already */
1089                         timeout = 1;    /* ASAP */
1090                 else if (timeout == 0 || timeout > deadline - now)
1091                         timeout = deadline - now;
1092         }
1093         RETURN(timeout);
1094 }
1095
1096 int ptlrpc_set_wait(struct ptlrpc_request_set *set)
1097 {
1098         struct list_head      *tmp;
1099         struct ptlrpc_request *req;
1100         struct l_wait_info     lwi;
1101         int                    rc, timeout;
1102         ENTRY;
1103
1104         if (list_empty(&set->set_requests))
1105                 RETURN(0);
1106
1107         list_for_each(tmp, &set->set_requests) {
1108                 req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
1109                 if (req->rq_phase == RQ_PHASE_NEW)
1110                         (void)ptlrpc_send_new_req(req);
1111         }
1112
1113         do {
1114                 timeout = ptlrpc_set_next_timeout(set);
1115
1116                 /* wait until all complete, interrupted, or an in-flight
1117                  * req times out */
1118                 CDEBUG(D_HA, "set %p going to sleep for %d seconds\n",
1119                        set, timeout);
1120                 lwi = LWI_TIMEOUT_INTR(cfs_time_seconds(timeout ? timeout : 1),
1121                                        ptlrpc_expired_set,
1122                                        ptlrpc_interrupted_set, set);
1123                 rc = l_wait_event(set->set_waitq, ptlrpc_check_set(set), &lwi);
1124
1125                 LASSERT(rc == 0 || rc == -EINTR || rc == -ETIMEDOUT);
1126
1127                 /* -EINTR => all requests have been flagged rq_intr so next
1128                  * check completes.
1129                  * -ETIMEOUTD => someone timed out.  When all reqs have
1130                  * timed out, signals are enabled allowing completion with
1131                  * EINTR.
1132                  * I don't really care if we go once more round the loop in
1133                  * the error cases -eeb. */
1134         } while (rc != 0 || set->set_remaining != 0);
1135
1136         LASSERT(set->set_remaining == 0);
1137
1138         rc = 0;
1139         list_for_each(tmp, &set->set_requests) {
1140                 req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
1141
1142                 LASSERT(req->rq_phase == RQ_PHASE_COMPLETE);
1143                 if (req->rq_status != 0)
1144                         rc = req->rq_status;
1145         }
1146
1147         if (set->set_interpret != NULL) {
1148                 int (*interpreter)(struct ptlrpc_request_set *set,void *,int) =
1149                         set->set_interpret;
1150                 rc = interpreter (set, set->set_arg, rc);
1151         }
1152
1153         RETURN(rc);
1154 }
1155
1156 static void __ptlrpc_free_req_to_pool(struct ptlrpc_request *request)
1157 {
1158         struct ptlrpc_request_pool *pool = request->rq_pool;
1159
1160         spin_lock(&pool->prp_lock);
1161         list_add_tail(&request->rq_list, &pool->prp_req_list);
1162         spin_unlock(&pool->prp_lock);
1163 }
1164
1165 static void __ptlrpc_free_req(struct ptlrpc_request *request, int locked)
1166 {
1167         ENTRY;
1168         if (request == NULL) {
1169                 EXIT;
1170                 return;
1171         }
1172
1173         LASSERTF(!request->rq_receiving_reply, "req %p\n", request);
1174         LASSERTF(request->rq_rqbd == NULL, "req %p\n",request);/* client-side */
1175         LASSERTF(list_empty(&request->rq_list), "req %p\n", request);
1176         LASSERTF(list_empty(&request->rq_set_chain), "req %p\n", request);
1177
1178         /* We must take it off the imp_replay_list first.  Otherwise, we'll set
1179          * request->rq_reqmsg to NULL while osc_close is dereferencing it. */
1180         if (request->rq_import != NULL) {
1181                 if (!locked)
1182                         spin_lock(&request->rq_import->imp_lock);
1183                 list_del_init(&request->rq_replay_list);
1184                 if (!locked)
1185                         spin_unlock(&request->rq_import->imp_lock);
1186         }
1187         LASSERTF(list_empty(&request->rq_replay_list), "req %p\n", request);
1188
1189         if (atomic_read(&request->rq_refcount) != 0) {
1190                 DEBUG_REQ(D_ERROR, request,
1191                           "freeing request with nonzero refcount");
1192                 LBUG();
1193         }
1194
1195         if (request->rq_repmsg != NULL) {
1196                 OBD_FREE(request->rq_repmsg, request->rq_replen);
1197                 request->rq_repmsg = NULL;
1198         }
1199         if (request->rq_export != NULL) {
1200                 class_export_put(request->rq_export);
1201                 request->rq_export = NULL;
1202         }
1203         if (request->rq_import != NULL) {
1204                 class_import_put(request->rq_import);
1205                 request->rq_import = NULL;
1206         }
1207         if (request->rq_bulk != NULL)
1208                 ptlrpc_free_bulk(request->rq_bulk);
1209
1210         if (request->rq_pool) {
1211                 __ptlrpc_free_req_to_pool(request);
1212         } else {
1213                 if (request->rq_reqmsg != NULL) {
1214                         OBD_FREE(request->rq_reqmsg, request->rq_reqlen);
1215                         request->rq_reqmsg = NULL;
1216                 }
1217                 OBD_FREE(request, sizeof(*request));
1218         }
1219         EXIT;
1220 }
1221
1222 void ptlrpc_free_req(struct ptlrpc_request *request)
1223 {
1224         __ptlrpc_free_req(request, 0);
1225 }
1226
1227 static int __ptlrpc_req_finished(struct ptlrpc_request *request, int locked);
1228 void ptlrpc_req_finished_with_imp_lock(struct ptlrpc_request *request)
1229 {
1230         LASSERT_SPIN_LOCKED(&request->rq_import->imp_lock);
1231         (void)__ptlrpc_req_finished(request, 1);
1232 }
1233
1234 static int __ptlrpc_req_finished(struct ptlrpc_request *request, int locked)
1235 {
1236         ENTRY;
1237         if (request == NULL)
1238                 RETURN(1);
1239
1240         if (request == LP_POISON ||
1241             request->rq_reqmsg == LP_POISON) {
1242                 CERROR("dereferencing freed request (bug 575)\n");
1243                 LBUG();
1244                 RETURN(1);
1245         }
1246
1247         DEBUG_REQ(D_INFO, request, "refcount now %u",
1248                   atomic_read(&request->rq_refcount) - 1);
1249
1250         if (atomic_dec_and_test(&request->rq_refcount)) {
1251                 __ptlrpc_free_req(request, locked);
1252                 RETURN(1);
1253         }
1254
1255         RETURN(0);
1256 }
1257
1258 void ptlrpc_req_finished(struct ptlrpc_request *request)
1259 {
1260         __ptlrpc_req_finished(request, 0);
1261 }
1262
1263 __u64 ptlrpc_req_xid(struct ptlrpc_request *request)
1264 {
1265         return request->rq_xid;
1266 }
1267 EXPORT_SYMBOL(ptlrpc_req_xid);
1268
1269 /* Disengage the client's reply buffer from the network
1270  * NB does _NOT_ unregister any client-side bulk.
1271  * IDEMPOTENT, but _not_ safe against concurrent callers.
1272  * The request owner (i.e. the thread doing the I/O) must call...
1273  */
1274 void ptlrpc_unregister_reply (struct ptlrpc_request *request)
1275 {
1276         int                rc;
1277         cfs_waitq_t       *wq;
1278         struct l_wait_info lwi;
1279         ENTRY;
1280
1281         LASSERT(!in_interrupt ());             /* might sleep */
1282
1283         if (!ptlrpc_client_receiving_reply(request))
1284                 return;
1285
1286         LNetMDUnlink (request->rq_reply_md_h);
1287
1288         /* We have to l_wait_event() whatever the result, to give liblustre
1289          * a chance to run reply_in_callback() */
1290
1291         if (request->rq_set != NULL)
1292                 wq = &request->rq_set->set_waitq;
1293         else
1294                 wq = &request->rq_reply_waitq;
1295
1296         for (;;) {
1297                 /* Network access will complete in finite time but the HUGE
1298                  * timeout lets us CWARN for visibility of sluggish NALs */
1299                 lwi = LWI_TIMEOUT(cfs_time_seconds(300), NULL, NULL);
1300                 rc = l_wait_event (*wq, !ptlrpc_client_receiving_reply(request), &lwi);
1301                 if (rc == 0)
1302                         return;
1303
1304                 LASSERT (rc == -ETIMEDOUT);
1305                 DEBUG_REQ(D_WARNING, request, "Unexpectedly long timeout");
1306         }
1307         EXIT;
1308 }
1309
1310 /* caller must hold imp->imp_lock */
1311 void ptlrpc_free_committed(struct obd_import *imp)
1312 {
1313         struct list_head *tmp, *saved;
1314         struct ptlrpc_request *req;
1315         struct ptlrpc_request *last_req = NULL; /* temporary fire escape */
1316         ENTRY;
1317
1318         LASSERT(imp != NULL);
1319
1320         LASSERT_SPIN_LOCKED(&imp->imp_lock);
1321
1322
1323         if (imp->imp_peer_committed_transno == imp->imp_last_transno_checked &&
1324             imp->imp_generation == imp->imp_last_generation_checked) {
1325                 CDEBUG(D_HA, "%s: skip recheck for last_committed "LPU64"\n",
1326                        imp->imp_obd->obd_name, imp->imp_peer_committed_transno);
1327                 return;
1328         }
1329
1330         CDEBUG(D_HA, "%s: committing for last_committed "LPU64" gen %d\n",
1331                imp->imp_obd->obd_name, imp->imp_peer_committed_transno,
1332                imp->imp_generation);
1333         imp->imp_last_transno_checked = imp->imp_peer_committed_transno;
1334         imp->imp_last_generation_checked = imp->imp_generation;
1335
1336         list_for_each_safe(tmp, saved, &imp->imp_replay_list) {
1337                 req = list_entry(tmp, struct ptlrpc_request, rq_replay_list);
1338
1339                 /* XXX ok to remove when 1357 resolved - rread 05/29/03  */
1340                 LASSERT(req != last_req);
1341                 last_req = req;
1342
1343                 if (req->rq_import_generation < imp->imp_generation) {
1344                         DEBUG_REQ(D_HA, req, "freeing request with old gen");
1345                         GOTO(free_req, 0);
1346                 }
1347
1348                 if (req->rq_replay) {
1349                         DEBUG_REQ(D_HA, req, "keeping (FL_REPLAY)");
1350                         continue;
1351                 }
1352
1353                 /* not yet committed */
1354                 if (req->rq_transno > imp->imp_peer_committed_transno) {
1355                         DEBUG_REQ(D_HA, req, "stopping search");
1356                         break;
1357                 }
1358
1359                 DEBUG_REQ(D_HA, req, "committing (last_committed "LPU64")",
1360                           imp->imp_peer_committed_transno);
1361 free_req:
1362                 spin_lock(&req->rq_lock);
1363                 req->rq_replay = 0;
1364                 spin_unlock(&req->rq_lock);
1365                 if (req->rq_commit_cb != NULL)
1366                         req->rq_commit_cb(req);
1367                 list_del_init(&req->rq_replay_list);
1368                 __ptlrpc_req_finished(req, 1);
1369         }
1370
1371         EXIT;
1372         return;
1373 }
1374
1375 void ptlrpc_cleanup_client(struct obd_import *imp)
1376 {
1377         ENTRY;
1378         EXIT;
1379         return;
1380 }
1381
1382 void ptlrpc_resend_req(struct ptlrpc_request *req)
1383 {
1384         DEBUG_REQ(D_HA, req, "going to resend");
1385         lustre_msg_set_handle(req->rq_reqmsg, &(struct lustre_handle){ 0 });
1386         req->rq_status = -EAGAIN;
1387
1388         spin_lock(&req->rq_lock);
1389         req->rq_resend = 1;
1390         req->rq_net_err = 0;
1391         req->rq_timedout = 0;
1392         if (req->rq_bulk) {
1393                 __u64 old_xid = req->rq_xid;
1394
1395                 /* ensure previous bulk fails */
1396                 req->rq_xid = ptlrpc_next_xid();
1397                 CDEBUG(D_HA, "resend bulk old x"LPU64" new x"LPU64"\n",
1398                        old_xid, req->rq_xid);
1399         }
1400         ptlrpc_wake_client_req(req);
1401         spin_unlock(&req->rq_lock);
1402 }
1403
1404 /* XXX: this function and rq_status are currently unused */
1405 void ptlrpc_restart_req(struct ptlrpc_request *req)
1406 {
1407         DEBUG_REQ(D_HA, req, "restarting (possibly-)completed request");
1408         req->rq_status = -ERESTARTSYS;
1409
1410         spin_lock(&req->rq_lock);
1411         req->rq_restart = 1;
1412         req->rq_timedout = 0;
1413         ptlrpc_wake_client_req(req);
1414         spin_unlock(&req->rq_lock);
1415 }
1416
1417 static int expired_request(void *data)
1418 {
1419         struct ptlrpc_request *req = data;
1420         ENTRY;
1421
1422         RETURN(ptlrpc_expire_one_request(req));
1423 }
1424
1425 static void interrupted_request(void *data)
1426 {
1427         struct ptlrpc_request *req = data;
1428         DEBUG_REQ(D_HA, req, "request interrupted");
1429         spin_lock(&req->rq_lock);
1430         req->rq_intr = 1;
1431         spin_unlock(&req->rq_lock);
1432 }
1433
1434 struct ptlrpc_request *ptlrpc_request_addref(struct ptlrpc_request *req)
1435 {
1436         ENTRY;
1437         atomic_inc(&req->rq_refcount);
1438         RETURN(req);
1439 }
1440
1441 void ptlrpc_retain_replayable_request(struct ptlrpc_request *req,
1442                                       struct obd_import *imp)
1443 {
1444         struct list_head *tmp;
1445
1446         LASSERT_SPIN_LOCKED(&imp->imp_lock);
1447
1448         /* clear this for new requests that were resent as well
1449            as resent replayed requests. */
1450         lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT);
1451
1452         /* don't re-add requests that have been replayed */
1453         if (!list_empty(&req->rq_replay_list))
1454                 return;
1455
1456         lustre_msg_add_flags(req->rq_reqmsg, MSG_REPLAY);
1457
1458         LASSERT(imp->imp_replayable);
1459         /* Balanced in ptlrpc_free_committed, usually. */
1460         ptlrpc_request_addref(req);
1461         list_for_each_prev(tmp, &imp->imp_replay_list) {
1462                 struct ptlrpc_request *iter =
1463                         list_entry(tmp, struct ptlrpc_request, rq_replay_list);
1464
1465                 /* We may have duplicate transnos if we create and then
1466                  * open a file, or for closes retained if to match creating
1467                  * opens, so use req->rq_xid as a secondary key.
1468                  * (See bugs 684, 685, and 428.)
1469                  * XXX no longer needed, but all opens need transnos!
1470                  */
1471                 if (iter->rq_transno > req->rq_transno)
1472                         continue;
1473
1474                 if (iter->rq_transno == req->rq_transno) {
1475                         LASSERT(iter->rq_xid != req->rq_xid);
1476                         if (iter->rq_xid > req->rq_xid)
1477                                 continue;
1478                 }
1479
1480                 list_add(&req->rq_replay_list, &iter->rq_replay_list);
1481                 return;
1482         }
1483
1484         list_add_tail(&req->rq_replay_list, &imp->imp_replay_list);
1485 }
1486
1487 int ptlrpc_queue_wait(struct ptlrpc_request *req)
1488 {
1489         int rc = 0;
1490         int brc;
1491         struct l_wait_info lwi;
1492         struct obd_import *imp = req->rq_import;
1493         cfs_duration_t timeout = 0;
1494         ENTRY;
1495
1496         LASSERT(req->rq_set == NULL);
1497         LASSERT(!req->rq_receiving_reply);
1498         atomic_inc(&imp->imp_inflight);
1499
1500         /* for distributed debugging */
1501         lustre_msg_set_status(req->rq_reqmsg, cfs_curproc_pid());
1502         LASSERT(imp->imp_obd != NULL);
1503         CDEBUG(D_RPCTRACE, "Sending RPC pname:cluuid:pid:xid:nid:opc "
1504                "%s:%s:%d:"LPU64":%s:%d\n", cfs_curproc_comm(),
1505                imp->imp_obd->obd_uuid.uuid,
1506                lustre_msg_get_status(req->rq_reqmsg), req->rq_xid,
1507                libcfs_nid2str(imp->imp_connection->c_peer.nid),
1508                lustre_msg_get_opc(req->rq_reqmsg));
1509
1510         /* Mark phase here for a little debug help */
1511         req->rq_phase = RQ_PHASE_RPC;
1512
1513         spin_lock(&imp->imp_lock);
1514         req->rq_import_generation = imp->imp_generation;
1515 restart:
1516         if (ptlrpc_import_delay_req(imp, req, &rc)) {
1517                 list_del(&req->rq_list);
1518
1519                 list_add_tail(&req->rq_list, &imp->imp_delayed_list);
1520                 spin_unlock(&imp->imp_lock);
1521
1522                 DEBUG_REQ(D_HA, req, "\"%s\" waiting for recovery: (%s != %s)",
1523                           cfs_curproc_comm(),
1524                           ptlrpc_import_state_name(req->rq_send_state),
1525                           ptlrpc_import_state_name(imp->imp_state));
1526                 lwi = LWI_INTR(interrupted_request, req);
1527                 rc = l_wait_event(req->rq_reply_waitq,
1528                                   (req->rq_send_state == imp->imp_state ||
1529                                    req->rq_err || req->rq_intr),
1530                                   &lwi);
1531                 DEBUG_REQ(D_HA, req, "\"%s\" awake: (%s == %s or %d/%d == 1)",
1532                           cfs_curproc_comm(),
1533                           ptlrpc_import_state_name(imp->imp_state),
1534                           ptlrpc_import_state_name(req->rq_send_state),
1535                           req->rq_err, req->rq_intr);
1536
1537                 spin_lock(&imp->imp_lock);
1538                 list_del_init(&req->rq_list);
1539
1540                 if (req->rq_err) {
1541                         rc = -EIO;
1542                 }
1543                 else if (req->rq_intr) {
1544                         rc = -EINTR;
1545                 }
1546                 else if (req->rq_no_resend) {
1547                         spin_unlock(&imp->imp_lock);
1548                         GOTO(out, rc = -ETIMEDOUT);
1549                 }
1550                 else {
1551                         GOTO(restart, rc);
1552                 }
1553         }
1554
1555         if (rc != 0) {
1556                 list_del_init(&req->rq_list);
1557                 spin_unlock(&imp->imp_lock);
1558                 req->rq_status = rc; // XXX this ok?
1559                 GOTO(out, rc);
1560         }
1561
1562         if (req->rq_resend) {
1563                 lustre_msg_add_flags(req->rq_reqmsg, MSG_RESENT);
1564
1565                 if (req->rq_bulk != NULL) {
1566                         ptlrpc_unregister_bulk (req);
1567
1568                         /* bulk requests are supposed to be
1569                          * idempotent, so we are free to bump the xid
1570                          * here, which we need to do before
1571                          * registering the bulk again (bug 6371).
1572                          * print the old xid first for sanity.
1573                          */
1574                         DEBUG_REQ(D_HA, req, "bumping xid for bulk: ");
1575                         req->rq_xid = ptlrpc_next_xid();
1576                 }
1577
1578                 DEBUG_REQ(D_HA, req, "resending: ");
1579         }
1580
1581         /* XXX this is the same as ptlrpc_set_wait */
1582         LASSERT(list_empty(&req->rq_list));
1583         list_add_tail(&req->rq_list, &imp->imp_sending_list);
1584         spin_unlock(&imp->imp_lock);
1585
1586         rc = ptl_send_rpc(req, 0);
1587         if (rc) {
1588                 DEBUG_REQ(D_HA, req, "send failed (%d); recovering", rc);
1589                 timeout = CFS_TICK;
1590         } else {
1591                 timeout = cfs_timeout_cap(cfs_time_seconds(req->rq_timeout));
1592                 DEBUG_REQ(D_NET, req,
1593                           "-- sleeping for "CFS_DURATION_T" jiffies", timeout);
1594         }
1595         lwi = LWI_TIMEOUT_INTR(timeout, expired_request, interrupted_request,
1596                                req);
1597         l_wait_event(req->rq_reply_waitq, ptlrpc_check_reply(req), &lwi);
1598
1599         CDEBUG(D_RPCTRACE, "Completed RPC pname:cluuid:pid:xid:nid:opc "
1600                "%s:%s:%d:"LPU64":%s:%d\n", cfs_curproc_comm(),
1601                imp->imp_obd->obd_uuid.uuid,
1602                lustre_msg_get_status(req->rq_reqmsg), req->rq_xid,
1603                libcfs_nid2str(imp->imp_connection->c_peer.nid),
1604                lustre_msg_get_opc(req->rq_reqmsg));
1605
1606         spin_lock(&imp->imp_lock);
1607         list_del_init(&req->rq_list);
1608         spin_unlock(&imp->imp_lock);
1609
1610         /* If the reply was received normally, this just grabs the spinlock
1611          * (ensuring the reply callback has returned), sees that
1612          * req->rq_receiving_reply is clear and returns. */
1613         ptlrpc_unregister_reply (req);
1614
1615         if (req->rq_err)
1616                 GOTO(out, rc = -EIO);
1617
1618         /* Resend if we need to, unless we were interrupted. */
1619         if (req->rq_resend && !req->rq_intr) {
1620                 /* ...unless we were specifically told otherwise. */
1621                 if (req->rq_no_resend)
1622                         GOTO(out, rc = -ETIMEDOUT);
1623                 spin_lock(&imp->imp_lock);
1624                 goto restart;
1625         }
1626
1627         if (req->rq_intr) {
1628                 /* Should only be interrupted if we timed out. */
1629                 if (!req->rq_timedout)
1630                         DEBUG_REQ(D_ERROR, req,
1631                                   "rq_intr set but rq_timedout not");
1632                 GOTO(out, rc = -EINTR);
1633         }
1634
1635         if (req->rq_timedout) {                 /* non-recoverable timeout */
1636                 GOTO(out, rc = -ETIMEDOUT);
1637         }
1638
1639         if (!req->rq_replied) {
1640                 /* How can this be? -eeb */
1641                 DEBUG_REQ(D_ERROR, req, "!rq_replied: ");
1642                 LBUG();
1643                 GOTO(out, rc = req->rq_status);
1644         }
1645
1646         rc = after_reply(req);
1647         /* NB may return +ve success rc */
1648         if (req->rq_resend) {
1649                 spin_lock(&imp->imp_lock);
1650                 goto restart;
1651         }
1652
1653  out:
1654         if (req->rq_bulk != NULL) {
1655                 if (rc >= 0) {
1656                         /* success so far.  Note that anything going wrong
1657                          * with bulk now, is EXTREMELY strange, since the
1658                          * server must have believed that the bulk
1659                          * tranferred OK before she replied with success to
1660                          * me. */
1661                         lwi = LWI_TIMEOUT(timeout, NULL, NULL);
1662                         brc = l_wait_event(req->rq_reply_waitq,
1663                                            !ptlrpc_bulk_active(req->rq_bulk),
1664                                            &lwi);
1665                         LASSERT(brc == 0 || brc == -ETIMEDOUT);
1666                         if (brc != 0) {
1667                                 LASSERT(brc == -ETIMEDOUT);
1668                                 DEBUG_REQ(D_ERROR, req, "bulk timed out");
1669                                 rc = brc;
1670                         } else if (!req->rq_bulk->bd_success) {
1671                                 DEBUG_REQ(D_ERROR, req, "bulk transfer failed");
1672                                 rc = -EIO;
1673                         }
1674                 }
1675                 if (rc < 0)
1676                         ptlrpc_unregister_bulk (req);
1677         }
1678
1679         LASSERT(!req->rq_receiving_reply);
1680         req->rq_phase = RQ_PHASE_INTERPRET;
1681
1682         atomic_dec(&imp->imp_inflight);
1683         cfs_waitq_signal(&imp->imp_recovery_waitq);
1684         RETURN(rc);
1685 }
1686
1687 struct ptlrpc_replay_async_args {
1688         int praa_old_state;
1689         int praa_old_status;
1690 };
1691
1692 static int ptlrpc_replay_interpret(struct ptlrpc_request *req,
1693                                     void * data, int rc)
1694 {
1695         struct ptlrpc_replay_async_args *aa = data;
1696         struct obd_import *imp = req->rq_import;
1697
1698         ENTRY;
1699         atomic_dec(&imp->imp_replay_inflight);
1700
1701         if (!req->rq_replied) {
1702                 CERROR("request replay timed out, restarting recovery\n");
1703                 GOTO(out, rc = -ETIMEDOUT);
1704         }
1705
1706         if (lustre_msg_get_type(req->rq_repmsg) == PTL_RPC_MSG_ERR &&
1707             lustre_msg_get_status(req->rq_repmsg) == -ENOTCONN)
1708                 GOTO(out, rc = lustre_msg_get_status(req->rq_repmsg));
1709
1710         /* The transno had better not change over replay. */
1711         LASSERT(lustre_msg_get_transno(req->rq_reqmsg) ==
1712                 lustre_msg_get_transno(req->rq_repmsg));
1713
1714         DEBUG_REQ(D_HA, req, "got rep");
1715
1716         /* let the callback do fixups, possibly including in the request */
1717         if (req->rq_replay_cb)
1718                 req->rq_replay_cb(req);
1719
1720         if (req->rq_replied &&
1721             lustre_msg_get_status(req->rq_repmsg) != aa->praa_old_status) {
1722                 DEBUG_REQ(D_ERROR, req, "status %d, old was %d",
1723                           lustre_msg_get_status(req->rq_repmsg),
1724                           aa->praa_old_status);
1725         } else {
1726                 /* Put it back for re-replay. */
1727                 lustre_msg_set_status(req->rq_repmsg, aa->praa_old_status);
1728         }
1729
1730         spin_lock(&imp->imp_lock);
1731         imp->imp_last_replay_transno = req->rq_transno;
1732         spin_unlock(&imp->imp_lock);
1733
1734         /* continue with recovery */
1735         rc = ptlrpc_import_recovery_state_machine(imp);
1736  out:
1737         req->rq_send_state = aa->praa_old_state;
1738
1739         if (rc != 0)
1740                 /* this replay failed, so restart recovery */
1741                 ptlrpc_connect_import(imp, NULL);
1742
1743         RETURN(rc);
1744 }
1745
1746
1747 int ptlrpc_replay_req(struct ptlrpc_request *req)
1748 {
1749         struct ptlrpc_replay_async_args *aa;
1750         ENTRY;
1751
1752         LASSERT(req->rq_import->imp_state == LUSTRE_IMP_REPLAY);
1753
1754         /* Not handling automatic bulk replay yet (or ever?) */
1755         LASSERT(req->rq_bulk == NULL);
1756
1757         DEBUG_REQ(D_HA, req, "REPLAY");
1758
1759         LASSERT (sizeof (*aa) <= sizeof (req->rq_async_args));
1760         aa = (struct ptlrpc_replay_async_args *)&req->rq_async_args;
1761         memset(aa, 0, sizeof *aa);
1762
1763         /* Prepare request to be resent with ptlrpcd */
1764         aa->praa_old_state = req->rq_send_state;
1765         req->rq_send_state = LUSTRE_IMP_REPLAY;
1766         req->rq_phase = RQ_PHASE_NEW;
1767         aa->praa_old_status = lustre_msg_get_status(req->rq_repmsg);
1768         req->rq_status = 0;
1769
1770         req->rq_interpret_reply = ptlrpc_replay_interpret;
1771         atomic_inc(&req->rq_import->imp_replay_inflight);
1772         ptlrpc_request_addref(req); /* ptlrpcd needs a ref */
1773
1774         ptlrpcd_add_req(req);
1775         RETURN(0);
1776 }
1777
1778 void ptlrpc_abort_inflight(struct obd_import *imp)
1779 {
1780         struct list_head *tmp, *n;
1781         ENTRY;
1782
1783         /* Make sure that no new requests get processed for this import.
1784          * ptlrpc_{queue,set}_wait must (and does) hold imp_lock while testing
1785          * this flag and then putting requests on sending_list or delayed_list.
1786          */
1787         spin_lock(&imp->imp_lock);
1788
1789         /* XXX locking?  Maybe we should remove each request with the list
1790          * locked?  Also, how do we know if the requests on the list are
1791          * being freed at this time?
1792          */
1793         list_for_each_safe(tmp, n, &imp->imp_sending_list) {
1794                 struct ptlrpc_request *req =
1795                         list_entry(tmp, struct ptlrpc_request, rq_list);
1796
1797                 DEBUG_REQ(D_HA, req, "inflight");
1798
1799                 spin_lock (&req->rq_lock);
1800                 if (req->rq_import_generation < imp->imp_generation) {
1801                         req->rq_err = 1;
1802                         ptlrpc_wake_client_req(req);
1803                 }
1804                 spin_unlock (&req->rq_lock);
1805         }
1806
1807         list_for_each_safe(tmp, n, &imp->imp_delayed_list) {
1808                 struct ptlrpc_request *req =
1809                         list_entry(tmp, struct ptlrpc_request, rq_list);
1810
1811                 DEBUG_REQ(D_HA, req, "aborting waiting req");
1812
1813                 spin_lock (&req->rq_lock);
1814                 if (req->rq_import_generation < imp->imp_generation) {
1815                         req->rq_err = 1;
1816                         ptlrpc_wake_client_req(req);
1817                 }
1818                 spin_unlock (&req->rq_lock);
1819         }
1820
1821         /* Last chance to free reqs left on the replay list, but we
1822          * will still leak reqs that haven't committed.  */
1823         if (imp->imp_replayable)
1824                 ptlrpc_free_committed(imp);
1825
1826         spin_unlock(&imp->imp_lock);
1827
1828         EXIT;
1829 }
1830
1831 static __u64 ptlrpc_last_xid = 0;
1832 spinlock_t ptlrpc_last_xid_lock;
1833
1834 __u64 ptlrpc_next_xid(void)
1835 {
1836         __u64 tmp;
1837         spin_lock(&ptlrpc_last_xid_lock);
1838         tmp = ++ptlrpc_last_xid;
1839         spin_unlock(&ptlrpc_last_xid_lock);
1840         return tmp;
1841 }
1842
1843 __u64 ptlrpc_sample_next_xid(void)
1844 {
1845         __u64 tmp;
1846         spin_lock(&ptlrpc_last_xid_lock);
1847         tmp = ptlrpc_last_xid + 1;
1848         spin_unlock(&ptlrpc_last_xid_lock);
1849         return tmp;
1850 }
1851 EXPORT_SYMBOL(ptlrpc_sample_next_xid);