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 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
  25 #include <linux/obd_support.h>
  26 #include <linux/obd_class.h>
  27 #include <linux/lustre_lib.h>
  28 #include <linux/lustre_ha.h>
  29 #include <linux/lustre_import.h>
  30
  31 void ptlrpc_init_client(int req_portal, int rep_portal, char *name,
  32                         struct ptlrpc_client *cl)
  33 {
  34         cl->cli_request_portal = req_portal;
  35         cl->cli_reply_portal   = rep_portal;
  36         cl->cli_name           = name;
  37 }
  38
  39 __u8 *ptlrpc_req_to_uuid(struct ptlrpc_request *req)
  40 {
  41         return req->rq_connection->c_remote_uuid;
  42 }
  43
  44 struct ptlrpc_connection *ptlrpc_uuid_to_connection(obd_uuid_t uuid)
  45 {
  46         struct ptlrpc_connection *c;
  47         struct lustre_peer peer;
  48         int err;
  49
  50         err = kportal_uuid_to_peer(uuid, &peer);
  51         if (err != 0) {
  52                 CERROR("cannot find peer %s!\n", uuid);
  53                 return NULL;
  54         }
  55
  56         c = ptlrpc_get_connection(&peer, uuid);
  57         if (c) {
  58                 memcpy(c->c_remote_uuid, uuid, sizeof(c->c_remote_uuid));
  59                 c->c_epoch++;
  60         }
  61
  62         CDEBUG(D_INFO, "%s -> %p\n", uuid, c);
  63
  64         return c;
  65 }
  66
  67 void ptlrpc_readdress_connection(struct ptlrpc_connection *conn,obd_uuid_t uuid)
  68 {
  69         struct lustre_peer peer;
  70         int err;
  71
  72         err = kportal_uuid_to_peer(uuid, &peer);
  73         if (err != 0) {
  74                 CERROR("cannot find peer %s!\n", uuid);
  75                 return;
  76         }
  77
  78         memcpy(&conn->c_peer, &peer, sizeof(peer));
  79         return;
  80 }
  81
  82 struct ptlrpc_bulk_desc *ptlrpc_prep_bulk(struct ptlrpc_connection *conn)
  83 {
  84         struct ptlrpc_bulk_desc *desc;
  85
  86         OBD_ALLOC(desc, sizeof(*desc));
  87         if (desc != NULL) {
  88                 desc->bd_connection = ptlrpc_connection_addref(conn);
  89                 atomic_set(&desc->bd_refcount, 1);
  90                 init_waitqueue_head(&desc->bd_waitq);
  91                 INIT_LIST_HEAD(&desc->bd_page_list);
  92                 ptl_set_inv_handle(&desc->bd_md_h);
  93                 ptl_set_inv_handle(&desc->bd_me_h);
  94         }
  95
  96         return desc;
  97 }
  98
  99 struct ptlrpc_bulk_page *ptlrpc_prep_bulk_page(struct ptlrpc_bulk_desc *desc)
 100 {
 101         struct ptlrpc_bulk_page *bulk;
 102
 103         OBD_ALLOC(bulk, sizeof(*bulk));
 104         if (bulk != NULL) {
 105                 bulk->bp_desc = desc;
 106                 list_add_tail(&bulk->bp_link, &desc->bd_page_list);
 107                 desc->bd_page_count++;
 108         }
 109         return bulk;
 110 }
 111
 112 void ptlrpc_free_bulk(struct ptlrpc_bulk_desc *desc)
 113 {
 114         struct list_head *tmp, *next;
 115         ENTRY;
 116         if (desc == NULL) {
 117                 EXIT;
 118                 return;
 119         }
 120
 121         list_for_each_safe(tmp, next, &desc->bd_page_list) {
 122                 struct ptlrpc_bulk_page *bulk;
 123                 bulk = list_entry(tmp, struct ptlrpc_bulk_page, bp_link);
 124                 ptlrpc_free_bulk_page(bulk);
 125         }
 126
 127         ptlrpc_put_connection(desc->bd_connection);
 128
 129         OBD_FREE(desc, sizeof(*desc));
 130         EXIT;
 131 }
 132
 133 void ptlrpc_free_bulk_page(struct ptlrpc_bulk_page *bulk)
 134 {
 135         ENTRY;
 136         if (bulk == NULL) {
 137                 EXIT;
 138                 return;
 139         }
 140
 141         list_del(&bulk->bp_link);
 142         bulk->bp_desc->bd_page_count--;
 143         OBD_FREE(bulk, sizeof(*bulk));
 144         EXIT;
 145 }
 146
 147 struct ptlrpc_request *ptlrpc_prep_req(struct obd_import *imp, int opcode,
 148                                        int count, int *lengths, char **bufs)
 149 {
 150         struct ptlrpc_connection *conn = imp->imp_connection;
 151         struct ptlrpc_request *request;
 152         int rc;
 153         ENTRY;
 154
 155         OBD_ALLOC(request, sizeof(*request));
 156         if (!request) {
 157                 CERROR("request allocation out of memory\n");
 158                 RETURN(NULL);
 159         }
 160
 161         rc = lustre_pack_msg(count, lengths, bufs,
 162                              &request->rq_reqlen, &request->rq_reqmsg);
 163         if (rc) {
 164                 CERROR("cannot pack request %d\n", rc);
 165                 OBD_FREE(request, sizeof(*request));
 166                 RETURN(NULL);
 167         }
 168
 169         request->rq_level = LUSTRE_CONN_FULL;
 170         request->rq_type = PTL_RPC_MSG_REQUEST;
 171         request->rq_import = imp;
 172         request->rq_connection = ptlrpc_connection_addref(conn);
 173
 174         INIT_LIST_HEAD(&request->rq_list);
 175         /*
 176          * This will be reduced once when the sender is finished (waiting for
 177          * reply, f.e.), and once when the request has been committed and is
 178          * removed from the to-be-committed list.
 179          *
 180          * Also, the refcount will be increased in ptl_send_rpc immediately
 181          * before we hand it off to portals, and there will be a corresponding
 182          * decrease in request_out_cb (which is called to indicate that portals
 183          * is finished with the request, and it can be safely freed).
 184          *
 185          * (Except in the DLM server case, where it will be dropped twice
 186          * by the sender, and then the last time by request_out_callback.)
 187          */
 188         atomic_set(&request->rq_refcount, 2);
 189
 190         spin_lock(&conn->c_lock);
 191         request->rq_xid = HTON__u32(++conn->c_xid_out);
 192         spin_unlock(&conn->c_lock);
 193
 194         request->rq_reqmsg->magic = PTLRPC_MSG_MAGIC;
 195         request->rq_reqmsg->version = PTLRPC_MSG_VERSION;
 196         request->rq_reqmsg->opc = HTON__u32(opcode);
 197
 198         ptlrpc_hdl2req(request, &imp->imp_handle);
 199         RETURN(request);
 200 }
 201
 202 void ptlrpc_req_finished(struct ptlrpc_request *request)
 203 {
 204         if (request == NULL)
 205                 return;
 206
 207         if (atomic_dec_and_test(&request->rq_refcount))
 208                 ptlrpc_free_req(request);
 209 }
 210
 211 void ptlrpc_free_req(struct ptlrpc_request *request)
 212 {
 213         ENTRY;
 214         if (request == NULL) {
 215                 EXIT;
 216                 return;
 217         }
 218
 219         if (atomic_read(&request->rq_refcount) != 0) {
 220                 CERROR("freeing request %p (%d->%s:%d) with refcount %d\n",
 221                        request, request->rq_reqmsg->opc,
 222                        request->rq_connection->c_remote_uuid,
 223                        request->rq_import->imp_client->cli_request_portal,
 224                        request->rq_refcount);
 225                 /* LBUG(); */
 226         }
 227
 228         if (request->rq_repmsg != NULL) {
 229                 OBD_FREE(request->rq_repmsg, request->rq_replen);
 230                 request->rq_repmsg = NULL;
 231                 request->rq_reply_md.start = NULL;
 232         }
 233         if (request->rq_reqmsg != NULL) {
 234                 OBD_FREE(request->rq_reqmsg, request->rq_reqlen);
 235                 request->rq_reqmsg = NULL;
 236         }
 237
 238         if (request->rq_connection) {
 239                 spin_lock(&request->rq_connection->c_lock);
 240                 list_del_init(&request->rq_list);
 241                 spin_unlock(&request->rq_connection->c_lock);
 242         }
 243
 244         ptlrpc_put_connection(request->rq_connection);
 245         OBD_FREE(request, sizeof(*request));
 246         EXIT;
 247 }
 248
 249 static int ptlrpc_check_reply(struct ptlrpc_request *req)
 250 {
 251         int rc = 0;
 252
 253         if (req->rq_repmsg != NULL) {
 254                 req->rq_transno = NTOH__u64(req->rq_repmsg->transno);
 255                 req->rq_flags |= PTL_RPC_FL_REPLIED;
 256                 GOTO(out, rc = 1);
 257         }
 258
 259         if (req->rq_flags & PTL_RPC_FL_RESEND) {
 260                 CERROR("-- RESTART --\n");
 261                 GOTO(out, rc = 1);
 262         }
 263
 264  out:
 265         CDEBUG(D_NET, "req = %p, rc = %d\n", req, rc);
 266         return rc;
 267 }
 268
 269 int ptlrpc_check_status(struct ptlrpc_request *req, int err)
 270 {
 271         ENTRY;
 272
 273         if (err != 0) {
 274                 CERROR("err is %d\n", err);
 275                 RETURN(err);
 276         }
 277
 278         if (req == NULL) {
 279                 CERROR("req == NULL\n");
 280                 RETURN(-ENOMEM);
 281         }
 282
 283         if (req->rq_repmsg == NULL) {
 284                 CERROR("req->rq_repmsg == NULL\n");
 285                 RETURN(-ENOMEM);
 286         }
 287
 288         err = req->rq_repmsg->status;
 289         if (req->rq_repmsg->type == NTOH__u32(PTL_RPC_MSG_ERR)) {
 290                 CERROR("req->rq_repmsg->type == PTL_RPC_MSG_ERR\n");
 291                 RETURN(err ? err : -EINVAL);
 292         }
 293
 294         if (err != 0) {
 295                 if (err < 0)
 296                         CERROR("req->rq_repmsg->status is %d\n", err);
 297                 else
 298                         CDEBUG(D_INFO, "req->rq_repmsg->status is %d\n", err);
 299                 /* XXX: translate this error from net to host */
 300                 RETURN(err);
 301         }
 302
 303         RETURN(0);
 304 }
 305
 306 static void ptlrpc_cleanup_request_buf(struct ptlrpc_request *request)
 307 {
 308         OBD_FREE(request->rq_reqmsg, request->rq_reqlen);
 309         request->rq_reqmsg = NULL;
 310         request->rq_reqlen = 0;
 311 }
 312
 313 /* Abort this request and cleanup any resources associated with it. */
 314 static int ptlrpc_abort(struct ptlrpc_request *request)
 315 {
 316         /* First remove the ME for the reply; in theory, this means
 317          * that we can tear down the buffer safely. */
 318         PtlMEUnlink(request->rq_reply_me_h);
 319         OBD_FREE(request->rq_reply_md.start, request->rq_replen);
 320         request->rq_repmsg = NULL;
 321         request->rq_replen = 0;
 322         return 0;
 323 }
 324
 325 /* caller must hold conn->c_lock */
 326 void ptlrpc_free_committed(struct ptlrpc_connection *conn)
 327 {
 328         struct list_head *tmp, *saved;
 329         struct ptlrpc_request *req;
 330
 331 restart:
 332         list_for_each_safe(tmp, saved, &conn->c_sending_head) {
 333                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
 334
 335                 if (req->rq_flags & PTL_RPC_FL_REPLAY) {
 336                         CDEBUG(D_INFO, "Keeping req %p xid "LPD64" for replay\n",
 337                                req, req->rq_xid);
 338                         continue;
 339                 }
 340
 341                 if (!(req->rq_flags & PTL_RPC_FL_REPLIED)) {
 342                         CDEBUG(D_INFO, "Keeping in-flight req %p xid "LPD64
 343                                " for replay\n", req, req->rq_xid);
 344                         continue;
 345                 }
 346
 347                 /* not yet committed */
 348                 if (req->rq_transno > conn->c_last_committed)
 349                         break;
 350
 351                 CDEBUG(D_INFO, "Marking request %p xid %Ld as committed "
 352                        "transno=%Lu, last_committed=%Lu\n", req,
 353                        (long long)req->rq_xid, (long long)req->rq_transno,
 354                        (long long)conn->c_last_committed);
 355                 if (atomic_dec_and_test(&req->rq_refcount)) {
 356                         /* We do this to prevent free_req deadlock.
 357                          * Restarting after each removal is not so bad, as we are
 358                          * almost always deleting the first item in the list.
 359                          *
 360                          * If we use a recursive lock here, we can skip the
 361                          * unlock/lock/restart sequence.
 362                          */
 363                         spin_unlock(&conn->c_lock);
 364                         ptlrpc_free_req(req);
 365                         spin_lock(&conn->c_lock);
 366                         goto restart;
 367                 } else {
 368                         list_del(&req->rq_list);
 369                         list_add(&req->rq_list, &conn->c_dying_head);
 370                 }
 371         }
 372
 373         EXIT;
 374         return;
 375 }
 376
 377 void ptlrpc_cleanup_client(struct obd_import *imp)
 378 {
 379         struct list_head *tmp, *saved;
 380         struct ptlrpc_request *req;
 381         struct ptlrpc_connection *conn = imp->imp_connection;
 382         ENTRY;
 383
 384         LASSERT(conn);
 385
 386 restart1:
 387         spin_lock(&conn->c_lock);
 388         list_for_each_safe(tmp, saved, &conn->c_sending_head) {
 389                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
 390                 if (req->rq_import != imp)
 391                         continue;
 392                 /* XXX we should make sure that nobody's sleeping on these! */
 393                 CDEBUG(D_INFO, "Cleaning req %p from sending list.\n", req);
 394                 list_del_init(&req->rq_list);
 395                 req->rq_import = NULL;
 396                 spin_unlock(&conn->c_lock);
 397                 ptlrpc_req_finished(req);
 398                 goto restart1;
 399         }
 400 restart2:
 401         list_for_each_safe(tmp, saved, &conn->c_dying_head) {
 402                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
 403                 if (req->rq_import != imp)
 404                         continue;
 405                 CERROR("Request %p is on the dying list at cleanup!\n", req);
 406                 list_del_init(&req->rq_list);
 407                 req->rq_import = NULL;
 408                 spin_unlock(&conn->c_lock);
 409                 ptlrpc_req_finished(req);
 410                 spin_lock(&conn->c_lock);
 411                 goto restart2;
 412         }
 413         spin_unlock(&conn->c_lock);
 414
 415         EXIT;
 416         return;
 417 }
 418
 419 void ptlrpc_continue_req(struct ptlrpc_request *req)
 420 {
 421         ENTRY;
 422         CDEBUG(D_HA, "continue delayed request "LPD64" opc %d\n",
 423                req->rq_xid, req->rq_reqmsg->opc);
 424         req->rq_reqmsg->addr = req->rq_import->imp_handle.addr;
 425         req->rq_reqmsg->cookie = req->rq_import->imp_handle.cookie;
 426         wake_up(&req->rq_wait_for_rep);
 427         EXIT;
 428 }
 429
 430 void ptlrpc_resend_req(struct ptlrpc_request *req)
 431 {
 432         ENTRY;
 433         CDEBUG(D_HA, "resend request "LPD64", opc %d\n",
 434                req->rq_xid, req->rq_reqmsg->opc);
 435         req->rq_reqmsg->addr = req->rq_import->imp_handle.addr;
 436         req->rq_reqmsg->cookie = req->rq_import->imp_handle.cookie;
 437         req->rq_status = -EAGAIN;
 438         req->rq_level = LUSTRE_CONN_RECOVD;
 439         req->rq_flags |= PTL_RPC_FL_RESEND;
 440         req->rq_flags &= ~PTL_RPC_FL_TIMEOUT;
 441         wake_up(&req->rq_wait_for_rep);
 442         EXIT;
 443 }
 444
 445 void ptlrpc_restart_req(struct ptlrpc_request *req)
 446 {
 447         ENTRY;
 448         CDEBUG(D_HA, "restart completed request "LPD64", opc %d\n",
 449                req->rq_xid, req->rq_reqmsg->opc);
 450         req->rq_status = -ERESTARTSYS;
 451         req->rq_flags |= PTL_RPC_FL_RECOVERY;
 452         req->rq_flags &= ~PTL_RPC_FL_TIMEOUT;
 453         wake_up(&req->rq_wait_for_rep);
 454         EXIT;
 455 }
 456
 457 static int expired_request(void *data)
 458 {
 459         struct ptlrpc_request *req = data;
 460
 461         ENTRY;
 462         CERROR("req xid "LPD64" op %d: timeout on conn to %s:%d\n",
 463                (unsigned long long)req->rq_xid, req->rq_reqmsg->opc,
 464                req->rq_connection->c_remote_uuid,
 465                req->rq_import->imp_client->cli_request_portal);
 466         req->rq_flags |= PTL_RPC_FL_TIMEOUT;
 467         if (!req->rq_import->imp_connection->c_recovd_data.rd_recovd)
 468                 RETURN(1);
 469
 470         req->rq_timeout = 0;
 471         req->rq_connection->c_level = LUSTRE_CONN_RECOVD;
 472         recovd_conn_fail(req->rq_import->imp_connection);
 473
 474         /* If this request is for recovery or other primordial tasks,
 475          * don't go back to sleep.
 476          */
 477         if (req->rq_level < LUSTRE_CONN_FULL)
 478                 RETURN(1);
 479         RETURN(0);
 480 }
 481
 482 static int interrupted_request(void *data)
 483 {
 484         struct ptlrpc_request *req = data;
 485         ENTRY;
 486         req->rq_flags |= PTL_RPC_FL_INTR;
 487         RETURN(1); /* ignored, as of this writing */
 488 }
 489
 490 int ptlrpc_queue_wait(struct ptlrpc_request *req)
 491 {
 492         int rc = 0;
 493         struct l_wait_info lwi;
 494         struct ptlrpc_client *cli = req->rq_import->imp_client;
 495         struct ptlrpc_connection *conn = req->rq_import->imp_connection;
 496         ENTRY;
 497
 498         init_waitqueue_head(&req->rq_wait_for_rep);
 499         CDEBUG(D_NET, "subsys: %s req "LPD64" opc %d level %d, conn level %d\n",
 500                cli->cli_name, req->rq_xid, req->rq_reqmsg->opc, req->rq_level,
 501                req->rq_connection->c_level);
 502
 503         /* XXX probably both an import and connection level are needed */
 504         if (req->rq_level > conn->c_level) {
 505                 spin_lock(&conn->c_lock);
 506                 if (conn->c_flags & CONN_INVALID) {
 507                         /* being torn down by "umount -f" */
 508                         CERROR("req xid "LPD64" op %d to %s:%d: CONN_INVALID\n",
 509                                (unsigned long long)req->rq_xid,
 510                                req->rq_reqmsg->opc,
 511                                req->rq_connection->c_remote_uuid,
 512                                req->rq_import->imp_client->cli_request_portal);
 513                         spin_unlock(&conn->c_lock);
 514                         RETURN(-EIO);
 515                 }
 516                 list_del(&req->rq_list);
 517                 list_add_tail(&req->rq_list, &conn->c_delayed_head);
 518                 spin_unlock(&conn->c_lock);
 519
 520                 CDEBUG(D_HA, "req xid "LPD64" op %d to %s:%d: waiting for "
 521                        "recovery (%d < %d)\n",
 522                        (unsigned long long)req->rq_xid, req->rq_reqmsg->opc,
 523                        req->rq_connection->c_remote_uuid,
 524                        req->rq_import->imp_client->cli_request_portal,
 525                        req->rq_level, conn->c_level);
 526
 527                 lwi = LWI_INTR(NULL, NULL);
 528                 rc = l_wait_event(req->rq_wait_for_rep,
 529                                   (req->rq_level <= conn->c_level) ||
 530                                   (req->rq_flags & PTL_RPC_FL_ERR), &lwi);
 531
 532                 spin_lock(&conn->c_lock);
 533                 list_del_init(&req->rq_list);
 534                 spin_unlock(&conn->c_lock);
 535
 536                 if (req->rq_flags & PTL_RPC_FL_ERR)
 537                         RETURN(-EIO);
 538
 539                 if (rc)
 540                         RETURN(rc);
 541
 542                 CERROR("process %d resumed\n", current->pid);
 543         }
 544  resend:
 545         req->rq_timeout = obd_timeout;
 546         spin_lock(&conn->c_lock);
 547         if (conn->c_flags & CONN_INVALID) {
 548                 CERROR("req xid "LPD64" op %d to %s:%d: CONN_INVALID\n",
 549                        (unsigned long long)req->rq_xid, req->rq_reqmsg->opc,
 550                        req->rq_connection->c_remote_uuid,
 551                        req->rq_import->imp_client->cli_request_portal);
 552                 spin_unlock(&conn->c_lock); /* being torn down by "umount -f" */
 553                 RETURN(-EIO);
 554         }
 555
 556         list_del(&req->rq_list);
 557         list_add_tail(&req->rq_list, &conn->c_sending_head);
 558         spin_unlock(&conn->c_lock);
 559         rc = ptl_send_rpc(req);
 560         if (rc) {
 561                 CDEBUG(D_HA, "error %d, opcode %d, need recovery\n", rc,
 562                        req->rq_reqmsg->opc);
 563                 /* the sleep below will time out, triggering recovery */
 564         }
 565
 566         CDEBUG(D_NET, "-- sleeping on req xid "LPD64" op %d to %s:%d\n",
 567                        (unsigned long long)req->rq_xid, req->rq_reqmsg->opc,
 568                        req->rq_connection->c_remote_uuid,
 569                        req->rq_import->imp_client->cli_request_portal);
 570         lwi = LWI_TIMEOUT_INTR(req->rq_timeout * HZ, expired_request,
 571                                interrupted_request,req);
 572         l_wait_event(req->rq_wait_for_rep, ptlrpc_check_reply(req), &lwi);
 573         CDEBUG(D_NET, "-- done sleeping on req xid "LPD64" op %d to %s:%d\n",
 574                        (unsigned long long)req->rq_xid, req->rq_reqmsg->opc,
 575                        req->rq_connection->c_remote_uuid,
 576                        req->rq_import->imp_client->cli_request_portal);
 577
 578         /* Don't resend if we were interrupted. */
 579         if ((req->rq_flags & (PTL_RPC_FL_RESEND | PTL_RPC_FL_INTR)) ==
 580             PTL_RPC_FL_RESEND) {
 581                 req->rq_flags &= ~PTL_RPC_FL_RESEND;
 582                 CDEBUG(D_HA, "req xid "LPD64" op %d to %s:%d\n",
 583                        (unsigned long long)req->rq_xid, req->rq_reqmsg->opc,
 584                        req->rq_connection->c_remote_uuid,
 585                        req->rq_import->imp_client->cli_request_portal);
 586                 /* we'll get sent again, so balance 2nd request_out_callback */
 587                 goto resend;
 588         }
 589
 590         // up(&cli->cli_rpc_sem);
 591         if (req->rq_flags & PTL_RPC_FL_INTR) {
 592                 if (!(req->rq_flags & PTL_RPC_FL_TIMEOUT))
 593                         LBUG(); /* should only be interrupted if we timed out */
 594                 /* Clean up the dangling reply buffers */
 595                 ptlrpc_abort(req);
 596                 GOTO(out, rc = -EINTR);
 597         }
 598
 599         if (req->rq_flags & PTL_RPC_FL_TIMEOUT)
 600                 GOTO(out, rc = -ETIMEDOUT);
 601
 602         if (!(req->rq_flags & PTL_RPC_FL_REPLIED))
 603                 GOTO(out, rc = req->rq_status);
 604
 605         rc = lustre_unpack_msg(req->rq_repmsg, req->rq_replen);
 606         if (rc) {
 607                 CERROR("unpack_rep failed: %d\n", rc);
 608                 GOTO(out, rc);
 609         }
 610 #if 0
 611         /* FIXME: Enable when BlueArc makes new release */
 612         if (req->rq_repmsg->type != PTL_RPC_MSG_REPLY &&
 613             req->rq_repmsg->type != PTL_RPC_MSG_ERR) {
 614                 CERROR("invalid packet type received (type=%u)\n",
 615                        req->rq_repmsg->type);
 616                 LBUG();
 617                 GOTO(out, rc = -EINVAL);
 618         }
 619 #endif
 620         CDEBUG(D_NET, "got rep "LPD64"\n", req->rq_xid);
 621         if (req->rq_repmsg->status == 0)
 622                 CDEBUG(D_NET, "--> buf %p len %d status %d\n", req->rq_repmsg,
 623                        req->rq_replen, req->rq_repmsg->status);
 624
 625         spin_lock(&conn->c_lock);
 626         conn->c_last_xid = req->rq_repmsg->last_xid;
 627         conn->c_last_committed = req->rq_repmsg->last_committed;
 628         ptlrpc_free_committed(conn);
 629         spin_unlock(&conn->c_lock);
 630
 631         EXIT;
 632  out:
 633         return rc;
 634 }
 635
 636 int ptlrpc_replay_req(struct ptlrpc_request *req)
 637 {
 638         int rc = 0;
 639         // struct ptlrpc_client *cli = req->rq_import->imp_client;
 640         struct l_wait_info lwi;
 641         ENTRY;
 642
 643         init_waitqueue_head(&req->rq_wait_for_rep);
 644         CDEBUG(D_NET, "req "LPD64" opc %d level %d, conn level %d\n",
 645                req->rq_xid, req->rq_reqmsg->opc, req->rq_level,
 646                req->rq_connection->c_level);
 647
 648         req->rq_timeout = obd_timeout;
 649         req->rq_reqmsg->addr = req->rq_import->imp_handle.addr;
 650         req->rq_reqmsg->cookie = req->rq_import->imp_handle.cookie;
 651
 652         rc = ptl_send_rpc(req);
 653         if (rc) {
 654                 CERROR("error %d, opcode %d\n", rc, req->rq_reqmsg->opc);
 655                 ptlrpc_cleanup_request_buf(req);
 656                 // up(&cli->cli_rpc_sem);
 657                 RETURN(-rc);
 658         }
 659
 660         CDEBUG(D_OTHER, "-- sleeping\n");
 661         lwi = LWI_INTR(NULL, NULL); /* XXX needs timeout, nested recovery */
 662         l_wait_event(req->rq_wait_for_rep, ptlrpc_check_reply(req), &lwi);
 663         CDEBUG(D_OTHER, "-- done\n");
 664
 665         // up(&cli->cli_rpc_sem);
 666
 667         if (!(req->rq_flags & PTL_RPC_FL_REPLIED)) {
 668                 CERROR("Unknown reason for wakeup\n");
 669                 /* XXX Phil - I end up here when I kill obdctl */
 670                 ptlrpc_abort(req);
 671                 GOTO(out, rc = -EINTR);
 672         }
 673
 674         rc = lustre_unpack_msg(req->rq_repmsg, req->rq_replen);
 675         if (rc) {
 676                 CERROR("unpack_rep failed: %d\n", rc);
 677                 GOTO(out, rc);
 678         }
 679
 680         CDEBUG(D_NET, "got rep "LPD64"\n", req->rq_xid);
 681         if (req->rq_repmsg->status == 0)
 682                 CDEBUG(D_NET, "--> buf %p len %d status %d\n", req->rq_repmsg,
 683                        req->rq_replen, req->rq_repmsg->status);
 684         else {
 685                 CERROR("recovery failed: ");
 686                 CERROR("req "LPD64" opc %d level %d, conn level %d\n",
 687                        req->rq_xid, req->rq_reqmsg->opc, req->rq_level,
 688                        req->rq_connection->c_level);
 689                 LBUG();
 690         }
 691
 692         if (req->rq_replay_cb)
 693                 req->rq_replay_cb(req, &req->rq_replay_cb_handle);
 694
 695  out:
 696         RETURN(rc);
 697 }