From 9d36fcd51b64d3f3309ca9df201261fac07dc921 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Sun, 26 Jul 2026 01:13:02 +0000 Subject: [PATCH] backend(audio-cpp): serialize runs with a wedge-aware guard audio.cpp sessions are not reentrant and a wedged CUDA call cannot be cancelled, so a plain mutex would pile every worker thread behind a stuck GPU. Callers waiting past the configured bound, or arriving while the holder has already overrun it, fail fast instead. A caller that queues behind a healthy run deliberately does not stamp the clock: only the thread that takes the lock does. Stamping on arrival would restart the wedge clock on every request and hide a stuck run from everyone behind it, which is the pile-up this guard exists to prevent. Assisted-by: Claude:claude-opus-5 [Claude Code] Signed-off-by: Ettore Di Giacinto --- backend/cpp/audio-cpp/CMakeLists.txt | 1 + backend/cpp/audio-cpp/run_guard.cpp | 70 +++++ backend/cpp/audio-cpp/run_guard.h | 76 +++++ backend/cpp/audio-cpp/run_guard_test.cpp | 364 +++++++++++++++++++++++ 4 files changed, 511 insertions(+) create mode 100644 backend/cpp/audio-cpp/run_guard.cpp create mode 100644 backend/cpp/audio-cpp/run_guard.h create mode 100644 backend/cpp/audio-cpp/run_guard_test.cpp diff --git a/backend/cpp/audio-cpp/CMakeLists.txt b/backend/cpp/audio-cpp/CMakeLists.txt index d2b6a7f57..f630f242f 100644 --- a/backend/cpp/audio-cpp/CMakeLists.txt +++ b/backend/cpp/audio-cpp/CMakeLists.txt @@ -90,6 +90,7 @@ add_executable(${TARGET} capability_routing.cpp audio_units.cpp transcript_assembly.cpp + run_guard.cpp ) target_include_directories(${TARGET} PRIVATE diff --git a/backend/cpp/audio-cpp/run_guard.cpp b/backend/cpp/audio-cpp/run_guard.cpp new file mode 100644 index 000000000..5cffdc28b --- /dev/null +++ b/backend/cpp/audio-cpp/run_guard.cpp @@ -0,0 +1,70 @@ +#include "run_guard.h" + +#include + +namespace audiocpp_backend { + +std::int64_t steady_now_ms() { + return std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count(); +} + +bool run_has_overrun(std::int64_t busy_since_ms, std::int64_t now_ms, + int timeout_ms) { + if (timeout_ms <= 0 || busy_since_ms == 0) { + return false; + } + return (now_ms - busy_since_ms) > timeout_ms; +} + +int resolve_busy_timeout_ms(int ceiling, int requested) { + if (requested <= 0) { + return ceiling; + } + if (ceiling <= 0) { + return requested; + } + return requested < ceiling ? requested : ceiling; +} + +RunGuard::Lock::Lock(RunGuard &guard, std::unique_lock lock) + : guard_(&guard), lock_(std::move(lock)) {} + +RunGuard::Lock::~Lock() { + if (guard_ != nullptr && lock_.owns_lock()) { + guard_->busy_since_ms_.store(0, std::memory_order_release); + } +} + +RunGuard::Lock RunGuard::acquire(int timeout_ms, const std::string &label) { + std::unique_lock lock(mutex_, std::defer_lock); + if (timeout_ms <= 0) { + lock.lock(); // guard disabled: unbounded wait + } else { + // If the current holder has already run past the bound, treat it as + // wedged and fail fast rather than queue behind it. This is what stops + // requests piling up on a stuck GPU. Nothing is stored here: only the + // thread that actually takes the lock stamps the clock, so a rejected + // caller cannot hide the wedge from the callers after it. + const auto since = busy_since_ms_.load(std::memory_order_acquire); + const auto now_ms = steady_now_ms(); + if (run_has_overrun(since, now_ms, timeout_ms)) { + throw BusyError("audio-cpp: model '" + label + + "' is busy: the current inference has run " + + std::to_string(now_ms - since) + + " ms (busy_timeout_ms=" + std::to_string(timeout_ms) + + "); it has likely wedged and cannot be cancelled"); + } + if (!lock.try_lock_for(std::chrono::milliseconds(timeout_ms))) { + throw BusyError("audio-cpp: model '" + label + + "' is busy: timed out after " + + std::to_string(timeout_ms) + + " ms waiting for the inference lock"); + } + } + busy_since_ms_.store(steady_now_ms(), std::memory_order_release); + return Lock(*this, std::move(lock)); +} + +} // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/run_guard.h b/backend/cpp/audio-cpp/run_guard.h new file mode 100644 index 000000000..1882925be --- /dev/null +++ b/backend/cpp/audio-cpp/run_guard.h @@ -0,0 +1,76 @@ +#pragma once + +// Serializes runs on one loaded model and bounds how long a caller waits. +// +// A single audio.cpp model runs one request at a time; its sessions are not +// reentrant. If an inference wedges the GPU, a CUDA call that never returns +// cannot be cancelled from userspace, so later requests would block forever +// and every gRPC worker thread would pile up behind it. With a positive +// timeout a caller instead gives up and throws BusyError, which grpc-server.cpp +// maps to UNAVAILABLE so the client can retry or fail over. A timeout of 0 +// restores unbounded std::mutex behaviour. +// +// This mirrors the behaviour upstream documents in app/server/busy_guard.h. +// That header is deliberately not copied: it is outside the public include +// path, and it is Apache-2.0 while this repository is MIT. + +#include +#include +#include +#include +#include + +namespace audiocpp_backend { + +class BusyError : public std::runtime_error { +public: + using std::runtime_error::runtime_error; +}; + +std::int64_t steady_now_ms(); + +// True when a caller should fail fast instead of queueing: the current holder +// acquired the guard at busy_since_ms and has already run longer than +// timeout_ms. busy_since_ms == 0 means idle; timeout_ms <= 0 disables the guard. +bool run_has_overrun(std::int64_t busy_since_ms, std::int64_t now_ms, + int timeout_ms); + +// Resolves the bound for one run. `ceiling` is the model's configured policy; +// `requested` is a per-request override where a value <= 0 means unset. A +// request may ask for a shorter bound than policy but never a longer one, so a +// client cannot weaken the guard. A ceiling of 0 means unbounded and therefore +// compares as infinity. +int resolve_busy_timeout_ms(int ceiling, int requested); + +class RunGuard { +public: + // Holds the guard for the duration of a run and marks it idle again on + // release, including when the run throws, so the fail-fast check never + // sees a stale timestamp. Move only. + class Lock { + public: + Lock() = default; + Lock(RunGuard &guard, std::unique_lock lock); + Lock(Lock &&) = default; + Lock &operator=(Lock &&) = default; + Lock(const Lock &) = delete; + Lock &operator=(const Lock &) = delete; + ~Lock(); + + private: + RunGuard *guard_ = nullptr; + std::unique_lock lock_; + }; + + // `label` identifies the model in the error message. Throws BusyError when + // timeout_ms is positive and either the current holder has already overrun + // (fail fast, no wait) or the wait itself times out. + Lock acquire(int timeout_ms, const std::string &label); + +private: + friend class Lock; + std::timed_mutex mutex_; + std::atomic busy_since_ms_{0}; +}; + +} // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/run_guard_test.cpp b/backend/cpp/audio-cpp/run_guard_test.cpp new file mode 100644 index 000000000..06e7ffe10 --- /dev/null +++ b/backend/cpp/audio-cpp/run_guard_test.cpp @@ -0,0 +1,364 @@ +// Unit tests for run_guard. Standard library only. The harness compiles this +// as a single translation unit, so the implementation is included directly. +// +// Every test that can block waits through AsyncAcquire, which bounds the wait +// and reports a named failure. A guard that never returns must fail this suite, +// not hang it: a hung test wedges CI far longer than a red one. + +#include "run_guard.cpp" + +#include +#include +#include +#include +#include +#include +#include + +static int failures = 0; + +static void check(bool ok, const std::string &name) { + if (!ok) { + failures++; + fprintf(stderr, "FAIL: %s\n", name.c_str()); + } else { + fprintf(stderr, "ok: %s\n", name.c_str()); + } +} + +using namespace audiocpp_backend; + +namespace { + +struct Outcome { + bool threw = false; + std::string message; +}; + +// Runs one acquire on a worker thread so that a guard which waits forever fails +// a named check instead of hanging the suite. The worker is joined by the +// future's destructor, so every test must first unblock whatever holds the +// guard before this object goes out of scope. +class AsyncAcquire { +public: + AsyncAcquire(RunGuard &guard, int timeout_ms, const std::string &label) + : future_(std::async(std::launch::async, [&guard, timeout_ms, label] { + Outcome out; + try { + auto lock = guard.acquire(timeout_ms, label); + (void)lock; + } catch (const BusyError &err) { + out.threw = true; + out.message = err.what(); + } + return out; + })) {} + + // True when the acquire returned within watchdog_ms. + bool returned_within(int watchdog_ms) { + return future_.wait_for(std::chrono::milliseconds(watchdog_ms)) == + std::future_status::ready; + } + + // Only valid once returned_within() reported true, or after the holder has + // been released; otherwise it blocks. + Outcome outcome() { return future_.get(); } + +private: + std::future future_; +}; + +// A run held open on its own thread until release() is called, so the main +// thread can observe a guard that is genuinely busy. +class HeldRun { +public: + explicit HeldRun(RunGuard &guard) { + thread_ = std::thread([this, &guard] { + auto lock = guard.acquire(0, "holder"); + held_.store(true); + while (!release_.load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + }); + while (!held_.load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + } + + void release() { + release_.store(true); + if (thread_.joinable()) { + thread_.join(); + } + } + + ~HeldRun() { release(); } + +private: + std::thread thread_; + std::atomic held_{false}; + std::atomic release_{false}; +}; + +const int kWatchdogMs = 5000; + +} // namespace + +static void test_steady_clock_advances() { + const std::int64_t before = steady_now_ms(); + std::this_thread::sleep_for(std::chrono::milliseconds(30)); + const std::int64_t after = steady_now_ms(); + check(after >= before + 10, "steady_now_ms advances with real time"); +} + +static void test_overrun_predicate() { + // busy_since_ms == 0 means idle: never an overrun. + check(!run_has_overrun(0, 100000, 1000), "idle is never an overrun"); + // timeout_ms <= 0 disables the guard. + check(!run_has_overrun(1000, 100000, 0), "zero timeout disables the check"); + check(!run_has_overrun(1000, 100000, -5), "negative timeout disables the check"); + // Held for 500ms under a 1000ms bound: fine. + check(!run_has_overrun(1000, 1500, 1000), "within the bound is not an overrun"); + // Exactly at the bound is not yet an overrun. + check(!run_has_overrun(1000, 2000, 1000), "exactly at the bound is not an overrun"); + // One millisecond past the bound is. + check(run_has_overrun(1000, 2001, 1000), "past the bound is an overrun"); + check(run_has_overrun(1000, 100000, 1000), "long past the bound is an overrun"); +} + +static void test_timeout_resolution() { + // No per-request value: policy applies. + check(resolve_busy_timeout_ms(5000, 0) == 5000, "unset request uses the ceiling"); + check(resolve_busy_timeout_ms(5000, -1) == 5000, "negative request uses the ceiling"); + // A request may ask for a shorter bound than policy. + check(resolve_busy_timeout_ms(5000, 1000) == 1000, "shorter request is honoured"); + // A request may never ask for a longer bound than policy. + check(resolve_busy_timeout_ms(1000, 5000) == 1000, "longer request is capped"); + // Equal on both sides is that value, not something else. + check(resolve_busy_timeout_ms(1000, 1000) == 1000, "an equal request is that value"); + // Unbounded policy honours whatever the request asks for. + check(resolve_busy_timeout_ms(0, 2000) == 2000, "unbounded policy honours the request"); + check(resolve_busy_timeout_ms(0, 0) == 0, "unbounded on both sides stays unbounded"); +} + +static void test_uncontended_acquire() { + RunGuard guard; + bool threw = false; + try { + auto lock = guard.acquire(1000, "model"); + (void)lock; + } catch (const BusyError &) { + threw = true; + } + check(!threw, "an uncontended acquire succeeds"); +} + +static void test_release_clears_the_busy_timestamp() { + RunGuard guard; + { + auto lock = guard.acquire(1000, "model"); + (void)lock; + } + // Sleep well past the bound the next caller will use. Releasing must mark + // the guard idle again: a stale timestamp would make this caller fail fast + // against a model that has not been running anything for 200ms. + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + + bool threw = false; + std::string message; + try { + auto lock = guard.acquire(20, "model"); + (void)lock; + } catch (const BusyError &err) { + threw = true; + message = err.what(); + } + check(!threw, "release clears the busy timestamp"); + if (threw) { + fprintf(stderr, " unexpected BusyError: %s\n", message.c_str()); + } +} + +static void test_exception_in_run_still_releases() { + RunGuard guard; + try { + auto lock = guard.acquire(1000, "model"); + (void)lock; + throw std::runtime_error("inference blew up"); + } catch (const std::runtime_error &) { + // swallowed + } + // Same reasoning as above: the unwind must clear the timestamp, not just + // drop the mutex, or the next caller fails fast against an idle model. + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + + bool threw = false; + try { + auto lock = guard.acquire(20, "model"); + (void)lock; + } catch (const BusyError &) { + threw = true; + } + check(!threw, "a throwing run still releases the guard"); +} + +static void test_contended_acquire_times_out() { + RunGuard guard; + HeldRun holder(guard); + + // 300ms is generous on purpose: the waiter must still be inside its bound + // when it looks, so that this exercises the wait path and not the fail-fast + // path. Only a 300ms stall between the holder acquiring and the waiter + // checking would confuse the two. + AsyncAcquire waiter(guard, 300, "my-model"); + const bool returned = waiter.returned_within(kWatchdogMs); + check(returned, "a contended acquire returns rather than waiting forever"); + + if (returned) { + const Outcome out = waiter.outcome(); + check(out.threw, "a contended acquire past its timeout throws BusyError"); + check(out.message.find("my-model") != std::string::npos, + "the error names the model"); + check(out.message.find("timed out after") != std::string::npos, + "waiting out the bound reports the wait, not a wedge"); + } + + holder.release(); + + // After the holder releases, the guard is usable again. + bool after_threw = false; + try { + auto lock = guard.acquire(1000, "my-model"); + (void)lock; + } catch (const BusyError &) { + after_threw = true; + } + check(!after_threw, "the guard is reusable once released"); +} + +static void test_fails_fast_when_the_holder_has_overrun() { + RunGuard guard; + HeldRun holder(guard); + + // Let the holder run past every bound the callers below use, so each of + // them must fail fast on the holder's own start time instead of queueing. + std::this_thread::sleep_for(std::chrono::milliseconds(250)); + + AsyncAcquire first(guard, 40, "wedged-model"); + const bool first_returned = first.returned_within(kWatchdogMs); + check(first_returned, "a caller meeting a wedged run returns"); + if (first_returned) { + const Outcome out = first.outcome(); + check(out.threw, "a caller meeting a wedged run throws BusyError"); + check(out.message.find("has run") != std::string::npos, + "the error reports the wedged run, not a plain wait timeout"); + check(out.message.find("wedged-model") != std::string::npos, + "the fail-fast error names the model"); + } + + // A second caller must see the same thing. A waiter that stamps its own + // arrival time over the holder's would hide the wedge from everyone after + // it, which is exactly the pile-up this guard exists to prevent. + AsyncAcquire second(guard, 40, "wedged-model"); + const bool second_returned = second.returned_within(kWatchdogMs); + check(second_returned, "a second caller meeting a wedged run returns"); + if (second_returned) { + const Outcome out = second.outcome(); + check(out.threw, "a second caller meeting a wedged run throws BusyError"); + check(out.message.find("has run") != std::string::npos, + "a rejected waiter does not reset the wedge clock"); + } + + holder.release(); +} + +// A caller that queues behind a healthy run must not stamp its own arrival over +// the holder's start time. If it did, the wedge clock would restart on every +// arrival and a run that never finishes would look young forever to everyone +// who arrives after: the pile-up this guard exists to prevent, back again. +static void test_a_queued_waiter_does_not_reset_the_wedge_clock() { + RunGuard guard; + HeldRun holder(guard); + + // 300ms in, a caller with a 5s bound is nowhere near its own limit, so it + // queues rather than failing fast. That is the caller that could clobber + // the clock. + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + AsyncAcquire queued(guard, 5000, "patient-model"); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + // 400ms in, this caller's own 250ms bound is already exceeded by the + // holder, so it must fail fast. It only can if the clock still reads the + // holder's start: 400ms of holding, not the 100ms since the queued caller + // arrived. + AsyncAcquire impatient(guard, 250, "wedged-model"); + const bool returned = impatient.returned_within(kWatchdogMs); + check(returned, "a caller arriving behind a queued waiter returns"); + if (returned) { + const Outcome out = impatient.outcome(); + check(out.threw, "a caller arriving behind a queued waiter throws BusyError"); + check(out.message.find("has run") != std::string::npos, + "a queued waiter does not reset the wedge clock"); + } + + holder.release(); + + const bool queued_returned = queued.returned_within(kWatchdogMs); + check(queued_returned, "the queued waiter returns once the holder releases"); + if (queued_returned) { + check(!queued.outcome().threw, "the queued waiter gets its turn"); + } +} + +static void test_runs_are_serialized() { + RunGuard guard; + std::atomic concurrent{0}; + std::atomic peak{0}; + std::atomic unexpected_busy{false}; + + std::vector threads; + for (int t = 0; t < 4; ++t) { + threads.emplace_back([&] { + for (int i = 0; i < 40; ++i) { + try { + auto lock = guard.acquire(0, "model"); + const int inside = ++concurrent; + int seen = peak.load(); + while (inside > seen && + !peak.compare_exchange_weak(seen, inside)) { + } + std::this_thread::sleep_for(std::chrono::microseconds(20)); + --concurrent; + } catch (const BusyError &) { + unexpected_busy.store(true); + return; + } + } + }); + } + for (auto &thread : threads) { + thread.join(); + } + + check(!unexpected_busy.load(), "an unbounded acquire never gives up"); + check(peak.load() == 1, "only one run holds the guard at a time"); +} + +int main() { + test_steady_clock_advances(); + test_overrun_predicate(); + test_timeout_resolution(); + test_uncontended_acquire(); + test_release_clears_the_busy_timestamp(); + test_exception_in_run_still_releases(); + test_contended_acquire_times_out(); + test_fails_fast_when_the_holder_has_overrun(); + test_a_queued_waiter_does_not_reset_the_wedge_clock(); + test_runs_are_serialized(); + if (failures) { + fprintf(stderr, "%d check(s) failed\n", failures); + return 1; + } + fprintf(stderr, "all run_guard checks passed\n"); + return 0; +}