From 5e37aaa2fb3633dc902820cdb529b667b7201e71 Mon Sep 17 00:00:00 2001
From: evgeny <evgeny@adaptto.ru>
Date: Wed, 14 Oct 2026 22:35:25 +0200
Subject: [PATCH] Add bounded cross-thread video port for platform call workers

---
 src/call/call_video_port.c | 190 +++++++++++++++++++++++++++++++++++++
 src/call/call_video_port.h |  31 ++++++
 2 files changed, 221 insertions(+)
 create mode 100644 src/call/call_video_port.c
 create mode 100644 src/call/call_video_port.h

diff --git a/src/call/call_video_port.c b/src/call/call_video_port.c
new file mode 100644
index 00000000..ffba4dcd
--- /dev/null
+++ b/src/call/call_video_port.c
@@ -0,0 +1,190 @@
+#include "call_video_port.h"
+#include "call_video_transport.h"
+#include "../utun_instance.h"
+#include "../../lib/u_async.h"
+#include "../../lib/mem.h"
+#include "../../lib/debug_config.h"
+#include <pthread.h>
+#include <string.h>
+
+static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
+static struct {
+    struct UTUN_INSTANCE *inst;
+    struct posted_task *pending;
+    uint64_t id, bound;
+    struct call_video_caps local;
+    struct call_video_port_status status;
+    struct call_video_format format;
+    uint32_t bps, applied_bps;
+    uint64_t applied_generation, applied_request;
+    int opening, configure, completed, resume, key;
+    struct call_video_frame tx, rx[8];
+    unsigned count, bytes;
+} port;
+
+static void process(void *arg);
+static int schedule(void) {
+    if (!port.inst) { DEBUG_WARN(DEBUG_CATEGORY_CALL, "video port: core detached"); return -1; }
+    if (port.pending) return 0;
+    struct posted_task *task = u_calloc(1, sizeof(*task));
+    if (!task) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video port task allocation failed"); return -1; }
+    task->callback = process; port.pending = task; uasync_post_reserved(port.inst->ua, task); return 0;
+}
+static void clear_frames(void) {
+    u_free((void *)port.tx.data); port.tx = (struct call_video_frame){0};
+    for (unsigned i = 0; i < port.count; ++i) u_free((void *)port.rx[i].data);
+    port.count = port.bytes = 0;
+}
+static void peer(void *arg, const struct call_video_caps *caps) {
+    pthread_mutex_lock(&lock);
+    if (port.id == (uint64_t)(uintptr_t)arg) { port.status.peer = *caps; port.status.have_peer = 1; }
+    pthread_mutex_unlock(&lock);
+}
+static int ready(void *arg, uint32_t bytes) {
+    pthread_mutex_lock(&lock);
+    int result = port.id == (uint64_t)(uintptr_t)arg && port.count < 8 && bytes <= 4u * 1024u * 1024u - port.bytes;
+    pthread_mutex_unlock(&lock); return result;
+}
+static void frame(void *arg, const struct call_video_frame *f) {
+    pthread_mutex_lock(&lock);
+    if (port.id == (uint64_t)(uintptr_t)arg && port.count < 8 && f->size <= 4u * 1024u * 1024u - port.bytes) {
+        uint8_t *data = u_malloc(f->size);
+        if (data) {
+            memcpy(data, f->data, f->size); port.rx[port.count] = *f; port.rx[port.count++].data = data; port.bytes += f->size;
+        } 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);
+}
+static void need_key(void *arg) {
+    pthread_mutex_lock(&lock);
+    if (port.id == (uint64_t)(uintptr_t)arg) ++port.status.key_serial;
+    pthread_mutex_unlock(&lock);
+}
+static void bitrate(void *arg, uint64_t generation, uint64_t request, uint32_t bps) {
+    pthread_mutex_lock(&lock);
+    if (port.id == (uint64_t)(uintptr_t)arg) {
+        port.status.rate_generation = generation; port.status.rate_request = request; port.status.rate_bps = bps;
+    }
+    pthread_mutex_unlock(&lock);
+}
+static const struct call_video_ops ops = {peer, ready, frame, need_key, bitrate};
+
+/* Снимок команды под mutex; API транспорта вызывается без mutex, callbacks могут войти обратно. */
+static void process(void *arg) {
+    (void)arg;
+    pthread_mutex_lock(&lock);
+    port.pending = NULL;
+    struct UTUN_INSTANCE *inst = port.inst;
+    uint64_t id = port.id, old = port.bound;
+    int opening = port.opening, configure = port.configure, completed = port.completed, resume = port.resume, key = port.key;
+    struct call_video_caps caps = port.local;
+    struct call_video_format format = port.format;
+    uint32_t bps = port.bps, applied_bps = port.applied_bps;
+    uint64_t generation = port.applied_generation, request = port.applied_request;
+    struct call_video_frame tx = port.tx; port.tx = (struct call_video_frame){0};
+    port.opening = port.configure = port.completed = port.resume = port.key = 0;
+    if (opening) port.bound = id;
+    pthread_mutex_unlock(&lock);
+    int result = 0, sent = 1;
+    if (inst) {
+        if (opening && old && old != id) {
+            struct call_video_caps empty = {0}; call_video_configure(inst, old, &empty, NULL, NULL);
+        }
+        if (opening && id) result = call_video_configure(inst, id, &caps, &ops, (void *)(uintptr_t)id);
+        if (id && configure) result = call_video_configure_rate(inst, id, &format, bps, format.maximum_bps);
+        if (id && completed) call_video_rate_applied(inst, id, generation, request, applied_bps);
+        if (id && resume) call_video_resume_receive(inst, id);
+        if (id && key) call_video_request_key(inst, id);
+        if (id && tx.data) sent = call_video_send_frame(inst, &tx);
+    }
+    u_free((void *)tx.data);
+    pthread_mutex_lock(&lock);
+    if (port.id == id) {
+        if (result < 0) { port.status.error = 1; DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video port configuration failed call=%llu", (unsigned long long)id); }
+        else if (configure) { port.status.configured_generation = format.generation; port.status.error = 0; }
+        if (sent != 1) ++port.status.key_serial;
+    }
+    pthread_mutex_unlock(&lock);
+}
+int call_video_port_attach(struct UTUN_INSTANCE *inst) {
+    pthread_mutex_lock(&lock);
+    int result = inst && !port.inst ? 0 : -1;
+    if (!result) { memset(&port, 0, sizeof(port)); port.inst = inst; DEBUG_INFO(DEBUG_CATEGORY_CALL, "video port attached"); }
+    else DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video port attach conflict");
+    pthread_mutex_unlock(&lock); return result;
+}
+void call_video_port_detach(struct UTUN_INSTANCE *inst) {
+    pthread_mutex_lock(&lock);
+    if (port.inst == inst) {
+        if (port.pending) uasync_cancel_post(inst->ua, port.pending);
+        clear_frames(); memset(&port, 0, sizeof(port));
+        DEBUG_INFO(DEBUG_CATEGORY_CALL, "video port detached; queues and pending task released");
+    }
+    pthread_mutex_unlock(&lock);
+}
+int call_video_port_open(uint64_t id, const struct call_video_caps *caps) {
+    if (!id || !call_video_caps_valid(caps)) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video port invalid open"); return -1; }
+    pthread_mutex_lock(&lock);
+    int result = schedule();
+    if (!result) {
+        if (port.id != id) { clear_frames(); port.status = (struct call_video_port_status){0}; }
+        port.id = id; port.local = *caps; port.opening = 1;
+        DEBUG_INFO(DEBUG_CATEGORY_CALL, "video port open requested call=%llu", (unsigned long long)id);
+    }
+    pthread_mutex_unlock(&lock); return result;
+}
+void call_video_port_close(uint64_t id) {
+    pthread_mutex_lock(&lock);
+    if (port.id == id) {
+        clear_frames(); port.id = 0; port.opening = 1; port.configure = port.completed = 0;
+        if (port.inst) schedule();
+        DEBUG_INFO(DEBUG_CATEGORY_CALL, "video port close requested call=%llu", (unsigned long long)id);
+    }
+    pthread_mutex_unlock(&lock);
+}
+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;
+    pthread_mutex_unlock(&lock); return result;
+}
+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);
+    int result = id && id == port.id ? schedule() : -1;
+    if (!result) { port.format = *format; port.bps = bps; port.configure = 1; port.status.error = 0; }
+    pthread_mutex_unlock(&lock); return result;
+}
+int call_video_port_applied(uint64_t id, uint64_t generation, uint64_t request, uint32_t bps) {
+    pthread_mutex_lock(&lock);
+    int result = id && id == port.id ? schedule() : -1;
+    if (!result) { port.applied_generation = generation; port.applied_request = request; port.applied_bps = bps; port.completed = 1; }
+    pthread_mutex_unlock(&lock); return result;
+}
+int call_video_port_send(const struct call_video_frame *f) {
+    if (!f || !f->data || !f->size || f->size > CALL_VIDEO_MAX_FRAME || f->offset || f->length != f->size) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video port invalid TX frame"); return -1;
+    }
+    pthread_mutex_lock(&lock);
+    int result = f->call_id && f->call_id == port.id && port.inst ? 0 : -1;
+    if (!result && !port.tx.data && !schedule()) {
+        uint8_t *data = u_malloc(f->size);
+        if (data) { memcpy(data, f->data, f->size); port.tx = *f; port.tx.data = data; result = 1; }
+        else { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video port TX allocation failed"); result = -1; }
+    }
+    pthread_mutex_unlock(&lock); return result;
+}
+int call_video_port_pull(uint64_t id, struct call_video_frame *out) {
+    pthread_mutex_lock(&lock);
+    int result = 0;
+    if (id && id == port.id && port.count && out) {
+        *out = port.rx[0]; port.bytes -= out->size; --port.count;
+        memmove(port.rx, port.rx + 1, port.count * sizeof(port.rx[0])); port.resume = 1; schedule(); result = 1;
+    }
+    pthread_mutex_unlock(&lock); return result;
+}
+void call_video_port_request_key(uint64_t id) {
+    pthread_mutex_lock(&lock);
+    if (id && id == port.id) { port.key = 1; schedule(); }
+    pthread_mutex_unlock(&lock);
+}
diff --git a/src/call/call_video_port.h b/src/call/call_video_port.h
new file mode 100644
index 00000000..5978ff7c
--- /dev/null
+++ b/src/call/call_video_port.h
@@ -0,0 +1,31 @@
+/* Один платформенный видеозвонок процесса. Attach/detach — uasync; остальные API
+ * потокобезопасны. Detach отменяет post до уничтожения ядра. Worker не хранит inst.
+ * Один TX mailbox + восемь RX кадров <=4MiB; владение pull.data передаётся через u_free.
+ * Никаких callbacks в UI: status/pull читает worker. Кодирование не блокирует uasync. */
+#ifndef CALL_VIDEO_PORT_H
+#define CALL_VIDEO_PORT_H
+#include "call_video.h"
+struct UTUN_INSTANCE;
+#ifdef __cplusplus
+extern "C" {
+#endif
+struct call_video_port_status {
+    struct call_video_caps peer;
+    uint64_t key_serial, rate_generation, rate_request, configured_generation;
+    uint32_t rate_bps;
+    int have_peer, error;
+};
+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);
+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);
+int call_video_port_send(const struct call_video_frame *frame);
+int call_video_port_pull(uint64_t id, struct call_video_frame *out);
+void call_video_port_request_key(uint64_t id);
+#ifdef __cplusplus
+}
+#endif
+#endif