Browse Source

call: continue bounded video flush without waiting for media ticks

master
evgeny 3 days ago
parent
commit
9886bc2fde
  1. 53
      src/call/call_video_transport.c
  2. 4
      tests/Makefile.am
  3. 119
      tests/test_call_video_flush.c

53
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;

4
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)

119
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 <assert.h>
#include <stdio.h>
/* Один 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;
}
Loading…
Cancel
Save