diff --git a/core/src/JobSystem.cpp b/core/src/JobSystem.cpp index f072136..bf341db 100644 --- a/core/src/JobSystem.cpp +++ b/core/src/JobSystem.cpp @@ -158,3 +158,207 @@ JobSystem::~JobSystem() { for (std::thread& worker : workers) worker.join(); } +void JobSystem::completeState(const std::shared_ptr& state, + std::exception_ptr error) { + { + std::lock_guard lock(state->mutex); + state->error = error; + state->finished = true; + } + state->cv.notify_all(); +} + +std::size_t JobSystem::queuedLocked() const { + std::size_t total = 0; + for (int i = 0; i < jobPriorityCount; ++i) total += queues[i].size(); + return total; +} + +JobSystem::QueuedJob JobSystem::popLocked() { + for (int i = 0; i < jobPriorityCount; ++i) { + if (queues[i].empty()) continue; + QueuedJob job = std::move(queues[i].front()); + queues[i].pop_front(); + return job; + } + return QueuedJob{}; +} + +void JobSystem::enqueue(std::function work, std::shared_ptr state, + JobPriority priority) { + { + std::lock_guard lock(mutex); + ++counters.submitted; + queues[static_cast(priority)].push_back(QueuedJob{std::move(work), std::move(state)}); + std::uint32_t depth = static_cast(queuedLocked()); + if (depth > counters.peakQueueDepth) counters.peakQueueDepth = depth; + } + workCv.notify_one(); +} + +void JobSystem::noteInlineStart() { + std::lock_guard lock(mutex); + ++counters.submitted; + if (counters.peakBusyWorkers < 1u) counters.peakBusyWorkers = 1u; +} + +void JobSystem::noteInlineFinish(bool failed) { + std::lock_guard lock(mutex); + ++counters.executed; + if (failed) ++counters.failed; +} + +void JobSystem::noteFailure() { + std::lock_guard lock(mutex); + ++counters.failed; +} + +void JobSystem::runInline(std::function work, + const std::shared_ptr& state) { + noteInlineStart(); + std::exception_ptr captured; + try { + work(); + } catch (...) { + captured = std::current_exception(); + } + completeState(state, captured); + noteInlineFinish(captured != nullptr); +} + +JobHandle JobSystem::submit(std::function work, JobPriority priority) { + auto state = std::make_shared(); + if (workers.empty()) { + runInline(std::move(work), state); + return JobHandle(state); + } + enqueue(std::move(work), state, priority); + return JobHandle(state); +} + +JobBatch JobSystem::submitBatch(std::vector> work, JobPriority priority) { + JobBatch batch; + for (std::function& job : work) batch.add(submit(std::move(job), priority)); + return batch; +} + +void JobSystem::workerLoop() { + for (;;) { + std::unique_lock lock(mutex); + workCv.wait(lock, [this] { return stopping || queuedLocked() > 0; }); + if (queuedLocked() == 0) { + if (stopping) return; + continue; + } + + QueuedJob job = popLocked(); + ++counters.queuePops; + ++busy; + if (busy > counters.peakBusyWorkers) counters.peakBusyWorkers = busy; + lock.unlock(); + + std::exception_ptr captured; + try { + job.work(); + } catch (...) { + captured = std::current_exception(); + } + if (job.state) completeState(job.state, captured); + + lock.lock(); + --busy; + ++counters.executed; + if (captured) ++counters.failed; + bool idle = busy == 0 && queuedLocked() == 0; + lock.unlock(); + if (idle) idleCv.notify_all(); + } +} + +void JobSystem::waitAll() { + if (workers.empty()) return; + std::unique_lock lock(mutex); + idleCv.wait(lock, [this] { return busy == 0 && queuedLocked() == 0; }); +} + +bool JobSystem::executeGraphNode(const std::shared_ptr& run, std::size_t index) { + std::exception_ptr captured; + if (run->depFailed[index].load(std::memory_order_acquire)) { + captured = std::make_exception_ptr( + std::runtime_error("dependency failed, skipped job " + run->names[index])); + } else { + try { + run->work[index](); + } catch (...) { + captured = std::current_exception(); + } + } + completeState(run->states[index], captured); + + for (std::size_t dependent : run->dependents[index]) { + if (captured) run->depFailed[dependent].store(true, std::memory_order_release); + if (run->remaining[dependent].fetch_sub(1, std::memory_order_acq_rel) == 1 && !run->serial) + scheduleGraphNode(run, dependent); + } + return captured != nullptr; +} + +void JobSystem::scheduleGraphNode(const std::shared_ptr& run, std::size_t index) { + JobPriority priority = run->priorities[index]; + enqueue([this, run, index] { + if (executeGraphNode(run, index)) noteFailure(); + }, + nullptr, priority); +} + +JobBatch JobSystem::runGraph(const JobGraph& graph) { + JobBatch batch; + std::size_t count = graph.nodeCount(); + std::vector order = graph.topologicalOrder(); + if (order.size() != count) { + batch.runnable = false; + return batch; + } + + auto run = std::make_shared(count); + run->serial = workers.empty(); + for (std::size_t i = 0; i < count; ++i) { + const JobNode& node = graph.nodeAt(i); + run->work[i] = node.work; + run->names[i] = node.name; + run->priorities[i] = node.priority; + run->remaining[i].store(static_cast(node.dependencies.size()), + std::memory_order_relaxed); + run->states[i] = std::make_shared(); + batch.add(JobHandle(run->states[i])); + } + for (std::size_t i = 0; i < count; ++i) + for (std::size_t dep : graph.nodeAt(i).dependencies) run->dependents[dep].push_back(i); + + if (run->serial) { + for (std::size_t index : order) { + noteInlineStart(); + bool failed = executeGraphNode(run, index); + noteInlineFinish(failed); + } + return batch; + } + + std::vector roots; + for (std::size_t i = 0; i < count; ++i) + if (run->remaining[i].load(std::memory_order_relaxed) == 0) roots.push_back(i); + for (std::size_t index : roots) scheduleGraphNode(run, index); + return batch; +} + +JobStats JobSystem::stats() const { + std::lock_guard lock(mutex); + return counters; +} + +void JobSystem::resetStats() { + std::lock_guard lock(mutex); + counters = JobStats{}; +} + +} // namespace vf