From a082c1d1814acb245036d061850d1d7ecbe13fba Mon Sep 17 00:00:00 2001 From: DualFroz Date: Wed, 2 Sep 2026 19:26:02 +0200 Subject: [PATCH] feat(vf): add JobStats and JobNode --- core/include/vf/JobSystem.h | 178 ++++++++++++++++++++++++++++++++++++ 1 file changed, 178 insertions(+) create mode 100644 core/include/vf/JobSystem.h diff --git a/core/include/vf/JobSystem.h b/core/include/vf/JobSystem.h new file mode 100644 index 0000000..bbea868 --- /dev/null +++ b/core/include/vf/JobSystem.h @@ -0,0 +1,178 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace vf { + +// Three service classes. High is drained before Normal, Normal before Low, so a +// chunk right under the observer never waits behind the far edge of the view. +enum class JobPriority : std::uint8_t { + High = 0, + Normal = 1, + Low = 2, +}; + +constexpr int jobPriorityCount = 3; + +struct JobStats { + std::uint64_t submitted = 0; + std::uint64_t executed = 0; + std::uint64_t failed = 0; + std::uint64_t queuePops = 0; + std::uint32_t peakBusyWorkers = 0; + std::uint32_t peakQueueDepth = 0; +}; + +// Completion token for one submitted job. Copyable and safe to wait on from any +// thread; an exception thrown by the job body is captured here instead of +// escaping into the worker thread. +class JobHandle { +public: + JobHandle() = default; + + bool valid() const { return static_cast(state); } + bool done() const; + bool failed() const; + void wait() const; + std::exception_ptr error() const; + std::string errorMessage() const; + void rethrowError() const; + +private: + friend class JobSystem; + + struct State; + + explicit JobHandle(std::shared_ptr shared) : state(std::move(shared)) {} + + std::shared_ptr state; +}; + +// A group of handles waited on as a unit. A batch returned for a graph that +// could not be scheduled reports valid() == false and holds no handles. +class JobBatch { +public: + JobBatch() = default; + + bool valid() const { return runnable; } + std::size_t size() const { return handles.size(); } + bool empty() const { return handles.empty(); } + const JobHandle& at(std::size_t index) const { return handles[index]; } + + void add(const JobHandle& handle) { handles.push_back(handle); } + void wait() const; + bool done() const; + std::size_t failedCount() const; + std::exception_ptr firstError() const; + +private: + friend class JobSystem; + + std::vector handles; + bool runnable = true; +}; + +struct JobNode { + std::string name; + std::function work; + JobPriority priority = JobPriority::Normal; + std::vector dependencies; +}; + +// Dependency graph of jobs. A node becomes runnable only once every node it +// depends on has finished. The graph is inert on its own: JobSystem::runGraph +// turns it into scheduled work. +class JobGraph { +public: + std::size_t addNode(std::string name, std::function work, + JobPriority priority = JobPriority::Normal); + void addDependency(std::size_t node, std::size_t dependsOn); + + std::size_t nodeCount() const { return nodes.size(); } + const JobNode& nodeAt(std::size_t index) const { return nodes[index]; } + + // Kahn sweep in ascending node order, so the order is a function of the + // graph alone. Returns fewer entries than nodeCount() when a cycle exists. + std::vector topologicalOrder() const; + bool hasCycle() const; + + void clear() { nodes.clear(); } + +private: + std::vector nodes; +}; + +// Thread pool with priority queues and dependency-aware graph execution. +// A thread count of zero means serial mode: every job runs inline on the +// calling thread at submit time, which is the reference behaviour the parallel +// modes are checked against. +class JobSystem { +public: + explicit JobSystem(unsigned threadCount = 0); + ~JobSystem(); + + JobSystem(const JobSystem&) = delete; + JobSystem& operator=(const JobSystem&) = delete; + + unsigned threadCount() const { return static_cast(workers.size()); } + bool serialMode() const { return workers.empty(); } + + JobHandle submit(std::function work, JobPriority priority = JobPriority::Normal); + JobBatch submitBatch(std::vector> work, + JobPriority priority = JobPriority::Normal); + + // Schedules every node of the graph, respecting dependencies. A cyclic + // graph is refused: the returned batch is invalid and nothing runs. When a + // node throws, its dependents are not executed and complete with an error + // of their own, so waiting on the batch can never hang. + JobBatch runGraph(const JobGraph& graph); + + void waitAll(); + + JobStats stats() const; + void resetStats(); + +private: + struct QueuedJob { + std::function work; + std::shared_ptr state; + }; + + struct GraphRun; + + void workerLoop(); + void enqueue(std::function work, std::shared_ptr state, + JobPriority priority); + void runInline(std::function work, const std::shared_ptr& state); + void scheduleGraphNode(const std::shared_ptr& run, std::size_t index); + bool executeGraphNode(const std::shared_ptr& run, std::size_t index); + void noteInlineStart(); + void noteInlineFinish(bool failed); + void noteFailure(); + std::size_t queuedLocked() const; + QueuedJob popLocked(); + + static void completeState(const std::shared_ptr& state, + std::exception_ptr error); + + mutable std::mutex mutex; + std::condition_variable workCv; + std::condition_variable idleCv; + std::vector workers; + std::deque queues[jobPriorityCount]; + JobStats counters; + unsigned busy = 0; + bool stopping = false; +}; + +} // namespace vf