Передавать готовую пачку GPU атомарно через TX mailbox

src/call/call_video_port.c

@@ -24,7 +24,8 @@ static struct {
     int opening, configure, completed, resume, key;
     int range;
     uint32_t minimum, maximum;
-    struct call_video_frame tx, rx[8];
+    struct call_video_frame tx[CALL_VIDEO_PORT_TX_FRAMES], rx[8];
+    unsigned tx_count;
     unsigned count, bytes;
     uint64_t decoded_generation;
     int decoded_pending, decoded_accepted;
@@ -39,7 +40,8 @@ static int schedule(void) {
     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.tx_count; ++i) u_free((void *)port.tx[i].data);
+    port.tx_count = 0;
     for (unsigned i = 0; i < port.count; ++i) u_free((void *)port.rx[i].data);
     port.count = port.bytes = 0;
 }
@@ -102,7 +104,9 @@ static void process(void *arg) {
     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};
+    struct call_video_frame tx[CALL_VIDEO_PORT_TX_FRAMES];
+    unsigned tx_count = port.tx_count;
+    memcpy(tx, port.tx, tx_count * sizeof(*tx)); port.tx_count = 0;
     port.opening = port.configure = port.completed = port.resume = port.key = 0;
     port.range = 0;
     port.decoded_pending = 0;
@@ -120,8 +124,8 @@ static void process(void *arg) {
         }
     }
     if (inst && id && !active) {
-        DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video port cancelled stale commands call=%llu state=%d tx_bytes=%u",
-                    (unsigned long long)id, session ? session->state : -1, tx.size);
+        DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video port cancelled stale commands call=%llu state=%d tx_frames=%u",
+                    (unsigned long long)id, session ? session->state : -1, tx_count);
         // Проверка id внутри close сохраняет порт следующего звонка, открытый worker во время обработки.
         call_video_port_close(id);
     }
@@ -138,9 +142,9 @@ static void process(void *arg) {
         if (id && resume) call_video_resume_receive(inst, id);
         if (id && key) call_video_request_key(inst, id);
         if (id && decoded_pending) call_video_decode_result(inst, id, decoded_generation, decoded_accepted);
-        if (id && tx.data) sent = call_video_send_frame(inst, &tx);
+        for (unsigned i = 0; i < tx_count && sent == 1; ++i) sent = call_video_send_frame(inst, &tx[i]);
     }
-    u_free((void *)tx.data);
+    for (unsigned i = 0; i < tx_count; ++i) u_free((void *)tx[i].data);
     pthread_mutex_lock(&lock);
     if (port.id == id) {
         port.status.minimum_bps = minimum; port.status.maximum_bps = maximum;
@@ -209,16 +213,37 @@ int call_video_port_applied(uint64_t id, uint64_t generation, uint64_t request,
     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;
+int call_video_port_send(const struct call_video_frame *frames, unsigned count) {
+    if (!frames || !count || count > CALL_VIDEO_PORT_TX_FRAMES) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video port invalid TX batch count=%u", count); return -1;
+    }
+    uint32_t bytes = 0;
+    for (unsigned i = 0; i < count; ++i) {
+        const struct call_video_frame *f = &frames[i];
+        if (!f->data || !f->size || f->size > CALL_VIDEO_MAX_FRAME - bytes || f->offset || f->length != f->size
+            || f->call_id != frames[0].call_id || !call_video_same_format(&f->format, &frames[0].format)
+            || (i && (f->frame_id <= frames[i - 1].frame_id || f->pts_us <= frames[i - 1].pts_us))) {
+            DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video port invalid TX batch frame=%u count=%u bytes=%u", i, count, bytes); return -1;
+        }
+        bytes += f->size;
     }
     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; }
+    int result = frames[0].call_id && frames[0].call_id == port.id && port.inst ? 0 : -1;
+    if (!result && !port.tx_count) {
+        unsigned copied = 0;
+        for (; copied < count; ++copied) {
+            uint8_t *data = u_malloc(frames[copied].size);
+            if (!data) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video port TX allocation failed frame=%u bytes=%u", copied, bytes); break; }
+            memcpy(data, frames[copied].data, frames[copied].size); port.tx[copied] = frames[copied]; port.tx[copied].data = data;
+        }
+        if (copied == count && !schedule()) {
+            port.tx_count = count; result = 1;
+            if (count > 1) DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video port TX batch accepted call=%llu frames=%u bytes=%u",
+                (unsigned long long)port.id, count, bytes);
+        } else {
+            for (unsigned i = 0; i < copied; ++i) u_free((void *)port.tx[i].data);
+            result = -1;
+        }
     }
     pthread_mutex_unlock(&lock); return result;
 }

src/call/call_video_port.h

