WorkList.h

// WARNING: Changes to this file must be contributed back to Sawyer or else they will
//          be clobbered by the next update from Sawyer.  The Sawyer repository is at
//          https://github.com/matzke1/sawyer.
 
 
 
 
#ifndef Sawyer_WorkList_H
#define Sawyer_WorkList_H
 
#include <Sawyer/Synchronization.h>
#include <deque>
 
namespace Sawyer {
 
template<class T>
class WorkFifo {
    mutable SAWYER_THREAD_TRAITS::Mutex mutex_;         // protects the following data members
    std::deque<T> items_;                               // ordered list of work to be completed
 
public:
    typedef T Item;
 
    void insert(const Item &item) {
        SAWYER_THREAD_TRAITS::LockGuard lock(mutex_);
        items_.push_back(item);
    }
 
    bool isEmpty() const {
        SAWYER_THREAD_TRAITS::LockGuard lock(mutex_);
        return items_.empty();
    }
 
    Item next() {
        SAWYER_THREAD_TRAITS::LockGuard lock(mutex_);
        ASSERT_forbid(items_.empty());
        Item item = items_.front();
        items_.pop_front();
        return item;
    }
};
 
template<class WorkItems, class Functor>
void processWorkList(WorkItems &workList, size_t maxWorkers, Functor f) {
#if SAWYER_MULTI_THREADED
    boost::mutex mutex;
    boost::condition_variable cond;
    size_t nActiveWorkers = 0;
 
    if (0 == maxWorkers)
        maxWorkers = boost::thread::hardware_concurrency();
    ASSERT_require(maxWorkers > 0);
 
    while (1) {
        // Wait for work items or workers to be available.
        boost::unique_lock<boost::mutex> lock(mutex);
        while (!workList.isEmpty() && nActiveWorkers >= maxWorkers)
            cond.wait(lock);
 
        // Start as much work as possible
        while (!workList.isEmpty() && nActiveWorkers < maxWorkers) {
            // Assuming lambda's are not available, we work around.
            struct Worker {
                static void doWork(Functor f, const typename WorkItems::Item &item, WorkItems *workList,
                                   boost::mutex *mutex, boost::condition_variable *cond, size_t *nActiveWorkers) {
                    f(item, *workList);
                    boost::lock_guard<boost::mutex> lock(*mutex);
                    --*nActiveWorkers;
                    cond->notify_one();
                }
            };
 
            boost::thread thrd(Worker::doWork, f, workList.next(), &workList, &mutex, &cond, &nActiveWorkers);
            thrd.detach();
            ++nActiveWorkers;
        }
 
        // Return if there's nothing to do
        if (workList.isEmpty() && 0 == nActiveWorkers)
            break;
    }
#else
    while (!workList.isEmpty())
        f(workList.next(), workList);
#endif
}
 
} // namespace
#endif