Remote Sink: threading refactoring

plugins/channelrx/CMakeLists.txt

@@ -13,9 +13,9 @@ if(LIBDSDCC_FOUND AND LIBMBE_FOUND)
     add_subdirectory(demoddsd)
 endif(LIBDSDCC_FOUND AND LIBMBE_FOUND)
 
-# if(CM256CC_FOUND)
+if(CM256CC_FOUND)
-#     add_subdirectory(remotesink)
+    add_subdirectory(remotesink)
-# endif(CM256CC_FOUND)
+endif(CM256CC_FOUND)
 
 if (CODEC2_FOUND)
     add_subdirectory(demodfreedv)

plugins/channelrx/remotesink/CMakeLists.txt

@@ -15,7 +15,8 @@ set(remotesink_SOURCES
     remotesinksink.cpp
     remotesinksettings.cpp
     remotesinkwebapiadapter.cpp
-	remotesinkthread.cpp
+    remotesinksender.cpp
+    remotesinkfifo.cpp
 	remotesinkplugin.cpp
 )
 
@@ -25,7 +26,8 @@ set(remotesink_HEADERS
     remotesinksink.h
     remotesinksettings.h
     remotesinkwebapiadapter.h
-	remotesinkthread.h
+    remotesinksender.h
+    remotesinkfifo.h
 	remotesinkplugin.h
 )
 

plugins/channelrx/remotesink/remotesink.cpp

@@ -34,6 +34,7 @@
 #include "dsp/dspcommands.h"
 #include "dsp/hbfilterchainconverter.h"
 #include "dsp/devicesamplemimo.h"
+#include "dsp/dspdevicesourceengine.h"
 #include "device/deviceapi.h"
 
 #include "remotesinkbaseband.h"
@@ -95,13 +96,13 @@ void RemoteSink::start()
     }
 
     m_thread->start();
-    m_basebandSink->startSink();
+    m_basebandSink->startSender();
 }
 
 void RemoteSink::stop()
 {
     qDebug("RemoteSink::stop");
-    m_basebandSink->stopSink();
+    m_basebandSink->stopSender();
 	m_thread->exit();
 	m_thread->wait();
 }
@@ -130,7 +131,7 @@ bool RemoteSink::handleMessage(const Message& cmd)
         if (getMessageQueueToGUI())
         {
             DSPSignalNotification* msgToGUI = new DSPSignalNotification(notif); // make a copy
-            getMessageQueueToGUI()->push(msgToBaseband);
+            getMessageQueueToGUI()->push(msgToGUI);
         }
 
 	    return true;

plugins/channelrx/remotesink/remotesinkbaseband.cpp

@@ -116,7 +116,8 @@ bool RemoteSinkBaseband::handleMessage(const Message& cmd)
         m_basebandSampleRate = notif.getSampleRate();
         qDebug() << "RemoteSinkBaseband::handleMessage: DSPSignalNotification: basebandSampleRate:" << m_basebandSampleRate;
         m_channelizer->setBasebandSampleRate(m_basebandSampleRate);
-        m_sink.applySampleRate(m_basebandSampleRate/ (1<<m_settings.m_log2Decim));
+        m_sink.applyBasebandSampleRate(m_basebandSampleRate);
+        m_sink.setDeviceCenterFrequency(notif.getCenterFrequency());
 
         return true;
     }
@@ -137,7 +138,6 @@ void RemoteSinkBaseband::applySettings(const RemoteSinkSettings& settings, bool
      || (settings.m_filterChainHash != m_settings.m_filterChainHash) || force)
     {
         m_channelizer->setDecimation(settings.m_log2Decim, settings.m_filterChainHash);
-        m_sink.applySampleRate(m_basebandSampleRate/ (1<<settings.m_log2Decim));
     }
 
     m_sink.applySettings(settings, force);
@@ -153,5 +153,5 @@ void RemoteSinkBaseband::setBasebandSampleRate(int sampleRate)
 {
     m_basebandSampleRate = sampleRate;
     m_channelizer->setBasebandSampleRate(m_basebandSampleRate);
-    m_sink.applySampleRate(m_basebandSampleRate/ (1<<m_settings.m_log2Decim));
+    m_sink.applyBasebandSampleRate(m_basebandSampleRate);
 }

plugins/channelrx/remotesink/remotesinkbaseband.h

