From 0eb743d62774296f48af49d6ccffed6908c16316 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 19 Oct 2026 17:22:51 +0200 Subject: [PATCH] Retain video mailbox tails and pause raw input under backpressure --- src/call/call_video_port.c | 37 +++++++++++++++++-------- src/call/call_video_port.h | 1 + tools/chatgui/src/call_video_engine.cpp | 30 ++++++++++++++++---- 3 files changed, 51 insertions(+), 17 deletions(-) diff --git a/src/call/call_video_port.c b/src/call/call_video_port.c index 4ffda4ab..aa9a2d9a 100644 --- a/src/call/call_video_port.c +++ b/src/call/call_video_port.c @@ -26,6 +26,7 @@ static struct { uint32_t minimum, maximum; struct call_video_frame tx[CALL_VIDEO_PORT_TX_FRAMES], rx[8]; unsigned tx_count; + int tx_processing; unsigned count, bytes; uint64_t decoded_generation; int decoded_pending, decoded_accepted; @@ -86,7 +87,15 @@ static void decoded(void *arg, uint64_t generation, int accepted) { } pthread_mutex_unlock(&lock); } -static const struct call_video_ops ops = {peer, ready, frame, need_key, bitrate, decoded}; +static void writable(void *arg, int admit) { + pthread_mutex_lock(&lock); + if (port.id == (uint64_t)(uintptr_t)arg) { + port.status.can_encode = admit; + if (port.tx_count && !port.tx_processing) schedule(); + } + pthread_mutex_unlock(&lock); +} +static const struct call_video_ops ops = {peer, ready, frame, need_key, bitrate, decoded, writable}; /* Снимок команды под mutex; API транспорта вызывается без mutex, callbacks могут войти обратно. */ static void process(void *arg) { @@ -106,7 +115,7 @@ static void process(void *arg) { uint64_t generation = port.applied_generation, request = port.applied_request; 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; + memcpy(tx, port.tx, tx_count * sizeof(*tx)); port.tx_count = 0; port.tx_processing = tx_count != 0; port.opening = port.configure = port.completed = port.resume = port.key = 0; port.range = 0; port.decoded_pending = 0; @@ -147,19 +156,25 @@ static void process(void *arg) { sent = call_video_send_frame(inst, &tx[sent_count]); if (sent != 1) break; } - if (sent_count < tx_count) - DEBUG_WARN(DEBUG_CATEGORY_CALL, "video port TX frames released call=%llu accepted=%u unsent=%u result=%d " + if (sent_count < tx_count && sent < 0) + DEBUG_WARN(DEBUG_CATEGORY_CALL, "video port TX chain failed call=%llu accepted=%u unsent=%u result=%d " "first_unsent=%llu last=%llu", (unsigned long long)id, sent_count, tx_count - sent_count, sent, (unsigned long long)tx[sent_count].frame_id, (unsigned long long)tx[tx_count - 1].frame_id); } - for (unsigned i = 0; i < tx_count; ++i) u_free((void *)tx[i].data); pthread_mutex_lock(&lock); + port.tx_processing = 0; if (port.id == id) { port.status.minimum_bps = minimum; port.status.maximum_bps = maximum; 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; + if (!sent && active) { + port.tx_count = tx_count - sent_count; + memcpy(port.tx, tx + sent_count, port.tx_count * sizeof(*tx)); + } + if (sent < 0) ++port.status.key_serial; } + unsigned release = !sent && active && port.id == id ? sent_count : tx_count; + for (unsigned i = 0; i < release; ++i) u_free((void *)tx[i].data); pthread_mutex_unlock(&lock); } int call_video_port_attach(struct UTUN_INSTANCE *inst) { @@ -205,7 +220,7 @@ 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; + if (!result) { *out = port.status; out->can_encode = out->can_encode && !port.tx_count && !port.tx_processing; } pthread_mutex_unlock(&lock); return result; } int call_video_port_format(uint64_t id, const struct call_video_format *format, uint32_t bps) { @@ -237,7 +252,7 @@ int call_video_port_send(const struct call_video_frame *frames, unsigned count) } pthread_mutex_lock(&lock); int result = frames[0].call_id && frames[0].call_id == port.id && port.inst ? 0 : -1; - if (!result && !port.tx_count) { + if (!result && !port.tx_count && !port.tx_processing) { unsigned copied = 0; for (; copied < count; ++copied) { uint8_t *data = u_malloc(frames[copied].size); @@ -253,9 +268,9 @@ int call_video_port_send(const struct call_video_frame *frames, unsigned count) result = -1; } } - if (result == 0) DEBUG_WARN(DEBUG_CATEGORY_CALL, - "video port TX mailbox busy call=%llu frames=%u first=%llu pending=%u; batch not queued", - (unsigned long long)frames[0].call_id, count, (unsigned long long)frames[0].frame_id, port.tx_count); + if (result == 0) DEBUG_TRACE(DEBUG_CATEGORY_CALL, + "video port TX retry call=%llu frames=%u first=%llu pending=%u processing=%d", + (unsigned long long)frames[0].call_id, count, (unsigned long long)frames[0].frame_id, port.tx_count, port.tx_processing); pthread_mutex_unlock(&lock); return result; } int call_video_port_pull(uint64_t id, struct call_video_frame *out) { diff --git a/src/call/call_video_port.h b/src/call/call_video_port.h index 42e0005c..89f656fc 100644 --- a/src/call/call_video_port.h +++ b/src/call/call_video_port.h @@ -16,6 +16,7 @@ struct call_video_port_status { uint32_t rate_bps; uint32_t minimum_bps, maximum_bps; int have_peer, error; + int can_encode; // Новые raw входы допустимы; готовый TX всё равно сохраняется и повторяется. uint64_t decoded_generation; int decoded_accepted; }; diff --git a/tools/chatgui/src/call_video_engine.cpp b/tools/chatgui/src/call_video_engine.cpp index 65be010d..6a80fcb1 100644 --- a/tools/chatgui/src/call_video_engine.cpp +++ b/tools/chatgui/src/call_video_engine.cpp @@ -191,6 +191,10 @@ void CallVideoEngine::transmit() { int64_t reportAt = nowUs(), encodeMax = 0, inputAgeMax = 0; uint64_t captureBase = 0; unsigned encodedCount = 0, keyCount = 0, sendBusy = 0, expired = 0; + QList pendingOutputs; + call_video_frame pendingPackets[CALL_VIDEO_PORT_TX_FRAMES]{}; + unsigned pendingCount = 0; + uint64_t pendingRevision = 0; uint64_t wakeSerial = 0; auto idle = [this, &wakeSerial] { std::unique_lock guard(m->lock); @@ -208,6 +212,18 @@ void CallVideoEngine::transmit() { clockCorrection = m->clockCorrection; image = m->capturedImage; wakeSerial = m->wakeSerial; } int64_t now = nowUs(); + if (pendingCount) { + if (!enabled || revision != pendingRevision) { + DEBUG_INFO(DEBUG_CATEGORY_CALL, "video TX cancelled by mode change call=%llu frames=%u", + (unsigned long long)m->id, pendingCount); + pendingCount = 0; pendingOutputs.clear(); forceKey = true; + } else { + int sent = call_video_port_send(pendingPackets, pendingCount); + if (!sent) { ++sendBusy; idle(); continue; } + if (sent < 0) { forceKey = true; m->report(tr("Video transport rejected encoded frames")); } + pendingCount = 0; pendingOutputs.clear(); + } + } if (now - reportAt >= 1000000) { DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "video capture flow call=%llu interval_us=%lld captured=%llu encoded=%u keys=%u send_busy=%u " @@ -258,6 +274,7 @@ void CallVideoEngine::transmit() { call_video_port_applied(m->id, format.generation, request, 0); } #ifdef HAVE_STREAM_CAPTURE + if (!state.can_encode) { idle(); continue; } if (image.isNull() || nowUs() - pts > CallCameraMaxFrameAgeUs) { DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX skips old capture age_us=%lld", (long long)(nowUs() - pts)); lastPts = pts; ++expired; continue; @@ -279,8 +296,9 @@ void CallVideoEngine::transmit() { 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) { + pendingOutputs = std::move(outputs); pendingRevision = revision; + auto* packets = pendingPackets; unsigned count = 0; + for (const auto& output : pendingOutputs) { ++encodedCount; if (output.key) ++keyCount; auto& packet = packets[count++]; packet.call_id = m->id; packet.frame_id = ++frameId; packet.pts_us = output.pts; packet.format = format; @@ -288,15 +306,15 @@ void CallVideoEngine::transmit() { packet.flags = output.key ? CALL_VIDEO_KEY : 0; } { std::lock_guard guard(m->lock); - if (!m->running || !m->enabled || m->revision != revision || nowUs() - packets[0].pts_us > CallCameraMaxFrameAgeUs) { + if (!m->running || !m->enabled || m->revision != revision) { 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; + expired += count; forceKey = true; pendingOutputs.clear(); continue; } - forceKey = call_video_port_send(packets, count) != 1; } + pendingCount = count; forceKey = false; } for (unsigned i = 0; i < count; ++i) { - const auto& output = outputs[i]; + const auto& output = pendingOutputs[i]; // Одинаковое окно исходного PTS на всех этапах: примерно один кадр в секунду, без трассы каждого кадра. if (output.pts % 1000000 < 34000) DEBUG_DEBUG(DEBUG_CATEGORY_CALL,