Clean up pulse audio and fix some threading issues

- improve threading handling, so auto tests flakiness has been fixed.
  (additional fixes required, namely atomic states comare_exchange)
- improve pulse audio operations managing with RAII
- remove a hack with a callback based on QObject::destroyed
  connection, the hack appeared to cause memory leak time to time.
- the main logic is not changed

The patch has been decoupled from
codereview.qt-project.org/c/qt/qtmultimedia/+/474485

Change-Id: I68002c243dea874c4300b14b1fbd6b618392ab4f
Reviewed-by: Lars Knoll <lars@knoll.priv.no>
(cherry picked from commit 93e7d81cdea07ad1e364e1a6d107022298be35ab)
Reviewed-by: Qt Cherry-pick Bot <cherrypick_bot@qt-project.org>

6 files changed, 147 insertions, 191 deletions

diff --git a/src/multimedia/pulseaudio/qaudioengine_pulse.cpp b/src/multimedia/pulseaudio/qaudioengine_pulse.cpp
index 3e822c125..01521a3d5 100644
--- a/src/multimedia/pulseaudio/qaudioengine_pulse.cpp
+++ b/src/multimedia/pulseaudio/qaudioengine_pulse.cpp
@@ -10,6 +10,7 @@
 #include "qpulsehelpers_p.h"
 #include <sys/types.h>
 #include <unistd.h>
+#include <mutex> // for lock_guard
 
 QT_BEGIN_NAMESPACE
 