@@ -59,13 +59,15 @@ public:
 
     RemoteSinkBaseband();
     ~RemoteSinkBaseband();
+
     void reset();
 	void feed(const SampleVector::const_iterator& begin, const SampleVector::const_iterator& end);
+    void startSender() { m_sink.startSender(); }
+    void stopSender() { m_sink.stopSender(); }
+
     MessageQueue *getInputMessageQueue() { return &m_inputMessageQueue; } //!< Get the queue for asynchronous inbound communication
     int getChannelSampleRate() const;
     void setBasebandSampleRate(int sampleRate);
-    void startSink() { m_sink.start(); }
-    void stopSink() { m_sink.stop(); }
 
 private:
     SampleSinkFifo m_sampleFifo;

plugins/channelrx/remotesink/remotesinkfifo.cpp

@@ -0,0 +1,95 @@
+///////////////////////////////////////////////////////////////////////////////////
+// Copyright (C) 2019 Edouard Griffiths, F4EXB                                   //
+//                                                                               //
+// This program is free software; you can redistribute it and/or modify          //
+// it under the terms of the GNU General Public License as published by          //
+// the Free Software Foundation as version 3 of the License, or                  //
+// (at your option) any later version.                                           //
+//                                                                               //
+// This program is distributed in the hope that it will be useful,               //
+// but WITHOUT ANY WARRANTY; without even the implied warranty of                //
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the                  //
+// GNU General Public License V3 for more details.                               //
+//                                                                               //
+// You should have received a copy of the GNU General Public License             //
+// along with this program. If not, see <http://www.gnu.org/licenses/>.          //
+///////////////////////////////////////////////////////////////////////////////////
+
+#include "remotesinkfifo.h"
+
+RemoteSinkFifo::RemoteSinkFifo(QObject *parent) :
+    QObject(parent)
+{}
+
+RemoteSinkFifo::RemoteSinkFifo(unsigned int size, QObject *parent) :
+    QObject(parent)
+{
+    resize(size);
+}
+
+RemoteSinkFifo::~RemoteSinkFifo()
+{}
+
+void RemoteSinkFifo::resize(unsigned int size)
+{
+    QMutexLocker mutexLocker(&m_mutex);
+    m_size = size;
+    m_data.resize(m_size);
+    m_readHead = 0;
+    m_servedHead = 0;
+    m_writeHead = 0;
+}
+
+void RemoteSinkFifo::reset()
+{
+    m_readHead = 0;
+    m_servedHead = 0;
+    m_writeHead = 0;
+}
+
+RemoteDataBlock *RemoteSinkFifo::getDataBlock()
+{
+    QMutexLocker mutexLocker(&m_mutex);
+    m_servedHead = m_writeHead;
+
+    if (m_writeHead < m_size - 1) {
+        m_writeHead++;
+    } else {
+        m_writeHead = 0;
+    }
+
+    emit dataBlockServed();
+    return &m_data[m_servedHead];
+}
+
+unsigned int RemoteSinkFifo::readDataBlock(RemoteDataBlock **dataBlock)
+{
+    QMutexLocker mutexLocker(&m_mutex);
+
+    if (calculateRemainder() == 0)
+    {
+        *dataBlock = nullptr;
+        return 0;
+    }
+    else
+    {
+        *dataBlock = &m_data[m_readHead];
+        m_readHead = m_readHead < m_size - 1 ? m_readHead + 1 : 0;
+        return calculateRemainder();
+    }
+}
+
+unsigned int RemoteSinkFifo::getRemainder()
+{
+    QMutexLocker mutexLocker(&m_mutex);
+    return calculateRemainder();
+}
+
+unsigned int RemoteSinkFifo::calculateRemainder()
+{
+    if (m_readHead <= m_servedHead) {
+        return m_servedHead - m_readHead;
+    } else {
+        return m_size - (m_readHead - m_servedHead);
+    }
+}

plugins/channelrx/remotesink/remotesinkfifo.h

