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

Preserve encoded chain on temporary video transport backpressure

Browse Source
This commit is contained in:
evgeny committed 2026-10-19 17:22:09 +02:00
1 parent 7cbe5375d4
commit 284a1268e4
3 files changed
+49 -63
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/call/call_video_playout.c
+27 -54
View File
@@ -5,8 +5,8 @@
#include <time.h>
#include <limits.h>
#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;
}
src/call/call_video_transport.c
+19 -7
View File
@@ -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) {
src/call/call_video_transport.h
+3 -2
View File
@@ -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. */