lustre/ptlrpc/client.c

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