diff --git a/src/call/call_video_transport.c b/src/call/call_video_transport.c index ff652e55..f0ca223c 100644 --- a/src/call/call_video_transport.c +++ b/src/call/call_video_transport.c @@ -11,15 +11,22 @@ #include 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) { diff --git a/tests/test_call.c b/tests/test_call.c index 397c031b..2a76f1de 100644 --- a/tests/test_call.c +++ b/tests/test_call.c @@ -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; }