add weak_ptr threaded work queue

pdbApp/tpool.cpp

@@ -0,0 +1,133 @@
+
+#include <typeinfo>
+#include <stdexcept>
+
+#include <epicsGuard.h>
+#include <errlog.h>
+
+#include <pv/sharedPtr.h>
+
+#define epicsExportSharedSymbols
+#include "helper.h"
+#include "tpool.h"
+
+typedef epicsGuard<epicsMutex> Guard;
+typedef epicsGuardRelease<epicsMutex> UnGuard;
+
+WorkQueue::WorkQueue(const std::string& name)
+    :name(name)
+    ,state(Idle)
+{}
+
+WorkQueue::~WorkQueue() { close(); }
+
+void WorkQueue::start(unsigned nworkers, unsigned prio)
+{
+    Guard G(mutex);
+
+    if(state!=Idle)
+        throw std::logic_error("Already started");
+
+    try {
+        state = Active;
+
+        for(unsigned i=0; i<nworkers; i++) {
+            p2p::auto_ptr<epicsThread> worker(new epicsThread(*this, name.c_str(),
+                                                              epicsThreadGetStackSize(epicsThreadStackSmall),
+                                                              prio));
+
+            worker->start();
+
+            workers.push_back(worker.get());
+            worker.release();
+        }
+    }catch(...){
+        UnGuard U(G); // unlock as close() blocks to join any workers which were started
+        close();
+        throw;
+    }
+}
+
+void WorkQueue::close()
+{
+    workers_t temp;
+
+    {
+        Guard G(mutex);
+        if(state!=Active)
+            return;
+
+        temp.swap(workers);
+        state = Stopping;
+    }
+
+    wakeup.signal();
+
+    for(workers_t::iterator it(temp.begin()), end(temp.end()); it!=end; ++it)
+    {
+        (*it)->exitWait();
+    }
+
+    {
+        Guard G(mutex);
+        state = Idle;
+    }
+}
+
+void WorkQueue::add(const value_type& work)
+{
+    bool empty;
+
+    {
+        Guard G(mutex);
+        if(state!=Active)
+            return;
+
+        empty = queue.empty();
+
+        queue.push_back(work);
+    }
+
+    if(empty) {
+        wakeup.signal();
+    }
+}
+
+void WorkQueue::run()
+{
+    Guard G(mutex);
+
+    std::tr1::shared_ptr<epicsThreadRunable> work;
+
+    while(state==Active) {
+
+        if(!queue.empty()) {
+            work = queue.front().lock();
+            queue.pop_front();
+        }
+
+        bool last = queue.empty();
+
+        {
+            UnGuard U(G);
+
+            if(work) {
+                try {
+                    work->run();
+                    work.reset();
+                }catch(std::exception& e){
+                    errlogPrintf("%s Unhandled exception from %s: %s\n",
+                                 name.c_str(), typeid(work.get()).name(), e.what());
+                    work.reset();
+                }
+            }
+
+            if(last) {
+                wakeup.wait();
+            }
+        }
+    }
+
+    // pass along the close() signal to next worker
+    wakeup.signal();
+}