Expose RX frame depth and wake decoding when complete video arrives

src/call/call_video_port.c

@@ -30,7 +30,9 @@ static struct {
     int64_t tx_wait_started;
     unsigned tx_retries;
     unsigned count, bytes;
-    int64_t rx_wait_started; // Полный FIFO возобновляет приём после drain, а не по одному освободившемуся месту.
+    int64_t rx_wait_started; // Диагностика backpressure; свободное место сразу доступно producer.
+    void (*rx_wake)(void *);
+    void *rx_wake_arg;
     uint64_t decoded_generation;
     int decoded_pending, decoded_accepted;
 } port;
@@ -60,15 +62,15 @@ static int ready(void *arg, uint32_t bytes) {
     pthread_mutex_lock(&lock);
     int result = 0;
     if (port.id == (uint64_t)(uintptr_t)arg) {
-        if (port.rx_wait_started && !port.count) {
-            DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video port RX drained call=%llu wait_us=%lld; admission resumed",
-                (unsigned long long)port.id, (long long)(call_video_playout_now_us() - port.rx_wait_started));
+        result = port.count < 8 && bytes <= 4u * 1024u * 1024u - port.bytes;
+        if (port.rx_wait_started && result) {
+            DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video port RX resumed call=%llu wait_us=%lld frames=%u",
+                (unsigned long long)port.id, (long long)(call_video_playout_now_us() - port.rx_wait_started), port.count);
             port.rx_wait_started = 0;
         }
-        result = !port.rx_wait_started && port.count < 8 && bytes <= 4u * 1024u * 1024u - port.bytes;
         if (!result && !port.rx_wait_started) {
             port.rx_wait_started = call_video_playout_now_us();
-            DEBUG_WARN(DEBUG_CATEGORY_CALL, "video port RX full call=%llu frames=%u bytes=%u next_bytes=%u; drain before resume",
+            DEBUG_WARN(DEBUG_CATEGORY_CALL, "video port RX full call=%llu frames=%u bytes=%u next_bytes=%u; retain until space",
                 (unsigned long long)port.id, port.count, port.bytes, bytes);
         }
     }
@@ -82,6 +84,7 @@ static void frame(void *arg, const struct call_video_frame *f) {
         if (data) {
             memcpy(data, f->data, f->size); port.rx[port.count] = *f;
             port.rx[port.count].received_us = received; port.rx[port.count++].data = data; port.bytes += f->size;
+            if (port.rx_wake) port.rx_wake(port.rx_wake_arg);
         } else { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video port RX allocation failed bytes=%u", f->size); port.key = 1; schedule(); }
     } else { DEBUG_WARN(DEBUG_CATEGORY_CALL, "video port RX changed during delivery"); port.key = 1; schedule(); }
     pthread_mutex_unlock(&lock);
@@ -228,6 +231,7 @@ int call_video_port_open(uint64_t id, const struct call_video_caps *caps) {
     if (!result) {
         if (port.id != id) {
             clear_frames(); port.status = (struct call_video_port_status){0};
+            port.rx_wake = NULL; port.rx_wake_arg = NULL;
             port.configure = port.completed = port.resume = port.key = port.range = port.decoded_pending = 0;
         }
         port.id = id; port.local = *caps; port.opening = 1;
@@ -238,6 +242,7 @@ int call_video_port_open(uint64_t id, const struct call_video_caps *caps) {
 void call_video_port_close(uint64_t id) {
     pthread_mutex_lock(&lock);
     if (port.id == id) {
+        port.rx_wake = NULL; port.rx_wake_arg = NULL;
         clear_frames(); port.id = 0; port.opening = 1;
         port.configure = port.completed = port.resume = port.key = port.range = port.decoded_pending = 0;
         if (port.inst) schedule();
@@ -248,9 +253,18 @@ void call_video_port_close(uint64_t id) {
 int call_video_port_status(uint64_t id, struct call_video_port_status *out) {
     pthread_mutex_lock(&lock);
     int result = id && id == port.id && port.inst && out ? 0 : -1;
-    if (!result) { *out = port.status; out->can_encode = out->can_encode && !port.tx_count && !port.tx_processing; }
+    if (!result) {
+        *out = port.status; out->can_encode = out->can_encode && !port.tx_count && !port.tx_processing;
+        out->rx_frames = port.count;
+    }
     pthread_mutex_unlock(&lock); return result;
 }
+int call_video_port_set_wake(uint64_t id, void (*wake)(void *), void *arg) {
+    pthread_mutex_lock(&lock);
+    int bound = id && id == port.id && port.inst;
+    if (bound) { port.rx_wake = wake; port.rx_wake_arg = arg; }
+    pthread_mutex_unlock(&lock); return bound;
+}
 int call_video_port_format(uint64_t id, const struct call_video_format *format, uint32_t bps) {
     if (!call_video_format_valid(format)) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video port invalid format"); return -1; }
     pthread_mutex_lock(&lock);

src/call/call_video_port.h

@@ -19,12 +19,16 @@ struct call_video_port_status {
     int can_encode; // Новые raw входы допустимы; готовый TX всё равно сохраняется и повторяется.
     uint64_t decoded_generation;
     int decoded_accepted;
+    unsigned rx_frames; // Полные encoded кадры порта; pending decoder/GUI учитывает RX worker.
 };
 int call_video_port_attach(struct UTUN_INSTANCE *inst);
 void call_video_port_detach(struct UTUN_INSTANCE *inst);
 int call_video_port_open(uint64_t id, const struct call_video_caps *caps);
 void call_video_port_close(uint64_t id);
 int call_video_port_status(uint64_t id, struct call_video_port_status *out);
+/* Уведомление RX без обращения к port API из callback; вызывается под mutex порта.
+ * 1 — установлен, 0 — звонок ещё не открыт/уже закрыт. NULL снимает callback до освобождения arg. */
+int call_video_port_set_wake(uint64_t id, void (*wake)(void *), void *arg);
 int call_video_port_format(uint64_t id, const struct call_video_format *format, uint32_t bps);
 int call_video_port_applied(uint64_t id, uint64_t generation, uint64_t request, uint32_t bps);
 #define CALL_VIDEO_PORT_TX_FRAMES 8u