GStreamer: AudioDecoder - fix race condition

The app sources lives on a gstreamer thread. `bufferAvailableChanged`
was called from both application thread and gstreamer thread, which is
not thread safe.

We simplify the application logic by deferring all mutation of the
`QGstreamerAudioDecoder` state to the application thread. This seems to
fix a spurious failure in tst_QAudioDecoderBackend::fileTest

Pick-to: 6.5 6.7
Change-Id: I992c39b071ebd48d5fc5ca28fae499c855b93304
Reviewed-by: Artem Dyomin <artem.dyomin@qt.io>

2 files changed, 44 insertions, 59 deletions

diff --git a/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder.cpp b/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder.cpp
index 3547f124a..0cfa28169 100644
--- a/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder.cpp
+++ b/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder.cpp
@@ -327,6 +327,7 @@ void QGstreamerAudioDecoder::start()
 void QGstreamerAudioDecoder::stop()
 {
     m_playbin.setState(GST_STATE_NULL);
+    m_currentSessionId += 1;
     removeAppSink();
 
     // GStreamer thread is stopped. Can safely access m_buffersAvailable
@@ -365,55 +366,38 @@ QAudioBuffer QGstreamerAudioDecoder::read()
 {
     QAudioBuffer audioBuffer;
 
-    int buffersAvailable;
-    {
-        QMutexLocker locker(&m_buffersMutex);
-        buffersAvailable = m_buffersAvailable;
+    if (m_buffersAvailable == 0)
+        return audioBuffer;
 
-        // need to decrement before pulling a buffer
-        // to make sure assert in QGstreamerAudioDecoderControl::new_buffer works
-        m_buffersAvailable--;
-    }
+    m_buffersAvailable -= 1;
 
+    if (m_buffersAvailable == 0)
+        bufferAvailableChanged(false);
 
-    if (buffersAvailable) {
-        if (buffersAvailable == 1)
-            bufferAvailableChanged(false);
-
-        const char* bufferData = nullptr;
-        int bufferSize = 0;
-
-        QGstSampleHandle sample = m_appSink.pullSample();
-        GstBuffer *buffer = gst_sample_get_buffer(sample.get());
-        GstMapInfo mapInfo;
-        gst_buffer_map(buffer, &mapInfo, GST_MAP_READ);
-        bufferData = (const char*)mapInfo.data;
-        bufferSize = mapInfo.size;
-        QAudioFormat format = QGstUtils::audioFormatForSample(sample.get());
-
-        if (format.isValid()) {
-            // XXX At the moment we have to copy data from GstBuffer into QAudioBuffer.
-            // We could improve performance by implementing QAbstractAudioBuffer for GstBuffer.
-            qint64 position = getPositionFromBuffer(buffer);
-            audioBuffer = QAudioBuffer(QByteArray((const char*)bufferData, bufferSize), format, position);
-            position /= 1000; // convert to milliseconds
-            if (position != m_position) {
-                m_position = position;
-                positionChanged(m_position);
-            }
+    QGstSampleHandle sample = m_appSink.pullSample();
+    GstBuffer *buffer = gst_sample_get_buffer(sample.get());
+    GstMapInfo mapInfo;
+    gst_buffer_map(buffer, &mapInfo, GST_MAP_READ);
+    const char *bufferData = (const char *)mapInfo.data;
+    int bufferSize = mapInfo.size;
+    QAudioFormat format = QGstUtils::audioFormatForSample(sample.get());
+
+    if (format.isValid()) {
+        // XXX At the moment we have to copy data from GstBuffer into QAudioBuffer.
+        // We could improve performance by implementing QAbstractAudioBuffer for GstBuffer.
+        qint64 position = getPositionFromBuffer(buffer);
+        audioBuffer = QAudioBuffer(QByteArray(bufferData, bufferSize), format, position);
+        position /= 1000; // convert to milliseconds
+        if (position != m_position) {
+            m_position = position;
+            positionChanged(m_position);
         }
-        gst_buffer_unmap(buffer, &mapInfo);
     }
+    gst_buffer_unmap(buffer, &mapInfo);
 
     return audioBuffer;
 }
 
-bool QGstreamerAudioDecoder::bufferAvailable() const
-{
-    QMutexLocker locker(&m_buffersMutex);
-    return m_buffersAvailable > 0;
-}
-
 qint64 QGstreamerAudioDecoder::position() const
 {
     return m_position;
@@ -430,30 +414,30 @@ void QGstreamerAudioDecoder::processInvalidMedia(QAudioDecoder::Error errorCode,
     error(int(errorCode), errorString);
 }
 
-GstFlowReturn QGstreamerAudioDecoder::new_sample(GstAppSink *, gpointer user_data)
+GstFlowReturn QGstreamerAudioDecoder::newSample(GstAppSink *)
 {
-    qCDebug(qLcGstreamerAudioDecoder) << "QGstreamerAudioDecoder::new_sample";
-
     // "Note that the preroll buffer will also be returned as the first buffer when calling
     // gst_app_sink_pull_buffer()."
-    QGstreamerAudioDecoder *decoder = reinterpret_cast<QGstreamerAudioDecoder*>(user_data);
-
-    int buffersAvailable;
-    {
-        QMutexLocker locker(&decoder->m_buffersMutex);
-        buffersAvailable = decoder->m_buffersAvailable;
-        decoder->m_buffersAvailable++;
-        Q_ASSERT(decoder->m_buffersAvailable <= MAX_BUFFERS_IN_QUEUE);
-    }
 
-    qCDebug(qLcGstreamerAudioDecoder) << "QGstreamerAudioDecoder::new_sample" << buffersAvailable;
+    QMetaObject::invokeMethod(this, [this, sessionId = m_currentSessionId] {
+        if (sessionId != m_currentSessionId)
+            return; // stop()ed before message is executed
+
+        m_buffersAvailable += 1;
+        bufferAvailableChanged(true);
+        bufferReady();
+    });
 
-    if (!buffersAvailable)
-        decoder->bufferAvailableChanged(true);
-    decoder->bufferReady();
     return GST_FLOW_OK;
 }
 
+GstFlowReturn QGstreamerAudioDecoder::new_sample(GstAppSink *sink, gpointer user_data)
+{
+    QGstreamerAudioDecoder *decoder = reinterpret_cast<QGstreamerAudioDecoder *>(user_data);
+    qCDebug(qLcGstreamerAudioDecoder) << "QGstreamerAudioDecoder::new_sample";
+    return decoder->newSample(sink);
+}
+
 void QGstreamerAudioDecoder::setAudioFlags(bool wantNativeAudio)
 {
     int flags = m_playbin.getInt("flags");
diff --git a/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder_p.h b/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder_p.h
index 2f6958947..eba1025fa 100644
--- a/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder_p.h
+++ b/src/plugins/multimedia/gstreamer/audio/qgstreameraudiodecoder_p.h
@@ -57,7 +57,6 @@ public:
     void setAudioFormat(const QAudioFormat &format) override;
 
     QAudioBuffer read() override;
-    bool bufferAvailable() const override;
 
     qint64 position() const override;
     qint64 duration() const override;
@@ -73,6 +72,8 @@ private:
 
 #if QT_CONFIG(gstreamer_app)
     static GstFlowReturn new_sample(GstAppSink *sink, gpointer user_data);
+    GstFlowReturn newSample(GstAppSink *sink);
+
     static void configureAppSrcElement(GObject *, GObject *, GParamSpec *,
                                        QGstreamerAudioDecoder *_this);
 #endif
@@ -96,14 +97,14 @@ private:
     QIODevice *mDevice = nullptr;
     QAudioFormat mFormat;
 
-    mutable QMutex m_buffersMutex;
     int m_buffersAvailable = 0;
-
     qint64 m_position = -1;
     qint64 m_duration = -1;
 
     int m_durationQueries = 0;
 
+    qint32 m_currentSessionId{};
+
     QGObjectHandlerScopedConnection m_deepNotifySourceConnection;
 };