Whamcloud - gitweb
- fixed typo in a comment.
[fs/lustre-release.git] / lnet / lnet / lib-eq.c
index 8a91860..5bae602 100644 (file)
@@ -5,9 +5,8 @@
  * Library level Event queue management routines
  *
  *  Copyright (c) 2001-2003 Cluster File Systems, Inc.
- *  Copyright (c) 2001-2002 Sandia National Laboratories
  *
- *   This file is part of Lustre, http://www.sf.net/projects/lustre/
+ *   This file is part of Lustre, http://www.lustre.org
  *
  *   Lustre is free software; you can redistribute it and/or
  *   modify it under the terms of version 2 of the GNU General Public
  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  */
 
-#define DEBUG_SUBSYSTEM S_PORTALS
-#include <portals/lib-p30.h>
-#include <portals/arg-blocks.h>
+#define DEBUG_SUBSYSTEM S_LNET
+#include <lnet/lib-lnet.h>
 
-int do_PtlEQAlloc_internal(nal_cb_t * nal, void *private, void *v_args,
-                           void *v_ret)
+int
+LNetEQAlloc(unsigned int count, lnet_eq_handler_t callback, 
+            lnet_handle_eq_t *handle)
 {
-        /*
-         * Incoming:
-         *      ptl_handle_ni_t ni_in
-         *      ptl_size_t count_in
-         *      void                    * base_in
-         *
-         * Outgoing:
-         *      ptl_handle_eq_t         * handle_out
-         */
-
-        PtlEQAlloc_in *args = v_args;
-        PtlEQAlloc_out *ret = v_ret;
-
-        lib_eq_t *eq;
-        unsigned long flags;
-
-        /* api should have rounded up */
-        if (args->count_in != LOWEST_BIT_SET (args->count_in))
-                return ret->rc = PTL_VAL_FAILED;
-
-        eq = lib_eq_alloc (nal);
-        if (eq == NULL)
-                return (ret->rc = PTL_NO_SPACE);
+        lnet_eq_t     *eq;
 
-        state_lock(nal, &flags);
+        LASSERT (the_lnet.ln_init);
+        LASSERT (the_lnet.ln_refcount > 0);
+        
+        /* We need count to be a power of 2 so that when eq_{enq,deq}_seq
+         * overflow, they don't skip entries, so the queue has the same
+         * apparant capacity at all times */
 
-        if (nal->cb_map != NULL) {
-                struct iovec iov = {
-                        .iov_base = args->base_in,
-                        .iov_len = args->count_in * sizeof (ptl_event_t) };
+        if (count != LOWEST_BIT_SET(count)) {   /* not a power of 2 already */
+                do {                    /* knock off all but the top bit... */
+                        count &= ~LOWEST_BIT_SET (count);
+                } while (count != LOWEST_BIT_SET(count));
 
-                ret->rc = nal->cb_map (nal, 1, &iov, &eq->eq_addrkey);
-                if (ret->rc != PTL_OK) {
-                        lib_eq_free (nal, eq);
-                        
-                        state_unlock (nal, &flags);
-                        return (ret->rc);
-                }
+                count <<= 1;                             /* ...and round up */
         }
 
-        eq->sequence = 1;
-        eq->base = args->base_in;
-        eq->size = args->count_in;
+        if (count == 0)        /* catch bad parameter / overflow on roundup */
+                return (-EINVAL);
+        
+        eq = lnet_eq_alloc();
+        if (eq == NULL)
+                return (-ENOMEM);
+
+        LIBCFS_ALLOC(eq->eq_events, count * sizeof(lnet_event_t));
+        if (eq->eq_events == NULL) {
+                LNET_LOCK();
+                lnet_eq_free (eq);
+                LNET_UNLOCK();
+
+                return -ENOMEM;
+        }
+
+        /* NB this resets all event sequence numbers to 0, to be earlier
+         * than eq_deq_seq */
+        memset(eq->eq_events, 0, count * sizeof(lnet_event_t));
+
+        eq->eq_deq_seq = 1;
+        eq->eq_enq_seq = 1;
+        eq->eq_size = count;
         eq->eq_refcount = 0;
-        eq->event_callback = args->callback_in;
+        eq->eq_callback = callback;
 
-        lib_initialise_handle (nal, &eq->eq_lh, PTL_COOKIE_TYPE_EQ);
-        list_add (&eq->eq_list, &nal->ni.ni_active_eqs);
+        LNET_LOCK();
 
-        state_unlock(nal, &flags);
+        lnet_initialise_handle (&eq->eq_lh, LNET_COOKIE_TYPE_EQ);
+        list_add (&eq->eq_list, &the_lnet.ln_active_eqs);
 
-        ptl_eq2handle(&ret->handle_out, eq);
-        return (ret->rc = PTL_OK);
+        LNET_UNLOCK();
+
+        lnet_eq2handle(handle, eq);
+        return (0);
 }
 
-int do_PtlEQFree_internal(nal_cb_t * nal, void *private, void *v_args,
-                          void *v_ret)
+int
+LNetEQFree(lnet_handle_eq_t eqh)
 {
-        /*
-         * Incoming:
-         *      ptl_handle_eq_t eventq_in
-         *
-         * Outgoing:
-         */
-
-        PtlEQFree_in *args = v_args;
-        PtlEQFree_out *ret = v_ret;
-        lib_eq_t *eq;
-        long flags;
+        lnet_eq_t     *eq;
+        int            size;
+        lnet_event_t  *events;
 
-        state_lock (nal, &flags);
+        LASSERT (the_lnet.ln_init);
+        LASSERT (the_lnet.ln_refcount > 0);
+        
+        LNET_LOCK();
 
-        eq = ptl_handle2eq(&args->eventq_in, nal);
+        eq = lnet_handle2eq(&eqh);
         if (eq == NULL) {
-                ret->rc = PTL_EQ_INVALID;
-        } else if (eq->eq_refcount != 0) {
-                ret->rc = PTL_EQ_IN_USE;
+                LNET_UNLOCK();
+                return (-ENOENT);
+        }
+
+        if (eq->eq_refcount != 0) {
+                LNET_UNLOCK();
+                return (-EBUSY);
+        }
+
+        /* stash for free after lock dropped */
+        events  = eq->eq_events;
+        size    = eq->eq_size;
+
+        lnet_invalidate_handle (&eq->eq_lh);
+        list_del (&eq->eq_list);
+        lnet_eq_free (eq);
+
+        LNET_UNLOCK();
+
+        LIBCFS_FREE(events, size * sizeof (lnet_event_t));
+
+        return 0;
+}
+
+int
+lib_get_event (lnet_eq_t *eq, lnet_event_t *ev)
+{
+        int           new_index = eq->eq_deq_seq & (eq->eq_size - 1);
+        lnet_event_t *new_event = &eq->eq_events[new_index];
+        int           rc;
+        ENTRY;
+
+        CDEBUG(D_INFO, "event: %p, sequence: %lu, eq->size: %u\n",
+               new_event, eq->eq_deq_seq, eq->eq_size);
+
+        if (LNET_SEQ_GT (eq->eq_deq_seq, new_event->sequence)) {
+                RETURN(0);
+        }
+
+        /* We've got a new event... */
+        *ev = *new_event;
+
+        /* ...but did it overwrite an event we've not seen yet? */
+        if (eq->eq_deq_seq == new_event->sequence) {
+                rc = 1;
         } else {
-                if (nal->cb_unmap != NULL) {
-                        struct iovec iov = {
-                                .iov_base = eq->base,
-                                .iov_len = eq->size * sizeof (ptl_event_t) };
-                        
-                        nal->cb_unmap(nal, 1, &iov, &eq->eq_addrkey);
+                /* don't complain with CERROR: some EQs are sized small
+                 * anyway; if it's important, the caller should complain */
+                CDEBUG(D_NET, "Event Queue Overflow: eq seq %lu ev seq %lu\n",
+                       eq->eq_deq_seq, new_event->sequence);
+                rc = -EOVERFLOW;
+        }
+
+        eq->eq_deq_seq = new_event->sequence + 1;
+        RETURN(rc);
+}
+
+
+int
+LNetEQGet (lnet_handle_eq_t eventq, lnet_event_t *event)
+{
+        int which;
+
+        return LNetEQPoll(&eventq, 1, 0, 
+                         event, &which);
+}
+
+int
+LNetEQWait (lnet_handle_eq_t eventq, lnet_event_t *event)
+{
+        int which;
+
+        return LNetEQPoll(&eventq, 1, LNET_TIME_FOREVER,
+                         event, &which);
+}
+
+int
+LNetEQPoll (lnet_handle_eq_t *eventqs, int neq, int timeout_ms,
+            lnet_event_t *event, int *which)
+{
+        int              i;
+        int              rc;
+#ifdef __KERNEL__
+        cfs_waitlink_t   wl;
+        cfs_time_t       now;
+#else
+        struct timeval   then;
+        struct timeval   now;
+# ifdef HAVE_LIBPTHREAD
+        struct timespec  ts;
+# endif
+        lnet_ni_t       *eqwaitni = the_lnet.ln_eqwaitni;
+#endif
+        ENTRY;
+
+        LASSERT (the_lnet.ln_init);
+        LASSERT (the_lnet.ln_refcount > 0);
+
+        if (neq < 1)
+                RETURN(-ENOENT);
+
+        LNET_LOCK();
+
+        for (;;) {
+                for (i = 0; i < neq; i++) {
+                        lnet_eq_t *eq = lnet_handle2eq(&eventqs[i]);
+
+                        if (eq == NULL) {
+                                LNET_UNLOCK();
+                                RETURN(-ENOENT);
+                        }
+
+                        rc = lib_get_event (eq, event);
+                        if (rc != 0) {
+                                LNET_UNLOCK();
+                                *which = i;
+                                RETURN(rc);
+                        }
+                }
+                
+#ifdef __KERNEL__
+                if (timeout_ms == 0) {
+                        LNET_UNLOCK ();
+                        RETURN (0);
                 }
 
-                lib_invalidate_handle (nal, &eq->eq_lh);
-                list_del (&eq->eq_list);
-                lib_eq_free (nal, eq);
-                ret->rc = PTL_OK;
-        }
+                cfs_waitlink_init(&wl);
+                set_current_state(TASK_INTERRUPTIBLE);
+                cfs_waitq_add(&the_lnet.ln_waitq, &wl);
+
+                LNET_UNLOCK();
 
-        state_unlock (nal, &flags);
+                if (timeout_ms < 0) {
+                        cfs_waitq_wait (&wl, CFS_TASK_INTERRUPTIBLE);
+                } else { 
+                        struct timeval tv;
 
-        return (ret->rc);
+                        now = cfs_time_current();
+                        cfs_waitq_timedwait(&wl, CFS_TASK_INTERRUPTIBLE,
+                                            cfs_time_seconds(timeout_ms)/1000);
+                        cfs_duration_usec(cfs_time_sub(cfs_time_current(), now), 
+                                          &tv); 
+                        timeout_ms -= tv.tv_sec * 1000 + tv.tv_usec / 1000;
+                        if (timeout_ms < 0)
+                                timeout_ms = 0;
+                }
+                
+                LNET_LOCK();
+                cfs_waitq_del(&the_lnet.ln_waitq, &wl);
+#else
+                if (eqwaitni != NULL) {
+                        /* I have a single NI that I have to call into, to get
+                         * events queued, or to block. */
+                        lnet_ni_addref_locked(eqwaitni);
+                        LNET_UNLOCK();
+
+                        if (timeout_ms <= 0) {
+                                (eqwaitni->ni_lnd->lnd_wait)(eqwaitni, timeout_ms);
+                        } else {
+                                gettimeofday(&then, NULL);
+
+                                (eqwaitni->ni_lnd->lnd_wait)(eqwaitni, timeout_ms);
+                                
+                                gettimeofday(&now, NULL);
+                                timeout_ms -= (now.tv_sec - then.tv_sec) * 1000 +
+                                              (now.tv_usec - then.tv_usec) / 1000;
+                                if (timeout_ms < 0)
+                                        timeout_ms = 0;
+                        }
+
+                        LNET_LOCK();
+                        lnet_ni_decref_locked(eqwaitni);
+
+                        /* don't call into eqwaitni again if timeout has
+                         * expired */
+                        if (timeout_ms == 0)
+                                eqwaitni = NULL;
+
+                        continue;               /* go back and check for events */
+                }
+
+                if (timeout_ms == 0) {
+                        LNET_UNLOCK();
+                        RETURN (0);
+                }
+
+# ifndef HAVE_LIBPTHREAD
+                /* If I'm single-threaded, LNET fails at startup if it can't
+                 * set the_lnet.ln_eqwaitni correctly.  */
+                LBUG();
+# else
+                if (timeout_ms < 0) {
+                        pthread_cond_wait(&the_lnet.ln_cond, 
+                                          &the_lnet.ln_lock);
+                } else {
+                        gettimeofday(&then, NULL);
+                        
+                        ts.tv_sec = then.tv_sec + timeout_ms/1000;
+                        ts.tv_nsec = then.tv_usec * 1000 + 
+                                     (timeout_ms%1000) * 1000000;
+                        if (ts.tv_nsec >= 1000000000) {
+                                ts.tv_sec++;
+                                ts.tv_nsec -= 1000000000;
+                        }
+                        
+                        pthread_cond_timedwait(&the_lnet.ln_cond,
+                                               &the_lnet.ln_lock, &ts);
+                        
+                        gettimeofday(&now, NULL);
+                        timeout_ms -= (now.tv_sec - then.tv_sec) * 1000 +
+                                      (now.tv_usec - then.tv_usec) / 1000;
+                        
+                        if (timeout_ms < 0)
+                                timeout_ms = 0;
+                }
+# endif
+#endif
+        }
 }