Airspy: refactored PerseusInputThread to PerseusInputWorker object moved to thread. Equivalent to FileInput changes

plugins/samplesource/airspy/CMakeLists.txt

@@ -5,7 +5,7 @@ set(airspy_SOURCES
     airspyplugin.cpp
     airspysettings.cpp
     airspywebapiadapter.cpp
-    airspythread.cpp
+    airspyworker.cpp
 )
 
 set(airspy_HEADERS
@@ -13,7 +13,7 @@ set(airspy_HEADERS
     airspyplugin.h
     airspysettings.h
     airspywebapiadapter.h
-    airspythread.h
+    airspyworker.h
 )
 
 include_directories(

plugins/samplesource/airspy/airspyinput.cpp

@@ -38,7 +38,7 @@
 #include "dsp/dspcommands.h"
 #include "dsp/dspengine.h"
 #include "airspysettings.h"
-#include "airspythread.h"
+#include "airspyworker.h"
 
 MESSAGE_CLASS_DEFINITION(AirspyInput::MsgConfigureAirspy, Message)
 MESSAGE_CLASS_DEFINITION(AirspyInput::MsgStartStop, Message)
@@ -51,7 +51,7 @@ AirspyInput::AirspyInput(DeviceAPI *deviceAPI) :
     m_deviceAPI(deviceAPI),
 	m_settings(),
 	m_dev(nullptr),
-	m_airspyThread(nullptr),
+	m_airspyWorker(nullptr),
 	m_deviceDescription("Airspy"),
 	m_running(false)
 {
@@ -176,24 +176,30 @@ bool AirspyInput::start()
         return false;
     }
 
-    if (m_running) { stop(); }
+    if (m_running) {
-
+        stop();
-	m_airspyThread = new AirspyThread(m_dev, &m_sampleFifo);
+    }
-	m_airspyThread->setSamplerate(m_sampleRates[m_settings.m_devSampleRateIndex]);
-	m_airspyThread->setLog2Decimation(m_settings.m_log2Decim);
-    m_airspyThread->setIQOrder(m_settings.m_iqOrder);
-	m_airspyThread->setFcPos((int) m_settings.m_fcPos);
-
-	m_airspyThread->startWork();
 
+	m_airspyWorker = new AirspyWorker(m_dev, &m_sampleFifo);
+    m_airspyWorker->moveToThread(&m_airspyWorkerThread);
+	m_airspyWorker->setSamplerate(m_sampleRates[m_settings.m_devSampleRateIndex]);
+	m_airspyWorker->setLog2Decimation(m_settings.m_log2Decim);
+    m_airspyWorker->setIQOrder(m_settings.m_iqOrder);
+	m_airspyWorker->setFcPos((int) m_settings.m_fcPos);
     mutexLocker.unlock();
 
-	applySettings(m_settings, true);
+    if (startWorker())
-
+    {
         qDebug("AirspyInput::startInput: started");
+        applySettings(m_settings, true);
         m_running = true;
+    }
+    else
+    {
+        m_running = false;
+    }
 
-	return true;
+	return m_running;
 }
 
 void AirspyInput::closeDevice()
@@ -214,16 +220,36 @@ void AirspyInput::stop()
 	qDebug("AirspyInput::stop");
 	QMutexLocker mutexLocker(&m_mutex);
 
-	if (m_airspyThread)
+	if (m_airspyWorker)
 	{
-		m_airspyThread->stopWork();
+		stopWorker();
-		delete m_airspyThread;
+		delete m_airspyWorker;
-		m_airspyThread = nullptr;
+		m_airspyWorker = nullptr;
 	}
 
 	m_running = false;
 }
 