@@ -0,0 +1,55 @@
+///////////////////////////////////////////////////////////////////////////////////
+// Copyright (C) 2019 Edouard Griffiths, F4EXB                                   //
+//                                                                               //
+// This program is free software; you can redistribute it and/or modify          //
+// it under the terms of the GNU General Public License as published by          //
+// the Free Software Foundation as version 3 of the License, or                  //
+// (at your option) any later version.                                           //
+//                                                                               //
+// This program is distributed in the hope that it will be useful,               //
+// but WITHOUT ANY WARRANTY; without even the implied warranty of                //
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the                  //
+// GNU General Public License V3 for more details.                               //
+//                                                                               //
+// You should have received a copy of the GNU General Public License             //
+// along with this program. If not, see <http://www.gnu.org/licenses/>.          //
+///////////////////////////////////////////////////////////////////////////////////
+
+#ifndef REMOTESINK_REMOTESINKFIFO_H_
+#define REMOTESINK_REMOTESINKFIFO_H_
+
+#include <vector>
+
+#include <QObject>
+#include <QMutex>
+
+#include "channel/remotedatablock.h"
+
+class RemoteSinkFifo : public QObject {
+    Q_OBJECT
+public:
+    RemoteSinkFifo(QObject *parent = nullptr);
+    RemoteSinkFifo(unsigned int size, QObject *parent = nullptr);
+    ~RemoteSinkFifo();
+    void resize(unsigned int size);
+    void reset();
+
+    RemoteDataBlock *getDataBlock();
+    unsigned int readDataBlock(RemoteDataBlock **dataBlock);
+    unsigned int getRemainder();
+
+signals:
+    void dataBlockServed();
+
+private:
+    std::vector<RemoteDataBlock> m_data;
+    int m_size;
+    unsigned int m_readHead;   //!< index of last data block processed
+    unsigned int m_servedHead; //!< index of last data block served
+    unsigned int m_writeHead;  //!< index of next data block to serve
+    QMutex m_mutex;
+
+    unsigned int calculateRemainder();
+};
+
+#endif // REMOTESINK_REMOTESINKFIFO_H_

plugins/channelrx/remotesink/remotesinkgui.cpp

@@ -296,6 +296,7 @@ void RemoteSinkGUI::onMenuDialogCalled(const QPoint &p)
 void RemoteSinkGUI::on_decimationFactor_currentIndexChanged(int index)
 {
     m_settings.m_log2Decim = index;
+    updateTxDelayTime();
     applyDecimation();
 }
 
@@ -368,7 +369,8 @@ void RemoteSinkGUI::updateTxDelayTime()
 {
     double txDelayRatio = m_settings.m_txDelay / 100.0;
     int samplesPerBlock = RemoteNbBytesPerBlock / sizeof(Sample);
-    double delay = m_basebandSampleRate == 0 ? 0.0 : (127*samplesPerBlock*txDelayRatio) / m_basebandSampleRate;
+    int channelSampleRate = m_basebandSampleRate / (1<<m_settings.m_log2Decim);
+    double delay = channelSampleRate == 0 ? 0.0 : (127*samplesPerBlock*txDelayRatio) / channelSampleRate;
     delay /= 128 + m_settings.m_nbFECBlocks;
     ui->txDelayTime->setText(tr("%1µs").arg(QString::number(delay*1e6, 'f', 0)));
 }

plugins/channelrx/remotesink/remotesinksender.cpp

@@ -21,90 +21,73 @@
 // along with this program. If not, see <http://www.gnu.org/licenses/>.          //
 ///////////////////////////////////////////////////////////////////////////////////
 
-#include "remotesinkthread.h"
 
-#include <channel/remotedatablock.h>
+#include <thread>
+#include <chrono>
+
 #include <QUdpSocket>
 
 #include "cm256cc/cm256.h"
 
-MESSAGE_CLASS_DEFINITION(RemoteSinkThread::MsgStartStop, Message)
+#include "channel/remotedatablock.h"
+#include "remotesinksender.h"
 
-RemoteSinkThread::RemoteSinkThread(QObject* parent) :
+RemoteSinkSender::RemoteSinkSender() :
-    QThread(parent),
+    m_fifo(20, this),
-    m_running(false),
     m_address(QHostAddress::LocalHost),
