From baeb4973dd57cffd039f001c891f78d7dec828be Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 19 Oct 2026 19:02:51 +0200 Subject: [PATCH] Expose RX frame depth and wake decoding when complete video arrives --- src/call/call_video_port.c | 28 +++++++++++++++++++++------- src/call/call_video_port.h | 4 ++++ 2 files changed, 25 insertions(+), 7 deletions(-) diff --git a/src/call/call_video_port.c b/src/call/call_video_port.c index 1c919285..17177097 100644 --- a/src/call/call_video_port.c +++ b/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); diff --git a/src/call/call_video_port.h b/src/call/call_video_port.h index 870ae10e..f9da9189 100644 --- a/src/call/call_video_port.h +++ b/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