Add parallel_for to ThreadPool and ParallelFor convenience function

.gitignore

@@ -15,3 +15,4 @@
 /active_project
 /test/assets/assets.db
 /test/products/
+_codeql_detected_source_root

cmake/thirdparty.cmake

@@ -37,7 +37,6 @@ if(EXISTS ${3RD_PATH})
     sky_find_3rd(TARGET crc32         DIR crc32c)
     sky_find_3rd(TARGET sfmt          DIR sfmt)
     sky_find_3rd(TARGET boost         DIR boost)
-    sky_find_3rd(TARGET taskflow      DIR taskflow)
     sky_find_3rd(TARGET mimalloc      DIR mimalloc)
 
     # framework

runtime/core/CMakeLists.txt

@@ -32,7 +32,6 @@ sky_add_library(TARGET Core STATIC
     LINK_LIBS
         3rdParty::sfmt
         3rdParty::crc32
-        3rdParty::taskflow
         3rdParty::boost
         3rdParty::rapidjson
 #        3rdParty::mimalloc

runtime/core/include/core/async/NamedThread.h

@@ -4,8 +4,7 @@
 
 #pragma once
 
-#include <taskflow/taskflow.hpp>
-
+#include <core/async/ThreadPool.h>
 #include <core/name/Name.h>
 #include <core/environment/Singleton.h>
 #include <core/async/Semaphore.h>
@@ -27,7 +26,7 @@ namespace sky {
         void Signal();
 
     private:
-        tf::Executor executor;
+        ThreadPool executor;
         Semaphore semaphore;
     };
 

runtime/core/include/core/async/Task.h

@@ -6,7 +6,7 @@
 
 #include <core/environment/Singleton.h>
 #include <core/template/ReferenceObject.h>
-#include <taskflow/taskflow.hpp>
+#include <core/async/ThreadPool.h>
 
 namespace sky {
 
@@ -24,7 +24,7 @@ namespace sky {
 
         bool IsWorking() const;
         void ResetTask();
-        tf::AsyncTask GetTask() const { return handle; }
+        AsyncTaskHandle GetTask() const { return handle; }
 
     protected:
         virtual bool DoWork() = 0;
@@ -33,9 +33,9 @@ namespace sky {
 
         friend class TaskExecutor;
 
-        std::atomic_bool           isDone{false};
-        tf::AsyncTask              handle;
-        std::vector<tf::AsyncTask> dependencies;
+        std::atomic_bool                isDone{false};
+        AsyncTaskHandle                 handle;
+        std::vector<AsyncTaskHandle>    dependencies;
 
     };
 
@@ -47,9 +47,15 @@ namespace sky {
 
         void WaitForAll();
 
-        tf::Executor &GetExecutor() { return executor; }
+        ThreadPool &GetExecutor() { return executor; }
     private:
-        tf::Executor executor;
+        ThreadPool executor;
     };
 
+    template <typename Index, typename Func>
+    void ParallelFor(Index begin, Index end, Func &&func)
+    {
+        TaskExecutor::Get()->GetExecutor().parallel_for(begin, end, std::forward<Func>(func));
+    }
+
 } // namespace sky

runtime/core/include/core/async/ThreadPool.h

@@ -0,0 +1,224 @@
+//
+// Created by blues on 2024/9/6.
+//
+
+#pragma once
+
+#include <atomic>
+#include <condition_variable>
+#include <functional>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <queue>
+#include <thread>
+#include <vector>
+
+namespace sky {
+
+    class AsyncTaskHandle {
+    public:
+        AsyncTaskHandle() = default;
+
+        explicit AsyncTaskHandle(std::shared_future<void> f)
+            : future(std::move(f))
+        {
+        }
+
+        bool empty() const { return !future.valid(); }
+
+        void reset() { future = {}; }
+
+        void Wait() const
+        {
+            if (future.valid()) {
+                future.wait();
+            }
+        }
+
+        bool IsReady() const
+        {
+            return !future.valid() ||
+                   future.wait_for(std::chrono::seconds(0)) == std::future_status::ready;
+        }
+
+    private:
+        friend class ThreadPool;
+        std::shared_future<void> future;
+    };
+
+    class ThreadPool {
+    public:
+        explicit ThreadPool(size_t numThreads = std::thread::hardware_concurrency());
+        ~ThreadPool();
+
+        // Submit a task and get a handle back
+        template <typename Func>
+        AsyncTaskHandle async(Func &&func)
+        {
+            auto task = std::make_shared<std::packaged_task<void()>>(std::forward<Func>(func));
+            std::shared_future<void> future = task->get_future().share();
+
+            taskCount.fetch_add(1, std::memory_order_relaxed);
+            {
+                std::lock_guard<std::mutex> lock(mutex);
+                tasks.emplace([task]() { (*task)(); });
+            }
+            condition.notify_one();
+            return AsyncTaskHandle(future);
+        }
+
+        // Submit a task without returning a handle (fire and forget)
+        template <typename Func>
+        void silent_async(Func &&func)
+        {
+            taskCount.fetch_add(1, std::memory_order_relaxed);
+            {
+                std::lock_guard<std::mutex> lock(mutex);
+                tasks.emplace([fn = std::forward<Func>(func)]() mutable {
+                    fn();
+                });
+            }
+            condition.notify_one();
+        }
+
+        // Submit a task that depends on other async tasks (variadic)
+        template <typename Func, typename ...Tasks>
+        std::pair<AsyncTaskHandle, std::future<void>> dependent_async(Func &&func, Tasks &&...deps)
+        {
+            std::vector<AsyncTaskHandle> depVec;
+            (depVec.emplace_back(std::forward<Tasks>(deps)), ...);
+            return dependent_async_impl(std::forward<Func>(func), depVec);
+        }
+
+        // Submit a task that depends on other async tasks (range)
+        template <typename Func, typename Iter>
+        std::pair<AsyncTaskHandle, std::future<void>> dependent_async(Func &&func, Iter begin, Iter end)
+        {
+            std::vector<AsyncTaskHandle> deps(begin, end);
+            return dependent_async_impl(std::forward<Func>(func), deps);
+        }
+
+        // Submit a dependent task without returning a future (fire and forget)
+        template <typename Func, typename ...Tasks>
+        AsyncTaskHandle silent_dependent_async(Func &&func, Tasks &&...deps)
+        {
+            std::vector<AsyncTaskHandle> depVec;
+            (depVec.emplace_back(std::forward<Tasks>(deps)), ...);
+            return silent_dependent_async_impl(std::forward<Func>(func), depVec);
+        }
+
+        // Parallel for loop: distributes [begin, end) across worker threads, calling func(index) for each index.
+        // Blocks until all iterations complete.
+        template <typename Index, typename Func>
+        void parallel_for(Index begin, Index end, Func &&func)
+        {
+            if (begin >= end) {
+                return;
+            }
+
+            const auto count = static_cast<size_t>(end - begin);
+            const auto numWorkers = workers.size();
+
+            if (numWorkers == 0 || count == 1) {
+                for (Index i = begin; i < end; ++i) {
+                    func(i);
+                }
+                return;
+            }
+
+            const auto chunks = std::min(count, numWorkers);
+            const auto chunkSize = count / chunks;
+            const auto remainder = count % chunks;
+
+            std::vector<AsyncTaskHandle> handles;
+            handles.reserve(chunks);
+
+            Index offset = begin;
+            for (size_t c = 0; c < chunks; ++c) {
+                Index chunkBegin = offset;
+                Index chunkEnd = chunkBegin + static_cast<Index>(chunkSize + (c < remainder ? 1 : 0));
+                offset = chunkEnd;
+
+                handles.emplace_back(async([&func, chunkBegin, chunkEnd]() {
+                    for (Index i = chunkBegin; i < chunkEnd; ++i) {
+                        func(i);
+                    }
+                }));
+            }
+
+            for (auto &h : handles) {
+                h.Wait();
+            }
+        }
+
+        // Wait for all submitted tasks to complete
+        void wait_for_all();
+
+        size_t NumWorkers() const { return workers.size(); }
+
+    private:
+        template <typename Func>
+        std::pair<AsyncTaskHandle, std::future<void>> dependent_async_impl(Func &&func, std::vector<AsyncTaskHandle> &deps)
+        {
+            auto task = std::make_shared<std::packaged_task<void()>>(
+                [fn = std::forward<Func>(func), dependencies = std::move(deps)]() mutable {
+                    for (auto &dep : dependencies) {
+                        dep.Wait();
+                    }
+                    fn();
+                });
+
+            std::future<void> rawFuture = task->get_future();
+            std::shared_future<void> sharedFuture = rawFuture.share();
+            AsyncTaskHandle handle(sharedFuture);
+
+            taskCount.fetch_add(1, std::memory_order_relaxed);
+            {
+                std::lock_guard<std::mutex> lock(mutex);
+                tasks.emplace([task]() { (*task)(); });
+            }
+            condition.notify_one();
+
+            auto resultFuture = std::async(std::launch::deferred, [sharedFuture]() {
+                sharedFuture.wait();
+            });
+
+            return {handle, std::move(resultFuture)};
+        }
+
+        template <typename Func>
+        AsyncTaskHandle silent_dependent_async_impl(Func &&func, std::vector<AsyncTaskHandle> &deps)
+        {
+            auto task = std::make_shared<std::packaged_task<void()>>(
+                [fn = std::forward<Func>(func), dependencies = std::move(deps)]() mutable {
+                    for (auto &dep : dependencies) {
+                        dep.Wait();
+                    }
+                    fn();
+                });
+
+            std::shared_future<void> future = task->get_future().share();
+
+            taskCount.fetch_add(1, std::memory_order_relaxed);
+            {
+                std::lock_guard<std::mutex> lock(mutex);
+                tasks.emplace([task]() { (*task)(); });
+            }
+            condition.notify_one();
+            return AsyncTaskHandle(future);
+        }
+
+        void workerLoop();
+
+        std::vector<std::thread> workers;
+        std::queue<std::function<void()>> tasks;
+        std::mutex mutex;
+        std::condition_variable condition;
+        std::atomic<bool> stopped{false};
+        std::atomic<int> taskCount{0};
+        std::mutex waitMutex;
+        std::condition_variable waitCondition;
+    };
+
+} // namespace sky

runtime/core/src/async/ThreadPool.cpp

@@ -0,0 +1,65 @@
+//
+// Created by blues on 2024/9/6.
+//
+
+#include <core/async/ThreadPool.h>
+
+namespace sky {
+
+    ThreadPool::ThreadPool(size_t numThreads)
+    {
+        for (size_t i = 0; i < numThreads; ++i) {
+            workers.emplace_back([this]() { workerLoop(); });
+        }
+    }
+
+    ThreadPool::~ThreadPool()
+    {
+        {
+            std::lock_guard<std::mutex> lock(mutex);
+            stopped.store(true, std::memory_order_relaxed);
+        }
+        condition.notify_all();
+        for (auto &worker : workers) {
+            if (worker.joinable()) {
+                worker.join();
+            }
+        }
+    }
+
+    void ThreadPool::workerLoop()
+    {
+        while (true) {
+            std::function<void()> task;
+            {
+                std::unique_lock<std::mutex> lock(mutex);
+                condition.wait(lock, [this]() {
+                    return stopped.load(std::memory_order_relaxed) || !tasks.empty();
+                });
+
+                if (stopped.load(std::memory_order_relaxed) && tasks.empty()) {
+                    return;
+                }
+
+                task = std::move(tasks.front());
+                tasks.pop();
+            }
+
+            task();
+
+            if (taskCount.fetch_sub(1, std::memory_order_acq_rel) == 1) {
+                std::lock_guard<std::mutex> lock(waitMutex);
+                waitCondition.notify_all();
+            }
+        }
+    }
+
+    void ThreadPool::wait_for_all()
+    {
+        std::unique_lock<std::mutex> lock(waitMutex);
+        waitCondition.wait(lock, [this]() {
+            return taskCount.load(std::memory_order_acquire) == 0;
+        });
+    }
+
+} // namespace sky