lnet/lnet/lib-eq.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  */
  30 /*
  31  * This file is part of Lustre, http://www.lustre.org/
  32  * Lustre is a trademark of Sun Microsystems, Inc.
  33  *
  34  * lnet/lnet/lib-eq.c
  35  *
  36  * Library level Event queue management routines
  37  */
  38
  39 #define DEBUG_SUBSYSTEM S_LNET
  40 #include <lnet/lib-lnet.h>
  41
  42 /**
  43  * Create an event queue that has room for \a count number of events.
  44  *
  45  * The event queue is circular and older events will be overwritten by new
  46  * ones if they are not removed in time by the user using the functions
  47  * LNetEQGet(), LNetEQWait(), or LNetEQPoll(). It is up to the user to
  48  * determine the appropriate size of the event queue to prevent this loss
  49  * of events. Note that when EQ handler is specified in \a callback, no
  50  * event loss can happen, since the handler is run for each event deposited
  51  * into the EQ.
  52  *
  53  * \param count The number of events to be stored in the event queue. It
  54  * will be rounded up to the next power of two.
  55  * \param callback A handler function that runs when an event is deposited
  56  * into the EQ. The constant value LNET_EQ_HANDLER_NONE can be used to
  57  * indicate that no event handler is desired.
  58  * \param handle On successful return, this location will hold a handle for
  59  * the newly created EQ.
  60  *
  61  * \retval 0       On success.
  62  * \retval -EINVAL If an parameter is not valid.
  63  * \retval -ENOMEM If memory for the EQ can't be allocated.
  64  *
  65  * \see lnet_eq_handler_t for the discussion on EQ handler semantics.
  66  */
  67 int
  68 LNetEQAlloc(unsigned int count, lnet_eq_handler_t callback,
  69             lnet_handle_eq_t *handle)
  70 {
  71         lnet_eq_t     *eq;
  72
  73         LASSERT (the_lnet.ln_init);
  74         LASSERT (the_lnet.ln_refcount > 0);
  75
  76         /* We need count to be a power of 2 so that when eq_{enq,deq}_seq
  77          * overflow, they don't skip entries, so the queue has the same
  78          * apparent capacity at all times */
  79
  80         if (count != LOWEST_BIT_SET(count)) {   /* not a power of 2 already */
  81                 do {                    /* knock off all but the top bit... */
  82                         count &= ~LOWEST_BIT_SET (count);
  83                 } while (count != LOWEST_BIT_SET(count));
  84
  85                 count <<= 1;                             /* ...and round up */
  86         }
  87
  88         if (count == 0)        /* catch bad parameter / overflow on roundup */
  89                 return (-EINVAL);
  90
  91         eq = lnet_eq_alloc();
  92         if (eq == NULL)
  93                 return (-ENOMEM);
  94
  95         LIBCFS_ALLOC(eq->eq_events, count * sizeof(lnet_event_t));
  96         if (eq->eq_events == NULL) {
  97                 lnet_eq_free(eq);
  98
  99                 return -ENOMEM;
 100         }
 101
 102         /* NB this resets all event sequence numbers to 0, to be earlier
 103          * than eq_deq_seq */
 104         memset(eq->eq_events, 0, count * sizeof(lnet_event_t));
 105
 106         eq->eq_deq_seq = 1;
 107         eq->eq_enq_seq = 1;
 108         eq->eq_size = count;
 109         eq->eq_refcount = 0;
 110         eq->eq_callback = callback;
 111
 112         LNET_LOCK();
 113
 114         lnet_initialise_handle (&eq->eq_lh, LNET_COOKIE_TYPE_EQ);
 115         cfs_list_add (&eq->eq_list, &the_lnet.ln_active_eqs);
 116
 117         LNET_UNLOCK();
 118
 119         lnet_eq2handle(handle, eq);
 120         return (0);
 121 }
 122
 123 /**
 124  * Release the resources associated with an event queue if it's idle;
 125  * otherwise do nothing and it's up to the user to try again.
 126  *
 127  * \param eqh A handle for the event queue to be released.
 128  *
 129  * \retval 0 If the EQ is not in use and freed.
 130  * \retval -ENOENT If \a eqh does not point to a valid EQ.
 131  * \retval -EBUSY  If the EQ is still in use by some MDs.
 132  */
 133 int
 134 LNetEQFree(lnet_handle_eq_t eqh)
 135 {
 136         lnet_eq_t     *eq;
 137         int            size;
 138         lnet_event_t  *events;
 139
 140         LASSERT (the_lnet.ln_init);
 141         LASSERT (the_lnet.ln_refcount > 0);
 142
 143         LNET_LOCK();
 144
 145         eq = lnet_handle2eq(&eqh);
 146         if (eq == NULL) {
 147                 LNET_UNLOCK();
 148                 return (-ENOENT);
 149         }
 150
 151         if (eq->eq_refcount != 0) {
 152                 CDEBUG(D_NET, "Event queue (%d) busy on destroy.\n",
 153                        eq->eq_refcount);
 154                 LNET_UNLOCK();
 155                 return (-EBUSY);
 156         }
 157
 158         /* stash for free after lock dropped */
 159         events  = eq->eq_events;
 160         size    = eq->eq_size;
 161
 162         lnet_invalidate_handle (&eq->eq_lh);
 163         cfs_list_del (&eq->eq_list);
 164         lnet_eq_free_locked(eq);
 165
 166         LNET_UNLOCK();
 167
 168         LIBCFS_FREE(events, size * sizeof (lnet_event_t));
 169
 170         return 0;
 171 }
 172
 173 int
 174 lib_get_event (lnet_eq_t *eq, lnet_event_t *ev)
 175 {
 176         int           new_index = eq->eq_deq_seq & (eq->eq_size - 1);
 177         lnet_event_t *new_event = &eq->eq_events[new_index];
 178         int           rc;
 179         ENTRY;
 180
 181         CDEBUG(D_INFO, "event: %p, sequence: %lu, eq->size: %u\n",
 182                new_event, eq->eq_deq_seq, eq->eq_size);
 183
 184         if (LNET_SEQ_GT (eq->eq_deq_seq, new_event->sequence)) {
 185                 RETURN(0);
 186         }
 187
 188         /* We've got a new event... */
 189         *ev = *new_event;
 190
 191         /* ...but did it overwrite an event we've not seen yet? */
 192         if (eq->eq_deq_seq == new_event->sequence) {
 193                 rc = 1;
 194         } else {
 195                 /* don't complain with CERROR: some EQs are sized small
 196                  * anyway; if it's important, the caller should complain */
 197                 CDEBUG(D_NET, "Event Queue Overflow: eq seq %lu ev seq %lu\n",
 198                        eq->eq_deq_seq, new_event->sequence);
 199                 rc = -EOVERFLOW;
 200         }
 201
 202         eq->eq_deq_seq = new_event->sequence + 1;
 203         RETURN(rc);
 204 }
 205
 206 /**
 207  * A nonblocking function that can be used to get the next event in an EQ.
 208  * If an event handler is associated with the EQ, the handler will run before
 209  * this function returns successfully. The event is removed from the queue.
 210  *
 211  * \param eventq A handle for the event queue.
 212  * \param event On successful return (1 or -EOVERFLOW), this location will
 213  * hold the next event in the EQ.
 214  *
 215  * \retval 0          No pending event in the EQ.
 216  * \retval 1          Indicates success.
 217  * \retval -ENOENT    If \a eventq does not point to a valid EQ.
 218  * \retval -EOVERFLOW Indicates success (i.e., an event is returned) and that
 219  * at least one event between this event and the last event obtained from the
 220  * EQ has been dropped due to limited space in the EQ.
 221  */
 222 int
 223 LNetEQGet (lnet_handle_eq_t eventq, lnet_event_t *event)
 224 {
 225         int which;
 226
 227         return LNetEQPoll(&eventq, 1, 0,
 228                          event, &which);
 229 }
 230
 231 /**
 232  * Block the calling process until there is an event in the EQ.
 233  * If an event handler is associated with the EQ, the handler will run before
 234  * this function returns successfully. This function returns the next event
 235  * in the EQ and removes it from the EQ.
 236  *
 237  * \param eventq A handle for the event queue.
 238  * \param event On successful return (1 or -EOVERFLOW), this location will
 239  * hold the next event in the EQ.
 240  *
 241  * \retval 1          Indicates success.
 242  * \retval -ENOENT    If \a eventq does not point to a valid EQ.
 243  * \retval -EOVERFLOW Indicates success (i.e., an event is returned) and that
 244  * at least one event between this event and the last event obtained from the
 245  * EQ has been dropped due to limited space in the EQ.
 246  */
 247 int
 248 LNetEQWait (lnet_handle_eq_t eventq, lnet_event_t *event)
 249 {
 250         int which;
 251
 252         return LNetEQPoll(&eventq, 1, LNET_TIME_FOREVER,
 253                          event, &which);
 254 }
 255
 256 /**
 257  * Block the calling process until there's an event from a set of EQs or
 258  * timeout happens.
 259  *
 260  * If an event handler is associated with the EQ, the handler will run before
 261  * this function returns successfully, in which case the corresponding event
 262  * is consumed.
 263  *
 264  * LNetEQPoll() provides a timeout to allow applications to poll, block for a
 265  * fixed period, or block indefinitely.
 266  *
 267  * \param eventqs,neq An array of EQ handles, and size of the array.
 268  * \param timeout_ms Time in milliseconds to wait for an event to occur on
 269  * one of the EQs. The constant LNET_TIME_FOREVER can be used to indicate an
 270  * infinite timeout.
 271  * \param event,which On successful return (1 or -EOVERFLOW), \a event will
 272  * hold the next event in the EQs, and \a which will contain the index of the
 273  * EQ from which the event was taken.
 274  *
 275  * \retval 0          No pending event in the EQs after timeout.
 276  * \retval 1          Indicates success.
 277  * \retval -EOVERFLOW Indicates success (i.e., an event is returned) and that
 278  * at least one event between this event and the last event obtained from the
 279  * EQ indicated by \a which has been dropped due to limited space in the EQ.
 280  * \retval -ENOENT    If there's an invalid handle in \a eventqs.
 281  */
 282 int
 283 LNetEQPoll (lnet_handle_eq_t *eventqs, int neq, int timeout_ms,
 284             lnet_event_t *event, int *which)
 285 {
 286         int              i;
 287         int              rc;
 288 #ifdef __KERNEL__
 289         cfs_waitlink_t   wl;
 290         cfs_time_t       now;
 291 #else
 292         struct timeval   then;
 293         struct timeval   now;
 294 # ifdef HAVE_LIBPTHREAD
 295         struct timespec  ts;
 296 # endif
 297         lnet_ni_t       *eqwaitni = the_lnet.ln_eqwaitni;
 298 #endif
 299         ENTRY;
 300
 301         LASSERT (the_lnet.ln_init);
 302         LASSERT (the_lnet.ln_refcount > 0);
 303
 304         if (neq < 1)
 305                 RETURN(-ENOENT);
 306
 307         LNET_LOCK();
 308
 309         for (;;) {
 310 #ifndef __KERNEL__
 311                 LNET_UNLOCK();
 312
 313                 /* Recursion breaker */
 314                 if (the_lnet.ln_rc_state == LNET_RC_STATE_RUNNING &&
 315                     !LNetHandleIsEqual(eventqs[0], the_lnet.ln_rc_eqh))
 316                         lnet_router_checker();
 317
 318                 LNET_LOCK();
 319 #endif
 320                 for (i = 0; i < neq; i++) {
 321                         lnet_eq_t *eq = lnet_handle2eq(&eventqs[i]);
 322
 323                         if (eq == NULL) {
 324                                 LNET_UNLOCK();
 325                                 RETURN(-ENOENT);
 326                         }
 327
 328                         rc = lib_get_event (eq, event);
 329                         if (rc != 0) {
 330                                 LNET_UNLOCK();
 331                                 *which = i;
 332                                 RETURN(rc);
 333                         }
 334                 }
 335
 336 #ifdef __KERNEL__
 337                 if (timeout_ms == 0) {
 338                         LNET_UNLOCK();
 339                         RETURN (0);
 340                 }
 341
 342                 cfs_waitlink_init(&wl);
 343                 cfs_set_current_state(CFS_TASK_INTERRUPTIBLE);
 344                 cfs_waitq_add(&the_lnet.ln_waitq, &wl);
 345
 346                 LNET_UNLOCK();
 347
 348                 if (timeout_ms < 0) {
 349                         cfs_waitq_wait (&wl, CFS_TASK_INTERRUPTIBLE);
 350                 } else {
 351                         struct timeval tv;
 352
 353                         now = cfs_time_current();
 354                         cfs_waitq_timedwait(&wl, CFS_TASK_INTERRUPTIBLE,
 355                                             cfs_time_seconds(timeout_ms)/1000);
 356                         cfs_duration_usec(cfs_time_sub(cfs_time_current(), now),
 357                                             &tv);
 358                         timeout_ms -= (int)(tv.tv_sec * 1000 + tv.tv_usec / 1000);
 359                         if (timeout_ms < 0)
 360                                 timeout_ms = 0;
 361                 }
 362
 363                 LNET_LOCK();
 364                 cfs_waitq_del(&the_lnet.ln_waitq, &wl);
 365 #else
 366                 if (eqwaitni != NULL) {
 367                         /* I have a single NI that I have to call into, to get
 368                          * events queued, or to block. */
 369                         lnet_ni_addref_locked(eqwaitni);
 370                         LNET_UNLOCK();
 371
 372                         if (timeout_ms <= 0) {
 373                                 (eqwaitni->ni_lnd->lnd_wait)(eqwaitni, timeout_ms);
 374                         } else {
 375                                 gettimeofday(&then, NULL);
 376
 377                                 (eqwaitni->ni_lnd->lnd_wait)(eqwaitni, timeout_ms);
 378
 379                                 gettimeofday(&now, NULL);
 380                                 timeout_ms -= (now.tv_sec - then.tv_sec) * 1000 +
 381                                               (now.tv_usec - then.tv_usec) / 1000;
 382                                 if (timeout_ms < 0)
 383                                         timeout_ms = 0;
 384                         }
 385
 386                         LNET_LOCK();
 387                         lnet_ni_decref_locked(eqwaitni);
 388
 389                         /* don't call into eqwaitni again if timeout has
 390                          * expired */
 391                         if (timeout_ms == 0)
 392                                 eqwaitni = NULL;
 393
 394                         continue;               /* go back and check for events */
 395                 }
 396
 397                 if (timeout_ms == 0) {
 398                         LNET_UNLOCK();
 399                         RETURN (0);
 400                 }
 401
 402 # ifndef HAVE_LIBPTHREAD
 403                 /* If I'm single-threaded, LNET fails at startup if it can't
 404                  * set the_lnet.ln_eqwaitni correctly.  */
 405                 LBUG();
 406 # else
 407                 if (timeout_ms < 0) {
 408                         pthread_cond_wait(&the_lnet.ln_cond,
 409                                           &the_lnet.ln_lock);
 410                 } else {
 411                         gettimeofday(&then, NULL);
 412
 413                         ts.tv_sec = then.tv_sec + timeout_ms/1000;
 414                         ts.tv_nsec = then.tv_usec * 1000 +
 415                                      (timeout_ms%1000) * 1000000;
 416                         if (ts.tv_nsec >= 1000000000) {
 417                                 ts.tv_sec++;
 418                                 ts.tv_nsec -= 1000000000;
 419                         }
 420
 421                         pthread_cond_timedwait(&the_lnet.ln_cond,
 422                                                &the_lnet.ln_lock, &ts);
 423
 424                         gettimeofday(&now, NULL);
 425                         timeout_ms -= (now.tv_sec - then.tv_sec) * 1000 +
 426                                       (now.tv_usec - then.tv_usec) / 1000;
 427
 428                         if (timeout_ms < 0)
 429                                 timeout_ms = 0;
 430                 }
 431 # endif
 432 #endif
 433         }
 434 }