Add opal_fifo_t class

This commit adds a new class: opal_fifo.h. The new class has atomic, non-atomic,
and opal_using_threads() conditoned routines. It should be used when first-in
first-out is required and should perform much better than using locks and an
opal_list_t. Like with opal_lifo_t there are two versions of the atomic
implementation: 128-bit compare-and-swap, and spin-locked. More implementations
can be added later (LL/SC comes to mind).

This commit also adds a unit test for the opal_fifo_t class. This test verifies
the fifo implementation when using multiple threads.

opal/class/Makefile.am

@@ -34,6 +34,7 @@ headers += \
         class/opal_object.h \
         class/opal_graph.h\
         class/opal_lifo.h \
+        class/opal_fifo.h \
         class/opal_pointer_array.h \
         class/opal_value_array.h \
         class/opal_ring_buffer.h \
@@ -50,6 +51,7 @@ lib@OPAL_LIB_PREFIX@open_pal_la_SOURCES += \
         class/opal_object.c \
         class/opal_graph.c\
         class/opal_lifo.c \
+        class/opal_fifo.c \
         class/opal_pointer_array.c \
         class/opal_value_array.c \
         class/opal_ring_buffer.c \

opal/class/opal_fifo.c

@@ -0,0 +1,41 @@
+/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
+/*
+ * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
+ *                         University Research and Technology
+ *                         Corporation.  All rights reserved.
+ * Copyright (c) 2004-2005 The University of Tennessee and The University
+ *                         of Tennessee Research Foundation.  All rights
+ *                         reserved.
+ * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
+ *                         University of Stuttgart.  All rights reserved.
+ * Copyright (c) 2004-2005 The Regents of the University of California.
+ *                         All rights reserved.
+ * Copyright (c) 2014      Los Alamos National Security, LLC. All rights
+ *                         reserved.
+ * $COPYRIGHT$
+ *
+ * Additional copyrights may follow
+ *
+ * $HEADER$
+ */
+
+#include "opal_config.h"
+#include "opal/class/opal_fifo.h"
+
+static void opal_fifo_construct (opal_fifo_t *fifo)
+{
+    OBJ_CONSTRUCT(&fifo->opal_fifo_ghost, opal_list_item_t);
+
+    fifo->opal_fifo_ghost.opal_list_next = &fifo->opal_fifo_ghost;
+
+    /** used to protect against ABA problems when not using a 128-bit compare-and-set */
+    fifo->opal_fifo_ghost.item_free = 0;
+
+    fifo->opal_fifo_head.data.counter = 0;
+    fifo->opal_fifo_head.data.item = &fifo->opal_fifo_ghost;
+
+    fifo->opal_fifo_tail.data.counter = 0;
+    fifo->opal_fifo_tail.data.item = &fifo->opal_fifo_ghost;
+}
+
+OBJ_CLASS_INSTANCE(opal_fifo_t, opal_object_t, opal_fifo_construct, NULL);

opal/class/opal_fifo.h