-    m_socket(0)
+    m_socket(nullptr)
 {
-
+    qDebug("RemoteSinkSender::RemoteSinkSender");
-    m_cm256p = m_cm256.isInitialized() ? &m_cm256 : 0;
+    m_cm256p = m_cm256.isInitialized() ? &m_cm256 : nullptr;
-    connect(&m_inputMessageQueue, SIGNAL(messageEnqueued()), this, SLOT(handleInputMessages()), Qt::QueuedConnection);
-}
-
-RemoteSinkThread::~RemoteSinkThread()
-{
-    qDebug("RemoteSinkThread::~RemoteSinkThread");
-}
-
-void RemoteSinkThread::startStop(bool start)
-{
-    MsgStartStop *msg = MsgStartStop::create(start);
-    m_inputMessageQueue.push(msg);
-}
-
-void RemoteSinkThread::startWork()
-{
-    qDebug("RemoteSinkThread::startWork");
-	m_startWaitMutex.lock();
     m_socket = new QUdpSocket(this);
-	start();
+
-	while(!m_running)
+    QObject::connect(
-		m_startWaiter.wait(&m_startWaitMutex, 100);
+        &m_fifo,
-	m_startWaitMutex.unlock();
+        &RemoteSinkFifo::dataBlockServed,
+        this,
+        &RemoteSinkSender::handleData,
+        Qt::QueuedConnection
+    );
 }
 
-void RemoteSinkThread::stopWork()
+RemoteSinkSender::~RemoteSinkSender()
 {
-	qDebug("RemoteSinkThread::stopWork");
+    qDebug("RemoteSinkSender::~RemoteSinkSender");
     delete m_socket;
-    m_socket = 0;
-	m_running = false;
-	wait();
 }
 
-void RemoteSinkThread::run()
+RemoteDataBlock *RemoteSinkSender::getDataBlock()
 {
-    qDebug("RemoteSinkThread::run: begin");
+    return m_fifo.getDataBlock();
-	m_running = true;
+}
-	m_startWaiter.wakeAll();
 
-    while (m_running)
+void RemoteSinkSender::handleData()
+{
+    RemoteDataBlock *dataBlock;
+    unsigned int remainder = m_fifo.getRemainder();
+
+    while (remainder != 0)
     {
-        sleep(1); // Do nothing as everything is in the data handler (dequeuer)
+        remainder = m_fifo.readDataBlock(&dataBlock);
+
+        if (dataBlock) {
+            sendDataBlock(dataBlock);
+        }
     }
-
-    m_running = false;
-    qDebug("RemoteSinkThread::run: end");
 }
 
-void RemoteSinkThread::processDataBlock(RemoteDataBlock *dataBlock)
+void RemoteSinkSender::sendDataBlock(RemoteDataBlock *dataBlock)
-{
-    handleDataBlock(*dataBlock);
-    delete dataBlock;
-}
-
-void RemoteSinkThread::handleDataBlock(RemoteDataBlock& dataBlock)
 {
 	CM256::cm256_encoder_params cm256Params;  //!< Main interface with CM256 encoder
 	CM256::cm256_block descriptorBlocks[256]; //!< Pointers to data for CM256 encoder
 	RemoteProtectedBlock fecBlocks[256];   //!< FEC data
 
-    uint16_t frameIndex = dataBlock.m_txControlBlock.m_frameIndex;
+    uint16_t frameIndex = dataBlock->m_txControlBlock.m_frameIndex;
-    int nbBlocksFEC = dataBlock.m_txControlBlock.m_nbBlocksFEC;
+    int nbBlocksFEC = dataBlock->m_txControlBlock.m_nbBlocksFEC;
-    int txDelay = dataBlock.m_txControlBlock.m_txDelay;
+    int txDelay = dataBlock->m_txControlBlock.m_txDelay;
-    m_address.setAddress(dataBlock.m_txControlBlock.m_dataAddress);
+    m_address.setAddress(dataBlock->m_txControlBlock.m_dataAddress);
-    uint16_t dataPort = dataBlock.m_txControlBlock.m_dataPort;
+    uint16_t dataPort = dataBlock->m_txControlBlock.m_dataPort;
-    RemoteSuperBlock *txBlockx = dataBlock.m_superBlocks;
+    RemoteSuperBlock *txBlockx = dataBlock->m_superBlocks;
 
     if ((nbBlocksFEC == 0) || !m_cm256p) // Do not FEC encode
     {
@@ -114,7 +97,7 @@ void RemoteSinkThread::handleDataBlock(RemoteDataBlock& dataBlock)
             {
                 // send block via UDP
                 m_socket->writeDatagram((const char*)&txBlockx[i], (qint64 ) RemoteUdpSize, m_address, dataPort);
-                //usleep(txDelay);
+                std::this_thread::sleep_for(std::chrono::microseconds(txDelay));
             }
         }
     }
