diff --git a/src/call/call.c b/src/call/call.c index 54d0b621..7f106b15 100644 --- a/src/call/call.c +++ b/src/call/call.c @@ -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); diff --git a/src/call/call_priv.h b/src/call/call_priv.h index 123c14bc..4387a21b 100644 --- a/src/call/call_priv.h +++ b/src/call/call_priv.h @@ -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. */