diff --git a/src/call/call_video_port.c b/src/call/call_video_port.c index f43f1503..f2e87269 100644 --- a/src/call/call_video_port.c +++ b/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; } diff --git a/src/call/call_video_port.h b/src/call/call_video_port.h index ba70d58e..42e0005c 100644 --- a/src/call/call_video_port.h +++ b/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); diff --git a/tools/chatgui/src/call_video_engine.cpp b/tools/chatgui/src/call_video_engine.cpp index 15c137de..f1c249a1 100644 --- a/tools/chatgui/src/call_video_engine.cpp +++ b/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(output.data.constData()); packet.flags = output.key ? CALL_VIDEO_KEY : 0; - { std::lock_guard 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 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