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