+bool AirspyInput::startWorker()
+{
+	if (m_airspyWorker->startWork())
+    {
+    	m_airspyWorkerThread.start();
+        return true;
+    }
+    else
+    {
+        return false;
+    }
+}
+
+void AirspyInput::stopWorker()
+{
+	m_airspyWorker->stopWork();
+	m_airspyWorkerThread.quit();
+	m_airspyWorkerThread.wait();
+}
+
 QByteArray AirspyInput::serialize() const
 {
     return m_settings.serialize();
@@ -402,10 +428,10 @@ bool AirspyInput::applySettings(const AirspySettings& settings, bool force)
 			{
 				qCritical("AirspyInput::applySettings: could not set sample rate index %u (%d S/s): %s", settings.m_devSampleRateIndex, m_sampleRates[settings.m_devSampleRateIndex], airspy_error_name(rc));
 			}
-			else if (m_airspyThread)
+			else if (m_airspyWorker)
 			{
 				qDebug("AirspyInput::applySettings: sample rate set to index: %u (%d S/s)", settings.m_devSampleRateIndex, m_sampleRates[settings.m_devSampleRateIndex]);
-				m_airspyThread->setSamplerate(m_sampleRates[settings.m_devSampleRateIndex]);
+				m_airspyWorker->setSamplerate(m_sampleRates[settings.m_devSampleRateIndex]);
 			}
 		}
 	}
@@ -415,9 +441,9 @@ bool AirspyInput::applySettings(const AirspySettings& settings, bool force)
         reverseAPIKeys.append("log2Decim");
 		forwardChange = true;
 
-		if (m_airspyThread)
+		if (m_airspyWorker)
 		{
-			m_airspyThread->setLog2Decimation(settings.m_log2Decim);
+			m_airspyWorker->setLog2Decimation(settings.m_log2Decim);
 			qDebug() << "AirspyInput: set decimation to " << (1<<settings.m_log2Decim);
 		}
 	}
