Wake video transmission on router queue readiness without polling delay

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

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 <assert.h>
 #include <stdio.h>
 
@@ -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");