Thread Wait For Parent

Question

I am implementing a simple thread pool mechanism for my ubuntu server (for my multi-client anonymous chat program), and I need to make my worker threads sleep until a job (in the form of a function pointer and parameter) needs to be performed.

My current system is going out the window. I'm(the worker thread is) asking the manager if a job is available, and if there isn't sleep for 5ms. If there is, add the job to the working queue and run through the function. Wretched waste of cycles.

What I'd like to do is make a simple event-like system. I'm thinking about having a vector of mutexes (one for each worker) and have the handle to the mutex passed in as a parameter at creation. Then in my manager class (which holds and hands out jobs), whenever a thread is created, lock the mutex. When a job needs to be performed unlock a the next mutex in line, wait for it to be locked and unlocked, and relock it. However I'm wondering if there's a much better means to this end.

---

tldr; So my question is this. What is the most efficient, effective, and safest way to make a thread wait for a job from a managing class? Is polling a technique I should even consider (more than 1000 clients at a time), is mutex locking decent? Or are there other techniques?

Answer 1

What you need is the condition variable.
All the worker threads call wait() which will suspend them.

The parent thread then puts a work item on a queue and calls signal on the condition variable. This will wake one thread that is sleeping. It can remove the job from the queue execute the job then call wait on the condition variable to go back to sleep.

Try:

#include <pthread.h>
#include <memory>
#include <list>

// Use RAII to do the lock/unlock
struct MutexLock
{
    MutexLock(pthread_mutex_t& m) : mutex(m)    { pthread_mutex_lock(&mutex); }
    ~MutexLock()                                { pthread_mutex_unlock(&mutex); }
    private:
        pthread_mutex_t&    mutex;
};

// The base class of all work we want to do.
struct Job
{
    virtual void doWork()  = 0;
};

// pthreads is a C library the call back must be a C function.
extern "C" void* threadPoolThreadStart(void*);

// The very basre minimal part of a thread pool
// It does not create the workers. You need to create the work threads
// then make them call workerStart(). I leave that as an exercise for you.
class ThreadPool
{

    public:
         ThreadPool(unsigned int threadCount=1);
        ~ThreadPool();

        void addWork(std::auto_ptr<Job> job);
    private:

        friend void* threadPoolThreadStart(void*);
        void workerStart();

        std::auto_ptr<Job>  getJob();

        bool                finished;   // Threads will re-wait while this is true.
        pthread_mutex_t     mutex;      // A lock so that we can sequence accesses.
        pthread_cond_t      cond;       // The condition variable that is used to hold worker threads.
        std::list<Job*>     workQueue;  // A queue of jobs.
        std::vector<pthread_t>threads;
};

// Create the thread pool
ThreadPool::ThreadPool(int unsigned threadCount)
    : finished(false)
    , threads(threadCount)
{
    // If we fail creating either pthread object than throw a fit.
    if (pthread_mutex_init(&mutex, NULL) != 0)
    {   throw int(1);
    }

    if (pthread_cond_init(&cond, NULL) != 0)
    {
        pthread_mutex_destroy(&mutex);
        throw int(2);
    }
    for(unsigned int loop=0; loop < threadCount;++loop)
    {
       if (pthread_create(threads[loop], NULL, threadPoolThreadStart, this) != 0)
       {
            // One thread failed: clean up
            for(unsigned int kill = loop -1; kill < loop /*unsigned will wrap*/;--kill)
            {
                pthread_kill(threads[kill], 9);
            }
            throw int(3);
       }
    }
}

