SDRdaemonSink: implemented Tx queue stabilization

plugins/samplesink/sdrdaemonsink/sdrdaemonsinkgui.cpp

@@ -23,6 +23,7 @@
 #include <QMessageBox>
 
 #include <boost/algorithm/string.hpp>
+#include <boost/lexical_cast.hpp>
 
 #include <nanomsg/nn.h>
 #include <nanomsg/pair.h>
@@ -48,6 +49,7 @@ SDRdaemonSinkGui::SDRdaemonSinkGui(DeviceSinkAPI *deviceAPI, QWidget* parent) :
 	m_sampleRate(0),
 	m_samplesCount(0),
 	m_tickCount(0),
+	m_nbSinceLastFlowCheck(0),
 	m_lastEngineState((DSPDeviceSinkEngine::State)-1),
 	m_doApplySettings(true),
 	m_forceSettings(true)
@@ -522,7 +524,7 @@ void SDRdaemonSinkGui::updateWithStreamTime()
 
 void SDRdaemonSinkGui::tick()
 {
-	if ((++m_tickCount & 0xf) == 0)
+	if ((++m_tickCount & 0xf) == 0) // 16*50ms ~800ms
 	{
 	    void *msgBuf = 0;
 
@@ -540,7 +542,36 @@ void SDRdaemonSinkGui::tick()
 
             if (nbTokens > 0) // at least the queue length is given
             {
-                ui->queueLengthText->setText(QString::fromStdString(strs[0]));
+                try
+                {
+                    int queueLength = boost::lexical_cast<int>(strs[0]);
+                    ui->queueLengthText->setText(QString::fromStdString(strs[0]));
+                    m_nbSinceLastFlowCheck++;
+                    int samplesCorr = 0;
+
+                    if (queueLength < 10)
+                    {
+//                        samplesCorr = ((10 - queueLength)*m_nbSinceLastFlowCheck)/10;
+                        samplesCorr = ((10 - queueLength)*16)/m_nbSinceLastFlowCheck;
+                    }
+                    else if (queueLength > 10)
+                    {
+//                        samplesCorr = ((10 - queueLength)*m_nbSinceLastFlowCheck)/10;
+                        samplesCorr = ((10 - queueLength)*16)/m_nbSinceLastFlowCheck;
+                    }
+
+                    if (samplesCorr != 0)
+                    {
+                        samplesCorr = samplesCorr < -50 ? -50 : samplesCorr > 50 ? 50 : samplesCorr;
+                        SDRdaemonSinkOutput::MsgConfigureSDRdaemonSinkChunkCorrection* message = SDRdaemonSinkOutput::MsgConfigureSDRdaemonSinkChunkCorrection::create(samplesCorr);
+                        m_deviceSampleSink->getInputMessageQueue()->push(message);
+                        m_nbSinceLastFlowCheck = 0;
+                    }
+                }
+                catch(const boost::bad_lexical_cast &)
+                {
+                    qDebug("SDRdaemonSinkGui::tick: queue length invalid: %s", strs[0].c_str());
+                }
             }
 
             if (nbTokens > 1) // the quality indicator is given also

plugins/samplesink/sdrdaemonsink/sdrdaemonsinkgui.h

@@ -62,6 +62,7 @@ private:
     quint64 m_deviceCenterFrequency; //!< Center frequency in device
 	int m_samplesCount;
 	std::size_t m_tickCount;
+	std::size_t m_nbSinceLastFlowCheck;
 	int m_lastEngineState;
     bool m_doApplySettings;
     bool m_forceSettings;

plugins/samplesink/sdrdaemonsink/sdrdaemonsinkgui.ui

@@ -434,7 +434,7 @@
         </size>
        </property>
        <property name="toolTip">
-        <string>Average number of blocks received / Average number of blocks used for recovery</string>
+        <string>Transmitter average number of blocks received / average number of blocks used for recovery</string>
        </property>
        <property name="text">
         <string>100.0/100.0</string>

plugins/samplesink/sdrdaemonsink/sdrdaemonsinkoutput.cpp

@@ -32,6 +32,7 @@
 MESSAGE_CLASS_DEFINITION(SDRdaemonSinkOutput::MsgConfigureSDRdaemonSink, Message)
 MESSAGE_CLASS_DEFINITION(SDRdaemonSinkOutput::MsgConfigureSDRdaemonSinkWork, Message)
 MESSAGE_CLASS_DEFINITION(SDRdaemonSinkOutput::MsgConfigureSDRdaemonSinkStreamTiming, Message)
+MESSAGE_CLASS_DEFINITION(SDRdaemonSinkOutput::MsgConfigureSDRdaemonSinkChunkCorrection, Message)
 MESSAGE_CLASS_DEFINITION(SDRdaemonSinkOutput::MsgReportSDRdaemonSinkStreamTiming, Message)
 
 SDRdaemonSinkOutput::SDRdaemonSinkOutput(DeviceSinkAPI *deviceAPI, const QTimer& masterTimer) :
@@ -149,6 +150,17 @@ bool SDRdaemonSinkOutput::handleMessage(const Message& message)
 
 		return true;
 	}
+	else if (MsgConfigureSDRdaemonSinkChunkCorrection::match(message))
+	{
+	    MsgConfigureSDRdaemonSinkChunkCorrection& conf = (MsgConfigureSDRdaemonSinkChunkCorrection&) message;
+
+	    if (m_sdrDaemonSinkThread != 0)
+        {
+	        m_sdrDaemonSinkThread->setChunkCorrection(conf.getChunkCorrection());
+        }
+
+	    return true;
+	}
 	else
 	{
 		return false;

plugins/samplesink/sdrdaemonsink/sdrdaemonsinkoutput.h

@@ -75,6 +75,26 @@ public:
 		{ }
 	};
 
+    class MsgConfigureSDRdaemonSinkChunkCorrection : public Message {
+        MESSAGE_CLASS_DECLARATION
+
+    public:
+        int getChunkCorrection() const { return m_chunkCorrection; }
+
+        static MsgConfigureSDRdaemonSinkChunkCorrection* create(int chunkCorrection)
+        {
+            return new MsgConfigureSDRdaemonSinkChunkCorrection(chunkCorrection);
+        }
+
+    private:
+        int m_chunkCorrection;
+
+        MsgConfigureSDRdaemonSinkChunkCorrection(int chunkCorrection) :
+            Message(),
+            m_chunkCorrection(chunkCorrection)
+        { }
+    };
+
 	class MsgConfigureSDRdaemonSinkStreamTiming : public Message {
 		MESSAGE_CLASS_DECLARATION
 

plugins/samplesink/sdrdaemonsink/sdrdaemonsinkthread.cpp

@@ -29,6 +29,7 @@ SDRdaemonSinkThread::SDRdaemonSinkThread(SampleSourceFifo* sampleFifo, QObject*
 	m_samplesChunkSize(0),
 	m_sampleFifo(sampleFifo),
 	m_samplesCount(0),
+	m_chunkCorrection(0),
     m_samplerate(0),
     m_throttlems(SDRDAEMONSINK_THROTTLE_MS),
     m_maxThrottlems(50),
@@ -121,6 +122,7 @@ void SDRdaemonSinkThread::tick()
         {
             m_throttlems = throttlems;
             m_samplesChunkSize = (m_samplerate * (m_throttlems+(m_throttleToggle ? 1 : 0))) / 1000;
+            m_samplesChunkSize = m_samplesChunkSize + m_chunkCorrection > 0 ? m_samplesChunkSize + m_chunkCorrection : m_samplesChunkSize;
             m_throttleToggle = !m_throttleToggle;
         }
 

plugins/samplesink/sdrdaemonsink/sdrdaemonsinkthread.h

@@ -56,6 +56,7 @@ public:
 
     std::size_t getSamplesCount() const { return m_samplesCount; }
     void setSamplesCount(int samplesCount) { m_samplesCount = samplesCount; }
+    void setChunkCorrection(int chunkCorrection) { m_chunkCorrection = chunkCorrection; }
 
 	void connectTimer(const QTimer& timer);
 
@@ -64,9 +65,10 @@ private:
 	QWaitCondition m_startWaiter;
 	bool m_running;
 
-	unsigned int m_samplesChunkSize;
+	int m_samplesChunkSize;
 	SampleSourceFifo* m_sampleFifo;
     std::size_t m_samplesCount;
+    int m_chunkCorrection;
 
 	int m_samplerate;
     int m_throttlems;