lustre/ptlrpc/events.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36
  37 #define DEBUG_SUBSYSTEM S_RPC
  38
  39 #ifndef __KERNEL__
  40 # include <liblustre.h>
  41 #else
  42 # ifdef __mips64__
  43 #  include <linux/kernel.h>
  44 # endif
  45 #endif
  46 #include <obd_class.h>
  47 #include <lustre_net.h>
  48 #include <lustre_sec.h>
  49 #include "ptlrpc_internal.h"
  50
  51 lnet_handle_eq_t   ptlrpc_eq_h;
  52
  53 /*
  54  *  Client's outgoing request callback
  55  */
  56 void request_out_callback(lnet_event_t *ev)
  57 {
  58         struct ptlrpc_cb_id   *cbid = ev->md.user_ptr;
  59         struct ptlrpc_request *req = cbid->cbid_arg;
  60         ENTRY;
  61
  62         LASSERT (ev->type == LNET_EVENT_SEND ||
  63                  ev->type == LNET_EVENT_UNLINK);
  64         LASSERT (ev->unlinked);
  65
  66         DEBUG_REQ((ev->status == 0) ? D_NET : D_ERROR, req,
  67                   "type %d, status %d", ev->type, ev->status);
  68
  69         sptlrpc_request_out_callback(req);
  70
  71         if (ev->type == LNET_EVENT_UNLINK || ev->status != 0) {
  72
  73                 /* Failed send: make it seem like the reply timed out, just
  74                  * like failing sends in client.c does currently...  */
  75
  76                 spin_lock(&req->rq_lock);
  77                 req->rq_net_err = 1;
  78                 spin_unlock(&req->rq_lock);
  79
  80                 ptlrpc_client_wake_req(req);
  81         }
  82
  83         ptlrpc_req_finished(req);
  84
  85         EXIT;
  86 }
  87
  88 /*
  89  * Client's incoming reply callback
  90  */
  91 void reply_in_callback(lnet_event_t *ev)
  92 {
  93         struct ptlrpc_cb_id   *cbid = ev->md.user_ptr;
  94         struct ptlrpc_request *req = cbid->cbid_arg;
  95         ENTRY;
  96
  97         DEBUG_REQ((ev->status == 0) ? D_NET : D_ERROR, req,
  98                   "type %d, status %d", ev->type, ev->status);
  99
 100         LASSERT (ev->type == LNET_EVENT_PUT || ev->type == LNET_EVENT_UNLINK);
 101         LASSERT (ev->md.start == req->rq_repbuf);
 102         LASSERT (ev->mlength <= req->rq_repbuf_len);
 103         /* We've set LNET_MD_MANAGE_REMOTE for all outgoing requests
 104            for adaptive timeouts' early reply. */
 105         LASSERT((ev->md.options & LNET_MD_MANAGE_REMOTE) != 0);
 106
 107         spin_lock(&req->rq_lock);
 108
 109         req->rq_receiving_reply = 0;
 110         req->rq_early = 0;
 111
 112         if (ev->status)
 113                 goto out_wake;
 114         if (ev->type == LNET_EVENT_UNLINK) {
 115                 req->rq_must_unlink = 0;
 116                 DEBUG_REQ(D_RPCTRACE, req, "unlink");
 117                 goto out_wake;
 118         }
 119
 120         if ((ev->offset == 0) &&
 121             ((lustre_msghdr_get_flags(req->rq_reqmsg) & MSGHDR_AT_SUPPORT))) {
 122                 /* Early reply */
 123                 DEBUG_REQ(D_ADAPTTO, req,
 124                           "Early reply received: mlen=%u offset=%d replen=%d "
 125                           "replied=%d unlinked=%d", ev->mlength, ev->offset,
 126                           req->rq_replen, req->rq_replied, ev->unlinked);
 127
 128                 req->rq_early_count++; /* number received, client side */
 129                 if (req->rq_replied) {
 130                         /* If we already got the real reply, then we need to
 131                          * check if lnet_finalize() unlinked the md.  In that
 132                          * case, there will be no further callback of type
 133                          * LNET_EVENT_UNLINK.
 134                          */
 135                         if (ev->unlinked)
 136                                 req->rq_must_unlink = 0;
 137                         else
 138                                 DEBUG_REQ(D_RPCTRACE, req, "unlinked in reply");
 139                         goto out_wake;
 140                 }
 141                 req->rq_early = 1;
 142                 req->rq_reply_off = ev->offset;
 143                 req->rq_nob_received = ev->mlength;
 144                 /* And we're still receiving */
 145                 req->rq_receiving_reply = 1;
 146         } else {
 147                 /* Real reply */
 148                 req->rq_replied = 1;
 149                 req->rq_reply_off = ev->offset;
 150                 req->rq_nob_received = ev->mlength;
 151                 /* LNetMDUnlink can't be called under the LNET_LOCK,
 152                    so we must unlink in ptlrpc_unregister_reply */
 153                 DEBUG_REQ(D_INFO, req,
 154                           "reply in flags=%x mlen=%u offset=%d replen=%d",
 155                           lustre_msg_get_flags(req->rq_reqmsg),
 156                           ev->mlength, ev->offset, req->rq_replen);
 157         }
 158
 159         req->rq_import->imp_last_reply_time = cfs_time_current_sec();
 160
 161 out_wake:
 162         /* NB don't unlock till after wakeup; req can disappear under us
 163          * since we don't have our own ref */
 164         ptlrpc_client_wake_req(req);
 165         spin_unlock(&req->rq_lock);
 166         EXIT;
 167 }
 168
 169 /*
 170  * Client's bulk has been written/read
 171  */
 172 void client_bulk_callback (lnet_event_t *ev)
 173 {
 174         struct ptlrpc_cb_id     *cbid = ev->md.user_ptr;
 175         struct ptlrpc_bulk_desc *desc = cbid->cbid_arg;
 176         ENTRY;
 177
 178         LASSERT ((desc->bd_type == BULK_PUT_SINK &&
 179                   ev->type == LNET_EVENT_PUT) ||
 180                  (desc->bd_type == BULK_GET_SOURCE &&
 181                   ev->type == LNET_EVENT_GET) ||
 182                  ev->type == LNET_EVENT_UNLINK);
 183         LASSERT (ev->unlinked);
 184
 185         CDEBUG((ev->status == 0) ? D_NET : D_ERROR,
 186                "event type %d, status %d, desc %p\n",
 187                ev->type, ev->status, desc);
 188
 189         spin_lock(&desc->bd_lock);
 190
 191         LASSERT(desc->bd_network_rw);
 192         desc->bd_network_rw = 0;
 193
 194         if (ev->type != LNET_EVENT_UNLINK && ev->status == 0) {
 195                 desc->bd_success = 1;
 196                 desc->bd_nob_transferred = ev->mlength;
 197                 desc->bd_sender = ev->sender;
 198         }
 199
 200         sptlrpc_enc_pool_put_pages(desc);
 201
 202         /* NB don't unlock till after wakeup; desc can disappear under us
 203          * otherwise */
 204         ptlrpc_client_wake_req(desc->bd_req);
 205
 206         spin_unlock(&desc->bd_lock);
 207         EXIT;
 208 }
 209
 210 /*
 211  * Server's incoming request callback
 212  */
 213 void request_in_callback(lnet_event_t *ev)
 214 {
 215         struct ptlrpc_cb_id               *cbid = ev->md.user_ptr;
 216         struct ptlrpc_request_buffer_desc *rqbd = cbid->cbid_arg;
 217         struct ptlrpc_service             *service = rqbd->rqbd_service;
 218         struct ptlrpc_request             *req;
 219         ENTRY;
 220
 221         LASSERT (ev->type == LNET_EVENT_PUT ||
 222                  ev->type == LNET_EVENT_UNLINK);
 223         LASSERT ((char *)ev->md.start >= rqbd->rqbd_buffer);
 224         LASSERT ((char *)ev->md.start + ev->offset + ev->mlength <=
 225                  rqbd->rqbd_buffer + service->srv_buf_size);
 226
 227         CDEBUG((ev->status == 0) ? D_NET : D_ERROR,
 228                "event type %d, status %d, service %s\n",
 229                ev->type, ev->status, service->srv_name);
 230
 231         if (ev->unlinked) {
 232                 /* If this is the last request message to fit in the
 233                  * request buffer we can use the request object embedded in
 234                  * rqbd.  Note that if we failed to allocate a request,
 235                  * we'd have to re-post the rqbd, which we can't do in this
 236                  * context. */
 237                 req = &rqbd->rqbd_req;
 238                 memset(req, 0, sizeof (*req));
 239         } else {
 240                 LASSERT (ev->type == LNET_EVENT_PUT);
 241                 if (ev->status != 0) {
 242                         /* We moaned above already... */
 243                         return;
 244                 }
 245                 OBD_ALLOC_GFP(req, sizeof(*req), CFS_ALLOC_ATOMIC_TRY);
 246                 if (req == NULL) {
 247                         CERROR("Can't allocate incoming request descriptor: "
 248                                "Dropping %s RPC from %s\n",
 249                                service->srv_name,
 250                                libcfs_id2str(ev->initiator));
 251                         return;
 252                 }
 253         }
 254
 255         /* NB we ABSOLUTELY RELY on req being zeroed, so pointers are NULL,
 256          * flags are reset and scalars are zero.  We only set the message
 257          * size to non-zero if this was a successful receive. */
 258         req->rq_xid = ev->match_bits;
 259         req->rq_reqbuf = ev->md.start + ev->offset;
 260         if (ev->type == LNET_EVENT_PUT && ev->status == 0)
 261                 req->rq_reqdata_len = ev->mlength;
 262         do_gettimeofday(&req->rq_arrival_time);
 263         req->rq_peer = ev->initiator;
 264         req->rq_self = ev->target.nid;
 265         req->rq_rqbd = rqbd;
 266         req->rq_phase = RQ_PHASE_NEW;
 267 #ifdef CRAY_XT3
 268         req->rq_uid = ev->uid;
 269 #endif
 270         spin_lock_init(&req->rq_lock);
 271         CFS_INIT_LIST_HEAD(&req->rq_timed_list);
 272         atomic_set(&req->rq_refcount, 1);
 273         if (ev->type == LNET_EVENT_PUT)
 274                 DEBUG_REQ(D_RPCTRACE, req, "incoming req");
 275
 276         CDEBUG(D_RPCTRACE, "peer: %s\n", libcfs_id2str(req->rq_peer));
 277
 278         spin_lock(&service->srv_lock);
 279
 280         req->rq_history_seq = service->srv_request_seq++;
 281         list_add_tail(&req->rq_history_list, &service->srv_request_history);
 282
 283         if (ev->unlinked) {
 284                 service->srv_nrqbd_receiving--;
 285                 CDEBUG(D_INFO, "Buffer complete: %d buffers still posted\n",
 286                        service->srv_nrqbd_receiving);
 287
 288                 /* Normally, don't complain about 0 buffers posted; LNET won't
 289                  * drop incoming reqs since we set the portal lazy */
 290                 if (test_req_buffer_pressure &&
 291                     ev->type != LNET_EVENT_UNLINK &&
 292                     service->srv_nrqbd_receiving == 0)
 293                         CWARN("All %s request buffers busy\n",
 294                               service->srv_name);
 295
 296                 /* req takes over the network's ref on rqbd */
 297         } else {
 298                 /* req takes a ref on rqbd */
 299                 rqbd->rqbd_refcount++;
 300         }
 301
 302         list_add_tail(&req->rq_list, &service->srv_req_in_queue);
 303         service->srv_n_queued_reqs++;
 304
 305         /* NB everything can disappear under us once the request
 306          * has been queued and we unlock, so do the wake now... */
 307         cfs_waitq_signal(&service->srv_waitq);
 308
 309         spin_unlock(&service->srv_lock);
 310         EXIT;
 311 }
 312
 313 /*
 314  *  Server's outgoing reply callback
 315  */
 316 void reply_out_callback(lnet_event_t *ev)
 317 {
 318         struct ptlrpc_cb_id       *cbid = ev->md.user_ptr;
 319         struct ptlrpc_reply_state *rs = cbid->cbid_arg;
 320         struct ptlrpc_service     *svc = rs->rs_service;
 321         ENTRY;
 322
 323         LASSERT (ev->type == LNET_EVENT_SEND ||
 324                  ev->type == LNET_EVENT_ACK ||
 325                  ev->type == LNET_EVENT_UNLINK);
 326
 327         if (!rs->rs_difficult) {
 328                 /* 'Easy' replies have no further processing so I drop the
 329                  * net's ref on 'rs' */
 330                 LASSERT (ev->unlinked);
 331                 ptlrpc_rs_decref(rs);
 332                 atomic_dec (&svc->srv_outstanding_replies);
 333                 EXIT;
 334                 return;
 335         }
 336
 337         LASSERT (rs->rs_on_net);
 338
 339         if (ev->unlinked) {
 340                 /* Last network callback.  The net's ref on 'rs' stays put
 341                  * until ptlrpc_server_handle_reply() is done with it */
 342                 spin_lock(&svc->srv_lock);
 343                 rs->rs_on_net = 0;
 344                 if (!rs->rs_no_ack ||
 345                     rs->rs_transno <= rs->rs_export->exp_obd->obd_last_committed)
 346                         ptlrpc_schedule_difficult_reply (rs);
 347                 spin_unlock(&svc->srv_lock);
 348         }
 349
 350         EXIT;
 351 }
 352
 353 /*
 354  * Server's bulk completion callback
 355  */
 356 void server_bulk_callback (lnet_event_t *ev)
 357 {
 358         struct ptlrpc_cb_id     *cbid = ev->md.user_ptr;
 359         struct ptlrpc_bulk_desc *desc = cbid->cbid_arg;
 360         ENTRY;
 361
 362         LASSERT (ev->type == LNET_EVENT_SEND ||
 363                  ev->type == LNET_EVENT_UNLINK ||
 364                  (desc->bd_type == BULK_PUT_SOURCE &&
 365                   ev->type == LNET_EVENT_ACK) ||
 366                  (desc->bd_type == BULK_GET_SINK &&
 367                   ev->type == LNET_EVENT_REPLY));
 368
 369         CDEBUG((ev->status == 0) ? D_NET : D_ERROR,
 370                "event type %d, status %d, desc %p\n",
 371                ev->type, ev->status, desc);
 372
 373         spin_lock(&desc->bd_lock);
 374
 375         if ((ev->type == LNET_EVENT_ACK ||
 376              ev->type == LNET_EVENT_REPLY) &&
 377             ev->status == 0) {
 378                 /* We heard back from the peer, so even if we get this
 379                  * before the SENT event (oh yes we can), we know we
 380                  * read/wrote the peer buffer and how much... */
 381                 desc->bd_success = 1;
 382                 desc->bd_nob_transferred = ev->mlength;
 383                 desc->bd_sender = ev->sender;
 384         }
 385
 386         if (ev->unlinked) {
 387                 /* This is the last callback no matter what... */
 388                 desc->bd_network_rw = 0;
 389                 cfs_waitq_signal(&desc->bd_waitq);
 390         }
 391
 392         spin_unlock(&desc->bd_lock);
 393         EXIT;
 394 }
 395
 396 static void ptlrpc_master_callback(lnet_event_t *ev)
 397 {
 398         struct ptlrpc_cb_id *cbid = ev->md.user_ptr;
 399         void (*callback)(lnet_event_t *ev) = cbid->cbid_fn;
 400
 401         /* Honestly, it's best to find out early. */
 402         LASSERT (cbid->cbid_arg != LP_POISON);
 403         LASSERT (callback == request_out_callback ||
 404                  callback == reply_in_callback ||
 405                  callback == client_bulk_callback ||
 406                  callback == request_in_callback ||
 407                  callback == reply_out_callback ||
 408                  callback == server_bulk_callback);
 409
 410         callback (ev);
 411 }
 412
 413 int ptlrpc_uuid_to_peer (struct obd_uuid *uuid,
 414                          lnet_process_id_t *peer, lnet_nid_t *self)
 415 {
 416         int               best_dist = 0;
 417         __u32             best_order = 0;
 418         int               count = 0;
 419         int               rc = -ENOENT;
 420         int               portals_compatibility;
 421         int               dist;
 422         __u32             order;
 423         lnet_nid_t        dst_nid;
 424         lnet_nid_t        src_nid;
 425
 426         portals_compatibility = LNetCtl(IOC_LIBCFS_PORTALS_COMPATIBILITY, NULL);
 427
 428         peer->pid = LUSTRE_SRV_LNET_PID;
 429
 430         /* Choose the matching UUID that's closest */
 431         while (lustre_uuid_to_peer(uuid->uuid, &dst_nid, count++) == 0) {
 432                 dist = LNetDist(dst_nid, &src_nid, &order);
 433                 if (dist < 0)
 434                         continue;
 435
 436                 if (dist == 0) {                /* local! use loopback LND */
 437                         peer->nid = *self = LNET_MKNID(LNET_MKNET(LOLND, 0), 0);
 438                         rc = 0;
 439                         break;
 440                 }
 441
 442                 if (rc < 0 ||
 443                     dist < best_dist ||
 444                     (dist == best_dist && order < best_order)) {
 445                         best_dist = dist;
 446                         best_order = order;
 447
 448                         if (portals_compatibility > 1) {
 449                                 /* Strong portals compatibility: Zero the nid's
 450                                  * NET, so if I'm reading new config logs, or
 451                                  * getting configured by (new) lconf I can
 452                                  * still talk to old servers. */
 453                                 dst_nid = LNET_MKNID(0, LNET_NIDADDR(dst_nid));
 454                                 src_nid = LNET_MKNID(0, LNET_NIDADDR(src_nid));
 455                         }
 456                         peer->nid = dst_nid;
 457                         *self = src_nid;
 458                         rc = 0;
 459                 }
 460         }
 461
 462         CDEBUG(D_NET,"%s->%s\n", uuid->uuid, libcfs_id2str(*peer));
 463         if (rc != 0)
 464                 CERROR("No NID found for %s\n", uuid->uuid);
 465         return rc;
 466 }
 467
 468 void ptlrpc_ni_fini(void)
 469 {
 470         cfs_waitq_t         waitq;
 471         struct l_wait_info  lwi;
 472         int                 rc;
 473         int                 retries;
 474
 475         /* Wait for the event queue to become idle since there may still be
 476          * messages in flight with pending events (i.e. the fire-and-forget
 477          * messages == client requests and "non-difficult" server
 478          * replies */
 479
 480         for (retries = 0;; retries++) {
 481                 rc = LNetEQFree(ptlrpc_eq_h);
 482                 switch (rc) {
 483                 default:
 484                         LBUG();
 485
 486                 case 0:
 487                         LNetNIFini();
 488                         return;
 489
 490                 case -EBUSY:
 491                         if (retries != 0)
 492                                 CWARN("Event queue still busy\n");
 493
 494                         /* Wait for a bit */
 495                         cfs_waitq_init(&waitq);
 496                         lwi = LWI_TIMEOUT(cfs_time_seconds(2), NULL, NULL);
 497                         l_wait_event(waitq, 0, &lwi);
 498                         break;
 499                 }
 500         }
 501         /* notreached */
 502 }
 503
 504 lnet_pid_t ptl_get_pid(void)
 505 {
 506         lnet_pid_t        pid;
 507
 508 #ifndef  __KERNEL__
 509         pid = getpid();
 510 #else
 511         pid = LUSTRE_SRV_LNET_PID;
 512 #endif
 513         return pid;
 514 }
 515
 516 int ptlrpc_ni_init(void)
 517 {
 518         int              rc;
 519         lnet_pid_t       pid;
 520
 521         pid = ptl_get_pid();
 522         CDEBUG(D_NET, "My pid is: %x\n", pid);
 523
 524         /* We're not passing any limits yet... */
 525         rc = LNetNIInit(pid);
 526         if (rc < 0) {
 527                 CDEBUG (D_NET, "Can't init network interface: %d\n", rc);
 528                 return (-ENOENT);
 529         }
 530
 531         /* CAVEAT EMPTOR: how we process portals events is _radically_
 532          * different depending on... */
 533 #ifdef __KERNEL__
 534         /* kernel portals calls our master callback when events are added to
 535          * the event queue.  In fact lustre never pulls events off this queue,
 536          * so it's only sized for some debug history. */
 537         rc = LNetEQAlloc(1024, ptlrpc_master_callback, &ptlrpc_eq_h);
 538 #else
 539         /* liblustre calls the master callback when it removes events from the
 540          * event queue.  The event queue has to be big enough not to drop
 541          * anything */
 542         rc = LNetEQAlloc(10240, LNET_EQ_HANDLER_NONE, &ptlrpc_eq_h);
 543 #endif
 544         if (rc == 0)
 545                 return 0;
 546
 547         CERROR ("Failed to allocate event queue: %d\n", rc);
 548         LNetNIFini();
 549
 550         return (-ENOMEM);
 551 }
 552
 553 #ifndef __KERNEL__
 554 CFS_LIST_HEAD(liblustre_wait_callbacks);
 555 CFS_LIST_HEAD(liblustre_idle_callbacks);
 556 void *liblustre_services_callback;
 557
 558 void *
 559 liblustre_register_waitidle_callback (struct list_head *callback_list,
 560                                       const char *name,
 561                                       int (*fn)(void *arg), void *arg)
 562 {
 563         struct liblustre_wait_callback *llwc;
 564
 565         OBD_ALLOC(llwc, sizeof(*llwc));
 566         LASSERT (llwc != NULL);
 567
 568         llwc->llwc_name = name;
 569         llwc->llwc_fn = fn;
 570         llwc->llwc_arg = arg;
 571         list_add_tail(&llwc->llwc_list, callback_list);
 572
 573         return (llwc);
 574 }
 575
 576 void
 577 liblustre_deregister_waitidle_callback (void *opaque)
 578 {
 579         struct liblustre_wait_callback *llwc = opaque;
 580
 581         list_del(&llwc->llwc_list);
 582         OBD_FREE(llwc, sizeof(*llwc));
 583 }
 584
 585 void *
 586 liblustre_register_wait_callback (const char *name,
 587                                   int (*fn)(void *arg), void *arg)
 588 {
 589         return liblustre_register_waitidle_callback(&liblustre_wait_callbacks,
 590                                                     name, fn, arg);
 591 }
 592
 593 void
 594 liblustre_deregister_wait_callback (void *opaque)
 595 {
 596         liblustre_deregister_waitidle_callback(opaque);
 597 }
 598
 599 void *
 600 liblustre_register_idle_callback (const char *name,
 601                                   int (*fn)(void *arg), void *arg)
 602 {
 603         return liblustre_register_waitidle_callback(&liblustre_idle_callbacks,
 604                                                     name, fn, arg);
 605 }
 606
 607 void
 608 liblustre_deregister_idle_callback (void *opaque)
 609 {
 610         liblustre_deregister_waitidle_callback(opaque);
 611 }
 612
 613 int
 614 liblustre_check_events (int timeout)
 615 {
 616         lnet_event_t ev;
 617         int         rc;
 618         int         i;
 619         ENTRY;
 620
 621         rc = LNetEQPoll(&ptlrpc_eq_h, 1, timeout * 1000, &ev, &i);
 622         if (rc == 0)
 623                 RETURN(0);
 624
 625         LASSERT (rc == -EOVERFLOW || rc == 1);
 626
 627         /* liblustre: no asynch callback so we can't affort to miss any
 628          * events... */
 629         if (rc == -EOVERFLOW) {
 630                 CERROR ("Dropped an event!!!\n");
 631                 abort();
 632         }
 633
 634         ptlrpc_master_callback (&ev);
 635         RETURN(1);
 636 }
 637
 638 int liblustre_waiting = 0;
 639
 640 int
 641 liblustre_wait_event (int timeout)
 642 {
 643         struct list_head               *tmp;
 644         struct liblustre_wait_callback *llwc;
 645         int                             found_something = 0;
 646
 647         /* single threaded recursion check... */
 648         liblustre_waiting = 1;
 649
 650         for (;;) {
 651                 /* Deal with all pending events */
 652                 while (liblustre_check_events(0))
 653                         found_something = 1;
 654
 655                 /* Give all registered callbacks a bite at the cherry */
 656                 list_for_each(tmp, &liblustre_wait_callbacks) {
 657                         llwc = list_entry(tmp, struct liblustre_wait_callback,
 658                                           llwc_list);
 659
 660                         if (llwc->llwc_fn(llwc->llwc_arg))
 661                                 found_something = 1;
 662                 }
 663
 664                 if (found_something || timeout == 0)
 665                         break;
 666
 667                 /* Nothing so far, but I'm allowed to block... */
 668                 found_something = liblustre_check_events(timeout);
 669                 if (!found_something)           /* still nothing */
 670                         break;                  /* I timed out */
 671         }
 672
 673         liblustre_waiting = 0;
 674
 675         return found_something;
 676 }
 677
 678 void
 679 liblustre_wait_idle(void)
 680 {
 681         static int recursed = 0;
 682
 683         struct list_head               *tmp;
 684         struct liblustre_wait_callback *llwc;
 685         int                             idle = 0;
 686
 687         LASSERT(!recursed);
 688         recursed = 1;
 689
 690         do {
 691                 liblustre_wait_event(0);
 692
 693                 idle = 1;
 694
 695                 list_for_each(tmp, &liblustre_idle_callbacks) {
 696                         llwc = list_entry(tmp, struct liblustre_wait_callback,
 697                                           llwc_list);
 698
 699                         if (!llwc->llwc_fn(llwc->llwc_arg)) {
 700                                 idle = 0;
 701                                 break;
 702                         }
 703                 }
 704
 705         } while (!idle);
 706
 707         recursed = 0;
 708 }
 709
 710 #endif /* __KERNEL__ */
 711
 712 int ptlrpc_init_portals(void)
 713 {
 714         int   rc = ptlrpc_ni_init();
 715
 716         if (rc != 0) {
 717                 CERROR("network initialisation failed\n");
 718                 return -EIO;
 719         }
 720 #ifndef __KERNEL__
 721         liblustre_services_callback =
 722                 liblustre_register_wait_callback("liblustre_check_services",
 723                                                  &liblustre_check_services,
 724                                                  NULL);
 725 #endif
 726         rc = ptlrpcd_addref();
 727         if (rc == 0)
 728                 return 0;
 729
 730         CERROR("rpcd initialisation failed\n");
 731 #ifndef __KERNEL__
 732         liblustre_deregister_wait_callback(liblustre_services_callback);
 733 #endif
 734         ptlrpc_ni_fini();
 735         return rc;
 736 }
 737
 738 void ptlrpc_exit_portals(void)
 739 {
 740 #ifndef __KERNEL__
 741         liblustre_deregister_wait_callback(liblustre_services_callback);
 742 #endif
 743         ptlrpcd_decref();
 744         ptlrpc_ni_fini();
 745 }