mason/src2/taskmanager.cpp

#include "taskmanager.h"

TaskManager::TaskManager() :
    running(false),
    stopping(false),
    maxThreads(std::thread::hardware_concurrency()),
    activeThreads(0),
    jobs(),
    mutex(),
    loopConditional(),
    waitConditional(),
    threads()
{
    threads.reserve(maxThreads);
}

TaskManager::~TaskManager() {
    stop();
}

void TaskManager::queue(const Job& job) {
    std::unique_lock lock(mutex);
    jobs.emplace(job);
    lock.unlock();
    loopConditional.notify_one();
}

void TaskManager::start() {
    std::lock_guard lock(mutex);
    if (running)
        return;

    for (uint32_t i = 0; i < maxThreads; ++i)
        threads.emplace_back(&TaskManager::loop, this);

    running = true;
}

void TaskManager::stop() {
    std::unique_lock lock(mutex);
    if (!running)
        return;

    stopping = true;
    lock.unlock();

    loopConditional.notify_all();
    for (std::thread& thread : threads)
        thread.join();

    threads.clear();
}

void TaskManager::loop() {
    while (true) {
        Job job;
        std::unique_lock lock(mutex);
        while (!stopping && jobs.empty())
            loopConditional.wait(lock);

        if (stopping)
            return;

        ++activeThreads;
        job = jobs.front();
        jobs.pop();
        lock.unlock();

        job();
        lock.lock();
        --activeThreads;
        lock.unlock();
        waitConditional.notify_all();
    }

}

bool TaskManager::busy() const {
    std::lock_guard lock(mutex);
    return activeThreads == 0;
}

void TaskManager::wait() const {
    std::unique_lock lock(mutex);
    while (activeThreads != 0)
        waitConditional.wait(lock);
}