diff --git a/core/src/JobSystem.cpp b/core/src/JobSystem.cpp new file mode 100644 index 0000000..f072136 --- /dev/null +++ b/core/src/JobSystem.cpp @@ -0,0 +1,160 @@ +#include "vf/JobSystem.h" + +#include +#include +#include + +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> work; + std::vector names; + std::vector priorities; + std::vector> dependents; + std::vector> remaining; + std::vector> depFailed; + std::vector> states; + bool serial = false; +}; + +bool JobHandle::done() const { + if (!state) return true; + std::lock_guard lock(state->mutex); + return state->finished; +} + +bool JobHandle::failed() const { + if (!state) return false; + std::lock_guard lock(state->mutex); + return state->finished && state->error != nullptr; +} + +void JobHandle::wait() const { + if (!state) return; + std::unique_lock lock(state->mutex); + state->cv.wait(lock, [this] { return state->finished; }); +} + +std::exception_ptr JobHandle::error() const { + if (!state) return nullptr; + std::lock_guard 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 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 JobGraph::topologicalOrder() const { + std::vector indegree(nodes.size(), 0); + std::vector> 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 ready; + for (std::size_t i = 0; i < nodes.size(); ++i) + if (indegree[i] == 0) ready.push_back(i); + + std::vector 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 lock(mutex); + stopping = true; + } + workCv.notify_all(); + for (std::thread& worker : workers) worker.join(); +} +