From 9886bc2fde19ef61ef198112379e8f15bbea7382 Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 15 Oct 2026 16:38:49 +0200 Subject: [PATCH] call: continue bounded video flush without waiting for media ticks --- src/call/call_video_transport.c | 53 +++++++++++--- tests/Makefile.am | 4 ++ tests/test_call_video_flush.c | 119 ++++++++++++++++++++++++++++++++ 3 files changed, 167 insertions(+), 9 deletions(-) create mode 100644 tests/test_call_video_flush.c diff --git a/src/call/call_video_transport.c b/src/call/call_video_transport.c index f2c53010..a11d39b1 100644 --- a/src/call/call_video_transport.c +++ b/src/call/call_video_transport.c @@ -16,12 +16,13 @@ struct video_tx_frame { struct call_video_frame frame; uint8_t *data; uint64_t started_tb; + unsigned budget_yields; }; struct call_video_transport { struct call_session *session; struct call_video_caps local, peer; struct call_video_ops ops; - void *arg, *timer; + void *arg, *timer, *flush_pending; struct call_video_receiver *receiver; struct video_tx_frame tx[VIDEO_TX_FRAMES]; unsigned tx_count, tx_bytes; @@ -41,6 +42,8 @@ struct call_video_transport { uint64_t report_us; unsigned offered, accepted, held_busy, transport_busy, key_wait; int64_t max_audio_ahead; + uint64_t flush_scheduled_us, flush_wait_max_us; + unsigned flush_yields, flush_runs, flush_net_wait, flush_audio_wait; }; static struct call_session *active(struct UTUN_INSTANCE *inst, uint64_t id) { @@ -86,9 +89,16 @@ 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])); } +/* call_soon остаётся в очереди с NULL callback до прохода uasync; v можно освободить сразу. */ +static void cancel_flush(struct call_video_transport *v) { + if (v->flush_pending && uasync_call_soon_cancel(v->session->ctx->inst->ua, v->flush_pending) != ERR_OK) + DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video flush cancel failed call=%llu", (unsigned long long)v->session->call_id); + v->flush_pending = NULL; v->flush_scheduled_us = 0; +} static void discard_tx(struct call_video_transport *v, const char *reason) { 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); + cancel_flush(v); while (v->tx_count) pop_tx(v); v->need_key = 1; if (v->ops.need_key) v->ops.need_key(v->arg); @@ -149,13 +159,16 @@ static void update_rate(struct call_video_transport *v) { 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 queue=%u bytes=%u audio_pts=%lld queue_age_us=%llu tail_ahead_us=%lld", + "audio_ahead_us=%lld held=%d queue=%u bytes=%u audio_pts=%lld queue_age_us=%llu tail_ahead_us=%lld " + "flush_yields=%u flush_runs=%u flush_wait_max_us=%llu flush_net_wait=%u flush_audio_wait=%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, v->tx_count, v->tx_bytes, (long long)v->audio_end_pts, (unsigned long long)(v->tx_count ? (get_time_tb() - v->tx[0].started_tb) * 100 : 0), - (long long)(v->tx_count && v->audio_end_pts >= 0 ? v->tx[v->tx_count - 1].frame.pts_us - v->audio_end_pts : 0)); + (long long)(v->tx_count && v->audio_end_pts >= 0 ? v->tx[v->tx_count - 1].frame.pts_us - v->audio_end_pts : 0), + v->flush_yields, v->flush_runs, (unsigned long long)v->flush_wait_max_us, v->flush_net_wait, v->flush_audio_wait); v->report_us = now; v->offered = v->accepted = v->held_busy = v->transport_busy = v->key_wait = 0; v->max_audio_ahead = 0; + v->flush_yields = v->flush_runs = v->flush_net_wait = v->flush_audio_wait = 0; v->flush_wait_max_us = 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; @@ -182,12 +195,25 @@ static void update_rate(struct call_video_transport *v) { v->ops.bitrate(v->arg, v->rate_format.generation, v->rate_request, result.target_bps); } } +static void flush(struct call_video_transport *v); +/* Снимаем handle до callback: повторный бюджет будет поставлен в следующий batch uasync. */ +static void continue_flush(void *arg) { + struct call_video_transport *v = arg; + v->flush_pending = NULL; v->flush_runs++; + flush(v); +} static void flush(struct call_video_transport *v) { + if (v->flush_scheduled_us) { + uint64_t wait = get_time_us() - v->flush_scheduled_us; + if (wait > v->flush_wait_max_us) v->flush_wait_max_us = wait; + } + // Другой источник flush мог успеть раньше отложенного вызова: оставляем только одно продолжение. + cancel_flush(v); if (!v->tx_count) return; struct call_session *s = v->session; struct ETCP_ROUTER_CONN *conn = NULL; uint8_t wire[CALL_VIDEO_HEADER_BYTES + CALL_VIDEO_FRAGMENT_BYTES]; - // Ограниченный burst одного вызова; повторяется по audio send и таймеру 50мс. + // Ограниченный burst одного вызова. Только budget ставит немедленное продолжение; backpressure — нет. 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; @@ -195,15 +221,15 @@ static void flush(struct call_video_transport *v) { 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 (v->acknowledged_revision != v->local_revision || f->pts_us > v->audio_end_pts) { v->flush_audio_wait++; 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; + >= debt_limit(v, net.baseline_rtt_us)) { v->flush_net_wait++; 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; + || queue_entry_count(conn->send_q) >= ROUTER_MAX_SEND_Q_PACKETS) { v->flush_net_wait++; return; } 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; @@ -214,12 +240,19 @@ static void flush(struct call_video_transport *v) { if (f->offset == f->size) { if (f->pts_us % 1000000 < 34000) DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, - "video timing stage=send call=%llu frame=%llu pts=%lld boot_us=%llu bytes=%u queue_age_us=%llu", + "video timing stage=send call=%llu frame=%llu pts=%lld boot_us=%llu bytes=%u queue_age_us=%llu budget_yields=%u", (unsigned long long)f->call_id, (unsigned long long)f->frame_id, (long long)f->pts_us, - (unsigned long long)get_time_us(), f->size, (unsigned long long)((get_time_tb() - head->started_tb) * 100)); + (unsigned long long)get_time_us(), f->size, (unsigned long long)((get_time_tb() - head->started_tb) * 100), + head->budget_yields); pop_tx(v); } } + if (v->tx_count && v->acknowledged_revision == v->local_revision && v->tx[0].frame.pts_us <= v->audio_end_pts) { + v->flush_pending = uasync_call_soon(s->ctx->inst->ua, v, continue_flush); + if (v->flush_pending) { + v->flush_scheduled_us = get_time_us(); v->flush_yields++; v->tx[0].budget_yields++; + } else DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video flush continuation allocation failed call=%llu", (unsigned long long)s->call_id); + } } static void tick(void *arg) { struct call_video_transport *v = arg; @@ -242,6 +275,7 @@ void call_video_transport_stop(struct call_session *s) { struct call_video_transport *v = s->video; if (!v) return; s->video = NULL; + cancel_flush(v); if (v->timer) uasync_cancel_timeout(s->ctx->inst->ua, v->timer); call_video_receiver_destroy(v->receiver); while (v->tx_count) pop_tx(v); @@ -326,6 +360,7 @@ int call_video_send_frame(struct UTUN_INSTANCE *inst, const struct call_video_fr 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->budget_yields = 0; 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; diff --git a/tests/Makefile.am b/tests/Makefile.am index 70c0babf..aa9331dd 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -13,6 +13,7 @@ check_PROGRAMS = \ test_call_video_rate \ test_call_video_playout \ test_call_video \ + test_call_video_flush \ test_memory_pool_and_config \ test_serialize \ test_reed_solomon \ @@ -688,6 +689,9 @@ test_call_video_playout_LDADD = $(LIBUTUN) $(COMMON_LIBS) test_call_video_SOURCES = test_call_video.c test_call_video_LDADD = $(LIBUTUN) $(COMMON_LIBS) + +test_call_video_flush_SOURCES = test_call_video_flush.c +test_call_video_flush_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) test_call_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/src/chat -I$(top_srcdir)/src/dm -I$(top_srcdir)/src/call -I$(top_srcdir)/src/routing_layer -I$(top_srcdir)/src/transport_layer -I$(top_srcdir)/lib test_call_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) diff --git a/tests/test_call_video_flush.c b/tests/test_call_video_flush.c new file mode 100644 index 00000000..841aa89f --- /dev/null +++ b/tests/test_call_video_flush.c @@ -0,0 +1,119 @@ +/* Настоящие transport/fragment/uasync, управляемый router без сетевых таймеров. */ +#define etcp_router_conn_get test_router_conn_get +#define etcp_router_conn_send test_router_conn_send +#define etcp_router_tx_stats test_router_tx_stats +#define etcp_router_send_q_has_room test_router_send_q_has_room +#define etcp_router_resume_receive test_router_resume_receive +#include "../src/call/call_video_transport.c" +#undef etcp_router_conn_get +#undef etcp_router_conn_send +#undef etcp_router_tx_stats +#undef etcp_router_send_q_has_room +#undef etcp_router_resume_receive +#include +#include + +/* Один batch, без poll/timeouts: тест не может случайно продолжить TX таймером 50мс. */ +extern void process_immediate_queue(struct UASYNC *ua); + +static struct ETCP_ROUTER_CONN router; +static unsigned packets, received, room, resumed; +static uint64_t debt; +enum { FRAME_BYTES = 45001 }; +static uint8_t output[FRAME_BYTES]; + +struct ETCP_ROUTER_CONN *test_router_conn_get(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t peer, uint8_t svc) { + (void)inst; (void)group; (void)peer; assert(svc == ETCP_RT_ID_CALL_VIDEO); return &router; +} +int test_router_conn_send(struct ETCP_ROUTER_CONN *conn, const uint8_t *data, size_t len, int mode) { + assert(conn == &router && !mode); + struct call_video_frame f; + assert(!call_video_fragment_parse(data, len, &f)); + assert(f.call_id == 7 && f.frame_id == 1 && f.size == FRAME_BYTES && f.offset == received); + assert(received + f.length <= sizeof(output)); + memcpy(output + received, f.data, f.length); received += f.length; packets++; return 0; +} +int test_router_tx_stats(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t peer, uint8_t svc, + uint64_t now_us, struct ETCP_ROUTER_TX_STATS *out) { + (void)inst; (void)group; (void)peer; (void)svc; (void)now_us; + *out = (struct ETCP_ROUTER_TX_STATS){.inflight_bytes = debt, .ready = 1}; return 0; +} +int test_router_send_q_has_room(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t peer, uint8_t svc) { + (void)inst; (void)group; (void)peer; (void)svc; return room; +} +void test_router_resume_receive(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t peer, uint8_t svc) { + (void)inst; (void)group; (void)peer; (void)svc; resumed++; +} + +static struct call_video_transport *setup(struct call_session *s) { + struct call_video_transport *v = u_calloc(1, sizeof(*v)); assert(v); + v->session = s; s->video = v; + v->receiver = call_video_receiver_create(); assert(v->receiver); + v->have_peer = 1; v->local_revision = v->acknowledged_revision = 1; + v->local.encode[0] = (struct call_video_limit){1280, 720, 30, 1280 * 720 * 30, 4000000, CALL_VIDEO_BASELINE, 31}; + v->peer.decode[0] = v->local.encode[0]; + v->audio_end_pts = 1000000; + v->tx_count = 1; v->tx_bytes = FRAME_BYTES; + struct video_tx_frame *head = &v->tx[0]; + head->data = u_malloc(FRAME_BYTES); assert(head->data); + for (unsigned i = 0; i < FRAME_BYTES; i++) head->data[i] = (uint8_t)(i * 17 + i / 257); + head->started_tb = get_time_tb(); + head->frame = (struct call_video_frame){.call_id = 7, .frame_id = 1, .pts_us = 1000000, + .format = {1, 640, 480, 30, 0, CALL_VIDEO_H264, 31, 4000000, CALL_VIDEO_BASELINE}, .size = FRAME_BYTES}; + packets = received = resumed = 0; room = 1; debt = 0; + memset(output, 0, sizeof(output)); return v; +} +static void check_complete(struct call_video_transport *v) { + assert(packets == 38 && received == FRAME_BYTES && !v->tx_count && !v->tx_bytes && !v->flush_pending); + for (unsigned i = 0; i < FRAME_BYTES; i++) assert(output[i] == (uint8_t)(i * 17 + i / 257)); +} +static void cleanup(struct call_session *s) { + call_video_transport_stop(s); assert(!s->video && resumed == 1); + unsigned before = packets; + uasync_drain_immediate(s->ctx->inst->ua); + assert(packets == before); + assert(s->ctx->inst->ua->timer_alloc_count == s->ctx->inst->ua->timer_free_count); +} +int main(void) { + debug_set_level(DEBUG_LEVEL_WARN); + struct UASYNC *ua = uasync_create(); assert(ua); + struct UTUN_INSTANCE inst = {.ua = ua}; + struct call_ctx ctx = {.inst = &inst}; + struct call_session s = {.ctx = &ctx, .call_id = 7}; + router.send_q = queue_new(ua, 0, 0, 0, "video_flush_test"); assert(router.send_q); + + struct call_video_transport *v = setup(&s); + flush(v); assert(packets == 16 && v->flush_pending); + process_immediate_queue(ua); assert(packets == 32 && v->flush_pending); + process_immediate_queue(ua); check_complete(v); cleanup(&s); + puts("PASS: 45001-byte frame completes in bounded batches without audio or periodic timers; bytes preserved"); + + for (unsigned gate = 0; gate < 2; gate++) { + v = setup(&s); flush(v); + if (gate) debt = 1000000; else room = 0; + process_immediate_queue(ua); assert(packets == 16 && !v->flush_pending && v->flush_net_wait == 1); + process_immediate_queue(ua); assert(packets == 16); + room = 1; debt = 0; flush(v); uasync_drain_immediate(ua); check_complete(v); cleanup(&s); + } + puts("PASS: router room and inflight debt stop continuation without spinning; TX resumes exactly once"); + + v = setup(&s); v->audio_end_pts--; + flush(v); assert(!packets && !v->flush_pending); + call_video_audio_sent(&s, 1000000); uasync_drain_immediate(ua); check_complete(v); cleanup(&s); + v = setup(&s); v->acknowledged_revision = 0; + flush(v); assert(!packets && !v->flush_pending); + v->acknowledged_revision = 1; flush(v); uasync_drain_immediate(ua); check_complete(v); cleanup(&s); + puts("PASS: audio position and CAPS acknowledgement remain required"); + + v = setup(&s); flush(v); + call_video_audio_sent(&s, 1020000); assert(packets == 32); + process_immediate_queue(ua); check_complete(v); cleanup(&s); + puts("PASS: audio flush supersedes pending continuation without duplicate fragments"); + + v = setup(&s); flush(v); cleanup(&s); assert(packets == 16); + v = setup(&s); flush(v); discard_tx(v, "test_detach"); + assert(!v->tx_count && !v->flush_pending && v->need_key); + uasync_drain_immediate(ua); assert(packets == 16); cleanup(&s); + puts("PASS: stop and discard cancel pending callbacks before freeing their owner/data"); + queue_free(router.send_q); uasync_destroy(ua, 0); return 0; +}