mirror/LocalAI
1
0
Fork 0
mirror of https://github.com/mudler/LocalAI.git synced 2026-07-31 02:18:50 -04:00
Code Issues Packages Projects Releases Wiki Activity

backend(audio-cpp): serialize inference through an InferenceLane

Browse Source 
One audio.cpp model is loaded per backend process and its sessions are not
reentrant, so concurrent gRPC handlers have to take turns. Serialization alone
is not enough: a wedged GPU call cannot be cancelled from userspace, so an
unbounded queue behind one stuck run would swallow every gRPC worker thread
until the process is useless.

InferenceLane gives handlers a lane with room for one runner. LaneEntry occupies
it for a scope and gives it back on every exit, including an exception, and is
the only way to take the lane at all: occupy/vacate are private with LaneEntry
as the sole friend, so a caller cannot acquire without holding something that
releases. LaneEntry is immovable on purpose, because a moved-from entry would
have to stop releasing while the lane still recorded it as occupied.

A caller either waits indefinitely or brings a millisecond budget. A bounded
caller that cannot get in fails instead of waiting on, and a bounded caller
whose budget is already shorter than the age of the run in the lane fails
immediately, which is what stops a queue forming behind a wedged run. The two
failures carry different text: one names the wait it exhausted, the other states
the measured age of the run without claiming to know why it is long, since a
short budget meeting a legitimately long run lands there too.

The run's age is stamped only after acquisition. A waiter that published itself
as holder would restart the measurement and hide a genuinely stuck holder from
every caller behind it.

Budget negotiation and the overrun decision are pure functions taking their
inputs explicitly, so both are covered without threads or sleeping. The
per-model ceiling arrives as an int of milliseconds; a request may tighten it
and may never loosen it.

Replaces the previous run_guard unit, which was a derivative of an
Apache-2.0 file upstream and could not stay in an MIT tree. Written from a
behaviour contract with no reference to the removed code.

Tests: 65 checks, standard library only, single translation unit, clean under
-Wall -Wextra. Mutation tested at 23/23 killed; two of those mutants exposed
missing coverage and the tests were extended until they died. ThreadSanitizer
clean.

Assisted-by: Claude:claude-opus-5 [Claude Code]
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
This commit is contained in:
Ettore Di Giacinto
2026-07-26 12:25:32 +00:00
parent 9d36fcd51b
commit 87387d91de
7 changed files with 831 additions and 511 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/cpp/audio-cpp/CMakeLists.txt
View File

