Fixed LocalSink and LocalInput samples threading model by adding one decoupling FIFO. Fixes issue #549

plugins/channelrx/localsink/localsinksink.cpp

@@ -33,6 +33,7 @@ LocalSinkSink::LocalSinkSink() :
         m_sampleRate(48000),
         m_deviceSampleRate(48000)
 {
+    m_sampleFifo.setSize(SampleSinkFifo::getSizePolicy(48000));
     applySettings(m_settings, true);
 }
 
@@ -42,7 +43,7 @@ LocalSinkSink::~LocalSinkSink()
 
 void LocalSinkSink::feed(const SampleVector::const_iterator& begin, const SampleVector::const_iterator& end)
 {
-    emit samplesAvailable((const quint8*) &(*begin), (end-begin)*sizeof(Sample));
+    m_sampleFifo.write(begin, end);
 }
 
 void LocalSinkSink::start(DeviceSampleSource *deviceSource)
@@ -54,16 +55,19 @@ void LocalSinkSink::start(DeviceSampleSource *deviceSource)
     }
 
     m_sinkThread = new LocalSinkThread();
+    m_sinkThread->setSampleFifo(&m_sampleFifo);
 
     if (deviceSource) {
-        m_sinkThread->setSampleFifo(deviceSource->getSampleFifo());
+        m_sinkThread->setDeviceSampleFifo(deviceSource->getSampleFifo());
     }
 
-    connect(this,
-            SIGNAL(samplesAvailable(const quint8*, uint)),
-            m_sinkThread,
-            SLOT(processSamples(const quint8*, uint)),
-            Qt::QueuedConnection);
+    QObject::connect(
+        &m_sampleFifo,
+        &SampleSinkFifo::dataReady,
+        m_sinkThread,
+        &LocalSinkThread::handleData,
+        Qt::QueuedConnection
+    );
 
     m_sinkThread->startStop(true);
     m_running = true;
@@ -73,10 +77,12 @@ void LocalSinkSink::stop()
 {
     qDebug("LocalSinkSink::stop");
 
-    disconnect(this,
-            SIGNAL(samplesAvailable(const quint8*, uint)),
-            m_sinkThread,
-            SLOT(processSamples(const quint8*, uint)));
+    QObject::disconnect(
+        &m_sampleFifo,
+        &SampleSinkFifo::dataReady,
+        m_sinkThread,
+        &LocalSinkThread::handleData
+    );
 
     if (m_sinkThread != 0)
     {

plugins/channelrx/localsink/localsinksink.h

@@ -40,10 +40,8 @@ public:
     void stop();
     bool isRunning() const { return m_running; }
 
-signals:
-    void samplesAvailable(const quint8* data, uint count);
-
 private:
+    SampleSinkFifo m_sampleFifo;
     LocalSinkSettings m_settings;
     LocalSinkThread *m_sinkThread;
     bool m_running;

plugins/channelrx/localsink/localsinkthread.cpp

@@ -72,10 +72,31 @@ void LocalSinkThread::run()
     qDebug("LocalSinkThread::run: end");
 }
 
-void LocalSinkThread::processSamples(const quint8* data, uint count)
+void LocalSinkThread::handleData()
 {
-    if (m_sampleFifo && m_running) {
-        m_sampleFifo->write(data, count);
+    while ((m_sampleFifo->fill() > 0) && (m_inputMessageQueue.size() == 0))
+    {
+		SampleVector::iterator part1begin;
+		SampleVector::iterator part1end;
+		SampleVector::iterator part2begin;
+		SampleVector::iterator part2end;
+
+        std::size_t count = m_sampleFifo->readBegin(m_sampleFifo->fill(), &part1begin, &part1end, &part2begin, &part2end);
+
+        if (m_deviceSampleFifo && m_running)
+        {
+            // first part of FIFO data
+            if (part1begin != part1end) {
+                m_deviceSampleFifo->write(part1begin,part1end);
+            }
+
+            // second part of FIFO data (used when block wraps around)
+            if(part2begin != part2end) {
+                m_deviceSampleFifo->write(part2begin, part2end);
+            }
+        }
+
+		m_sampleFifo->readCommit((unsigned int) count);
     }
 }
 

plugins/channelrx/localsink/localsinkthread.h

@@ -56,15 +56,17 @@ public:
 
     void startStop(bool start);
     void setSampleFifo(SampleSinkFifo *sampleFifo) { m_sampleFifo = sampleFifo; }
+    void setDeviceSampleFifo(SampleSinkFifo *sampleFifo) { m_deviceSampleFifo  = sampleFifo; }
 
 public slots:
-    void processSamples(const quint8* data, uint count);
+    void handleData(); //!< Handle data when samples have to be processed
 
 private:
 	QMutex m_startWaitMutex;
 	QWaitCondition m_startWaiter;
 	volatile bool m_running;
     SampleSinkFifo *m_sampleFifo;
+    SampleSinkFifo *m_deviceSampleFifo;
 
     MessageQueue m_inputMessageQueue;