npu: a worker the audio pool cannot start is an exception, not std::terminate (onto #2035)

npu/audio_parallel.h

@@ -33,6 +33,7 @@
 #include <cstdlib>
 #include <exception>
 #include <thread>
+#include <utility>
 #include <vector>
 
 namespace onebit::npu::pack_import {
@@ -49,30 +50,53 @@ inline size_t audio_thread_count() {
     return hw == 0 ? 1 : (hw > 8 ? 8 : hw);
 }
 
+// How a worker is started. It is a parameter so a test can make the Nth start fail: std::thread's constructor throws
+// std::system_error when the OS refuses a thread (EAGAIN at RLIMIT_NPROC, which a box running many lanes under one user
+// can reach), and that failure cannot be arranged deterministically from a test any other way.
+struct StdThreadSpawn {
+    template <typename F>
+    std::thread operator()(F&& f) const { return std::thread(std::forward<F>(f)); }
+};
+
 // Run fn(0) .. fn(units-1), each exactly once, on at most `want` threads. Worker t takes units t, t+want, t+2*want, ...
-template <typename Fn>
-void parallel_units(size_t units, size_t want, Fn&& fn) {
+//
+// A worker that cannot be STARTED is handled like a unit that throws (engine#2035 review): the workers already running
+// are joined and the start's exception is rethrown on the caller. Without the join, the unwind destroys `pool` while it
+// still holds joinable threads, and a joinable std::thread's destructor calls std::terminate. The units of the workers
+// that never started do not run, which is why this is an exception and not a quiet fallback to fewer threads.
+template <typename Fn, typename Spawn>
+void parallel_units_spawn(size_t units, size_t want, Fn&& fn, Spawn&& spawn) {
     if (want > units) want = units;
     if (want <= 1) {
         for (size_t u = 0; u < units; ++u) fn(u);
         return;
     }
     std::vector<std::exception_ptr> failed(want);
     std::vector<std::thread> pool;
-    pool.reserve(want);
-    for (size_t t = 0; t < want; ++t)
-        pool.emplace_back([&, t]() {
-            try {
-                for (size_t u = t; u < units; u += want) fn(u);
-            } catch (...) {
-                failed[t] = std::current_exception();
-            }
-        });
+    pool.reserve(want);   // no reallocation below, so a start failure cannot also lose a started thread
+    try {
+        for (size_t t = 0; t < want; ++t)
+            pool.push_back(spawn([&, t]() {
+                try {
+                    for (size_t u = t; u < units; u += want) fn(u);
+                } catch (...) {
+                    failed[t] = std::current_exception();
+                }
+            }));
+    } catch (...) {
+        for (auto& th : pool) th.join();
+        throw;
+    }
     for (auto& th : pool) th.join();
     for (const auto& ep : failed)
         if (ep) std::rethrow_exception(ep);
 }
 
+template <typename Fn>
+void parallel_units(size_t units, size_t want, Fn&& fn) {
+    parallel_units_spawn(units, want, static_cast<Fn&&>(fn), StdThreadSpawn{});
+}
+
 template <typename Fn>
 void parallel_units(size_t units, Fn&& fn) {
     parallel_units(units, audio_thread_count(), static_cast<Fn&&>(fn));

tests/audio_parallel_test.cpp

@@ -29,6 +29,9 @@
 #include <cstring>
 #include <stdexcept>
 #include <string>
+#include <system_error>
+#include <thread>
+#include <utility>
 #include <vector>
 
 #include "audio_exec.h"
@@ -187,12 +190,40 @@ void test_a_throwing_unit() {
     }
 }
 
+// engine#2035 review: a worker that cannot be started (std::thread throws std::system_error at EAGAIN) used to leave the
+// threads started before it joinable while `pool` unwound -- std::terminate. The spawner fails on its Nth call, so the
+// case is deterministic; without the join-then-rethrow the process dies by SIGABRT here.
+void test_a_failed_start() {
+    using onebit::npu::pack_import::parallel_units_spawn;
+    for (size_t fail_at : {size_t(0), size_t(1), size_t(3)}) {
+        const size_t units = 40, want = 4;
+        std::atomic<int> ran{0};
+        size_t calls = 0;
+        auto spawn = [&](auto&& f) {
+            if (calls++ == fail_at) throw std::system_error(std::make_error_code(std::errc::resource_unavailable_try_again));
+            return std::thread(std::forward<decltype(f)>(f));
+        };
+        bool caught = false;
+        try {
+            parallel_units_spawn(units, want, [&](size_t) { ran.fetch_add(1); }, spawn);
+        } catch (const std::system_error& e) {
+            caught = e.code() == std::errc::resource_unavailable_try_again;
+        }
+        check(caught, "a worker that cannot be started (start " + std::to_string(fail_at) +
+                          " of 4) is a std::system_error the caller catches, not std::terminate");
+        // the workers started before the failure ran all their units and were joined before the rethrow
+        check(ran.load() == int(fail_at * (units / want)),
+              "the " + std::to_string(fail_at) + " worker(s) started before the failure finish before the caller sees it");
+    }
+}
+
 }  // namespace
 
 int main() {
     test_bit_identity();
     test_every_unit_once();
     test_a_throwing_unit();
+    test_a_failed_start();
     if (g_failures == 0) std::printf("audio_parallel: ok\n");
     return g_failures == 0 ? 0 : 1;
 }