diff --git a/src/call/call.c b/src/call/call.c index 7f106b15c..86dcd7a90 100644 --- a/src/call/call.c +++ b/src/call/call.c @@ -28,6 +28,8 @@ #include "../routing_layer/topo_node.h" #include "../transport_layer/etcp_api.h" #include "../transport_layer/etcp.h" +#include "../transport_layer/etcp_tx_stats.h" +#include "call_audio_history.h" #include "../../lib/mem.h" #include "../../lib/ll_queue.h" #include "../../lib/debug_config.h" @@ -1012,6 +1014,21 @@ static void call_audio_batch_tick(void* arg) { struct ETCP_ROUTER_TX_STATS net = {0}; uint64_t now = get_time_us(); int found = etcp_router_tx_stats(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_AUDIO, now, &net) == 0; + const struct call_audio_ops* audio = s->ctx->inst->call_audio_ops; + if (audio && audio->record_tx) { + struct call_history_tx snapshot = {.at_ms = now / 1000, .queued_bytes = s->tx_q ? s->tx_q->total_bytes : 0}; + struct ll_queue* queues = s->ctx->inst->router_conns; + for (struct ll_entry* e = queues ? queues->head : NULL; e; e = e->next) { + struct ETCP_ROUTER_CONN* r = (struct ETCP_ROUTER_CONN*)e; + if (r->closed) continue; + if (r->send_q) snapshot.queued_bytes += r->send_q->total_bytes; + if (r->inflight_q) snapshot.inflight_bytes += r->inflight_q->total_bytes; + } + struct etcp_tx_snapshot transport; + if (etcp_tx_get_snapshot(s->ctx->inst, now, &transport) == 0) + snapshot.transport_bytes = transport.queued_bytes + transport.inflight_bytes; + audio->record_tx(s->call_id, &snapshot); + } if (call_audio_batch_step(&s->audio_batch, now, found ? &net : NULL)) DEBUG_ERROR(DEBUG_CATEGORY_CALL, "audio batch control failed call=%llx; retaining target=%u", (unsigned long long)s->call_id, s->audio_batch.target); diff --git a/tools/chatgui/tests/test_call_jitter.cpp b/tools/chatgui/tests/test_call_jitter.cpp index b286bb696..27af4acad 100644 --- a/tools/chatgui/tests/test_call_jitter.cpp +++ b/tools/chatgui/tests/test_call_jitter.cpp @@ -2,6 +2,7 @@ #include "call/call_jitter.h" #include "call/call_jitter_window.h" #include "call/call_playout_policy.h" +#include "call/call_history_store.h" #include "call/call_audio.h" #include "utun_instance.h" #include "../../../lib/opus_codec.h" @@ -195,7 +196,33 @@ static void test_render_readiness() { call_audio_release(id); opus_codec_encoder_destroy(enc); } +static void test_audio_history() { + CallHistoryStore history; + call_audio_history snapshot; + history.receive(0, 100, 200, 100, 0, 0); // Подготовка не получает веса тишины. + history.receive(10, 100, 200, 100, 480, 240); + history.receive(20, 200, 200, 100, 480, 0); + history.receive(10020, 300, 200, 100, 960, 0); + history.transmit({10020, 100, 300, 250}); + history.snapshot(10020, snapshot); + CHECK(snapshot.rx_count == 2 && snapshot.tx_count == 1); + CHECK(snapshot.rx[0].low_ms == 100 && snapshot.rx[0].high_ms == 200); + CHECK(std::abs(snapshot.observed_ms - 30) < .001); + CHECK(std::abs(snapshot.missing_ms - 2.5) < .001); + CHECK(std::abs(snapshot.histogram[2] - 1.0 / 6) < .001); + CHECK(std::abs(snapshot.histogram[6] - 2.0 / 3) < .001); + CHECK(snapshot.tx[0].queued_bytes + snapshot.tx[0].inflight_bytes == 400); // Транспорт не складывается повторно. + history.snapshot(40020, snapshot); + CHECK(!snapshot.rx_count && !snapshot.tx_count && snapshot.observed_ms == 0); + for (int t = 50000; t < 90000; t += 50) history.receive(t, 4000, 200, 100, 2400, 0); + history.snapshot(89950, snapshot); + CHECK(snapshot.rx_count == 600 && snapshot.rx[0].at_ms == 60000); + CHECK(std::abs(snapshot.histogram[60] - 1) < .001); + puts("audio history: time weights, startup exclusion, gaps, overflow bin and ring expiry PASS"); +} + int main() { + test_audio_history(); test_startup_policy(); test_packet_arrival_sampling(); CHECK(call_audio_init(&audio_owner) == 0);