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