From 284a1268e41b62b17f020ea71226bd2766e51242 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 19 Oct 2026 17:22:09 +0200 Subject: [PATCH] Preserve encoded chain on temporary video transport backpressure --- src/call/call_video_playout.c | 81 +++++++++++---------------------- src/call/call_video_transport.c | 26 ++++++++--- src/call/call_video_transport.h | 5 +- 3 files changed, 49 insertions(+), 63 deletions(-) diff --git a/src/call/call_video_playout.c b/src/call/call_video_playout.c index c8116231..491c28ea 100644 --- a/src/call/call_video_playout.c +++ b/src/call/call_video_playout.c @@ -5,8 +5,8 @@ #include #include -#define VIDEO_LATE_US 50000 #define VIDEO_AHEAD_US 40000 +#define VIDEO_MAX_SPEED_PPM 1600000 int64_t call_video_playout_now_us(void) { #ifdef _WIN32 @@ -21,19 +21,9 @@ int64_t call_video_playout_now_us(void) { void call_video_playout_init(struct call_video_playout *p, uint64_t call_id) { memset(p, 0, sizeof(*p)); p->call_id = call_id; p->last_pts = -1; p->interval_min = p->av_min = INT64_MAX; p->av_max = INT64_MIN; - DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video playout init call=%llu speed=1.04 av_us=-50000..40000", (unsigned long long)call_id); -} - -/* 25/26 без накопления времени поздних вызовов и без переполнения delta*25. */ -static int64_t interval(int64_t delta) { return delta / 26 * 25 + delta % 26 * 25 / 26; } - -/* Аудио ограничивает фазу, а не квантует каждый кадр на 20мс. Проекция на 1x лишь - * назначает срок следующей проверки; при SHOW границы проверяются по свежему audio PTS. */ -static int64_t bounded(int64_t deadline, int64_t pts, int64_t now, int64_t audio) { - int64_t earliest = now + (pts - audio) - VIDEO_AHEAD_US, latest = now + (pts - audio) + VIDEO_LATE_US; - if (deadline < earliest) return earliest; - if (deadline > latest) return latest; - return deadline; + p->speed_ppm = 1000000; + DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video playout init call=%llu sequential=1 max_speed_ppm=%d ahead_us=%d", + (unsigned long long)call_id, VIDEO_MAX_SPEED_PPM, VIDEO_AHEAD_US); } /* Сводка агрегирует штатные ожидания и пропуски; per-frame spam не нужен. */ @@ -41,18 +31,15 @@ static void report(struct call_video_playout *p, int64_t now) { if (!p->report_us) p->report_us = now; if (now - p->report_us < 1000000) return; DEBUG_DEBUG(DEBUG_CATEGORY_CALL, - "video playout call=%llu interval_us=%lld shown=%u drop_late=%u drop_schedule=%u drop_order=%u " + "video playout call=%llu interval_us=%lld shown=%u speed_ppm=%u drop_order=%u " "cadence_us=%lld..%lld gap_pts_us=%lld gap_rx_us=%lld gap_dispatch_us=%lld " - "av_us=%lld..%lld dispatch_max_us=%lld data_shift_us=%lld audio_wait=%u clock_wait=%u " - "late_av_us=%lld..%lld late_rx_age_us=%lld..%lld", - (unsigned long long)p->call_id, (long long)(now - p->report_us), p->shown, p->late, p->skipped, p->order, + "av_us=%lld..%lld dispatch_max_us=%lld data_shift_us=%lld audio_wait=%u clock_wait=%u", + (unsigned long long)p->call_id, (long long)(now - p->report_us), p->shown, p->speed_ppm, p->order, (long long)(p->interval_min == INT64_MAX ? 0 : p->interval_min), (long long)p->interval_max, (long long)p->gap_pts, (long long)p->gap_received, (long long)p->gap_dispatch, (long long)(p->shown ? p->av_min : 0), (long long)(p->shown ? p->av_max : 0), - (long long)p->dispatch_max, (long long)p->data_shift, p->audio_wait, p->clock_wait, - (long long)(p->late ? p->late_av_min : 0), (long long)(p->late ? p->late_av_max : 0), - (long long)(p->late ? p->late_age_min : 0), (long long)(p->late ? p->late_age_max : 0)); - p->report_us = now; p->shown = p->late = p->skipped = p->order = p->audio_wait = p->clock_wait = 0; + (long long)p->dispatch_max, (long long)p->data_shift, p->audio_wait, p->clock_wait); + p->report_us = now; p->shown = p->order = p->audio_wait = p->clock_wait = 0; p->interval_min = p->av_min = INT64_MAX; p->av_max = INT64_MIN; p->interval_max = p->dispatch_max = p->data_shift = 0; p->gap_pts = p->gap_received = p->gap_dispatch = 0; @@ -60,7 +47,10 @@ static void report(struct call_video_playout *p, int64_t now) { enum call_video_playout_action call_video_playout_poll(struct call_video_playout *p, const struct call_video_playout_frame *f, const struct call_video_playout_frame *next, - int64_t now, int audio_valid, int64_t audio) { + int64_t now, const struct call_clock_snapshot *clock) { + (void)next; + int audio_valid = clock && clock->valid; + int64_t audio = clock ? clock->pts_us : -1; report(p, now); p->deadline_us = now + 5000; if (p->audio_valid != !!audio_valid) { DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video playout clock call=%llu valid=%d pts=%lld", @@ -75,39 +65,22 @@ enum call_video_playout_action call_video_playout_poll(struct call_video_playout (long long)f->received_us, (long long)now); p->order++; return CALL_VIDEO_PLAYOUT_DROP_ORDER; } - int64_t planned = p->anchored ? p->last_deadline + interval(f->pts_us - p->last_pts) : f->received_us; - int64_t available = planned > f->received_us ? planned : f->received_us; - int64_t deadline = bounded(available, f->pts_us, now, audio), av = f->pts_us - audio; + // Аудио задаёт базовый темп; долг видео снимается с постоянной 0.5с, максимум 1.6x. + int64_t av = f->pts_us - audio; + uint32_t base = clock->rate_ppm > 1000000 ? clock->rate_ppm : 1000000; + int64_t speed = base + (av < 0 ? -av * 2 : 0); + p->speed_ppm = speed < VIDEO_MAX_SPEED_PPM ? (uint32_t)speed : VIDEO_MAX_SPEED_PPM; + int64_t delta = p->last_pts >= 0 ? f->pts_us - p->last_pts : 0; + int64_t step = delta / p->speed_ppm * 1000000 + delta % p->speed_ppm * 1000000 / p->speed_ppm; + int64_t deadline = p->anchored ? p->last_deadline + step : now; p->deadline_us = deadline; - enum call_video_playout_action action = CALL_VIDEO_PLAYOUT_SHOW; - if (av < -VIDEO_LATE_US) action = CALL_VIDEO_PLAYOUT_DROP_LATE; - else if (deadline > now) { - if (av > VIDEO_AHEAD_US) p->audio_wait++; + if (av > VIDEO_AHEAD_US) { + p->deadline_us = now + 5000; p->audio_wait++; return CALL_VIDEO_PLAYOUT_WAIT; - } else if (next && next->pts_us > f->pts_us && next->received_us <= now) { - int64_t due = deadline + interval(next->pts_us - f->pts_us); - if (due < next->received_us) due = next->received_us; - if (bounded(due, next->pts_us, now, audio) <= now) action = CALL_VIDEO_PLAYOUT_DROP_SCHEDULE; } - p->last_pts = f->pts_us; p->last_deadline = deadline; p->anchored = 1; - if (available > planned) p->data_shift += available - planned; - if (action == CALL_VIDEO_PLAYOUT_DROP_LATE) { - int64_t age = now - f->received_us; - if (!p->late) { - p->late_av_min = p->late_av_max = av; p->late_age_min = p->late_age_max = age; - DEBUG_WARN(DEBUG_CATEGORY_CALL, "video late drop call=%llu pts=%lld audio_pts=%lld av_us=%lld limit_us=%d " - "rx_age_us=%lld received_us=%lld mono_us=%lld planned_us=%lld deadline_us=%lld", - (unsigned long long)p->call_id, (long long)f->pts_us, (long long)audio, (long long)av, VIDEO_LATE_US, - (long long)age, (long long)f->received_us, (long long)now, (long long)planned, (long long)deadline); - } - if (av < p->late_av_min) p->late_av_min = av; - if (av > p->late_av_max) p->late_av_max = av; - if (age < p->late_age_min) p->late_age_min = age; - if (age > p->late_age_max) p->late_age_max = age; - p->late++; - } - else if (action == CALL_VIDEO_PLAYOUT_DROP_SCHEDULE) p->skipped++; - else { + if (deadline > now) return CALL_VIDEO_PLAYOUT_WAIT; + p->last_pts = f->pts_us; p->last_deadline = now; p->anchored = 1; + { if (now - deadline > p->dispatch_max) p->dispatch_max = now - deadline; if (av < p->av_min) p->av_min = av; if (av > p->av_max) p->av_max = av; @@ -121,5 +94,5 @@ enum call_video_playout_action call_video_playout_poll(struct call_video_playout } p->last_show_us = now; p->last_show_pts = f->pts_us; p->last_received = f->received_us; p->shown++; } - return action; + return CALL_VIDEO_PLAYOUT_SHOW; } diff --git a/src/call/call_video_transport.c b/src/call/call_video_transport.c index fd3d7e99..aac39635 100644 --- a/src/call/call_video_transport.c +++ b/src/call/call_video_transport.c @@ -51,6 +51,11 @@ static struct call_session *active(struct UTUN_INSTANCE *inst, uint64_t id) { return s && s->state == CALL_ST_ACTIVE ? s : NULL; } static void tick(void *arg); +/* Admission ограничивает новые входы encoder, а готовые кадры продолжают передаваться. */ +static void writable(struct call_video_transport *v) { + if (v->ops.writable) v->ops.writable(v->arg, + v->admit && !v->tx_rejected && v->have_peer && v->acknowledged_revision == v->local_revision && !v->tx_count); +} static struct call_video_transport *ensure(struct call_session *s) { if (s->video) return s->video; struct call_video_transport *v = u_calloc(1, sizeof(*v)); @@ -200,7 +205,7 @@ static void flush(struct call_video_transport *v); static void continue_flush(void *arg) { struct call_video_transport *v = arg; v->flush_pending = NULL; v->flush_runs++; - flush(v); + flush(v); writable(v); } static void flush(struct call_video_transport *v) { if (v->flush_scheduled_us) { @@ -263,6 +268,7 @@ static void tick(void *arg) { if (v->decode_pending && !control_send(v, v->decode_result ? VIDEO_DECODE_OK : VIDEO_DECODE_REJECT)) v->decode_pending = 0; if (!v->tx_rejected) update_rate(v); flush(v); + writable(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); } @@ -270,6 +276,7 @@ void call_video_audio_sent(struct call_session *s, int64_t end_pts_us) { if (!s->video || end_pts_us < 0) return; s->video->audio_end_pts = end_pts_us; flush(s->video); + writable(s->video); } void call_video_transport_stop(struct call_session *s) { struct call_video_transport *v = s->video; @@ -323,8 +330,8 @@ int call_video_send_frame(struct UTUN_INSTANCE *inst, const struct call_video_fr (unsigned long long)(v ? v->tx_last_id : 0), (long long)(f ? f->pts_us : -1), (long long)(v ? v->tx_last_pts : -1), f ? f->size : 0, f ? f->offset : 0, f ? f->length : 0, f && call_video_format_valid(&f->format)); return -1; } - if (!v->have_peer || v->acknowledged_revision != v->local_revision) { v->need_key = 1; return 0; } - if (v->tx_rejected) return 0; + if (!v->have_peer || v->acknowledged_revision != v->local_revision) return 0; + if (v->tx_rejected) return -1; if (!call_video_format_supported(&v->local, &v->peer, &f->format)) { DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX unsupported format codec=%u level=%u bitrate_cap=%u profile=%u", f->format.codec, f->format.level, f->format.maximum_bps, f->format.profile); return -1; @@ -336,26 +343,30 @@ int call_video_send_frame(struct UTUN_INSTANCE *inst, const struct call_video_fr 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; } } v->offered++; flush(v); if (v->tx_count == VIDEO_TX_FRAMES || v->tx_bytes + f->size > CALL_VIDEO_MAX_FRAME || (v->tx_count && f->pts_us - v->tx[0].frame.pts_us > VIDEO_TX_SPAN_US)) { if (v->tx_count && v->tx[0].frame.pts_us > v->audio_end_pts) v->held_busy++; else v->transport_busy++; - v->need_key = 1; return 0; + return 0; } struct ETCP_ROUTER_TX_STATS net; if (etcp_router_tx_stats(inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO, get_time_us(), &net) == 0) { uint64_t debt = media_bytes(net.queued_bytes + net.inflight_bytes, (uint64_t)net.queued_packets + net.inflight_packets); - if (debt + v->tx_bytes + f->size > CALL_VIDEO_MAX_FRAME) { v->need_key = 1; return 0; } + if (debt + v->tx_bytes + f->size > CALL_VIDEO_MAX_FRAME) return 0; } int changed = !call_video_same_format(&f->format, &v->tx_last_format); if (f->format.generation < v->tx_last_format.generation || (changed && f->format.generation == v->tx_last_format.generation)) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video TX format changed without new generation"); return -1; } if (changed || (v->tx_last_id && f->frame_id != v->tx_last_id + 1)) v->need_key = 1; - if (v->need_key && !(f->flags & CALL_VIDEO_KEY)) { v->key_wait++; return 0; } + if (v->need_key && !(f->flags & CALL_VIDEO_KEY)) { + v->key_wait++; + DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX broken chain call=%llu frame=%llu last=%llu; KEY required", + (unsigned long long)f->call_id, (unsigned long long)f->frame_id, (unsigned long long)v->tx_last_id); + return -1; + } uint8_t *data = u_malloc(f->size); if (!data) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video TX allocation failed bytes=%u", f->size); v->need_key = 1; return -1; } memcpy(data, f->data, f->size); @@ -491,6 +502,7 @@ static void receive(struct ETCP_CONN *conn, struct ll_entry *entry) { } else if (data[0] == VIDEO_CAPS_ACK && length == 17 && revision && revision <= v->local_revision) { if (revision < v->local_revision) goto done; v->acknowledged_revision = revision; + writable(v); DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video capabilities acknowledged call=%llu revision=%llu", (unsigned long long)id, (unsigned long long)revision); } else if (data[0] == VIDEO_KEY_REQUEST && length == 17 && !revision) { diff --git a/src/call/call_video_transport.h b/src/call/call_video_transport.h index 2e324471..f244e502 100644 --- a/src/call/call_video_transport.h +++ b/src/call/call_video_transport.h @@ -17,13 +17,14 @@ struct call_video_ops { void (*need_key)(void *arg); void (*bitrate)(void *arg, uint64_t generation, uint64_t request, uint32_t bps); void (*decoded)(void *arg, uint64_t generation, int accepted); + void (*writable)(void *arg, int admit); // Разрешение новых входов encoder; также будит retry сохранённой пачки. }; /* Настройка активного звонка, включая receive-only. NULL ops снимает платформенные callbacks. * capabilities отражают только реально проверенные профили. 0 — успешно, -1 — ошибка. */ int call_video_configure(struct UTUN_INSTANCE *inst, uint64_t call_id, const struct call_video_caps *caps, const struct call_video_ops *ops, void *arg); -/* Копирует один полный кадр. 1 принят, 0 backpressure/ожидание KEY, -1 неверный формат/сессия. - * После 0 producer обязан запросить KEY: пропуск P-кадра разрывает цепочку references. +/* Копирует один полный кадр. 1 принят, 0 временная занятость, -1 неверный формат/сессия/цепочка. + * После 0 producer сохраняет кадр и повторяет с теми же ID/PTS, не запрашивая KEY. * FIFO ожидания аудио/сети: до 32 кадров, 500мс PTS и CALL_VIDEO_MAX_FRAME байт суммарно. * Каждый кадр выходит только после соответствующего аудио; deadline головы 2с. * При истечении deadline сбрасывается вся зависимая цепочка и запрашивается KEY. */