Bound desktop codec work and drain delayed frames through EOF

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")

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<int16_t>(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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> guard(p.lock); resync = p.resync; }
         if (resync) reset(true);
         observe();
+        drainVideo();
+        if (!p.running) break;
+        bool space;
+        { std::lock_guard<std::mutex> 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<std::mutex> guard(p.lock);
+                    if (p.lastInputPts >= 0) p.framePeriod = std::clamp<int64_t>(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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<const uint8_t*>(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<std::mutex> guard(p.lock);
-          if (p.lastInputPts >= 0) p.framePeriod = std::clamp<int64_t>((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<std::mutex> guard(p.lock);
-              p.decoderLead = std::max<int64_t>(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<std::mutex> 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<std::mutex> 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<std::mutex> guard(p.lock);

tools/chatgui/tests/test_stream_view.cpp

@@ -27,6 +27,16 @@ extern "C" {
 }
 
 static QList<QByteArray> snapshots;
+#if defined(TEST_NULL_AUDIO) && defined(HAVE_FFMPEG)
+static std::atomic<bool> 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<ma_device_data_proc> originalAudioCallback{nullptr};
 static std::atomic<uint64_t> 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<const char*>(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<const uint8_t*>(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;