@@ -153,26 +154,22 @@ static void event_cb(pa_context* context, pa_subscription_event_type_t t, uint32
     case PA_SUBSCRIPTION_EVENT_CHANGE:
         switch (facility) {
         case PA_SUBSCRIPTION_EVENT_SERVER: {
-            pa_operation *op = pa_context_get_server_info(context, serverInfoCallback, userdata);
-            if (op)
-                pa_operation_unref(op);
-            else
+            PAOperationUPtr op(pa_context_get_server_info(context, serverInfoCallback, userdata));
+            if (!op)
                 qWarning("PulseAudioService: failed to get server info");
             break;
         }
         case PA_SUBSCRIPTION_EVENT_SINK: {
-            pa_operation *op = pa_context_get_sink_info_by_index(context, index, sinkInfoCallback, userdata);
-            if (op)
-                pa_operation_unref(op);
-            else
+            PAOperationUPtr op(
+                    pa_context_get_sink_info_by_index(context, index, sinkInfoCallback, userdata));
+            if (!op)
                 qWarning("PulseAudioService: failed to get sink info");
             break;
         }
         case PA_SUBSCRIPTION_EVENT_SOURCE: {
-            pa_operation *op = pa_context_get_source_info_by_index(context, index, sourceInfoCallback, userdata);
-            if (op)
-                pa_operation_unref(op);
-            else
+            PAOperationUPtr op(pa_context_get_source_info_by_index(context, index,
+                                                                   sourceInfoCallback, userdata));
+            if (!op)
                 qWarning("PulseAudioService: failed to get source info");
             break;
         }
@@ -324,14 +321,12 @@ void QPulseAudioEngine::prepare()
         pa_context_set_state_callback(m_context, contextStateCallback, this);
 
         pa_context_set_subscribe_callback(m_context, event_cb, this);
-        pa_operation *op = pa_context_subscribe(m_context,
-                                                pa_subscription_mask_t(PA_SUBSCRIPTION_MASK_SINK |
-                                                                       PA_SUBSCRIPTION_MASK_SOURCE |
-                                                                       PA_SUBSCRIPTION_MASK_SERVER),
-                                                nullptr, nullptr);
-        if (op)
-            pa_operation_unref(op);
-        else
+        PAOperationUPtr op(pa_context_subscribe(
+                m_context,
+                pa_subscription_mask_t(PA_SUBSCRIPTION_MASK_SINK | PA_SUBSCRIPTION_MASK_SOURCE
+                                       | PA_SUBSCRIPTION_MASK_SERVER),
+                nullptr, nullptr));
+        if (!op)
             qWarning("PulseAudioService: failed to subscribe to context notifications");
     } else {
         pa_context_unref(m_context);
@@ -372,39 +367,34 @@ void QPulseAudioEngine::release()
 
 void QPulseAudioEngine::updateDevices()
 {
-    lock();
+    std::lock_guard lock(*this);
 
     // Get default input and output devices
-    pa_operation *operation = pa_context_get_server_info(m_context, serverInfoCallback, this);
+    PAOperationUPtr operation(pa_context_get_server_info(m_context, serverInfoCallback, this));
     if (operation) {
-        while (pa_operation_get_state(operation) == PA_OPERATION_RUNNING)
+        while (pa_operation_get_state(operation.get()) == PA_OPERATION_RUNNING)
             pa_threaded_mainloop_wait(m_mainLoop);
-        pa_operation_unref(operation);
     } else {
         qWarning("PulseAudioService: failed to get server info");
     }
 
     // Get output devices
-    operation = pa_context_get_sink_info_list(m_context, sinkInfoCallback, this);
+    operation.reset(pa_context_get_sink_info_list(m_context, sinkInfoCallback, this));
     if (operation) {
-        while (pa_operation_get_state(operation) == PA_OPERATION_RUNNING)
+        while (pa_operation_get_state(operation.get()) == PA_OPERATION_RUNNING)
             pa_threaded_mainloop_wait(m_mainLoop);
-        pa_operation_unref(operation);
     } else {
         qWarning("PulseAudioService: failed to get sink info");
     }
 
     // Get input devices
-    operation = pa_context_get_source_info_list(m_context, sourceInfoCallback, this);
+    operation.reset(pa_context_get_source_info_list(m_context, sourceInfoCallback, this));
     if (operation) {
-        while (pa_operation_get_state(operation) == PA_OPERATION_RUNNING)
+        while (pa_operation_get_state(operation.get()) == PA_OPERATION_RUNNING)
             pa_threaded_mainloop_wait(m_mainLoop);
-        pa_operation_unref(operation);
     } else {
         qWarning("PulseAudioService: failed to get source info");
     }
-
-    unlock();
 }
 
 void QPulseAudioEngine::onContextFailed()
diff --git a/src/multimedia/pulseaudio/qpulseaudiosink.cpp b/src/multimedia/pulseaudio/qpulseaudiosink.cpp
index fea1f5afb..132b9fbc1 100644
--- a/src/multimedia/pulseaudio/qpulseaudiosink.cpp
+++ b/src/multimedia/pulseaudio/qpulseaudiosink.cpp
@@ -12,7 +12,7 @@
 #include "qpulsehelpers_p.h"
 #include <sys/types.h>
 #include <unistd.h>
-
+#include <mutex> // for st::lock_guard
 
 QT_BEGIN_NAMESPACE
 
@@ -98,10 +98,8 @@ static void outputStreamDrainComplete(pa_stream *stream, int success, void *user
 
     qCDebug(qLcPulseAudioOut) << "Stream drained:" << bool(success) << userdata;
 
-    std::unique_ptr<QObject> object(static_cast<QObject *>(userdata));
-
-    if (object && !success)
-        object->disconnect();
+    if (userdata && success)
+        static_cast<QPulseAudioSink *>(userdata)->streamDrainedCallback();
 }
 
 static void streamAdjustPrebufferCallback(pa_stream *stream, int success, void *userdata)
@@ -126,27 +124,24 @@ QPulseAudioSink::~QPulseAudioSink()
     QCoreApplication::processEvents();
 }
 
-void QPulseAudioSink::setError(QAudio::Error error)
-{
-    if (m_errorState == error)
-        return;
-
-    m_errorState = error;
-    emit errorChanged(error);
-}
-
 QAudio::Error QPulseAudioSink::error() const
 {
     return m_errorState;
 }
 
-void QPulseAudioSink::setState(QAudio::State state)
+void QPulseAudioSink::setStateAndError(QAudio::State state, QAudio::Error error,
+                                       bool forceEmitState)
 {
-    if (m_deviceState == state)
-        return;
+    // TODO: reimplement with atomic state changings (compare_exchange)
+
+    const auto isStateChanged = m_deviceState.exchange(state) != state;
+    const auto isErrorChanged = m_errorState.exchange(error) != error;
 
-    m_deviceState = state;
-    emit stateChanged(state);
+    if (isStateChanged || forceEmitState)
+        emit stateChanged(state);
+
+    if (isErrorChanged)
+        emit errorChanged(error);
 }
 
 QAudio::State QPulseAudioSink::state() const
@@ -159,34 +154,23 @@ void QPulseAudioSink::streamUnderflowCallback()
     if (m_audioSource && m_audioSource->atEnd()) {
         qCDebug(qLcPulseAudioOut) << "Draining stream at end of buffer";
 
-        auto object = std::make_unique<QObject>();
-        connect(object.get(), &QObject::destroyed, this, &QPulseAudioSink::streamDrainedCallback);
-        pa_operation *o = pa_stream_drain(m_stream, outputStreamDrainComplete, object.get());
-        if (o) {
-            object.release();
-            pa_operation_unref(o);
-        }
+        exchangeDrainOperation(pa_stream_drain(m_stream, outputStreamDrainComplete, this));
     } else if (m_deviceState != QAudio::IdleState && !m_resuming) {
-        setError(QAudio::UnderrunError);
-        setState(QAudio::IdleState);
+        setStateAndError(QAudio::IdleState, QAudio::UnderrunError);
     }
 }
 
 void QPulseAudioSink::streamDrainedCallback()
 {
-    setError(QAudio::NoError);
-    setState(QAudio::IdleState);
+    if (!exchangeDrainOperation(nullptr))
+        return;
+
+    setStateAndError(QAudio::IdleState, QAudio::NoError);
 }
 
 void QPulseAudioSink::start(QIODevice *device)
 {
-    setState(QAudio::StoppedState);
-    setError(QAudio::NoError);
-
-    // Handle change of mode
-    if (m_audioSource && !m_pullMode)
-        delete m_audioSource;
-    m_audioSource = nullptr;
+    setStateAndError(QAudio::StoppedState, QAudio::NoError);
 
     close();
 
@@ -203,7 +187,7 @@ void QPulseAudioSink::start(QIODevice *device)
     lastProcessedUSecs = 0;
 
     connect(m_audioSource, &QIODevice::readyRead, this, &QPulseAudioSink::startReading);
-    setState(QAudio::ActiveState);
+    setStateAndError(QAudio::ActiveState, QAudio::NoError);
 }
 
 void QPulseAudioSink::startReading()
