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

Require the adaptive frame count before requesting or using a Router slot

Browse Source
This commit is contained in:
evgeny committed 2026-10-30 19:28:22 +01:00
1 parent ee5c258b3c
commit 6ae080d417
2 files changed
+33 -9
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/call/call.c
+31 -9
View File
@@ -47,6 +47,7 @@ struct call_audio_pending {
uint16_t seq;
};
static void call_tx_flush(struct call_session* s);
static void call_audio_batch_tick(void* arg);
static void call_tx_flush(struct call_session* s);
@@ -150,6 +151,7 @@ static struct call_session* call_session_new(struct call_ctx* ctx, uint64_t call
s->state = CALL_ST_IDLE;
s->started_tb = get_time_tb();
s->ctx = ctx;
call_audio_batch_init(&s->audio_batch, get_time_us());
s->tx_q = queue_new(ctx->inst->ua, 0, 0, 0, "call_txq");
if (!s->tx_q) { queue_entry_free(qe); return NULL; }
// При провале queue_data_put_with_index сам освобождает qe (вместе с s), поэтому
@@ -171,6 +173,8 @@ static void call_session_free_cb(void* arg) {
* дренировать tx_q (без отправки). Безопасно вызывать из uasync-потока. */
static void call_session_cleanup(struct call_ctx* ctx, struct call_session* s) {
call_video_transport_stop(s);
if (s->audio_batch_timer) uasync_cancel_timeout(ctx->inst->ua, s->audio_batch_timer);
s->audio_batch_timer = NULL;
etcp_router_cancel_request(&s->audio_send);
if (s->tx_q) {
struct ll_entry* e;
@@ -427,8 +431,9 @@ static void call_activate(struct call_session* s) {
if (s->state == CALL_ST_ACTIVE || s->state == CALL_ST_ENDED) return;
struct call_ctx* ctx = s->ctx;
s->state = CALL_ST_ACTIVE;
DEBUG_INFO(DEBUG_CATEGORY_CALL, "audio batch startup call=%llx mode=router-ready max_frames=%u max_bytes=%u",
(unsigned long long)s->call_id, CALL_AUDIO_BATCH_MAX, CALL_MEDIA_MAX_BYTES);
call_audio_batch_tick(s);
DEBUG_INFO(DEBUG_CATEGORY_CALL, "audio batch startup call=%llx mode=full-and-router-ready target=%u max_bytes=%u",
(unsigned long long)s->call_id, s->audio_batch.target, CALL_MEDIA_MAX_BYTES);
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: accepted id=%016llx -> ACTIVE (router delivery)", CALL_ID,
(unsigned long long)s->call_id);
call_event_id(ctx, CHAT_EVT_CALL_ACCEPTED, s->call_id, 0);
@@ -534,11 +539,12 @@ static void call_post_stats(struct call_session* s) {
s->tx_wait_frames = s->tx_wait_max_ms = 0; s->tx_wait_total_ms = 0;
struct ETCP_ROUTER_TX_STATS net = {0};
etcp_router_tx_stats(ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_AUDIO, get_time_us(), &net);
const char *wait = !s->audio_send.conn ? "idle" : !net.ready ? "handshake" : net.no_route ? "no-route"
const char *wait = s->tx_q && s->tx_q->count < (int)s->audio_batch.target ? "frames"
: !s->audio_send.conn ? "unregistered" : net.no_route ? "no-route" : !net.ready ? "handshake"
: net.receive_busy ? "peer-busy" : rconn && net.inflight_packets >= rconn->inflight_limit ? "window" : "transport";
DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "audio batch flow call=%llx wait=%s queued=%u/%llu inflight=%u/%llu "
DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "audio batch flow call=%llx target=%u wait=%s queued=%u/%llu inflight=%u/%llu "
"enqueue_ack_us=%llu submit_ack_us=%u ack_bytes_500ms=%llu tx_frames=%u tx_batches=%u rx_frames=%u rx_batches=%u",
(unsigned long long)s->call_id, wait, net.queued_packets, (unsigned long long)net.queued_age_us,
(unsigned long long)s->call_id, s->audio_batch.target, wait, net.queued_packets, (unsigned long long)net.queued_age_us,
net.inflight_packets, (unsigned long long)net.inflight_age_us, (unsigned long long)net.delivery_rtt_us,
net.rtt_us, (unsigned long long)net.window.bytes, s->c_tx_frames, s->tx_batches, s->c_rx_frames, s->rx_batches);
call_tx_flush(s); /* Повтор регистрации после reset/ошибки выделения; обычный drain работает по событиям. */
@@ -1000,10 +1006,25 @@ void call_set_audio_ops(struct UTUN_INSTANCE* inst, const struct call_audio_ops*
/* ─────────────────── TX медиа: tx_q + backpressure ─────────────────── */
static void call_audio_batch_tick(void* arg) {
struct call_session *s = arg; s->audio_batch_timer = NULL;
if (s->state != CALL_ST_ACTIVE) return;
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;
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);
call_tx_flush(s);
s->audio_batch_timer = uasync_set_timeout(s->ctx->inst->ua, 500, s, call_audio_batch_tick, "audio_batch");
if (!s->audio_batch_timer) DEBUG_ERROR(DEBUG_CATEGORY_CALL, "audio batch control timer allocation failed");
}
/* Пачка формируется по готовности Router; до принятия все кадры остаются в FIFO сервиса. */
static void call_tx_ready(struct ETCP_ROUTER_CONN* rconn, void* arg) {
struct call_session *s = arg;
if (s->state != CALL_ST_ACTIVE || !s->tx_q || !s->tx_q->head) return;
if (s->tx_q->count < (int)s->audio_batch.target) return; /* target мог вырасти после регистрации. */
struct call_audio_pending *first = (struct call_audio_pending*)s->tx_q->head->data;
struct call_audio_batch_view batch = {.header = {.subcmd = CALL_SUBCMD_MEDIA, .call_id = s->call_id,
.track_id = CALL_TRACK_AUDIO, .capture_epoch = first->timing.epoch,
@@ -1011,7 +1032,7 @@ static void call_tx_ready(struct ETCP_ROUTER_CONN* rconn, void* arg) {
size_t size = CALL_MEDIA_HDR_SIZE;
struct ll_entry *next = s->tx_q->head;
const char *reason = "available";
while (next && batch.header.count < CALL_AUDIO_BATCH_MAX) {
while (next && batch.header.count < s->audio_batch.target) {
struct call_audio_pending *p = (struct call_audio_pending*)next->data;
if (p->timing.epoch != first->timing.epoch || p->timing.delay_samples != first->timing.delay_samples) {
reason = "capture-change"; break;
@@ -1021,7 +1042,7 @@ static void call_tx_ready(struct ETCP_ROUTER_CONN* rconn, void* arg) {
.header = {p->seq, p->timing.pts_us, next->len}, .opus = next->dgram};
size += CALL_AUDIO_FRAME_HDR_SIZE + next->len; next = next->next;
}
if (batch.header.count == CALL_AUDIO_BATCH_MAX) reason = "frames";
if (batch.header.count == s->audio_batch.target) reason = "target";
uint8_t wire[CALL_MEDIA_MAX_BYTES];
size_t encoded = 0;
if (call_audio_batch_encode(&batch, wire, sizeof(wire), &encoded)) return;
@@ -1042,8 +1063,8 @@ static void call_tx_ready(struct ETCP_ROUTER_CONN* rconn, void* arg) {
queue_dgram_free(e); queue_entry_free(e);
}
s->c_tx_frames += count; s->tx_batches++;
DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "audio batch TX call=%llx frames=%u bytes=%zu oldest_us=%llu first_seq=%u reason=%s remaining=%d",
(unsigned long long)s->call_id, count, encoded, (unsigned long long)age,
DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "audio batch TX call=%llx frames=%u target=%u bytes=%zu oldest_us=%llu first_seq=%u reason=%s remaining=%d",
(unsigned long long)s->call_id, count, s->audio_batch.target, encoded, (unsigned long long)age,
batch.frames[0].header.seq, reason, queue_entry_count(s->tx_q));
if (last_pts != INT64_MIN) call_video_audio_sent(s, last_pts + 20000);
call_tx_flush(s);
@@ -1051,6 +1072,7 @@ static void call_tx_ready(struct ETCP_ROUTER_CONN* rconn, void* arg) {
static void call_tx_flush(struct call_session* s) {
if (s->state != CALL_ST_ACTIVE || !s->tx_q || !s->tx_q->head || s->audio_send.conn) return;
if (s->tx_q->count < (int)s->audio_batch.target) return;
struct ETCP_ROUTER_CONN *rconn = etcp_router_conn_get(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_AUDIO);
if (etcp_router_request_send(rconn, &s->audio_send, call_tx_ready, s))
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "audio router request failed call=%llx; retaining Opus", (unsigned long long)s->call_id);
src/call/call_priv.h
+2
View File
@@ -35,6 +35,8 @@ struct call_session {
uint8_t connection_reported, connection_type; /* Было событие CONNECTION / последний router path.type (0..3). */
uint64_t connection_next_hop, last_media_tb; /* Кэш next hop и время последнего RX; watchdog использует traffic_timer. */
uint16_t tx_media_seq; /* Следующий sequence аудиопакета с wrap uint16; расходуется до выделения TX entry. */
struct call_audio_batch_control audio_batch; /* Прежний адаптивный выбор обязательного числа кадров. */
void *audio_batch_timer; /* Снимок Router и пересчёт target каждые 50мс; не отправляет неполную пачку. */
uint32_t tx_batches, rx_batches; /* Число Router-кодограмм, отдельно от аудиокадров. */
struct ll_queue *tx_q; /* FIFO Opus с локальными seq/PTS/created_us в entry->data; до 750, overflow удаляет голову. */
struct router_send_request audio_send; /* Слот отправки: до callback кадры остаются в tx_q. */