diff --git a/patches/metal--deps--indium.patch b/patches/metal--deps--indium.patch index 63b9a2c..09cd9a4 100644 --- a/patches/metal--deps--indium.patch +++ b/patches/metal--deps--indium.patch @@ -2447,7 +2447,7 @@ index 43f4a84..aaa1acb 100644 + if ((calls.fetch_add(1,std::memory_order_relaxed) & 63)==0) + bindingReuse.store(access("/tmp/metal-no-binding-reuse",F_OK)!=0,std::memory_order_relaxed); +} -+// OFF by default (round-3 behaviour: a presenting commit waits for the whole worker queue with drain()). Guest flag file ++// Frame-end drain waits for a snapshot of queued work. The opt-in guest flag file +// /tmp/metal-present-dispatch-wait (checked once per process) makes the frame-end sync narrower: the caller waits only +// until this commit's own task has run (its present dispatch, and by FIFO order every replay queued before it), not for +// tasks queued after it. Round 4 had this on by default; that round regressed in matches, so it is opt-in now. @@ -2487,7 +2487,7 @@ index 43f4a84..aaa1acb 100644 + token.wait("presentDispatchWait"); + return; + } -+ // Default (round 3): post first, then check the flag file and wait for the whole worker queue. ++ // Post first, then wait for the work already queued; later producers cannot extend this wait. + SubmitThread::post([self]() { + AsyncEncode::PoolScope pool; + self->commit(); @@ -8050,7 +8050,7 @@ diff --git a/private-include/indium/submit-thread.private.hpp b/private-include/ new file mode 100644 --- /dev/null +++ b/private-include/indium/submit-thread.private.hpp -@@ -0,0 +1,159 @@ +@@ -0,0 +1,175 @@ +#pragma once + +// Ordered queue-submission worker. @@ -8067,6 +8067,8 @@ new file mode 100644 + +#include +#include ++#include ++#include +#include +#include +#include @@ -8080,6 +8082,9 @@ new file mode 100644 + std::condition_variable idle; + std::deque> queue; + size_t active = 0; ++ uint64_t posted = 0, completed = 0; ++ size_t drainWaiters = 0; ++ uint64_t drainTarget = std::numeric_limits::max(); + bool started = false; + const bool enabled = access("/tmp/metal-submit-thread", F_OK) == 0; + }; @@ -8095,8 +8100,10 @@ new file mode 100644 + { + std::lock_guard lock(mutex); + done = true; ++ // The waiter may destroy this stack token as soon as it sees done. ++ // Finish notifying while the token's mutex still protects its lifetime. ++ condvar.notify_all(); + } -+ condvar.notify_all(); + } + // blocks until the task carrying this token has finished (counting the block under `site`) + void wait(const char* site) { @@ -8133,7 +8140,8 @@ new file mode 100644 + { + std::lock_guard lock(s.mutex); + --s.active; -+ if (s.queue.empty() && s.active == 0) s.idle.notify_all(); ++ ++s.completed; ++ if (s.drainWaiters && s.completed >= s.drainTarget) s.idle.notify_all(); + } + } + } @@ -8158,6 +8166,7 @@ new file mode 100644 + // mutex as the worker's wait to avoid losing that transition. + needsWake = s.active == 0 && s.queue.empty(); + s.queue.push_back(std::move(task)); ++ ++s.posted; + } + if (needsWake) s.wake.notify_one(); + } @@ -8190,6 +8199,7 @@ new file mode 100644 + // mutex as the worker's wait to avoid losing that transition. + needsWake = s.active == 0 && s.queue.empty(); + s.queue.push_back(std::move(task)); ++ ++s.posted; + } + if (needsWake) s.wake.notify_one(); + } @@ -8206,7 +8216,13 @@ new file mode 100644 + auto& s = state(); + if (onWorker()) return; + std::unique_lock lock(s.mutex); -+ Waits::blockOn(site, s.idle, lock, [&] { return s.queue.empty() && s.active == 0; }); ++ const uint64_t target = s.posted; ++ if (s.completed >= target) return; ++ ++s.drainWaiters; ++ if (target < s.drainTarget) s.drainTarget = target; ++ // FIFO completion is sufficient: later producers must not extend this wait. ++ Waits::blockOn(site, s.idle, lock, [&] { return s.completed >= target; }); ++ if (--s.drainWaiters == 0) s.drainTarget = std::numeric_limits::max(); + } + } +} diff --git a/tools/run-deferred-handoff-tests.sh b/tools/run-deferred-handoff-tests.sh index 431832e..2928e10 100755 --- a/tools/run-deferred-handoff-tests.sh +++ b/tools/run-deferred-handoff-tests.sh @@ -3,8 +3,9 @@ # # tools/test-deferred-handoff.cpp the ordered worker's FIFO, the present-dispatch token, the wait counters # tools/test-deferred-encoder-model.cpp the deferred blit/render op hand-off: order, resource lifetime, drop safety +# tools/test-submit-drain.cpp overlapping waits complete despite blocked later producers # -# Both compile the real headers from deps/indium/private-include with the host compiler. The counter test writes its +# These compile the real headers from deps/indium/private-include with the host compiler. The counter test writes its # summary to /tmp/opencode/waits-test.log (DARLING_WAITS_LOG) and needs /tmp/metal-debug-waits for the length of the # run; it removes the flag again if it created it (and first checks, in a forked child, that the counters stay off and # write nothing without it). @@ -17,7 +18,7 @@ project_dir=$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd) out_dir=${RIOT_TEST_OUT:-/tmp/opencode} include="$project_dir/build/darling-src/src/external/metal/deps/indium/private-include" mkdir -p "$out_dir" -for name in test-deferred-handoff test-deferred-encoder-model; do +for name in test-deferred-handoff test-deferred-encoder-model test-submit-drain; do c++ -std=c++17 -O1 -pthread -I "$include" "$project_dir/tools/$name.cpp" -o "$out_dir/$name" echo "== $name" "$out_dir/$name" diff --git a/tools/test-submit-drain.cpp b/tools/test-submit-drain.cpp new file mode 100644 index 0000000..2706a7e --- /dev/null +++ b/tools/test-submit-drain.cpp @@ -0,0 +1,66 @@ +// Exercise the real worker with overlapping drains and producers. A frame's +// drain must finish even while later work is blocked indefinitely. +#include +#include +#include +#include +#include +#include +#include + +using namespace std::chrono_literals; +using namespace Indium; + +static void awaitDrains(size_t count) { + const auto deadline = std::chrono::steady_clock::now() + 5s; + for (;;) { + { + auto& s = SubmitThread::state(); + std::lock_guard lock(s.mutex); + if (s.drainWaiters == count) return; + } + assert(std::chrono::steady_clock::now() < deadline); + std::this_thread::yield(); + } +} + +int main() { + std::promise releaseFirst, releaseSecond, releaseThird; + auto first = releaseFirst.get_future().share(); + auto second = releaseSecond.get_future().share(); + auto third = releaseThird.get_future().share(); + std::atomic completed{0}; + SubmitThread::post([&] { first.wait(); ++completed; }); + auto drain1 = std::async(std::launch::async, [] { SubmitThread::drain(); }); + awaitDrains(1); // drain1 has taken its snapshot before the next post + SubmitThread::post([&] { second.wait(); ++completed; }); + auto drain2 = std::async(std::launch::async, [] { SubmitThread::drain(); }); + awaitDrains(2); + SubmitThread::post([&] { third.wait(); ++completed; }); + releaseFirst.set_value(); + assert(drain1.wait_for(2s) == std::future_status::ready); + drain1.get(); + assert(completed == 1); + assert(drain2.wait_for(0s) == std::future_status::timeout); + releaseSecond.set_value(); + assert(drain2.wait_for(2s) == std::future_status::ready); + drain2.get(); + assert(completed == 2); + assert(SubmitThread::pending()); + releaseThird.set_value(); + SubmitThread::drain(); + assert(completed == 3); + // The worker itself cannot wait for a snapshot containing its active task. + SubmitThread::post([] { SubmitThread::drain(); }); + SubmitThread::drain(); + // Repeated idle/active transitions exercise notification registration races. + for (unsigned i=0; i<100000; ++i) { + SubmitThread::post([&] { ++completed; }); + SubmitThread::drain(); + } + assert(completed == 100003); + std::puts("PASS submit drain: overlapping snapshots, blocked later producer, reentrant drain, 100000 handoffs"); + std::fflush(stdout); + // The production worker is detached and process-lived. + std::_Exit(0); +}