Bound submit-worker drains to work already queued

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 <indium/submit-thread.private.hpp>
+#include <atomic>
+#include <cassert>
+#include <chrono>
+#include <cstdio>
+#include <future>
+#include <thread>
+
+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<std::mutex> lock(s.mutex);
+            if (s.drainWaiters == count) return;
+        }
+        assert(std::chrono::steady_clock::now() < deadline);
+        std::this_thread::yield();
+    }
+}
+
+int main() {
+    std::promise<void> releaseFirst, releaseSecond, releaseThird;
+    auto first = releaseFirst.get_future().share();
+    auto second = releaseSecond.get_future().share();
+    auto third = releaseThird.get_future().share();
+    std::atomic<unsigned> 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);
+}