@@ -90,7 +90,7 @@ add_executable(${TARGET}
capability_routing.cpp
audio_units.cpp
transcript_assembly.cpp
run_guard.cpp
inference_lane.cpp
)
target_include_directories(${TARGET} PRIVATE

114
backend/cpp/audio-cpp/inference_lane.cpp Normal file
View File

@@ -0,0 +1,114 @@
#include "inference_lane.h"
#include <algorithm>
#include <chrono>
namespace audiocpp_backend {
std::int64_t monotonic_millis() {
return std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::steady_clock::now().time_since_epoch())
.count();
}
int resolve_wait_budget_ms(int policy_ceiling_ms, int request_hint_ms) {
// Both sides normalise to 0 for "not specified", which is also the value
// that means unbounded on the way out, so the unspecified cases fall out of
// the arithmetic instead of needing their own branches.
const int ceiling = policy_ceiling_ms > 0 ? policy_ceiling_ms : 0;
const int hint = request_hint_ms > 0 ? request_hint_ms : 0;
if (ceiling == 0) {
return hint;
}
if (hint == 0) {
return ceiling;
}
// The hint can tighten the ceiling but never loosen it.
return std::min(ceiling, hint);
}
bool run_exceeds_budget(bool lane_occupied, std::int64_t run_started_ms,
std::int64_t now_ms, int budget_ms) {
if (!lane_occupied || budget_ms <= 0) {
return false;
}
return now_ms - run_started_ms > static_cast<std::int64_t>(budget_ms);
}
namespace {
std::string busy_prefix(const std::string &model_label) {
return "inference lane for model '" + model_label + "' is busy: ";
}
// States the measurement and nothing else. A short-budget caller meeting a
// legitimately long run lands here too, so this text must not declare the run
// broken; the numbers let a reader decide that for themselves.
std::string overrun_message(const std::string &model_label,
std::int64_t run_age_ms, int budget_ms) {
return busy_prefix(model_label) + "the in-flight run has been running for " +
std::to_string(run_age_ms) + " ms, longer than this request's " +
std::to_string(budget_ms) + " ms wait budget";
}
std::string wait_exhausted_message(const std::string &model_label,
int budget_ms) {
return busy_prefix(model_label) + "timed out after " +
std::to_string(budget_ms) +
" ms waiting for the in-flight run to finish";
}
} // namespace
void InferenceLane::occupy(int budget_ms) {
std::unique_lock<std::mutex> lock(state_mutex_);
// Checked once, on arrival: if the run already in the lane has outlived what
// this caller brought, no amount of waiting can help it, and queueing here
// is exactly how a wedged run swallows every handler thread.
const std::int64_t arrived_ms = monotonic_millis();
if (run_exceeds_budget(occupied_, run_started_ms_, arrived_ms, budget_ms)) {
throw LaneUnavailable(overrun_message(
model_label_, arrived_ms - run_started_ms_, budget_ms));
}
if (budget_ms > 0) {
if (!vacated_.wait_for(lock, std::chrono::milliseconds(budget_ms),
[this] { return !occupied_; })) {
throw LaneUnavailable(
wait_exhausted_message(model_label_, budget_ms));
}
} else {
vacated_.wait(lock, [this] { return !occupied_; });
}
// Only now, holding both the mutex and the lane. Stamping any earlier would
// restart the age of the run for every caller behind this one and make a
// genuinely wedged holder look freshly started forever.
occupied_ = true;
run_started_ms_ = monotonic_millis();
}
void InferenceLane::vacate() {
{
std::lock_guard<std::mutex> lock(state_mutex_);
occupied_ = false;
}
// notify_all, not notify_one: a waiter that times out concurrently with a
// notification can consume it, and losing the only wakeup would park the
// remaining waiters for the rest of the run's lifetime. Waking all of them
// still admits exactly one, since the rest re-test occupancy under the mutex
// and go back to waiting, and ordering among waiters is not a requirement.
vacated_.notify_all();
}
LaneEntry::LaneEntry(InferenceLane &lane, int budget_ms) : lane_(lane) {
// If this throws, the object never existed, so ~LaneEntry does not run and
// cannot hand back a lane this caller never held.
lane_.occupy(budget_ms);
}
LaneEntry::~LaneEntry() { lane_.vacate(); }
} // namespace audiocpp_backend

118
backend/cpp/audio-cpp/inference_lane.h Normal file
View File

