diff --git a/src/call/call_video_transport.c b/src/call/call_video_transport.c index bf47df1c..2e5faa32 100644 --- a/src/call/call_video_transport.c +++ b/src/call/call_video_transport.c @@ -22,6 +22,11 @@ struct call_video_transport { uint64_t local_revision, peer_revision, acknowledged_revision, caps_sent_tb; int64_t audio_end_pts, tx_last_pts; int have_peer, need_key; + struct call_video_rate rate; + struct route_tx_meter production; + uint64_t rate_generation, rate_request, produced_id, route_local_epoch, route_peer_epoch; + uint32_t applied_bps, maximum_bps; + int rate_pending, admit; }; static struct call_session *active(struct UTUN_INSTANCE *inst, uint64_t id) { @@ -36,6 +41,7 @@ static struct call_video_transport *ensure(struct call_session *s) { v->receiver = call_video_receiver_create(); if (!v->receiver) { u_free(v); return NULL; } v->session = s; v->need_key = 1; v->audio_end_pts = v->tx_last_pts = -1; + v->admit = 1; v->timer = uasync_set_timeout(s->ctx->inst->ua, 500, v, tick, "call_video"); if (!v->timer) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video timer allocation failed"); @@ -66,6 +72,47 @@ static void discard_tx(struct call_video_transport *v, const char *reason) { u_free(v->data); v->data = NULL; v->need_key = 1; if (v->ops.need_key) v->ops.need_key(v->arg); } +static uint64_t media_bytes(uint64_t bytes, uint64_t packets) { + uint64_t headers = packets * CALL_VIDEO_HEADER_BYTES; + return bytes >= headers ? bytes - headers : 0; +} +static void update_rate(struct call_video_transport *v) { + if (!v->rate_generation || !v->applied_bps) return; + struct call_session *s = v->session; + uint64_t now = get_time_us(); + struct ETCP_ROUTER_TX_STATS net = {0}; + int found = etcp_router_tx_stats(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO, now, &net) == 0; + if (found && (net.local_epoch != v->route_local_epoch || net.peer_epoch != v->route_peer_epoch)) { + v->route_local_epoch = net.local_epoch; v->route_peer_epoch = net.peer_epoch; + call_video_rate_init(&v->rate, now); route_tx_meter_reset(&v->production, now); + DEBUG_INFO(DEBUG_CATEGORY_CALL, "video rate transport epoch changed call=%llu local=%llu peer=%llu", + (unsigned long long)s->call_id, (unsigned long long)net.local_epoch, (unsigned long long)net.peer_epoch); + } + struct route_tx_window produced = {0}; route_tx_meter_window(&v->production, now, &produced); + int held = v->data && v->tx.pts_us > v->audio_end_pts; + struct call_video_rate_sample sample = {0}; + sample.now_us = now; sample.window_end_us = now / ROUTE_TX_SAMPLE_US * ROUTE_TX_SAMPLE_US; + sample.delivered_bytes = media_bytes(net.window.bytes, net.window.packets); sample.produced_bytes = produced.bytes; + sample.debt_bytes = media_bytes(net.queued_bytes + net.inflight_bytes, (uint64_t)net.queued_packets + net.inflight_packets); + if (v->data && !held) sample.debt_bytes += v->tx.size - v->tx.offset; + sample.applied_bps = v->applied_bps; sample.minimum_bps = 64000; sample.maximum_bps = v->maximum_bps; + sample.baseline_rtt_us = net.baseline_rtt_us; sample.rtt_us = net.rtt_us; + sample.frame_interval_us = v->tx.format.fps ? 1000000 / v->tx.format.fps : 33333; + sample.debt_limit_bytes = CALL_VIDEO_MAX_FRAME; + sample.valid = found && net.window.valid && produced.valid && net.window.end_us == produced.end_us; + // Первый KEY должен создать логический канал; до этого нет транспортных измерений. + sample.ready = found ? net.ready : 1; + sample.no_route = found ? net.no_route : !etcp_router_has_route(s->ctx->inst, s->group_id, s->peer_node_id); + sample.receive_busy = net.receive_busy; sample.probing = net.probing; + sample.locally_held = held; sample.command_pending = v->rate_pending; + struct call_video_rate_result result; + if (call_video_rate_step(&v->rate, &sample, &result)) { v->admit = 0; return; } + v->admit = result.admit; + if (result.target_bps != v->applied_bps && v->ops.bitrate) { + v->rate_pending = 1; ++v->rate_request; + v->ops.bitrate(v->arg, v->rate_generation, v->rate_request, result.target_bps); + } +} static void flush(struct call_video_transport *v) { if (!v->data) return; struct call_session *s = v->session; @@ -96,6 +143,7 @@ static void tick(void *arg) { // Отдельный control может обогнать ACCEPT: повторяем до подтверждения приложения. if (v->local_revision && v->acknowledged_revision != v->local_revision && get_time_tb() - v->caps_sent_tb >= 5000) control_send(v, VIDEO_CAPS); + update_rate(v); flush(v); v->timer = uasync_set_timeout(v->session->ctx->inst->ua, 500, v, tick, "call_video"); if (!v->timer) DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video timer rearm failed call=%llu", (unsigned long long)v->session->call_id); @@ -148,6 +196,13 @@ int call_video_send_frame(struct UTUN_INSTANCE *inst, const struct call_video_fr if (!v->have_peer || v->acknowledged_revision != v->local_revision) { v->need_key = 1; return 0; } if (call_video_choose(&v->local, &v->peer, f->format.codec, f->format.width, f->format.height, f->format.fps) != f->format.codec) { DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX unsupported peer format"); return -1; } + if (v->rate_generation) { + if (f->format.generation != v->rate_generation) { DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX stale encoder generation"); return -1; } + if (f->frame_id > v->produced_id) { + route_tx_meter_ack(&v->production, get_time_us(), f->size); v->produced_id = f->frame_id; + } + if (!v->admit) { v->need_key = 1; return 0; } + } if (v->data) { v->need_key = 1; return 0; } int changed = !call_video_same_format(&f->format, &v->tx.format); if (f->format.generation < v->tx.format.generation || (changed && f->format.generation == v->tx.format.generation)) { @@ -161,6 +216,42 @@ int call_video_send_frame(struct UTUN_INSTANCE *inst, const struct call_video_fr v->tx_started_tb = get_time_tb(); v->tx_last_id = f->frame_id; v->tx_last_pts = f->pts_us; v->need_key = 0; flush(v); return 1; } +int call_video_configure_rate(struct UTUN_INSTANCE *inst, uint64_t id, uint64_t generation, + uint32_t applied, uint32_t maximum) { + struct call_session *s = active(inst, id); + struct call_video_transport *v = s ? s->video : NULL; + if (!v || !v->ops.bitrate || !generation || generation < v->rate_generation || applied < 64000 || applied > 16000000 + || maximum < 64000 || maximum > 16000000) { + DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video rate configure invalid call=%llu generation=%llu bitrate=%u cap=%u", + (unsigned long long)id, (unsigned long long)generation, applied, maximum); return -1; + } + if (v->rate_generation != generation) { + v->rate_generation = generation; v->rate_pending = 0; v->admit = 1; + call_video_rate_init(&v->rate, get_time_us()); route_tx_meter_reset(&v->production, get_time_us()); + } + v->applied_bps = applied; v->maximum_bps = maximum; + DEBUG_INFO(DEBUG_CATEGORY_CALL, "video rate configured call=%llu generation=%llu applied=%u cap=%u", + (unsigned long long)id, (unsigned long long)generation, applied, maximum); return 0; +} +int call_video_rate_applied(struct UTUN_INSTANCE *inst, uint64_t id, uint64_t generation, uint64_t request, uint32_t bps) { + struct call_session *s = active(inst, id); + struct call_video_transport *v = s ? s->video : NULL; + if (!v || generation != v->rate_generation || request != v->rate_request || !v->rate_pending) { + DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video rate stale completion call=%llu generation=%llu request=%llu", + (unsigned long long)id, (unsigned long long)generation, (unsigned long long)request); return -1; + } + if (bps && (bps < 64000 || bps > 16000000)) { + DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video rate invalid encoder result bps=%u", bps); return -1; + } + v->rate_pending = 0; v->applied_bps = bps; + if (!bps) { + v->admit = 0; + DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video encoder rejected bitrate call=%llu generation=%llu request=%llu", + (unsigned long long)id, (unsigned long long)generation, (unsigned long long)request); return -1; + } + DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video bitrate applied call=%llu generation=%llu request=%llu bps=%u", + (unsigned long long)id, (unsigned long long)generation, (unsigned long long)request, bps); return 0; +} void call_video_resume_receive(struct UTUN_INSTANCE *inst, uint64_t id) { struct call_session *s = active(inst, id); if (s) etcp_router_resume_receive(inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO); diff --git a/src/call/call_video_transport.h b/src/call/call_video_transport.h index b3c5e516..2c106c13 100644 --- a/src/call/call_video_transport.h +++ b/src/call/call_video_transport.h @@ -15,6 +15,7 @@ struct call_video_ops { int (*receive_ready)(void *arg, uint32_t bytes); void (*frame)(void *arg, const struct call_video_frame *frame); void (*need_key)(void *arg); + void (*bitrate)(void *arg, uint64_t generation, uint64_t request, uint32_t bps); }; /* Настройка активного звонка, включая receive-only. NULL ops снимает платформенные callbacks. * capabilities отражают только реально проверенные профили. 0 — успешно, -1 — ошибка. */ @@ -27,6 +28,13 @@ int call_video_send_frame(struct UTUN_INSTANCE *inst, const struct call_video_fr void call_video_resume_receive(struct UTUN_INSTANCE *inst, uint64_t call_id); /* Decoder потерял references. Запрос идемпотентный, отправляется не чаще раза в 500мс. */ void call_video_request_key(struct UTUN_INSTANCE *inst, uint64_t call_id); +/* После открытия encoder; тот же generation позволяет изменить пользовательский предел. + * Callback bitrate только ставит команду worker в очередь. Подтверждение — отдельным post. + * applied=0 в rate_applied означает отказ encoder, выключает регулятор до configure_rate. */ +int call_video_configure_rate(struct UTUN_INSTANCE *inst, uint64_t call_id, uint64_t generation, + uint32_t applied_bps, uint32_t maximum_bps); +int call_video_rate_applied(struct UTUN_INSTANCE *inst, uint64_t call_id, uint64_t generation, + uint64_t request, uint32_t applied_bps); /* Только call.c: управление общей сессией и граница уже отправленного аудио. */ int call_video_transport_bind(struct UTUN_INSTANCE *inst);