@@ -214,13 +198,7 @@ void QPulseAudioSink::startReading()
 
 QIODevice *QPulseAudioSink::start()
 {
-    setState(QAudio::StoppedState);
-    setError(QAudio::NoError);
-
-    // Handle change of mode
-    if (m_audioSource && !m_pullMode)
-        delete m_audioSource;
-    m_audioSource = nullptr;
+    setStateAndError(QAudio::StoppedState, QAudio::NoError);
 
     close();
 
@@ -236,7 +214,7 @@ QIODevice *QPulseAudioSink::start()
     gettimeofday(&lastTimingInfo, nullptr);
     lastProcessedUSecs = 0;
 
-    setState(QAudio::IdleState);
+    setStateAndError(QAudio::IdleState, QAudio::NoError);
 
     return m_audioSource;
 }
@@ -249,9 +227,7 @@ bool QPulseAudioSink::open()
     QPulseAudioEngine *pulseEngine = QPulseAudioEngine::instance();
 
     if (!pulseEngine->context() || pa_context_get_state(pulseEngine->context()) != PA_CONTEXT_READY) {
-        setError(QAudio::FatalError);
-        setState(QAudio::StoppedState);
-        emit stateChanged(m_deviceState);
+        setStateAndError(QAudio::StoppedState, QAudio::FatalError, true);
         return false;
     }
 
@@ -260,9 +236,7 @@ bool QPulseAudioSink::open()
     Q_ASSERT(spec.channels == channel_map.channels);
 
     if (!pa_sample_spec_valid(&spec)) {
-        setError(QAudio::OpenError);
-        setState(QAudio::StoppedState);
-        emit stateChanged(m_deviceState);
+        setStateAndError(QAudio::StoppedState, QAudio::OpenError, true);
         return false;
     }
 
@@ -305,17 +279,16 @@ bool QPulseAudioSink::open()
 #endif
 
     m_stream = pa_stream_new_with_proplist(pulseEngine->context(), m_streamName.constData(), &m_spec, &channel_map, propList);