@@ -142,7 +125,7 @@ void RemoteSinkThread::handleDataBlock(RemoteDataBlock& dataBlock)
         // Encode FEC blocks
         if (m_cm256p->cm256_encode(cm256Params, descriptorBlocks, fecBlocks))
         {
-            qWarning("RemoteSinkThread::handleDataBlock: CM256 encode failed. No transmission.");
+            qWarning("RemoteSinkSender::handleDataBlock: CM256 encode failed. No transmission.");
             // TODO: send without FEC changing meta data to set indication of no FEC
         }
 
@@ -159,32 +142,10 @@ void RemoteSinkThread::handleDataBlock(RemoteDataBlock& dataBlock)
             {
                 // send block via UDP
                 m_socket->writeDatagram((const char*)&txBlockx[i], (qint64 ) RemoteUdpSize, m_address, dataPort);
-                //usleep(txDelay);
+                std::this_thread::sleep_for(std::chrono::microseconds(txDelay));
             }
         }
     }
 
-    dataBlock.m_txControlBlock.m_processed = true;
+    dataBlock->m_txControlBlock.m_processed = true;
-}
-
-void RemoteSinkThread::handleInputMessages()
-{
-    Message* message;
-
-    while ((message = m_inputMessageQueue.pop()) != 0)
-    {
-        if (MsgStartStop::match(*message))
-        {
-            MsgStartStop* notif = (MsgStartStop*) message;
-            qDebug("RemoteSinkThread::handleInputMessages: MsgStartStop: %s", notif->getStartStop() ? "start" : "stop");
-
-            if (notif->getStartStop()) {
-                startWork();
-            } else {
-                stopWork();
-            }
-
-            delete message;
-        }
-    }
 }

plugins/channelrx/remotesink/remotesinksender.h

@@ -21,10 +21,10 @@
 // along with this program. If not, see <http://www.gnu.org/licenses/>.          //
 ///////////////////////////////////////////////////////////////////////////////////
 
-#ifndef PLUGINS_CHANNELRX_REMOTESINK_REMOTESINKTHREAD_H_
+#ifndef PLUGINS_CHANNELRX_REMOTESINK_REMOTESINKSENDER_H_
-#define PLUGINS_CHANNELRX_REMOTESINK_REMOTESINKTHREAD_H_
+#define PLUGINS_CHANNELRX_REMOTESINK_REMOTESINKSENDER_H_
 
-#include <QThread>
+#include <QObject>
 #include <QMutex>
 #include <QWaitCondition>
 #include <QHostAddress>
@@ -34,63 +34,34 @@
 #include "util/message.h"
 #include "util/messagequeue.h"
 
+#include "remotesinkfifo.h"
+
 class RemoteDataBlock;
 class CM256;
 class QUdpSocket;
 
