PlutoSDR: serialize sample thread lifecycle across buddies

stop() tore the sample thread down with stopWork() + delete and only then
cleared m_deviceShared.m_thread. The buddy's applySettings() suspends and
resumes that thread through m_thread from another thread, without holding
this device's mutex, so it could restart the thread between stopWork() and
delete (QThread: Destroyed while thread is still running -> qFatal/abort),
or call through the pointer after the delete. The device's own
applySettings() had the same exposure on its own thread pointer, since it
does not take m_mutex either.

Seen when the Satellite Tracker loads presets into the Rx and Tx device
sets of one Pluto at AOS: abort in PlutoSDRInputThread::~PlutoSDRInputThread
<- PlutoSDRInput::stop <- DSPDeviceSourceEngine::gotoIdle.

Add DevicePlutoSDRShared::m_threadsMutex (recursive, shared by all PlutoSDR
device sets) and take it in start() around startWork()/publish, in stop()
and handleError() around unpublish + teardown (unpublishing first), in
suspendBuddies()/resumeBuddies(), and for the whole suspend -> apply ->
resume sequence in applySettings(). Also wait() before deleting the thread
and check the own-thread pointer before resuming it.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

devices/plutosdr/deviceplutosdrshared.cpp

@@ -22,3 +22,4 @@
 MESSAGE_CLASS_DEFINITION(DevicePlutoSDRShared::MsgCrossReportToBuddy, Message)
 
 const unsigned int DevicePlutoSDRShared::m_sampleFifoMinRate = 48000;
+QRecursiveMutex DevicePlutoSDRShared::m_threadsMutex;

devices/plutosdr/deviceplutosdrshared.h

@@ -22,6 +22,8 @@
 
 #include <stdint.h>
 
+#include <QRecursiveMutex>
+
 #include "util/message.h"
 #include "export.h"
 
@@ -94,6 +96,15 @@ public:
 
     static const unsigned int m_sampleFifoMinRate;
 
+    /**
+     * Serializes every access to m_thread across all PlutoSDR device sets.
+     * A buddy suspends and resumes this device's sample thread through m_thread, from another
+     * thread and without holding this device's own mutex. Without a common lock it can restart
+     * a thread that stop() is in the middle of tearing down (QThread: Destroyed while thread is
+     * still running) or call through a pointer that has just been deleted.
+     */
+    static QRecursiveMutex m_threadsMutex;
+
     DevicePlutoSDRShared() :
         m_deviceParams(0),
         m_thread(0),

plugins/samplesink/plutosdroutput/plutosdroutput.cpp

@@ -126,8 +126,11 @@ bool PlutoSDROutput::start()
     qDebug("PlutoSDROutput::start: thread created");
 
     m_plutoSDROutputThread->setLog2Interpolation(m_settings.m_log2Interp);
-    m_plutoSDROutputThread->startWork();
-    m_deviceShared.m_thread = m_plutoSDROutputThread;
+    {
+        QMutexLocker threadsLocker(&DevicePlutoSDRShared::m_threadsMutex);
+        m_plutoSDROutputThread->startWork();
+        m_deviceShared.m_thread = m_plutoSDROutputThread;
+    }
     m_running = true;
     mutexLocker.unlock();
 
@@ -146,14 +149,20 @@ void PlutoSDROutput::stop()
 
     m_running = false;
 
-    if (m_plutoSDROutputThread != 0)
+    // Unpublish the thread before tearing it down, and keep the shared lock through
+    // stop/wait/delete. applySettings() holds the same lock across its suspend ->
+    // apply -> resume transaction, so teardown must not interleave with that sequence.
+    QMutexLocker threadsLocker(&DevicePlutoSDRShared::m_threadsMutex);
+    m_deviceShared.m_thread = nullptr;
+    m_deviceShared.m_threadWasRunning = false;
+
+    if (m_plutoSDROutputThread)
     {
         m_plutoSDROutputThread->stopWork();
+        m_plutoSDROutputThread->wait();
         delete m_plutoSDROutputThread;
-        m_plutoSDROutputThread = 0;
+        m_plutoSDROutputThread = nullptr;
     }
-
-    m_deviceShared.m_thread = 0;
 }
 
 void PlutoSDROutput::handleError(int errorCode)
@@ -161,6 +170,9 @@ void PlutoSDROutput::handleError(int errorCode)
     QMutexLocker mutexLocker(&m_mutex);
 
     m_running = false;
+    QMutexLocker threadsLocker(&DevicePlutoSDRShared::m_threadsMutex);
+    m_deviceShared.m_thread = nullptr;
+    m_deviceShared.m_threadWasRunning = false;
     if (m_plutoSDROutputThread)
     {
         if (m_plutoSDROutputThread->isRunning())
@@ -171,7 +183,6 @@ void PlutoSDROutput::handleError(int errorCode)
         delete m_plutoSDROutputThread;
         m_plutoSDROutputThread = nullptr;
     }
-    m_deviceShared.m_thread = 0;
 
     const QString errorMessage =
         tr("Radio needs a restart: %1 (%2)")
