feat: add JobSystem

core/src/JobSystem.cpp

@@ -0,0 +1,160 @@
+#include "vf/JobSystem.h"
+
+#include <atomic>
+#include <stdexcept>
+#include <utility>
+
+namespace vf {
+
+struct JobHandle::State {
+    std::mutex mutex;
+    std::condition_variable cv;
+    std::exception_ptr error;
+    bool finished = false;
+};
+
+struct JobSystem::GraphRun {
+    explicit GraphRun(std::size_t count)
+        : work(count),
+          names(count),
+          priorities(count, JobPriority::Normal),
+          dependents(count),
+          remaining(count),
+          depFailed(count),
+          states(count) {}
+
+    std::vector<std::function<void()>> work;
+    std::vector<std::string> names;
+    std::vector<JobPriority> priorities;
+    std::vector<std::vector<std::size_t>> dependents;
+    std::vector<std::atomic<int>> remaining;
+    std::vector<std::atomic<bool>> depFailed;
+    std::vector<std::shared_ptr<JobHandle::State>> states;
+    bool serial = false;
+};
+
+bool JobHandle::done() const {
+    if (!state) return true;
+    std::lock_guard<std::mutex> lock(state->mutex);
+    return state->finished;
+}
+
+bool JobHandle::failed() const {
+    if (!state) return false;
+    std::lock_guard<std::mutex> lock(state->mutex);
+    return state->finished && state->error != nullptr;
+}
+
+void JobHandle::wait() const {
+    if (!state) return;
+    std::unique_lock<std::mutex> lock(state->mutex);
+    state->cv.wait(lock, [this] { return state->finished; });
+}
+
+std::exception_ptr JobHandle::error() const {
+    if (!state) return nullptr;
+    std::lock_guard<std::mutex> lock(state->mutex);
+    return state->error;
+}
+
+std::string JobHandle::errorMessage() const {
+    std::exception_ptr captured = error();
+    if (!captured) return std::string();
+    try {
+        std::rethrow_exception(captured);
+    } catch (const std::exception& ex) {
+        return ex.what();
+    } catch (...) {
+        return std::string("unknown exception");
+    }
+}
+
+void JobHandle::rethrowError() const {
+    std::exception_ptr captured = error();
+    if (captured) std::rethrow_exception(captured);
+}
+
+void JobBatch::wait() const {
+    for (const JobHandle& handle : handles) handle.wait();
+}
+
+bool JobBatch::done() const {
+    for (const JobHandle& handle : handles)
+        if (!handle.done()) return false;
+    return true;
+}
+
+std::size_t JobBatch::failedCount() const {
+    std::size_t count = 0;
+    for (const JobHandle& handle : handles)
+        if (handle.failed()) ++count;
+    return count;
+}
+
+std::exception_ptr JobBatch::firstError() const {
+    for (const JobHandle& handle : handles) {
+        std::exception_ptr captured = handle.error();
+        if (captured) return captured;
+    }
+    return nullptr;
+}
+
+std::size_t JobGraph::addNode(std::string name, std::function<void()> work, JobPriority priority) {
+    JobNode node;
+    node.name = std::move(name);
+    node.work = std::move(work);
+    node.priority = priority;
+    nodes.push_back(std::move(node));
+    return nodes.size() - 1;
+}
+
+void JobGraph::addDependency(std::size_t node, std::size_t dependsOn) {
+    nodes[node].dependencies.push_back(dependsOn);
+}
+
+std::vector<std::size_t> JobGraph::topologicalOrder() const {
+    std::vector<int> indegree(nodes.size(), 0);
+    std::vector<std::vector<std::size_t>> dependents(nodes.size());
+    for (std::size_t i = 0; i < nodes.size(); ++i) {
+        for (std::size_t dep : nodes[i].dependencies) {
+            ++indegree[i];
+            dependents[dep].push_back(i);
+        }
+    }
+
+    std::vector<std::size_t> ready;
+    for (std::size_t i = 0; i < nodes.size(); ++i)
+        if (indegree[i] == 0) ready.push_back(i);
+
+    std::vector<std::size_t> order;
+    order.reserve(nodes.size());
+    for (std::size_t cursor = 0; cursor < ready.size(); ++cursor) {
+        std::size_t current = ready[cursor];
+        order.push_back(current);
+        for (std::size_t dependent : dependents[current])
+            if (--indegree[dependent] == 0) ready.push_back(dependent);
+    }
+    return order;
+}
+
+bool JobGraph::hasCycle() const {
+    return topologicalOrder().size() != nodes.size();
+}
+
+JobSystem::JobSystem(unsigned threadCount) {
+    workers.reserve(threadCount);
+    for (unsigned i = 0; i < threadCount; ++i)
+        workers.emplace_back([this] { workerLoop(); });
+}
+
+JobSystem::~JobSystem() {
+    if (workers.empty()) return;
+    waitAll();
+    {
+        std::lock_guard<std::mutex> lock(mutex);
+        stopping = true;
+    }
+    workCv.notify_all();
+    for (std::thread& worker : workers) worker.join();
+}
+