lustre/ptlrpc/client.c

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