mason/src2/taskmanager.cpp

#include "taskmanager.h"

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

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

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

void TaskManager::queue(const Job& job, const Result& result) {
    std::unique_lock lock(mutex);
    tasks.emplace(job, result);
    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) {
        Task task;
        std::unique_lock lock(mutex);
        while (!stopping && tasks.empty())
            loopConditional.wait(lock);

        if (stopping)
            return;

        ++activeThreads;
        task = tasks.front();
        tasks.pop();
        lock.unlock();

        executeTask(task);

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

}

void TaskManager::executeTask(const Task& task) const {
    try {
        task.first();
    } catch (const std::exception& e) {
        if (task.second.has_value())
            task.second.value()(&e);
    }

    try {
        if (task.second.has_value())
            task.second.value()(std::nullopt);
    } catch (...) {}
}


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