QRtAudioEngine: introduce overflow buffersHEADdev

QRtAudioEngine uses fixed sized ringbuffers for communication
between rt and non-rt threads. In the unlikely case that these
ringbuffers fill up, we need to queue on the corresponding thread and
retry the next time we want to send data (rt->non-rt) or via a backoff
timer (non-rt->rt). This is rare and should ideally not happen in
practice, but if it does, we should handle it gracefully.

Pick-to: 6.10
Change-Id: I703811552d9e59a2f9dd770120b9eb3369ef8759
Reviewed-by: Timur Pocheptsov <[email protected]>

4 files changed, 144 insertions, 14 deletions

diff --git a/src/multimedia/audio/q_pmr_emulation_p.h b/src/multimedia/audio/q_pmr_emulation_p.h
index c31737536..5cd15599f 100644
--- a/src/multimedia/audio/q_pmr_emulation_p.h
+++ b/src/multimedia/audio/q_pmr_emulation_p.h
@@ -25,6 +25,11 @@
 #  endif
 #endif
 
+#if defined(Q_OS_VXWORKS)
+// std::pmr::deque<variant> doesn't seem to compile on VxWorks, so we use our emulation
+#  define QT_MM_PMR_EMULATION
+#endif
+
 #if !__has_include(<memory_resource>)
 #  define QT_MM_PMR_EMULATION
 #endif
