diff --git a/npu/audio_parallel.h b/npu/audio_parallel.h index 30ebfc05c..c4173e144 100644 --- a/npu/audio_parallel.h +++ b/npu/audio_parallel.h @@ -33,6 +33,7 @@ #include #include #include +#include #include namespace onebit::npu::pack_import { @@ -49,9 +50,22 @@ 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 + std::thread operator()(F&& f) const { return std::thread(std::forward(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 -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 +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); @@ -59,20 +73,30 @@ void parallel_units(size_t units, size_t want, Fn&& fn) { } std::vector failed(want); std::vector 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 +void parallel_units(size_t units, size_t want, Fn&& fn) { + parallel_units_spawn(units, want, static_cast(fn), StdThreadSpawn{}); +} + template void parallel_units(size_t units, Fn&& fn) { parallel_units(units, audio_thread_count(), static_cast(fn)); diff --git a/tests/audio_parallel_test.cpp b/tests/audio_parallel_test.cpp index 93b71a551..d5dc36df6 100644 --- a/tests/audio_parallel_test.cpp +++ b/tests/audio_parallel_test.cpp @@ -29,6 +29,9 @@ #include #include #include +#include +#include +#include #include #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 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(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; }