tpool.cpp

#include <typeinfo>
#include <stdexcept>

#include <epicsEvent.h>
#include <epicsGuard.h>
#include <epicsThread.h>
#include <errlog.h>

#include <pv/sharedPtr.h>

#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();
        delete *it;
    }

    {
        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();
}