src/buildtool/multithreading/task_system.cpp


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

#include "src/buildtool/multithreading/task_system.hpp"

#include "gsl-lite/gsl-lite.hpp"
#include "src/buildtool/multithreading/task.hpp"

TaskSystem::TaskSystem() : TaskSystem(std::thread::hardware_concurrency()) {}

TaskSystem::TaskSystem(std::size_t number_of_threads)
    : thread_count_{std::max(std::size_t{1}, number_of_threads)},
      total_workload_{thread_count_} {
    for (std::size_t index = 0; index < thread_count_; ++index) {
        queues_.emplace_back(&total_workload_);
    }
    for (std::size_t index = 0; index < thread_count_; ++index) {
        threads_.emplace_back([&, index]() { Run(index); });
    }
}

TaskSystem::~TaskSystem() {
    Finish();    // wait for tasks to finish
    Shutdown();  // stop the threads
    for (auto& t : threads_) {
        t.join();
    }
}

void TaskSystem::Shutdown() noexcept {
    shutdown_ = true;
    // Abort the workload counter in case a system shut down was requested while
    // the task queues (workload) are not yet empty and someone is still waiting
    // on this counter to become zero.
    total_workload_.Abort();
    for (auto& q : queues_) {
        q.done();
    }
}

void TaskSystem::Finish() noexcept {
    // When starting a new task system all spawned threads will immediately go
    // to sleep and wait for tasks. Even after adding some tasks, it can take a
    // while until the first thread wakes up. Therefore, we need to wait for the
    // total workload (number of active threads _and_ total number of queued
    // tasks) to become zero.
    total_workload_.WaitForZero();
}

void TaskSystem::Run(std::size_t idx) {
    gsl_Expects(thread_count_ > 0);

    while (not shutdown_) {
        std::optional<Task> t{};
        for (std::size_t i = 0; i < thread_count_; ++i) {
            t = queues_[(idx + i) % thread_count_].try_pop();
            if (t) {
                break;
            }
        }

        // NOLINTNEXTLINE(clang-analyzer-core.DivideZero)
        t = t ? t : queues_[idx % thread_count_].pop();

        if (not t or shutdown_) {
            break;
        }

        (*t)();
    }
}