@@ -0,0 +1,118 @@
#pragma once
// Serializes inference against the single audio.cpp model a backend process
// owns. Standard library only.
//
// Why a plain mutex is not enough: an audio.cpp session is not reentrant, so
// concurrent gRPC handlers have to take turns. But a GPU call that wedges cannot
// be cancelled from userspace, so an unbounded queue behind one wedged run would
// swallow every gRPC worker thread until the process is useless. A caller
// therefore needs to be able to walk away, and needs to be able to tell "the
// lane is busy with normal work and I ran out of patience" apart from "the run
// in the lane has already outlived the patience I brought".
#include <condition_variable>
#include <cstdint>
#include <mutex>
#include <stdexcept>
#include <string>
namespace audiocpp_backend {
// Reading off a monotonic clock, in milliseconds. Monotonic on purpose: a wall
// clock adjustment must never make an in-flight run look younger or older than
// it is, because that reading decides whether callers give up.
std::int64_t monotonic_millis();
// Collapses the per-model configured ceiling and the optional per-request hint
// into the wait budget a caller actually gets. Returns 0 for "wait
// indefinitely".
//
// Either input may be non-positive, which means "not specified":
// - an unspecified hint yields the ceiling,
// - an unspecified ceiling means no policy limit, so the hint stands,
// - unspecified on both sides is unbounded.
// A specified hint may only tighten the ceiling. A client asking for a longer
// wait than the model's policy allows does not get it, because that would let a
// request weaken an operator's choice.
int resolve_wait_budget_ms(int policy_ceiling_ms, int request_hint_ms);
// True when a caller carrying budget_ms should give up on arrival rather than
// queue up. Deliberately a pure function of the lane's observable state so the
// decision can be tested without threads or sleeping.
//
// Only occupancy makes a start timestamp meaningful: a lane nobody holds is
// never overrunning, whatever timestamp the last holder left behind. An
// unbounded caller (non-positive budget) has no budget to exceed. And the
// comparison is strict, so a run whose age exactly equals the budget still has
// its last millisecond.
bool run_exceeds_budget(bool lane_occupied, std::int64_t run_started_ms,
std::int64_t now_ms, int budget_ms);
// Thrown when a caller cannot take the lane, in either of the two situations
// resolve_wait_budget_ms allows for. The message distinguishes them; callers
// that need to report a status code can treat them alike.
class LaneUnavailable : public std::runtime_error {
public:
explicit LaneUnavailable(const std::string &reason)
: std::runtime_error(reason) {}
};
class LaneEntry;
// One lane per loaded model. Shared by every handler thread; not copyable.
class InferenceLane {
public:
explicit InferenceLane(std::string model_label)
: model_label_(std::move(model_label)) {}
InferenceLane(const InferenceLane &) = delete;
InferenceLane &operator=(const InferenceLane &) = delete;
const std::string &model_label() const { return model_label_; }
private:
// Occupancy is only reachable through LaneEntry, so there is no way to take
// the lane without also having something that gives it back.
friend class LaneEntry;
void occupy(int budget_ms);
void vacate();
const std::string model_label_;
std::mutex state_mutex_;
std::condition_variable vacated_;
bool occupied_ = false;
// Only meaningful while occupied_ is true.
std::int64_t run_started_ms_ = 0;
};
// Scoped occupancy of a lane. Construct it where the inference happens and it
// is given back on every exit from that scope, including an exception and
// including a caller that returns from the middle of a long stream. Throws
// LaneUnavailable if the lane could not be taken, in which case there is no
// object and nothing to release.
class LaneEntry {
public:
// budget_ms <= 0 waits indefinitely. Pass the output of
// resolve_wait_budget_ms.
LaneEntry(InferenceLane &lane, int budget_ms);
~LaneEntry();
LaneEntry(const LaneEntry &) = delete;
LaneEntry &operator=(const LaneEntry &) = delete;
// Deliberately immovable rather than carefully movable. A moved-from entry
// would have to stop releasing the lane while the lane still records it as
// occupied, and that hazard is not worth the convenience: the lane can only
// be recovered by whoever took it. Callers needing to extend the lifetime
// past a scope should hold the entry indirectly instead.
LaneEntry(LaneEntry &&) = delete;
LaneEntry &operator=(LaneEntry &&) = delete;
private:
InferenceLane &lane_;
};
} // namespace audiocpp_backend

598
backend/cpp/audio-cpp/inference_lane_test.cpp Normal file
View File