@@ -426,8 +452,8 @@ bool AirspyInput::applySettings(const AirspySettings& settings, bool force)
 	{
         reverseAPIKeys.append("iqOrder");
 
-		if (m_airspyThread) {
+		if (m_airspyWorker) {
-			m_airspyThread->setIQOrder(settings.m_iqOrder);
+			m_airspyWorker->setIQOrder(settings.m_iqOrder);
 		}
 	}
 
@@ -472,9 +498,9 @@ bool AirspyInput::applySettings(const AirspySettings& settings, bool force)
 
 	if ((m_settings.m_fcPos != settings.m_fcPos) || force)
 	{
-		if (m_airspyThread)
+		if (m_airspyWorker)
 		{
-			m_airspyThread->setFcPos((int) settings.m_fcPos);
+			m_airspyWorker->setFcPos((int) settings.m_fcPos);
 			qDebug() << "AirspyInput: set fc pos (enum) to " << (int) settings.m_fcPos;
 		}
 	}

plugins/samplesource/airspy/airspyinput.h

@@ -21,6 +21,7 @@
 #include <QString>
 #include <QByteArray>
 #include <QNetworkRequest>
+#include <QThread>
 
 #include <libairspy/airspy.h>
 #include <dsp/devicesamplesource.h>
@@ -29,7 +30,7 @@
 class QNetworkAccessManager;
 class QNetworkReply;
 class DeviceAPI;
-class AirspyThread;
+class AirspyWorker;
 class FileRecord;
 
 class AirspyInput : public DeviceSampleSource {
@@ -162,7 +163,8 @@ private:
 	QMutex m_mutex;
 	AirspySettings m_settings;
 	struct airspy_device* m_dev;
-	AirspyThread* m_airspyThread;
+	AirspyWorker* m_airspyWorker;
+    QThread m_airspyWorkerThread;
 	QString m_deviceDescription;
 	std::vector<uint32_t> m_sampleRates;
 	bool m_running;
@@ -170,6 +172,8 @@ private:
     QNetworkAccessManager *m_networkManager;
     QNetworkRequest m_networkRequest;
 
+	bool startWorker();
+	void stopWorker();
 	bool openDevice();
 	void closeDevice();
 	bool applySettings(const AirspySettings& settings, bool force);

plugins/samplesource/airspy/airspyworker.cpp

@@ -19,14 +19,14 @@
 #include <errno.h>
 #include <algorithm>
 
-#include "airspythread.h"
+#include "airspyworker.h"
 
 #include "dsp/samplesinkfifo.h"
 
-AirspyThread *AirspyThread::m_this = 0;
+AirspyWorker *AirspyWorker::m_this = 0;
 
-AirspyThread::AirspyThread(struct airspy_device* dev, SampleSinkFifo* sampleFifo, QObject* parent) :
+AirspyWorker::AirspyWorker(struct airspy_device* dev, SampleSinkFifo* sampleFifo, QObject* parent) :
-	QThread(parent),
+	QObject(parent),
 	m_running(false),
 	m_dev(dev),
 	m_convertBuffer(AIRSPY_BLOCKSIZE),
@@ -40,80 +40,61 @@ AirspyThread::AirspyThread(struct airspy_device* dev, SampleSinkFifo* sampleFifo
 	std::fill(m_buf, m_buf + 2*AIRSPY_BLOCKSIZE, 0);
 }
 
-AirspyThread::~AirspyThread()
+AirspyWorker::~AirspyWorker()
 {
 	stopWork();
 	m_this = 0;
 }
 
-void AirspyThread::startWork()
+bool AirspyWorker::startWork()
 {
-	m_startWaitMutex.lock();
+	airspy_error rc;
-	start();
-	while(!m_running)
-		m_startWaiter.wait(&m_startWaitMutex, 100);
-	m_startWaitMutex.unlock();
-}
 
-void AirspyThread::stopWork()
+	rc = (airspy_error) airspy_start_rx(m_dev, rx_callback, NULL);
+
+	if (rc == AIRSPY_SUCCESS)
 	{
-	qDebug("AirspyThread::stopWork");
+    	m_running = (airspy_is_streaming(m_dev) == AIRSPY_TRUE);
+	}
+    else
+    {
+		qCritical("AirspyWorker::run: failed to start Airspy Rx: %s", airspy_error_name(rc));
         m_running = false;
-	wait();
     }
 
-void AirspyThread::setSamplerate(uint32_t samplerate)
+    return m_running;
+}
+
+void AirspyWorker::stopWork()
+{
+	airspy_error rc = (airspy_error) airspy_stop_rx(m_dev);
+
+	if (rc == AIRSPY_SUCCESS) {
+		qDebug("AirspyWorker::run: stopped Airspy Rx");
+	} else {
+		qDebug("AirspyWorker::run: failed to stop Airspy Rx: %s", airspy_error_name(rc));
+	}
+
+	m_running = false;
+}
+
+void AirspyWorker::setSamplerate(uint32_t samplerate)
 {
 	m_samplerate = samplerate;
 }
 
-void AirspyThread::setLog2Decimation(unsigned int log2_decim)
+void AirspyWorker::setLog2Decimation(unsigned int log2_decim)
 {
 	m_log2Decim = log2_decim;
 }
 
-void AirspyThread::setFcPos(int fcPos)
+void AirspyWorker::setFcPos(int fcPos)
 {
 	m_fcPos = fcPos;
 }
 
-void AirspyThread::run()
-{
-	airspy_error rc;
-
-	m_running = true;
-	m_startWaiter.wakeAll();
-
-	rc = (airspy_error) airspy_start_rx(m_dev, rx_callback, NULL);
-
-	if (rc != AIRSPY_SUCCESS)
-	{
-		qCritical("AirspyThread::run: failed to start Airspy Rx: %s", airspy_error_name(rc));
-	}
-	else
-	{
-		while ((m_running) && (airspy_is_streaming(m_dev) == AIRSPY_TRUE))
-		{
-			sleep(1);
-		}
-	}
-
-	rc = (airspy_error) airspy_stop_rx(m_dev);
-
-	if (rc == AIRSPY_SUCCESS)
-	{
-		qDebug("AirspyThread::run: stopped Airspy Rx");
-	}
-	else
-	{
-		qDebug("AirspyThread::run: failed to stop Airspy Rx: %s", airspy_error_name(rc));
-	}
-
-	m_running = false;
-}
-
 //  Decimate according to specified log2 (ex: log2=4 => decim=16)
-void AirspyThread::callbackIQ(const qint16* buf, qint32 len)
+void AirspyWorker::callbackIQ(const qint16* buf, qint32 len)
 {
 	SampleVector::iterator it = m_convertBuffer.begin();
 
@@ -206,7 +187,7 @@ void AirspyThread::callbackIQ(const qint16* buf, qint32 len)
 	m_sampleFifo->write(m_convertBuffer.begin(), it);
 }
 
-void AirspyThread::callbackQI(const qint16* buf, qint32 len)
+void AirspyWorker::callbackQI(const qint16* buf, qint32 len)
 {
 	SampleVector::iterator it = m_convertBuffer.begin();
 
@@ -299,7 +280,7 @@ void AirspyThread::callbackQI(const qint16* buf, qint32 len)
 	m_sampleFifo->write(m_convertBuffer.begin(), it);
 }
 
-int AirspyThread::rx_callback(airspy_transfer_t* transfer)
+int AirspyWorker::rx_callback(airspy_transfer_t* transfer)
 {
 	qint32 bytes_to_write = transfer->sample_count * sizeof(qint16);
 

plugins/samplesource/airspy/airspyworker.h

@@ -15,12 +15,10 @@
 // along with this program. If not, see <http://www.gnu.org/licenses/>.          //
 ///////////////////////////////////////////////////////////////////////////////////
 
-#ifndef INCLUDE_AIRSPYTHREAD_H
+#ifndef INCLUDE_AIRSPYWORKER_H
-#define INCLUDE_AIRSPYTHREAD_H
+#define INCLUDE_AIRSPYWORKER_H
 
-#include <QThread>
+#include <QObject>
-#include <QMutex>
-#include <QWaitCondition>
 #include <libairspy/airspy.h>
 
 #include "dsp/samplesinkfifo.h"
@@ -28,14 +26,14 @@
 
 #define AIRSPY_BLOCKSIZE (1<<17)
 
-class AirspyThread : public QThread {
+class AirspyWorker : public QObject {
 	Q_OBJECT
 
 public:
-	AirspyThread(struct airspy_device* dev, SampleSinkFifo* sampleFifo, QObject* parent = NULL);
+	AirspyWorker(struct airspy_device* dev, SampleSinkFifo* sampleFifo, QObject* parent = NULL);
-	~AirspyThread();
+	~AirspyWorker();
 
-	void startWork();
+	bool startWork();
 	void stopWork();
 	void setSamplerate(uint32_t samplerate);
 	void setLog2Decimation(unsigned int log2_decim);
@@ -43,8 +41,6 @@ public:
     void setIQOrder(bool iqOrder) { m_iqOrder = iqOrder; }
 
 private:
-	QMutex m_startWaitMutex;
-	QWaitCondition m_startWaiter;
 	bool m_running;
 
 	struct airspy_device* m_dev;
@@ -56,15 +52,14 @@ private:
 	unsigned int m_log2Decim;
 	int m_fcPos;
     bool m_iqOrder;
-	static AirspyThread *m_this;
+	static AirspyWorker *m_this;
 
 	Decimators<qint32, qint16, SDR_RX_SAMP_SZ, 12, true> m_decimatorsIQ;
 	Decimators<qint32, qint16, SDR_RX_SAMP_SZ, 12, false> m_decimatorsQI;
 
-	void run();
 	void callbackIQ(const qint16* buf, qint32 len);
 	void callbackQI(const qint16* buf, qint32 len);
 	static int rx_callback(airspy_transfer_t* transfer);
 };
 
-#endif // INCLUDE_AIRSPYTHREAD_H
+#endif // INCLUDE_AIRSPYWORKER_H