1
#include <condition_variable>
2
#include <functional>
3
#include <future>
4
#include <mutex>
5
#include <queue>
6
#include <thread>
7
#include <vector>
8

9
class MiniPool {
10
public:
11
    explicit MiniPool(const size_t n) {
12
        for (size_t i = 0; i < n; ++i) {
13
            workers.emplace_back([this] {
14
                for (;;) {
15
                    std::function<void()> task;
16
                    {
17
                        std::unique_lock<std::mutex> lock(mtx);
18
                        cv.wait(lock, [this] { return stopping || !tasks.empty(); });
19
                        if (stopping && tasks.empty()) return;
20
                        task = std::move(tasks.front());
21
                        tasks.pop();
22
                    }
23
                    task();
24
                }
25
            });
26
        }
27
    }
28

29
    template <typename F>
30
    auto enqueue(F f) -> std::future<std::invoke_result_t<F>> {
31
        using R            = std::invoke_result_t<F>;
32
        auto t             = std::make_shared<std::packaged_task<R()>>(std::move(f));
33
        std::future<R> res = t->get_future();
34
        {
35
            std::lock_guard<std::mutex> lock(mtx);
36
            tasks.emplace([t] { (*t)(); });
37
        }
38
        cv.notify_one();
39
        return res;
40
    }
41

42
    ~MiniPool() {
43
        {
44
            std::lock_guard<std::mutex> lock(mtx);
45
            stopping = true;
46
        }
47
        cv.notify_all();
48
        for (auto& w : workers) w.join();
49
    }
50

51
private:
52
    std::vector<std::thread> workers;
53
    std::queue<std::function<void()>> tasks;
54
    std::mutex mtx;
55
    std::condition_variable cv;
56
    bool stopping = false;
57
};

1
class ThreadPool {
2
public:
3
    explicit ThreadPool(std::size_t thread_count = std::thread::hardware_concurrency()) : stop_(false), active_(0) {
4
        if (thread_count == 0) {
5
            thread_count = 1;
6
        }
7
        workers_.reserve(thread_count);
8
        for (std::size_t i = 0; i < thread_count; ++i) {
9
            workers_.emplace_back([this] { worker_loop(); });
10
        }
11
    }
12

13
    ThreadPool(const ThreadPool&)            = delete;
14
    ThreadPool& operator=(const ThreadPool&) = delete;
15
    ThreadPool(ThreadPool&&)                 = delete;
16
    ThreadPool& operator=(ThreadPool&&)      = delete;
17

18
    ~ThreadPool() {
19
        {
20
            std::lock_guard<std::mutex> lock(mutex_);
21
            stop_ = true;
22
        }
23
        cv_.notify_all();
24
        for (auto& t : workers_) {
25
            if (t.joinable()) {
26
                t.join();
27
            }
28
        }
29
    }
30

31
    template <class F, class... Args>
32
    auto enqueue(F&& f, Args&&... args) -> std::future<std::invoke_result_t<F, Args...>> {
33
        using R = std::invoke_result_t<F, Args...>;
34

35
        auto task = std::make_shared<std::packaged_task<R()>>(make_task_functor(std::forward<F>(f), std::forward<Args>(args)...));
36

37
        std::future<R> fut = task->get_future();
38

39
        {
40
            std::lock_guard<std::mutex> lock(mutex_);
41
            if (stop_) {
42
                throw std::runtime_error("enqueue on stopped ThreadPool");
43
            }
44
            tasks_.emplace_back([task] { (*task)(); });
45
        }
46

47
        cv_.notify_one();
48
        return fut;
49
    }
50

51
    void wait_idle() {
52
        std::unique_lock<std::mutex> lock(mutex_);
53
        idle_cv_.wait(lock, [this] { return tasks_.empty() && active_ == 0; });
54
    }
55

56
    std::size_t size() const noexcept {
57
        return workers_.size();
58
    }
59

60
private:
61
    using Task = std::function<void()>;
62

63
    template <class F, class... Args>
64
    static auto make_task_functor(F&& f, Args&&... args) {
65
        using Fn  = std::decay_t<F>;
66
        using Tup = std::tuple<std::decay_t<Args>...>;
67
        return [fn = Fn(std::forward<F>(f)), tup = Tup(std::forward<Args>(args)...)]() mutable { return std::apply(std::move(fn), std::move(tup)); };
68
    }
69

70
    void worker_loop() {
71
        for (;;) {
72
            Task task;
73
            {
74
                std::unique_lock<std::mutex> lock(mutex_);
75
                cv_.wait(lock, [this] { return stop_ || !tasks_.empty(); });
76
                if (stop_ && tasks_.empty()) {
77
                    return;
78
                }
79
                task = std::move(tasks_.front());
80
                tasks_.pop_front();
81
                ++active_;
82
            }
83

84
            task();
85

86
            {
87
                std::lock_guard<std::mutex> lock(mutex_);
88
                --active_;
89
                if (tasks_.empty() && active_ == 0) {
90
                    idle_cv_.notify_all();
91
                }
92
            }
93
        }
94
    }
95

96
    std::vector<std::thread> workers_;
97
    std::deque<Task> tasks_;
98
    mutable std::mutex mutex_;
99
    std::condition_variable cv_;
100
    std::condition_variable idle_cv_;
101
    bool stop_;
102
    std::size_t active_;
103
};