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