线程池和异步任务

src/core/async/async_task.cpp

@@ -0,0 +1 @@
+#include "async_task.h"

src/core/async/async_task.h

@@ -0,0 +1,5 @@
+#pragma once
+
+class async_task {
+
+};

src/core/async/queued_thread_pool.cpp

@@ -0,0 +1,32 @@
+#include "queued_thread_pool.h"
+
+queued_thread_pool::queued_thread_pool(const size_t num_threads) : stop(false) {
+    // 创建线程
+    for (std::size_t i = 0; i < num_threads; ++i) {
+        threads.emplace_back(&queued_thread_pool::worker_thread, this);
+    }
+}
+
+queued_thread_pool::~queued_thread_pool() {
+    stop = true;
+    // 唤醒所有线程
+    task_available.broadcast_signal();
+}
+
+void queued_thread_pool::worker_thread() {
+    while (true) {
+        // 如果线程池停止且任务队列为空，则退出
+        if (stop && tasks.empty()) {
+            return;
+        }
+
+        // 从任务队列中取出任务并执行
+        if (const auto& task = tasks.pop()) {
+            task.value()();
+            continue;
+        }
+
+        // 如果任务队列为空，则等待任务
+        task_available.wait();
+    }
+}

src/core/async/queued_thread_pool.h

@@ -0,0 +1,44 @@
+#pragma once
+#include <thread>
+#include <queue>
+#include <mutex>
+#include <functional>
+#include <future>
+#include <memory>
+
+#include "thread_event.h"
+#include "containers/safe_queue.h"
+#include "containers/safe_vector.h"
+
+class queued_thread_pool {
+
+public:
+    explicit queued_thread_pool(size_t num_threads = std::thread::hardware_concurrency());
+    ~queued_thread_pool();
+
+    template<typename F, typename... Args>
+    auto submit(F&& in_func, Args&&... in_args) {
+        using return_type = std::invoke_result_t<F, Args...>;
+
+        auto bind_task = std::make_shared<std::packaged_task<return_type()>>(
+            std::bind(std::forward<F>(in_func), std::forward<Args>(in_args)...)
+        );
+
+        auto res = bind_task->get_future();
+        if (stop) { throw std::runtime_error("submit on stopped ThreadPool"); }
+
+        tasks.push([bind_task] { (*bind_task)(); });
+        task_available.signal();
+
+        return res;
+    }
+private:
+    void worker_thread();
+
+    std::vector<std::jthread> threads;
+    safe_queue<std::function<void()>> tasks;
+
+    // 同步原语
+    thread_event task_available;
+    std::atomic_bool stop;
+};

src/core/async/thread_event.h

@@ -0,0 +1,35 @@
+#pragma once
+#include <condition_variable>
+#include <mutex>
+#include <thread>
+
+class thread_event {
+public:
+    void wait() {
+        std::unique_lock lock(mutex);
+        cv.wait(lock, [this] { return signaled; });
+        signaled = false;
+    }
+
+    void wait(std::chrono::milliseconds timeout) {
+        std::unique_lock lock(mutex);
+        cv.wait_for(lock, timeout, [this] { return signaled; });
+        signaled = false;
+    }
+
+    void signal() {
+        std::lock_guard lock(mutex);
+        signaled = true;
+        cv.notify_one();
+    }
+
+    void broadcast_signal() {
+        std::lock_guard lock(mutex);
+        signaled = true;
+        cv.notify_all();
+    }
+private:
+    std::condition_variable cv;
+    std::mutex mutex;
+    bool signaled = false;
+};