+    pa_proplist_free(propList);
+
     if (!m_stream) {
         qCWarning(qLcPulseAudioOut) << "QAudioSink: pa_stream_new_with_proplist() failed!";
         pulseEngine->unlock();
-        setError(QAudio::OpenError);
-        setState(QAudio::StoppedState);
-        emit stateChanged(m_deviceState);
+
+        setStateAndError(QAudio::StoppedState, QAudio::OpenError, true);
         return false;
     }
 
-    pa_proplist_free(propList);
-
     pa_stream_set_state_callback(m_stream, outputStreamStateCallback, this);
     pa_stream_set_write_callback(m_stream, outputStreamWriteCallback, this);
 
@@ -336,9 +309,7 @@ bool QPulseAudioSink::open()
         pa_stream_unref(m_stream);
         m_stream = nullptr;
         pulseEngine->unlock();
-        setError(QAudio::OpenError);
-        setState(QAudio::StoppedState);
-        emit stateChanged(m_deviceState);
+        setStateAndError(QAudio::StoppedState, QAudio::OpenError, true);
         return false;
     }
 
@@ -349,17 +320,15 @@ bool QPulseAudioSink::open()
     m_periodTime = SinkPeriodTimeMs;
     m_periodSize = pa_usec_to_bytes(m_periodTime * 1000, &m_spec);
     m_bufferSize = buffer->tlength;
-    m_maxBufferSize = buffer->maxlength;
-    m_audioBuffer = new char[m_maxBufferSize];
+    m_audioBuffer.resize(buffer->maxlength);
 
     const qint64 streamSize = m_audioSource ? m_audioSource->size() : 0;
     if (m_pullMode && streamSize > 0 && static_cast<qint64>(buffer->prebuf) > streamSize) {
         pa_buffer_attr newBufferAttr;
         newBufferAttr = *buffer;
         newBufferAttr.prebuf = streamSize;
-        pa_operation *o = pa_stream_set_buffer_attr(m_stream, &newBufferAttr, streamAdjustPrebufferCallback, nullptr);
-        if (o)
-            pa_operation_unref(o);
+        PAOperationUPtr(pa_stream_set_buffer_attr(m_stream, &newBufferAttr,
+                                                  streamAdjustPrebufferCallback, nullptr));
     }
 
     if (Q_UNLIKELY(qLcPulseAudioOut().isEnabled(QtDebugMsg))) {
@@ -394,7 +363,7 @@ void QPulseAudioSink::close()
     QPulseAudioEngine *pulseEngine = QPulseAudioEngine::instance();
 
     if (m_stream) {
-        pulseEngine->lock();
+        std::lock_guard lock(*pulseEngine);
 
         pa_stream_set_state_callback(m_stream, nullptr, nullptr);
         pa_stream_set_write_callback(m_stream, nullptr, nullptr);
@@ -402,15 +371,15 @@ void QPulseAudioSink::close()
         pa_stream_set_overflow_callback(m_stream, nullptr, nullptr);
         pa_stream_set_latency_update_callback(m_stream, nullptr, nullptr);
 
-        pa_operation *o = pa_stream_drain(m_stream, outputStreamDrainComplete, nullptr);
-        if (o)
-            pa_operation_unref(o);
+        if (auto prevOp = exchangeDrainOperation(nullptr))
+            // cancel the draining callback that is not relevant already
+            pa_operation_cancel(prevOp.get());
+
+        PAOperationUPtr(pa_stream_drain(m_stream, outputStreamDrainComplete, nullptr));
 
         pa_stream_disconnect(m_stream);
         pa_stream_unref(m_stream);
         m_stream = nullptr;
-
-        pulseEngine->unlock();
     }
 
     disconnect(pulseEngine, &QPulseAudioEngine::contextFailed, this, &QPulseAudioSink::onPulseContextFailed);
@@ -424,10 +393,8 @@ void QPulseAudioSink::close()
         }
     }
     m_opened = false;
-    if (m_audioBuffer) {
-        delete[] m_audioBuffer;
-        m_audioBuffer = nullptr;
-    }
+    m_resuming = false;
+    m_audioBuffer.clear();
 }
 
 void QPulseAudioSink::timerEvent(QTimerEvent *event)