// Cleanup any left overs.
// Note. This does not deal with worker threads.
//       You need to add a method to flush all worker threads
//       out of this pobject before you let the destructor destroy it.
ThreadPool::~ThreadPool()
{
    finished = true;
    for(std::vector<pthread_t>::iterator loop = threads.begin();loop != threads.end(); ++loop)
    {
        // Send enough signals to free all threads.
        pthread_cond_signal(&cond);
    }
    for(std::vector<pthread_t>::iterator loop = threads.begin();loop != threads.end(); ++loop)
    {
        // Wait for all threads to exit (they will as finished is true and
        //                               we sent enough signals to make sure
        //                               they are running).
        void*  result;
        pthread_join(*loop, &result);
    }
    // Destroy the pthread objects.
    pthread_cond_destroy(&cond);
    pthread_mutex_destroy(&mutex);

    // Delete all re-maining jobs.
    // Notice how we took ownership of the jobs.
    for(std::list<Job*>::const_iterator loop = workQueue.begin(); loop != workQueue.end();++loop)
    {
        delete *loop;
    }
}

// Add a new job to the queue
// Signal the condition variable. This will flush a waiting worker
// otherwise the job will wait for a worker to finish processing its current job.
void ThreadPool::addWork(std::auto_ptr<Job> job)
{
    MutexLock  lock(mutex);

    workQueue.push_back(job.release());
    pthread_cond_signal(&cond);
}

// Start a thread.
// Make sure no exceptions escape as that is bad.
void* threadPoolThreadStart(void* data)
{
    ThreadPool* pool = reinterpret_cast<ThreadPool*>(workerStart);
    try
    {
        pool->workerStart();
    }
    catch(...){}
    return NULL;
}

// This is the main worker loop.
void ThreadPool::workerStart()
{
    while(!finished)
    {
        std::auto_ptr<Job>    job    = getJob();
        if (job.get() != NULL)
        {
            job->doWork();
        }
    }
}

// The workers come here to get a job.
// If there are non in the queue they are suspended waiting on cond
// until a new job is added above.
std::auto_ptr<Job> ThreadPool::getJob()
{
    MutexLock  lock(mutex);

    while((workQueue.empty()) && (!finished))
    {
        pthread_cond_wait(&cond, &mutex);
        // The wait releases the mutex lock and suspends the thread (until a signal).
        // When a thread wakes up it is help until it can acquire the mutex so when we
        // get here the mutex is again locked.
        //
        // Note: You must use while() here. This is because of the situation.
        //   Two workers:  Worker A processing job A.
        //                 Worker B suspended on condition variable.
        //   Parent adds a new job and calls signal.
        //   This wakes up thread B. But it is possible for Worker A to finish its
        //   work and lock the mutex before the Worker B is released from the above call.
        //
        //   If that happens then Worker A will see that the queue is not empty
        //   and grab the work item in the queue and start processing. Worker B will
        //   then lock the mutext and proceed here. If the above is not a while then
        //   it would try and remove an item from an empty queue. With a while it sees
        //   that the queue is empty and re-suspends on the condition variable above.
    }
    std::auto_ptr<Job>  result;
    if (!finished)
    {    result.reset(workQueue.front());
         workQueue.pop_front();
    }

    return result;
}

Answer 2

The usual way this is implemented is to have a queue queue of outstanding work, a mutex mutex protecting the queue, and a wait condition queue_not_empty. Then, each worker thread does the following (using pseudo-api):

while (true) {
    Work * work = 0;
    mutex.lock();
    while ( queue.empty() )
       if ( !queue_not_empty.wait( &mutex, timeout ) )
           return; // timeout - exit the worker thread
    work = queue.front();
    queue.pop_front();
    mutex.unlock();
    work->perform();
}

The wait( &mutex, timeout ) call blocks until either the wait condition is signaled, or the call times out. The mutex passed is atomically unlocked inside wait(), and locked again before returning from the call, to provide a consistent view of the queue to all participants. timeout would be chosen rather large (seconds), and would lead to the thread exiting (the thread pool would start new ones if more work came in).

Meanwhile, the thread pool's work insertion function does this:

Work * work = ...;
mutex.lock();
queue.push_back( work );
if ( worker.empty() )
    start_a_new_worker();
queue_not_empty.wake_one();
mutex.unlock();