diff --git a/tests/testJobSystem.cpp b/tests/testJobSystem.cpp index d61fea9..d8512a5 100644 --- a/tests/testJobSystem.cpp +++ b/tests/testJobSystem.cpp @@ -154,3 +154,162 @@ VF_TEST(graphNeverStartsNodeBeforeItsDependencies) { } } +// A wide graph: one root, a fan-out layer, and a join node that must observe +// every fan-out node as finished before it starts. +VF_TEST(graphJoinNodeWaitsForTheWholeFanOut) { + for (unsigned threads : {0u, 2u, 6u}) { + JobSystem jobs(threads); + std::atomic liveFanOut{0}; + std::atomic peakLiveFanOut{0}; + std::atomic doneFanOut{0}; + std::atomic joinSawLive{-1}; + std::atomic joinSawDone{-1}; + std::atomic rootDone{0}; + std::atomic fanOutSawRoot{0}; + + JobGraph graph; + std::size_t root = graph.addNode("root", [&] { rootDone.store(1); }); + std::vector fanOut; + for (int i = 0; i < 12; ++i) { + fanOut.push_back(graph.addNode("fan" + std::to_string(i), [&] { + fanOutSawRoot.fetch_add(rootDone.load()); + int live = liveFanOut.fetch_add(1) + 1; + int seen = peakLiveFanOut.load(); + while (live > seen && !peakLiveFanOut.compare_exchange_weak(seen, live)) { + } + liveFanOut.fetch_sub(1); + doneFanOut.fetch_add(1); + })); + } + std::size_t join = graph.addNode("join", [&] { + joinSawLive.store(liveFanOut.load()); + joinSawDone.store(doneFanOut.load()); + }); + for (std::size_t node : fanOut) { + graph.addDependency(node, root); + graph.addDependency(join, node); + } + + JobBatch batch = jobs.runGraph(graph); + batch.wait(); + CHECK_EQ(joinSawLive.load(), 0); + CHECK_EQ(joinSawDone.load(), 12); // the join really ran last + CHECK_EQ(fanOutSawRoot.load(), 12); // and every fan-out node ran after the root + CHECK(peakLiveFanOut.load() >= 1); + if (threads == 0u) CHECK_EQ(peakLiveFanOut.load(), 1); + CHECK_EQ(batch.failedCount(), static_cast(0)); + } +} + +VF_TEST(graphWithCycleIsRejectedInsteadOfHanging) { + for (unsigned threads : {0u, 4u}) { + JobSystem jobs(threads); + std::atomic ran{0}; + JobGraph graph; + for (int i = 0; i < 3; ++i) + graph.addNode("loop" + std::to_string(i), [&ran] { ran.fetch_add(1); }); + graph.addDependency(0, 2); + graph.addDependency(1, 0); + graph.addDependency(2, 1); + + CHECK(graph.hasCycle()); + CHECK_EQ(graph.topologicalOrder().size(), static_cast(0)); + JobBatch batch = jobs.runGraph(graph); + CHECK(!batch.valid()); + CHECK(batch.empty()); + batch.wait(); // returns at once, nothing was scheduled + CHECK_EQ(ran.load(), 0); + } +} + +VF_TEST(selfDependentNodeCountsAsACycle) { + JobSystem jobs(2); + JobGraph graph; + std::atomic ran{0}; + graph.addNode("lonely", [&ran] { ran.fetch_add(1); }); + graph.addDependency(0, 0); + CHECK(graph.hasCycle()); + JobBatch batch = jobs.runGraph(graph); + CHECK(!batch.valid()); + CHECK_EQ(ran.load(), 0); +} + +VF_TEST(jobExceptionTravelsBackThroughItsHandle) { + for (unsigned threads : {0u, 3u}) { + JobSystem jobs(threads); + JobHandle bad = jobs.submit([] { throw std::runtime_error("chunk generator blew up"); }); + bad.wait(); + CHECK(bad.done()); + CHECK(bad.failed()); + CHECK(bad.errorMessage() == std::string("chunk generator blew up")); + + bool rethrown = false; + try { + bad.rethrowError(); + } catch (const std::runtime_error& ex) { + rethrown = std::string(ex.what()) == "chunk generator blew up"; + } + CHECK(rethrown); + + std::atomic after{0}; + JobBatch survivors; + for (int i = 0; i < 32; ++i) + survivors.add(jobs.submit([&after] { after.fetch_add(1); })); + survivors.wait(); + CHECK_EQ(after.load(), 32); + CHECK_EQ(survivors.failedCount(), static_cast(0)); + CHECK_EQ(jobs.stats().failed, static_cast(1)); + } +} + +VF_TEST(graphNodeWithFailedDependencyIsSkippedNotHung) { + for (unsigned threads : {0u, 4u}) { + JobSystem jobs(threads); + std::atomic ranDependent{0}; + std::atomic ranIndependent{0}; + JobGraph graph; + std::size_t broken = graph.addNode("broken", [] { throw std::runtime_error("stage lost"); }); + std::size_t child = graph.addNode("child", [&] { ranDependent.fetch_add(1); }); + std::size_t grandchild = graph.addNode("grandchild", [&] { ranDependent.fetch_add(1); }); + std::size_t sibling = graph.addNode("sibling", [&] { ranIndependent.fetch_add(1); }); + graph.addDependency(child, broken); + graph.addDependency(grandchild, child); + + JobBatch batch = jobs.runGraph(graph); + CHECK(batch.valid()); + batch.wait(); // must not deadlock on the skipped nodes + + CHECK(batch.at(broken).failed()); + CHECK(batch.at(child).failed()); + CHECK(batch.at(grandchild).failed()); + CHECK(!batch.at(sibling).failed()); + CHECK_EQ(ranDependent.load(), 0); + CHECK_EQ(ranIndependent.load(), 1); + CHECK_EQ(batch.failedCount(), static_cast(3)); + CHECK(batch.firstError() != nullptr); + } +} + +VF_TEST(highPriorityJobsAreServedBeforeLowOnesOnOneWorker) { + JobSystem jobs(1); + Latch started; + Latch release; + std::mutex orderMutex; + std::vector order; + + JobHandle blocker = jobs.submit([&] { + started.open(); + release.await(); + }); + started.await(); // the single worker is now occupied + + auto record = [&](int tag) { + return [&, tag] { + std::lock_guard lock(orderMutex); + order.push_back(tag); + }; + }; + JobHandle low = jobs.submit(record(3), JobPriority::Low); + JobHandle normal = jobs.submit(record(2), JobPriority::Normal); + JobHandle high = jobs.submit(record(1), JobPriority::High); +