@@ -440,25 +407,25 @@ void QPulseAudioSink::timerEvent(QTimerEvent *event)
 
 void QPulseAudioSink::userFeed()
 {
-    if (m_deviceState == QAudio::StoppedState || m_deviceState == QAudio::SuspendedState)
+    const auto state = this->state();
+    if (state == QAudio::StoppedState || state == QAudio::SuspendedState)
         return;
 
     m_resuming = false;
 
     if (m_pullMode) {
-        setError(QAudio::NoError);
-        setState(QAudio::ActiveState);
+        setStateAndError(QAudio::ActiveState, QAudio::NoError);
         int writableSize = bytesFree();
         int chunks = writableSize / m_periodSize;
         if (chunks == 0)
             return;
 
-        int input = m_periodSize; // always request 1 chunk of data from user
-        if (input > m_maxBufferSize)
-            input = m_maxBufferSize;
+        const int input = std::min(
+                m_periodSize,
+                static_cast<int>(m_audioBuffer.size())); // always request 1 chunk of data from user
 
-        Q_ASSERT(m_audioBuffer);
-        int audioBytesPulled = m_audioSource->read(m_audioBuffer, input);
+        Q_ASSERT(!m_audioBuffer.empty());
+        int audioBytesPulled = m_audioSource->read(m_audioBuffer.data(), input);
         Q_ASSERT(audioBytesPulled <= input);
         if (audioBytesPulled > 0) {
             if (audioBytesPulled > input) {
@@ -466,7 +433,7 @@ void QPulseAudioSink::userFeed()
                                             << audioBytesPulled << "should be less than" << input;
                 audioBytesPulled = input;
             }
-            qint64 bytesWritten = write(m_audioBuffer, audioBytesPulled);
+            qint64 bytesWritten = write(m_audioBuffer.data(), audioBytesPulled);
             Q_ASSERT(bytesWritten == audioBytesPulled); //unfinished write should not happen since the data provided is less than writableSize
             Q_UNUSED(bytesWritten);
 
@@ -476,17 +443,14 @@ void QPulseAudioSink::userFeed()
             }
         } else if (audioBytesPulled == 0) {
             m_tickTimer.stop();
-            qCDebug(qLcPulseAudioOut) << "No more data available, source is done:" << m_audioSource->atEnd();
-            setError(m_audioSource->atEnd() ? QAudio::NoError : QAudio::UnderrunError);
-            setState(QAudio::IdleState);
+            const auto atEnd = m_audioSource->atEnd();
+            qCDebug(qLcPulseAudioOut) << "No more data available, source is done:" << atEnd;
+            setStateAndError(QAudio::IdleState, atEnd ? QAudio::NoError : QAudio::UnderrunError);
         }
     } else {
-        if (state() == QAudio::IdleState)
-            setError(QAudio::UnderrunError);
+        if (state == QAudio::IdleState)
+            setStateAndError(state, QAudio::UnderrunError);
     }
-
-    if (m_deviceState != QAudio::ActiveState)
-        return;
 }
 
 qint64 QPulseAudioSink::write(const char *data, qint64 len)
@@ -499,9 +463,10 @@ qint64 QPulseAudioSink::write(const char *data, qint64 len)
     void *dest = nullptr;
 
     if (pa_stream_begin_write(m_stream, &dest, &nbytes) < 0) {
+        pulseEngine->unlock();
         qCWarning(qLcPulseAudioOut) << "pa_stream_begin_write error:"
                                     << pa_strerror(pa_context_errno(pulseEngine->context()));
-        setError(QAudio::IOError);
+        setStateAndError(state(), QAudio::IOError);
         return 0;
     }
 
@@ -518,17 +483,17 @@ qint64 QPulseAudioSink::write(const char *data, qint64 len)
     data = reinterpret_cast<char *>(dest);
 
     if ((pa_stream_write(m_stream, data, len, nullptr, 0, PA_SEEK_RELATIVE)) < 0) {
+        pulseEngine->unlock();
         qCWarning(qLcPulseAudioOut) << "pa_stream_write error:"
                                     << pa_strerror(pa_context_errno(pulseEngine->context()));
-        setError(QAudio::IOError);
+        setStateAndError(state(), QAudio::IOError);
         return 0;
     }
 
     pulseEngine->unlock();
     m_totalTimeValue += len;
 
