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

Queue video frames while awaiting matching audio timestamps

master
evgeny 5 days ago
parent
95ab911caf
commit
209f2ceb7c
2 changed files with 92 additions and 49 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 106
      src/call/call_video_transport.c
  2. 35
      tests/test_call.c

106
src/call/call_video_transport.c
Unescape View File

@ -11,15 +11,22 @@
#include <string.h>
enum { VIDEO_CAPS = 1, VIDEO_KEY_REQUEST = 2, VIDEO_CAPS_ACK = 3, VIDEO_DECODE_OK = 4, VIDEO_DECODE_REJECT = 5 };
enum { VIDEO_TX_FRAMES = 32, VIDEO_TX_SPAN_US = 500000 };
struct video_tx_frame {
struct call_video_frame frame;
uint8_t *data;
uint64_t started_tb;
};
struct call_video_transport {
struct call_session *session;
struct call_video_caps local, peer;
struct call_video_ops ops;
void *arg, *timer;
struct call_video_receiver *receiver;
struct call_video_frame tx;
uint8_t *data;
uint64_t tx_started_tb, tx_last_id, key_request_tb;
struct video_tx_frame tx[VIDEO_TX_FRAMES];
unsigned tx_count, tx_bytes;
struct call_video_format tx_last_format;
uint64_t tx_last_id, key_request_tb;
uint64_t local_revision, peer_revision, acknowledged_revision, caps_sent_tb;
int64_t audio_end_pts, tx_last_pts;
int have_peer, need_key;
@ -74,10 +81,16 @@ static int control_send(struct call_video_transport *v, uint8_t type) {
if (type == VIDEO_CAPS) v->caps_sent_tb = get_time_tb();
return 0;
}
/* Удаление головы сохраняет порядок и владение остальными кадрами. */
static void pop_tx(struct call_video_transport *v) {
v->tx_bytes -= v->tx[0].frame.size; u_free(v->tx[0].data); --v->tx_count;
memmove(v->tx, v->tx + 1, v->tx_count * sizeof(v->tx[0]));
}
static void discard_tx(struct call_video_transport *v, const char *reason) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX discard call=%llu frame=%llu offset=%u total=%u reason=%s",
(unsigned long long)v->session->call_id, (unsigned long long)v->tx.frame_id, v->tx.offset, v->tx.size, reason);
u_free(v->data); v->data = NULL; v->need_key = 1;
DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX discard call=%llu frames=%u bytes=%u reason=%s",
(unsigned long long)v->session->call_id, v->tx_count, v->tx_bytes, reason);
while (v->tx_count) pop_tx(v);
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) {
@ -130,24 +143,26 @@ static void update_rate(struct call_video_transport *v) {
(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;
if (v->audio_end_pts >= 0 && held && v->tx.pts_us - v->audio_end_pts > v->max_audio_ahead)
v->max_audio_ahead = v->tx.pts_us - v->audio_end_pts;
int held = v->tx_count && v->tx[0].frame.pts_us > v->audio_end_pts;
if (v->audio_end_pts >= 0 && held && v->tx[0].frame.pts_us - v->audio_end_pts > v->max_audio_ahead)
v->max_audio_ahead = v->tx[0].frame.pts_us - v->audio_end_pts;
if (now - v->report_us >= 1000000) {
DEBUG_DEBUG(DEBUG_CATEGORY_CALL,
"video TX flow call=%llu offered=%u accepted=%u audio_busy=%u transport_busy=%u key_wait=%u audio_ahead_us=%lld held=%d",
"video TX flow call=%llu offered=%u accepted=%u audio_busy=%u transport_busy=%u key_wait=%u audio_ahead_us=%lld held=%d queue=%u bytes=%u",
(unsigned long long)s->call_id, v->offered, v->accepted, v->held_busy, v->transport_busy, v->key_wait,
(long long)v->max_audio_ahead, held);
(long long)v->max_audio_ahead, held, v->tx_count, v->tx_bytes);
v->report_us = now; v->offered = v->accepted = v->held_busy = v->transport_busy = v->key_wait = 0; v->max_audio_ahead = 0;
}
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;
// Ожидание аудио не является перегрузкой сети; считаем только готовые к отправке кадры.
for (unsigned i = 0; i < v->tx_count && v->tx[i].frame.pts_us <= v->audio_end_pts; i++)
sample.debt_bytes += v->tx[i].frame.size - v->tx[i].frame.offset;
sample.applied_bps = v->applied_bps; sample.minimum_bps = minimum; sample.maximum_bps = maximum;
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.frame_interval_us = v->tx_last_format.fps ? 1000000 / v->tx_last_format.fps : 33333;
sample.debt_limit_bytes = debt_limit(v, net.baseline_rtt_us);
sample.valid = found && net.window.valid && produced.valid && net.window.end_us == produced.end_us;
// Первый KEY должен создать логический канал; до этого нет транспортных измерений.
@ -164,34 +179,35 @@ static void update_rate(struct call_video_transport *v) {
}
}
static void flush(struct call_video_transport *v) {
if (!v->data) return;
if (!v->tx_count) return;
struct call_session *s = v->session;
if (get_time_tb() - v->tx_started_tb > 20000) { discard_tx(v, "deadline"); return; }
if (!v->have_peer || !call_video_format_supported(&v->local, &v->peer, &v->tx.format)) {
discard_tx(v, "capabilities_changed"); return;
}
if (v->acknowledged_revision != v->local_revision) return;
// Видео не обгоняет соответствующий аудиокадр. Отказ камеры не задерживает звук.
if (v->tx.pts_us > v->audio_end_pts) return;
struct ETCP_ROUTER_CONN *conn = etcp_router_conn_get(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO);
if (!conn) { discard_tx(v, "router_allocation"); return; }
struct ETCP_ROUTER_CONN *conn = NULL;
uint8_t wire[CALL_VIDEO_HEADER_BYTES + CALL_VIDEO_FRAGMENT_BYTES];
// Ограниченный burst одного вызова; повторяется по audio send и таймеру 50мс.
for (unsigned budget = 0; v->data && budget < 16; budget++) {
for (unsigned budget = 0; v->tx_count && budget < 16; budget++) {
struct video_tx_frame *head = &v->tx[0];
struct call_video_frame *f = &head->frame;
if (get_time_tb() - head->started_tb > 20000) { discard_tx(v, "deadline"); return; }
if (!v->have_peer || !call_video_format_supported(&v->local, &v->peer, &f->format)) {
discard_tx(v, "capabilities_changed"); return;
}
if (v->acknowledged_revision != v->local_revision || f->pts_us > v->audio_end_pts) return;
if (!conn) conn = etcp_router_conn_get(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO);
if (!conn) { discard_tx(v, "router_allocation"); return; }
struct ETCP_ROUTER_TX_STATS net;
if (!etcp_router_tx_stats(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO, get_time_us(), &net)
&& media_bytes(net.queued_bytes + net.inflight_bytes, (uint64_t)net.queued_packets + net.inflight_packets)
>= debt_limit(v, net.baseline_rtt_us)) return;
if (!etcp_router_send_q_has_room(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO)
|| queue_entry_count(conn->send_q) >= ROUTER_MAX_SEND_Q_PACKETS) return;
uint32_t remaining = v->tx.size - v->tx.offset;
v->tx.length = remaining < CALL_VIDEO_FRAGMENT_BYTES ? remaining : CALL_VIDEO_FRAGMENT_BYTES;
v->tx.data = v->data + v->tx.offset;
uint32_t remaining = f->size - f->offset;
f->length = remaining < CALL_VIDEO_FRAGMENT_BYTES ? remaining : CALL_VIDEO_FRAGMENT_BYTES;
f->data = head->data + f->offset;
size_t length;
if (call_video_fragment_encode(&v->tx, wire, sizeof(wire), &length)
if (call_video_fragment_encode(f, wire, sizeof(wire), &length)
|| etcp_router_conn_send(conn, wire, length, 0)) { discard_tx(v, "router_send"); return; }
v->tx.offset += v->tx.length;
if (v->tx.offset == v->tx.size) { u_free(v->data); v->data = NULL; }
f->offset += f->length;
if (f->offset == f->size) pop_tx(v);
}
}
static void tick(void *arg) {
@ -216,7 +232,9 @@ void call_video_transport_stop(struct call_session *s) {
if (!v) return;
s->video = NULL;
if (v->timer) uasync_cancel_timeout(s->ctx->inst->ua, v->timer);
call_video_receiver_destroy(v->receiver); u_free(v->data); u_free(v);
call_video_receiver_destroy(v->receiver);
while (v->tx_count) pop_tx(v);
u_free(v);
// Уже принятые роутером пакеты содержат call_id и не принадлежат следующему звонку.
etcp_router_resume_receive(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO);
}
@ -235,7 +253,7 @@ int call_video_configure(struct UTUN_INSTANCE *inst, uint64_t id, const struct c
if (!ops) {
memset(&v->local, 0, sizeof(v->local));
v->rate_format = (struct call_video_format){0}; v->applied_bps = 0; v->rate_pending = 0; v->admit = 0;
if (v->data) discard_tx(v, "platform_detached");
if (v->tx_count) discard_tx(v, "platform_detached");
call_video_receiver_reset(v->receiver);
}
if (v->have_peer && v->ops.capabilities) v->ops.capabilities(v->arg, &v->peer);
@ -276,25 +294,29 @@ int call_video_send_frame(struct UTUN_INSTANCE *inst, const struct call_video_fr
if (!v->admit) { v->need_key = 1; return 0; }
}
v->offered++;
if (v->data) {
if (v->tx.pts_us > v->audio_end_pts) v->held_busy++; else v->transport_busy++;
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;
}
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 + f->size > CALL_VIDEO_MAX_FRAME) { v->need_key = 1; return 0; }
if (debt + v->tx_bytes + f->size > CALL_VIDEO_MAX_FRAME) { 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)) {
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; }
v->data = u_malloc(f->size);
if (!v->data) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video TX allocation failed bytes=%u", f->size); return -1; }
memcpy(v->data, f->data, f->size); v->tx = *f; v->tx.data = v->data;
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;
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);
struct video_tx_frame *tail = &v->tx[v->tx_count++];
tail->frame = *f; tail->frame.data = tail->data = data; tail->started_tb = get_time_tb(); v->tx_bytes += f->size;
v->tx_last_id = f->frame_id; v->tx_last_pts = f->pts_us; v->tx_last_format = f->format; v->need_key = 0;
v->accepted++; flush(v); return 1;
}
int call_video_configure_rate(struct UTUN_INSTANCE *inst, uint64_t id, const struct call_video_format *format,
@ -433,7 +455,7 @@ static void receive(struct ETCP_CONN *conn, struct ll_entry *entry) {
int accepted = data[0] == VIDEO_DECODE_OK;
DEBUG_INFO(DEBUG_CATEGORY_CALL, "video remote decoder result call=%llu generation=%llu accepted=%d",
(unsigned long long)id, (unsigned long long)revision, accepted);
if (!accepted) { v->tx_rejected = 1; v->admit = 0; if (v->data) discard_tx(v, "decoder_rejected"); }
if (!accepted) { v->tx_rejected = 1; v->admit = 0; if (v->tx_count) discard_tx(v, "decoder_rejected"); }
if (v->ops.decoded) v->ops.decoded(v->arg, revision, accepted);
} else goto invalid;
} else if (v->ops.frame) {

35
tests/test_call.c
Unescape View File

@ -119,6 +119,7 @@ static const struct call_audio_ops audio_ops = {.on_media = audio_recv_cb};
static struct call_video_caps g_video_peer[2];
static uint8_t g_video_data[9001];
static int g_video_frames, g_video_ready, g_video_error;
static unsigned g_video_preroll;
static uint64_t g_rate_generation, g_rate_request;
static uint32_t g_rate_bps;
static unsigned g_rate_commands;
@ -134,6 +135,11 @@ static void video_need_key(void *arg) { g_video_keys[(size_t)arg]++; }
static int video_ready(void *arg, uint32_t bytes) { (void)arg; return bytes <= sizeof(g_video_data) && g_video_ready; }
static void video_frame(void *arg, const struct call_video_frame *f) {
(void)arg;
if (f->format.generation == 1 && f->frame_id >= 2 && f->frame_id <= 7) {
if (f->frame_id != g_video_preroll + 2 || f->pts_us != sent_timing.pts_us + (int64_t)f->frame_id * 30000
|| f->flags || f->size != sizeof(g_video_data) || memcmp(f->data, g_video_data, sizeof(g_video_data))) g_video_error = 1;
g_video_preroll++; return;
}
g_video_frames++;
if (f->size != sizeof(g_video_data) || memcmp(f->data, g_video_data, sizeof(g_video_data))
|| f->format.generation != (uint64_t)g_video_frames || f->format.codec != (g_video_frames == 1 ? CALL_VIDEO_H265 : CALL_VIDEO_H264))
@ -147,7 +153,8 @@ static const struct call_video_ops video_ops = {
};
static struct call_video_frame video_test_frame(uint64_t id, unsigned generation) {
struct call_video_frame f = {0};
f.call_id = id; f.frame_id = generation; f.pts_us = sent_timing.pts_us + generation * 30000;
f.call_id = id; f.frame_id = generation == 1 ? 1 : generation + 6;
f.pts_us = sent_timing.pts_us + (int64_t)f.frame_id * 30000;
f.format = (struct call_video_format){generation, 640, 480, 30, 0,
generation == 1 ? CALL_VIDEO_H265 : CALL_VIDEO_H264, generation == 1 ? 93 : 31, 4000000, 1};
f.flags = CALL_VIDEO_KEY; f.size = f.length = sizeof(g_video_data); f.data = g_video_data;
@ -539,9 +546,23 @@ static void call_tick(void* arg) {
break;
}
case P_VIDEO_HOLD: {
if (g_video_frames) { fprintf(stderr, "video escaped audio barrier\n"); t->result = 2; break; }
if (g_video_frames || g_video_preroll) { fprintf(stderr, "video escaped audio barrier\n"); t->result = 2; break; }
if (!t->phase_ticks) {
for (unsigned i = 2; i <= 7; i++) {
struct call_video_frame f = video_test_frame(t->call_id, 1);
f.frame_id = i; f.pts_us = sent_timing.pts_us + i * 30000; f.flags = 0;
if (call_video_send_frame(A, &f) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video audio wait lost P-frame %u", i); t->result = 2; break;
}
}
struct call_video_frame overflow = video_test_frame(t->call_id, 1);
overflow.frame_id = 8; overflow.pts_us += 700000;
if (call_video_send_frame(A, &overflow) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video audio wait exceeded queue time bound"); t->result = 2; break;
}
}
if (++t->phase_ticks < 10) break;
struct call_capture_time timing = sent_timing; timing.pts_us += 20000;
struct call_capture_time timing = sent_timing; timing.pts_us += 200000;
uint8_t opus[16] = {1};
if (call_send_media(A, t->call_id, opus, sizeof(opus), &timing)) { t->result = 2; break; }
t->phase_ticks = 0; t->phase = P_VIDEO_BUSY; break;
@ -551,7 +572,7 @@ static void call_tick(void* arg) {
if (++t->phase_ticks < 10 || g_media[IDX_B].frames != 2) break;
g_video_ready = 1; call_video_resume_receive(B, t->call_id); t->phase = P_VIDEO_RESUME; break;
case P_VIDEO_RESUME:
if (!g_video_frames || !g_rate_commands) break;
if (!g_video_frames || g_video_preroll != 6 || !g_rate_commands) break;
if (g_video_error) { t->result = 2; break; }
// Настройка меняется при незавершённой команде; старый ACK остаётся действительным.
if (chat_setting_set(A, "call_video_bitrate_range_kbps", "300:350")) { t->result = 2; break; }
@ -579,7 +600,7 @@ static void call_tick(void* arg) {
if (call_video_configure_rate(A, t->call_id, &f.format, 350000, 400000)
|| call_video_rate_applied(A, t->call_id, 1, g_rate_request, g_rate_bps) != -1) { t->result = 2; break; }
if (call_video_send_frame(A, &f) != 1) { t->result = 2; break; }
struct call_capture_time timing = sent_timing; timing.pts_us += 60000;
struct call_capture_time timing = sent_timing; timing.pts_us += 240000;
uint8_t opus[16] = {2};
if (call_send_media(A, t->call_id, opus, sizeof(opus), &timing)) { t->result = 2; break; }
t->phase = P_VIDEO_SWITCH_WAIT; break;
@ -615,7 +636,7 @@ static void call_tick(void* arg) {
case P_VIDEO_PORT_OPEN: {
struct call_video_port_status status;
if (call_video_port_status(t->call_id, &status) || !status.have_peer) break;
struct call_capture_time timing = sent_timing; timing.pts_us += 200000;
struct call_capture_time timing = sent_timing; timing.pts_us += 400000;
uint8_t opus[16] = {3};
if (call_send_media(A, t->call_id, opus, sizeof(opus), &timing)) { t->result = 2; break; }
t->phase = P_VIDEO_PORT_RECEIVE; break;
@ -623,7 +644,7 @@ static void call_tick(void* arg) {
case P_VIDEO_PORT_RECEIVE: {
struct call_video_frame received;
if (!call_video_port_pull(t->call_id, &received)) break;
int valid = received.frame_id == 4 && received.format.maximum_bps == 3000000
int valid = received.frame_id == 10 && received.format.maximum_bps == 3000000
&& received.size == sizeof(g_video_data) && !memcmp(received.data, g_video_data, sizeof(g_video_data));
u_free((void *)received.data);
if (!valid) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video port RX ownership/metadata mismatch"); t->result = 2; break; }

Write Preview
Loading…
Cancel
Save