jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Adapt call batching to queue pressure and verify protected audio startup

Browse Source
This commit is contained in:
evgeny committed 2026-10-30 19:54:58 +01:00
1 parent 1960377e5c
commit b092e09b43
6 files changed
+106 -34
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/call/call_audio_batch.c
+15 -8
View File
@@ -71,25 +71,32 @@ int call_audio_batch_step(struct call_audio_batch_control *c, uint64_t now, cons
c->delay_us = net ? net->delivery_rtt_us : 0;
if (net && net->queued_age_us > c->delay_us) c->delay_us = net->queued_age_us;
if (net && net->inflight_age_us > c->delay_us) c->delay_us = net->inflight_age_us;
uint64_t interval = net && net->delivery_rtt_us > 250000 ? net->delivery_rtt_us * 2 : 500000;
uint64_t fresh_us = interval > 2000000 ? interval : 2000000;
uint64_t interval = net ? net->delivery_rtt_us : 250000;
if (interval < 250000) interval = 250000;
if (interval > 750000) interval = 750000;
uint64_t fresh_us = net && net->delivery_rtt_us > 1000000 ? net->delivery_rtt_us * 2 : 2000000;
int measured = net && net->ready && !net->no_route && !net->receive_busy && !net->probing
&& net->acked_packets && now - net->last_ack_us <= fresh_us;
if (measured) c->feedback_received = 1;
int usable = measured && net->window.valid;
// Постоянный RTT не означает перегрузку: смотрим ожидание отправки и превышение обычной доставки.
int pressure = net && (net->queued_age_us > 100000 || net->inflight_age_us > net->delivery_rtt_us + 100000);
const char *reason = "hold";
if (!usable) { c->high = c->low = 0; reason = "unavailable"; }
else if (c->delay_us > 100000) {
else if (pressure) {
c->low = 0;
if (!c->high) { c->high = 1; c->high_since_us = now; }
if (c->target < CALL_AUDIO_BATCH_MAX && now >= c->next_change_us && now - c->high_since_us >= 200000) {
c->target++; c->next_change_us = now + interval; reason = "enqueue_ack_delay";
if (c->target < CALL_AUDIO_BATCH_MAX && now >= c->next_change_us && now - c->high_since_us >= 200000
&& net->acked_packets > c->changed_ack_packets) {
c->target++; c->next_change_us = now + interval; c->changed_ack_packets = net->acked_packets; reason = "queue_pressure";
}
} else if (c->delay_us < 40000 && !net->queued_packets) {
} else if (!net->queued_packets && net->inflight_age_us <= net->delivery_rtt_us + 40000) {
c->high = 0;
if (!c->low) { c->low = 1; c->low_since_us = now; }
if (c->target > 1 && now >= c->next_change_us && now - c->low_since_us >= 2000000) {
c->target--; c->next_change_us = now + interval; c->low_since_us = now; reason = "recovered";
if (c->target > 1 && now >= c->next_change_us && now - c->low_since_us >= 5000000
&& net->acked_packets > c->changed_ack_packets) {
c->target--; c->next_change_us = now + interval; c->low_since_us = now;
c->changed_ack_packets = net->acked_packets; reason = "recovered";
}
} else c->high = c->low = 0;
if (c->target != previous) DEBUG_INFO(DEBUG_CATEGORY_CALL,
src/call/call_audio_batch.h
+1
View File
@@ -18,6 +18,7 @@ struct call_audio_batch_view {
struct ETCP_ROUTER_TX_STATS;
struct call_audio_batch_control {
uint64_t local_epoch, peer_epoch, last_us, high_since_us, low_since_us, next_change_us;
uint64_t changed_ack_packets; /* После изменения ждём новых подтверждений, не реагируем повторно на старый снимок. */
uint64_t delay_us; /* max полного RTT и возраста Router-очередей; локальное накопление не входит. */
unsigned target; /* Необходимое число кадров, 1..10; старт с 4. */
int feedback_received; /* Первый пригодный ACK в текущей Router epoch. */
tests/test_call_audio_batch.c
+13 -6
View File
@@ -45,7 +45,7 @@ static void wire_test(void) {
}
static void step(struct call_audio_batch_control *c, struct ETCP_ROUTER_TX_STATS *n, uint64_t t) {
n->last_ack_us = t; CHECK(call_audio_batch_step(c, t, n) == 0);
n->last_ack_us = t; ++n->acked_packets; CHECK(call_audio_batch_step(c, t, n) == 0);
}
static void control_test(void) {
struct call_audio_batch_control c;
@@ -64,19 +64,26 @@ static void control_test(void) {
/* Изолированная задержка не включает упаковку. */
n.delivery_rtt_us = 150000; step(&c, &n, 2000000); step(&c, &n, 2050000);
n.delivery_rtt_us = 20000; step(&c, &n, 2100000); CHECK(c.target == 4);
/* Даже без queued/inflight высокий обычный RTT включает пачки. */
/* Постоянный RTT без роста очередей не заставляет увеличивать пачку. */
n.delivery_rtt_us = 150000;
for (unsigned t = 2150000; t <= 2350000; t += 50000) step(&c, &n, t);
CHECK(c.target == 5);
CHECK(c.target == 4);
n.queued_age_us = 150000; n.queued_packets = 2;
for (unsigned t = 2400000; t <= 2850000; t += 50000) step(&c, &n, t);
CHECK(c.target == 6);
uint64_t ack = n.acked_packets;
for (unsigned t = 2900000; t <= 3500000; t += 50000) {
n.last_ack_us = t; CHECK(call_audio_batch_step(&c, t, &n) == 0);
CHECK(c.target == 6 && n.acked_packets == ack); // Один снимок не вызывает несколько увеличений.
}
n.delivery_rtt_us = 300000;
for (unsigned t = 2900000; t <= 9000000; t += 50000) step(&c, &n, t);
for (unsigned t = 3550000; t <= 9000000; t += 50000) step(&c, &n, t);
CHECK(c.target == 10);
n.queued_age_us = 0; n.queued_packets = 0;
n.delivery_rtt_us = 10000; step(&c, &n, 9050000);
CHECK(c.target == 10);
for (unsigned t = 9100000; t <= 11050000; t += 50000) step(&c, &n, t);
CHECK(c.target == 9);
CHECK(c.target == 10);
n.peer_epoch = 3; step(&c, &n, 11100000); CHECK(c.target == 4);
n.inflight_age_us = 150000;
for (unsigned t = 11150000; t <= 11350000; t += 50000) step(&c, &n, t);
@@ -88,7 +95,7 @@ static void control_test(void) {
CHECK(call_audio_batch_step(&c, 19999999, &n) == -1);
CHECK(call_audio_batch_step(&c, 20000001, NULL) == 0 && c.target == 4);
n.inflight_age_us = 0;
for (unsigned t = 20100000; t <= 27000000; t += 50000) step(&c, &n, t);
for (unsigned t = 20100000; t <= 37000000; t += 50000) step(&c, &n, t);
CHECK(c.target == 1);
}
int main(void) {
tools/chatgui/src/voice_audio_io.cpp
+16 -11
View File
@@ -133,17 +133,6 @@ void VoiceAudioIo::capture(const int16_t* pcm, unsigned frames, int64_t firstTim
void VoiceAudioIo::reference(const float* pcm, unsigned frames, int64_t firstTimeUs) {
if (!m_capturing) return;
if (!firstTimeUs) firstTimeUs = voice_time_us();
if (m_kind == Kind::Call && m_outputStartedEpoch != epoch) {
unsigned required = std::max(1920u, callbackFrames ? callbackFrames : frames);
if (m_playback.size() * 960 < required) {
// PREBUFFER: устройство работает, но речь ещё не началась. Это не starvation.
PlaybackProgress waiting{}; waiting.epoch = epoch; waiting.callbackFrames = callbackFrames ? callbackFrames : frames;
if (!m_playProgress.push(waiting)) m_progressDropped.fetch_add(1, std::memory_order_relaxed);
m_wake.notify_all();
return;
}
m_outputStartedEpoch = epoch;
}
for (unsigned offset = 0; offset < frames;) {
Event event{}; event.frames = std::min(frames - offset, 960u);
event.time = firstTimeUs + (int64_t)offset * 1000000 / 48000;
@@ -162,6 +151,22 @@ void VoiceAudioIo::reference(const float* pcm, unsigned frames, int64_t firstTim
void VoiceAudioIo::mix(float* stereo, unsigned frames, int64_t firstTimeUs, bool measured, unsigned callbackFrames) {
uint64_t epoch = m_playEpoch.load();
if (!firstTimeUs) firstTimeUs = voice_time_us();
if (m_kind == Kind::Call && m_outputStartedEpoch != epoch) {
// Извлечь только первый актуальный блок, сохранив его до разрешения вывода.
if (m_playCurrent.epoch != epoch) m_playPosition = 960;
for (int attempt = 0; m_playPosition == 960 && attempt < 12 && m_playback.pop(m_playCurrent); ++attempt)
if (m_playCurrent.epoch == epoch) m_playPosition = 0;
unsigned required = std::max(1920u, callbackFrames ? callbackFrames : frames);
unsigned ready = unsigned(m_playback.size() * 960) + (m_playCurrent.epoch == epoch ? 960 - m_playPosition : 0);
if (ready < required) {
// PREBUFFER: устройство работает, но речь ещё не началась. Это не starvation.
PlaybackProgress waiting{}; waiting.epoch = epoch; waiting.callbackFrames = callbackFrames ? callbackFrames : frames;
if (!m_playProgress.push(waiting)) m_progressDropped.fetch_add(1, std::memory_order_relaxed);
m_wake.notify_all();
return;
}
m_outputStartedEpoch = epoch;
}
unsigned consumed = 0;
for (unsigned i = 0; i < frames; ++i) {
if (m_playCurrent.epoch != epoch) m_playPosition = 960;
tools/chatgui/tests/test_call_jitter.cpp
+46 -9
View File
@@ -1,6 +1,7 @@
// Сохранность FIFO, десятисекундный burst, догон при продолжающемся входе и повторное накопление.
#include "call/call_jitter.h"
#include "call/call_jitter_window.h"
#include "call/call_playout_policy.h"
#include "call/call_audio.h"
#include "utun_instance.h"
#include "../../../lib/opus_codec.h"
@@ -133,7 +134,42 @@ static void test_packet_arrival_sampling() {
call_jitter_destroy(j);
}
static void test_startup_policy() {
call_playout_policy p;
CHECK(p.reserve_ms == 200);
for (int t = 0; t < 10000; t += 80) p.observe(t, 20, 1000, false);
CHECK(p.reserve_ms == 200); // Нет проигрывания — нет доказательства устойчивости.
p.recover(10000, 1000); CHECK(p.reserve_ms == 220);
for (int t = 10000; t < 15000; t += 80) p.observe(t, 20, 1000, true);
CHECK(p.reserve_ms == 220);
p.observe(15100, 20, 1000, true); CHECK(p.reserve_ms == 200);
p.observe(15101, 350, 1000, true); CHECK(p.reserve_ms == 350);
p.observe(15102, 20, 1000, true); CHECK(p.reserve_ms == 350);
decoder d; auto* j = call_jitter_create(decode, &d, 1000);
int next = 0, packets = 0, batch = 4, nominal = 80, due = nominal;
int16_t pcm[480]; bool played = false;
for (int now = 0; now < 20000; now += 10) {
if (now >= due) {
for (int i = 0; i < batch; i++, next++)
call_jitter_push(j, (uint8_t*)&next, sizeof(next), nominal, now, nullptr, i == 0);
if (++packets == 60) batch = 10;
nominal += batch * 20; due = nominal + (packets % 4 == 0 ? 120 : 0);
}
int n = call_jitter_pull(j, pcm, 480);
played |= n > 0;
if (n < 480) call_jitter_silence(j, 480 - n);
call_jitter_diagnostics state{}; call_jitter_get_diagnostics(j, &state, 0);
CHECK(state.underruns == 0);
if (now < 5000) CHECK(state.reserve_ms >= 200);
}
CHECK(played);
call_jitter_destroy(j);
puts("startup policy: cold start, held reserve, packet size change and jitter PASS");
}
int main() {
test_startup_policy();
test_packet_arrival_sampling();
CHECK(call_audio_init(&audio_owner) == 0);
{
@@ -187,10 +223,12 @@ int main() {
call_jitter_get_stats(j, &depth, nullptr, nullptr, &under, nullptr);
CHECK(depth == 180 && under == 0);
push(j, 9);
CHECK(call_jitter_pull(j, pcm, 480) == 0); // 200мс защищённого запаса плюс следующий блок.
push(j, 10);
CHECK(call_jitter_pull(j, pcm, 480) == 480);
call_jitter_diagnostics diagnostic{};
call_jitter_get_diagnostics(j, &diagnostic, 0);
CHECK(diagnostic.resumed == 1 && diagnostic.pcm_samples == 480 && diagnostic.reserve_ms == 20);
CHECK(diagnostic.resumed == 1 && diagnostic.pcm_samples == 480 && diagnostic.reserve_ms == 200);
CHECK(diagnostic.encoded_ms >= 0 && diagnostic.processing_ms >= 0 && diagnostic.arrival_spread_ms == 0);
call_jitter_get_diagnostics(j, &diagnostic, 0);
CHECK(diagnostic.resumed == 1); // Снимок playback не потребляет события для uasync.
@@ -199,7 +237,6 @@ int main() {
call_jitter_get_diagnostics(j, &diagnostic, 0);
CHECK(diagnostic.resumed == 0);
call_jitter_range range;
push(j, 10);
for (int i = 11; i < 511; i++) push(j, i, 10220); // 10с сетевой задержки, затем пачка
call_jitter_get_stats(j, nullptr, nullptr, nullptr, nullptr, &range);
CHECK(range.reserve_ms == 1000);
@@ -224,7 +261,7 @@ int main() {
push(j, next++);
CHECK(call_jitter_pull(j, pcm, 480) == 0);
CHECK(d.seen.size() == before);
push(j, next++);
for (int i = 0; i < 11; i++) push(j, next++);
int refill = call_jitter_pull(j, pcm, 480);
if (refill != 480) {
call_jitter_get_stats(j, &depth, nullptr, nullptr, nullptr, &range);
@@ -244,7 +281,7 @@ int main() {
for (int i = 0; i < 60; i++) push(j, i);
call_jitter_range range;
call_jitter_get_stats(j, &depth, nullptr, nullptr, nullptr, &range);
CHECK(depth == 1200 && range.reserve_ms == 20);
CHECK(depth == 1200 && range.reserve_ms == 200);
CHECK(call_jitter_pull(j, pcm, 480) == 480); // первый pull сразу воспроизводит речь
int next = 60, short_pulls = 0;
for (int tick = 0; tick < 6000; tick++) {
@@ -259,7 +296,7 @@ int main() {
state.tempo_x100, (long long)state.arrival_spread_ms);
}
call_jitter_get_stats(j, &depth, &tempo, &dropped, &under, &range);
CHECK(range.reserve_ms == 20); // догон не увеличивает сетевой резерв
CHECK(range.reserve_ms <= 200); // догон не увеличивает сетевой резерв; начальная цель уходит постепенно.
}
CHECK(short_pulls == 0);
CHECK(depth < 150 && tempo <= 110 && dropped == 0 && under == 0);
@@ -325,7 +362,7 @@ int main() {
call_audio_on_media(&audio_owner, id, 0, 0, encoded, len, nullptr, 1);
CHECK(call_audio_pull_pcm(id, pcm, 480) == 480);
for (auto value : pcm) CHECK(value == 0);
for (int i = 1; i < 10; i++) call_audio_on_media(&audio_owner, id, i, 0, encoded, len, nullptr, 1);
for (int i = 1; i < 11; i++) call_audio_on_media(&audio_owner, id, i, 0, encoded, len, nullptr, 1);
CHECK(call_audio_pull_pcm(id, pcm, 480) == 480);
bool audible = false;
for (auto value : pcm) if (value) audible = true;
@@ -338,14 +375,14 @@ int main() {
}
CHECK(audible);
// Малого запаса недостаточно для восстановления: сигнал продолжает звучать.
call_audio_on_media(&audio_owner, id, 10, 0, encoded, len, nullptr, 1);
call_audio_on_media(&audio_owner, id, 11, 0, encoded, len, nullptr, 1);
audible = false;
for (int tick = 0; tick < 100; tick++) {
CHECK(call_audio_pull_pcm(id, pcm, 480) == 480);
for (auto value : pcm) if (std::abs((int)value) > 1000) audible = true;
}
CHECK(audible);
for (int i = 11; i < 34; i++) call_audio_on_media(&audio_owner, id, i, 0, encoded, len, nullptr, 1);
for (int i = 12; i < 35; i++) call_audio_on_media(&audio_owner, id, i, 0, encoded, len, nullptr, 1);
CHECK(call_audio_pull_pcm(id, pcm, 480) == 480);
call_audio_begin_end(id); // явное завершение прекращает догон и включает финальный тон
CHECK(call_audio_pull_pcm(id, pcm, 480) > 0);
@@ -361,7 +398,7 @@ int main() {
uint8_t encoded[512];
int len = opus_codec_encode(encoder, silence, 960, encoded, sizeof(encoded));
CHECK(len > 0);
for (int i = 0; i < 10; i++) call_audio_on_media(&audio_owner, 124, i, 0, encoded, len, nullptr, 1);
for (int i = 0; i < 11; i++) call_audio_on_media(&audio_owner, 124, i, 0, encoded, len, nullptr, 1);
std::vector<int16_t> output(777);
under = 0;
for (int tick = 0; tick < 30 && !under; tick++) {
tools/chatgui/tests/test_voice_audio_io.cpp
+15
View File
@@ -163,6 +163,21 @@ static void small_chunks_aec(unsigned captureChunk, unsigned renderChunk) {
}
int main() {
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
{
VoiceAudioIo cold;
mediaReady = false; pulls = 0; output_position = 0;
REQUIRE(cold.start(VoiceAudioIo::Kind::Call, 99));
float out[4096]{};
for (int i = 0; i < 20; i++) { cold.mix(out, 2048); std::this_thread::sleep_for(std::chrono::milliseconds(2)); }
REQUIRE(pulls == 0 && output_position == 0);
REQUIRE(std::all_of(std::begin(out), std::end(out), [](float v) { return v == 0; }));
mediaReady = true;
await([] { return pulls >= 3; });
cold.mix(out, 2048);
await([] { return output_position == 2048; });
REQUIRE(std::all_of(std::begin(out), std::end(out), [](float v) { return v == 1000 / 32768.f; }));
cold.stop();
}
// Резерв уменьшается только при стабильной работе; крупный callback и starvation повышают минимум.
VoicePlaybackBuffer buffering;
REQUIRE(buffering.blocks == 1);