ThreadPool

Change-Id: I09bf867a2dbb45e063e4785f5b2b1f705fae72c8
Reviewed-on: http://gfw160.austin.ibm.com:8080/gerrit/680
Tested-by: Jenkins Server
Reviewed-by: Douglas R. Gilbert <dgilbert@us.ibm.com>
Reviewed-by: Bradley W. Bishop <bradleyb@us.ibm.com>
Reviewed-by: Terry J. Opie <opiet@us.ibm.com>
Reviewed-by: MIKE J. JONES <mjjones@us.ibm.com>
Reviewed-by: A. Patrick Williams III <iawillia@us.ibm.com>

6 files changed, 393 insertions, 1 deletions

diff --git a/src/usr/initservice/extinitsvc/extinitsvctasks.H b/src/usr/initservice/extinitsvc/extinitsvctasks.H
index 5d4e0db3c..c84627cdb 100644
--- a/src/usr/initservice/extinitsvc/extinitsvctasks.H
+++ b/src/usr/initservice/extinitsvc/extinitsvctasks.H
@@ -43,6 +43,19 @@ const TaskInfo g_exttaskinfolist[]   =   {
         //  -----   Extended Image  ------------------------------------------
 
         /**
+         * @brief  util library.
+         */
+        {
+            "libutil.so",                          // library name
+            NULL,                                  // no function to run
+            {
+                INIT_TASK,                         // init only
+                EXT_IMAGE,                         // extended image module
+                START_UTIL_ERRL_ID,                // error module id
+            }
+        },
+
+        /**
          * @brief   targeting task,
          */
         {
diff --git a/src/usr/makefile b/src/usr/makefile
index 60a40d488..579a2159a 100644
--- a/src/usr/makefile
+++ b/src/usr/makefile
@@ -27,6 +27,6 @@ OBJS = module_init.o
 SUBDIRS = example.d trace.d cxxtest.d testcore.d errl.d devicefw.d  \
           scom.d xscom.d targeting.d initservice.d hwpf.d \
           ecmddatabuffer.d pnor.d i2c.d vfs.d fsi.d hwas.d fsiscom.d \
-	  intr.d spd.d pore.d
+	  intr.d spd.d pore.d util.d
 
 include ${ROOTPATH}/config.mk
diff --git a/src/usr/util/makefile b/src/usr/util/makefile
new file mode 100644
index 000000000..d48edb34a
--- /dev/null
+++ b/src/usr/util/makefile
@@ -0,0 +1,30 @@
+#  IBM_PROLOG_BEGIN_TAG
+#  This is an automatically generated prolog.
+#
+#  $Source: src/usr/util/makefile $
+#
+#  IBM CONFIDENTIAL
+#
+#  COPYRIGHT International Business Machines Corp. 2012
+#
+#  p1
+#
+#  Object Code Only (OCO) source materials
+#  Licensed Internal Code Source Materials
+#  IBM HostBoot Licensed Internal Code
+#
+#  The source code for this program is not published or other-
+#  wise divested of its trade secrets, irrespective of what has
+#  been deposited with the U.S. Copyright Office.
+#
+#  Origin: 30
+#
+#  IBM_PROLOG_END
+ROOTPATH = ../../..
+MODULE = util
+
+OBJS = threadpool.o
+
+SUBDIRS = test.d
+
+include ${ROOTPATH}/config.mk
diff --git a/src/usr/util/test/makefile b/src/usr/util/test/makefile
new file mode 100644
index 000000000..e8b5ace45
--- /dev/null
+++ b/src/usr/util/test/makefile
@@ -0,0 +1,30 @@
+#  IBM_PROLOG_BEGIN_TAG
+#  This is an automatically generated prolog.
+#
+#  $Source: src/usr/util/test/makefile $
+#
+#  IBM CONFIDENTIAL
+#
+#  COPYRIGHT International Business Machines Corp. 2012
+#
+#  p1
+#
+#  Object Code Only (OCO) source materials
+#  Licensed Internal Code Source Materials
+#  IBM HostBoot Licensed Internal Code
+#
+#  The source code for this program is not published or other-
+#  wise divested of its trade secrets, irrespective of what has
+#  been deposited with the U.S. Copyright Office.
+#
+#  Origin: 30
+#
+#  IBM_PROLOG_END
+ROOTPATH = ../../../..
+
+MODULE = testutil
+
+TESTS = *.H
+
+include ${ROOTPATH}/config.mk
+
diff --git a/src/usr/util/test/threadpool.H b/src/usr/util/test/threadpool.H
new file mode 100644
index 000000000..412d4e546
--- /dev/null
+++ b/src/usr/util/test/threadpool.H
@@ -0,0 +1,196 @@
+//  IBM_PROLOG_BEGIN_TAG
+//  This is an automatically generated prolog.
+//
+//  $Source: src/usr/util/test/threadpool.H $
+//
+//  IBM CONFIDENTIAL
+//
+//  COPYRIGHT International Business Machines Corp. 2012
+//
+//  p1
+//
+//  Object Code Only (OCO) source materials
+//  Licensed Internal Code Source Materials
+//  IBM HostBoot Licensed Internal Code
+//
+//  The source code for this program is not published or other-
+//  wise divested of its trade secrets, irrespective of what has
+//  been deposited with the U.S. Copyright Office.
+//
+//  Origin: 30
+//
+//  IBM_PROLOG_END
+#ifndef __UTIL_TEST_THREADPOOL_H
+#define __UTIL_TEST_THREADPOOL_H
+
+#include <cxxtest/TestSuite.H>
+#include <util/threadpool.H>
+
+namespace __ThreadPoolTest
+{
+    /** WorkItem that simply modifies a pointer. */
+    struct Simple
+    {
+        explicit Simple(uint64_t* _v) : iv_value(_v) {};
+        void operator()() { (*iv_value) = 0xABCD; };
+
+        private:
+            uint64_t* iv_value;
+    };
+
+    /** WorkItem that modifies a value and has an order function. */
+    struct SimpleOrdered : public Simple
+    {
+        explicit SimpleOrdered(uint64_t* _v) : Simple(_v) {};
+        bool operator<(const SimpleOrdered& rhs) { return false; };
+    };
+
+    /** WorkItem that touches a barrier to ensure at least a certain number
+     *  of threads have been spawned. */
+    struct EnsureThreads
+    {
+        explicit EnsureThreads(barrier_t* _b) : iv_barrier(_b) {};
+        void operator()() { barrier_wait(iv_barrier); };
+
+        private:
+            barrier_t* iv_barrier;
+    };
+
+    /** WorkItem with an order function that modifies a value in order
+     *  sequence.
+     *
+     *  An EnsureOrder workitem is created with 0, 1, 2, etc.  The order
+     *  function ensures that item-0 should run before item-1, etc.  To
+     *  perform the test, a read-modify-write of a shared value is performed.
+     *  The value should be at the item-n value and the WorkItem increments
+     *  the value.
+     *
+     *  There is no issues with weak consistency with this object because only
+     *  a single thread is executing against the pool at this time.
+     */
+    struct EnsureOrder
+    {
+        EnsureOrder(uint64_t _i, uint64_t* _v) : iv_id(_i), iv_value(_v) {};
+        void operator()()
+            {
+                if ((*iv_value) != iv_id)
+                {
+                    TS_FAIL("ID number wrong. %d:%d", (*iv_value), iv_id);
+                }
+                (*iv_value) = iv_id + 1;
+            }
+        bool operator<(const EnsureOrder& rhs) { return iv_id < rhs.iv_id; };
+
+        private:
+            uint64_t iv_id;
+            uint64_t* iv_value;
+    };
+
+};
+
+
+class ThreadPoolTest: public CxxTest::TestSuite
+{
+    public:
+        /** Test that we can construct a ThreadPool for both FIFO and non-FIFO
+         *  ordered pools. */
+        void testSimpleWorkItem()
+        {
+            Util::ThreadPool<__ThreadPoolTest::Simple> instance;
+            uint64_t value = 0;
+
+            instance.insert(new __ThreadPoolTest::Simple(&value));
+            instance.start();
+            instance.shutdown();
+
+            if (value == 0)
+            {
+                TS_FAIL("Value was not changed by Simple worker thread.");
+            }
+
+            Util::ThreadPool<__ThreadPoolTest::SimpleOrdered> instance2;
+            value = 0;
+
+            instance2.insert(new __ThreadPoolTest::SimpleOrdered(&value));
+            instance2.start();
+            instance2.shutdown();
+
+            if (value == 0)
+            {
+                TS_FAIL("Value was not changed by SimpleOrdered"
+                        "worker thread.");
+            }
+
+        }
+
+        /** Test that the thread manager function creates at least as many
+         *  threads as directed. */
+        void testThreadManager()
+        {
+            Util::ThreadPool<__ThreadPoolTest::EnsureThreads> instance;
+            barrier_t barrier;
+
+            for (size_t count = 1; count < 5; count++)
+            {
+                Util::ThreadPoolManager::setThreadCount(count);
+                barrier_init(&barrier, count);
+
+                for (size_t i = 0; i < count; i++)
+                {
+                    instance.insert(
+                        new __ThreadPoolTest::EnsureThreads(&barrier));
+                }
+
+                instance.start();
+                instance.shutdown();
+
+                barrier_destroy(&barrier);
+            }
+        }
+
+        /** Test that the order functions work on the thread pool. */
+        void testThreadOrder()
+        {
+            Util::ThreadPool<__ThreadPoolTest::EnsureOrder> instance;
+            uint64_t value = 0;
+
+            // Ensure that adding work items in order works.
+            Util::ThreadPoolManager::setThreadCount(1);
+            for (uint64_t i = 0; i < 29; i++)
+            {
+                instance.insert(new __ThreadPoolTest::EnsureOrder(i, &value));
+            }
+
+            instance.start();
+            instance.shutdown();
+
+            // Ensure that adding work items in reverse order works.
+            value = 0;
+            for (uint64_t i = 0; i < 29; i++)
+            {
+                instance.insert(
+                    new __ThreadPoolTest::EnsureOrder(28 - i, &value));
+            }
+
+            instance.start();
+            instance.shutdown();
+
+            // Ensure that adding work items in a unsorted order works.
+            //
+            // Since 29 is a prime, i*n mod 29 generates the additive group of
+            // all integers (mod 29).  Use this to create a pseudo-random
+            // ordering of the integers 0-28: 0, 23, 17, 11, 5, 28, 22, ...
+            value = 0;
+            for (uint64_t i = 0; i < 29; i++)
+            {
+                instance.insert(
+                    new __ThreadPoolTest::EnsureOrder((i*23) % 29, &value));
+
+            }
+
+            instance.start();
+            instance.shutdown();
+        }
+};
+
+#endif
diff --git a/src/usr/util/threadpool.C b/src/usr/util/threadpool.C
new file mode 100644
index 000000000..e11d8807b
--- /dev/null
+++ b/src/usr/util/threadpool.C
@@ -0,0 +1,123 @@
+//  IBM_PROLOG_BEGIN_TAG
+//  This is an automatically generated prolog.
+//
+//  $Source: src/usr/util/threadpool.C $
+//
+//  IBM CONFIDENTIAL
+//
+//  COPYRIGHT International Business Machines Corp. 2012
+//
+//  p1
+//
+//  Object Code Only (OCO) source materials
+//  Licensed Internal Code Source Materials
+//  IBM HostBoot Licensed Internal Code
+//
+//  The source code for this program is not published or other-
+//  wise divested of its trade secrets, irrespective of what has
+//  been deposited with the U.S. Copyright Office.
+//
+//  Origin: 30
+//
+//  IBM_PROLOG_END
+#include <util/threadpool.H>
+#include <sys/task.h>
+#include <sys/misc.h>
+
+void Util::__Util_ThreadPool_Impl::ThreadPoolImpl::__init()
+{
+    // Initialize all member variables.
+    iv_worklist.clear();
+    mutex_init(&iv_mutex);
+    sync_cond_init(&iv_condvar);
+    iv_children.clear();
+    iv_shutdown = false;
+}
+
+void Util::__Util_ThreadPool_Impl::ThreadPoolImpl::__insert(void* i_workItem)
+{
+    mutex_lock(&iv_mutex);
+
+    // Insert item on to the work list and signal any worker thread that may
+    // be waiting.
+    iv_worklist.push_back(i_workItem);
+    sync_cond_signal(&iv_condvar);
+
+    mutex_unlock(&iv_mutex);
+}
+
+void* Util::__Util_ThreadPool_Impl::ThreadPoolImpl::__remove(order_fn_t fn)
+{
+    void* val = NULL;
+
+    mutex_lock(&iv_mutex);
+    do
+    {
+        // Wait until the worklist is empty or told to shutdown.
+        while(iv_worklist.empty() && !iv_shutdown)
+        {
+            sync_cond_wait(&iv_condvar, &iv_mutex);
+        }
+
+        // If told to shutdown and no work remains, end thread.
+        if (iv_worklist.empty() && iv_shutdown)
+        {
+            break;
+        }
+
+        // Otherwise, obtain next item from worklist, using order function
+        // passed to us.
+        worklist_itr_t itr = fn(iv_worklist.begin(), iv_worklist.end());
+
+        val = *itr;
+        iv_worklist.erase(itr);
+
+    } while(0);
+
+    mutex_unlock(&iv_mutex);
+
+    return val;
+}
+
+void Util::__Util_ThreadPool_Impl::ThreadPoolImpl::__start(
+        Util::__Util_ThreadPool_Impl::ThreadPoolImpl::start_fn_t fn,
+        void* instance)
+{
+    mutex_lock(&iv_mutex);
+
+    size_t thread_count = Util::ThreadPoolManager::getThreadCount();
+    while(thread_count--)
+    {
+        // Create children and add to a queue for later joining (during
+        // shutdown)
+        iv_children.push_back(task_create(fn, instance));
+    }
+
+    mutex_unlock(&iv_mutex);
+}
+
+void Util::__Util_ThreadPool_Impl::ThreadPoolImpl::__shutdown()
+{
+    mutex_lock(&iv_mutex);
+
+    // Set shutdown status and signal all children to release from their
+    // condition variable.
+    iv_shutdown = true;
+    sync_cond_broadcast(&iv_condvar);
+
+    // Join on all the children.
+    while(!iv_children.empty())
+    {
+        tid_t child = iv_children.front();
+        iv_children.pop_front();
+
+        mutex_unlock(&iv_mutex);
+        task_wait_tid(child, NULL, NULL);  // Don't need status.
+        mutex_lock(&iv_mutex);
+    }
+
+    mutex_unlock(&iv_mutex);
+}
+
+// Default thread count of one per HW thread.
+size_t Util::ThreadPoolManager::cv_size = cpu_thread_count();