diff --git a/src/call/call_audio_batch.c b/src/call/call_audio_batch.c index 6454f6342..ae5ae3d46 100644 --- a/src/call/call_audio_batch.c +++ b/src/call/call_audio_batch.c @@ -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, diff --git a/src/call/call_audio_batch.h b/src/call/call_audio_batch.h index 72e739ef6..93f5918a8 100644 --- a/src/call/call_audio_batch.h +++ b/src/call/call_audio_batch.h @@ -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. */ diff --git a/tests/test_call_audio_batch.c b/tests/test_call_audio_batch.c index 4f408b6f1..c7fcff18a 100644 --- a/tests/test_call_audio_batch.c +++ b/tests/test_call_audio_batch.c @@ -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) { diff --git a/tools/chatgui/src/voice_audio_io.cpp b/tools/chatgui/src/voice_audio_io.cpp index d19cd5eb3..ef6e2a86d 100644 --- a/tools/chatgui/src/voice_audio_io.cpp +++ b/tools/chatgui/src/voice_audio_io.cpp @@ -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; diff --git a/tools/chatgui/tests/test_call_jitter.cpp b/tools/chatgui/tests/test_call_jitter.cpp index ef2728ce4..02625f38a 100644 --- a/tools/chatgui/tests/test_call_jitter.cpp +++ b/tools/chatgui/tests/test_call_jitter.cpp @@ -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 output(777); under = 0; for (int tick = 0; tick < 30 && !under; tick++) { diff --git a/tools/chatgui/tests/test_voice_audio_io.cpp b/tools/chatgui/tests/test_voice_audio_io.cpp index 4e749c512..97cb9b8ee 100644 --- a/tools/chatgui/tests/test_voice_audio_io.cpp +++ b/tools/chatgui/tests/test_voice_audio_io.cpp @@ -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);