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_LOCK();
  98                 lnet_eq_free (eq);
  99                 LNET_UNLOCK();
 100
 101                 return -ENOMEM;
 102         }
 103
 104         /* NB this resets all event sequence numbers to 0, to be earlier
 105          * than eq_deq_seq */
 106         memset(eq->eq_events, 0, count * sizeof(lnet_event_t));
 107
 108         eq->eq_deq_seq = 1;
 109         eq->eq_enq_seq = 1;
 110         eq->eq_size = count;
 111         eq->eq_refcount = 0;
 112         eq->eq_callback = callback;
 113
 114         LNET_LOCK();
 115
 116         lnet_initialise_handle (&eq->eq_lh, LNET_COOKIE_TYPE_EQ);
 117         cfs_list_add (&eq->eq_list, &the_lnet.ln_active_eqs);
 118
 119         LNET_UNLOCK();
 120
 121         lnet_eq2handle(handle, eq);
 122         return (0);
 123 }
 124
 125 /**
 126  * Release the resources associated with an event queue if it's idle;
 127  * otherwise do nothing and it's up to the user to try again.
 128  *
 129  * \param eqh A handle for the event queue to be released.
 130  *
 131  * \retval 0 If the EQ is not in use and freed.
 132  * \retval -ENOENT If \a eqh does not point to a valid EQ.
 133  * \retval -EBUSY  If the EQ is still in use by some MDs.
 134  */
 135 int
 136 LNetEQFree(lnet_handle_eq_t eqh)
 137 {
 138         lnet_eq_t     *eq;
 139         int            size;
 140         lnet_event_t  *events;
 141
 142         LASSERT (the_lnet.ln_init);
 143         LASSERT (the_lnet.ln_refcount > 0);
 144
 145         LNET_LOCK();
 146
 147         eq = lnet_handle2eq(&eqh);
 148         if (eq == NULL) {
 149                 LNET_UNLOCK();
 150                 return (-ENOENT);
 151         }
 152
 153         if (eq->eq_refcount != 0) {
 154                 CDEBUG(D_NET, "Event queue (%d) busy on destroy.\n",
 155                        eq->eq_refcount);
 156                 LNET_UNLOCK();
 157                 return (-EBUSY);
 158         }
 159
 160         /* stash for free after lock dropped */
 161         events  = eq->eq_events;
 162         size    = eq->eq_size;
 163
 164         lnet_invalidate_handle (&eq->eq_lh);
 165         cfs_list_del (&eq->eq_list);
 166         lnet_eq_free (eq);
 167
 168         LNET_UNLOCK();
 169
 170         LIBCFS_FREE(events, size * sizeof (lnet_event_t));
 171
 172         return 0;
 173 }
 174
 175 int
 176 lib_get_event (lnet_eq_t *eq, lnet_event_t *ev)
 177 {
 178         int           new_index = eq->eq_deq_seq & (eq->eq_size - 1);
 179         lnet_event_t *new_event = &eq->eq_events[new_index];
 180         int           rc;
 181         ENTRY;
 182
 183         CDEBUG(D_INFO, "event: %p, sequence: %lu, eq->size: %u\n",
 184                new_event, eq->eq_deq_seq, eq->eq_size);
 185
 186         if (LNET_SEQ_GT (eq->eq_deq_seq, new_event->sequence)) {
 187                 RETURN(0);
 188         }
 189
 190         /* We've got a new event... */
 191         *ev = *new_event;
 192
 193         /* ...but did it overwrite an event we've not seen yet? */
 194         if (eq->eq_deq_seq == new_event->sequence) {
 195                 rc = 1;
 196         } else {
 197                 /* don't complain with CERROR: some EQs are sized small
 198                  * anyway; if it's important, the caller should complain */
 199                 CDEBUG(D_NET, "Event Queue Overflow: eq seq %lu ev seq %lu\n",
 200                        eq->eq_deq_seq, new_event->sequence);
 201                 rc = -EOVERFLOW;
 202         }
 203
 204         eq->eq_deq_seq = new_event->sequence + 1;
 205         RETURN(rc);
 206 }
 207
 208 /**
 209  * A nonblocking function that can be used to get the next event in an EQ.
 210  * If an event handler is associated with the EQ, the handler will run before
 211  * this function returns successfully. The event is removed from the queue.
 212  *
 213  * \param eventq A handle for the event queue.
 214  * \param event On successful return (1 or -EOVERFLOW), this location will
 215  * hold the next event in the EQ.
 216  *
 217  * \retval 0          No pending event in the EQ.
 218  * \retval 1          Indicates success.
 219  * \retval -ENOENT    If \a eventq does not point to a valid EQ.
 220  * \retval -EOVERFLOW Indicates success (i.e., an event is returned) and that
 221  * at least one event between this event and the last event obtained from the
 222  * EQ has been dropped due to limited space in the EQ.
 223  */
 224 int
 225 LNetEQGet (lnet_handle_eq_t eventq, lnet_event_t *event)
 226 {
 227         int which;
 228
 229         return LNetEQPoll(&eventq, 1, 0,
 230                          event, &which);
 231 }
 232
 233 /**
 234  * Block the calling process until there is an event in the EQ.
 235  * If an event handler is associated with the EQ, the handler will run before
 236  * this function returns successfully. This function returns the next event
 237  * in the EQ and removes it from the EQ.
 238  *
 239  * \param eventq A handle for the event queue.
 240  * \param event On successful return (1 or -EOVERFLOW), this location will
 241  * hold the next event in the EQ.
 242  *
 243  * \retval 1          Indicates success.
 244  * \retval -ENOENT    If \a eventq does not point to a valid EQ.
 245  * \retval -EOVERFLOW Indicates success (i.e., an event is returned) and that
 246  * at least one event between this event and the last event obtained from the
 247  * EQ has been dropped due to limited space in the EQ.
 248  */
 249 int
 250 LNetEQWait (lnet_handle_eq_t eventq, lnet_event_t *event)
 251 {
 252         int which;
 253
 254         return LNetEQPoll(&eventq, 1, LNET_TIME_FOREVER,
 255                          event, &which);
 256 }
 257
 258 /**
 259  * Block the calling process until there's an event from a set of EQs or
 260  * timeout happens.
 261  *
 262  * If an event handler is associated with the EQ, the handler will run before
 263  * this function returns successfully, in which case the corresponding event
 264  * is consumed.
 265  *
 266  * LNetEQPoll() provides a timeout to allow applications to poll, block for a
 267  * fixed period, or block indefinitely.
 268  *
 269  * \param eventqs,neq An array of EQ handles, and size of the array.
 270  * \param timeout_ms Time in milliseconds to wait for an event to occur on
 271  * one of the EQs. The constant LNET_TIME_FOREVER can be used to indicate an
 272  * infinite timeout.
 273  * \param event,which On successful return (1 or -EOVERFLOW), \a event will
 274  * hold the next event in the EQs, and \a which will contain the index of the
 275  * EQ from which the event was taken.
 276  *
 277  * \retval 0          No pending event in the EQs after timeout.
 278  * \retval 1          Indicates success.
 279  * \retval -EOVERFLOW Indicates success (i.e., an event is returned) and that
 280  * at least one event between this event and the last event obtained from the
 281  * EQ indicated by \a which has been dropped due to limited space in the EQ.
 282  * \retval -ENOENT    If there's an invalid handle in \a eventqs.
 283  */
 284 int
 285 LNetEQPoll (lnet_handle_eq_t *eventqs, int neq, int timeout_ms,
 286             lnet_event_t *event, int *which)
 287 {
 288         int              i;
 289         int              rc;
 290 #ifdef __KERNEL__
 291         cfs_waitlink_t   wl;
 292         cfs_time_t       now;
 293 #else
 294         struct timeval   then;
 295         struct timeval   now;
 296 # ifdef HAVE_LIBPTHREAD
 297         struct timespec  ts;
 298 # endif
 299         lnet_ni_t       *eqwaitni = the_lnet.ln_eqwaitni;
 300 #endif
 301         ENTRY;
 302
 303         LASSERT (the_lnet.ln_init);
 304         LASSERT (the_lnet.ln_refcount > 0);
 305
 306         if (neq < 1)
 307                 RETURN(-ENOENT);
 308
 309         LNET_LOCK();
 310
 311         for (;;) {
 312 #ifndef __KERNEL__
 313                 LNET_UNLOCK();
 314
 315                 /* Recursion breaker */
 316                 if (the_lnet.ln_rc_state == LNET_RC_STATE_RUNNING &&
 317                     !LNetHandleIsEqual(eventqs[0], the_lnet.ln_rc_eqh))
 318                         lnet_router_checker();
 319
 320                 LNET_LOCK();
 321 #endif
 322                 for (i = 0; i < neq; i++) {
 323                         lnet_eq_t *eq = lnet_handle2eq(&eventqs[i]);
 324
 325                         if (eq == NULL) {
 326                                 LNET_UNLOCK();
 327                                 RETURN(-ENOENT);
 328                         }
 329
 330                         rc = lib_get_event (eq, event);
 331                         if (rc != 0) {
 332                                 LNET_UNLOCK();
 333                                 *which = i;
 334                                 RETURN(rc);
 335                         }
 336                 }
 337
 338 #ifdef __KERNEL__
 339                 if (timeout_ms == 0) {
 340                         LNET_UNLOCK();
 341                         RETURN (0);
 342                 }
 343
 344                 cfs_waitlink_init(&wl);
 345                 cfs_set_current_state(CFS_TASK_INTERRUPTIBLE);
 346                 cfs_waitq_add(&the_lnet.ln_waitq, &wl);
 347
 348                 LNET_UNLOCK();
 349
 350                 if (timeout_ms < 0) {
 351                         cfs_waitq_wait (&wl, CFS_TASK_INTERRUPTIBLE);
 352                 } else {
 353                         struct timeval tv;
 354
 355                         now = cfs_time_current();
 356                         cfs_waitq_timedwait(&wl, CFS_TASK_INTERRUPTIBLE,
 357                                             cfs_time_seconds(timeout_ms)/1000);
 358                         cfs_duration_usec(cfs_time_sub(cfs_time_current(), now),
 359                                             &tv);
 360                         timeout_ms -= (int)(tv.tv_sec * 1000 + tv.tv_usec / 1000);
 361                         if (timeout_ms < 0)
 362                                 timeout_ms = 0;
 363                 }
 364
 365                 LNET_LOCK();
 366                 cfs_waitq_del(&the_lnet.ln_waitq, &wl);
 367 #else
 368                 if (eqwaitni != NULL) {
 369                         /* I have a single NI that I have to call into, to get
 370                          * events queued, or to block. */
 371                         lnet_ni_addref_locked(eqwaitni);
 372                         LNET_UNLOCK();
 373
 374                         if (timeout_ms <= 0) {
 375                                 (eqwaitni->ni_lnd->lnd_wait)(eqwaitni, timeout_ms);
 376                         } else {
 377                                 gettimeofday(&then, NULL);
 378
 379                                 (eqwaitni->ni_lnd->lnd_wait)(eqwaitni, timeout_ms);
 380
 381                                 gettimeofday(&now, NULL);
 382                                 timeout_ms -= (now.tv_sec - then.tv_sec) * 1000 +
 383                                               (now.tv_usec - then.tv_usec) / 1000;
 384                                 if (timeout_ms < 0)
 385                                         timeout_ms = 0;
 386                         }
 387
 388                         LNET_LOCK();
 389                         lnet_ni_decref_locked(eqwaitni);
 390
 391                         /* don't call into eqwaitni again if timeout has
 392                          * expired */
 393                         if (timeout_ms == 0)
 394                                 eqwaitni = NULL;
 395
 396                         continue;               /* go back and check for events */
 397                 }
 398
 399                 if (timeout_ms == 0) {
 400                         LNET_UNLOCK();
 401                         RETURN (0);
 402                 }
 403
 404 # ifndef HAVE_LIBPTHREAD
 405                 /* If I'm single-threaded, LNET fails at startup if it can't
 406                  * set the_lnet.ln_eqwaitni correctly.  */
 407                 LBUG();
 408 # else
 409                 if (timeout_ms < 0) {
 410                         pthread_cond_wait(&the_lnet.ln_cond,
 411                                           &the_lnet.ln_lock);
 412                 } else {
 413                         gettimeofday(&then, NULL);
 414
 415                         ts.tv_sec = then.tv_sec + timeout_ms/1000;
 416                         ts.tv_nsec = then.tv_usec * 1000 +
 417                                      (timeout_ms%1000) * 1000000;
 418                         if (ts.tv_nsec >= 1000000000) {
 419                                 ts.tv_sec++;
 420                                 ts.tv_nsec -= 1000000000;
 421                         }
 422
 423                         pthread_cond_timedwait(&the_lnet.ln_cond,
 424                                                &the_lnet.ln_lock, &ts);
 425
 426                         gettimeofday(&now, NULL);
 427                         timeout_ms -= (now.tv_sec - then.tv_sec) * 1000 +
 428                                       (now.tv_usec - then.tv_usec) / 1000;
 429
 430                         if (timeout_ms < 0)
 431                                 timeout_ms = 0;
 432                 }
 433 # endif
 434 #endif
 435         }
 436 }