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