-    setError(QAudio::NoError);
-    setState(QAudio::ActiveState);
+    setStateAndError(QAudio::ActiveState, QAudio::NoError);
 
     return len;
 }
@@ -540,20 +505,17 @@ void QPulseAudioSink::stop()
 
     close();
 
-    setError(QAudio::NoError);
-    setState(QAudio::StoppedState);
+    setStateAndError(QAudio::StoppedState, QAudio::NoError);
 }
 
 qsizetype QPulseAudioSink::bytesFree() const
 {
-    if (m_deviceState != QAudio::ActiveState && m_deviceState != QAudio::IdleState)
+    const auto state = this->state();
+    if (state != QAudio::ActiveState && state != QAudio::IdleState)
         return 0;
 
-    QPulseAudioEngine *pulseEngine = QPulseAudioEngine::instance();
-    pulseEngine->lock();
-    int writableSize = pa_stream_writable_size(m_stream);
-    pulseEngine->unlock();
-    return writableSize;
+    std::lock_guard lock(*QPulseAudioEngine::instance());
+    return pa_stream_writable_size(m_stream);
 }
 
 void QPulseAudioSink::setBufferSize(qsizetype value)
@@ -574,9 +536,10 @@ static qint64 operator-(timeval t1, timeval t2)
 
 qint64 QPulseAudioSink::processedUSecs() const
 {
-    if (!m_stream || m_deviceState == QAudio::StoppedState)
+    const auto state = this->state();
+    if (!m_stream || state == QAudio::StoppedState)
         return 0;
-    if (m_deviceState == QAudio::SuspendedState)
+    if (state == QAudio::SuspendedState)
         return lastProcessedUSecs;
 
     auto info = pa_stream_get_timing_info(m_stream);
@@ -637,24 +600,22 @@ void QPulseAudioSink::resume()
     if (m_deviceState == QAudio::SuspendedState) {
         m_resuming = true;
 
-        QPulseAudioEngine *pulseEngine = QPulseAudioEngine::instance();
-
-        pulseEngine->lock();
+        {
+            QPulseAudioEngine *pulseEngine = QPulseAudioEngine::instance();
 
-        pa_operation *operation = pa_stream_cork(m_stream, 0, outputStreamSuccessCallback, nullptr);
-        pulseEngine->wait(operation);
-        pa_operation_unref(operation);
+            std::lock_guard lock(*pulseEngine);
 
-        operation = pa_stream_trigger(m_stream, outputStreamSuccessCallback, nullptr);
-        pulseEngine->wait(operation);
-        pa_operation_unref(operation);
+            PAOperationUPtr operation(
+                    pa_stream_cork(m_stream, 0, outputStreamSuccessCallback, nullptr));
+            pulseEngine->wait(operation.get());
 
-        pulseEngine->unlock();
+            operation.reset(pa_stream_trigger(m_stream, outputStreamSuccessCallback, nullptr));
+            pulseEngine->wait(operation.get());
+        }
 
         m_tickTimer.start(m_periodTime, this);
 
-        setState(m_suspendedInState);
-        setError(QAudio::NoError);
+        setStateAndError(m_suspendedInState, QAudio::NoError);
     }
 }
 