@@ -1,6 +1,7 @@
 /* Один платформенный видеозвонок процесса. Attach/detach — uasync; остальные API
  * потокобезопасны. Detach отменяет post до уничтожения ядра. Worker не хранит inst.
- * Один TX mailbox + восемь RX кадров <=4MiB; владение pull.data передаётся через u_free.
+ * Один TX mailbox (готовая пачка до восьми кадров <=2MiB) + восемь RX кадров <=4MiB;
+ * владение pull.data передаётся через u_free. TX не ждёт накопления пачки.
  * Никаких callbacks в UI: status/pull читает worker. Кодирование не блокирует uasync. */
 #ifndef CALL_VIDEO_PORT_H
 #define CALL_VIDEO_PORT_H
@@ -25,7 +26,10 @@ 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);
+#define CALL_VIDEO_PORT_TX_FRAMES 8u
+/* Атомарно копирует готовую пачку одного звонка. 1 — принята целиком, 0 — mailbox занят,
+ * -1 — ошибка; на 0/-1 ни один кадр не принят. count: 1..CALL_VIDEO_PORT_TX_FRAMES. */
+int call_video_port_send(const struct call_video_frame *frames, unsigned count);
 int call_video_port_pull(uint64_t id, struct call_video_frame *out);
 void call_video_port_request_key(uint64_t id);
 int call_video_port_range(uint64_t id, uint32_t minimum_bps, uint32_t maximum_bps);

tools/chatgui/src/call_video_engine.cpp

@@ -250,27 +250,37 @@ void CallVideoEngine::transmit() {
         }
         if (!encoded) { forceKey = true; m->report(encoder.error()); continue; }
         // GPU может вернуть несколько уже готовых кадров: отправляем все, не добавляя очередь в TX worker.
+        if (outputs.isEmpty()) continue;
+        if (outputs.size() > CALL_VIDEO_PORT_TX_FRAMES) {
+            DEBUG_ERROR(DEBUG_CATEGORY_CALL, "GPU output batch exceeds mailbox frames=%lld", (long long)outputs.size());
+            forceKey = true; continue;
+        }
+        call_video_frame packets[CALL_VIDEO_PORT_TX_FRAMES]{}; unsigned count = 0;
         for (const auto& output : outputs) {
             ++encodedCount; if (output.key) ++keyCount;
-            call_video_frame packet{};
+            auto& packet = packets[count++];
             packet.call_id = m->id; packet.frame_id = ++frameId; packet.pts_us = output.pts; packet.format = format;
             packet.size = packet.length = output.data.size(); packet.data = reinterpret_cast<const uint8_t*>(output.data.constData());
             packet.flags = output.key ? CALL_VIDEO_KEY : 0;
-            { std::lock_guard<std::mutex> guard(m->lock);
-              if (!m->running || !m->enabled || m->revision != revision || nowUs() - output.pts > CallCameraMaxFrameAgeUs) {
-                  DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX discards obsolete encoded frame revision=%llu/%llu enabled=%d age_us=%lld",
-                      (unsigned long long)revision, (unsigned long long)m->revision, m->enabled, (long long)(nowUs() - output.pts));
-                  ++expired; forceKey = true; continue;
-              }
-              forceKey = call_video_port_send(&packet) != 1; }
+        }
+        { std::lock_guard<std::mutex> guard(m->lock);
+          if (!m->running || !m->enabled || m->revision != revision || nowUs() - packets[0].pts_us > CallCameraMaxFrameAgeUs) {
+              DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX discards obsolete batch revision=%llu/%llu enabled=%d age_us=%lld frames=%u",
+                  (unsigned long long)revision, (unsigned long long)m->revision, m->enabled,
+                  (long long)(nowUs() - packets[0].pts_us), count);
+              expired += count; forceKey = true; continue;
+          }
+          forceKey = call_video_port_send(packets, count) != 1; }
+        for (const auto& output : outputs) {
             // Одинаковое окно исходного PTS на всех этапах: примерно один кадр в секунду, без трассы каждого кадра.
             if (output.pts % 1000000 < 34000)
                 DEBUG_DEBUG(DEBUG_CATEGORY_CALL,
                     "video timing stage=encode call=%llu frame=%llu pts=%lld mono_us=%lld encode_us=%lld accepted=%d",
-                    (unsigned long long)m->id, (unsigned long long)frameId, (long long)output.pts, (long long)nowUs(),
+                    (unsigned long long)m->id, (unsigned long long)(frameId - count + 1 + &output - outputs.constData()),
+                    (long long)output.pts, (long long)nowUs(),
                     (long long)(nowUs() - encodeStarted), !forceKey);
-            if (forceKey) { ++sendBusy; break; }
         }
+        if (forceKey) ++sendBusy;
 #else
         idle();
 #endif