@@ -0,0 +1,310 @@
+/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
+/*
+ * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
+ *                         University Research and Technology
+ *                         Corporation.  All rights reserved.
+ * Copyright (c) 2004-2007 The University of Tennessee and The University
+ *                         of Tennessee Research Foundation.  All rights
+ *                         reserved.
+ * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
+ *                         University of Stuttgart.  All rights reserved.
+ * Copyright (c) 2004-2005 The Regents of the University of California.
+ *                         All rights reserved.
+ * Copyright (c) 2007      Voltaire All rights reserved.
+ * Copyright (c) 2010      IBM Corporation.  All rights reserved.
+ * Copyright (c) 2014      Los Alamos National Security, LLC. All rights
+ *                         reseved.
+ * $COPYRIGHT$
+ *
+ * Additional copyrights may follow
+ *
+ * $HEADER$
+ */
+
+#ifndef OPAL_FIFO_H_HAS_BEEN_INCLUDED
+#define OPAL_FIFO_H_HAS_BEEN_INCLUDED
+
+#include "opal_config.h"
+#include "opal/class/opal_lifo.h"
+
+#include "opal/sys/atomic.h"
+#include "opal/threads/mutex.h"
+
+BEGIN_C_DECLS
+
+/* Atomic First In First Out lists. If we are in a multi-threaded environment then the
+ * atomicity is insured via the compare-and-swap operation, if not we simply do a read
+ * and/or a write.
+ *
+ * There is a trick. The ghost element at the end of the list. This ghost element has
+ * the next pointer pointing to itself, therefore we cannot go past the end of the list.
+ * With this approach we will never have a NULL element in the list, so we never have
+ * to test for the NULL.
+ */
+struct opal_fifo_t {
+    opal_object_t super;
+
+    /** first element on the fifo */
+    volatile opal_counted_pointer_t opal_fifo_head;
+    /** last element on the fifo */
+    volatile opal_counted_pointer_t opal_fifo_tail;
+
+    /** list sentinel (always points to self) */
+    opal_list_item_t opal_fifo_ghost;
+};
+
+typedef struct opal_fifo_t opal_fifo_t;
+
+OPAL_DECLSPEC OBJ_CLASS_DECLARATION(opal_fifo_t);
+
+static inline opal_list_item_t *opal_fifo_head (opal_fifo_t* fifo)
+{
+    return (opal_list_item_t *) fifo->opal_fifo_head.data.item;
+}
+
+static inline opal_list_item_t *opal_fifo_tail (opal_fifo_t* fifo)
+{
+    return (opal_list_item_t *) fifo->opal_fifo_tail.data.item;
+}
+
+/* The ghost pointer will never change. The head will change via an atomic
+ * compare-and-swap. On most architectures the reading of a pointer is an
+ * atomic operation so we don't have to protect it.
+ */
+static inline bool opal_fifo_is_empty( opal_fifo_t* fifo )
+{
+    return opal_fifo_head (fifo) == &fifo->opal_fifo_ghost;
+}
+
+#if OPAL_HAVE_ATOMIC_CMPSET_128
+
+/* Add one element to the FIFO. We will return the last head of the list
+ * to allow the upper level to detect if this element is the first one in the
+ * list (if the list was empty before this operation).
+ */
+static inline opal_list_item_t *opal_fifo_push_atomic (opal_fifo_t *fifo,
+                                                       opal_list_item_t *item)
+{
+    opal_counted_pointer_t tail;
+
+    item->opal_list_next = &fifo->opal_fifo_ghost;
+
+    do {
+        tail.value = fifo->opal_fifo_tail.value;
+
+        if (opal_update_counted_pointer (&fifo->opal_fifo_tail, tail, item)) {
+            break;
+        }
+    } while (1);
+
+    opal_atomic_wmb ();
+
+    if (&fifo->opal_fifo_ghost == tail.data.item) {
+        /* update the head */
+        fifo->opal_fifo_head.data.item = item;
+    } else {
+        /* update previous item */
+        tail.data.item->opal_list_next = item;
+    }
+
+    return (opal_list_item_t *) tail.data.item;
+}
+
+/* Retrieve one element from the FIFO. If we reach the ghost element then the FIFO
+ * is empty so we return NULL.
+ */
+static inline opal_list_item_t *opal_fifo_pop_atomic (opal_fifo_t *fifo)
+{
+    opal_list_item_t *item, *next;
+    opal_counted_pointer_t head, tail;
+
+    do {
+        head.value = fifo->opal_fifo_head.value;
+        tail.value = fifo->opal_fifo_tail.value;
+        opal_atomic_rmb ();
+
+        item = (opal_list_item_t *) head.data.item;
+        next = (opal_list_item_t *) item->opal_list_next;
+
+        if (&fifo->opal_fifo_ghost == tail.data.item && &fifo->opal_fifo_ghost == item) {
+            return NULL;
+        }
+
+        /* the head or next pointer are in an inconsistent state. keep looping. */
+        if (tail.data.item != item && &fifo->opal_fifo_ghost != tail.data.item &&
+            &fifo->opal_fifo_ghost == next) {
+            continue;
+        }
+
+        /* try popping the head */
+        if (opal_update_counted_pointer (&fifo->opal_fifo_head, head, next)) {
+            break;
+        }
+    } while (1);
+
+    opal_atomic_wmb ();
+
+    /* check for tail and head consistency */
+    if (&fifo->opal_fifo_ghost == next) {
+        /* the head was just set to &fifo->opal_fifo_ghost. try to update the tail as well */
+        if (!opal_update_counted_pointer (&fifo->opal_fifo_tail, tail, &fifo->opal_fifo_ghost)) {
+            /* tail was changed by a push operation. wait for the item's next pointer to be se then
+             * update the head */
+
+            /* wait for next pointer to be updated by push */
+            while (&fifo->opal_fifo_ghost == item->opal_list_next) {
+                opal_atomic_rmb ();
+            }
+
+            /* update the head with the real next value. note that no other thread
+             * will be attempting to update the head until after it has been updated
+             * with the next pointer. push will not see an empty list and other pop
+             * operations will loop until the head is consistent. */
+            head.value = fifo->opal_fifo_head.value;
+            next = (opal_list_item_t *) item->opal_list_next;
+
+            assert (&fifo->opal_fifo_ghost == head.data.item);
+
+            fifo->opal_fifo_head.data.item = next;
+        }
+    }
+
+    item->opal_list_next = NULL;
+
+    return item;
+}
+
+#else
+
+/* When compare-and-set 128 is not available we avoid the ABA problem by
+ * using a spin-lock on the head (using the head counter). Otherwise
+ * the algorithm is identical to the compare-and-set 128 version. */
+static inline opal_list_item_t *opal_fifo_push_atomic (opal_fifo_t *fifo,
+                                                       opal_list_item_t *item)
+{
+    opal_list_item_t *tail_item;
+
+    item->opal_list_next = &fifo->opal_fifo_ghost;
+
+    /* try to get the tail */
+    tail_item = opal_atomic_swap_ptr (&fifo->opal_fifo_tail.data.item, item);
+
+    opal_atomic_wmb ();
+
+    if (&fifo->opal_fifo_ghost == tail_item) {
+        /* update the head */
+        fifo->opal_fifo_head.data.item = item;
+    } else {
+        /* update previous item */
+        tail_item->opal_list_next = item;
+    }
+
+    opal_atomic_wmb ();
+
+    return (opal_list_item_t *) tail_item;
+}
+
+/* Retrieve one element from the FIFO. If we reach the ghost element then the FIFO
+ * is empty so we return NULL.
+ */
+static inline opal_list_item_t *opal_fifo_pop_atomic (opal_fifo_t *fifo)
+{
+    opal_list_item_t *item, *next;
+
+    do {
+        if (opal_atomic_cmpset_32 ((int32_t *) &fifo->opal_fifo_head.data.counter, 0, 1)) {
+            break;
+        }
+
+        opal_atomic_wmb ();
+    } while (1);
+
+    item = opal_fifo_head (fifo);
+    if (&fifo->opal_fifo_ghost == item) {
+        fifo->opal_fifo_head.data.counter = 0;
+        return NULL;
+    }
+
+    next = (opal_list_item_t *) item->opal_list_next;
+    fifo->opal_fifo_head.data.item = next;
+
+    if (&fifo->opal_fifo_ghost == next) {
+        if (!opal_atomic_cmpset_ptr (&fifo->opal_fifo_tail.data.item, item, &fifo->opal_fifo_ghost)) {
+            while (&fifo->opal_fifo_ghost == item->opal_list_next) {
+                opal_atomic_rmb ();
+            }
+
+            fifo->opal_fifo_head.data.item = (opal_list_item_t *) item->opal_list_next;
+        }
+    }
+
+    opal_atomic_wmb ();
+
+    /* unlock the head */
+    fifo->opal_fifo_head.data.counter = 0;
+
+    item->opal_list_next = NULL;
+
+    return item;
+}
+
+#endif
+
+/* single threaded versions of push/pop */
+static inline opal_list_item_t *opal_fifo_push_st (opal_fifo_t *fifo,
+                                                   opal_list_item_t *item)
+{
+    opal_list_item_t *prev = opal_fifo_tail (fifo);
+
+    item->opal_list_next = &fifo->opal_fifo_ghost;
+
+    fifo->opal_fifo_tail.data.item = item;
+    if (&fifo->opal_fifo_ghost == opal_fifo_head (fifo)) {
+        fifo->opal_fifo_head.data.item = item;
+    } else {
+        prev->opal_list_next = item;
+    }
+
+    return (opal_list_item_t *) item->opal_list_next;
+}
+
+static inline opal_list_item_t *opal_fifo_pop_st (opal_fifo_t *fifo)
+{
+    opal_list_item_t *item = opal_fifo_head (fifo);
+
+    if (item == &fifo->opal_fifo_ghost) {
+        return NULL;
+    }
+
+    fifo->opal_fifo_head.data.item = (opal_list_item_t *) item->opal_list_next;
+    if (&fifo->opal_fifo_ghost == opal_fifo_head (fifo)) {
+        fifo->opal_fifo_tail.data.item = &fifo->opal_fifo_ghost;
+    }
+
+    item->opal_list_next = NULL;
+    return item;
+}
+
+/* push/pop versions conditioned off opal_using_threads() */
+static inline opal_list_item_t *opal_fifo_push (opal_fifo_t *fifo,
+                                                opal_list_item_t *item)
+{
+    if (opal_using_threads ()) {
+        return opal_fifo_push_atomic (fifo, item);
+    }
+
+    return opal_fifo_push_st (fifo, item);
+}
+
+static inline opal_list_item_t *opal_fifo_pop (opal_fifo_t *fifo)
+{
+    if (opal_using_threads ()) {
+        return opal_fifo_pop_atomic (fifo);
+    }
+
+    return opal_fifo_pop_st (fifo);
+}
+
+END_C_DECLS
+
+#endif  /* OPAL_FIFO_H_HAS_BEEN_INCLUDED */