@@ -670,23 +631,20 @@ QAudioFormat QPulseAudioSink::format() const
 
 void QPulseAudioSink::suspend()
 {
-    if (m_deviceState == QAudio::ActiveState || m_deviceState == QAudio::IdleState) {
-        m_suspendedInState = m_deviceState;
-        setError(QAudio::NoError);
-        setState(QAudio::SuspendedState);
+    const auto state = this->state();
+    if (state == QAudio::ActiveState || state == QAudio::IdleState) {
+        m_suspendedInState = state;
+        setStateAndError(QAudio::SuspendedState, QAudio::NoError);
 
         m_tickTimer.stop();
 
         QPulseAudioEngine *pulseEngine = QPulseAudioEngine::instance();
-        pa_operation *operation;
 
-        pulseEngine->lock();
+        std::lock_guard lock(*pulseEngine);
 
-        operation = pa_stream_cork(m_stream, 1, outputStreamSuccessCallback, nullptr);
-        pulseEngine->wait(operation);
-        pa_operation_unref(operation);
-
-        pulseEngine->unlock();
+        PAOperationUPtr operation(
+                pa_stream_cork(m_stream, 1, outputStreamSuccessCallback, nullptr));
+        pulseEngine->wait(operation.get());
     }
 }
 
@@ -712,9 +670,9 @@ qint64 PulseOutputPrivate::writeData(const char *data, qint64 len)
 {
     qint64 written = 0;
 
-    if ((m_audioDevice->m_deviceState == QAudio::ActiveState
-         || m_audioDevice->m_deviceState == QAudio::IdleState)) {
-         while(written < len) {
+    const auto state = m_audioDevice->state();
+    if (state == QAudio::ActiveState || state == QAudio::IdleState) {
+        while (written < len) {
             int chunk = m_audioDevice->write(data+written, (len-written));
             if (chunk <= 0)
                 return written;
@@ -742,8 +700,12 @@ void QPulseAudioSink::onPulseContextFailed()
 {
     close();
 
-    setError(QAudio::FatalError);
-    setState(QAudio::StoppedState);
+    setStateAndError(QAudio::StoppedState, QAudio::FatalError);
+}
+
+PAOperationUPtr QPulseAudioSink::exchangeDrainOperation(pa_operation *newOperation)
+{
+    return PAOperationUPtr(m_drainOperation.exchange(newOperation));
 }
 
 QT_END_NAMESPACE
diff --git a/src/multimedia/pulseaudio/qpulseaudiosink_p.h b/src/multimedia/pulseaudio/qpulseaudiosink_p.h
index 4b3b1cf91..45ca4fd94 100644
--- a/src/multimedia/pulseaudio/qpulseaudiosink_p.h
+++ b/src/multimedia/pulseaudio/qpulseaudiosink_p.h
@@ -24,9 +24,12 @@
 
 #include "qaudio.h"
 #include "qaudiodevice.h"
+#include "pulseaudio/qpulsehelpers_p.h"
+
 #include <private/qaudiosystem_p.h>
 
 #include <pulse/pulseaudio.h>
+#include <atomic>
 
 QT_BEGIN_NAMESPACE
 
@@ -64,8 +67,7 @@ protected:
     void timerEvent(QTimerEvent *event) override;
 
 private:
-    void setState(QAudio::State state);
-    void setError(QAudio::Error error);
+    void setStateAndError(QAudio::State state, QAudio::Error error, bool forceEmitState = false);
     void startReading();
 
     bool open();
@@ -76,6 +78,8 @@ private Q_SLOTS:
     void userFeed();
     void onPulseContextFailed();
 
+    PAOperationUPtr exchangeDrainOperation(pa_operation *newOperation);
+
 private:
     pa_sample_spec m_spec = {};
     // calculate timing manually, as pulseaudio doesn't give us good enough data
@@ -90,7 +94,7 @@ private:
 
     QIODevice *m_audioSource = nullptr;
     pa_stream *m_stream = nullptr;
-    char *m_audioBuffer = nullptr;
+    std::vector<char> m_audioBuffer;
 
     qint64 m_totalTimeValue = 0;
     qint64 m_elapsedTimeOffset = 0;
@@ -98,12 +102,12 @@ private:
     mutable qint64 lastProcessedUSecs = 0;
     qreal m_volume = 1.0;
 
-    QAudio::Error m_errorState = QAudio::NoError;
-    QAudio::State m_deviceState = QAudio::StoppedState;
+    std::atomic<QAudio::Error> m_errorState = QAudio::NoError;
+    std::atomic<QAudio::State> m_deviceState = QAudio::StoppedState;
     QAudio::State m_suspendedInState = QAudio::SuspendedState;
+    std::atomic<pa_operation *> m_drainOperation = nullptr;
     int m_periodSize = 0;
     int m_bufferSize = 0;
-    int m_maxBufferSize = 0;
     int m_periodTime = 0;
     bool m_pullMode = true;
     bool m_opened = false;
diff --git a/src/multimedia/pulseaudio/qpulseaudiosource.cpp b/src/multimedia/pulseaudio/qpulseaudiosource.cpp
index 9a4cf85e3..b9b7c8fbb 100644
--- a/src/multimedia/pulseaudio/qpulseaudiosource.cpp
+++ b/src/multimedia/pulseaudio/qpulseaudiosource.cpp
@@ -12,6 +12,7 @@
 #include "qpulsehelpers_p.h"
 #include <sys/types.h>
 #include <unistd.h>
+#include <mutex> // for lock_guard
 
 QT_BEGIN_NAMESPACE
 
@@ -326,7 +327,7 @@ void QPulseAudioSource::close()
     QPulseAudioEngine *pulseEngine = QPulseAudioEngine::instance();
 
     if (m_stream) {
-        pulseEngine->lock();
+        std::lock_guard lock(*pulseEngine);
 
         pa_stream_set_state_callback(m_stream, nullptr, nullptr);
         pa_stream_set_read_callback(m_stream, nullptr, nullptr);
@@ -336,8 +337,6 @@ void QPulseAudioSource::close()
         pa_stream_disconnect(m_stream);
         pa_stream_unref(m_stream);
         m_stream = nullptr;
-
-        pulseEngine->unlock();
     }
 
     disconnect(pulseEngine, &QPulseAudioEngine::contextFailed, this, &QPulseAudioSource::onPulseContextFailed);
@@ -477,15 +476,13 @@ void QPulseAudioSource::resume()
 {
     if (m_deviceState == QAudio::SuspendedState || m_deviceState == QAudio::IdleState) {
         QPulseAudioEngine *pulseEngine = QPulseAudioEngine::instance();
-        pa_operation *operation;
-
-        pulseEngine->lock();
-
-        operation = pa_stream_cork(m_stream, 0, inputStreamSuccessCallback, nullptr);
-        pulseEngine->wait(operation);
-        pa_operation_unref(operation);
 
-        pulseEngine->unlock();
+        {
+            std::lock_guard lock(*pulseEngine);
+            PAOperationUPtr operation(
+                    pa_stream_cork(m_stream, 0, inputStreamSuccessCallback, nullptr));
+            pulseEngine->wait(operation.get());
+        }
 
         m_timer->start(m_periodTime);
 
@@ -539,15 +536,11 @@ void QPulseAudioSource::suspend()
         m_timer->stop();
 
         QPulseAudioEngine *pulseEngine = QPulseAudioEngine::instance();
-        pa_operation *operation;
 
-        pulseEngine->lock();
+        std::lock_guard lock(*pulseEngine);
 
-        operation = pa_stream_cork(m_stream, 1, inputStreamSuccessCallback, nullptr);
-        pulseEngine->wait(operation);
-        pa_operation_unref(operation);
-
-        pulseEngine->unlock();
+        PAOperationUPtr operation(pa_stream_cork(m_stream, 1, inputStreamSuccessCallback, nullptr));
+        pulseEngine->wait(operation.get());
     }
 }
 
diff --git a/src/multimedia/pulseaudio/qpulseaudiosource_p.h b/src/multimedia/pulseaudio/qpulseaudiosource_p.h
index e367cb006..81f3936ae 100644
--- a/src/multimedia/pulseaudio/qpulseaudiosource_p.h
+++ b/src/multimedia/pulseaudio/qpulseaudiosource_p.h
@@ -102,7 +102,7 @@ class PulseInputPrivate : public QIODevice
     Q_OBJECT
 public:
     PulseInputPrivate(QPulseAudioSource *audio);
-    ~PulseInputPrivate() {};
+    ~PulseInputPrivate() override = default;
 
     qint64 readData(char *data, qint64 len) override;
     qint64 writeData(const char *data, qint64 len) override;
diff --git a/src/multimedia/pulseaudio/qpulsehelpers_p.h b/src/multimedia/pulseaudio/qpulsehelpers_p.h
index 61907a361..762d11db3 100644
--- a/src/multimedia/pulseaudio/qpulsehelpers_p.h
+++ b/src/multimedia/pulseaudio/qpulsehelpers_p.h
@@ -24,6 +24,13 @@ QT_BEGIN_NAMESPACE
 
 Q_DECLARE_LOGGING_CATEGORY(qLcPulseAudioOut)
 
+struct PAOperationDeleter
+{
+    void operator()(pa_operation *op) const { pa_operation_unref(op); }
+};
+
+using PAOperationUPtr = std::unique_ptr<pa_operation, PAOperationDeleter>;
+
 namespace QPulseAudioInternal
 {
 pa_sample_spec audioFormatToSampleSpec(const QAudioFormat &format);