lustre/ptlrpc/events.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
  25 #ifdef __KERNEL__
  26 #include <linux/module.h>
  27 #else
  28 #include <liblustre.h>
  29 #endif
  30 #include <linux/obd_class.h>
  31 #include <linux/lustre_net.h>
  32 #include "ptlrpc_internal.h"
  33
  34 #if !defined(__KERNEL__) && CRAY_PORTALS
  35 /* forward ref in events.c */
  36 static void cray_portals_callback(ptl_event_t *ev);
  37 #endif
  38
  39
  40 struct ptlrpc_ni  ptlrpc_interfaces[8];
  41 int               ptlrpc_ninterfaces;
  42
  43 /*
  44  *  Client's outgoing request callback
  45  */
  46 void request_out_callback(ptl_event_t *ev)
  47 {
  48         struct ptlrpc_cb_id   *cbid = ev->md.user_ptr;
  49         struct ptlrpc_request *req = cbid->cbid_arg;
  50         unsigned long          flags;
  51         ENTRY;
  52
  53         LASSERT (ev->type == PTL_EVENT_SEND_END ||
  54                  ev->type == PTL_EVENT_UNLINK);
  55         LASSERT (ev->unlinked);
  56
  57         DEBUG_REQ((ev->ni_fail_type == PTL_NI_OK) ? D_NET : D_ERROR, req,
  58                   "type %d, status %d", ev->type, ev->ni_fail_type);
  59
  60         if (ev->type == PTL_EVENT_UNLINK ||
  61             ev->ni_fail_type != PTL_NI_OK) {
  62
  63                 /* Failed send: make it seem like the reply timed out, just
  64                  * like failing sends in client.c does currently...  */
  65
  66                 spin_lock_irqsave(&req->rq_lock, flags);
  67                 req->rq_net_err = 1;
  68                 spin_unlock_irqrestore(&req->rq_lock, flags);
  69
  70                 ptlrpc_wake_client_req(req);
  71         }
  72
  73         /* this balances the atomic_inc in ptl_send_rpc() */
  74         ptlrpc_req_finished(req);
  75         EXIT;
  76 }
  77
  78 /*
  79  * Client's incoming reply callback
  80  */
  81 void reply_in_callback(ptl_event_t *ev)
  82 {
  83         struct ptlrpc_cb_id   *cbid = ev->md.user_ptr;
  84         struct ptlrpc_request *req = cbid->cbid_arg;
  85         unsigned long flags;
  86         ENTRY;
  87
  88         LASSERT (ev->type == PTL_EVENT_PUT_END ||
  89                  ev->type == PTL_EVENT_UNLINK);
  90         LASSERT (ev->unlinked);
  91         LASSERT (ev->md.start == req->rq_repbuf);
  92         LASSERT (ev->offset == 0);
  93         LASSERT (ev->mlength <= req->rq_repbuf_len);
  94
  95         DEBUG_REQ((ev->ni_fail_type == PTL_NI_OK) ? D_NET : D_ERROR, req,
  96                   "type %d, status %d", ev->type, ev->ni_fail_type);
  97
  98         spin_lock_irqsave (&req->rq_lock, flags);
  99
 100         LASSERT (req->rq_receiving_reply);
 101         req->rq_receiving_reply = 0;
 102
 103         if (ev->type == PTL_EVENT_PUT_END &&
 104             ev->ni_fail_type == PTL_NI_OK) {
 105                 req->rq_replied = 1;
 106                 req->rq_nob_received = ev->mlength;
 107         }
 108
 109         /* NB don't unlock till after wakeup; req can disappear under us
 110          * since we don't have our own ref */
 111         ptlrpc_wake_client_req(req);
 112
 113         spin_unlock_irqrestore (&req->rq_lock, flags);
 114         EXIT;
 115 }
 116
 117 /*
 118  * Client's bulk has been written/read
 119  */
 120 void client_bulk_callback (ptl_event_t *ev)
 121 {
 122         struct ptlrpc_cb_id     *cbid = ev->md.user_ptr;
 123         struct ptlrpc_bulk_desc *desc = cbid->cbid_arg;
 124         unsigned long            flags;
 125         ENTRY;
 126
 127         LASSERT ((desc->bd_type == BULK_PUT_SINK &&
 128                   ev->type == PTL_EVENT_PUT_END) ||
 129                  (desc->bd_type == BULK_GET_SOURCE &&
 130                   ev->type == PTL_EVENT_GET_END) ||
 131                  ev->type == PTL_EVENT_UNLINK);
 132         LASSERT (ev->unlinked);
 133
 134         CDEBUG((ev->ni_fail_type == PTL_NI_OK) ? D_NET : D_ERROR,
 135                "event type %d, status %d, desc %p\n",
 136                ev->type, ev->ni_fail_type, desc);
 137
 138         spin_lock_irqsave (&desc->bd_lock, flags);
 139
 140         LASSERT(desc->bd_network_rw);
 141         desc->bd_network_rw = 0;
 142
 143         if (ev->type != PTL_EVENT_UNLINK &&
 144             ev->ni_fail_type == PTL_NI_OK) {
 145                 desc->bd_success = 1;
 146                 desc->bd_nob_transferred = ev->mlength;
 147         }
 148
 149         /* NB don't unlock till after wakeup; desc can disappear under us
 150          * otherwise */
 151         ptlrpc_wake_client_req(desc->bd_req);
 152
 153         spin_unlock_irqrestore (&desc->bd_lock, flags);
 154         EXIT;
 155 }
 156
 157 /*
 158  * Server's incoming request callback
 159  */
 160 void request_in_callback(ptl_event_t *ev)
 161 {
 162         struct ptlrpc_cb_id               *cbid = ev->md.user_ptr;
 163         struct ptlrpc_request_buffer_desc *rqbd = cbid->cbid_arg;
 164         struct ptlrpc_srv_ni              *srv_ni = rqbd->rqbd_srv_ni;
 165         struct ptlrpc_service             *service = srv_ni->sni_service;
 166         struct ptlrpc_request             *req;
 167         char                              str[PTL_NALFMT_SIZE];
 168         unsigned long                     flags;
 169         ENTRY;
 170
 171         LASSERT (ev->type == PTL_EVENT_PUT_END ||
 172                  ev->type == PTL_EVENT_UNLINK);
 173         LASSERT ((char *)ev->md.start >= rqbd->rqbd_buffer);
 174         LASSERT ((char *)ev->md.start + ev->offset + ev->mlength <=
 175                  rqbd->rqbd_buffer + service->srv_buf_size);
 176
 177         CDEBUG((ev->ni_fail_type == PTL_OK) ? D_NET : D_ERROR,
 178                "event type %d, status %d, service %s\n",
 179                ev->type, ev->ni_fail_type, service->srv_name);
 180
 181         if (ev->unlinked) {
 182                 /* If this is the last request message to fit in the
 183                  * request buffer we can use the request object embedded in
 184                  * rqbd.  Note that if we failed to allocate a request,
 185                  * we'd have to re-post the rqbd, which we can't do in this
 186                  * context. */
 187                 req = &rqbd->rqbd_req;
 188                 memset(req, 0, sizeof (*req));
 189         } else {
 190                 LASSERT (ev->type == PTL_EVENT_PUT_END);
 191                 if (ev->ni_fail_type != PTL_NI_OK) {
 192                         /* We moaned above already... */
 193                         return;
 194                 }
 195                 OBD_ALLOC_GFP(req, sizeof(*req), GFP_ATOMIC);
 196                 if (req == NULL) {
 197                         CERROR("Can't allocate incoming request descriptor: "
 198                                "Dropping %s RPC from %s\n",
 199                                service->srv_name,
 200                                portals_id2str(srv_ni->sni_ni->pni_number,
 201                                               ev->initiator, str));
 202                         return;
 203                 }
 204         }
 205
 206         /* NB we ABSOLUTELY RELY on req being zeroed, so pointers are NULL,
 207          * flags are reset and scalars are zero.  We only set the message
 208          * size to non-zero if this was a successful receive. */
 209         req->rq_xid = ev->match_bits;
 210         req->rq_reqbuf = ev->md.start + ev->offset;
 211         if (ev->type == PTL_EVENT_PUT_END &&
 212             ev->ni_fail_type == PTL_NI_OK)
 213                 req->rq_reqbuf_len = ev->mlength;
 214         do_gettimeofday(&req->rq_arrival_time);
 215         req->rq_peer.peer_id = ev->initiator;
 216         req->rq_peer.peer_ni = rqbd->rqbd_srv_ni->sni_ni;
 217         ptlrpc_id2str(&req->rq_peer, req->rq_peerstr);
 218         req->rq_rqbd = rqbd;
 219 #if CRAY_PORTALS
 220         req->rq_uid = ev->uid;
 221 #endif
 222
 223         spin_lock_irqsave (&service->srv_lock, flags);
 224
 225         if (ev->unlinked) {
 226                 srv_ni->sni_nrqbd_receiving--;
 227                 if (ev->type != PTL_EVENT_UNLINK &&
 228                     srv_ni->sni_nrqbd_receiving == 0) {
 229                         /* This service is off-air on this interface because
 230                          * all its request buffers are busy.  Portals will
 231                          * start dropping incoming requests until more buffers
 232                          * get posted.  NB don't moan if it's because we're
 233                          * tearing down the service. */
 234                         CWARN("All %s %s request buffers busy\n",
 235                               service->srv_name, srv_ni->sni_ni->pni_name);
 236                 }
 237                 /* req takes over the network's ref on rqbd */
 238         } else {
 239                 /* req takes a ref on rqbd */
 240                 rqbd->rqbd_refcount++;
 241         }
 242
 243         list_add_tail(&req->rq_list, &service->srv_request_queue);
 244         service->srv_n_queued_reqs++;
 245
 246         /* NB everything can disappear under us once the request
 247          * has been queued and we unlock, so do the wake now... */
 248         wake_up(&service->srv_waitq);
 249
 250         spin_unlock_irqrestore(&service->srv_lock, flags);
 251         EXIT;
 252 }
 253
 254 /*
 255  *  Server's outgoing reply callback
 256  */
 257 void reply_out_callback(ptl_event_t *ev)
 258 {
 259         struct ptlrpc_cb_id       *cbid = ev->md.user_ptr;
 260         struct ptlrpc_reply_state *rs = cbid->cbid_arg;
 261         struct ptlrpc_srv_ni      *sni = rs->rs_srv_ni;
 262         struct ptlrpc_service     *svc = sni->sni_service;
 263         unsigned long              flags;
 264         ENTRY;
 265
 266         LASSERT (ev->type == PTL_EVENT_SEND_END ||
 267                  ev->type == PTL_EVENT_ACK ||
 268                  ev->type == PTL_EVENT_UNLINK);
 269
 270         if (!rs->rs_difficult) {
 271                 /* I'm totally responsible for freeing "easy" replies */
 272                 LASSERT (ev->unlinked);
 273                 lustre_free_reply_state (rs);
 274                 atomic_dec (&svc->srv_outstanding_replies);
 275                 EXIT;
 276                 return;
 277         }
 278
 279         LASSERT (rs->rs_on_net);
 280
 281         if (ev->unlinked) {
 282                 /* Last network callback */
 283                 spin_lock_irqsave (&svc->srv_lock, flags);
 284                 rs->rs_on_net = 0;
 285                 ptlrpc_schedule_difficult_reply (rs);
 286                 spin_unlock_irqrestore (&svc->srv_lock, flags);
 287         }
 288
 289         EXIT;
 290 }
 291
 292 /*
 293  * Server's bulk completion callback
 294  */
 295 void server_bulk_callback (ptl_event_t *ev)
 296 {
 297         struct ptlrpc_cb_id     *cbid = ev->md.user_ptr;
 298         struct ptlrpc_bulk_desc *desc = cbid->cbid_arg;
 299         unsigned long            flags;
 300         ENTRY;
 301
 302         LASSERT (ev->type == PTL_EVENT_SEND_END ||
 303                  ev->type == PTL_EVENT_UNLINK ||
 304                  (desc->bd_type == BULK_PUT_SOURCE &&
 305                   ev->type == PTL_EVENT_ACK) ||
 306                  (desc->bd_type == BULK_GET_SINK &&
 307                   ev->type == PTL_EVENT_REPLY_END));
 308
 309         CDEBUG((ev->ni_fail_type == PTL_NI_OK) ? D_NET : D_ERROR,
 310                "event type %d, status %d, desc %p\n",
 311                ev->type, ev->ni_fail_type, desc);
 312
 313         spin_lock_irqsave (&desc->bd_lock, flags);
 314
 315         if ((ev->type == PTL_EVENT_ACK ||
 316              ev->type == PTL_EVENT_REPLY_END) &&
 317             ev->ni_fail_type == PTL_NI_OK) {
 318                 /* We heard back from the peer, so even if we get this
 319                  * before the SENT event (oh yes we can), we know we
 320                  * read/wrote the peer buffer and how much... */
 321                 desc->bd_success = 1;
 322                 desc->bd_nob_transferred = ev->mlength;
 323         }
 324
 325         if (ev->unlinked) {
 326                 /* This is the last callback no matter what... */
 327                 desc->bd_network_rw = 0;
 328                 wake_up(&desc->bd_waitq);
 329         }
 330
 331         spin_unlock_irqrestore (&desc->bd_lock, flags);
 332         EXIT;
 333 }
 334
 335 void rawrpc_request_out_callback(ptl_event_t *ev)
 336 {
 337         struct ptlrpc_cb_id   *cbid = ev->md.user_ptr;
 338         struct ptlrpc_request *req = cbid->cbid_arg;
 339
 340         /* we suppose request_out_callback() will drop the
 341          * reference by 1
 342          */
 343         request_out_callback(ev);
 344         rawrpc_req_finished(req);
 345 }
 346
 347 static void ptlrpc_master_callback(ptl_event_t *ev)
 348 {
 349         struct ptlrpc_cb_id *cbid = ev->md.user_ptr;
 350         void (*callback)(ptl_event_t *ev) = cbid->cbid_fn;
 351
 352         /* Honestly, it's best to find out early. */
 353         LASSERT (cbid->cbid_arg != LP_POISON);
 354         LASSERT (callback == request_out_callback ||
 355                  callback == reply_in_callback ||
 356                  callback == client_bulk_callback ||
 357                  callback == request_in_callback ||
 358                  callback == reply_out_callback ||
 359                  callback == server_bulk_callback ||
 360                  callback == rawrpc_request_out_callback);
 361
 362         callback (ev);
 363 }
 364
 365 int ptlrpc_uuid_to_peer (struct obd_uuid *uuid, struct ptlrpc_peer *peer)
 366 {
 367         struct ptlrpc_ni   *pni;
 368         __u32               peer_nal;
 369         ptl_nid_t           peer_nid;
 370         int                 i;
 371         char                str[PTL_NALFMT_SIZE];
 372         int                 rc = lustre_uuid_to_peer((char *)uuid->uuid,
 373                                                      &peer_nal, &peer_nid);
 374         if (rc != 0)
 375                 RETURN (rc);
 376
 377         for (i = 0; i < ptlrpc_ninterfaces; i++) {
 378                 pni = &ptlrpc_interfaces[i];
 379
 380 #ifndef CRAY_PORTALS
 381                 if (pni->pni_number == peer_nal) {
 382 #else
 383                 /* compatible nals but may be from different bridges */
 384                 if (NALID_FROM_IFACE(pni->pni_number) ==
 385                     NALID_FROM_IFACE(peer_nal)) {
 386 #endif
 387                         peer->peer_id.nid = peer_nid;
 388                         peer->peer_id.pid = LUSTRE_SRV_PTL_PID;
 389                         peer->peer_ni = pni;
 390                         return (0);
 391                 }
 392         }
 393
 394         CERROR("Can't find ptlrpc interface for NAL %x, NID %s\n",
 395                peer_nal, portals_nid2str(peer_nal, peer_nid, str));
 396         return (-ENOENT);
 397 }
 398
 399 void ptlrpc_ni_fini(struct ptlrpc_ni *pni)
 400 {
 401         wait_queue_head_t   waitq;
 402         struct l_wait_info  lwi;
 403         int                 rc;
 404         int                 retries;
 405
 406         /* Wait for the event queue to become idle since there may still be
 407          * messages in flight with pending events (i.e. the fire-and-forget
 408          * messages == client requests and "non-difficult" server
 409          * replies */
 410
 411         for (retries = 0;; retries++) {
 412                 rc = PtlEQFree(pni->pni_eq_h);
 413                 switch (rc) {
 414                 default:
 415                         LBUG();
 416
 417                 case PTL_OK:
 418                         PtlNIFini(pni->pni_ni_h);
 419                         return;
 420
 421                 case PTL_EQ_IN_USE:
 422                         if (retries != 0)
 423                                 CWARN("Event queue for %s still busy\n",
 424                                       pni->pni_name);
 425
 426                         /* Wait for a bit */
 427                         init_waitqueue_head(&waitq);
 428                         lwi = LWI_TIMEOUT(2*HZ, NULL, NULL);
 429                         l_wait_event(waitq, 0, &lwi);
 430                         break;
 431                 }
 432         }
 433         /* notreached */
 434 }
 435
 436 ptl_pid_t ptl_get_pid(void)
 437 {
 438         ptl_pid_t        pid;
 439
 440 #ifndef  __KERNEL__
 441         pid = getpid();
 442 #ifdef CRAY_PORTALS
 443         /* hack to keep pid in range accepted by ernal */
 444         pid &= 0xFF;
 445         if (pid == LUSTRE_SRV_PTL_PID)
 446                 pid++;
 447 #endif
 448 #else
 449         pid = LUSTRE_SRV_PTL_PID;
 450 #endif
 451         return pid;
 452 }
 453
 454 int ptlrpc_ni_init(int number, char *name, struct ptlrpc_ni *pni)
 455 {
 456         int              rc;
 457         char             str[20];
 458         ptl_handle_ni_t  nih;
 459         ptl_pid_t        pid;
 460
 461         pid = ptl_get_pid();
 462
 463         /* We're not passing any limits yet... */
 464         rc = PtlNIInit(number, pid, NULL, NULL, &nih);
 465         if (rc != PTL_OK && rc != PTL_IFACE_DUP) {
 466                 CDEBUG (D_NET, "Can't init network interface %s: %d\n",
 467                         name, rc);
 468                 return (-ENOENT);
 469         }
 470
 471         CDEBUG(D_NET, "My pid is: %x\n", ptl_get_pid());
 472
 473         PtlSnprintHandle(str, sizeof(str), nih);
 474         CDEBUG (D_NET, "init %x %s: %s\n", number, name, str);
 475
 476         pni->pni_name = name;
 477         pni->pni_number = number;
 478         pni->pni_ni_h = nih;
 479
 480         pni->pni_eq_h = PTL_INVALID_HANDLE;
 481
 482         /* CAVEAT EMPTOR: how we process portals events is _radically_
 483          * different depending on... */
 484 #ifdef __KERNEL__
 485         /* kernel portals calls our master callback when events are added to
 486          * the event queue.  In fact lustre never pulls events off this queue,
 487          * so it's only sized for some debug history. */
 488         rc = PtlEQAlloc(pni->pni_ni_h, 1024, ptlrpc_master_callback,
 489                         &pni->pni_eq_h);
 490 #else
 491         /* liblustre calls the master callback when it removes events from the
 492          * event queue.  The event queue has to be big enough not to drop
 493          * anything */
 494 # if CRAY_PORTALS
 495         /* cray portals implements a non-standard callback to notify us there
 496          * are buffered events even when the app is not doing a filesystem
 497          * call. */
 498         rc = PtlEQAlloc(pni->pni_ni_h, 10240, cray_portals_callback,
 499                         &pni->pni_eq_h);
 500 # else
 501         rc = PtlEQAlloc(pni->pni_ni_h, 10240, PTL_EQ_HANDLER_NONE,
 502                         &pni->pni_eq_h);
 503 # endif
 504 #endif
 505         if (rc != PTL_OK)
 506                 GOTO (fail, rc = -ENOMEM);
 507
 508         return (0);
 509  fail:
 510         CERROR ("Failed to initialise network interface %s: %d\n",
 511                 name, rc);
 512
 513         /* OK to do complete teardown since we invalidated the handles above */
 514         ptlrpc_ni_fini (pni);
 515         return (rc);
 516 }
 517
 518 #ifndef __KERNEL__
 519 LIST_HEAD(liblustre_wait_callbacks);
 520 void *liblustre_services_callback;
 521
 522 void *
 523 liblustre_register_wait_callback (int (*fn)(void *arg), void *arg)
 524 {
 525         struct liblustre_wait_callback *llwc;
 526
 527         OBD_ALLOC(llwc, sizeof(*llwc));
 528         LASSERT (llwc != NULL);
 529
 530         llwc->llwc_fn = fn;
 531         llwc->llwc_arg = arg;
 532         list_add_tail(&llwc->llwc_list, &liblustre_wait_callbacks);
 533
 534         return (llwc);
 535 }
 536
 537 void
 538 liblustre_deregister_wait_callback (void *opaque)
 539 {
 540         struct liblustre_wait_callback *llwc = opaque;
 541
 542         list_del(&llwc->llwc_list);
 543         OBD_FREE(llwc, sizeof(*llwc));
 544 }
 545
 546 int
 547 liblustre_check_events (int timeout)
 548 {
 549         ptl_event_t ev;
 550         int         rc;
 551         int         i;
 552         ENTRY;
 553
 554         rc = PtlEQPoll(&ptlrpc_interfaces[0].pni_eq_h, 1, timeout * 1000,
 555                        &ev, &i);
 556         if (rc == PTL_EQ_EMPTY)
 557                 RETURN(0);
 558
 559         LASSERT (rc == PTL_EQ_DROPPED || rc == PTL_OK);
 560
 561         /* liblustre: no asynch callback so we can't affort to miss any
 562          * events... */
 563         if (rc == PTL_EQ_DROPPED) {
 564                 CERROR ("Dropped an event!!!\n");
 565                 abort();
 566         }
 567
 568         ptlrpc_master_callback (&ev);
 569         RETURN(1);
 570 }
 571
 572 int liblustre_waiting = 0;
 573
 574 int
 575 liblustre_wait_event (int timeout)
 576 {
 577         struct list_head               *tmp;
 578         struct liblustre_wait_callback *llwc;
 579         int                             found_something = 0;
 580
 581         /* single threaded recursion check... */
 582         liblustre_waiting = 1;
 583
 584         for (;;) {
 585                 /* Deal with all pending events */
 586                 while (liblustre_check_events(0))
 587                         found_something = 1;
 588
 589                 /* Give all registered callbacks a bite at the cherry */
 590                 list_for_each(tmp, &liblustre_wait_callbacks) {
 591                         llwc = list_entry(tmp, struct liblustre_wait_callback,
 592                                           llwc_list);
 593
 594                         if (llwc->llwc_fn(llwc->llwc_arg))
 595                                 found_something = 1;
 596                 }
 597
 598                 if (found_something || timeout == 0)
 599                         break;
 600
 601                 /* Nothing so far, but I'm allowed to block... */
 602                 found_something = liblustre_check_events(timeout);
 603                 if (!found_something)           /* still nothing */
 604                         break;                  /* I timed out */
 605         }
 606
 607         liblustre_waiting = 0;
 608
 609         return found_something;
 610 }
 611
 612 #if CRAY_PORTALS
 613 static void cray_portals_callback(ptl_event_t *ev)
 614 {
 615         /* We get a callback from the client Cray portals implementation
 616          * whenever anyone calls PtlEQPoll(), and an event queue with a
 617          * callback handler has outstanding events.
 618          *
 619          * If it's not liblustre calling PtlEQPoll(), this lets us know we
 620          * have outstanding events which we handle with
 621          * liblustre_wait_event().
 622          *
 623          * Otherwise, we're already eagerly consuming events and we'd
 624          * handle events out of order if we recursed. */
 625         if (!liblustre_waiting)
 626                 liblustre_wait_event(0);
 627 }
 628 #endif
 629 #endif /* __KERNEL__ */
 630
 631 int ptlrpc_default_nal(void)
 632 {
 633         if (ptlrpc_ninterfaces == 0)
 634                 return (-ENOENT);
 635
 636         return (ptlrpc_interfaces[0].pni_number);
 637 }
 638
 639 int ptlrpc_init_portals(void)
 640 {
 641         /* Add new portals network interfaces here.
 642          * Order is irrelevent! */
 643         static struct {
 644                 int   number;
 645                 char *name;
 646         } ptl_nis[] = {
 647 #ifndef CRAY_PORTALS
 648                 {QSWNAL,    "qswnal"},
 649                 {SOCKNAL,   "socknal"},
 650                 {GMNAL,     "gmnal"},
 651                 {OPENIBNAL, "openibnal"},
 652                 {IIBNAL,    "iibnal"},
 653                 {VIBNAL,    "vibnal"},
 654                 {TCPNAL,    "tcpnal"},
 655                 {LONAL,     "lonal"},
 656                 {RANAL,     "ranal"},
 657 #else
 658                 {CRAY_KERN_NAL, "cray_kern_nal"},
 659                 {CRAY_QK_NAL, "cray_qk_nal"},
 660                 {CRAY_USER_NAL, "cray_user_nal"},
 661 #endif
 662         };
 663         int   rc;
 664         int   i;
 665
 666         LASSERT(ptlrpc_ninterfaces == 0);
 667
 668         for (i = 0; i < sizeof (ptl_nis) / sizeof (ptl_nis[0]); i++) {
 669                 LASSERT(ptlrpc_ninterfaces < (sizeof(ptlrpc_interfaces) /
 670                                               sizeof(ptlrpc_interfaces[0])));
 671
 672                 rc = ptlrpc_ni_init(ptl_nis[i].number, ptl_nis[i].name,
 673                                     &ptlrpc_interfaces[ptlrpc_ninterfaces]);
 674                 if (rc == 0)
 675                         ptlrpc_ninterfaces++;
 676         }
 677
 678         if (ptlrpc_ninterfaces == 0) {
 679                 CERROR("network initialisation failed: is a NAL module "
 680                        "loaded?\n");
 681                 return -EIO;
 682         }
 683 #ifndef __KERNEL__
 684         liblustre_services_callback =
 685                 liblustre_register_wait_callback(&liblustre_check_services, NULL);
 686 #endif
 687         return 0;
 688 }
 689
 690 void ptlrpc_exit_portals(void)
 691 {
 692 #ifndef __KERNEL__
 693         liblustre_deregister_wait_callback(liblustre_services_callback);
 694 #endif
 695         while (ptlrpc_ninterfaces > 0)
 696                 ptlrpc_ni_fini (&ptlrpc_interfaces[--ptlrpc_ninterfaces]);
 697 }