liblava/liblava/util/thread.hpp at flow · liblava/liblava · GitHub

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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
/**
 * @file         liblava/util/thread.hpp
 * @brief        Thread pool
 * @authors      Lava Block OÜ and contributors
 * @copyright    Copyright (c) 2018-present, MIT License
 */

#pragma once

#include "liblava/core/id.hpp"
#include "liblava/core/time.hpp"
#include <condition_variable>
#include <deque>
#include <mutex>
#include <thread>

namespace lava {

/**
 * @brief Sleep for seconds
 * @param time    Seconds to sleep
 */
inline void sleep(seconds time) {
    std::this_thread::sleep_for(time);
}

/**
 * @brief Sleep for milliseconds
 * @param time    Milliseconds to sleep
 */
inline void sleep(ms time) {
    std::this_thread::sleep_for(time);
}

/**
 * @brief Sleep for microseconds
 * @param time    Microseconds to sleep
 */
inline void sleep(us time) {
    std::this_thread::sleep_for(time);
}

/**
 * @brief Thread pool
 */
struct thread_pool {
    /// Task function (with thread id)
    using task = std::function<void(id::ref)>;

    /**
     * @brief Set up the thread pool
     * @param count    Number of threads
     */
    void setup(ui32 count = 2) {
        for (auto i = 0u; i < count; ++i)
            m_workers.emplace_back(worker(*this));
    }

    /**
     * @brief Tear down the thread pool
     */
    void teardown() {
        m_stop = true;
        m_condition.notify_all();

        for (auto& worker : m_workers)
            worker.join();

        m_workers.clear();
    }

    /**
     * @brief Enqueue a task
     * @param f    Task function
     */
    void enqueue(auto f) {
        {
            std::unique_lock<std::mutex> lock(m_queue_mutex);
            m_tasks.push_back(task(f));
        }
        m_condition.notify_one();
    }

private:
    /**
     * @brief Thread worker
     */
    struct worker {
        /**
         * @brief Construct a new worker
         * @param pool    Thread pool
         */
        explicit worker(thread_pool& pool)
        : m_pool(pool) {}

        /**
         * @brief Run task operator
         */
        void operator()() {
            auto thread_id = ids::instance().next();

            task task;
            while (true) {
                {
                    std::unique_lock<std::mutex> lock(m_pool.m_queue_mutex);

                    while (!m_pool.m_stop && m_pool.m_tasks.empty())
                        m_pool.m_condition.wait(lock);

                    if (m_pool.m_stop)
                        break;

                    task = m_pool.m_tasks.front();
                    m_pool.m_tasks.pop_front();
                }

                task(thread_id);
            }
        }

    private:
        /// Thread pool
        thread_pool& m_pool;
    };

    /// List of workers
    std::vector<std::thread> m_workers;

    /// List of tasks
    std::deque<task> m_tasks;

    /// Queue mutex
    std::mutex m_queue_mutex;

    /// Condition variable
    std::condition_variable m_condition;

    /// Stop state
    bool m_stop = false;
};

} // namespace lava