@@ -400,6 +411,7 @@ void PlutoSDROutput::closeDevice()
 
 void PlutoSDROutput::suspendBuddies()
 {
+    QMutexLocker threadsLocker(&DevicePlutoSDRShared::m_threadsMutex);
     // suspend Rx buddy's thread
 
     for (unsigned int i = 0; i < m_deviceAPI->getSourceBuddies().size(); i++)
@@ -415,6 +427,7 @@ void PlutoSDROutput::suspendBuddies()
 
 void PlutoSDROutput::resumeBuddies()
 {
+    QMutexLocker threadsLocker(&DevicePlutoSDRShared::m_threadsMutex);
     // resume Rx buddy's thread
 
     for (unsigned int i = 0; i < m_deviceAPI->getSourceBuddies().size(); i++)
@@ -438,6 +451,10 @@ bool PlutoSDROutput::applySettings(const PlutoSDROutputSettings& settings, const
 
     qDebug().noquote() << "PlutoSDROutput::applySettings: force:" << force << settings.getDebugString(settingsKeys, force);
 
+    // Held for the whole suspend -> apply -> resume sequence below (own thread and buddies'),
+    // so that a concurrent stop()/start() of either device set cannot interleave with it.
+    QMutexLocker threadsLocker(&DevicePlutoSDRShared::m_threadsMutex);
+
     bool forwardChangeOwnDSP    = false;
     bool forwardChangeOtherDSP  = false;
     bool ownThreadWasRunning    = false;
@@ -625,7 +642,7 @@ bool PlutoSDROutput::applySettings(const PlutoSDROutputSettings& settings, const
         }
     }
 
-    if (ownThreadWasRunning) {
+    if (ownThreadWasRunning && m_plutoSDROutputThread) {
         m_plutoSDROutputThread->startWork();
     }
 

plugins/samplesource/plutosdrinput/plutosdrinput.cpp

@@ -127,8 +127,11 @@ bool PlutoSDRInput::start()
 
     m_plutoSDRInputThread->setLog2Decimation(m_settings.m_log2Decim);
     m_plutoSDRInputThread->setIQOrder(m_settings.m_iqOrder);
-    m_plutoSDRInputThread->startWork();
-    m_deviceShared.m_thread = m_plutoSDRInputThread;
+    {
+        QMutexLocker threadsLocker(&DevicePlutoSDRShared::m_threadsMutex);
+        m_plutoSDRInputThread->startWork();
+        m_deviceShared.m_thread = m_plutoSDRInputThread;
+    }
     m_running = true;
 
     mutexLocker.unlock();
@@ -147,14 +150,20 @@ void PlutoSDRInput::stop()
 
     m_running = false;
 
+    // Unpublish the thread before tearing it down, and keep the shared lock through
+    // stop/wait/delete. applySettings() holds the same lock across its suspend ->
+    // apply -> resume transaction, so teardown must not interleave with that sequence.
+    QMutexLocker threadsLocker(&DevicePlutoSDRShared::m_threadsMutex);
+    m_deviceShared.m_thread = nullptr;
+    m_deviceShared.m_threadWasRunning = false;
+
     if (m_plutoSDRInputThread)
     {
         m_plutoSDRInputThread->stopWork();
+        m_plutoSDRInputThread->wait();
         delete m_plutoSDRInputThread;
         m_plutoSDRInputThread = nullptr;
     }
-
-    m_deviceShared.m_thread = nullptr;
 }
 
 void PlutoSDRInput::handleError(int errorCode)
@@ -162,6 +171,9 @@ void PlutoSDRInput::handleError(int errorCode)
     QMutexLocker mutexLocker(&m_mutex);
 
     m_running = false;
+    QMutexLocker threadsLocker(&DevicePlutoSDRShared::m_threadsMutex);
+    m_deviceShared.m_thread = nullptr;
+    m_deviceShared.m_threadWasRunning = false;
     if (m_plutoSDRInputThread)
     {
         if (m_plutoSDRInputThread->isRunning())
@@ -172,7 +184,6 @@ void PlutoSDRInput::handleError(int errorCode)
         delete m_plutoSDRInputThread;
         m_plutoSDRInputThread = nullptr;
     }
-    m_deviceShared.m_thread = nullptr;
 
     if (m_deviceAPI->getSinkBuddies().size() == 0)
     {
@@ -419,6 +430,7 @@ void PlutoSDRInput::closeDevice()
 
 void PlutoSDRInput::suspendBuddies()
 {
+    QMutexLocker threadsLocker(&DevicePlutoSDRShared::m_threadsMutex);
     // suspend Tx buddy's thread
 
     for (unsigned int i = 0; i < m_deviceAPI->getSinkBuddies().size(); i++)
@@ -434,6 +446,7 @@ void PlutoSDRInput::suspendBuddies()
 
 void PlutoSDRInput::resumeBuddies()
 {
+    QMutexLocker threadsLocker(&DevicePlutoSDRShared::m_threadsMutex);
     // resume Tx buddy's thread
 
     for (unsigned int i = 0; i < m_deviceAPI->getSinkBuddies().size(); i++)
@@ -455,6 +468,10 @@ bool PlutoSDRInput::applySettings(const PlutoSDRInputSettings& settings, const Q
         return false;
     }
 
+    // Held for the whole suspend -> apply -> resume sequence below (own thread and buddies'),
+    // so that a concurrent stop()/start() of either device set cannot interleave with it.
+    QMutexLocker threadsLocker(&DevicePlutoSDRShared::m_threadsMutex);
+
     bool forwardChangeOwnDSP    = false;
     bool forwardChangeOtherDSP  = false;
     bool ownThreadWasRunning    = false;
@@ -687,7 +704,7 @@ bool PlutoSDRInput::applySettings(const PlutoSDRInputSettings& settings, const Q
         }
     }
 
-    if (ownThreadWasRunning) {
+    if (ownThreadWasRunning && m_plutoSDRInputThread) {
         m_plutoSDRInputThread->startWork();
     }