mirror of
https://github.com/mudler/LocalAI.git
synced 2026-07-30 18:09:05 -04:00
d31e923bb7
AudioTranscriptionLive holds the model's inference lane for the whole stream,
which is correct (the streaming session is stateful and a concurrent run would
interleave two callers' audio) and newly dangerous. Every other RPC holds the
lane across compute, or across a write to a slow reader, and both of those
terminate on their own. A live stream instead blocks in a client-driven read,
and a peer that goes silent WITHOUT closing the stream never terminates
anything: the lane stays taken and every other request against that model queues
behind a client that stopped speaking.
live_watchdog is a one-shot idle timer that ends the stream when no frame has
arrived inside a window. It is standard library only, so it is unit tested
without an engine. gRPC's synchronous Read has no timeout and cannot be given
one, so the only way to unblock it is ServerContext::TryCancel, which decides
the wire status itself: the client sees CANCELLED rather than the
DEADLINE_EXCEEDED the handler returns, the reason is logged, and the lane coming
back is the point. When it fires the read loop throws rather than reporting
end-of-input, so the driver does not go on to finalize a decode nobody is
waiting for.
It is armed only after the lane is taken and disarmed as soon as the read side
closes, and both ends matter. Arming earlier would cover acquire(), which
legitimately blocks while another live stream runs, so a queued caller would be
cancelled for waiting its turn. Disarming later would cover our own decode,
where a window overrun is not a peer going quiet and cancelling would throw away
the transcript the client is waiting for.
The window is the new live_idle_timeout_ms option, 30 s by default, 0 meaning no
limit. core/http/endpoints/openai/realtime.go drives a 300 ms ticker and feeds
every tick that produced new audio while a turn is open, so 30 s of silence is a
hundred ticks that delivered nothing. It is also longer than any pause a speaker
takes mid-utterance, which is the case that must never be cut off, and
backend.proto lets one stream span many utterances, so a client that pauses
longer between them raises the option rather than discovering it.
Two smaller corrections in the same handler:
- check_can_serve now runs BEFORE the sample rate check.
pkg/grpc/grpcerrors/errors.go degrades to the file path on UNIMPLEMENTED and
on nothing else, so a live-incapable model asked at a wrong rate was
answering INVALID_ARGUMENT and costing the caller its fallback.
- a negative sample rate is refused instead of silently becoming 16000. Zero
still means 16000, which is what the proto documents; -1 is malformed rather
than absent and gets the same refusal every other bad rate gets.
And one thing recorded rather than changed, at the handler: "live" here means
incremental INPUT, not low latency, and with the pinned families it does not yet
mean incremental OUTPUT either. nemotron_asr's process_audio_chunk only appends
to its buffer, so its whole decode and every delta happen inside finalize(),
after the client closes its send side. The policy-window buffering is inert for
that family and matters only for vibevoice_asr and higgs_audio_stt.
Verified on the wire with live_idle_timeout_ms:3000. A silent client acked at
371 ms and was cancelled at 3.371 s; a second live stream opened one second
later received its ack 2.37 s in, i.e. at the instant the first was cancelled,
and then transcribed successfully on the same cached session. Without the
watchdog it would still be waiting. Re-ran the live transcription (ready first,
59 incremental deltas, concat equal to the final text, word timestamps in
nanoseconds, eou and eob false), the citrinet refusal at both a right and a
wrong rate (UNIMPLEMENTED either way now), and Task 12's AudioTranscriptionStream
on nemotron_asr, which is unchanged.
Mutation testing the watchdog found a weakness in its own test: the destructor
test slept past the window inside the watched scope, so a destructor that
DETACHED the thread instead of joining it passed unnoticed. The test now uses a
window longer than the scope, which kills that mutant, and says why.
Assisted-by: Claude:claude-opus-5 [Claude Code]
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
166 lines
7.0 KiB
C++
166 lines
7.0 KiB
C++
// Unit tests for model_options. Standard library only, so
|
|
// backend/cpp/run-unit-tests.sh picks this up with no engine checkout.
|
|
//
|
|
// The harness compiles this file as a single translation unit with no other
|
|
// sources, so the implementation is included directly rather than linked.
|
|
//
|
|
// Build and run standalone:
|
|
// g++ -std=c++17 -I. model_options_test.cpp -o t && ./t
|
|
|
|
#include "model_options.cpp"
|
|
|
|
#include <cstdio>
|
|
#include <map>
|
|
#include <string>
|
|
#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());
|
|
}
|
|
}
|
|
|
|
// Returns the mapped value, or an empty string when the key is absent. map::at
|
|
// would throw on a miss and abort the whole binary, so one prefix off-by-one
|
|
// would hide every check that follows it instead of failing a single one.
|
|
static std::string lookup(const std::map<std::string, std::string> &values,
|
|
const std::string &key) {
|
|
const auto found = values.find(key);
|
|
return found == values.end() ? std::string() : found->second;
|
|
}
|
|
|
|
using audiocpp_backend::parse_model_options;
|
|
|
|
static void test_defaults() {
|
|
auto r = parse_model_options({});
|
|
check(r.error.empty(), "empty option list is not an error");
|
|
check(r.options.family.empty(), "family defaults to empty");
|
|
check(r.options.task.empty(), "task defaults to empty");
|
|
check(r.options.backend == "cpu", "backend defaults to cpu");
|
|
check(r.options.device == 0, "device defaults to 0");
|
|
check(r.options.threads == 0, "threads defaults to 0");
|
|
check(r.options.busy_timeout_ms == 0, "busy_timeout_ms defaults to 0");
|
|
// NOT zero, unlike every other numeric option here. A live stream holds the
|
|
// model's lane while it waits on the client, so the default has to bound
|
|
// that wait; 0 is the explicit "no limit" the operator opts into.
|
|
check(r.options.live_idle_timeout_ms == 30000,
|
|
"live_idle_timeout_ms defaults to 30000");
|
|
check(r.options.load_options.empty(), "load_options defaults empty");
|
|
check(r.options.session_options.empty(), "session_options defaults empty");
|
|
}
|
|
|
|
static void test_scalar_options() {
|
|
auto r = parse_model_options({
|
|
"family:qwen3_tts",
|
|
"task:tts",
|
|
"backend:cuda",
|
|
"device:1",
|
|
"threads:8",
|
|
"busy_timeout_ms:30000",
|
|
"live_idle_timeout_ms:5000",
|
|
});
|
|
check(r.error.empty(), "scalar options parse without error");
|
|
check(r.options.family == "qwen3_tts", "family parsed");
|
|
check(r.options.task == "tts", "task parsed");
|
|
check(r.options.backend == "cuda", "backend parsed");
|
|
check(r.options.device == 1, "device parsed");
|
|
check(r.options.threads == 8, "threads parsed");
|
|
check(r.options.busy_timeout_ms == 30000, "busy_timeout_ms parsed");
|
|
check(r.options.live_idle_timeout_ms == 5000, "live_idle_timeout_ms parsed");
|
|
check(parse_model_options({"live_idle_timeout_ms:0"}).options.live_idle_timeout_ms == 0,
|
|
"an explicit 0 turns the live idle limit off rather than reverting to "
|
|
"the default");
|
|
}
|
|
|
|
// Values containing colons must survive: split on the FIRST colon only.
|
|
static void test_value_containing_colon() {
|
|
auto r = parse_model_options({"model_spec_override:/models/a:b/spec.json"});
|
|
check(r.error.empty(), "colon-bearing value is not an error");
|
|
check(r.options.model_spec_override == "/models/a:b/spec.json",
|
|
"value keeps every colon after the first separator");
|
|
}
|
|
|
|
static void test_namespaced_options() {
|
|
auto r = parse_model_options({
|
|
"load.weight_type:q8_0",
|
|
"session.miocodec.weight_type:f16",
|
|
"session.graph_capacity:tiered",
|
|
});
|
|
check(r.error.empty(), "namespaced options parse without error");
|
|
check(r.options.load_options.size() == 1, "one load option");
|
|
check(lookup(r.options.load_options, "weight_type") == "q8_0", "load prefix stripped");
|
|
check(r.options.session_options.size() == 2, "two session options");
|
|
check(lookup(r.options.session_options, "miocodec.weight_type") == "f16",
|
|
"session prefix stripped, inner dots kept");
|
|
check(lookup(r.options.session_options, "graph_capacity") == "tiered",
|
|
"second session option parsed");
|
|
}
|
|
|
|
static void test_errors() {
|
|
check(!parse_model_options({"family"}).error.empty(),
|
|
"entry without a colon is rejected");
|
|
check(!parse_model_options({"nonsense:1"}).error.empty(),
|
|
"unknown key is rejected");
|
|
check(!parse_model_options({"device:abc"}).error.empty(),
|
|
"non-numeric device is rejected");
|
|
check(!parse_model_options({"threads:-1"}).error.empty(),
|
|
"negative threads is rejected");
|
|
check(!parse_model_options({"load.:x"}).error.empty(),
|
|
"empty load key is rejected");
|
|
check(!parse_model_options({"session.:x"}).error.empty(),
|
|
"empty session key is rejected");
|
|
check(!parse_model_options({"busy_timeout_ms:abc"}).error.empty(),
|
|
"non-numeric busy_timeout_ms is rejected");
|
|
check(!parse_model_options({"live_idle_timeout_ms:abc"}).error.empty(),
|
|
"non-numeric live_idle_timeout_ms is rejected");
|
|
check(!parse_model_options({"live_idle_timeout_ms:-1"}).error.empty(),
|
|
"negative live_idle_timeout_ms is rejected");
|
|
check(!parse_model_options({"device:-1"}).error.empty(),
|
|
"negative device is rejected");
|
|
check(!parse_model_options({"threads:x"}).error.empty(),
|
|
"non-numeric threads is rejected");
|
|
|
|
// Values too large for int must be rejected, not silently wrapped into a
|
|
// negative device index that then reaches the ggml backend selector.
|
|
check(!parse_model_options({"device:2147483648"}).error.empty(),
|
|
"device above INT_MAX is rejected");
|
|
check(!parse_model_options({"threads:99999999999999"}).error.empty(),
|
|
"threads above INT_MAX is rejected");
|
|
|
|
// The error text must name the offending entry so a user can fix their YAML.
|
|
const auto r = parse_model_options({"nonsense:1"});
|
|
check(r.error.find("nonsense") != std::string::npos,
|
|
"error names the offending key");
|
|
|
|
// An empty key still has to give the user something to grep for.
|
|
const auto empty_key = parse_model_options({":value"});
|
|
check(empty_key.error.find(":value") != std::string::npos,
|
|
"unknown-key error names the entry even when the key is empty");
|
|
}
|
|
|
|
static void test_blank_entries_ignored() {
|
|
auto r = parse_model_options({"", " ", "family:supertonic"});
|
|
check(r.error.empty(), "blank entries are skipped, not rejected");
|
|
check(r.options.family == "supertonic", "real entry still parsed");
|
|
}
|
|
|
|
int main() {
|
|
test_defaults();
|
|
test_scalar_options();
|
|
test_value_containing_colon();
|
|
test_namespaced_options();
|
|
test_errors();
|
|
test_blank_entries_ignored();
|
|
if (failures) {
|
|
fprintf(stderr, "%d check(s) failed\n", failures);
|
|
return 1;
|
|
}
|
|
fprintf(stderr, "all model_options checks passed\n");
|
|
return 0;
|
|
}
|