BladerRF2 input support. Trying to fix threading issue

devices/bladerf2/devicebladerf2shared.cpp

@@ -21,8 +21,7 @@ MESSAGE_CLASS_DEFINITION(DeviceBladeRF2Shared::MsgReportBuddyChange, Message)
 DeviceBladeRF2Shared::DeviceBladeRF2Shared() :
     m_dev(0),
     m_channel(-1),
-    m_inputThread(0),
+    m_source(0)
-    m_outputThread(0)
 {}
 
 DeviceBladeRF2Shared::~DeviceBladeRF2Shared()

devices/bladerf2/devicebladerf2shared.h

@@ -22,6 +22,7 @@
 
 class SampleSinkFifo;
 class SampleSourceFifo;
+class BladeRF2Input;
 
 /**
  * Structure shared by a buddy with other buddies
@@ -29,34 +30,6 @@ class SampleSourceFifo;
 class DEVICES_API DeviceBladeRF2Shared
 {
 public:
-    class InputThreadInterface
-    {
-    public:
-        virtual ~InputThreadInterface() {}
-        virtual void startWork() = 0;
-        virtual void stopWork() = 0;
-        virtual bool isRunning() const = 0;
-        virtual unsigned int getNbChannels() const = 0;
-        virtual void setLog2Decimation(unsigned int channel, unsigned int log2_decim) = 0;
-        virtual unsigned int getLog2Decimation(unsigned int channel) const = 0;
-        virtual void setFcPos(unsigned int channel, int fcPos) = 0;
-        virtual int getFcPos(unsigned int channel) const = 0;
-        virtual void setFifo(unsigned int channel, SampleSinkFifo *fifo) = 0;
-        virtual SampleSinkFifo *getFifo(unsigned int channel) = 0;
-    };
-
-    class OutputThreadInterface
-    {
-    public:
-        virtual ~OutputThreadInterface() {}
-        virtual void startWork() = 0;
-        virtual void stopWork() = 0;
-        virtual bool isRunning() = 0;
-        virtual unsigned int getNbChannels() const = 0;
-        virtual void setFifo(unsigned int channel, SampleSourceFifo *fifo) = 0;
-        virtual SampleSourceFifo *getFifo(unsigned int channel) = 0;
-    };
-
     class MsgReportBuddyChange : public Message {
         MESSAGE_CLASS_DECLARATION
 
@@ -82,8 +55,7 @@ public:
 
     DeviceBladeRF2 *m_dev;
     int m_channel; //!< allocated channel (-1 if none)
-    InputThreadInterface *m_inputThread;   //!< The SISO/MIMO input thread
+    BladeRF2Input *m_source;
-    OutputThreadInterface *m_outputThread; //!< The SISO/MIMO output thread
 };
 
 

plugins/samplesource/bladerf2input/bladerf2input.cpp

@@ -189,6 +189,7 @@ bool BladeRF2Input::openDevice()
         }
     }
 
+    m_deviceShared.m_source = this;
     m_deviceAPI->setBuddySharedPtr(&m_deviceShared); // propagate common parameters to API
     return true;
 }
@@ -204,6 +205,7 @@ void BladeRF2Input::closeDevice()
     }
 
     m_deviceShared.m_channel = -1;
+    m_deviceShared.m_source = 0;
 
     // No buddies so effectively close the device
 
@@ -220,6 +222,38 @@ void BladeRF2Input::init()
     applySettings(m_settings, true);
 }
 
+BladeRF2InputThread *BladeRF2Input::findThread()
+{
+    if (m_thread == 0) // this does not own the thread
+    {
+        BladeRF2InputThread *bladerf2InputThread = 0;
+
+        // find a buddy that has allocated the thread
+        const std::vector<DeviceSourceAPI*>& sourceBuddies = m_deviceAPI->getSourceBuddies();
+        std::vector<DeviceSourceAPI*>::const_iterator it = sourceBuddies.begin();
+
+        for (; it != sourceBuddies.end(); ++it)
+        {
+            BladeRF2Input *buddySource = ((DeviceBladeRF2Shared*) (*it)->getBuddySharedPtr())->m_source;
+
+            if (buddySource)
+            {
+                bladerf2InputThread = buddySource->getThread();
+
+                if (bladerf2InputThread) {
+                    break;
+                }
+            }
+        }
+
+        return bladerf2InputThread;
+    }
+    else
+    {
+        return m_thread; // own thread
+    }
+}
+
 bool BladeRF2Input::start()
 {
     if (!m_deviceShared.m_dev)
@@ -228,36 +262,21 @@ bool BladeRF2Input::start()
         return false;
     }
 
-    Bladerf2InputThread *bladerf2InputThread = 0;
+    BladeRF2InputThread *bladerf2InputThread = findThread();
     bool needsStart = false;
 
-    // find thread allocated by a buddy
+    if (bladerf2InputThread) // if thread is already allocated
-    const std::vector<DeviceSourceAPI*>& sourceBuddies = m_deviceAPI->getSourceBuddies();
-    std::vector<DeviceSourceAPI*>::const_iterator it = sourceBuddies.begin();
-
-    for (; it != sourceBuddies.end(); ++it)
     {
-        bladerf2InputThread = (Bladerf2InputThread*) ((DeviceBladeRF2Shared*) (*it)->getBuddySharedPtr())->m_inputThread;
-
-        if (bladerf2InputThread) {
-            break;
-        }
-    }
-
-    if (bladerf2InputThread) // if thread was allocated by a buddy
-    {
-        DeviceSourceAPI *sourceBuddy = m_deviceAPI->getSourceBuddies()[0];
-        DeviceBladeRF2Shared *deviceBladeRF2Shared = (DeviceBladeRF2Shared*) sourceBuddy->getBuddySharedPtr();
-        bladerf2InputThread = (Bladerf2InputThread*) deviceBladeRF2Shared->m_inputThread;
         int nbOriginalChannels = bladerf2InputThread->getNbChannels();
 
-        if (m_deviceShared.m_channel+1 > nbOriginalChannels) // expansion
+        if (m_deviceShared.m_channel+1 > nbOriginalChannels) // expansion by deleting and re-creating the thread
         {
             SampleSinkFifo **fifos = new SampleSinkFifo*[nbOriginalChannels];
             unsigned int *log2Decims = new unsigned int[nbOriginalChannels];
             int *fcPoss = new int[nbOriginalChannels];
 
-            for (int i = 0; i < nbOriginalChannels; i++) { // save original FIFO references and data
+            for (int i = 0; i < nbOriginalChannels; i++) // save original FIFO references and data
+            {
                 fifos[i] = bladerf2InputThread->getFifo(i);
                 log2Decims[i] = bladerf2InputThread->getLog2Decimation(i);
                 fcPoss[i] = bladerf2InputThread->getFcPos(i);
@@ -265,20 +284,22 @@ bool BladeRF2Input::start()
 
             bladerf2InputThread->stopWork();
             delete bladerf2InputThread;
-            bladerf2InputThread = new Bladerf2InputThread(m_deviceShared.m_dev->getDev(), m_deviceShared.m_channel+1);
+            bladerf2InputThread = new BladeRF2InputThread(m_deviceShared.m_dev->getDev(), m_deviceShared.m_channel+1);
+            m_thread = bladerf2InputThread; // take ownership
 
-            for (int i = 0; i < nbOriginalChannels; i++) { // restore original FIFO references
+            for (int i = 0; i < nbOriginalChannels; i++) // restore original FIFO references
+            {
                 bladerf2InputThread->setFifo(i, fifos[i]);
                 bladerf2InputThread->setLog2Decimation(i, log2Decims[i]);
                 bladerf2InputThread->setFcPos(i, fcPoss[i]);
             }
 
-            // propagate new thread address to buddies
+            // remove old thread address from buddies (reset in all buddies)
             const std::vector<DeviceSourceAPI*>& sourceBuddies = m_deviceAPI->getSourceBuddies();
             std::vector<DeviceSourceAPI*>::const_iterator it = sourceBuddies.begin();
 
             for (; it != sourceBuddies.end(); ++it) {
-                ((DeviceBladeRF2Shared*) (*it)->getBuddySharedPtr())->m_inputThread = bladerf2InputThread;
+                ((DeviceBladeRF2Shared*) (*it)->getBuddySharedPtr())->m_source->resetThread();
             }
 
             needsStart = true;
@@ -286,14 +307,14 @@ bool BladeRF2Input::start()
     }
     else // first allocation
     {
-        bladerf2InputThread = new Bladerf2InputThread(m_deviceShared.m_dev->getDev(), m_deviceShared.m_channel+1);
+        bladerf2InputThread = new BladeRF2InputThread(m_deviceShared.m_dev->getDev(), m_deviceShared.m_channel+1);
+        m_thread = bladerf2InputThread; // take ownership
         needsStart = true;
     }
 
     bladerf2InputThread->setFifo(m_deviceShared.m_channel, &m_sampleFifo);
     bladerf2InputThread->setLog2Decimation(m_deviceShared.m_channel, m_settings.m_log2Decim);
     bladerf2InputThread->setFcPos(m_deviceShared.m_channel, (int) m_settings.m_fcPos);
-    m_deviceShared.m_inputThread = bladerf2InputThread;
 
     if (needsStart) {
         bladerf2InputThread->startWork();
@@ -301,7 +322,7 @@ bool BladeRF2Input::start()
 
     applySettings(m_settings, true);
 
-    qDebug("BladerfInput::startInput: started");
+    qDebug("BladeRF2Input::start: started");
     m_running = true;
 
     return true;
@@ -313,55 +334,70 @@ void BladeRF2Input::stop()
         return;
     }
 
-    int nbOriginalChannels = m_deviceShared.m_inputThread->getNbChannels();
+    BladeRF2InputThread *bladerf2InputThread = findThread();
-    Bladerf2InputThread *bladerf2InputThread = 0;
 
-    if (nbOriginalChannels == 1) // SI mode
+    if (bladerf2InputThread == 0) // no thread allocated
     {
-        m_deviceShared.m_inputThread->stopWork();
+        return;
-        bladerf2InputThread = (Bladerf2InputThread*) m_deviceShared.m_inputThread;
-        delete bladerf2InputThread;
-        m_deviceShared.m_inputThread = 0;
-        m_running = false;
     }
-    else if (m_deviceShared.m_channel == nbOriginalChannels - 1) // remove last MI channel => reduce
+
+    int nbOriginalChannels = bladerf2InputThread->getNbChannels();
+
+    if (nbOriginalChannels == 1) // SI mode => just stop and delete the thread
     {
-        m_deviceShared.m_inputThread->stopWork();
+        bladerf2InputThread->stopWork();
-        SampleSinkFifo **fifos = new SampleSinkFifo*[nbOriginalChannels-1];
-        unsigned int *log2Decims = new unsigned int[nbOriginalChannels-1];
-        int *fcPoss = new int[nbOriginalChannels-1];
-
-        for (int i = 0; i < nbOriginalChannels-1; i++) { // save original FIFO references
-            fifos[i] = m_deviceShared.m_inputThread->getFifo(i);
-            log2Decims[i] = m_deviceShared.m_inputThread->getLog2Decimation(i);
-            fcPoss[i] = m_deviceShared.m_inputThread->getFcPos(i);
-        }
-
-        bladerf2InputThread = (Bladerf2InputThread*) m_deviceShared.m_inputThread;
         delete bladerf2InputThread;
-        bladerf2InputThread = new Bladerf2InputThread(m_deviceShared.m_dev->getDev(), nbOriginalChannels-1);
+        m_thread = 0;
-        m_deviceShared.m_inputThread = bladerf2InputThread;
 
-        for (int i = 0; i < nbOriginalChannels-1; i++) { // restore original FIFO references
+        // remove old thread address from buddies (reset in all buddies)
-            m_deviceShared.m_inputThread->setFifo(i, fifos[i]);
-            m_deviceShared.m_inputThread->setLog2Decimation(i, log2Decims[i]);
-            m_deviceShared.m_inputThread->setFcPos(i, fcPoss[i]);
-        }
-
-        // propagate new thread address to buddies
         const std::vector<DeviceSourceAPI*>& sourceBuddies = m_deviceAPI->getSourceBuddies();
         std::vector<DeviceSourceAPI*>::const_iterator it = sourceBuddies.begin();
 
         for (; it != sourceBuddies.end(); ++it) {
-            ((DeviceBladeRF2Shared*) (*it)->getBuddySharedPtr())->m_inputThread = m_deviceShared.m_inputThread;
+            ((DeviceBladeRF2Shared*) (*it)->getBuddySharedPtr())->m_source->resetThread();
+        }
+    }
+    else if (m_deviceShared.m_channel == nbOriginalChannels - 1) // remove last MI channel => reduce by deleting and re-creating the thread
+    {
+        bladerf2InputThread->stopWork();
+        SampleSinkFifo **fifos = new SampleSinkFifo*[nbOriginalChannels-1];
+        unsigned int *log2Decims = new unsigned int[nbOriginalChannels-1];
+        int *fcPoss = new int[nbOriginalChannels-1];
+
+        for (int i = 0; i < nbOriginalChannels-1; i++) // save original FIFO references
+        {
+            fifos[i] = bladerf2InputThread->getFifo(i);
+            log2Decims[i] = bladerf2InputThread->getLog2Decimation(i);
+            fcPoss[i] = bladerf2InputThread->getFcPos(i);
         }
 
-        m_deviceShared.m_inputThread->startWork();
+        delete bladerf2InputThread;
+        bladerf2InputThread = new BladeRF2InputThread(m_deviceShared.m_dev->getDev(), nbOriginalChannels-1);
+        m_thread = bladerf2InputThread; // take ownership
+
+        for (int i = 0; i < nbOriginalChannels-1; i++)  // restore original FIFO references
+        {
+            bladerf2InputThread->setFifo(i, fifos[i]);
+            bladerf2InputThread->setLog2Decimation(i, log2Decims[i]);
+            bladerf2InputThread->setFcPos(i, fcPoss[i]);
+        }
+
+        // remove old thread address from buddies (reset in all buddies)
+        const std::vector<DeviceSourceAPI*>& sourceBuddies = m_deviceAPI->getSourceBuddies();
+        std::vector<DeviceSourceAPI*>::const_iterator it = sourceBuddies.begin();
+
+        for (; it != sourceBuddies.end(); ++it) {
+            ((DeviceBladeRF2Shared*) (*it)->getBuddySharedPtr())->m_source->resetThread();
+        }
+
+        bladerf2InputThread->startWork();
     }
-    else
+    else // remove channel from existing thread
     {
-        m_deviceShared.m_inputThread->setFifo(m_deviceShared.m_channel, 0); // remove FIFO
+        bladerf2InputThread->setFifo(m_deviceShared.m_channel, 0); // remove FIFO
     }
+
+    m_running = false;
 }
 
 QByteArray BladeRF2Input::serialize() const
@@ -670,9 +706,11 @@ bool BladeRF2Input::applySettings(const BladeRF2InputSettings& settings, bool fo
 
     if ((m_settings.m_fcPos != settings.m_fcPos) || force)
     {
-        if (m_deviceShared.m_inputThread != 0)
+        BladeRF2InputThread *inputThread = findThread();
+
+        if (inputThread != 0)
         {
-            m_deviceShared.m_inputThread->setFcPos(m_deviceShared.m_channel, (int) settings.m_fcPos);
+            inputThread->setFcPos(m_deviceShared.m_channel, (int) settings.m_fcPos);
             qDebug() << "BladeRF2Input::applySettings: set fc pos (enum) to " << (int) settings.m_fcPos;
         }
     }
@@ -680,10 +718,11 @@ bool BladeRF2Input::applySettings(const BladeRF2InputSettings& settings, bool fo
     if ((m_settings.m_log2Decim != settings.m_log2Decim) || force)
     {
         forwardChangeOwnDSP = true;
+        BladeRF2InputThread *inputThread = findThread();
 
-        if (m_deviceShared.m_inputThread != 0)
+        if (inputThread != 0)
         {
-            m_deviceShared.m_inputThread->setLog2Decimation(m_deviceShared.m_channel, settings.m_log2Decim);
+            inputThread->setLog2Decimation(m_deviceShared.m_channel, settings.m_log2Decim);
             qDebug() << "BladeRF2Input::applySettings: set decimation to " << (1<<settings.m_log2Decim);
         }
     }

plugins/samplesource/bladerf2input/bladerf2input.h

@@ -100,6 +100,8 @@ public:
     virtual void init();
     virtual bool start();
     virtual void stop();
+    BladeRF2InputThread *getThread() { return m_thread; }
+    void resetThread() { m_thread = 0; }
 
     virtual QByteArray serialize() const;
     virtual bool deserialize(const QByteArray& data);
@@ -152,6 +154,7 @@ private:
 
     bool openDevice();
     void closeDevice();
+    BladeRF2InputThread *findThread();
     bool applySettings(const BladeRF2InputSettings& settings, bool force = false);
     void webapiFormatDeviceSettings(SWGSDRangel::SWGDeviceSettings& response, const BladeRF2InputSettings& settings);
     void webapiFormatDeviceReport(SWGSDRangel::SWGDeviceReport& response);

plugins/samplesource/bladerf2input/bladerf2inputthread.cpp

@@ -16,7 +16,7 @@
 
 #include "bladerf2inputthread.h"
 
-Bladerf2InputThread::Bladerf2InputThread(struct bladerf* dev, unsigned int nbRxChannels, QObject* parent) :
+BladeRF2InputThread::BladeRF2InputThread(struct bladerf* dev, unsigned int nbRxChannels, QObject* parent) :
     QThread(parent),
     m_running(false),
     m_dev(dev),
@@ -31,9 +31,9 @@ Bladerf2InputThread::Bladerf2InputThread(struct bladerf* dev, unsigned int nbRxC
     m_buf = new qint16[2*DeviceBladeRF2::blockSize*nbRxChannels];
 }
 
-Bladerf2InputThread::~Bladerf2InputThread()
+BladeRF2InputThread::~BladeRF2InputThread()
 {
-    qDebug("Bladerf2InputThread::~Bladerf2InputThread");
+    qDebug("BladeRF2InputThread::~BladeRF2InputThread");
 
     if (m_running) {
         stopWork();
@@ -43,7 +43,7 @@ Bladerf2InputThread::~Bladerf2InputThread()
     delete[] m_channels;
 }
 
-void Bladerf2InputThread::startWork()
+void BladeRF2InputThread::startWork()
 {
     m_startWaitMutex.lock();
     start();
@@ -55,13 +55,13 @@ void Bladerf2InputThread::startWork()
     m_startWaitMutex.unlock();
 }
 
-void Bladerf2InputThread::stopWork()
+void BladeRF2InputThread::stopWork()
 {
     m_running = false;
     wait();
 }
 
-void Bladerf2InputThread::run()
+void BladeRF2InputThread::run()
 {
     int res;
 
@@ -82,7 +82,7 @@ void Bladerf2InputThread::run()
 
         if (status < 0)
         {
-            qCritical("Bladerf2InputThread::run: cannot configure streams: %s", bladerf_strerror(status));
+            qCritical("BladeRF2InputThread::run: cannot configure streams: %s", bladerf_strerror(status));
         }
         else
         {
@@ -92,7 +92,7 @@ void Bladerf2InputThread::run()
 
                 if (res < 0)
                 {
-                    qCritical("BladerfThread::run sync Rx error: %s", bladerf_strerror(res));
+                    qCritical("BladeRF2InputThread::run sync Rx error: %s", bladerf_strerror(res));
                     break;
                 }
 
@@ -106,14 +106,14 @@ void Bladerf2InputThread::run()
     }
     else
     {
-        qWarning("Bladerf2InputThread::run: no channels or FIFO allocated. Aborting");
+        qWarning("BladeRF2InputThread::run: no channels or FIFO allocated. Aborting");
     }
 
 
     m_running = false;
 }
 
-unsigned int Bladerf2InputThread::getNbFifos()
+unsigned int BladeRF2InputThread::getNbFifos()
 {
     unsigned int fifoCount = 0;
 
@@ -127,14 +127,14 @@ unsigned int Bladerf2InputThread::getNbFifos()
     return fifoCount;
 }
 
-void Bladerf2InputThread::setLog2Decimation(unsigned int channel, unsigned int log2_decim)
+void BladeRF2InputThread::setLog2Decimation(unsigned int channel, unsigned int log2_decim)
 {
     if (channel < m_nbChannels) {
         m_channels[channel].m_log2Decim = log2_decim;
     }
 }
 
-unsigned int Bladerf2InputThread::getLog2Decimation(unsigned int channel) const
+unsigned int BladeRF2InputThread::getLog2Decimation(unsigned int channel) const
 {
     if (channel < m_nbChannels) {
         return m_channels[channel].m_log2Decim;
@@ -143,14 +143,14 @@ unsigned int Bladerf2InputThread::getLog2Decimation(unsigned int channel) const
     }
 }
 
-void Bladerf2InputThread::setFcPos(unsigned int channel, int fcPos)
+void BladeRF2InputThread::setFcPos(unsigned int channel, int fcPos)
 {
     if (channel < m_nbChannels) {
         m_channels[channel].m_fcPos = fcPos;
     }
 }
 
-int Bladerf2InputThread::getFcPos(unsigned int channel) const
+int BladeRF2InputThread::getFcPos(unsigned int channel) const
 {
     if (channel < m_nbChannels) {
         return m_channels[channel].m_fcPos;
@@ -159,14 +159,14 @@ int Bladerf2InputThread::getFcPos(unsigned int channel) const
     }
 }
 
-void Bladerf2InputThread::setFifo(unsigned int channel, SampleSinkFifo *sampleFifo)
+void BladeRF2InputThread::setFifo(unsigned int channel, SampleSinkFifo *sampleFifo)
 {
     if (channel < m_nbChannels) {
         m_channels[channel].m_sampleFifo = sampleFifo;
     }
 }
 
-SampleSinkFifo *Bladerf2InputThread::getFifo(unsigned int channel)
+SampleSinkFifo *BladeRF2InputThread::getFifo(unsigned int channel)
 {
     if (channel < m_nbChannels) {
         return m_channels[channel].m_sampleFifo;
@@ -175,14 +175,14 @@ SampleSinkFifo *Bladerf2InputThread::getFifo(unsigned int channel)
     }
 }
 
-void Bladerf2InputThread::callbackMI(const qint16* buf, qint32 samplesPerChannel)
+void BladeRF2InputThread::callbackMI(const qint16* buf, qint32 samplesPerChannel)
 {
     // TODO: write a set of decimators that can take interleaved samples in input directly
     int status = bladerf_deinterleave_stream_buffer(BLADERF_RX_X2, BLADERF_FORMAT_SC16_Q11 , samplesPerChannel*m_nbChannels, (void *) buf);
 
     if (status < 0)
     {
-        qCritical("Bladerf2InputThread::callbackMI: cannot de-interleave buffer: %s", bladerf_strerror(status));
+        qCritical("BladeRF2InputThread::callbackMI: cannot de-interleave buffer: %s", bladerf_strerror(status));
         return;
     }
 
@@ -194,7 +194,7 @@ void Bladerf2InputThread::callbackMI(const qint16* buf, qint32 samplesPerChannel
     }
 }
 
-void Bladerf2InputThread::callbackSI(const qint16* buf, qint32 len, unsigned int channel)
+void BladeRF2InputThread::callbackSI(const qint16* buf, qint32 len, unsigned int channel)
 {
     SampleVector::iterator it = m_channels[channel].m_convertBuffer.begin();
 

plugins/samplesource/bladerf2input/bladerf2inputthread.h

@@ -32,12 +32,12 @@
 #include "dsp/samplesinkfifo.h"
 #include "dsp/decimators.h"
 
-class Bladerf2InputThread : public QThread, public DeviceBladeRF2Shared::InputThreadInterface {
+class BladeRF2InputThread : public QThread {
     Q_OBJECT
 
 public:
-    Bladerf2InputThread(struct bladerf* dev, unsigned int nbRxChannels, QObject* parent = NULL);
+    BladeRF2InputThread(struct bladerf* dev, unsigned int nbRxChannels, QObject* parent = NULL);
-    ~Bladerf2InputThread();
+    ~BladeRF2InputThread();
 
     virtual void startWork();
     virtual void stopWork();