Exercise audio continuity during video stalls and generation changes

tools/chatgui/src/streamplayer_window.cpp

@@ -255,7 +255,7 @@ void StreamPlayerWindow::decode() {
     uint64_t audioPackets = 0, videoPackets = 0, batches = 0, batchMax = 0;
     int64_t feedMax = 0, pullMax = 0, workMax = 0;
     auto fail = [&](const QString& reason, bool retry = true) {
-        p.failed = true; p.complete = true;
+        p.failed = retry; p.complete = true; p.running = false;
         DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop audio worker failed id=%llu retry=%d: %s",
             (unsigned long long)p.id, retry, qPrintable(reason));
         emit statusChanged(reason); p.notify();
@@ -455,10 +455,10 @@ void StreamPlayerWindow::decodeVideo() {
     while (p.running && !p.failed) {
         uint64_t version;
         { std::lock_guard<std::mutex> guard(p.wakeLock); version = p.wakeVersion; }
-        uint64_t current; bool space;
+        uint64_t current; bool space, videoDone;
         {
             std::lock_guard<std::mutex> guard(p.lock);
-            current = p.videoGeneration; space = p.videos.size() < 2;
+            current = p.videoGeneration; space = p.videos.size() < 2; videoDone = p.videoEnded;
         }
         if (current != generation) {
             avcodec_free_context(&decoder); av_packet_unref(packet); revision = 0; eosSent = false;
@@ -466,16 +466,22 @@ void StreamPlayerWindow::decodeVideo() {
             DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "desktop video generation id=%llu generation=%llu",
                 (unsigned long long)p.id, (unsigned long long)generation);
         }
+        if (videoDone) {
+            std::unique_lock<std::mutex> guard(p.wakeLock);
+            p.wake.wait_for(guard, std::chrono::milliseconds(50), [&] { return !p.running || p.failed || p.wakeVersion != version; });
+            continue;
+        }
         bool worked = false;
         if (decoder && space) {
             int64_t at = nowUs(); int result = avcodec_receive_frame(decoder, frame); codecTime(at);
+            { std::lock_guard<std::mutex> guard(p.lock); if (generation != p.videoGeneration) continue; }
             if (result == AVERROR_EOF) {
                 { std::lock_guard<std::mutex> guard(p.lock);
                   if (generation != p.videoGeneration) continue;
                   p.videoEnded = true; p.videoBusy = false; }
                 DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop decoder drained id=%llu decoded=%llu discarded=%llu",
                     (unsigned long long)p.id, (unsigned long long)decoded, (unsigned long long)discarded);
-                p.notify(); break;
+                p.notify(); continue;
             }
             if (result < 0 && result != AVERROR(EAGAIN)) { fail(text("Stream video decoder failed")); break; }
             if (result == 0) {
@@ -588,6 +594,7 @@ void StreamPlayerWindow::decodeVideo() {
         }
         if (space && decoder && (packet->size || (ending && !eosSent))) {
             int64_t at = nowUs(); int result = avcodec_send_packet(decoder, packet->size ? packet : nullptr); codecTime(at);
+            { std::lock_guard<std::mutex> guard(p.lock); if (generation != p.videoGeneration) continue; }
             if (result == 0) {
                 worked = true;
                 {
@@ -608,7 +615,7 @@ void StreamPlayerWindow::decodeVideo() {
         }
         if (ending && !decoder && !packet->size) {
             { std::lock_guard<std::mutex> guard(p.lock); if (generation != p.videoGeneration) continue; p.videoEnded = true; }
-            p.notify(); break;
+            p.notify(); continue;
         }
         int64_t now = nowUs();
         if (now - reportAt >= 1000000) {

tools/chatgui/tests/test_stream_view.cpp

@@ -24,6 +24,7 @@
 #include <cmath>
 #include <chrono>
 #include <deque>
+#include <thread>
 
 extern "C" {
 #include "utun_instance.h"
@@ -82,6 +83,7 @@ static QImage nativeCrop(const QImage& frame, const VideoRenderWidget& view, QPo
 }
 #if defined(TEST_NULL_AUDIO) && defined(HAVE_FFMPEG)
 static std::atomic<bool> delayedDecoder{false};
+static std::atomic<unsigned> videoPauseMs{0}, videoPauses{0};
 static AVCodecContext* delayedContext = nullptr;
 static std::deque<AVFrame*> delayedFrames;
 static bool delayedDraining = false;
@@ -96,6 +98,11 @@ extern "C" int __wrap_avcodec_open2(AVCodecContext* context, const AVCodec* code
 }
 extern "C" int __real_avcodec_receive_frame(AVCodecContext*, AVFrame*);
 extern "C" int __wrap_avcodec_receive_frame(AVCodecContext* context, AVFrame* frame) {
+    if (unsigned pause = videoPauseMs.exchange(0)) {
+        ++videoPauses;
+        DEBUG_INFO(DEBUG_CATEGORY_STREAM, "fixture video decoder stall=%ums; audio worker must continue", pause);
+        std::this_thread::sleep_for(std::chrono::milliseconds(pause));
+    }
     if (context != delayedContext) return __real_avcodec_receive_frame(context, frame);
     int result = __real_avcodec_receive_frame(context, frame);
     if (!result) {
@@ -687,6 +694,14 @@ private slots:
         QVERIFY2(shown.count() >= beforeSteady + 30, "30fps video must keep advancing with 200ms physical audio latency");
 #ifdef TEST_NULL_AUDIO
         QCOMPARE(silenceGaps.load(), gapsBefore);
+        uint64_t beforePause = nonzeroSamples.load(); unsigned pausesBefore = videoPauses.load();
+        videoPauseMs = 500;
+        QTest::qWait(650);
+        QCOMPARE(videoPauses.load(), pausesBefore + 1);
+        QCOMPARE(silenceGaps.load(), gapsBefore);
+        QVERIFY(nonzeroSamples.load() > beforePause + 40000);
+        DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] video stalled 500ms: audio continued samples=%llu new_gaps=0",
+            (unsigned long long)(nonzeroSamples.load() - beforePause));
 #endif
         {
             StreamPlayerWindow second(group, inst->node_id, epoch, "Second reader"); second.show();