diff --git a/src/multimedia/audio/qrtaudioengine.cpp b/src/multimedia/audio/qrtaudioengine.cpp
index fdbb5ef37..875a4c279 100644
--- a/src/multimedia/audio/qrtaudioengine.cpp
+++ b/src/multimedia/audio/qrtaudioengine.cpp
@@ -26,6 +26,7 @@ QT_BEGIN_NAMESPACE
 namespace QtMultimediaPrivate {
 
 using namespace QtPrivate;
+using namespace std::chrono_literals;
 
 ///////////////////////////////////////////////////////////////////////////////////////////////////
 
@@ -114,8 +115,18 @@ QRtAudioEngine::QRtAudioEngine(const QAudioDevice &device, const QAudioFormat &f
         moveToThread(appThread);
         m_sink.moveToThread(appThread);
         m_notificationEvent.moveToThread(appThread);
+        m_pendingCommandsTimer.moveToThread(appThread);
     }
 
+    m_pendingCommandsTimer.setInterval(10ms);
+    m_pendingCommandsTimer.setTimerType(Qt::CoarseTimer);
+    m_pendingCommandsTimer.callOnTimeout(&m_pendingCommandsTimer, [this] {
+        auto lock = std::lock_guard{ m_mutex };
+        sendPendingRtCommands();
+        if (m_appToRtOverflowBuffer.empty())
+            m_pendingCommandsTimer.stop();
+    });
+
     QPlatformAudioSink *platformSink = QPlatformAudioSink::get(m_sink);
 
     platformSink->start([this](QSpan<float> outputBuffer) {
@@ -149,7 +160,7 @@ void QRtAudioEngine::play(SharedVoice voice)
 
     m_voices.insert(voice);
 
-    m_appToRt.write(PlayCommand{
+    sendAppToRtCommand(PlayCommand{
             std::move(voice),
     });
 }
@@ -162,7 +173,7 @@ void QRtAudioEngine::stop(const SharedVoice &voice)
 void QRtAudioEngine::stop(VoiceId voiceId)
 {
     auto lock = std::lock_guard{ m_mutex };
-    m_appToRt.write(StopCommand{ voiceId });
+    sendAppToRtCommand(StopCommand{ voiceId });
 }
 
 void QRtAudioEngine::visitVoiceRt(VoiceId voiceId, RtVoiceVisitor fn, bool visitorIsTrivial)
@@ -170,13 +181,13 @@ void QRtAudioEngine::visitVoiceRt(VoiceId voiceId, RtVoiceVisitor fn, bool visit
     auto lock = std::lock_guard{ m_mutex };
 
     if (visitorIsTrivial) {
-        m_appToRt.write(VisitCommandTrivial{
+        sendAppToRtCommand(VisitCommandTrivial{
                 voiceId,
                 std::move(fn),
         });
 
     } else {
-        m_appToRt.write(VisitCommand{
+        sendAppToRtCommand(VisitCommand{
                 voiceId,
                 std::move(fn),
         });
@@ -192,6 +203,7 @@ VoiceId QRtAudioEngine::allocateVoiceId()
 void QRtAudioEngine::audioCallback(QSpan<float> outputBuffer) noexcept QT_MM_NONBLOCKING
 {
     runRtCommands();
+    bool sendNotification = sendPendingAppNotifications();
 
     std::fill(outputBuffer.begin(), outputBuffer.end(), 0.f);
 
@@ -211,15 +223,18 @@ void QRtAudioEngine::audioCallback(QSpan<float> outputBuffer) noexcept QT_MM_NON
         withRTSanDisabled([&] {
             for (const SharedVoice &voice : finishedVoices) {
                 m_rtVoiceRegistry.erase(voice);
-                m_rtToApp.write(StopNotification{ voice });
+                bool stopSent = sendRtToAppNotification(StopNotification{ voice });
+                if (stopSent)
+                    sendNotification = true;
             }
         });
-        m_notificationEvent.set();
     }
 
     // TODO: we should probably (soft)clip the output buffer
 
     cleanupRetiredVoices();
+    if (sendNotification)
+        m_notificationEvent.set();
 }
 
 void QRtAudioEngine::cleanupRetiredVoices() noexcept QT_MM_NONBLOCKING
@@ -243,10 +258,8 @@ void QRtAudioEngine::cleanupRetiredVoices() noexcept QT_MM_NONBLOCKING
     withRTSanDisabled([&] {
         erase_if(m_rtVoiceRegistry, [&](const SharedVoice &voice) {
             bool voiceIsActive = voice->isActive();
-            if (!voiceIsActive) {
-                m_rtToApp.write(StopNotification{ voice });
-                notifyApp = true;
-            }
+            if (!voiceIsActive)
+                notifyApp = sendRtToAppNotification(StopNotification{ voice });
 
             return false;
         });
@@ -282,10 +295,11 @@ void QRtAudioEngine::runRtCommand(StopCommand cmd) noexcept QT_MM_NONBLOCKING
     SharedVoice voice = *it;
     m_rtVoiceRegistry.erase(it);
 
-    m_rtToApp.write(StopNotification{
+    bool emitNotify = sendRtToAppNotification(StopNotification{
             std::move(voice),
     });
-    m_notificationEvent.set();
+    if (emitNotify)
+        m_notificationEvent.set();
 }
 
 void QRtAudioEngine::runRtCommand(VisitCommand cmd) noexcept QT_MM_NONBLOCKING
@@ -296,10 +310,11 @@ void QRtAudioEngine::runRtCommand(VisitCommand cmd) noexcept QT_MM_NONBLOCKING
     cmd.callback(**it);
 
     // send callback back to application for destruction
-    m_rtToApp.write(VisitReply{
+    bool emitNotify = sendRtToAppNotification(VisitReply{
             std::move(cmd.callback),
     });
-    m_notificationEvent.set();
+    if (emitNotify)
+        m_notificationEvent.set();
 }
 
 void QRtAudioEngine::runRtCommand(VisitCommandTrivial cmd) noexcept QT_MM_NONBLOCKING
@@ -345,6 +360,77 @@ void QRtAudioEngine::runNonRtNotification(VisitReply)
     // nop (just making sure to delete on the application thread);
 }
 
+void QRtAudioEngine::sendAppToRtCommand(RtCommand cmd)
+{
+    // first write all pending commands from overflow buffer
+    sendPendingRtCommands();
+
+    bool written = m_appToRt.produceOne([&] {
+        return std::move(cmd);
+    });
+
+    if (written)
+        return;
+
+    m_appToRtOverflowBuffer.emplace_back(std::move(cmd));
+
+    QMetaObject::invokeMethod(&m_pendingCommandsTimer, [this] {
+        if (!m_pendingCommandsTimer.isActive())
+            m_pendingCommandsTimer.start();
+    });
+}
+
+bool QRtAudioEngine::sendRtToAppNotification(Notification cmd)
+{
+    // first write all pending commands from overflow buffer
+    bool emitNotification = sendPendingAppNotifications();
+
+    bool written = m_rtToApp.produceOne([&] {
+        return std::move(cmd);
+    });
+
+    if (written)
+        return true;
+
+    m_rtToAppOverflowBuffer.emplace_back(std::move(cmd));
+
+    return emitNotification;
+}
+
+void QRtAudioEngine::sendPendingRtCommands()
+{
+    while (!m_appToRtOverflowBuffer.empty()) {
+        Q_UNLIKELY_BRANCH;
+        bool written = m_appToRt.produceOne([&] {
+            return std::move(m_appToRtOverflowBuffer.front());
+        });
+        if (!written)
+            return;
+
+        m_appToRtOverflowBuffer.pop_front();
+    }
+}
+
+bool QRtAudioEngine::sendPendingAppNotifications()
+{
+    bool emitNotification = false;
+    while (!m_rtToAppOverflowBuffer.empty()) {
+        Q_UNLIKELY_BRANCH;
+
+        // first write all pending commands from overflow buffer
+        bool written = m_rtToApp.produceOne([&] {
+            return std::move(m_rtToAppOverflowBuffer.front());
+        });
+        if (!written)
+            break;
+
+        m_rtToAppOverflowBuffer.pop_front();
+        emitNotification = true;
+    }
+
+    return emitNotification;
+}
+
 } // namespace QtMultimediaPrivate
 
 QT_END_NAMESPACE
diff --git a/src/multimedia/audio/qrtaudioengine_p.h b/src/multimedia/audio/qrtaudioengine_p.h
index 04787decf..b2214db79 100644
--- a/src/multimedia/audio/qrtaudioengine_p.h
+++ b/src/multimedia/audio/qrtaudioengine_p.h
@@ -16,6 +16,7 @@
 //
 
 #include <QtCore/qtclasshelpermacros.h>
+#include <QtCore/qtimer.h>
 #include <QtCore/qmutex.h>
 
 #include <QtMultimedia/qaudiosink.h>
@@ -26,8 +27,10 @@
 #include <QtMultimedia/private/q_pmr_emulation_p.h>
 
 #include <cstdint>
+#include <deque>
 #include <set>
 #include <variant>
+#include <vector>
 
 QT_BEGIN_NAMESPACE
 
@@ -272,6 +275,15 @@ private:
     static constexpr size_t commandBuffersSize = 2048;
     QtPrivate::QAudioRingBuffer<RtCommand> m_appToRt{ commandBuffersSize };
     QtPrivate::QAudioRingBuffer<Notification> m_rtToApp{ commandBuffersSize };
+    std::deque<RtCommand> m_appToRtOverflowBuffer;
+    std::deque<Notification, pmr::polymorphic_allocator<Notification>> m_rtToAppOverflowBuffer{
+        m_rtMemoryPool.get(),
+    };
+    void sendAppToRtCommand(RtCommand cmd);
+    bool sendRtToAppNotification(Notification cmd);
+    void sendPendingRtCommands();
+    bool sendPendingAppNotifications();
+    QTimer m_pendingCommandsTimer;
 
     QtPrivate::QAutoResetEvent m_notificationEvent;
     std::vector<VoiceId> m_finishedVoices;
diff --git a/tests/auto/integration/qrtaudioengine/tst_qrtaudioengine.cpp b/tests/auto/integration/qrtaudioengine/tst_qrtaudioengine.cpp
index e7e75a984..474545628 100644
--- a/tests/auto/integration/qrtaudioengine/tst_qrtaudioengine.cpp
+++ b/tests/auto/integration/qrtaudioengine/tst_qrtaudioengine.cpp
@@ -92,6 +92,7 @@ private slots:
     void play_and_stop_byPlaybackStatus();
 
     void play_validateDuration();
+    void visit_overloadCommandBuffers();
 
 private:
     // helpers
@@ -237,6 +238,32 @@ void tst_QRtAudioEngine::play_validateDuration()
     QCOMPARE_LT(voice->m_sampleCount, upperBound);
 }
 
+void tst_QRtAudioEngine::visit_overloadCommandBuffers()
+{
+    std::shared_ptr<QRtAudioEngine> engine = makeEngine();
+
+    auto voice = makeMockVoice(getFormat());
+    QSignalSpy finishedSpy(engine.get(), &QRtAudioEngine::voiceFinished);
+    engine->play(voice);
+
+    constexpr size_t commandCount = 16384; // 8 times the command buffer size
+
+    size_t iteration = 0;
+    auto refcountedDummyObject = std::make_shared<int>(0);
+    for (size_t i = 0; i < commandCount; ++i) {
+        engine->visitVoiceRt(
+                voice,
+                [&, payload = std::array<char, 128>{}, refcountedDummyObject](NullMockVoice &) {
+            Q_UNUSED(payload);
+            iteration += 1;
+        });
+    }
+
+    QTRY_COMPARE_EQ(iteration, commandCount); // all commands should have been executed
+    QTRY_COMPARE_EQ(refcountedDummyObject.use_count(),
+                    1); // all commands should have been destroyed on the app thread
+}
+
 QTEST_MAIN(tst_QRtAudioEngine)
 
 #include "tst_qrtaudioengine.moc"