diff --git a/tools/chatgui/CMakeLists.txt b/tools/chatgui/CMakeLists.txt index 5e6d1351..3409698d 100644 --- a/tools/chatgui/CMakeLists.txt +++ b/tools/chatgui/CMakeLists.txt @@ -536,6 +536,9 @@ if(CMAKE_SYSTEM_NAME STREQUAL "Linux") target_compile_definitions(test_stream_view PRIVATE TEST_NULL_AUDIO) target_link_options(test_stream_view PRIVATE "-Wl,--wrap=ma_context_init" "-Wl,--wrap=ma_device_init" "-Wl,--wrap=audio_stream_frame_time") + if(FFMPEG_FOUND) + target_link_options(test_stream_view PRIVATE "-Wl,--wrap=avcodec_open2") + endif() endif() add_test(NAME test_stream_view COMMAND test_stream_view) set_tests_properties(test_stream_view PROPERTIES TIMEOUT 30 ENVIRONMENT "QT_QPA_PLATFORM=offscreen") diff --git a/tools/chatgui/src/streamplayer_window.cpp b/tools/chatgui/src/streamplayer_window.cpp index 32516aa5..e085c4f0 100644 --- a/tools/chatgui/src/streamplayer_window.cpp +++ b/tools/chatgui/src/streamplayer_window.cpp @@ -54,7 +54,8 @@ struct StreamPlayerWindow::Playback { int64_t position = -1, clockAt = 0, ready = -1, audioDacAt = 0, deviceLead = 0, lastShownPts = -1; int64_t decoderLead = 0, framePeriod = 33333, lastInputPts = -1, lastDecodedPts = -1; uint64_t audioFrameStart = 0, shown = 0, underruns = 0, endFrames = UINT64_MAX; - bool started = false, presenting = false, videoReady = false, ended = false, activeAudio = false, holdAudio = false, resync = false; + bool started = false, presenting = false, videoReady = false, ended = false, videoEnded = false; + bool activeAudio = false, holdAudio = false, resync = false; ~Playback() { stream_playback_audio_destroy(sound); } uint64_t played(int64_t now) const { @@ -127,7 +128,7 @@ void StreamPlayerWindow::openAudio() { if (count < 0) { p.resync = true; count = 0; } if ((unsigned)count < frames) { stream_playback_audio_silence(p.sound, frames - (unsigned)count); - if (!p.ended) p.underruns++; + if (p.presenting && !p.ended) p.underruns++; } float volume = p.volume.load(); for (int i = 0; i < count * 2; i++) pcm[i] = static_cast(pcm[i] * volume); @@ -159,7 +160,7 @@ void StreamPlayerWindow::tick() { } if (!image.isNull()) first = p.shown++ == 0; bool audioDone = !p.activeAudio || !p.audioAvailable || (p.endFrames != UINT64_MAX && p.played(nowUs()) >= p.endFrames); - finished = p.ended && p.videos.empty() && audioDone; + finished = p.ended && p.videoEnded && p.videos.empty() && audioDone; } if (!image.isNull()) { m_render->setFrame(image); m_status->setText(text("Live")); @@ -185,7 +186,7 @@ void StreamPlayerWindow::keyPressEvent(QKeyEvent* event) { else QWidget::keyPressEvent(event); } -/* Worker декодирует по запросу ближайшего времени; общая минутная адаптация не зависит от сетевого cursor. */ +/* Worker заполняет codec по готовности входа; GUI показывает только ближайшие декодированные кадры. */ void StreamPlayerWindow::decode() { auto& p = *m_playback; #ifndef HAVE_FFMPEG @@ -194,17 +195,20 @@ void StreamPlayerWindow::decode() { #else AVCodecContext* decoder = nullptr; AVFrame* frame = av_frame_alloc(); AVPacket* packet = av_packet_alloc(); SwsContext* converter = nullptr; - bool waitKey = true, opened = false; + bool waitKey = true, opened = false, finishing = false, eosSent = false; + stream_media_format videoFormat{}; uint32_t revision = 0; uint64_t decoded = 0, discarded = 0, reportedShown = 0; - int64_t deadline = nowUs() + 10000000, report = nowUs(); + int64_t deadline = nowUs() + 10000000, report = nowUs(), decoderProgress = -1, lastInputAt = -1; auto reset = [&](bool request) { - avcodec_free_context(&decoder); waitKey = true; + avcodec_free_context(&decoder); av_packet_unref(packet); waitKey = true; finishing = eosSent = false; + decoderProgress = lastInputAt = -1; { std::lock_guard guard(p.lock); stream_playback_audio_destroy(p.sound); p.sound = nullptr; p.videos.clear(); p.videoBytes = 0; p.started = p.presenting = p.videoReady = false; p.position = p.ready = -1; p.decoderLead = 0; p.framePeriod = 33333; p.lastInputPts = p.lastDecodedPts = -1; + p.ended = p.videoEnded = false; p.audioDacAt = 0; p.audioFrameStart = 0; p.holdAudio = p.resync = false; p.endFrames = UINT64_MAX; } if (request && !gui_stream_resync(p.id)) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop playback resync refused"); @@ -231,6 +235,50 @@ void StreamPlayerWindow::decode() { } p.ready = p.activeAudio ? status.ready_us : status.video_us; }; + /* Забираем выход прежде следующего входа. При FIN тот же проход дренирует codec до EOF. */ + auto drainVideo = [&] { + if (!decoder) return; + for (unsigned work = 0; work < 8 && p.running; work++) { + { std::lock_guard guard(p.lock); if (p.videos.size() >= 2) return; } + int result = avcodec_receive_frame(decoder, frame); + if (result == AVERROR(EAGAIN)) return; + if (result == AVERROR_EOF) { + std::lock_guard guard(p.lock); p.videoEnded = true; + DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop decoder drained id=%llu decoded=%llu", (unsigned long long)p.id, + (unsigned long long)decoded); return; + } + if (result < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop decoder output failed id=%llu codec=%u result=%d", + (unsigned long long)p.id, videoFormat.video_codec, result); + fail(text("Stream video decoder failed")); p.running = false; return; + } + decoded++; decoderProgress = nowUs(); + if (frame->width <= 0 || frame->width > 4096 || frame->height <= 0 || frame->height > 4096) { + DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop decoder invalid size=%dx%d", frame->width, frame->height); + fail(text("Invalid stream video dimensions")); p.running = false; return; + } + int64_t pts; + { std::lock_guard guard(p.lock); + pts = frame->best_effort_timestamp == AV_NOPTS_VALUE ? p.lastInputPts : frame->best_effort_timestamp; + p.decoderLead = std::max(p.decoderLead, p.lastInputPts - pts); p.lastDecodedPts = pts; p.videoReady = true; + if (!p.presenting) { p.presenting = true; p.clockAt = nowUs(); + DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop first decoded frame id=%llu pts=%lld decoder_lead=%lld", + (unsigned long long)p.id, (long long)pts, (long long)p.decoderLead); } + if (pts < p.clock(nowUs()) - 100000) { discarded++; continue; } } + converter = sws_getCachedContext(converter, frame->width, frame->height, (AVPixelFormat)frame->format, + frame->width, frame->height, AV_PIX_FMT_RGB24, SWS_BILINEAR, nullptr, nullptr, nullptr); + QImage image(frame->width, frame->height, QImage::Format_RGB888); + if (!converter || image.isNull()) { fail(text("Stream video conversion failed")); p.running = false; return; } + uint8_t* destination[] = {image.bits(), nullptr, nullptr, nullptr}; + int strides[] = {(int)image.bytesPerLine(), 0, 0, 0}; + if (sws_scale(converter, frame->data, frame->linesize, 0, frame->height, destination, strides) != frame->height) { + fail(text("Stream video conversion failed")); p.running = false; return; + } + if (videoFormat.rotation) image = image.transformed(QTransform().rotate(videoFormat.rotation)); + std::lock_guard guard(p.lock); + p.videoBytes += (size_t)image.sizeInBytes(); p.videos.push_back({std::move(image), pts}); + } + }; if (!frame || !packet) fail(text("Stream decoder allocation failed")); else if (!gui_stream_open(p.id, p.group, p.source, p.epoch)) fail(text("Could not connect to the stream")); else while (p.running) { @@ -238,11 +286,38 @@ void StreamPlayerWindow::decode() { { std::lock_guard guard(p.lock); resync = p.resync; } if (resync) reset(true); observe(); + drainVideo(); + if (!p.running) break; + bool space; + { std::lock_guard guard(p.lock); space = p.videos.size() < 2; } + if (space && decoder && (packet->size || (finishing && !eosSent))) { + int sent = avcodec_send_packet(decoder, packet->size ? packet : nullptr); + if (sent == 0) { + if (packet->size) { + std::lock_guard guard(p.lock); + if (p.lastInputPts >= 0) p.framePeriod = std::clamp(packet->pts - p.lastInputPts, 1, 1000000); + p.lastInputPts = packet->pts; lastInputAt = nowUs(); + } else eosSent = true; + av_packet_unref(packet); drainVideo(); + } else if (sent != AVERROR(EAGAIN)) { + DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop decoder input failed id=%llu codec=%u bytes=%d result=%d", + (unsigned long long)p.id, videoFormat.video_codec, packet->size, sent); + fail(text("Stream video decoder failed")); break; + } + } + { std::lock_guard guard(p.lock); + if (finishing && p.videoEnded) break; + if (p.videos.empty() && decoderProgress >= 0 && nowUs() - decoderProgress > 10000000 + && (packet->size || finishing || (lastInputAt >= 0 && nowUs() - lastInputAt < 1000000))) { + DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop decoder stalled id=%llu input_PTS=%lld output_PTS=%lld pending=%d FIN=%d", + (unsigned long long)p.id, (long long)p.lastInputPts, (long long)p.lastDecodedPts, packet->size, finishing); + fail(text("Stream video decoder failed")); break; + } } int64_t position, audioAhead; bool wantVideo; { std::lock_guard guard(p.lock); int64_t now = nowUs(); position = p.started ? p.clock(now) : -1; - wantVideo = p.videos.size() < 2; + wantVideo = !packet->size && p.videos.size() < 2; audioAhead = !p.activeAudio || !p.sound || stream_playback_audio_needs_input(p.sound) ? p.deviceLead + 80000 : -1; if (p.presenting && p.ready >= 0) { uint64_t played = p.played(now); @@ -262,7 +337,7 @@ void StreamPlayerWindow::decode() { } } StreamRecord record; - if (!gui_stream_pull(p.id, record, position, wantVideo, audioAhead)) { + if (finishing || !gui_stream_pull(p.id, record, position, wantVideo, audioAhead)) { if (!gui_stream_ready()) { fail(text("Chat service is stopped")); break; } if ((!opened || waitKey) && nowUs() > deadline) { fail(text("Stream connection timed out")); break; } std::this_thread::sleep_for(std::chrono::milliseconds(5)); continue; @@ -279,7 +354,7 @@ void StreamPlayerWindow::decode() { if (record.kind == StreamRecordKind::End) { std::lock_guard guard(p.lock); p.ended = true; p.holdAudio = false; if (p.sound) stream_playback_audio_finish(p.sound); - break; + finishing = true; p.videoEnded = !decoder; continue; } observe(); if (record.pts > INT64_MAX) { fail(text("Invalid stream timestamp")); break; } @@ -294,10 +369,12 @@ void StreamPlayerWindow::decode() { } if (revision != record.format.revision) { avcodec_free_context(&decoder); waitKey = true; revision = record.format.revision; + decoderProgress = lastInputAt = -1; { std::lock_guard guard(p.lock); p.videos.clear(); p.videoBytes = 0; p.videoReady = false; p.decoderLead = 0; p.framePeriod = 33333; p.lastInputPts = p.lastDecodedPts = -1; } deadline = nowUs() + 10000000; } + videoFormat = record.format; if (waitKey && !(record.flags & STREAM_MEDIA_KEY)) continue; const uint8_t* data = reinterpret_cast(record.data.constData()); const uint8_t* configuration = nullptr; size_t size = (size_t)record.data.size(), configurationSize = 0; @@ -338,50 +415,7 @@ void StreamPlayerWindow::decode() { if (av_new_packet(packet, (int)size) < 0) { fail(text("Stream decoder allocation failed")); break; } std::memcpy(packet->data, data, size); packet->pts = packet->dts = (int64_t)record.pts; - int sent = avcodec_send_packet(decoder, packet); - if (sent < 0) { - DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop video packet rejected codec=%u bytes=%zu result=%d", record.format.video_codec, size, sent); - reset(true); continue; - } - { std::lock_guard guard(p.lock); - if (p.lastInputPts >= 0) p.framePeriod = std::clamp((int64_t)record.pts - p.lastInputPts, 1, 1000000); - p.lastInputPts = (int64_t)record.pts; } - int received; - while ((received = avcodec_receive_frame(decoder, frame)) == 0) { - decoded++; - if (frame->width <= 0 || frame->width > 4096 || frame->height <= 0 || frame->height > 4096) { - DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop video dimensions invalid id=%llu codec=%u size=%dx%d", - (unsigned long long)p.id, record.format.video_codec, frame->width, frame->height); - fail(text("Invalid stream video dimensions")); p.running = false; break; - } - int64_t pts = frame->best_effort_timestamp == AV_NOPTS_VALUE ? (int64_t)record.pts : frame->best_effort_timestamp; - { std::lock_guard guard(p.lock); - p.decoderLead = std::max(0, p.lastInputPts - pts); p.lastDecodedPts = pts; p.videoReady = true; - if (!p.presenting) { p.presenting = true; p.clockAt = nowUs(); - DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop first decoded frame id=%llu pts=%lld decoder_lead=%lld", - (unsigned long long)p.id, (long long)pts, (long long)p.decoderLead); } } - { std::lock_guard guard(p.lock); if (pts < p.clock(nowUs()) - 100000) { discarded++; continue; } } - converter = sws_getCachedContext(converter, frame->width, frame->height, (AVPixelFormat)frame->format, - frame->width, frame->height, AV_PIX_FMT_RGB24, SWS_BILINEAR, nullptr, nullptr, nullptr); - QImage image(frame->width, frame->height, QImage::Format_RGB888); - if (!converter || image.isNull()) { fail(text("Stream video conversion failed")); p.running = false; break; } - uint8_t* destination[] = {image.bits(), nullptr, nullptr, nullptr}; - int strides[] = {(int)image.bytesPerLine(), 0, 0, 0}; - if (sws_scale(converter, frame->data, frame->linesize, 0, frame->height, destination, strides) != frame->height) { - fail(text("Stream video conversion failed")); p.running = false; break; - } - if (record.format.rotation) image = image.transformed(QTransform().rotate(record.format.rotation)); - std::lock_guard guard(p.lock); - if (p.videos.size() >= 4) { - DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop video preparation exceeded id=%llu frames=%zu", - (unsigned long long)p.id, p.videos.size()); - fail(text("Stream video decoder failed")); p.running = false; break; - } - p.videoBytes += (size_t)image.sizeInBytes(); p.videos.push_back({std::move(image), pts}); - } - if (received != AVERROR(EAGAIN) && received != AVERROR_EOF && p.running) { - DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop video decode rejected codec=%u result=%d", record.format.video_codec, received); reset(true); - } + if (lastInputAt < 0 || nowUs() - lastInputAt > 1000000) decoderProgress = nowUs(); if (nowUs() - report >= 1000000) { int64_t now = nowUs(); double seconds = (now - report) / 1000000.0; report = now; std::lock_guard guard(p.lock); diff --git a/tools/chatgui/tests/test_stream_view.cpp b/tools/chatgui/tests/test_stream_view.cpp index dab21507..d98d6e56 100644 --- a/tools/chatgui/tests/test_stream_view.cpp +++ b/tools/chatgui/tests/test_stream_view.cpp @@ -27,6 +27,16 @@ extern "C" { } static QList snapshots; +#if defined(TEST_NULL_AUDIO) && defined(HAVE_FFMPEG) +static std::atomic delayedDecoder{false}; +extern "C" int __real_avcodec_open2(AVCodecContext*, const AVCodec*, AVDictionary**); +extern "C" int __wrap_avcodec_open2(AVCodecContext* context, const AVCodec* codec, AVDictionary** options) { + if (delayedDecoder && av_codec_is_decoder(codec)) { + context->thread_count = 4; context->thread_type = FF_THREAD_FRAME; + } + return __real_avcodec_open2(context, codec, options); +} +#endif #ifdef TEST_NULL_AUDIO static std::atomic originalAudioCallback{nullptr}; static std::atomic nonzeroSamples{0}; @@ -309,6 +319,51 @@ private slots: #endif } + void delayedDecoderAndShortStreamDrain() { +#if defined(TEST_NULL_AUDIO) && defined(HAVE_FFMPEG) + const AVCodec* encoder = avcodec_find_encoder_by_name("libx264"); QVERIFY(encoder); + AVCodecContext* context = avcodec_alloc_context3(encoder); QVERIFY(context); + context->width = 160; context->height = 90; context->pix_fmt = AV_PIX_FMT_YUV420P; + context->time_base = {1, 25}; context->max_b_frames = 0; + AVDictionary* options = nullptr; av_dict_set(&options, "preset", "ultrafast", 0); av_dict_set(&options, "tune", "zerolatency", 0); + QCOMPARE(avcodec_open2(context, encoder, &options), 0); av_dict_free(&options); + AVFrame* frame = av_frame_alloc(); QVERIFY(frame); frame->width = 160; frame->height = 90; frame->format = context->pix_fmt; + QCOMPARE(av_frame_get_buffer(frame, 32), 0); + for (int plane = 0; plane < 3; plane++) for (int y = 0; y < (plane ? 45 : 90); y++) + memset(frame->data[plane] + y * frame->linesize[plane], plane ? 128 : 80, plane ? 80 : 160); + frame->pts = 0; QCOMPARE(avcodec_send_frame(context, frame), 0); + AVPacket* packet = av_packet_alloc(); QVERIFY(packet); QCOMPARE(avcodec_receive_packet(context, packet), 0); + QByteArray access(reinterpret_cast(packet->data), packet->size); + av_packet_free(&packet); av_frame_free(&frame); avcodec_free_context(&context); + struct Delay { Delay() { delayedDecoder = true; } ~Delay() { delayedDecoder = false; } } delay; + for (unsigned count : {3u, 8u}) { + QCOMPARE(stream_publish_finish(publisher), STREAM_OK); stream_close(publisher); publisher = nullptr; + QCOMPARE(stream_publish_open(inst, group, nullptr, &publisher), STREAM_OK); + stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); epoch = info.descriptor.source_epoch; + StreamPlayerWindow window(group, inst->node_id, epoch, "Delayed decoder"); window.show(); + QSignalSpy shown(&window, &StreamPlayerWindow::framePresented); + for (unsigned i = 0; i < count; i++) { + QByteArray data = i ? access : QByteArray(4, '\0') + access; + stream_media_block media{}; media.format = {160, 90, 0, STREAM_MEDIA_H264, 0, 1}; media.video_count = 1; + media.video[0] = {i + 1, i * 40000u, (uint32_t)data.size(), 0, (uint32_t)data.size(), + (uint8_t)(i ? 0 : STREAM_MEDIA_KEY), reinterpret_cast(data.constData())}; + uint8_t wire[7345]; size_t length = 0; + QCOMPARE(stream_media_encode(&media, wire, sizeof(wire), &length), STREAM_OK); + QCOMPARE(stream_publish_block(publisher, wire, length), STREAM_OK); QTest::qWait(i ? 1 : 30); + } + QTest::qWait(100); + if (count == 3) QCOMPARE(shown.count(), 0); + else QTRY_VERIFY_WITH_TIMEOUT(shown.count() > 0, 1500); + QCOMPARE(stream_publish_finish(publisher), STREAM_OK); + QTRY_COMPARE_WITH_TIMEOUT(shown.count(), (int)count, 2000); + QTRY_VERIFY_WITH_TIMEOUT(stream_get_info(publisher, &info) == STREAM_OK && info.owners == 1, 2000); + window.close(); QTest::qWait(10); + } +#else + QSKIP("Delayed FFmpeg decoder fixture requires Linux and FFmpeg"); +#endif + } + void finishAndDetachPendingOwner() { QCOMPARE(stream_publish_finish(publisher), STREAM_OK); QTRY_COMPARE(snapshots.last().mid(9), QByteArray("[]")); stream_close(publisher); publisher = nullptr;