lustre/ptlrpc/client.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (c) 2002, 2003 Cluster File Systems, Inc.
   5  *
   6  *   This file is part of Lustre, http://www.lustre.org.
   7  *
   8  *   Lustre is free software; you can redistribute it and/or
   9  *   modify it under the terms of version 2 of the GNU General Public
  10  *   License as published by the Free Software Foundation.
  11  *
  12  *   Lustre is distributed in the hope that it will be useful,
  13  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  *   GNU General Public License for more details.
  16  *
  17  *   You should have received a copy of the GNU General Public License
  18  *   along with Lustre; if not, write to the Free Software
  19  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  20  *
  21  */
  22
  23 #define DEBUG_SUBSYSTEM S_RPC
  24 #ifndef __KERNEL__
  25 #include <errno.h>
  26 #include <signal.h>
  27 #include <liblustre.h>
  28 #endif
  29
  30 #include <linux/obd_support.h>
  31 #include <linux/obd_class.h>
  32 #include <linux/lustre_lib.h>
  33 #include <linux/lustre_ha.h>
  34 #include <linux/lustre_import.h>
  35
  36 #include "ptlrpc_internal.h"
  37
  38 void ptlrpc_init_client(int req_portal, int rep_portal, char *name,
  39                         struct ptlrpc_client *cl)
  40 {
  41         cl->cli_request_portal = req_portal;
  42         cl->cli_reply_portal   = rep_portal;
  43         cl->cli_name           = name;
  44 }
  45
  46 struct obd_uuid *ptlrpc_req_to_uuid(struct ptlrpc_request *req)
  47 {
  48         return &req->rq_connection->c_remote_uuid;
  49 }
  50
  51 struct ptlrpc_connection *ptlrpc_uuid_to_connection(struct obd_uuid *uuid)
  52 {
  53         struct ptlrpc_connection *c;
  54         struct ptlrpc_peer peer;
  55         int err;
  56
  57         err = ptlrpc_uuid_to_peer(uuid, &peer);
  58         if (err != 0) {
  59                 CERROR("cannot find peer %s!\n", uuid->uuid);
  60                 return NULL;
  61         }
  62
  63         c = ptlrpc_get_connection(&peer, uuid);
  64         if (c) {
  65                 memcpy(c->c_remote_uuid.uuid,
  66                        uuid->uuid, sizeof(c->c_remote_uuid.uuid));
  67                 c->c_epoch++;
  68         }
  69
  70         CDEBUG(D_INFO, "%s -> %p\n", uuid->uuid, c);
  71
  72         return c;
  73 }
  74
  75 void ptlrpc_readdress_connection(struct ptlrpc_connection *conn,
  76                                  struct obd_uuid *uuid)
  77 {
  78         struct ptlrpc_peer peer;
  79         int err;
  80
  81         err = ptlrpc_uuid_to_peer(uuid, &peer);
  82         if (err != 0) {
  83                 CERROR("cannot find peer %s!\n", uuid->uuid);
  84                 return;
  85         }
  86
  87         memcpy(&conn->c_peer, &peer, sizeof (peer));
  88         return;
  89 }
  90
  91 static inline struct ptlrpc_bulk_desc *new_bulk(void)
  92 {
  93         struct ptlrpc_bulk_desc *desc;
  94
  95         OBD_ALLOC(desc, sizeof(*desc));
  96         if (!desc)
  97                 return NULL;
  98
  99         spin_lock_init(&desc->bd_lock);
 100         init_waitqueue_head(&desc->bd_waitq);
 101         INIT_LIST_HEAD(&desc->bd_page_list);
 102         desc->bd_md_h = PTL_HANDLE_NONE;
 103         desc->bd_me_h = PTL_HANDLE_NONE;
 104
 105         return desc;
 106 }
 107
 108 struct ptlrpc_bulk_desc *ptlrpc_prep_bulk_imp (struct ptlrpc_request *req,
 109                                                int type, int portal)
 110 {
 111         struct obd_import *imp = req->rq_import;
 112         struct ptlrpc_bulk_desc *desc;
 113
 114         LASSERT(type == BULK_PUT_SINK || type == BULK_GET_SOURCE);
 115
 116         desc = new_bulk();
 117         if (desc == NULL)
 118                 RETURN(NULL);
 119
 120         desc->bd_import_generation = req->rq_import_generation;
 121         desc->bd_import = class_import_get(imp);
 122         desc->bd_req = req;
 123         desc->bd_type = type;
 124         desc->bd_portal = portal;
 125
 126         /* This makes req own desc, and free it when she frees herself */
 127         req->rq_bulk = desc;
 128
 129         return desc;
 130 }
 131
 132 struct ptlrpc_bulk_desc *ptlrpc_prep_bulk_exp (struct ptlrpc_request *req,
 133                                                int type, int portal)
 134 {
 135         struct obd_export *exp = req->rq_export;
 136         struct ptlrpc_bulk_desc *desc;
 137
 138         LASSERT(type == BULK_PUT_SOURCE || type == BULK_GET_SINK);
 139
 140         desc = new_bulk();
 141         if (desc == NULL)
 142                 RETURN(NULL);
 143
 144         desc->bd_export = class_export_get(exp);
 145         desc->bd_req = req;
 146         desc->bd_type = type;
 147         desc->bd_portal = portal;
 148
 149         /* NB we don't assign rq_bulk here; server-side requests are
 150          * re-used, and the handler frees the bulk desc explicitly. */
 151
 152         return desc;
 153 }
 154
 155 int ptlrpc_prep_bulk_page(struct ptlrpc_bulk_desc *desc,
 156                           struct page *page, int pageoffset, int len)
 157 {
 158         struct ptlrpc_bulk_page *bulk;
 159
 160         OBD_ALLOC(bulk, sizeof(*bulk));
 161         if (bulk == NULL)
 162                 return -ENOMEM;
 163
 164         LASSERT(page != NULL);
 165         LASSERT(pageoffset >= 0);
 166         LASSERT(len > 0);
 167         LASSERT(pageoffset + len <= PAGE_SIZE);
 168
 169         bulk->bp_page = page;
 170         bulk->bp_pageoffset = pageoffset;
 171         bulk->bp_buflen = len;
 172
 173         bulk->bp_desc = desc;
 174         list_add_tail(&bulk->bp_link, &desc->bd_page_list);
 175         desc->bd_page_count++;
 176         return 0;
 177 }
 178
 179 void ptlrpc_free_bulk(struct ptlrpc_bulk_desc *desc)
 180 {
 181         struct list_head *tmp, *next;
 182         ENTRY;
 183
 184         LASSERT(desc != NULL);
 185         LASSERT(desc->bd_page_count != 0x5a5a5a5a); /* not freed already */
 186         LASSERT(!desc->bd_network_rw);         /* network hands off or */
 187
 188         list_for_each_safe(tmp, next, &desc->bd_page_list) {
 189                 struct ptlrpc_bulk_page *bulk;
 190                 bulk = list_entry(tmp, struct ptlrpc_bulk_page, bp_link);
 191                 ptlrpc_free_bulk_page(bulk);
 192         }
 193
 194         LASSERT(desc->bd_page_count == 0);
 195         LASSERT((desc->bd_export != NULL) ^ (desc->bd_import != NULL));
 196
 197         if (desc->bd_export)
 198                 class_export_put(desc->bd_export);
 199         else
 200                 class_import_put(desc->bd_import);
 201
 202         OBD_FREE(desc, sizeof(*desc));
 203         EXIT;
 204 }
 205
 206 void ptlrpc_free_bulk_page(struct ptlrpc_bulk_page *bulk)
 207 {
 208         LASSERT(bulk != NULL);
 209
 210         list_del(&bulk->bp_link);
 211         bulk->bp_desc->bd_page_count--;
 212         OBD_FREE(bulk, sizeof(*bulk));
 213 }
 214
 215 struct ptlrpc_request *ptlrpc_prep_req(struct obd_import *imp, int opcode,
 216                                        int count, int *lengths, char **bufs)
 217 {
 218         struct ptlrpc_request *request;
 219         int rc;
 220         ENTRY;
 221
 222         LASSERT((unsigned long)imp > 0x1000);
 223
 224         OBD_ALLOC(request, sizeof(*request));
 225         if (!request) {
 226                 CERROR("request allocation out of memory\n");
 227                 RETURN(NULL);
 228         }
 229
 230         rc = lustre_pack_msg(count, lengths, bufs,
 231                              &request->rq_reqlen, &request->rq_reqmsg);
 232         if (rc) {
 233                 CERROR("cannot pack request %d\n", rc);
 234                 OBD_FREE(request, sizeof(*request));
 235                 RETURN(NULL);
 236         }
 237
 238         request->rq_timeout = obd_timeout;
 239         request->rq_level = LUSTRE_CONN_FULL;
 240         request->rq_type = PTL_RPC_MSG_REQUEST;
 241         request->rq_import = class_import_get(imp);
 242         request->rq_phase = RQ_PHASE_NEW;
 243
 244         /* XXX FIXME bug 249 */
 245         request->rq_request_portal = imp->imp_client->cli_request_portal;
 246         request->rq_reply_portal = imp->imp_client->cli_reply_portal;
 247
 248         request->rq_connection = ptlrpc_connection_addref(imp->imp_connection);
 249
 250         spin_lock_init(&request->rq_lock);
 251         INIT_LIST_HEAD(&request->rq_list);
 252         init_waitqueue_head(&request->rq_wait_for_rep);
 253         request->rq_xid = ptlrpc_next_xid();
 254         atomic_set(&request->rq_refcount, 1);
 255
 256         request->rq_reqmsg->opc = opcode;
 257         request->rq_reqmsg->flags = 0;
 258
 259         RETURN(request);
 260 }
 261
 262 struct ptlrpc_request_set *ptlrpc_prep_set(void)
 263 {
 264         struct ptlrpc_request_set *set;
 265
 266         OBD_ALLOC(set, sizeof *set);
 267         if (!set)
 268                 RETURN(NULL);
 269         INIT_LIST_HEAD(&set->set_requests);
 270         init_waitqueue_head(&set->set_waitq);
 271         set->set_remaining = 0;
 272
 273         RETURN(set);
 274 }
 275
 276 /* Finish with this set; opposite of prep_set. */
 277 void ptlrpc_set_destroy(struct ptlrpc_request_set *set)
 278 {
 279         struct list_head *tmp;
 280         struct list_head *next;
 281         int               expected_phase;
 282         int               n = 0;
 283         ENTRY;
 284
 285         /* Requests on the set should either all be completed, or all be new */
 286         expected_phase = (set->set_remaining == 0) ?
 287                          RQ_PHASE_COMPLETE : RQ_PHASE_NEW;
 288         list_for_each (tmp, &set->set_requests) {
 289                 struct ptlrpc_request *req =
 290                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 291
 292                 LASSERT(req->rq_phase == expected_phase);
 293                 n++;
 294         }
 295
 296         LASSERT(set->set_remaining == 0 || set->set_remaining == n);
 297
 298         list_for_each_safe(tmp, next, &set->set_requests) {
 299                 struct ptlrpc_request *req =
 300                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 301                 list_del_init(&req->rq_set_chain);
 302
 303                 LASSERT(req->rq_phase == expected_phase);
 304
 305                 if (req->rq_phase == RQ_PHASE_NEW) {
 306
 307                         if (req->rq_interpret_reply != NULL) {
 308                                 int (*interpreter)(struct ptlrpc_request *,
 309                                                    void *, int) =
 310                                         req->rq_interpret_reply;
 311
 312                                 /* higher level (i.e. LOV) failed;
 313                                  * let the sub reqs clean up */
 314                                 req->rq_status = -EBADR;
 315                                 interpreter(req, &req->rq_async_args,
 316                                             req->rq_status);
 317                         }
 318                         set->set_remaining--;
 319                 }
 320
 321                 req->rq_set = NULL;
 322                 ptlrpc_req_finished (req);
 323         }
 324
 325         LASSERT(set->set_remaining == 0);
 326
 327         OBD_FREE(set, sizeof(*set));
 328         EXIT;
 329 }
 330
 331 void ptlrpc_set_add_req(struct ptlrpc_request_set *set,
 332                         struct ptlrpc_request *req)
 333 {
 334         /* The set takes over the caller's request reference */
 335         list_add_tail(&req->rq_set_chain, &set->set_requests);
 336         req->rq_set = set;
 337         set->set_remaining++;
 338 }
 339
 340 static int ptlrpc_check_reply(struct ptlrpc_request *req)
 341 {
 342         unsigned long flags;
 343         int rc = 0;
 344         ENTRY;
 345
 346         /* serialise with network callback */
 347         spin_lock_irqsave (&req->rq_lock, flags);
 348
 349         if (req->rq_replied) {
 350                 DEBUG_REQ(D_NET, req, "REPLIED:");
 351                 GOTO(out, rc = 1);
 352         }
 353
 354         if (req->rq_err) {
 355                 DEBUG_REQ(D_ERROR, req, "ABORTED:");
 356                 GOTO(out, rc = 1);
 357         }
 358
 359         if (req->rq_resend) {
 360                 DEBUG_REQ(D_ERROR, req, "RESEND:");
 361                 GOTO(out, rc = 1);
 362         }
 363
 364         if (req->rq_restart) {
 365                 DEBUG_REQ(D_ERROR, req, "RESTART:");
 366                 GOTO(out, rc = 1);
 367         }
 368         EXIT;
 369  out:
 370         spin_unlock_irqrestore (&req->rq_lock, flags);
 371         DEBUG_REQ(D_NET, req, "rc = %d for", rc);
 372         return rc;
 373 }
 374
 375 static int ptlrpc_check_status(struct ptlrpc_request *req)
 376 {
 377         int err;
 378         ENTRY;
 379
 380         err = req->rq_repmsg->status;
 381         if (req->rq_repmsg->type == PTL_RPC_MSG_ERR) {
 382                 DEBUG_REQ(D_ERROR, req, "type == PTL_RPC_MSG_ERR (%d)", err);
 383                 if (err >= 0)
 384                         CERROR("Error Reply has >= zero status\n");
 385                 RETURN(err < 0 ? err : -EINVAL);
 386         }
 387
 388         if (err < 0) {
 389                 DEBUG_REQ(D_INFO, req, "status is %d", err);
 390         } else if (err > 0) {
 391                 /* XXX: translate this error from net to host */
 392                 DEBUG_REQ(D_INFO, req, "status is %d", err);
 393         }
 394
 395         RETURN(err);
 396 }
 397
 398 static int after_reply(struct ptlrpc_request *req, int *restartp)
 399 {
 400         unsigned long flags;
 401         struct obd_import *imp = req->rq_import;
 402         int rc;
 403         ENTRY;
 404
 405         LASSERT(!req->rq_receiving_reply);
 406         LASSERT(req->rq_replied);
 407
 408         if (restartp != NULL)
 409                 *restartp = 0;
 410
 411         /* NB Until this point, the whole of the incoming message,
 412          * including buflens, status etc is in the sender's byte order. */
 413
 414 #if SWAB_PARANOIA
 415         /* Clear reply swab mask; this is a new reply in sender's byte order */
 416         req->rq_rep_swab_mask = 0;
 417 #endif
 418         rc = lustre_unpack_msg(req->rq_repmsg, req->rq_replen);
 419         if (rc) {
 420                 CERROR("unpack_rep failed: %d\n", rc);
 421                 RETURN(-EPROTO);
 422         }
 423
 424         if (req->rq_repmsg->type != PTL_RPC_MSG_REPLY &&
 425             req->rq_repmsg->type != PTL_RPC_MSG_ERR) {
 426                 CERROR("invalid packet type received (type=%u)\n",
 427                        req->rq_repmsg->type);
 428                 RETURN(-EPROTO);
 429         }
 430
 431         /* Store transno in reqmsg for replay. */
 432         req->rq_reqmsg->transno = req->rq_transno = req->rq_repmsg->transno;
 433
 434         rc = ptlrpc_check_status(req);
 435
 436         /* Either we've been evicted, or the server has failed for
 437          * some reason. Try to reconnect, and if that fails, punt to the
 438          * upcall. */
 439         if (rc == -ENOTCONN) {
 440                 if (req->rq_level < LUSTRE_CONN_FULL || req->rq_no_recov ||
 441                     imp->imp_obd->obd_no_recov || imp->imp_dlm_fake) {
 442                         RETURN(-ENOTCONN);
 443                 }
 444
 445                 ptlrpc_request_handle_notconn(req);
 446
 447                 if (req->rq_err)
 448                         RETURN(-EIO);
 449
 450                 if (req->rq_no_resend)
 451                         RETURN(rc); /* -ENOTCONN */
 452
 453                 if (req->rq_resend) {
 454                         if (restartp == NULL)
 455                                 LBUG(); /* async resend not supported yet */
 456                         spin_lock_irqsave (&req->rq_lock, flags);
 457                         req->rq_resend = 0;
 458                         spin_unlock_irqrestore (&req->rq_lock, flags);
 459                         *restartp = 1;
 460                         lustre_msg_add_flags(req->rq_reqmsg, MSG_RESENT);
 461                         DEBUG_REQ(D_HA, req, "resending: ");
 462                         RETURN(0);
 463                 }
 464
 465                 CERROR("request should be err or resend: %p\n", req);
 466                 LBUG();
 467         }
 468
 469         if (req->rq_import->imp_replayable) {
 470                 spin_lock_irqsave(&imp->imp_lock, flags);
 471                 if ((req->rq_replay || req->rq_transno != 0) && rc >= 0)
 472                         ptlrpc_retain_replayable_request(req, imp);
 473
 474                 if (req->rq_transno > imp->imp_max_transno)
 475                         imp->imp_max_transno = req->rq_transno;
 476
 477                 /* Replay-enabled imports return commit-status information. */
 478                 if (req->rq_repmsg->last_committed)
 479                         imp->imp_peer_committed_transno =
 480                                 req->rq_repmsg->last_committed;
 481                 ptlrpc_free_committed(imp);
 482                 spin_unlock_irqrestore(&imp->imp_lock, flags);
 483         }
 484
 485         RETURN(rc);
 486 }
 487
 488 int ptlrpc_check_set(struct ptlrpc_request_set *set)
 489 {
 490         unsigned long flags;
 491         struct list_head *tmp;
 492         ENTRY;
 493
 494         if (set->set_remaining == 0)
 495                 RETURN(1);
 496
 497         list_for_each(tmp, &set->set_requests) {
 498                 struct ptlrpc_request *req =
 499                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 500                 struct obd_import *imp = req->rq_import;
 501                 int rc = 0;
 502
 503                 if (!(req->rq_phase == RQ_PHASE_RPC ||
 504                       req->rq_phase == RQ_PHASE_BULK ||
 505                       req->rq_phase == RQ_PHASE_INTERPRET ||
 506                       req->rq_phase == RQ_PHASE_COMPLETE)) {
 507                         DEBUG_REQ(D_ERROR, req, "bad phase %x", req->rq_phase);
 508                         LBUG();
 509                 }
 510
 511                 if (req->rq_phase == RQ_PHASE_COMPLETE)
 512                         continue;
 513
 514                 if (req->rq_phase == RQ_PHASE_INTERPRET)
 515                         GOTO(interpret, req->rq_status);
 516
 517                 if (req->rq_err) {
 518                         ptlrpc_unregister_reply(req);
 519                         if (req->rq_status == 0)
 520                                 req->rq_status = -EIO;
 521                         req->rq_phase = RQ_PHASE_INTERPRET;
 522
 523                         spin_lock_irqsave(&imp->imp_lock, flags);
 524                         list_del_init(&req->rq_list);
 525                         spin_unlock_irqrestore(&imp->imp_lock, flags);
 526
 527                         GOTO(interpret, req->rq_status);
 528                 }
 529
 530                 if (req->rq_intr) {
 531                         /* NB could be on delayed list */
 532                         ptlrpc_unregister_reply(req);
 533                         req->rq_status = -EINTR;
 534                         req->rq_phase = RQ_PHASE_INTERPRET;
 535
 536                         spin_lock_irqsave(&imp->imp_lock, flags);
 537                         list_del_init(&req->rq_list);
 538                         spin_unlock_irqrestore(&imp->imp_lock, flags);
 539
 540                         GOTO(interpret, req->rq_status);
 541                 }
 542
 543                 if (req->rq_phase == RQ_PHASE_RPC) {
 544                         int do_restart = 0;
 545                         if (req->rq_waiting || req->rq_resend) {
 546                                 spin_lock_irqsave(&imp->imp_lock, flags);
 547
 548                                 if (req->rq_level > imp->imp_level) {
 549                                         spin_unlock_irqrestore(&imp->imp_lock,
 550                                                                flags);
 551                                         continue;
 552                                 }
 553
 554                                 list_del(&req->rq_list);
 555                                 list_add_tail(&req->rq_list,
 556                                               &imp->imp_sending_list);
 557
 558                                 if (req->rq_import_generation <
 559                                     imp->imp_generation) {
 560                                         req->rq_status = -EIO;
 561                                         req->rq_phase = RQ_PHASE_INTERPRET;
 562                                         spin_unlock_irqrestore(&imp->imp_lock,
 563                                                                flags);
 564                                         GOTO(interpret, req->rq_status);
 565                                 }
 566                                 spin_unlock_irqrestore(&imp->imp_lock, flags);
 567
 568                                 req->rq_waiting = 0;
 569                                 if (req->rq_resend) {
 570                                         lustre_msg_add_flags(req->rq_reqmsg,
 571                                                              MSG_RESENT);
 572                                         spin_lock_irqsave(&req->rq_lock, flags);
 573                                         req->rq_resend = 0;
 574                                         spin_unlock_irqrestore(&req->rq_lock,
 575                                                                flags);
 576
 577                                         ptlrpc_unregister_reply(req);
 578                                         if (req->rq_bulk)
 579                                                 ptlrpc_unregister_bulk(req);
 580                                 }
 581
 582                                 rc = ptl_send_rpc(req);
 583                                 if (rc) {
 584                                         req->rq_status = rc;
 585                                         req->rq_phase = RQ_PHASE_INTERPRET;
 586                                         GOTO(interpret, req->rq_status);
 587                                 }
 588
 589                         }
 590
 591                         /* Ensure the network callback returned */
 592                         spin_lock_irqsave (&req->rq_lock, flags);
 593                         if (!req->rq_replied) {
 594                                 spin_unlock_irqrestore (&req->rq_lock, flags);
 595                                 continue;
 596                         }
 597                         spin_unlock_irqrestore (&req->rq_lock, flags);
 598
 599                         spin_lock_irqsave(&imp->imp_lock, flags);
 600                         list_del_init(&req->rq_list);
 601                         spin_unlock_irqrestore(&imp->imp_lock, flags);
 602
 603                         req->rq_status = after_reply(req, &do_restart);
 604                         if (do_restart) {
 605                                 spin_lock_irqsave (&req->rq_lock, flags);
 606                                 req->rq_resend = 1; /* ugh */
 607                                 spin_unlock_irqrestore (&req->rq_lock, flags);
 608                                 continue;
 609                         }
 610
 611                         /* If there is no bulk associated with this request,
 612                          * then we're done and should let the interpreter
 613                          * process the reply.  Similarly if the RPC returned
 614                          * an error, and therefore the bulk will never arrive.
 615                          */
 616                         if (req->rq_bulk == NULL || req->rq_status != 0) {
 617                                 req->rq_phase = RQ_PHASE_INTERPRET;
 618                                 GOTO(interpret, req->rq_status);
 619                         }
 620
 621                         req->rq_phase = RQ_PHASE_BULK;
 622                 }
 623
 624                 LASSERT(req->rq_phase == RQ_PHASE_BULK);
 625                 if (!ptlrpc_bulk_complete (req->rq_bulk))
 626                         continue;
 627
 628                 req->rq_phase = RQ_PHASE_INTERPRET;
 629
 630         interpret:
 631                 LASSERT(req->rq_phase == RQ_PHASE_INTERPRET);
 632                 LASSERT(!req->rq_receiving_reply);
 633
 634                 ptlrpc_unregister_reply(req);
 635                 if (req->rq_bulk != NULL)
 636                         ptlrpc_unregister_bulk (req);
 637
 638                 if (req->rq_interpret_reply != NULL) {
 639                         int (*interpreter)(struct ptlrpc_request *,void *,int) =
 640                                 req->rq_interpret_reply;
 641                         req->rq_status = interpreter(req, &req->rq_async_args,
 642                                                      req->rq_status);
 643                 }
 644
 645                 CDEBUG(D_RPCTRACE, "Completed RPC pname:cluuid:pid:xid:ni:nid:"
 646                        "opc %s:%s:%d:"LPU64":%s:"LPX64":%d\n", current->comm,
 647                        imp->imp_obd->obd_uuid.uuid, req->rq_reqmsg->status,
 648                        req->rq_xid,
 649                        imp->imp_connection->c_peer.peer_ni->pni_name,
 650                        imp->imp_connection->c_peer.peer_nid,
 651                        req->rq_reqmsg->opc);
 652
 653                 req->rq_phase = RQ_PHASE_COMPLETE;
 654                 set->set_remaining--;
 655         }
 656
 657         RETURN(set->set_remaining == 0);
 658 }
 659
 660 int ptlrpc_expire_one_request(struct ptlrpc_request *req)
 661 {
 662         unsigned long      flags;
 663         struct obd_import *imp = req->rq_import;
 664         ENTRY;
 665
 666         DEBUG_REQ(D_ERROR, req, "timeout");
 667
 668         spin_lock_irqsave (&req->rq_lock, flags);
 669         req->rq_timedout = 1;
 670         spin_unlock_irqrestore (&req->rq_lock, flags);
 671
 672         ptlrpc_unregister_reply (req);
 673
 674         if (imp == NULL) {
 675                 DEBUG_REQ(D_HA, req, "NULL import: already cleaned up?");
 676                 RETURN(1);
 677         }
 678
 679         /* The DLM server doesn't want recovery run on its imports. */
 680         if (imp->imp_dlm_fake)
 681                 RETURN(1);
 682
 683         /* If this request is for recovery or other primordial tasks,
 684          * don't go back to sleep, and don't start recovery again.. */
 685         if (req->rq_level < LUSTRE_CONN_FULL || req->rq_no_recov ||
 686             imp->imp_obd->obd_no_recov)
 687                 RETURN(1);
 688
 689         ptlrpc_fail_import(imp, req->rq_import_generation);
 690
 691         RETURN(0);
 692 }
 693
 694 static int expired_set(void *data)
 695 {
 696         struct ptlrpc_request_set *set = data;
 697         struct list_head          *tmp;
 698         time_t                     now = LTIME_S (CURRENT_TIME);
 699         ENTRY;
 700
 701         LASSERT(set != NULL);
 702
 703         /* A timeout expired; see which reqs it applies to... */
 704         list_for_each (tmp, &set->set_requests) {
 705                 struct ptlrpc_request *req =
 706                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 707
 708                 /* request in-flight? */
 709                 if (!((req->rq_phase == RQ_PHASE_RPC && !req->rq_waiting) ||
 710                       (req->rq_phase == RQ_PHASE_BULK)))
 711                         continue;
 712
 713                 if (req->rq_timedout ||           /* already dealt with */
 714                     req->rq_sent + req->rq_timeout > now) /* not expired */
 715                         continue;
 716
 717                 /* deal with this guy */
 718                 ptlrpc_expire_one_request (req);
 719         }
 720
 721         /* When waiting for a whole set, we always to break out of the
 722          * sleep so we can recalculate the timeout, or enable interrupts
 723          * iff everyone's timed out.
 724          */
 725         RETURN(1);
 726 }
 727
 728 static void interrupted_set(void *data)
 729 {
 730         struct ptlrpc_request_set *set = data;
 731         struct list_head *tmp;
 732         unsigned long flags;
 733
 734         LASSERT(set != NULL);
 735         CERROR("INTERRUPTED SET %p\n", set);
 736
 737         list_for_each(tmp, &set->set_requests) {
 738                 struct ptlrpc_request *req =
 739                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 740
 741                 if (req->rq_phase != RQ_PHASE_RPC)
 742                         continue;
 743
 744                 spin_lock_irqsave (&req->rq_lock, flags);
 745                 req->rq_intr = 1;
 746                 spin_unlock_irqrestore (&req->rq_lock, flags);
 747         }
 748 }
 749
 750 int ptlrpc_set_wait(struct ptlrpc_request_set *set)
 751 {
 752         struct list_head      *tmp;
 753         struct obd_import     *imp;
 754         struct ptlrpc_request *req;
 755         struct l_wait_info     lwi;
 756         unsigned long          flags;
 757         int                    rc;
 758         time_t                 now;
 759         time_t                 deadline;
 760         int                    timeout;
 761         ENTRY;
 762
 763         SIGNAL_MASK_ASSERT(); /* XXX BUG 1511 */
 764         LASSERT(!list_empty(&set->set_requests));
 765         list_for_each(tmp, &set->set_requests) {
 766                 req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 767
 768                 LASSERT(req->rq_level == LUSTRE_CONN_FULL);
 769                 LASSERT(req->rq_phase == RQ_PHASE_NEW);
 770                 req->rq_phase = RQ_PHASE_RPC;
 771
 772                 imp = req->rq_import;
 773                 spin_lock_irqsave(&imp->imp_lock, flags);
 774
 775                 if (imp->imp_invalid) {
 776                         spin_unlock_irqrestore(&imp->imp_lock, flags);
 777                         req->rq_status = -EIO;
 778                         req->rq_phase = RQ_PHASE_INTERPRET;
 779                         continue;
 780                 }
 781
 782                 req->rq_import_generation = imp->imp_generation;
 783
 784                 if (req->rq_level > imp->imp_level) {
 785                         if (req->rq_no_recov || imp->imp_obd->obd_no_recov ||
 786                             imp->imp_dlm_fake) {
 787                                 spin_unlock_irqrestore(&imp->imp_lock, flags);
 788                                 req->rq_status = -EWOULDBLOCK;
 789                                 req->rq_phase = RQ_PHASE_INTERPRET;
 790                                 continue;
 791                         }
 792
 793                         spin_lock (&req->rq_lock);
 794                         req->rq_waiting = 1;
 795                         spin_unlock (&req->rq_lock);
 796                         LASSERT(list_empty (&req->rq_list));
 797                         // list_del(&req->rq_list);
 798                         list_add_tail(&req->rq_list, &imp->imp_delayed_list);
 799                         spin_unlock_irqrestore(&imp->imp_lock, flags);
 800                         continue;
 801                 }
 802
 803                 /* XXX this is the same as ptlrpc_queue_wait */
 804                 LASSERT(list_empty(&req->rq_list));
 805                 list_add_tail(&req->rq_list, &imp->imp_sending_list);
 806                 spin_unlock_irqrestore(&imp->imp_lock, flags);
 807
 808                 req->rq_reqmsg->status = current->pid;
 809                 CDEBUG(D_RPCTRACE, "Sending RPC pname:cluuid:pid:xid:ni:nid:opc"
 810                        " %s:%s:%d:"LPU64":%s:"LPX64":%d\n", current->comm,
 811                        imp->imp_obd->obd_uuid.uuid, req->rq_reqmsg->status,
 812                        req->rq_xid,
 813                        imp->imp_connection->c_peer.peer_ni->pni_name,
 814                        imp->imp_connection->c_peer.peer_nid,
 815                        req->rq_reqmsg->opc);
 816
 817                 rc = ptl_send_rpc(req);
 818                 if (rc) {
 819                         req->rq_status = rc;
 820                         req->rq_phase = RQ_PHASE_INTERPRET;
 821                 }
 822         }
 823
 824         do {
 825                 now = LTIME_S (CURRENT_TIME);
 826                 timeout = 0;
 827                 list_for_each (tmp, &set->set_requests) {
 828                         req = list_entry(tmp, struct ptlrpc_request,
 829                                          rq_set_chain);
 830
 831                         /* request in-flight? */
 832                         if (!((req->rq_phase == RQ_PHASE_RPC &&
 833                                !req->rq_waiting) ||
 834                               (req->rq_phase == RQ_PHASE_BULK)))
 835                                 continue;
 836
 837                         if (req->rq_timedout)   /* already timed out */
 838                                 continue;
 839
 840                         deadline = req->rq_sent + req->rq_timeout;
 841                         if (deadline <= now)    /* actually expired already */
 842                                 timeout = 1;    /* ASAP */
 843                         else if (timeout == 0 || timeout > deadline - now)
 844                                 timeout = deadline - now;
 845                 }
 846
 847                 /* wait until all complete, interrupted, or an in-flight
 848                  * req times out */
 849                 CDEBUG(D_HA, "set %p going to sleep for %d seconds\n",
 850                        set, timeout);
 851                 lwi = LWI_TIMEOUT_INTR((timeout ? timeout : 1) * HZ,
 852                                        expired_set, interrupted_set, set);
 853                 rc = l_wait_event(set->set_waitq, ptlrpc_check_set(set), &lwi);
 854
 855                 LASSERT(rc == 0 || rc == -EINTR || rc == -ETIMEDOUT);
 856
 857                 /* -EINTR => all requests have been flagged rq_intr so next
 858                  * check completes.
 859                  * -ETIMEOUTD => someone timed out.  When all reqs have
 860                  * timed out, signals are enabled allowing completion with
 861                  * EINTR.
 862                  * I don't really care if we go once more round the loop in
 863                  * the error cases -eeb. */
 864         } while (rc != 0);
 865
 866         LASSERT(set->set_remaining == 0);
 867
 868         rc = 0;
 869         list_for_each(tmp, &set->set_requests) {
 870                 req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 871
 872                 LASSERT(req->rq_phase == RQ_PHASE_COMPLETE);
 873                 if (req->rq_status != 0)
 874                         rc = req->rq_status;
 875         }
 876
 877         if (set->set_interpret != NULL) {
 878                 int (*interpreter)(struct ptlrpc_request_set *set,void *,int) =
 879                         set->set_interpret;
 880                 rc = interpreter (set, &set->set_args, rc);
 881         }
 882
 883         RETURN(rc);
 884 }
 885
 886 static void __ptlrpc_free_req(struct ptlrpc_request *request, int locked)
 887 {
 888         ENTRY;
 889         if (request == NULL) {
 890                 EXIT;
 891                 return;
 892         }
 893
 894         LASSERT(!request->rq_receiving_reply);
 895
 896         /* We must take it off the imp_replay_list first.  Otherwise, we'll set
 897          * request->rq_reqmsg to NULL while osc_close is dereferencing it. */
 898         if (request->rq_import != NULL) {
 899                 unsigned long flags = 0;
 900                 if (!locked)
 901                         spin_lock_irqsave(&request->rq_import->imp_lock, flags);
 902                 list_del_init(&request->rq_list);
 903                 if (!locked)
 904                         spin_unlock_irqrestore(&request->rq_import->imp_lock,
 905                                                flags);
 906         }
 907
 908         if (atomic_read(&request->rq_refcount) != 0) {
 909                 DEBUG_REQ(D_ERROR, request,
 910                           "freeing request with nonzero refcount");
 911                 LBUG();
 912         }
 913
 914         if (request->rq_repmsg != NULL) {
 915                 OBD_FREE(request->rq_repmsg, request->rq_replen);
 916                 request->rq_repmsg = NULL;
 917         }
 918         if (request->rq_reqmsg != NULL) {
 919                 OBD_FREE(request->rq_reqmsg, request->rq_reqlen);
 920                 request->rq_reqmsg = NULL;
 921         }
 922         if (request->rq_export != NULL) {
 923                 class_export_put(request->rq_export);
 924                 request->rq_export = NULL;
 925         }
 926         if (request->rq_import != NULL) {
 927                 class_import_put(request->rq_import);
 928                 request->rq_import = NULL;
 929         }
 930         if (request->rq_bulk != NULL)
 931                 ptlrpc_free_bulk(request->rq_bulk);
 932
 933         ptlrpc_put_connection(request->rq_connection);
 934         OBD_FREE(request, sizeof(*request));
 935         EXIT;
 936 }
 937
 938 void ptlrpc_free_req(struct ptlrpc_request *request)
 939 {
 940         __ptlrpc_free_req(request, 0);
 941 }
 942
 943 static int __ptlrpc_req_finished(struct ptlrpc_request *request, int locked)
 944 {
 945         ENTRY;
 946         if (request == NULL)
 947                 RETURN(1);
 948
 949         if (request == (void *)(long)(0x5a5a5a5a5a5a5a5a) ||
 950             request->rq_obd == (void *)(long)(0x5a5a5a5a5a5a5a5a)) {
 951                 CERROR("dereferencing freed request (bug 575)\n");
 952                 LBUG();
 953                 RETURN(1);
 954         }
 955
 956         DEBUG_REQ(D_INFO, request, "refcount now %u",
 957                   atomic_read(&request->rq_refcount) - 1);
 958
 959         if (atomic_dec_and_test(&request->rq_refcount)) {
 960                 __ptlrpc_free_req(request, locked);
 961                 RETURN(1);
 962         }
 963
 964         RETURN(0);
 965 }
 966
 967 void ptlrpc_req_finished(struct ptlrpc_request *request)
 968 {
 969         __ptlrpc_req_finished(request, 0);
 970 }
 971
 972 static void ptlrpc_cleanup_request_buf(struct ptlrpc_request *request)
 973 {
 974         OBD_FREE(request->rq_reqmsg, request->rq_reqlen);
 975         request->rq_reqmsg = NULL;
 976         request->rq_reqlen = 0;
 977 }
 978
 979 /* Disengage the client's reply buffer from the network
 980  * NB does _NOT_ unregister any client-side bulk.
 981  * IDEMPOTENT, but _not_ safe against concurrent callers.
 982  * The request owner (i.e. the thread doing the I/O) must call...
 983  */
 984 void ptlrpc_unregister_reply (struct ptlrpc_request *request)
 985 {
 986         unsigned long flags;
 987         int           rc;
 988         ENTRY;
 989
 990         LASSERT(!in_interrupt ());             /* might sleep */
 991
 992         spin_lock_irqsave (&request->rq_lock, flags);
 993         if (!request->rq_receiving_reply) {     /* not waiting for a reply */
 994                 spin_unlock_irqrestore (&request->rq_lock, flags);
 995                 EXIT;
 996                 /* NB reply buffer not freed here */
 997                 return;
 998         }
 999
1000         LASSERT(!request->rq_replied);         /* callback hasn't completed */
1001         spin_unlock_irqrestore (&request->rq_lock, flags);
1002
1003         rc = PtlMDUnlink (request->rq_reply_md_h);
1004         switch (rc) {
1005         default:
1006                 LBUG ();
1007
1008         case PTL_OK:                            /* unlinked before completion */
1009                 LASSERT(request->rq_receiving_reply);
1010                 LASSERT(!request->rq_replied);
1011                 spin_lock_irqsave (&request->rq_lock, flags);
1012                 request->rq_receiving_reply = 0;
1013                 spin_unlock_irqrestore (&request->rq_lock, flags);
1014                 OBD_FREE(request->rq_repmsg, request->rq_replen);
1015                 request->rq_repmsg = NULL;
1016                 EXIT;
1017                 return;
1018
1019         case PTL_MD_INUSE:                      /* callback in progress */
1020                 for (;;) {
1021                         /* Network access will complete in finite time but
1022                          * the timeout lets us CERROR for visibility */
1023                         struct l_wait_info lwi = LWI_TIMEOUT(10*HZ, NULL, NULL);
1024
1025                         rc = l_wait_event (request->rq_wait_for_rep,
1026                                            request->rq_replied, &lwi);
1027                         LASSERT(rc == 0 || rc == -ETIMEDOUT);
1028                         if (rc == 0) {
1029                                 spin_lock_irqsave (&request->rq_lock, flags);
1030                                 /* Ensure the callback has completed scheduling
1031                                  * me and taken its hands off the request */
1032                                 spin_unlock_irqrestore(&request->rq_lock,flags);
1033                                 break;
1034                         }
1035
1036                         CERROR ("Unexpectedly long timeout: req %p\n", request);
1037                 }
1038                 /* fall through */
1039
1040         case PTL_INV_MD:                        /* callback completed */
1041                 LASSERT(!request->rq_receiving_reply);
1042                 LASSERT(request->rq_replied);
1043                 EXIT;
1044                 return;
1045         }
1046         /* Not Reached */
1047 }
1048
1049 /* caller must hold imp->imp_lock */
1050 void ptlrpc_free_committed(struct obd_import *imp)
1051 {
1052         struct list_head *tmp, *saved;
1053         struct ptlrpc_request *req;
1054         struct ptlrpc_request *last_req = NULL; /* temporary fire escape */
1055         ENTRY;
1056
1057         LASSERT(imp != NULL);
1058
1059 #ifdef CONFIG_SMP
1060         LASSERT(spin_is_locked(&imp->imp_lock));
1061 #endif
1062
1063         CDEBUG(D_HA, "%s: committing for last_committed "LPU64"\n",
1064                imp->imp_obd->obd_name, imp->imp_peer_committed_transno);
1065
1066         list_for_each_safe(tmp, saved, &imp->imp_replay_list) {
1067                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
1068
1069                 /* XXX ok to remove when 1357 resolved - rread 05/29/03  */
1070                 LASSERT(req != last_req);
1071                 last_req = req;
1072
1073                 if (req->rq_import_generation < imp->imp_generation) {
1074                         DEBUG_REQ(D_HA, req, "freeing request with old gen");
1075                         GOTO(free_req, 0);
1076                 }
1077
1078                 if (req->rq_replay) {
1079                         DEBUG_REQ(D_HA, req, "keeping (FL_REPLAY)");
1080                         continue;
1081                 }
1082
1083                 /* not yet committed */
1084                 if (req->rq_transno > imp->imp_peer_committed_transno) {
1085                         DEBUG_REQ(D_HA, req, "stopping search");
1086                         break;
1087                 }
1088
1089                 DEBUG_REQ(D_HA, req, "committing (last_committed "LPU64")",
1090                           imp->imp_peer_committed_transno);
1091 free_req:
1092                 list_del_init(&req->rq_list);
1093                 __ptlrpc_req_finished(req, 1);
1094         }
1095
1096         EXIT;
1097         return;
1098 }
1099
1100 void ptlrpc_cleanup_client(struct obd_import *imp)
1101 {
1102         ENTRY;
1103         EXIT;
1104         return;
1105 }
1106
1107 void ptlrpc_resend_req(struct ptlrpc_request *req)
1108 {
1109         unsigned long flags;
1110
1111         DEBUG_REQ(D_HA, req, "resending");
1112         req->rq_reqmsg->handle.cookie = 0;
1113         ptlrpc_put_connection(req->rq_connection);
1114         req->rq_connection =
1115                 ptlrpc_connection_addref(req->rq_import->imp_connection);
1116         req->rq_status = -EAGAIN;
1117
1118         spin_lock_irqsave (&req->rq_lock, flags);
1119         req->rq_resend = 1;
1120         req->rq_timedout = 0;
1121         if (req->rq_set != NULL)
1122                 wake_up (&req->rq_set->set_waitq);
1123         else
1124                 wake_up(&req->rq_wait_for_rep);
1125         spin_unlock_irqrestore (&req->rq_lock, flags);
1126 }
1127
1128 /* XXX: this function and rq_status are currently unused */
1129 void ptlrpc_restart_req(struct ptlrpc_request *req)
1130 {
1131         unsigned long flags;
1132
1133         DEBUG_REQ(D_HA, req, "restarting (possibly-)completed request");
1134         req->rq_status = -ERESTARTSYS;
1135
1136         spin_lock_irqsave (&req->rq_lock, flags);
1137         req->rq_restart = 1;
1138         req->rq_timedout = 0;
1139         if (req->rq_set != NULL)
1140                 wake_up (&req->rq_set->set_waitq);
1141         else
1142                 wake_up(&req->rq_wait_for_rep);
1143         spin_unlock_irqrestore (&req->rq_lock, flags);
1144 }
1145
1146 static int expired_request(void *data)
1147 {
1148         struct ptlrpc_request *req = data;
1149         ENTRY;
1150
1151         RETURN(ptlrpc_expire_one_request(req));
1152 }
1153
1154 static void interrupted_request(void *data)
1155 {
1156         unsigned long flags;
1157
1158         struct ptlrpc_request *req = data;
1159         DEBUG_REQ(D_HA, req, "request interrupted");
1160         spin_lock_irqsave (&req->rq_lock, flags);
1161         req->rq_intr = 1;
1162         spin_unlock_irqrestore (&req->rq_lock, flags);
1163 }
1164
1165 struct ptlrpc_request *ptlrpc_request_addref(struct ptlrpc_request *req)
1166 {
1167         ENTRY;
1168         atomic_inc(&req->rq_refcount);
1169         RETURN(req);
1170 }
1171
1172 void ptlrpc_retain_replayable_request(struct ptlrpc_request *req,
1173                                       struct obd_import *imp)
1174 {
1175         struct list_head *tmp;
1176
1177 #ifdef CONFIG_SMP
1178         LASSERT(spin_is_locked(&imp->imp_lock));
1179 #endif
1180
1181         LASSERT(imp->imp_replayable);
1182         /* Balanced in ptlrpc_free_committed, usually. */
1183         ptlrpc_request_addref(req);
1184         list_for_each_prev(tmp, &imp->imp_replay_list) {
1185                 struct ptlrpc_request *iter =
1186                         list_entry(tmp, struct ptlrpc_request, rq_list);
1187
1188                 /* We may have duplicate transnos if we create and then
1189                  * open a file, or for closes retained if to match creating
1190                  * opens, so use req->rq_xid as a secondary key.
1191                  * (See bugs 684, 685, and 428.)
1192                  * XXX no longer needed, but all opens need transnos!
1193                  */
1194                 if (iter->rq_transno > req->rq_transno)
1195                         continue;
1196
1197                 if (iter->rq_transno == req->rq_transno) {
1198                         LASSERT(iter->rq_xid != req->rq_xid);
1199                         if (iter->rq_xid > req->rq_xid)
1200                                 continue;
1201                 }
1202
1203                 list_add(&req->rq_list, &iter->rq_list);
1204                 return;
1205         }
1206
1207         list_add_tail(&req->rq_list, &imp->imp_replay_list);
1208 }
1209
1210 int ptlrpc_queue_wait(struct ptlrpc_request *req)
1211 {
1212         int rc = 0;
1213         int brc;
1214         struct l_wait_info lwi;
1215         struct obd_import *imp = req->rq_import;
1216         struct obd_device *obd = imp->imp_obd;
1217         unsigned long flags;
1218         int do_restart = 0;
1219         int timeout = 0;
1220         ENTRY;
1221
1222         LASSERT(req->rq_set == NULL);
1223         LASSERT(!req->rq_receiving_reply);
1224
1225         /* for distributed debugging */
1226         req->rq_reqmsg->status = current->pid;
1227         LASSERT(imp->imp_obd != NULL);
1228         CDEBUG(D_RPCTRACE, "Sending RPC pname:cluuid:pid:xid:ni:nid:opc "
1229                "%s:%s:%d:"LPU64":%s:"LPX64":%d\n", current->comm,
1230                imp->imp_obd->obd_uuid.uuid,
1231                req->rq_reqmsg->status, req->rq_xid,
1232                imp->imp_connection->c_peer.peer_ni->pni_name,
1233                imp->imp_connection->c_peer.peer_nid,
1234                req->rq_reqmsg->opc);
1235
1236         /* Mark phase here for a little debug help */
1237         req->rq_phase = RQ_PHASE_RPC;
1238
1239         spin_lock_irqsave(&imp->imp_lock, flags);
1240         req->rq_import_generation = imp->imp_generation;
1241 restart:
1242         /*
1243          * If the import has been invalidated (such as by an OST failure), the
1244          * request must fail with -EIO.  Recovery requests are allowed to go
1245          * through, though, so that they have a chance to revalidate the
1246          * import.
1247          */
1248         if (req->rq_import->imp_invalid && req->rq_level == LUSTRE_CONN_FULL) {
1249                 DEBUG_REQ(D_ERROR, req, "IMP_INVALID:");
1250                 spin_unlock_irqrestore(&imp->imp_lock, flags);
1251                 GOTO(out, rc = -EIO);
1252         }
1253
1254         if (req->rq_import_generation < imp->imp_generation) {
1255                 DEBUG_REQ(D_ERROR, req, "req old gen:");
1256                 spin_unlock_irqrestore(&imp->imp_lock, flags);
1257                 GOTO(out, rc = -EIO);
1258         }
1259
1260         if (req->rq_level > imp->imp_level) {
1261                 list_del(&req->rq_list);
1262                 if (req->rq_no_recov || obd->obd_no_recov ||
1263                     imp->imp_dlm_fake) {
1264                         spin_unlock_irqrestore(&imp->imp_lock, flags);
1265                         GOTO(out, rc = -EWOULDBLOCK);
1266                 }
1267
1268                 list_add_tail(&req->rq_list, &imp->imp_delayed_list);
1269                 spin_unlock_irqrestore(&imp->imp_lock, flags);
1270
1271                 DEBUG_REQ(D_HA, req, "\"%s\" waiting for recovery: (%d > %d)",
1272                           current->comm, req->rq_level, imp->imp_level);
1273                 lwi = LWI_INTR(NULL, NULL);
1274                 rc = l_wait_event(req->rq_wait_for_rep,
1275                                   (req->rq_level <= imp->imp_level ||
1276                                    req->rq_err),
1277                                   &lwi);
1278                 DEBUG_REQ(D_HA, req, "\"%s\" awake: (%d > %d or %d == 1)",
1279                           current->comm, imp->imp_level, req->rq_level,
1280                           req->rq_err);
1281
1282                 spin_lock_irqsave(&imp->imp_lock, flags);
1283                 list_del_init(&req->rq_list);
1284
1285                 if (req->rq_err ||
1286                     req->rq_import_generation < imp->imp_generation)
1287                         rc = -EIO;
1288
1289                 if (rc) {
1290                         spin_unlock_irqrestore(&imp->imp_lock, flags);
1291                         GOTO(out, rc);
1292                 }
1293
1294                 DEBUG_REQ(D_HA, req, "resumed");
1295         }
1296
1297         /* XXX this is the same as ptlrpc_set_wait */
1298         LASSERT(list_empty(&req->rq_list));
1299         list_add_tail(&req->rq_list, &imp->imp_sending_list);
1300         spin_unlock_irqrestore(&imp->imp_lock, flags);
1301
1302         rc = ptl_send_rpc(req);
1303         if (rc) {
1304                 /* The DLM's fake imports want to avoid all forms of
1305                  * recovery. */
1306                 if (imp->imp_dlm_fake) {
1307                         spin_lock_irqsave(&imp->imp_lock, flags);
1308                         list_del_init(&req->rq_list);
1309                         spin_unlock_irqrestore(&imp->imp_lock, flags);
1310                         GOTO(out, rc);
1311                 }
1312
1313                 DEBUG_REQ(D_ERROR, req, "send failed (%d); recovering", rc);
1314
1315                 ptlrpc_fail_import(imp, req->rq_import_generation);
1316
1317                 /* If we've been told to not wait, we're done. */
1318                 if (req->rq_level < LUSTRE_CONN_FULL || req->rq_no_recov ||
1319                     obd->obd_no_recov) {
1320                         spin_lock_irqsave(&imp->imp_lock, flags);
1321                         list_del_init(&req->rq_list);
1322                         spin_unlock_irqrestore(&imp->imp_lock, flags);
1323                         GOTO(out, rc);
1324                 }
1325
1326                 /* If we errored, allow the user to interrupt immediately */
1327                 timeout = 1;
1328         } else {
1329                 timeout = req->rq_timeout * HZ;
1330                 DEBUG_REQ(D_NET, req, "-- sleeping");
1331         }
1332 #ifdef __KERNEL__
1333         lwi = LWI_TIMEOUT_INTR(timeout, expired_request, interrupted_request,
1334                                req);
1335         l_wait_event(req->rq_wait_for_rep, ptlrpc_check_reply(req), &lwi);
1336 #else
1337         {
1338                 extern int reply_in_callback(ptl_event_t *ev);
1339                 ptl_event_t reply_ev;
1340                 PtlEQWait(req->rq_connection->c_peer.peer_ni->pni_reply_in_eq_h,
1341                           &reply_ev);
1342                 reply_in_callback(&reply_ev);
1343
1344                 LASSERT(reply_ev.mem_desc.user_ptr == (void *)req);
1345                 // ptlrpc_check_reply(req);
1346                 // not required now it only tests
1347         }
1348 #endif
1349
1350         DEBUG_REQ(D_NET, req, "-- done sleeping");
1351
1352         CDEBUG(D_RPCTRACE, "Completed RPC pname:cluuid:pid:xid:ni:nid:opc "
1353                "%s:%s:%d:"LPU64":%s:"LPX64":%d\n", current->comm,
1354                imp->imp_obd->obd_uuid.uuid,
1355                req->rq_reqmsg->status, req->rq_xid,
1356                imp->imp_connection->c_peer.peer_ni->pni_name,
1357                imp->imp_connection->c_peer.peer_nid,
1358                req->rq_reqmsg->opc);
1359
1360         spin_lock_irqsave(&imp->imp_lock, flags);
1361         list_del_init(&req->rq_list);
1362         spin_unlock_irqrestore(&imp->imp_lock, flags);
1363
1364         /* If the reply was received normally, this just grabs the spinlock
1365          * (ensuring the reply callback has returned), sees that
1366          * req->rq_receiving_reply is clear and returns. */
1367         ptlrpc_unregister_reply (req);
1368
1369         if (req->rq_err)
1370                 GOTO(out, rc = -EIO);
1371
1372         /* Resend if we need to, unless we were interrupted. */
1373         if (req->rq_resend && !req->rq_intr) {
1374                 /* ...unless we were specifically told otherwise. */
1375                 if (req->rq_no_resend) {
1376                         spin_lock_irqsave (&req->rq_lock, flags);
1377                         req->rq_no_resend = 0;
1378                         spin_unlock_irqrestore (&req->rq_lock, flags);
1379                         GOTO(out, rc = -ETIMEDOUT);
1380                 }
1381                 spin_lock_irqsave (&req->rq_lock, flags);
1382                 req->rq_resend = 0;
1383                 spin_unlock_irqrestore (&req->rq_lock, flags);
1384                 lustre_msg_add_flags(req->rq_reqmsg, MSG_RESENT);
1385
1386                 if (req->rq_bulk != NULL)
1387                         ptlrpc_unregister_bulk (req);
1388
1389                 DEBUG_REQ(D_HA, req, "resending: ");
1390                 spin_lock_irqsave(&imp->imp_lock, flags);
1391                 goto restart;
1392         }
1393
1394         if (req->rq_intr) {
1395                 /* Should only be interrupted if we timed out. */
1396                 if (!req->rq_timedout)
1397                         DEBUG_REQ(D_ERROR, req,
1398                                   "rq_intr set but rq_timedout not");
1399                 GOTO(out, rc = -EINTR);
1400         }
1401
1402         if (req->rq_timedout) {                 /* non-recoverable timeout */
1403                 GOTO(out, rc = -ETIMEDOUT);
1404         }
1405
1406         if (!req->rq_replied) {
1407                 /* How can this be? -eeb */
1408                 DEBUG_REQ(D_ERROR, req, "!rq_replied: ");
1409                 LBUG();
1410                 GOTO(out, rc = req->rq_status);
1411         }
1412
1413         rc = after_reply (req, &do_restart);
1414         /* NB may return +ve success rc */
1415         if (do_restart) {
1416                 if (req->rq_bulk != NULL)
1417                         ptlrpc_unregister_bulk (req);
1418                 DEBUG_REQ(D_HA, req, "resending: ");
1419                 spin_lock_irqsave(&imp->imp_lock, flags);
1420                 goto restart;
1421         }
1422
1423  out:
1424         if (req->rq_bulk != NULL) {
1425                 if (rc >= 0) {                  /* success so far */
1426                         lwi = LWI_TIMEOUT(timeout, NULL, NULL);
1427                         brc = l_wait_event(req->rq_wait_for_rep,
1428                                            ptlrpc_bulk_complete(req->rq_bulk),
1429                                            &lwi);
1430                         if (brc != 0) {
1431                                 LASSERT(brc == -ETIMEDOUT);
1432                                 CERROR ("Timed out waiting for bulk\n");
1433                                 rc = brc;
1434                         }
1435                 }
1436                 if (rc < 0) {
1437                         /* MDS blocks for put ACKs before replying */
1438                         /* OSC sets rq_no_resend for the time being */
1439                         LASSERT(req->rq_no_resend);
1440                         ptlrpc_unregister_bulk (req);
1441                 }
1442         }
1443
1444         LASSERT(!req->rq_receiving_reply);
1445         req->rq_phase = RQ_PHASE_INTERPRET;
1446         RETURN(rc);
1447 }
1448
1449 int ptlrpc_replay_req(struct ptlrpc_request *req)
1450 {
1451         int rc = 0, old_level, old_status = 0;
1452         // struct ptlrpc_client *cli = req->rq_import->imp_client;
1453         struct l_wait_info lwi;
1454         ENTRY;
1455
1456         /* I don't touch rq_phase here, so the debug log can show what
1457          * state it was left in */
1458
1459         /* Not handling automatic bulk replay yet (or ever?) */
1460         LASSERT(req->rq_bulk == NULL);
1461
1462         DEBUG_REQ(D_NET, req, "about to replay");
1463
1464         /* Update request's state, since we might have a new connection. */
1465         ptlrpc_put_connection(req->rq_connection);
1466         req->rq_connection =
1467                 ptlrpc_connection_addref(req->rq_import->imp_connection);
1468
1469         /* temporarily set request to RECOVD level (reset at out:) */
1470         old_level = req->rq_level;
1471         if (req->rq_replied)
1472                 old_status = req->rq_repmsg->status;
1473         req->rq_level = LUSTRE_CONN_RECOVER;
1474         rc = ptl_send_rpc(req);
1475         if (rc) {
1476                 CERROR("error %d, opcode %d\n", rc, req->rq_reqmsg->opc);
1477                 ptlrpc_cleanup_request_buf(req);
1478                 // up(&cli->cli_rpc_sem);
1479                 GOTO(out, rc = -rc);
1480         }
1481
1482         CDEBUG(D_OTHER, "-- sleeping\n");
1483         lwi = LWI_INTR(NULL, NULL); /* XXX needs timeout, nested recovery */
1484         l_wait_event(req->rq_wait_for_rep, ptlrpc_check_reply(req), &lwi);
1485         CDEBUG(D_OTHER, "-- done\n");
1486
1487         // up(&cli->cli_rpc_sem);
1488
1489         /* If the reply was received normally, this just grabs the spinlock
1490          * (ensuring the reply callback has returned), sees that
1491          * req->rq_receiving_reply is clear and returns. */
1492         ptlrpc_unregister_reply (req);
1493
1494         if (!req->rq_replied) {
1495                 CERROR("Unknown reason for wakeup\n");
1496                 /* XXX Phil - I end up here when I kill obdctl */
1497                 /* ...that's because signals aren't all masked in
1498                  * l_wait_event() -eeb */
1499                 GOTO(out, rc = -EINTR);
1500         }
1501
1502 #if SWAB_PARANOIA
1503         /* Clear reply swab mask; this is a new reply in sender's byte order */
1504         req->rq_rep_swab_mask = 0;
1505 #endif
1506         rc = lustre_unpack_msg(req->rq_repmsg, req->rq_replen);
1507         if (rc) {
1508                 CERROR("unpack_rep failed: %d\n", rc);
1509                 GOTO(out, rc = -EPROTO);
1510         }
1511 #if 0
1512         /* FIXME: Enable when BlueArc makes new release */
1513         if (req->rq_repmsg->type != PTL_RPC_MSG_REPLY &&
1514             req->rq_repmsg->type != PTL_RPC_MSG_ERR) {
1515                 CERROR("invalid packet type received (type=%u)\n",
1516                        req->rq_repmsg->type);
1517                 GOTO(out, rc = -EPROTO);
1518         }
1519 #endif
1520
1521         /* The transno had better not change over replay. */
1522         LASSERT(req->rq_reqmsg->transno == req->rq_repmsg->transno);
1523
1524         CDEBUG(D_NET, "got rep "LPD64"\n", req->rq_xid);
1525
1526         /* let the callback do fixups, possibly including in the request */
1527         if (req->rq_replay_cb)
1528                 req->rq_replay_cb(req);
1529
1530         if (req->rq_replied && req->rq_repmsg->status != old_status) {
1531                 DEBUG_REQ(D_HA, req, "status %d, old was %d",
1532                           req->rq_repmsg->status, old_status);
1533         }
1534
1535  out:
1536         req->rq_level = old_level;
1537         RETURN(rc);
1538 }
1539
1540 void ptlrpc_abort_inflight(struct obd_import *imp)
1541 {
1542         unsigned long flags;
1543         struct list_head *tmp, *n;
1544         ENTRY;
1545
1546         /* Make sure that no new requests get processed for this import.
1547          * ptlrpc_{queue,set}_wait must (and does) hold imp_lock while testing
1548          * this flag and then putting requests on sending_list or delayed_list.
1549          */
1550         spin_lock_irqsave(&imp->imp_lock, flags);
1551
1552         /* XXX locking?  Maybe we should remove each request with the list
1553          * locked?  Also, how do we know if the requests on the list are
1554          * being freed at this time?
1555          */
1556         list_for_each_safe(tmp, n, &imp->imp_sending_list) {
1557                 struct ptlrpc_request *req =
1558                         list_entry(tmp, struct ptlrpc_request, rq_list);
1559
1560                 DEBUG_REQ(D_HA, req, "inflight");
1561
1562                 spin_lock (&req->rq_lock);
1563                 if (req->rq_import_generation < imp->imp_generation) {
1564                         req->rq_err = 1;
1565                         if (req->rq_set != NULL)
1566                                 wake_up(&req->rq_set->set_waitq);
1567                         else
1568                                 wake_up(&req->rq_wait_for_rep);
1569                 }
1570                 spin_unlock (&req->rq_lock);
1571         }
1572
1573         list_for_each_safe(tmp, n, &imp->imp_delayed_list) {
1574                 struct ptlrpc_request *req =
1575                         list_entry(tmp, struct ptlrpc_request, rq_list);
1576
1577                 DEBUG_REQ(D_HA, req, "aborting waiting req");
1578
1579                 spin_lock (&req->rq_lock);
1580                 if (req->rq_import_generation < imp->imp_generation) {
1581                         req->rq_err = 1;
1582                         if (req->rq_set != NULL)
1583                                 wake_up(&req->rq_set->set_waitq);
1584                         else
1585                                 wake_up(&req->rq_wait_for_rep);
1586                 }
1587                 spin_unlock (&req->rq_lock);
1588         }
1589
1590         /* Last chance to free reqs left on the replay list, but we
1591          * will still leak reqs that haven't comitted.  */
1592         if (imp->imp_replayable)
1593                 ptlrpc_free_committed(imp);
1594
1595         spin_unlock_irqrestore(&imp->imp_lock, flags);
1596
1597         EXIT;
1598 }
1599
1600 static __u64 ptlrpc_last_xid = 0;
1601 static spinlock_t ptlrpc_last_xid_lock = SPIN_LOCK_UNLOCKED;
1602
1603 __u64 ptlrpc_next_xid(void)
1604 {
1605         __u64 tmp;
1606         spin_lock(&ptlrpc_last_xid_lock);
1607         tmp = ++ptlrpc_last_xid;
1608         spin_unlock(&ptlrpc_last_xid_lock);
1609         return tmp;
1610 }
1611
1612