test/class/Makefile.am

@@ -34,7 +34,8 @@ check_PROGRAMS = \
 	opal_list \
 	opal_value_array \
 	opal_pointer_array \
-	opal_lifo
+	opal_lifo \
+	opal_fifo
 
 # The tests are built, but since they're not actively maintained, they
 # are not run during "make check".
@@ -94,5 +95,11 @@ opal_lifo_LDADD = \
 	$(top_builddir)/test/support/libsupport.a
 opal_lifo_DEPENDENCIES = $(opal_lifo_LDADD)
 
+opal_fifo_SOURCES = opal_fifo.c
+opal_fifo_LDADD = \
+        $(top_builddir)/opal/libopen-pal.la \
+	$(top_builddir)/test/support/libsupport.a
+opal_fifo_DEPENDENCIES = $(opal_fifo_LDADD)
+
 clean-local:
 	rm -f opal_bitmap_test_out.txt opal_hash_table_test_out.txt opal_proc_table_test_out.txt

test/class/opal_fifo.c

@@ -0,0 +1,252 @@
+/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
+/*
+ * Copyright (c) 2014      Los Alamos National Security, LLC. All rights
+ *                         reserved.
+ * $COPYRIGHT$
+ *
+ * Additional copyrights may follow
+ *
+ * $HEADER$
+ */
+
+#include "opal_config.h"
+#include <assert.h>
+
+#include "support.h"
+#include "opal/class/opal_fifo.h"
+#include "opal/runtime/opal.h"
+#include "opal/constants.h"
+
+#include <stdlib.h>
+#include <stdio.h>
+#include <stddef.h>
+
+#include <sys/time.h>
+
+#define OPAL_FIFO_TEST_THREAD_COUNT 8
+#define ITERATIONS 1000000
+#define ITEM_COUNT 100
+#define ITEMS_PER_LOOP 30
+
+static void *thread_test (void *arg) {
+    opal_fifo_t *fifo = (opal_fifo_t *) arg;
+    opal_list_item_t *item;
+    struct timeval start, stop, total;
+    double timing;
+
+    gettimeofday (&start, NULL);
+    for (int i = 0 ; i < ITERATIONS ; ++i) {
+        item = opal_fifo_pop_atomic (fifo);
+        if (NULL != item) {
+            (void) opal_fifo_push_atomic (fifo, item);
+        }
+    }
+    gettimeofday (&stop, NULL);
+
+    timersub(&stop, &start, &total);
+
+    timing = ((double) total.tv_sec + (double) total.tv_usec * 1e-6) / (double) ITERATIONS;
+
+    printf ("Atomics thread finished. Time: %d s %d us %d nsec/poppush\n", (int) total.tv_sec,
+            total.tv_usec, (int)(timing / 1e-9));
+
+    return NULL;
+}
+
+static void *thread_test_exhaust (void *arg) {
+  opal_fifo_t *fifo = (opal_fifo_t *) arg;
+  opal_list_item_t *items[ITEMS_PER_LOOP];
+  struct timeval start, stop, total;
+  int item_count = 0;
+  double timing;
+
+  gettimeofday (&start, NULL);
+
+  for (int i = 0 ; i < ITERATIONS ; i += ITEMS_PER_LOOP) {
+    for (int j = 0 ; j < ITEMS_PER_LOOP ; ++j) {
+      items[j] = opal_fifo_pop_atomic (fifo);
+      if (items[j]) {
+	++item_count;
+      }
+    }
+
+    for (int j = 0 ; j < ITEMS_PER_LOOP ; ++j) {
+      if (items[j]) {
+	(void) opal_fifo_push_atomic (fifo, items[j]);
+      }
+    }
+  }
+
+  gettimeofday (&stop, NULL);
+
+  timersub(&stop, &start, &total);
+
+  timing = ((double) total.tv_sec + (double) total.tv_usec * 1e-6) / (double) item_count;
+
+  fprintf (stderr, "Exhaustive atomics thread finished. Popped %d items. Time: %d s %d us %d nsec/poppush\n", item_count,
+	   (int) total.tv_sec, total.tv_usec, (int)(timing / 1e-9));
+
+  return NULL;
+}
+
+static bool check_fifo_consistency (opal_fifo_t *fifo, int expected_count)
+{
+    opal_list_item_t *item;
+    int count;
+
+    for (count = 0, item = fifo->opal_fifo_head.data.item ; item != &fifo->opal_fifo_ghost ;
+         item = opal_list_get_next(item), count++);
+
+    return count == expected_count;
+}
+
+int main (int argc, char *argv[]) {
+    pthread_t threads[OPAL_FIFO_TEST_THREAD_COUNT];
+    opal_list_item_t *item, *prev, *item2;
+    struct timeval start, stop, total;
+    opal_fifo_t fifo;
+    bool success;
+    double timing;
+    int rc;
+
+    rc = opal_init (&argc, &argv);
+    test_verify_int(OPAL_SUCCESS, rc);
+    if (OPAL_SUCCESS != rc) {
+        test_finalize();
+        exit (1);
+    }
+
+    test_init("opal_fifo_t");
+
+    OBJ_CONSTRUCT(&fifo, opal_fifo_t);
+
+    item = OBJ_NEW(opal_list_item_t);
+    prev = opal_fifo_push_st (&fifo, item);
+    if (&fifo.opal_fifo_ghost == prev) {
+        test_success ();
+    } else {
+        test_failure (" opal_fifo_push_st on empty fifo");
+    }
+
+    item2 = opal_fifo_pop_st (&fifo);
+    if (item == item2) {
+        test_success ();
+    } else {
+        test_failure (" opal_fifo_pop_st");
+    }
+
+    OBJ_RELEASE(item);
+
+    for (int i = 0 ; i < ITEM_COUNT ; ++i) {
+        item = OBJ_NEW(opal_list_item_t);
+        item->item_free = 0;
+        opal_fifo_push_st (&fifo, item);
+    }
+
+    if (check_fifo_consistency (&fifo, ITEM_COUNT)) {
+        test_success ();
+    } else {
+        test_failure (" opal_fifo_push_st(multiple items)");
+    }
+
+    gettimeofday (&start, NULL);
+    for (int i = 0 ; i < ITERATIONS ; ++i) {
+        item = opal_fifo_pop_st (&fifo);
+        (void) opal_fifo_push_st (&fifo, item);
+    }
+    gettimeofday (&stop, NULL);
+
+    timersub(&stop, &start, &total);
+
+    timing = ((double) total.tv_sec + (double) total.tv_usec * 1e-6) / (double) ITERATIONS;
+
+    if (check_fifo_consistency (&fifo, ITEM_COUNT)) {
+        test_success ();
+    } else {
+        test_failure (" fifo push/pop");
+    }
+
+    printf ("Single thread test. Time: %d s %d us %d nsec/poppush\n", (int) total.tv_sec,
+            total.tv_usec, (int)(timing / 1e-9));
+
+    thread_test (&fifo);
+
+    if (check_fifo_consistency (&fifo, ITEM_COUNT)) {
+        test_success ();
+    } else {
+        test_failure (" fifo push/pop single-threaded with atomics");
+    }
+
+    gettimeofday (&start, NULL);
+    for (int i = 0 ; i < OPAL_FIFO_TEST_THREAD_COUNT ; ++i) {
+        pthread_create (threads + i, NULL, thread_test, &fifo);
+    }
+
+    for (int i = 0 ; i < OPAL_FIFO_TEST_THREAD_COUNT ; ++i) {
+        void *ret;
+
+        pthread_join (threads[i], &ret);
+    }
+    gettimeofday (&stop, NULL);
+
+    timersub(&stop, &start, &total);
+
+    timing = ((double) total.tv_sec + (double) total.tv_usec * 1e-6) / (double) (ITERATIONS * OPAL_FIFO_TEST_THREAD_COUNT);
+
+    if (check_fifo_consistency (&fifo, ITEM_COUNT)) {
+        test_success ();
+    } else {
+        test_failure (" fifo push/pop multi-threaded with atomics");
+    }
+
+    printf ("All threads finished. Thread count: %d Time: %d s %d us %d nsec/poppush\n",
+             OPAL_FIFO_TEST_THREAD_COUNT, (int) total.tv_sec, total.tv_usec, (int)(timing / 1e-9));
+
+
+    gettimeofday (&start, NULL);
+    for (int i = 0 ; i < OPAL_FIFO_TEST_THREAD_COUNT ; ++i) {
+        pthread_create (threads + i, NULL, thread_test_exhaust, &fifo);
+    }
+
+    for (int i = 0 ; i < OPAL_FIFO_TEST_THREAD_COUNT ; ++i) {
+        void *ret;
+
+        pthread_join (threads[i], &ret);
+    }
+    gettimeofday (&stop, NULL);
+
+    timersub(&stop, &start, &total);
+
+    timing = ((double) total.tv_sec + (double) total.tv_usec * 1e-6) / (double) (ITERATIONS * OPAL_FIFO_TEST_THREAD_COUNT);
+
+    if (check_fifo_consistency (&fifo, ITEM_COUNT)) {
+        test_success ();
+    } else {
+        test_failure (" fifo push/pop multi-threaded with atomics when there are insufficient items");
+    }
+
+    printf ("All threads finished. Thread count: %d Time: %d s %d us %d nsec/poppush\n",
+             OPAL_FIFO_TEST_THREAD_COUNT, (int) total.tv_sec, total.tv_usec, (int)(timing / 1e-9));
+
+    success = true;
+    for (int i = 0 ; i < ITEM_COUNT ; ++i) {
+        item = opal_fifo_pop_st (&fifo);
+        if (NULL == item) {
+            success = false;
+            break;
+        }
+        OBJ_RELEASE(item);
+    }
+
+    if (success) {
+        test_success ();
+    } else {
+        test_failure (" fifo pop all items");
+    }
+
+    OBJ_DESTRUCT(&fifo);
+
+    opal_finalize ();
+
+    return test_finalize ();
+}