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, req->rq_status);
 316                         }
 317                         set->set_remaining--;
 318                 }
 319
 320                 req->rq_set = NULL;
 321                 ptlrpc_req_finished (req);
 322         }
 323
 324         LASSERT(set->set_remaining == 0);
 325
 326         OBD_FREE(set, sizeof(*set));
 327         EXIT;
 328 }
 329
 330 void ptlrpc_set_add_req(struct ptlrpc_request_set *set,
 331                         struct ptlrpc_request *req)
 332 {
 333         /* The set takes over the caller's request reference */
 334         list_add_tail(&req->rq_set_chain, &set->set_requests);
 335         req->rq_set = set;
 336         set->set_remaining++;
 337 }
 338
 339 static int ptlrpc_check_reply(struct ptlrpc_request *req)
 340 {
 341         unsigned long flags;
 342         int rc = 0;
 343         ENTRY;
 344
 345         /* serialise with network callback */
 346         spin_lock_irqsave (&req->rq_lock, flags);
 347
 348         if (req->rq_replied) {
 349                 DEBUG_REQ(D_NET, req, "REPLIED:");
 350                 GOTO(out, rc = 1);
 351         }
 352
 353         if (req->rq_err) {
 354                 DEBUG_REQ(D_ERROR, req, "ABORTED:");
 355                 GOTO(out, rc = 1);
 356         }
 357
 358         if (req->rq_resend) {
 359                 DEBUG_REQ(D_ERROR, req, "RESEND:");
 360                 GOTO(out, rc = 1);
 361         }
 362
 363         if (req->rq_restart) {
 364                 DEBUG_REQ(D_ERROR, req, "RESTART:");
 365                 GOTO(out, rc = 1);
 366         }
 367         EXIT;
 368  out:
 369         spin_unlock_irqrestore (&req->rq_lock, flags);
 370         DEBUG_REQ(D_NET, req, "rc = %d for", rc);
 371         return rc;
 372 }
 373
 374 static int ptlrpc_check_status(struct ptlrpc_request *req)
 375 {
 376         int err;
 377         ENTRY;
 378
 379         err = req->rq_repmsg->status;
 380         if (req->rq_repmsg->type == PTL_RPC_MSG_ERR) {
 381                 DEBUG_REQ(D_ERROR, req, "type == PTL_RPC_MSG_ERR (%d)", err);
 382                 if (err >= 0)
 383                         CERROR("Error Reply has >= zero status\n");
 384                 RETURN(err < 0 ? err : -EINVAL);
 385         }
 386
 387         if (err < 0) {
 388                 DEBUG_REQ(D_INFO, req, "status is %d", err);
 389         } else if (err > 0) {
 390                 /* XXX: translate this error from net to host */
 391                 DEBUG_REQ(D_INFO, req, "status is %d", err);
 392         }
 393
 394         RETURN(err);
 395 }
 396
 397 #warning this needs to change after robert fixes eviction handling
 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_eviction(req);
 446
 447                 if (req->rq_err)
 448                         RETURN(-EIO);
 449
 450                 if (req->rq_resend) {
 451                         if (restartp == NULL)
 452                                 LBUG(); /* async resend not supported yet */
 453                         spin_lock_irqsave (&req->rq_lock, flags);
 454                         req->rq_resend = 0;
 455                         spin_unlock_irqrestore (&req->rq_lock, flags);
 456                         *restartp = 1;
 457                         lustre_msg_add_flags(req->rq_reqmsg, MSG_RESENT);
 458                         DEBUG_REQ(D_HA, req, "resending: ");
 459                         RETURN (0);
 460                 }
 461
 462                 CERROR("request should be err or resend: %p\n", req);
 463                 LBUG();
 464         }
 465
 466         if (req->rq_import->imp_replayable) {
 467                 spin_lock_irqsave(&imp->imp_lock, flags);
 468                 if ((req->rq_replay || req->rq_transno != 0) && rc >= 0)
 469                         ptlrpc_retain_replayable_request(req, imp);
 470
 471                 if (req->rq_transno > imp->imp_max_transno)
 472                         imp->imp_max_transno = req->rq_transno;
 473
 474                 /* Replay-enabled imports return commit-status information. */
 475                 if (req->rq_repmsg->last_committed) {
 476                         imp->imp_peer_committed_transno =
 477                                 req->rq_repmsg->last_committed;
 478                 }
 479                 ptlrpc_free_committed(imp);
 480                 spin_unlock_irqrestore(&imp->imp_lock, flags);
 481         }
 482
 483         RETURN(rc);
 484 }
 485
 486 int ptlrpc_check_set(struct ptlrpc_request_set *set)
 487 {
 488         unsigned long flags;
 489         struct list_head *tmp;
 490         ENTRY;
 491
 492         if (set->set_remaining == 0)
 493                 RETURN(1);
 494
 495         list_for_each(tmp, &set->set_requests) {
 496                 struct ptlrpc_request *req =
 497                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 498                 struct obd_import *imp = req->rq_import;
 499                 int rc = 0;
 500
 501                 if (!(req->rq_phase == RQ_PHASE_RPC ||
 502                       req->rq_phase == RQ_PHASE_BULK ||
 503                       req->rq_phase == RQ_PHASE_INTERPRET ||
 504                       req->rq_phase == RQ_PHASE_COMPLETE)) {
 505                         DEBUG_REQ(D_ERROR, req, "bad phase %x", req->rq_phase);
 506                         LBUG();
 507                 }
 508
 509                 if (req->rq_phase == RQ_PHASE_COMPLETE)
 510                         continue;
 511
 512                 if (req->rq_phase == RQ_PHASE_INTERPRET)
 513                         GOTO (interpret, req->rq_status);
 514
 515                 if (req->rq_err) {
 516                         ptlrpc_unregister_reply(req);
 517                         if (req->rq_status == 0)
 518                                 req->rq_status = -EIO;
 519                         req->rq_phase = RQ_PHASE_INTERPRET;
 520
 521                         spin_lock_irqsave(&imp->imp_lock, flags);
 522                         list_del_init(&req->rq_list);
 523                         spin_unlock_irqrestore(&imp->imp_lock, flags);
 524
 525                         GOTO (interpret, req->rq_status);
 526                 }
 527
 528                 if (req->rq_intr) {
 529                         /* NB could be on delayed list */
 530                         ptlrpc_unregister_reply(req);
 531                         req->rq_status = -EINTR;
 532                         req->rq_phase = RQ_PHASE_INTERPRET;
 533
 534                         spin_lock_irqsave(&imp->imp_lock, flags);
 535                         list_del_init(&req->rq_list);
 536                         spin_unlock_irqrestore(&imp->imp_lock, flags);
 537
 538                         GOTO (interpret, req->rq_status);
 539                 }
 540
 541                 if (req->rq_phase == RQ_PHASE_RPC) {
 542                         int do_restart = 0;
 543                         if (req->rq_waiting || req->rq_resend) {
 544                                 spin_lock_irqsave(&imp->imp_lock, flags);
 545
 546                                 if (req->rq_level > imp->imp_level) {
 547                                         spin_unlock_irqrestore(&imp->imp_lock,
 548                                                                flags);
 549                                         continue;
 550                                 }
 551
 552                                 list_del(&req->rq_list);
 553                                 list_add_tail(&req->rq_list,
 554                                               &imp->imp_sending_list);
 555
 556                                 if (req->rq_import_generation <
 557                                     imp->imp_generation) {
 558                                         req->rq_status = -EIO;
 559                                         req->rq_phase = RQ_PHASE_INTERPRET;
 560                                         spin_unlock_irqrestore(&imp->imp_lock,
 561                                                                flags);
 562                                         GOTO (interpret, req->rq_status);
 563                                 }
 564                                 spin_unlock_irqrestore(&imp->imp_lock, flags);
 565
 566                                 req->rq_waiting = 0;
 567                                 if (req->rq_resend) {
 568                                         lustre_msg_add_flags(req->rq_reqmsg,
 569                                                              MSG_RESENT);
 570                                         spin_lock_irqsave(&req->rq_lock, flags);
 571                                         req->rq_resend = 0;
 572                                         spin_unlock_irqrestore(&req->rq_lock,
 573                                                                flags);
 574                                         ptlrpc_unregister_reply(req);
 575                                         if (req->rq_bulk)
 576                                                 ptlrpc_unregister_bulk(req);
 577                                }
 578
 579                                 rc = ptl_send_rpc(req);
 580                                 if (rc) {
 581                                         req->rq_status = rc;
 582                                         req->rq_phase = RQ_PHASE_INTERPRET;
 583                                         GOTO (interpret, req->rq_status);
 584                                 }
 585
 586                         }
 587
 588                         /* Ensure the network callback returned */
 589                         spin_lock_irqsave (&req->rq_lock, flags);
 590                         if (!req->rq_replied) {
 591                                 spin_unlock_irqrestore (&req->rq_lock, flags);
 592                                 continue;
 593                         }
 594                         spin_unlock_irqrestore (&req->rq_lock, flags);
 595
 596                         spin_lock_irqsave(&imp->imp_lock, flags);
 597                         list_del_init(&req->rq_list);
 598                         spin_unlock_irqrestore(&imp->imp_lock, flags);
 599
 600                         req->rq_status = after_reply(req, &do_restart);
 601                         if (do_restart) {
 602                                 spin_lock_irqsave (&req->rq_lock, flags);
 603                                 req->rq_resend = 1; /* ugh */
 604                                 spin_unlock_irqrestore (&req->rq_lock, flags);
 605                                 continue;
 606                         }
 607
 608                         /* If there is no bulk associated with this request,
 609                          * then we're done and should let the interpreter
 610                          * process the reply.  Similarly if the RPC returned
 611                          * an error, and therefore the bulk will never arrive.
 612                          */
 613                         if (req->rq_bulk == NULL || req->rq_status != 0) {
 614                                 req->rq_phase = RQ_PHASE_INTERPRET;
 615                                 GOTO (interpret, req->rq_status);
 616                         }
 617
 618                         req->rq_phase = RQ_PHASE_BULK;
 619                 }
 620
 621                 LASSERT (req->rq_phase == RQ_PHASE_BULK);
 622                 if (!ptlrpc_bulk_complete (req->rq_bulk))
 623                         continue;
 624
 625                 req->rq_phase = RQ_PHASE_INTERPRET;
 626
 627         interpret:
 628                 LASSERT (req->rq_phase == RQ_PHASE_INTERPRET);
 629                 LASSERT (!req->rq_receiving_reply);
 630
 631                 ptlrpc_unregister_reply(req);
 632                 if (req->rq_bulk != NULL)
 633                         ptlrpc_unregister_bulk (req);
 634
 635                 if (req->rq_interpret_reply != NULL) {
 636                         int (*interpreter)(struct ptlrpc_request *,void *,int) =
 637                                 req->rq_interpret_reply;
 638                         req->rq_status = interpreter(req, &req->rq_async_args,
 639                                                      req->rq_status);
 640                 }
 641
 642                 CDEBUG(D_RPCTRACE, "Completed RPC pname:cluuid:pid:xid:ni:nid:"
 643                        "opc %s:%s:%d:"LPU64":%s:"LPX64":%d\n", current->comm,
 644                        imp->imp_obd->obd_uuid.uuid, req->rq_reqmsg->status,
 645                        req->rq_xid,
 646                        imp->imp_connection->c_peer.peer_ni->pni_name,
 647                        imp->imp_connection->c_peer.peer_nid,
 648                        req->rq_reqmsg->opc);
 649
 650                 req->rq_phase = RQ_PHASE_COMPLETE;
 651                 set->set_remaining--;
 652         }
 653
 654         RETURN (set->set_remaining == 0);
 655 }
 656
 657 int ptlrpc_expire_one_request(struct ptlrpc_request *req)
 658 {
 659         unsigned long      flags;
 660         struct obd_import *imp = req->rq_import;
 661         ENTRY;
 662
 663         DEBUG_REQ(D_ERROR, req, "timeout");
 664
 665         spin_lock_irqsave (&req->rq_lock, flags);
 666         req->rq_timedout = 1;
 667         spin_unlock_irqrestore (&req->rq_lock, flags);
 668
 669         ptlrpc_unregister_reply (req);
 670
 671         if (imp == NULL) {
 672                 DEBUG_REQ(D_HA, req, "NULL import: already cleaned up?");
 673                 RETURN(1);
 674         }
 675
 676         /* The DLM server doesn't want recovery run on its imports. */
 677         if (imp->imp_dlm_fake)
 678                 RETURN(1);
 679
 680         /* If this request is for recovery or other primordial tasks,
 681          * don't go back to sleep, and don't start recovery again.. */
 682         if (req->rq_level < LUSTRE_CONN_FULL || req->rq_no_recov ||
 683             imp->imp_obd->obd_no_recov)
 684                 RETURN(1);
 685
 686         ptlrpc_fail_import(imp, req->rq_import_generation);
 687
 688         RETURN(0);
 689 }
 690
 691 static int expired_set(void *data)
 692 {
 693         struct ptlrpc_request_set *set = data;
 694         struct list_head          *tmp;
 695         time_t                     now = LTIME_S (CURRENT_TIME);
 696         ENTRY;
 697
 698         LASSERT (set != NULL);
 699
 700         /* A timeout expired; see which reqs it applies to... */
 701         list_for_each (tmp, &set->set_requests) {
 702                 struct ptlrpc_request *req =
 703                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 704
 705                 /* request in-flight? */
 706                 if (!((req->rq_phase == RQ_PHASE_RPC && !req->rq_waiting) ||
 707                       (req->rq_phase == RQ_PHASE_BULK)))
 708                         continue;
 709
 710                 if (req->rq_timedout ||           /* already dealt with */
 711                     req->rq_sent + req->rq_timeout > now) /* not expired */
 712                         continue;
 713
 714                 /* deal with this guy */
 715                 ptlrpc_expire_one_request (req);
 716         }
 717
 718         /* When waiting for a whole set, we always to break out of the
 719          * sleep so we can recalculate the timeout, or enable interrupts
 720          * iff everyone's timed out.
 721          */
 722         RETURN(1);
 723 }
 724
 725 static void interrupted_set(void *data)
 726 {
 727         struct ptlrpc_request_set *set = data;
 728         struct list_head *tmp;
 729         unsigned long flags;
 730
 731         LASSERT (set != NULL);
 732         CERROR("INTERRUPTED SET %p\n", set);
 733
 734         list_for_each(tmp, &set->set_requests) {
 735                 struct ptlrpc_request *req =
 736                         list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 737
 738                 if (req->rq_phase != RQ_PHASE_RPC)
 739                         continue;
 740
 741                 spin_lock_irqsave (&req->rq_lock, flags);
 742                 req->rq_intr = 1;
 743                 spin_unlock_irqrestore (&req->rq_lock, flags);
 744         }
 745 }
 746
 747 int ptlrpc_set_wait(struct ptlrpc_request_set *set)
 748 {
 749         struct list_head      *tmp;
 750         struct obd_import     *imp;
 751         struct ptlrpc_request *req;
 752         struct l_wait_info     lwi;
 753         unsigned long          flags;
 754         int                    rc;
 755         time_t                 now;
 756         time_t                 deadline;
 757         int                    timeout;
 758         ENTRY;
 759
 760         LASSERT(!list_empty(&set->set_requests));
 761         list_for_each(tmp, &set->set_requests) {
 762                 req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 763
 764                 LASSERT (req->rq_level == LUSTRE_CONN_FULL);
 765                 LASSERT (req->rq_phase == RQ_PHASE_NEW);
 766                 req->rq_phase = RQ_PHASE_RPC;
 767
 768                 imp = req->rq_import;
 769                 spin_lock_irqsave(&imp->imp_lock, flags);
 770
 771                 if (imp->imp_invalid) {
 772                         spin_unlock_irqrestore(&imp->imp_lock, flags);
 773                         req->rq_status = -EIO;
 774                         req->rq_phase = RQ_PHASE_INTERPRET;
 775                         continue;
 776                 }
 777
 778                 req->rq_import_generation = imp->imp_generation;
 779
 780                 if (req->rq_level > imp->imp_level) {
 781                         if (req->rq_no_recov || imp->imp_obd->obd_no_recov ||
 782                             imp->imp_dlm_fake) {
 783                                 spin_unlock_irqrestore(&imp->imp_lock, flags);
 784                                 req->rq_status = -EWOULDBLOCK;
 785                                 req->rq_phase = RQ_PHASE_INTERPRET;
 786                                 continue;
 787                         }
 788
 789                         spin_lock (&req->rq_lock);
 790                         req->rq_waiting = 1;
 791                         spin_unlock (&req->rq_lock);
 792                         LASSERT (list_empty (&req->rq_list));
 793                         // list_del(&req->rq_list);
 794                         list_add_tail(&req->rq_list, &imp->imp_delayed_list);
 795                         spin_unlock_irqrestore(&imp->imp_lock, flags);
 796                         continue;
 797                 }
 798
 799                 /* XXX this is the same as ptlrpc_queue_wait */
 800                 LASSERT(list_empty(&req->rq_list));
 801                 list_add_tail(&req->rq_list, &imp->imp_sending_list);
 802                 spin_unlock_irqrestore(&imp->imp_lock, flags);
 803
 804                 CDEBUG(D_RPCTRACE, "Sending RPC pname:cluuid:pid:xid:ni:nid:opc"
 805                        " %s:%s:%d:"LPU64":%s:"LPX64":%d\n", current->comm,
 806                        imp->imp_obd->obd_uuid.uuid, req->rq_reqmsg->status,
 807                        req->rq_xid,
 808                        imp->imp_connection->c_peer.peer_ni->pni_name,
 809                        imp->imp_connection->c_peer.peer_nid,
 810                        req->rq_reqmsg->opc);
 811
 812                 rc = ptl_send_rpc(req);
 813                 if (rc) {
 814                         req->rq_status = rc;
 815                         req->rq_phase = RQ_PHASE_INTERPRET;
 816                 }
 817         }
 818
 819         do {
 820                 now = LTIME_S (CURRENT_TIME);
 821                 timeout = 0;
 822                 list_for_each (tmp, &set->set_requests) {
 823                         req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 824
 825                         /* request in-flight? */
 826                         if (!((req->rq_phase == RQ_PHASE_RPC &&
 827                                !req->rq_waiting) ||
 828                               (req->rq_phase == RQ_PHASE_BULK)))
 829                                 continue;
 830
 831                         if (req->rq_timedout)   /* already timed out */
 832                                 continue;
 833
 834                         deadline = req->rq_sent + req->rq_timeout;
 835                         if (deadline <= now)    /* actually expired already */
 836                                 timeout = 1;    /* ASAP */
 837                         else if (timeout == 0 || timeout > deadline - now)
 838                                 timeout = deadline - now;
 839                 }
 840
 841                 /* wait until all complete, interrupted, or an in-flight
 842                  * req times out */
 843                 CDEBUG(D_HA, "set %p going to sleep for %d seconds\n",
 844                        set, timeout);
 845                 lwi = LWI_TIMEOUT_INTR((timeout ? timeout : 1) * HZ,
 846                                        expired_set, interrupted_set, set);
 847                 rc = l_wait_event(set->set_waitq, ptlrpc_check_set(set), &lwi);
 848
 849                 LASSERT (rc == 0 || rc == -EINTR || rc == -ETIMEDOUT);
 850
 851                 /* -EINTR => all requests have been flagged rq_intr so next
 852                  * check completes.
 853                  * -ETIMEOUTD => someone timed out.  When all reqs have
 854                  * timed out, signals are enabled allowing completion with
 855                  * EINTR.
 856                  * I don't really care if we go once more round the loop in
 857                  * the error cases -eeb. */
 858         } while (rc != 0);
 859
 860         LASSERT (set->set_remaining == 0);
 861
 862         rc = 0;
 863         list_for_each(tmp, &set->set_requests) {
 864                 req = list_entry(tmp, struct ptlrpc_request, rq_set_chain);
 865
 866                 LASSERT (req->rq_phase == RQ_PHASE_COMPLETE);
 867                 if (req->rq_status != 0)
 868                         rc = req->rq_status;
 869         }
 870
 871         if (set->set_interpret != NULL) {
 872                 int (*interpreter)(struct ptlrpc_request_set *set,void *,int) =
 873                         set->set_interpret;
 874                 rc = interpreter (set, &set->set_args, rc);
 875         }
 876
 877         RETURN(rc);
 878 }
 879
 880 static void __ptlrpc_free_req(struct ptlrpc_request *request, int locked)
 881 {
 882         ENTRY;
 883         if (request == NULL) {
 884                 EXIT;
 885                 return;
 886         }
 887
 888         LASSERT (!request->rq_receiving_reply);
 889
 890         /* We must take it off the imp_replay_list first.  Otherwise, we'll set
 891          * request->rq_reqmsg to NULL while osc_close is dereferencing it. */
 892         if (request->rq_import != NULL) {
 893                 unsigned long flags = 0;
 894                 if (!locked)
 895                         spin_lock_irqsave(&request->rq_import->imp_lock, flags);
 896                 list_del_init(&request->rq_list);
 897                 if (!locked)
 898                         spin_unlock_irqrestore(&request->rq_import->imp_lock,
 899                                                flags);
 900         }
 901
 902         if (atomic_read(&request->rq_refcount) != 0) {
 903                 DEBUG_REQ(D_ERROR, request,
 904                           "freeing request with nonzero refcount");
 905                 LBUG();
 906         }
 907
 908         if (request->rq_repmsg != NULL) {
 909                 OBD_FREE(request->rq_repmsg, request->rq_replen);
 910                 request->rq_repmsg = NULL;
 911         }
 912         if (request->rq_reqmsg != NULL) {
 913                 OBD_FREE(request->rq_reqmsg, request->rq_reqlen);
 914                 request->rq_reqmsg = NULL;
 915         }
 916         if (request->rq_export != NULL) {
 917                 class_export_put(request->rq_export);
 918                 request->rq_export = NULL;
 919         }
 920         if (request->rq_import != NULL) {
 921                 class_import_put(request->rq_import);
 922                 request->rq_import = NULL;
 923         }
 924         if (request->rq_bulk != NULL)
 925                 ptlrpc_free_bulk(request->rq_bulk);
 926
 927         ptlrpc_put_connection(request->rq_connection);
 928         OBD_FREE(request, sizeof(*request));
 929         EXIT;
 930 }
 931
 932 void ptlrpc_free_req(struct ptlrpc_request *request)
 933 {
 934         __ptlrpc_free_req(request, 0);
 935 }
 936
 937 static int __ptlrpc_req_finished(struct ptlrpc_request *request, int locked)
 938 {
 939         ENTRY;
 940         if (request == NULL)
 941                 RETURN(1);
 942
 943         if (request == (void *)(long)(0x5a5a5a5a5a5a5a5a) ||
 944             request->rq_obd == (void *)(long)(0x5a5a5a5a5a5a5a5a)) {
 945                 CERROR("dereferencing freed request (bug 575)\n");
 946                 LBUG();
 947                 RETURN(1);
 948         }
 949
 950         DEBUG_REQ(D_INFO, request, "refcount now %u",
 951                   atomic_read(&request->rq_refcount) - 1);
 952
 953         if (atomic_dec_and_test(&request->rq_refcount)) {
 954                 __ptlrpc_free_req(request, locked);
 955                 RETURN(1);
 956         }
 957
 958         RETURN(0);
 959 }
 960
 961 void ptlrpc_req_finished(struct ptlrpc_request *request)
 962 {
 963         __ptlrpc_req_finished(request, 0);
 964 }
 965
 966 static void ptlrpc_cleanup_request_buf(struct ptlrpc_request *request)
 967 {
 968         OBD_FREE(request->rq_reqmsg, request->rq_reqlen);
 969         request->rq_reqmsg = NULL;
 970         request->rq_reqlen = 0;
 971 }
 972
 973 /* Disengage the client's reply buffer from the network
 974  * NB does _NOT_ unregister any client-side bulk.
 975  * IDEMPOTENT, but _not_ safe against concurrent callers.
 976  * The request owner (i.e. the thread doing the I/O) must call...
 977  */
 978 void ptlrpc_unregister_reply (struct ptlrpc_request *request)
 979 {
 980         unsigned long flags;
 981         int           rc;
 982         ENTRY;
 983
 984         LASSERT (!in_interrupt ());             /* might sleep */
 985
 986         spin_lock_irqsave (&request->rq_lock, flags);
 987         if (!request->rq_receiving_reply) {     /* not waiting for a reply */
 988                 spin_unlock_irqrestore (&request->rq_lock, flags);
 989                 EXIT;
 990                 /* NB reply buffer not freed here */
 991                 return;
 992         }
 993
 994         LASSERT (!request->rq_replied);         /* callback hasn't completed */
 995         spin_unlock_irqrestore (&request->rq_lock, flags);
 996
 997         rc = PtlMDUnlink (request->rq_reply_md_h);
 998         switch (rc) {
 999         default:
1000                 LBUG ();
1001
1002         case PTL_OK:                            /* unlinked before completion */
1003                 LASSERT (request->rq_receiving_reply);
1004                 LASSERT (!request->rq_replied);
1005                 spin_lock_irqsave (&request->rq_lock, flags);
1006                 request->rq_receiving_reply = 0;
1007                 spin_unlock_irqrestore (&request->rq_lock, flags);
1008                 OBD_FREE(request->rq_repmsg, request->rq_replen);
1009                 request->rq_repmsg = NULL;
1010                 EXIT;
1011                 return;
1012
1013         case PTL_MD_INUSE:                      /* callback in progress */
1014                 for (;;) {
1015                         /* Network access will complete in finite time but
1016                          * the timeout lets us CERROR for visibility */
1017                         struct l_wait_info lwi = LWI_TIMEOUT(10*HZ, NULL, NULL);
1018
1019                         rc = l_wait_event (request->rq_wait_for_rep,
1020                                            request->rq_replied, &lwi);
1021                         LASSERT (rc == 0 || rc == -ETIMEDOUT);
1022                         if (rc == 0) {
1023                                 spin_lock_irqsave (&request->rq_lock, flags);
1024                                 /* Ensure the callback has completed scheduling
1025                                  * me and taken its hands off the request */
1026                                 spin_unlock_irqrestore(&request->rq_lock,flags);
1027                                 break;
1028                         }
1029
1030                         CERROR ("Unexpectedly long timeout: req %p\n", request);
1031                 }
1032                 /* fall through */
1033
1034         case PTL_INV_MD:                        /* callback completed */
1035                 LASSERT (!request->rq_receiving_reply);
1036                 LASSERT (request->rq_replied);
1037                 EXIT;
1038                 return;
1039         }
1040         /* Not Reached */
1041 }
1042
1043 /* caller must hold imp->imp_lock */
1044 void ptlrpc_free_committed(struct obd_import *imp)
1045 {
1046         struct list_head *tmp, *saved;
1047         struct ptlrpc_request *req;
1048         struct ptlrpc_request *last_req = NULL; /* temporary fire escape */
1049         ENTRY;
1050
1051         LASSERT(imp != NULL);
1052
1053 #ifdef CONFIG_SMP
1054         LASSERT(spin_is_locked(&imp->imp_lock));
1055 #endif
1056
1057         CDEBUG(D_HA, "%s: committing for last_committed "LPU64"\n",
1058                imp->imp_obd->obd_name, imp->imp_peer_committed_transno);
1059
1060         list_for_each_safe(tmp, saved, &imp->imp_replay_list) {
1061                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
1062
1063                 /* XXX ok to remove when 1357 resolved - rread 05/29/03  */
1064                 LASSERT (req != last_req);
1065                 last_req = req;
1066
1067                 if (req->rq_import_generation < imp->imp_generation) {
1068                         DEBUG_REQ(D_HA, req, "freeing request with old gen");
1069                         GOTO(free_req, 0);
1070                 }
1071
1072                 if (req->rq_replay) {
1073                         DEBUG_REQ(D_HA, req, "keeping (FL_REPLAY)");
1074                         continue;
1075                 }
1076
1077                 /* not yet committed */
1078                 if (req->rq_transno > imp->imp_peer_committed_transno) {
1079                         DEBUG_REQ(D_HA, req, "stopping search");
1080                         break;
1081                 }
1082
1083                 DEBUG_REQ(D_HA, req, "committing (last_committed "LPU64")",
1084                           imp->imp_peer_committed_transno);
1085 free_req:
1086                 list_del_init(&req->rq_list);
1087                 __ptlrpc_req_finished(req, 1);
1088         }
1089
1090         EXIT;
1091         return;
1092 }
1093
1094 void ptlrpc_cleanup_client(struct obd_import *imp)
1095 {
1096         ENTRY;
1097         EXIT;
1098         return;
1099 }
1100
1101 void ptlrpc_resend_req(struct ptlrpc_request *req)
1102 {
1103         unsigned long flags;
1104
1105         DEBUG_REQ(D_HA, req, "resending");
1106         req->rq_reqmsg->handle.cookie = 0;
1107         ptlrpc_put_connection(req->rq_connection);
1108         req->rq_connection =
1109                 ptlrpc_connection_addref(req->rq_import->imp_connection);
1110         req->rq_status = -EAGAIN;
1111
1112         spin_lock_irqsave (&req->rq_lock, flags);
1113         req->rq_resend = 1;
1114         req->rq_timedout = 0;
1115         if (req->rq_set != NULL)
1116                 wake_up (&req->rq_set->set_waitq);
1117         else
1118                 wake_up(&req->rq_wait_for_rep);
1119         spin_unlock_irqrestore (&req->rq_lock, flags);
1120 }
1121
1122 /* XXX: this function and rq_status are currently unused */
1123 void ptlrpc_restart_req(struct ptlrpc_request *req)
1124 {
1125         unsigned long flags;
1126
1127         DEBUG_REQ(D_HA, req, "restarting (possibly-)completed request");
1128         req->rq_status = -ERESTARTSYS;
1129
1130         spin_lock_irqsave (&req->rq_lock, flags);
1131         req->rq_restart = 1;
1132         req->rq_timedout = 0;
1133         if (req->rq_set != NULL)
1134                 wake_up (&req->rq_set->set_waitq);
1135         else
1136                 wake_up(&req->rq_wait_for_rep);
1137         spin_unlock_irqrestore (&req->rq_lock, flags);
1138 }
1139
1140 static int expired_request(void *data)
1141 {
1142         struct ptlrpc_request *req = data;
1143         ENTRY;
1144
1145         RETURN(ptlrpc_expire_one_request(req));
1146 }
1147
1148 static void interrupted_request(void *data)
1149 {
1150         unsigned long flags;
1151
1152         struct ptlrpc_request *req = data;
1153         DEBUG_REQ(D_HA, req, "request interrupted");
1154         spin_lock_irqsave (&req->rq_lock, flags);
1155         req->rq_intr = 1;
1156         spin_unlock_irqrestore (&req->rq_lock, flags);
1157 }
1158
1159 struct ptlrpc_request *ptlrpc_request_addref(struct ptlrpc_request *req)
1160 {
1161         ENTRY;
1162         atomic_inc(&req->rq_refcount);
1163         RETURN(req);
1164 }
1165
1166 void ptlrpc_retain_replayable_request(struct ptlrpc_request *req,
1167                                       struct obd_import *imp)
1168 {
1169         struct list_head *tmp;
1170
1171 #ifdef CONFIG_SMP
1172         LASSERT(spin_is_locked(&imp->imp_lock));
1173 #endif
1174
1175         LASSERT(imp->imp_replayable);
1176         /* Balanced in ptlrpc_free_committed, usually. */
1177         ptlrpc_request_addref(req);
1178         list_for_each_prev(tmp, &imp->imp_replay_list) {
1179                 struct ptlrpc_request *iter =
1180                         list_entry(tmp, struct ptlrpc_request, rq_list);
1181
1182                 /* We may have duplicate transnos if we create and then
1183                  * open a file, or for closes retained if to match creating
1184                  * opens, so use req->rq_xid as a secondary key.
1185                  * (See bugs 684, 685, and 428.)
1186                  * XXX no longer needed, but all opens need transnos!
1187                  */
1188                 if (iter->rq_transno > req->rq_transno)
1189                         continue;
1190
1191                 if (iter->rq_transno == req->rq_transno) {
1192                         LASSERT(iter->rq_xid != req->rq_xid);
1193                         if (iter->rq_xid > req->rq_xid)
1194                                 continue;
1195                 }
1196
1197                 list_add(&req->rq_list, &iter->rq_list);
1198                 return;
1199         }
1200
1201         list_add_tail(&req->rq_list, &imp->imp_replay_list);
1202 }
1203
1204 int ptlrpc_queue_wait(struct ptlrpc_request *req)
1205 {
1206         int rc = 0;
1207         int brc;
1208         struct l_wait_info lwi;
1209         struct obd_import *imp = req->rq_import;
1210         struct obd_device *obd = imp->imp_obd;
1211         struct ptlrpc_connection *conn = imp->imp_connection;
1212         unsigned int flags;
1213         int do_restart = 0;
1214         int timeout = 0;
1215         ENTRY;
1216
1217         LASSERT (req->rq_set == NULL);
1218         LASSERT (!req->rq_receiving_reply);
1219
1220         /* for distributed debugging */
1221         req->rq_reqmsg->status = current->pid;
1222         LASSERT(imp->imp_obd != NULL);
1223         CDEBUG(D_RPCTRACE, "Sending RPC pname:cluuid:pid:xid:ni:nid:opc "
1224                "%s:%s:%d:"LPU64":%s:"LPX64":%d\n", current->comm,
1225                imp->imp_obd->obd_uuid.uuid,
1226                req->rq_reqmsg->status, req->rq_xid,
1227                conn->c_peer.peer_ni->pni_name, conn->c_peer.peer_nid,
1228                req->rq_reqmsg->opc);
1229
1230         /* Mark phase here for a little debug help */
1231         req->rq_phase = RQ_PHASE_RPC;
1232
1233         spin_lock_irqsave(&imp->imp_lock, flags);
1234         req->rq_import_generation = imp->imp_generation;
1235 restart:
1236         /*
1237          * If the import has been invalidated (such as by an OST failure), the
1238          * request must fail with -EIO.  Recovery requests are allowed to go
1239          * through, though, so that they have a chance to revalidate the
1240          * import.
1241          */
1242         if (req->rq_import->imp_invalid && req->rq_level == LUSTRE_CONN_FULL) {
1243                 DEBUG_REQ(D_ERROR, req, "IMP_INVALID:");
1244                 spin_unlock_irqrestore(&imp->imp_lock, flags);
1245                 GOTO (out, rc = -EIO);
1246         }
1247
1248         if (req->rq_import_generation < imp->imp_generation) {
1249                 DEBUG_REQ(D_ERROR, req, "req old gen:");
1250                 spin_unlock_irqrestore(&imp->imp_lock, flags);
1251                 GOTO (out, rc = -EIO);
1252         }
1253
1254         if (req->rq_level > imp->imp_level) {
1255                 list_del(&req->rq_list);
1256                 if (req->rq_no_recov || obd->obd_no_recov ||
1257                     imp->imp_dlm_fake) {
1258                         spin_unlock_irqrestore(&imp->imp_lock, flags);
1259                         GOTO (out, rc = -EWOULDBLOCK);
1260                 }
1261
1262                 list_add_tail(&req->rq_list, &imp->imp_delayed_list);
1263                 spin_unlock_irqrestore(&imp->imp_lock, flags);
1264
1265                 DEBUG_REQ(D_HA, req, "\"%s\" waiting for recovery: (%d > %d)",
1266                           current->comm, req->rq_level, imp->imp_level);
1267                 lwi = LWI_INTR(NULL, NULL);
1268                 rc = l_wait_event(req->rq_wait_for_rep,
1269                                   (req->rq_level <= imp->imp_level ||
1270                                    req->rq_err),
1271                                   &lwi);
1272                 DEBUG_REQ(D_HA, req, "\"%s\" awake: (%d > %d)",
1273                           current->comm, req->rq_level, imp->imp_level);
1274
1275                 spin_lock_irqsave(&imp->imp_lock, flags);
1276                 list_del_init(&req->rq_list);
1277
1278                 if (req->rq_err ||
1279                     req->rq_import_generation < imp->imp_generation)
1280                         rc = -EIO;
1281
1282
1283                 if (rc) {
1284                         spin_unlock_irqrestore(&imp->imp_lock, flags);
1285                         GOTO (out, rc);
1286                 }
1287
1288                 CERROR("process %d resumed\n", current->pid);
1289         }
1290
1291         /* XXX this is the same as ptlrpc_set_wait */
1292         LASSERT(list_empty(&req->rq_list));
1293         list_add_tail(&req->rq_list, &imp->imp_sending_list);
1294         spin_unlock_irqrestore(&imp->imp_lock, flags);
1295
1296         rc = ptl_send_rpc(req);
1297         if (rc) {
1298                 /* The DLM's fake imports want to avoid all forms of
1299                  * recovery. */
1300                 if (imp->imp_dlm_fake) {
1301                         spin_lock_irqsave(&imp->imp_lock, flags);
1302                         list_del_init(&req->rq_list);
1303                         spin_unlock_irqrestore(&imp->imp_lock, flags);
1304                         GOTO(out, rc);
1305                 }
1306
1307                 DEBUG_REQ(D_ERROR, req, "send failed (%d); recovering", rc);
1308
1309                 ptlrpc_fail_import(imp, req->rq_import_generation);
1310
1311                 /* If we've been told to not wait, we're done. */
1312                 if (req->rq_level < LUSTRE_CONN_FULL || req->rq_no_recov ||
1313                     obd->obd_no_recov) {
1314                         spin_lock_irqsave(&imp->imp_lock, flags);
1315                         list_del_init(&req->rq_list);
1316                         spin_unlock_irqrestore(&imp->imp_lock, flags);
1317                         GOTO(out, rc);
1318                 }
1319
1320                 /* If we errored, allow the user to interrupt immediately */
1321                 timeout = 1;
1322         } else {
1323                 timeout = req->rq_timeout * HZ;
1324                 DEBUG_REQ(D_NET, req, "-- sleeping");
1325         }
1326 #ifdef __KERNEL__
1327         lwi = LWI_TIMEOUT_INTR(timeout, expired_request, interrupted_request,
1328                                req);
1329         l_wait_event(req->rq_wait_for_rep, ptlrpc_check_reply(req), &lwi);
1330 #else
1331         {
1332                 extern int reply_in_callback(ptl_event_t *ev);
1333                 ptl_event_t reply_ev;
1334                 PtlEQWait(req->rq_connection->c_peer.peer_ni->pni_reply_in_eq_h,
1335                           &reply_ev);
1336                 reply_in_callback(&reply_ev);
1337
1338                 LASSERT (reply_ev.mem_desc.user_ptr == (void *)req);
1339                 // ptlrpc_check_reply(req);
1340                 // not required now it only tests
1341         }
1342 #endif
1343
1344         DEBUG_REQ(D_NET, req, "-- done sleeping");
1345
1346         CDEBUG(D_RPCTRACE, "Completed RPC pname:cluuid:pid:xid:ni:nid:opc "
1347                "%s:%s:%d:"LPU64":%s:"LPX64":%d\n", current->comm,
1348                imp->imp_obd->obd_uuid.uuid,
1349                req->rq_reqmsg->status, req->rq_xid,
1350                conn->c_peer.peer_ni->pni_name, conn->c_peer.peer_nid,
1351                req->rq_reqmsg->opc);
1352
1353         spin_lock_irqsave(&imp->imp_lock, flags);
1354         list_del_init(&req->rq_list);
1355         spin_unlock_irqrestore(&imp->imp_lock, flags);
1356
1357         /* If the reply was received normally, this just grabs the spinlock
1358          * (ensuring the reply callback has returned), sees that
1359          * req->rq_receiving_reply is clear and returns. */
1360         ptlrpc_unregister_reply (req);
1361
1362         if (req->rq_err)
1363                 GOTO(out, rc = -EIO);
1364
1365         /* Resend if we need to, unless we were interrupted. */
1366         if (req->rq_resend && !req->rq_intr) {
1367                 /* ...unless we were specifically told otherwise. */
1368                 if (req->rq_no_resend) {
1369                         spin_lock_irqsave (&req->rq_lock, flags);
1370                         req->rq_no_resend = 0;
1371                         spin_unlock_irqrestore (&req->rq_lock, flags);
1372                         GOTO(out, rc = -ETIMEDOUT);
1373                 }
1374                 spin_lock_irqsave (&req->rq_lock, flags);
1375                 req->rq_resend = 0;
1376                 spin_unlock_irqrestore (&req->rq_lock, flags);
1377                 lustre_msg_add_flags(req->rq_reqmsg, MSG_RESENT);
1378
1379                 if (req->rq_bulk != NULL)
1380                         ptlrpc_unregister_bulk (req);
1381
1382                 DEBUG_REQ(D_HA, req, "resending: ");
1383                 spin_lock_irqsave(&imp->imp_lock, flags);
1384                 goto restart;
1385         }
1386
1387         if (req->rq_intr) {
1388                 /* Should only be interrupted if we timed out. */
1389                 if (!req->rq_timedout)
1390                         DEBUG_REQ(D_ERROR, req,
1391                                   "rq_intr set but rq_timedout not");
1392                 GOTO(out, rc = -EINTR);
1393         }
1394
1395         if (req->rq_timedout) {                 /* non-recoverable timeout */
1396                 GOTO(out, rc = -ETIMEDOUT);
1397         }
1398
1399         if (!req->rq_replied) {
1400                 /* How can this be? -eeb */
1401                 DEBUG_REQ(D_ERROR, req, "!rq_replied: ");
1402                 LBUG();
1403                 GOTO(out, rc = req->rq_status);
1404         }
1405
1406         rc = after_reply (req, &do_restart);
1407         /* NB may return +ve success rc */
1408         if (do_restart) {
1409                 if (req->rq_bulk != NULL)
1410                         ptlrpc_unregister_bulk (req);
1411                 DEBUG_REQ(D_HA, req, "resending: ");
1412                 spin_lock_irqsave(&imp->imp_lock, flags);
1413                 goto restart;
1414         }
1415
1416  out:
1417         if (req->rq_bulk != NULL) {
1418                 if (rc >= 0) {                  /* success so far */
1419                         lwi = LWI_TIMEOUT(timeout, NULL, NULL);
1420                         brc = l_wait_event(req->rq_wait_for_rep,
1421                                            ptlrpc_bulk_complete(req->rq_bulk),
1422                                            &lwi);
1423                         if (brc != 0) {
1424                                 LASSERT (brc == -ETIMEDOUT);
1425                                 CERROR ("Timed out waiting for bulk\n");
1426                                 rc = brc;
1427                         }
1428                 }
1429                 if (rc < 0) {
1430                         /* MDS blocks for put ACKs before replying */
1431                         /* OSC sets rq_no_resend for the time being */
1432                         LASSERT (req->rq_no_resend);
1433                         ptlrpc_unregister_bulk (req);
1434                 }
1435         }
1436
1437         LASSERT (!req->rq_receiving_reply);
1438         req->rq_phase = RQ_PHASE_INTERPRET;
1439         RETURN (rc);
1440 }
1441
1442 int ptlrpc_replay_req(struct ptlrpc_request *req)
1443 {
1444         int rc = 0, old_level, old_status = 0;
1445         // struct ptlrpc_client *cli = req->rq_import->imp_client;
1446         struct l_wait_info lwi;
1447         ENTRY;
1448
1449         /* I don't touch rq_phase here, so the debug log can show what
1450          * state it was left in */
1451
1452         /* Not handling automatic bulk replay yet (or ever?) */
1453         LASSERT (req->rq_bulk == NULL);
1454
1455         DEBUG_REQ(D_NET, req, "about to replay");
1456
1457         /* Update request's state, since we might have a new connection. */
1458         ptlrpc_put_connection(req->rq_connection);
1459         req->rq_connection =
1460                 ptlrpc_connection_addref(req->rq_import->imp_connection);
1461
1462         /* temporarily set request to RECOVD level (reset at out:) */
1463         old_level = req->rq_level;
1464         if (req->rq_replied)
1465                 old_status = req->rq_repmsg->status;
1466         req->rq_level = LUSTRE_CONN_RECOVER;
1467         rc = ptl_send_rpc(req);
1468         if (rc) {
1469                 CERROR("error %d, opcode %d\n", rc, req->rq_reqmsg->opc);
1470                 ptlrpc_cleanup_request_buf(req);
1471                 // up(&cli->cli_rpc_sem);
1472                 GOTO(out, rc = -rc);
1473         }
1474
1475         CDEBUG(D_OTHER, "-- sleeping\n");
1476         lwi = LWI_INTR(NULL, NULL); /* XXX needs timeout, nested recovery */
1477         l_wait_event(req->rq_wait_for_rep, ptlrpc_check_reply(req), &lwi);
1478         CDEBUG(D_OTHER, "-- done\n");
1479
1480         // up(&cli->cli_rpc_sem);
1481
1482         /* If the reply was received normally, this just grabs the spinlock
1483          * (ensuring the reply callback has returned), sees that
1484          * req->rq_receiving_reply is clear and returns. */
1485         ptlrpc_unregister_reply (req);
1486
1487         if (!req->rq_replied) {
1488                 CERROR("Unknown reason for wakeup\n");
1489                 /* XXX Phil - I end up here when I kill obdctl */
1490                 /* ...that's because signals aren't all masked in
1491                  * l_wait_event() -eeb */
1492                 GOTO(out, rc = -EINTR);
1493         }
1494
1495 #if SWAB_PARANOIA
1496         /* Clear reply swab mask; this is a new reply in sender's byte order */
1497         req->rq_rep_swab_mask = 0;
1498 #endif
1499         rc = lustre_unpack_msg(req->rq_repmsg, req->rq_replen);
1500         if (rc) {
1501                 CERROR("unpack_rep failed: %d\n", rc);
1502                 GOTO(out, rc = -EPROTO);
1503         }
1504 #if 0
1505         /* FIXME: Enable when BlueArc makes new release */
1506         if (req->rq_repmsg->type != PTL_RPC_MSG_REPLY &&
1507             req->rq_repmsg->type != PTL_RPC_MSG_ERR) {
1508                 CERROR("invalid packet type received (type=%u)\n",
1509                        req->rq_repmsg->type);
1510                 GOTO(out, rc = -EPROTO);
1511         }
1512 #endif
1513
1514         /* The transno had better not change over replay. */
1515         LASSERT(req->rq_reqmsg->transno == req->rq_repmsg->transno);
1516
1517         CDEBUG(D_NET, "got rep "LPD64"\n", req->rq_xid);
1518
1519         /* let the callback do fixups, possibly including in the request */
1520         if (req->rq_replay_cb)
1521                 req->rq_replay_cb(req);
1522
1523         if (req->rq_replied && req->rq_repmsg->status != old_status) {
1524                 DEBUG_REQ(D_HA, req, "status %d, old was %d",
1525                           req->rq_repmsg->status, old_status);
1526         }
1527
1528  out:
1529         req->rq_level = old_level;
1530         RETURN(rc);
1531 }
1532
1533 void ptlrpc_abort_inflight(struct obd_import *imp)
1534 {
1535         unsigned long flags;
1536         struct list_head *tmp, *n;
1537         ENTRY;
1538
1539         /* Make sure that no new requests get processed for this import.
1540          * ptlrpc_{queue,set}_wait must (and does) hold imp_lock while testing
1541          * this flag and then putting requests on sending_list or delayed_list.
1542          */
1543         spin_lock_irqsave(&imp->imp_lock, flags);
1544
1545         /* XXX locking?  Maybe we should remove each request with the list
1546          * locked?  Also, how do we know if the requests on the list are
1547          * being freed at this time?
1548          */
1549         list_for_each_safe(tmp, n, &imp->imp_sending_list) {
1550                 struct ptlrpc_request *req =
1551                         list_entry(tmp, struct ptlrpc_request, rq_list);
1552
1553                 DEBUG_REQ(D_HA, req, "inflight");
1554
1555                 spin_lock (&req->rq_lock);
1556                 if (req->rq_import_generation < imp->imp_generation) {
1557                         req->rq_err = 1;
1558                         if (req->rq_set != NULL)
1559                                 wake_up(&req->rq_set->set_waitq);
1560                         else
1561                                 wake_up(&req->rq_wait_for_rep);
1562                 }
1563                 spin_unlock (&req->rq_lock);
1564         }
1565
1566         list_for_each_safe(tmp, n, &imp->imp_delayed_list) {
1567                 struct ptlrpc_request *req =
1568                         list_entry(tmp, struct ptlrpc_request, rq_list);
1569
1570                 DEBUG_REQ(D_HA, req, "aborting waiting req");
1571
1572                 spin_lock (&req->rq_lock);
1573                 if (req->rq_import_generation < imp->imp_generation) {
1574                         req->rq_err = 1;
1575                         if (req->rq_set != NULL)
1576                                 wake_up(&req->rq_set->set_waitq);
1577                         else
1578                                 wake_up(&req->rq_wait_for_rep);
1579                         spin_unlock (&req->rq_lock);
1580                 }
1581         }
1582
1583         /* Last chance to free reqs left on the replay list, but we
1584          * will still leak reqs that haven't comitted.  */
1585         if (imp->imp_replayable)
1586                 ptlrpc_free_committed(imp);
1587
1588         spin_unlock_irqrestore(&imp->imp_lock, flags);
1589
1590         EXIT;
1591 }
1592
1593 static __u64 ptlrpc_last_xid = 0;
1594 static spinlock_t ptlrpc_last_xid_lock = SPIN_LOCK_UNLOCKED;
1595
1596 __u64 ptlrpc_next_xid(void)
1597 {
1598         __u64 tmp;
1599         spin_lock(&ptlrpc_last_xid_lock);
1600         tmp = ++ptlrpc_last_xid;
1601         spin_unlock(&ptlrpc_last_xid_lock);
1602         return tmp;
1603 }
1604
1605