@@ -0,0 +1,598 @@
// Unit tests for inference_lane. Standard library only. The harness compiles
// this file as a single translation unit, so the implementation is included
// directly rather than linked.
//
// Two kinds of test live here:
//
// * The pure ones (budget negotiation, the overrun predicate) run with no
// threads and no sleeping. They carry the arithmetic, so they are the tests
// that must be exhaustive.
// * The threaded ones exercise the lane itself. Every one of them is bounded:
// contenders use a generous wait budget instead of the unbounded mode
// wherever the point of the test does not require unbounded, and a watchdog
// in main() puts a ceiling on the whole file. A broken implementation must
// go red, not hang, because a hung job costs CI far more than a red one.
//
// Wall-clock margins are called out individually. The rule applied throughout:
// a margin is only allowed if a slow or loaded machine pushes the measurement
// deeper into the passing region.
#include "inference_lane.cpp"
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdio>
#include <cstdlib>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
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 audiocpp_backend::InferenceLane;
using audiocpp_backend::LaneEntry;
using audiocpp_backend::LaneUnavailable;
using audiocpp_backend::resolve_wait_budget_ms;
using audiocpp_backend::run_exceeds_budget;
// A one-shot, level-triggered signal with a bounded wait. Preferred over sleeps
// for "the other thread got there" so the tests do not encode a guess about
// scheduling.
class Signal {
public:
void raise() {
{
std::lock_guard<std::mutex> lock(mutex_);
raised_ = true;
}
cv_.notify_all();
}
bool await(int timeout_ms) {
std::unique_lock<std::mutex> lock(mutex_);
return cv_.wait_for(lock, std::chrono::milliseconds(timeout_ms),
[this] { return raised_; });
}
private:
std::mutex mutex_;
std::condition_variable cv_;
bool raised_ = false;
};
static std::int64_t elapsed_ms_since(
const std::chrono::steady_clock::time_point &start) {
return std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::steady_clock::now() - start)
.count();
}
static void nap(int ms) {
std::this_thread::sleep_for(std::chrono::milliseconds(ms));
}
static bool mentions(const std::string &haystack, const std::string &needle) {
return haystack.find(needle) != std::string::npos;
}
// Wording that separates the two failure modes. Kept here so a message reword
// that erases the distinction breaks these tests loudly.
static const char *const kTimedOutPhrase = "timed out after";
static const char *const kStillRunningPhrase = "has been running for";
// Attempts an entry and reports what happened, so the threaded tests can assert
// on the message rather than only on the exception type.
struct EntryOutcome {
bool acquired = false;
std::string message;
std::int64_t took_ms = 0;
};
static EntryOutcome try_entry(InferenceLane &lane, int budget_ms) {
EntryOutcome out;
const auto start = std::chrono::steady_clock::now();
try {
LaneEntry entry(lane, budget_ms);
out.acquired = true;
} catch (const LaneUnavailable &refused) {
out.message = refused.what();
}
out.took_ms = elapsed_ms_since(start);
return out;
}
// ---------------------------------------------------------------------------
// B9: budget negotiation. Pure, no threads.
// ---------------------------------------------------------------------------
static void test_budget_negotiation() {
check(resolve_wait_budget_ms(0, 0) == 0,
"B9 no ceiling and no request hint is unbounded");
check(resolve_wait_budget_ms(-1, -1) == 0,
"B9 negative ceiling and negative hint is unbounded");
check(resolve_wait_budget_ms(5000, 0) == 5000,
"B9 absent hint yields the ceiling");
check(resolve_wait_budget_ms(5000, -250) == 5000,
"B9 negative hint yields the ceiling");
check(resolve_wait_budget_ms(5000, 1200) == 1200,
"B9 a shorter request hint is granted");
check(resolve_wait_budget_ms(5000, 1) == 1,
"B9 a much shorter request hint is granted");
check(resolve_wait_budget_ms(5000, 9000) == 5000,
"B9 a longer request hint cannot weaken the ceiling");
check(resolve_wait_budget_ms(5000, 5001) == 5000,
"B9 a hint one ms over the ceiling is clamped");
check(resolve_wait_budget_ms(5000, 5000) == 5000,
"B9 a hint equal to the ceiling is the ceiling");
check(resolve_wait_budget_ms(0, 1200) == 1200,
"B9 without a policy limit the request hint applies");
check(resolve_wait_budget_ms(-5, 1200) == 1200,
"B9 a negative ceiling is no policy limit");
check(resolve_wait_budget_ms(1, 0) == 1,
"B9 a one ms ceiling survives negotiation");
}
// ---------------------------------------------------------------------------
// B5 and B6: the overrun predicate. Pure, no threads.
// ---------------------------------------------------------------------------
static void test_overrun_predicate() {
// B5: strictly longer.
check(run_exceeds_budget(true, 1000, 1100, 100) == false,
"B5 elapsed exactly equal to the budget is not an overrun");
check(run_exceeds_budget(true, 1000, 1101, 100) == true,
"B5 one ms past the budget is an overrun");
check(run_exceeds_budget(true, 1000, 1099, 100) == false,
"B5 one ms short of the budget is not an overrun");
check(run_exceeds_budget(true, 0, 1, 1) == false,
"B5 a one ms budget at one ms elapsed is not an overrun");
check(run_exceeds_budget(true, 0, 2, 1) == true,
"B5 a one ms budget at two ms elapsed is an overrun");
// B6: an unoccupied lane is never stuck, whatever the leftover timestamp
// says. This is the pure half of B6; the wiring half is threaded below.
check(run_exceeds_budget(false, 0, 10000000, 1) == false,
"B6 an idle lane with an ancient start timestamp is not an overrun");
check(run_exceeds_budget(false, 5, 5, 5) == false,
"B6 an idle lane is not an overrun at any elapsed value");
// An unbounded caller has no budget to exceed, so it never fails fast.
check(run_exceeds_budget(true, 0, 10000000, 0) == false,
"B2 an unbounded caller never sees an overrun");
check(run_exceeds_budget(true, 0, 10000000, -1) == false,
"B2 a negative budget never sees an overrun");
}
// ---------------------------------------------------------------------------
// B1: mutual exclusion under real contention, and a release admits a waiter.
// ---------------------------------------------------------------------------
static void test_mutual_exclusion() {
InferenceLane lane("exclusion-model");
constexpr int kContenders = 4; // "at least three simultaneous contenders"
constexpr int kHoldMs = 15;
// Generous on purpose: the point of this test is exclusion, not timeouts.
// A larger budget only makes a healthy run more likely to pass, while still
// bounding a broken one at roughly five seconds instead of forever.
constexpr int kBudgetMs = 5000;
std::atomic<int> in_flight{0};
std::atomic<int> peak_in_flight{0};
std::atomic<int> completed{0};
std::atomic<int> refused{0};
Signal go;
std::vector<std::thread> contenders;
for (int i = 0; i < kContenders; i++) {
contenders.emplace_back([&] {
go.await(5000);
try {
LaneEntry entry(lane, kBudgetMs);
const int now_inside = in_flight.fetch_add(1) + 1;
int seen = peak_in_flight.load();
while (now_inside > seen &&
!peak_in_flight.compare_exchange_weak(seen, now_inside)) {
// retry with the refreshed value
}
nap(kHoldMs);
in_flight.fetch_sub(1);
completed.fetch_add(1);
} catch (const LaneUnavailable &) {
refused.fetch_add(1);
}
});
}
go.raise();
for (auto &t : contenders) {
t.join();
}
check(refused.load() == 0, "B1 no contender was refused within its budget");
check(completed.load() == kContenders,
"B1 every contender eventually got the lane");
check(peak_in_flight.load() == 1,
"B1 never more than one holder inside the lane at once");
}
// ---------------------------------------------------------------------------
// B2: unbounded mode waits out a run longer than any bound would allow.
// ---------------------------------------------------------------------------
static void test_unbounded_waits_out_the_holder() {
InferenceLane lane("patient-model");
constexpr int kHoldMs = 300;
Signal held;
Signal patient_done;
std::int64_t patient_wait_ms = -1;
bool patient_acquired = false;
std::thread holder([&] {
LaneEntry entry(lane, 0);
held.raise();
nap(kHoldMs);
});
check(held.await(5000), "B2 holder took the lane");
std::thread patient([&] {
const auto start = std::chrono::steady_clock::now();
try {
LaneEntry entry(lane, 0);
patient_acquired = true;
} catch (const LaneUnavailable &) {
patient_acquired = false;
}
patient_wait_ms = elapsed_ms_since(start);
patient_done.raise();
});
// Bound on the unbounded mode: a lane that never wakes its waiters goes red
// here instead of hanging in join(). The named failure is printed before the
// join, so even a hard hang leaves a diagnosis behind for the watchdog.
check(patient_done.await(10000), "B2 unbounded caller returned at all");
patient.join();
holder.join();
check(patient_acquired, "B2 unbounded caller acquired instead of failing");
// Margin: the holder holds for 300 ms, so the waiter must block for about
// that long. Asserting only half of it means a loaded machine, which makes
// the wait longer, drifts further into passing.
check(patient_wait_ms >= kHoldMs / 2,
"B2 unbounded caller actually waited for the in-flight run");
}
// ---------------------------------------------------------------------------
// B3: a bounded caller that cannot get in gives up with the timeout wording.
// ---------------------------------------------------------------------------
static void test_bounded_wait_times_out() {
InferenceLane lane("impatient-model");
constexpr int kBudgetMs = 120;
Signal held;
Signal release;
std::thread holder([&] {
LaneEntry entry(lane, 0);
held.raise();
release.await(10000);
});
check(held.await(5000), "B3 holder took the lane");
// The waiter arrives immediately, so the holder's elapsed time is far below
// the budget and the fail-fast path must not trigger here.
const EntryOutcome outcome = try_entry(lane, kBudgetMs);
release.raise();
holder.join();
check(!outcome.acquired, "B3 bounded caller did not acquire a held lane");
check(mentions(outcome.message, kTimedOutPhrase),
"B3 failure names the exhausted wait, not an overrunning run");
check(!mentions(outcome.message, kStillRunningPhrase),
"B3 failure is not worded as an overrun");
check(mentions(outcome.message, "impatient-model"),
"B3 failure names the model");
check(mentions(outcome.message, "120"),
"B3 failure reports the budget it waited out");
// Margin: wait_for cannot return before its deadline, so the true value is
// at least 120 ms and load only raises it. Asserting 100 leaves room for
// clock granularity while still catching an implementation that returns
// early without waiting.
check(outcome.took_ms >= 100, "B3 bounded caller waited out its budget");
}
// ---------------------------------------------------------------------------
// B4: a caller whose budget is already exceeded fails at once.
// ---------------------------------------------------------------------------
static void test_fail_fast_against_a_long_run() {
InferenceLane lane("wedged-model");
constexpr int kBudgetMs = 200;
constexpr int kRunAgeMs = 400;
Signal held;
Signal release;
std::thread holder([&] {
LaneEntry entry(lane, 0);
held.raise();
release.await(10000);
});
check(held.await(5000), "B4 holder took the lane");
// Margin: the arriving caller needs the holder's elapsed time to exceed
// 200 ms. Sleeping 400 ms means a loaded machine oversleeps and pushes the
// elapsed time further past the budget, never below it.
nap(kRunAgeMs);
const EntryOutcome outcome = try_entry(lane, kBudgetMs);
release.raise();
holder.join();
check(!outcome.acquired, "B4 caller did not acquire a long-running lane");
check(mentions(outcome.message, kStillRunningPhrase),
"B4 failure states the measured age of the in-flight run");
check(!mentions(outcome.message, kTimedOutPhrase),
"B4 failure is not worded as an exhausted wait");
check(mentions(outcome.message, "wedged-model"),
"B4 failure names the model");
// Margin: a fail-fast return takes microseconds. 150 ms of headroom under a
// 200 ms budget separates "returned at once" from "waited out the budget"
// by a wide enough gap that scheduler noise cannot close it. The message
// assertions above are the load-independent proof; this one pins the timing.
check(outcome.took_ms < 150, "B4 caller failed without waiting out its budget");
}
// ---------------------------------------------------------------------------
// B6 wiring: an idle lane never looks stuck, however old the last run is.
// ---------------------------------------------------------------------------
static void test_idle_lane_is_never_stuck() {
InferenceLane lane("idle-model");
{
LaneEntry entry(lane, 0);
}
// Ages the leftover start timestamp well past the tiny budget used below.
// A longer sleep only makes a stale-timestamp bug more visible, so load
// helps this test rather than hurting it.
nap(80);
const EntryOutcome first = try_entry(lane, 20);
check(first.acquired, "B6 tiny budget still acquires an idle lane");
nap(80);
const EntryOutcome second = try_entry(lane, 1);
check(second.acquired, "B6 a one ms budget still acquires an idle lane");
}
// ---------------------------------------------------------------------------
// B7: every ownership exit clears the busy state, including an exception
// thrown from inside the guarded region.
// ---------------------------------------------------------------------------
static void test_release_on_exception() {
InferenceLane lane("throwing-model");
struct GuardedRegionFailure {};
bool propagated = false;
try {
LaneEntry entry(lane, 0);
throw GuardedRegionFailure{};
} catch (const GuardedRegionFailure &) {
propagated = true;
}
check(propagated, "B7 the guarded region's own exception propagated");
// If the throw had leaked the busy state, this tiny budget would fail.
const EntryOutcome after_throw = try_entry(lane, 20);
check(after_throw.acquired, "B7 lane is free after an exception unwound it");
// Same check for a holder that unwinds on another thread, which is the shape
// a gRPC handler failing mid-inference actually has.
Signal thrown;
std::thread unlucky([&] {
try {
LaneEntry entry(lane, 0);
throw GuardedRegionFailure{};
} catch (const GuardedRegionFailure &) {
thrown.raise();
}
});
check(thrown.await(5000), "B7 worker thread unwound its guarded region");
unlucky.join();
const EntryOutcome after_worker = try_entry(lane, 20);
check(after_worker.acquired,
"B7 lane is free after a worker thread unwound it");
}
// ---------------------------------------------------------------------------
// B8: a waiter must not publish itself as the holder. If it did, its arrival
// would restart the elapsed-time measurement and hide the real holder.
//
// Timeline, with the holder taking the lane at t0 and never letting go:
//
// t0 holder acquires, elapsed measurement starts here and only here
// t0+100 waiter arrives with a 400 ms budget, blocks, and times out
// t0+500 late caller arrives with a 450 ms budget
//
// A correct lane measures 500 ms of holding at the late arrival, which is more
// than 450, so the late caller fails fast. An implementation that let the
// waiter stamp itself as holder measures only the 400 ms since the waiter
// arrived, which is under 450, so the late caller would queue behind a stuck
// run instead. The two paths are told apart by their wording.
// ---------------------------------------------------------------------------
static void test_waiter_does_not_become_the_holder() {
InferenceLane lane("stamp-model");
constexpr int kWaiterArrivesAfterMs = 100;
constexpr int kWaiterBudgetMs = 400;
constexpr int kLateBudgetMs = 450;
Signal held;
Signal release;
std::thread holder([&] {
LaneEntry entry(lane, 0);
held.raise();
release.await(10000);
});
check(held.await(5000), "B8 holder took the lane");
// Margin: the waiter must not fail fast on arrival, which needs the
// holder's elapsed time to stay under 400 ms. Arriving at 100 ms leaves
// 300 ms of slack, so oversleeping under load does not flip the path.
nap(kWaiterArrivesAfterMs);
EntryOutcome waiter_outcome;
std::thread waiter([&] { waiter_outcome = try_entry(lane, kWaiterBudgetMs); });
waiter.join();
check(!waiter_outcome.acquired, "B8 mid-queue waiter did not acquire");
check(mentions(waiter_outcome.message, kTimedOutPhrase),
"B8 mid-queue waiter waited out its budget and timed out");
// Margin: the holder has now been in the lane for at least 500 ms against a
// 450 ms budget. Load lengthens both sleeps, so the measured age only grows
// and the fail-fast path only becomes more certain.
const EntryOutcome late = try_entry(lane, kLateBudgetMs);
release.raise();
holder.join();
check(!late.acquired, "B8 late caller did not acquire");
check(mentions(late.message, kStillRunningPhrase),
"B8 elapsed time is still measured from the real holder's acquisition");
check(late.took_ms < 200,
"B8 late caller failed fast rather than queueing behind the holder");
}
// ---------------------------------------------------------------------------
// B8, other direction: the age of a run is measured from the moment its holder
// acquired, not from the moment that holder arrived. A caller that queued for a
// while and then got in is starting a fresh run, and its time in the queue must
// not be billed to it: if it were, every handover would hand the new holder a
// head start towards looking overrun, and short-budget callers would be turned
// away from a run that has barely begun.
//
// t0 first holder acquires and holds for 300 ms
// t0 second caller arrives and queues
// t0+300 second caller acquires, so its own run age is ~0 here
// t0+300 a third caller arrives with a 300 ms budget
//
// Correct: the third caller sees a run that just started, so it queues and then
// times out. Billing the queue time to the second caller would show a 300 ms old
// run instead, and the third caller would be turned away as an overrun.
// ---------------------------------------------------------------------------
static void test_run_age_starts_at_acquisition() {
InferenceLane lane("handover-model");
constexpr int kFirstHoldMs = 300;
constexpr int kThirdBudgetMs = 200;
Signal first_held;
Signal handed_over;
Signal release_second;
std::thread first([&] {
LaneEntry entry(lane, 0);
first_held.raise();
nap(kFirstHoldMs);
});
// Ordering matters only for the diagnosis, not for the assertion: waiting
// for the first holder guarantees the second caller really does queue, which
// is what gives it queue time to be wrongly billed for.
check(first_held.await(5000), "B8 first holder took the lane");
std::thread second([&] {
LaneEntry entry(lane, 0);
handed_over.raise();
release_second.await(10000);
});
check(handed_over.await(10000), "B8 queued caller was handed the lane");
// Margin: the new holder's run is a few ms old against a 200 ms budget, so
// this caller must queue. Load can only add a few ms of handover latency,
// well inside that slack, while it lengthens the queue time that the buggy
// version would bill, making the bug more visible rather than less.
const EntryOutcome third = try_entry(lane, kThirdBudgetMs);
release_second.raise();
second.join();
first.join();
check(!third.acquired, "B8 lane was still held by the queued caller");
check(mentions(third.message, kTimedOutPhrase),
"B8 a fresh holder's run age excludes the time it spent queueing");
}
// ---------------------------------------------------------------------------
// B10: both failure modes carry a usable message, and the fail-fast one reports
// a measurement rather than diagnosing a cause.
// ---------------------------------------------------------------------------
static void test_failure_messages_are_diagnosable() {
InferenceLane lane("diagnosable-model");
Signal held;
Signal release;
std::thread holder([&] {
LaneEntry entry(lane, 0);
held.raise();
release.await(10000);
});
check(held.await(5000), "B10 holder took the lane");
nap(120);
const EntryOutcome fast = try_entry(lane, 30);
const EntryOutcome slow = try_entry(lane, 60);
release.raise();
holder.join();
check(fast.message != slow.message,
"B10 the two failure modes do not share one message");
check(!fast.message.empty() && !slow.message.empty(),
"B10 both failures carry text");
check(mentions(fast.message, "diagnosable-model") &&
mentions(slow.message, "diagnosable-model"),
"B10 both failures name the model");
// The fail-fast wording must not accuse the run of being stuck: a short
// budget meeting a legitimately long run reaches this path too.
for (const char *verdict : {"stuck", "wedged", "hung", "deadlock"}) {
check(!mentions(fast.message, verdict),
std::string("B10 fail-fast message avoids diagnosing '") +
verdict + "'");
}
}
int main() {
// Last resort only. Every threaded test above is individually bounded, so
// this should never fire; it exists so that an implementation which parks a
// thread forever still ends the job instead of occupying a CI runner.
std::thread watchdog([] {
std::this_thread::sleep_for(std::chrono::seconds(30));
fprintf(stderr, "FAIL: watchdog fired, an inference_lane test hung\n");
fflush(stderr);
std::_Exit(1);
});
watchdog.detach();
test_budget_negotiation();
test_overrun_predicate();
test_mutual_exclusion();
test_unbounded_waits_out_the_holder();
test_bounded_wait_times_out();
test_fail_fast_against_a_long_run();
test_idle_lane_is_never_stuck();
test_release_on_exception();
test_waiter_does_not_become_the_holder();
test_run_age_starts_at_acquisition();
test_failure_messages_are_diagnosable();
if (failures == 0) {
fprintf(stderr, "\nAll inference_lane tests passed.\n");
return 0;
}
fprintf(stderr, "\n%d inference_lane test(s) failed.\n", failures);
return 1;
}

70
backend/cpp/audio-cpp/run_guard.cpp
View File

@@ -1,70 +0,0 @@
#include "run_guard.h"
#include <chrono>
namespace audiocpp_backend {
std::int64_t steady_now_ms() {
return std::chrono::duration_cast<std::chrono::milliseconds>(
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<std::timed_mutex> 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<std::timed_mutex> 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

76
backend/cpp/audio-cpp/run_guard.h
View File

@@ -1,76 +0,0 @@
#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 <atomic>
#include <cstdint>
#include <mutex>
#include <stdexcept>
#include <string>
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<std::timed_mutex> 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<std::timed_mutex> 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<std::int64_t> busy_since_ms_{0};
};
} // namespace audiocpp_backend

364
backend/cpp/audio-cpp/run_guard_test.cpp
View File

@@ -1,364 +0,0 @@
// 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 <atomic>
#include <chrono>
#include <cstdio>
#include <future>
#include <string>
#include <thread>
#include <vector>
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<Outcome> 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<bool> held_{false};
std::atomic<bool> 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<int> concurrent{0};
std::atomic<int> peak{0};
std::atomic<bool> unexpected_busy{false};
std::vector<std::thread> 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;
}