-class RemoteSinkThread : public QThread {
+class RemoteSinkSender : public QObject {
     Q_OBJECT
 
 public:
-    class MsgStartStop : public Message {
+    RemoteSinkSender();
-        MESSAGE_CLASS_DECLARATION
+    ~RemoteSinkSender();
 
-    public:
+    RemoteDataBlock *getDataBlock();
-        bool getStartStop() const { return m_startStop; }
-
-        static MsgStartStop* create(bool startStop) {
-            return new MsgStartStop(startStop);
-        }
-
-    protected:
-        bool m_startStop;
-
-        MsgStartStop(bool startStop) :
-            Message(),
-            m_startStop(startStop)
-        { }
-    };
-
-    RemoteSinkThread(QObject* parent = 0);
-    ~RemoteSinkThread();
-
-    void startStop(bool start);
-
-public slots:
-    void processDataBlock(RemoteDataBlock *dataBlock);
 
 private:
-	QMutex m_startWaitMutex;
+    RemoteSinkFifo m_fifo;
-	QWaitCondition m_startWaiter;
-	volatile bool m_running;
-
     CM256 m_cm256;
     CM256 *m_cm256p;
 
     QHostAddress m_address;
     QUdpSocket *m_socket;
 
-    MessageQueue m_inputMessageQueue;
+    void sendDataBlock(RemoteDataBlock *dataBlock);
-
-    void startWork();
-    void stopWork();
-
-    void run();
-    void handleDataBlock(RemoteDataBlock& dataBlock);
 
 private slots:
-    void handleInputMessages();
+    void handleData();
 };
 
-#endif // PLUGINS_CHANNELRX_REMOTESINK_REMOTESINKTHREAD_H_
+#endif // PLUGINS_CHANNELRX_REMOTESINK_REMOTESINKSENDER_H_
 

plugins/channelrx/remotesink/remotesinksink.cpp

@@ -16,6 +16,7 @@
 ///////////////////////////////////////////////////////////////////////////////////
 
 #include <QMutexLocker>
+#include <QThread>
 
 #include <boost/crc.hpp>
 #include <boost/cstdint.hpp>
@@ -23,47 +24,64 @@
 #include "dsp/hbfilterchainconverter.h"
 #include "util/timeutil.h"
 
-#include "remotesinkthread.h"
+#include "remotesinksender.h"
 #include "remotesinksink.h"
 
 RemoteSinkSink::RemoteSinkSink() :
-        m_running(false),
-        m_remoteSinkThread(nullptr),
         m_txBlockIndex(0),
         m_frameCount(0),
         m_sampleIndex(0),
         m_dataBlock(nullptr),
-        m_centerFrequency(0),
+        m_deviceCenterFrequency(0),
         m_frequencyOffset(0),
-        m_sampleRate(48000),
+        m_basebandSampleRate(48000),
         m_nbBlocksFEC(0),
         m_txDelay(35),
         m_dataAddress("127.0.0.1"),
         m_dataPort(9090)
 {
+    qDebug("RemoteSinkSink::RemoteSinkSink");
+
+    m_senderThread = new QThread(this);
+    m_remoteSinkSender = new RemoteSinkSender();
+    m_remoteSinkSender->moveToThread(m_senderThread);
+
     applySettings(m_settings, true);
 }
 
 RemoteSinkSink::~RemoteSinkSink()
 {
-    QMutexLocker mutexLocker(&m_dataBlockMutex);
+    qDebug("RemoteSinkSink::~RemoteSinkSink");
 
-    if (m_dataBlock && !m_dataBlock->m_txControlBlock.m_complete) {
+    delete m_remoteSinkSender;
-        delete m_dataBlock;
+    delete m_senderThread;
-    }
 }
 
-void RemoteSinkSink::setTxDelay(int txDelay, int nbBlocksFEC)
+void RemoteSinkSink::startSender()
+{
+    qDebug("RemoteSinkSink::startSender");
+    m_senderThread->start();
+}
+
+void RemoteSinkSink::stopSender()
+{
+    qDebug("RemoteSinkSink::stopSender");
+	m_senderThread->exit();
+	m_senderThread->wait();
+}
+
+void RemoteSinkSink::setTxDelay(int txDelay, int nbBlocksFEC, int log2Decim)
 {
     double txDelayRatio = txDelay / 100.0;
     int samplesPerBlock = RemoteNbBytesPerBlock / sizeof(Sample);
-    double delay = m_sampleRate == 0 ? 1.0 : (127*samplesPerBlock*txDelayRatio) / m_sampleRate;
+    int sampleRate = m_basebandSampleRate / (1<<log2Decim);
+    double delay = sampleRate == 0 ? 1.0 : (127*samplesPerBlock*txDelayRatio) / sampleRate;
     delay /= 128 + nbBlocksFEC;
     m_txDelay = roundf(delay*1e6); // microseconds
     qDebug() << "RemoteSinkSink::setTxDelay:"
         << "txDelay:" << txDelay << "%"
         << "m_txDelay:" << m_txDelay << "us"
-        << "m_sampleRate: " << m_sampleRate << "S/s";
+        << "sampleRate: " << sampleRate << "S/s";
 }
 
 void RemoteSinkSink::setNbBlocksFEC(int nbBlocksFEC)
@@ -88,8 +106,8 @@ void RemoteSinkSink::feed(const SampleVector::const_iterator& begin, const Sampl
             uint64_t nowus = TimeUtil::nowus();
             // gettimeofday(&tv, 0);
 
-            metaData.m_centerFrequency = m_centerFrequency + m_frequencyOffset;
+            metaData.m_centerFrequency = m_deviceCenterFrequency + m_frequencyOffset;
-            metaData.m_sampleRate = m_sampleRate;
+            metaData.m_sampleRate = m_basebandSampleRate / (1<<m_settings.m_log2Decim);
             metaData.m_sampleBytes = (SDR_RX_SAMP_SZ <= 16 ? 2 : 4);
             metaData.m_sampleBits = SDR_RX_SAMP_SZ;
             metaData.m_nbOriginalBlocks = RemoteNbOrginalBlocks;
@@ -98,7 +116,7 @@ void RemoteSinkSink::feed(const SampleVector::const_iterator& begin, const Sampl
             metaData.m_tv_usec = nowus % 1000000UL; // tv.tv_usec;
 
             if (!m_dataBlock) { // on the very first cycle there is no data block allocated
-                m_dataBlock = new RemoteDataBlock();
+                m_dataBlock = m_remoteSinkSender->getDataBlock(); // ask a new block to sender
             }
 
             boost::crc_32_type crc32;
@@ -158,7 +176,6 @@ void RemoteSinkSink::feed(const SampleVector::const_iterator& begin, const Sampl
 
             if (m_txBlockIndex == RemoteNbOrginalBlocks - 1) // frame complete
             {
-                m_dataBlockMutex.lock();
                 m_dataBlock->m_txControlBlock.m_frameIndex = m_frameCount;
                 m_dataBlock->m_txControlBlock.m_processed = false;
                 m_dataBlock->m_txControlBlock.m_complete = true;
@@ -167,9 +184,7 @@ void RemoteSinkSink::feed(const SampleVector::const_iterator& begin, const Sampl
                 m_dataBlock->m_txControlBlock.m_dataAddress = m_dataAddress;
                 m_dataBlock->m_txControlBlock.m_dataPort = m_dataPort;
 
-                emit dataBlockAvailable(m_dataBlock);
+                m_dataBlock = m_remoteSinkSender->getDataBlock(); // ask a new block to sender
-                m_dataBlock = new RemoteDataBlock(); // create a new one immediately
-                m_dataBlockMutex.unlock();
 
                 m_txBlockIndex = 0;
                 m_frameCount++;
@@ -182,39 +197,6 @@ void RemoteSinkSink::feed(const SampleVector::const_iterator& begin, const Sampl
     }
 }
 
-void RemoteSinkSink::start()
-{
-    qDebug("RemoteSinkSink::start");
-
-    memset((void *) &m_currentMetaFEC, 0, sizeof(RemoteMetaDataFEC));
-
-    if (m_running) {
-        stop();
-    }
-
-    m_remoteSinkThread = new RemoteSinkThread();
-    connect(this,
-            SIGNAL(dataBlockAvailable(RemoteDataBlock *)),
-            m_remoteSinkThread,
-            SLOT(processDataBlock(RemoteDataBlock *)),
-            Qt::QueuedConnection);
-    m_remoteSinkThread->startStop(true);
-    m_running = true;
-}
-
-void RemoteSinkSink::stop()
-{
-    qDebug("RemoteSinkSink::stop");
-
-    if (m_remoteSinkThread)
-    {
-        m_remoteSinkThread->startStop(false);
-        m_remoteSinkThread->deleteLater();
-    }
-
-    m_running = false;
-}
-
 void RemoteSinkSink::applySettings(const RemoteSinkSettings& settings, bool force)
 {
     qDebug() << "RemoteSinkSink::applySettings:"
@@ -225,16 +207,6 @@ void RemoteSinkSink::applySettings(const RemoteSinkSettings& settings, bool forc
             << " m_streamIndex: " << settings.m_streamIndex
             << " force: " << force;
 
-    if ((m_settings.m_nbFECBlocks != settings.m_nbFECBlocks) || force)
-    {
-        setNbBlocksFEC(settings.m_nbFECBlocks);
-        setTxDelay(settings.m_txDelay, settings.m_nbFECBlocks);
-    }
-
-    if ((m_settings.m_txDelay != settings.m_txDelay) || force) {
-        setTxDelay(settings.m_txDelay, settings.m_nbFECBlocks);
-    }
-
     if ((m_settings.m_dataAddress != settings.m_dataAddress) || force) {
         m_dataAddress = settings.m_dataAddress;
     }
@@ -243,11 +215,24 @@ void RemoteSinkSink::applySettings(const RemoteSinkSettings& settings, bool forc
         m_dataPort = settings.m_dataPort;
     }
 
+    if ((m_settings.m_log2Decim != settings.m_log2Decim)
+     || (m_settings.m_filterChainHash != settings.m_filterChainHash)
+     || (m_settings.m_nbFECBlocks != settings.m_nbFECBlocks)
+     || (m_settings.m_txDelay != settings.m_txDelay) || force)
+    {
+        double shiftFactor = HBFilterChainConverter::getShiftFactor(settings.m_log2Decim, settings.m_filterChainHash);
+        m_frequencyOffset = round(shiftFactor*m_basebandSampleRate);
+        setNbBlocksFEC(settings.m_nbFECBlocks);
+        setTxDelay(settings.m_txDelay, settings.m_nbFECBlocks, settings.m_log2Decim);
+    }
+
     m_settings = settings;
 }
 
-void RemoteSinkSink::applySampleRate(uint32_t sampleRate)
+void RemoteSinkSink::applyBasebandSampleRate(uint32_t sampleRate)
 {
-    m_sampleRate = sampleRate;
+    m_basebandSampleRate = sampleRate;
-    setTxDelay(m_settings.m_txDelay, m_settings.m_nbFECBlocks);
+    double shiftFactor = HBFilterChainConverter::getShiftFactor(m_settings.m_log2Decim, m_settings.m_filterChainHash);
+    m_frequencyOffset = round(shiftFactor*m_basebandSampleRate);
+    setTxDelay(m_settings.m_txDelay, m_settings.m_nbFECBlocks, m_settings.m_log2Decim);
 }

plugins/channelrx/remotesink/remotesinksink.h

@@ -19,7 +19,6 @@
 #define INCLUDE_REMOTESINKSINK_H_
 
 #include <QObject>
-#include <QMutex>
 
 #include "dsp/channelsamplesink.h"
 #include "channel/remotedatablock.h"
@@ -28,7 +27,8 @@
 #include "remotesinksettings.h"
 
 class DeviceSampleSource;
-class RemoteSinkThread;
+class RemoteSinkSender;
+class QThread;
 
 class RemoteSinkSink : public QObject, public ChannelSampleSink {
     Q_OBJECT
@@ -38,29 +38,26 @@ public:
 
 	virtual void feed(const SampleVector::const_iterator& begin, const SampleVector::const_iterator& end);
 
-    void applySettings(const RemoteSinkSettings& settings, bool force = false);
+    void startSender();
-    void applySampleRate(uint32_t sampleRate);
+    void stopSender();
-    void start();
-    void stop();
-    bool isRunning() const { return m_running; }
 
-    /** Set center frequency given in Hz */
+    void applySettings(const RemoteSinkSettings& settings, bool force = false);
-    void setCenterFrequency(uint64_t centerFrequency) { m_centerFrequency = centerFrequency; }
+    void applyBasebandSampleRate(uint32_t sampleRate);
+
+    /** Set device center frequency given in Hz */
+    void setDeviceCenterFrequency(uint64_t centerFrequency) { m_deviceCenterFrequency = centerFrequency; }
 
     /** Set sample rate given in Hz */
 
     void setNbBlocksFEC(int nbBlocksFEC);
-    void setTxDelay(int txDelay, int nbBlocksFEC);
+    void setTxDelay(int txDelay, int nbBlocksFEC, int log2Decim);
     void setDataAddress(const QString& address) { m_dataAddress = address; }
     void setDataPort(uint16_t port) { m_dataPort = port; }
 
-signals:
-    void dataBlockAvailable(RemoteDataBlock *dataBlock);
-
 private:
     RemoteSinkSettings m_settings;
-    RemoteSinkThread *m_remoteSinkThread;
+    QThread *m_senderThread;
-    bool m_running;
+    RemoteSinkSender *m_remoteSinkSender;
 
     int m_txBlockIndex;                  //!< Current index in blocks to transmit in the Tx row
     uint16_t m_frameCount;               //!< transmission frame count
@@ -68,11 +65,10 @@ private:
     RemoteSuperBlock m_superBlock;
     RemoteMetaDataFEC m_currentMetaFEC;
     RemoteDataBlock *m_dataBlock;
-    QMutex m_dataBlockMutex;
 
-    uint64_t m_centerFrequency;
+    uint64_t m_deviceCenterFrequency;
     int64_t m_frequencyOffset;
-    uint32_t m_sampleRate;
+    uint32_t m_basebandSampleRate;
     int m_nbBlocksFEC;
     int m_txDelay;
     QString m_dataAddress;