diff --git a/src/call/call_video_transport.c b/src/call/call_video_transport.c index aac39635..cececfaf 100644 --- a/src/call/call_video_transport.c +++ b/src/call/call_video_transport.c @@ -23,6 +23,7 @@ struct call_video_transport { struct call_video_caps local, peer; struct call_video_ops ops; void *arg, *timer, *flush_pending; + struct queue_waiter_handle send_waiter; // Одноразовое событие освобождения router send_q. struct call_video_receiver *receiver; struct video_tx_frame tx[VIDEO_TX_FRAMES]; unsigned tx_count, tx_bytes; @@ -96,6 +97,8 @@ static void pop_tx(struct call_video_transport *v) { } /* call_soon остаётся в очереди с NULL callback до прохода uasync; v можно освободить сразу. */ static void cancel_flush(struct call_video_transport *v) { + struct call_session *s = v->session; + etcp_router_cancel_send_ready(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO, &v->send_waiter); 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; @@ -201,6 +204,10 @@ static void update_rate(struct call_video_transport *v) { } } static void flush(struct call_video_transport *v); +static void send_ready(struct ll_queue *queue, void *arg) { + (void)queue; + struct call_video_transport *v = arg; flush(v); writable(v); +} /* Снимаем handle до callback: повторный бюджет будет поставлен в следующий batch uasync. */ static void continue_flush(void *arg) { struct call_video_transport *v = arg; @@ -234,7 +241,12 @@ static void flush(struct call_video_transport *v) { && media_bytes(net.queued_bytes + net.inflight_bytes, (uint64_t)net.queued_packets + net.inflight_packets) >= 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) { v->flush_net_wait++; return; } + || queue_entry_count(conn->send_q) >= ROUTER_MAX_SEND_Q_PACKETS) { + v->flush_net_wait++; + etcp_router_on_send_ready(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO, + &v->send_waiter, send_ready, v); + 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; @@ -502,7 +514,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); + flush(v); 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) { diff --git a/tests/test_call_video_flush.c b/tests/test_call_video_flush.c index 841aa89f..cda3b8eb 100644 --- a/tests/test_call_video_flush.c +++ b/tests/test_call_video_flush.c @@ -4,12 +4,16 @@ #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 +#define etcp_router_on_send_ready test_router_on_send_ready +#define etcp_router_cancel_send_ready test_router_cancel_send_ready #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 +#undef etcp_router_on_send_ready +#undef etcp_router_cancel_send_ready #include #include @@ -19,6 +23,8 @@ extern void process_immediate_queue(struct UASYNC *ua); static struct ETCP_ROUTER_CONN router; static unsigned packets, received, room, resumed; static uint64_t debt; +static queue_threshold_callback_fn ready_callback; +static void *ready_arg; enum { FRAME_BYTES = 45001 }; static uint8_t output[FRAME_BYTES]; @@ -44,6 +50,14 @@ int test_router_send_q_has_room(struct UTUN_INSTANCE *inst, uint64_t group, uint 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++; } +void test_router_on_send_ready(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t peer, uint8_t svc, + struct queue_waiter_handle *h, queue_threshold_callback_fn callback, void *arg) { + (void)inst; (void)group; (void)peer; (void)svc; (void)h; ready_callback = callback; ready_arg = arg; +} +void test_router_cancel_send_ready(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t peer, uint8_t svc, + struct queue_waiter_handle *h) { + (void)inst; (void)group; (void)peer; (void)svc; (void)h; ready_callback = NULL; ready_arg = NULL; +} static struct call_video_transport *setup(struct call_session *s) { struct call_video_transport *v = u_calloc(1, sizeof(*v)); assert(v); @@ -93,7 +107,10 @@ int main(void) { 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); + room = 1; debt = 0; + if (!gate) { assert(ready_callback && ready_arg == v); ready_callback(router.send_q, ready_arg); } + else 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");