SDR->Demod->Audio chain all using std::thread

2% CPU usage drop on 3.2G i7 desktop

CMakeLists.txt

@@ -91,8 +91,6 @@ SET (cubicsdr_sources
 	src/AppFrame.cpp
 	src/sdr/SDRThread.cpp
 	src/demod/DemodulatorThread.cpp
-	src/demod/DemodulatorThreadQueue.cpp
-	src/demod/DemodulatorThreadTask.cpp
 	src/demod/DemodulatorMgr.cpp
 	src/audio/AudioThread.cpp
 	src/util/Gradient.cpp
@@ -112,8 +110,6 @@ SET (cubicsdr_headers
 	src/AppFrame.h
 	src/sdr/SDRThread.h
 	src/demod/DemodulatorThread.h
-	src/demod/DemodulatorThreadQueue.h
-	src/demod/DemodulatorThreadTask.h
 	src/demod/DemodulatorMgr.h
 	src/audio/AudioThread.h
 	src/util/Gradient.h

src/AppFrame.cpp

@@ -65,11 +65,14 @@ AppFrame::AppFrame() :
 
     demodulatorTest = demodMgr.newThread(this);
     demodulatorTest->params.audioInputQueue = audioInputQueue;
-    demodulatorTest->run();
+    demodulatorTest->init();
+
+    audioVisualQueue = new DemodulatorThreadOutputQueue();
+    demodulatorTest->setVisualOutputQueue(audioVisualQueue);
 
     threadCmdQueueSDR = new SDRThreadCommandQueue;
     sdrThread = new SDRThread(threadCmdQueueSDR);
-    sdrThread->bindDemodulator(*demodulatorTest);
+    sdrThread->bindDemodulator(demodulatorTest);
 
     iqVisualQueue = new SDRThreadIQDataQueue;
     sdrThread->setIQVisualQueue(iqVisualQueue);
@@ -101,40 +104,15 @@ void AppFrame::OnNewWindow(wxCommandEvent& WXUNUSED(event)) {
 
 void AppFrame::OnThread(wxCommandEvent& event) {
     SDRThreadIQData *iqData;
-    DemodulatorThreadAudioData *demodAudioData;
 
     std::vector<signed char> *new_uc_buffer;
     std::vector<float> *new_float_buffer;
     std::string asdf("beep");
 
-    switch (event.GetId()) {
-
     // SDR IQ -> Demodulator
 
-    case DemodulatorThreadTask::DEMOD_THREAD_AUDIO_DATA:
-        demodAudioData = (DemodulatorThreadAudioData *) event.GetClientData();
-        new_float_buffer = &(demodAudioData->data);
-        if (new_float_buffer != NULL && new_float_buffer->size()) {
-
-            if (scopeCanvas->waveform_points.size() != new_float_buffer->size() * 2) {
-                scopeCanvas->waveform_points.resize(new_float_buffer->size() * 2);
-            }
-
-            for (int i = 0, iMax = new_float_buffer->size(); i < iMax; i++) {
-                scopeCanvas->waveform_points[i * 2 + 1] = (*new_float_buffer)[i] * 0.5f;
-                scopeCanvas->waveform_points[i * 2] = ((double) i / (double) iMax);
-            }
-
-        } else {
-            std::cout << "Incoming Demodulator data empty?" << std::endl;
-        }
-        delete demodAudioData;
-
-        break;
-    default:
     event.Skip();
 }
-}
 
 void AppFrame::OnIdle(wxIdleEvent& event) {
     bool work_done = false;
@@ -152,6 +130,26 @@ void AppFrame::OnIdle(wxIdleEvent& event) {
         work_done = true;
     }
 
+    if (!audioVisualQueue->empty()) {
+        AudioThreadInput demodAudioData;
+        audioVisualQueue->pop(demodAudioData);
+        if (demodAudioData.data.size()) {
+
+            if (scopeCanvas->waveform_points.size() != demodAudioData.data.size() * 2) {
+                scopeCanvas->waveform_points.resize(demodAudioData.data.size() * 2);
+            }
+
+            for (int i = 0, iMax = demodAudioData.data.size(); i < iMax; i++) {
+                scopeCanvas->waveform_points[i * 2 + 1] = demodAudioData.data[i] * 0.5f;
+                scopeCanvas->waveform_points[i * 2] = ((double) i / (double) iMax);
+            }
+
+        } else {
+            std::cout << "Incoming Demodulator data empty?" << std::endl;
+        }
+        work_done = true;
+    }
+
     if (!work_done) {
         event.Skip();
     }

src/AppFrame.h

@@ -45,6 +45,7 @@ private:
     SDRThread *sdrThread;
     SDRThreadCommandQueue* threadCmdQueueSDR;
     SDRThreadIQDataQueue* iqVisualQueue;
+    DemodulatorThreadOutputQueue* audioVisualQueue;
 
     std::thread *t1;
     std::thread *t_SDR;

src/demod/DemodulatorMgr.h

@@ -4,44 +4,36 @@
 #include <map>
 
 #include "DemodulatorThread.h"
-#include "DemodulatorThreadQueue.h"
-#include "DemodulatorThreadTask.h"
 
 class DemodulatorInstance {
 public:
-    DemodulatorThread *t_Demod;
-    DemodulatorThreadQueue* threadQueueDemod;
-    DemodulatorThreadParameters params;
-    wxEvtHandler* parent;
+    DemodulatorThread *demodulatorThread;
+    std::thread *t_Demod;
 
-    DemodulatorInstance(wxEvtHandler* pParent) :
-            t_Demod(NULL), threadQueueDemod(NULL), parent(pParent) {
+    DemodulatorThreadInputQueue* threadQueueDemod;
+    DemodulatorThreadParameters params;
+
+    DemodulatorInstance() :
+            t_Demod(NULL), threadQueueDemod(NULL), demodulatorThread(NULL) {
+    }
+    void setVisualOutputQueue(DemodulatorThreadOutputQueue *tQueue) {
+        demodulatorThread->setVisualOutputQueue(tQueue);
     }
 
     void init() {
-        threadQueueDemod = new DemodulatorThreadQueue(parent);
-        t_Demod = new DemodulatorThread(threadQueueDemod, &params);
-    }
-
-    void addTask(const DemodulatorThreadTask& task, const DemodulatorThreadQueue::DEMOD_PRIORITY& priority) {
-        threadQueueDemod->addTask(task, priority);
-    }
-
-    bool run() {
-        init();
-        if (t_Demod->Run() != wxTHREAD_NO_ERROR) {
-            wxLogError
-            ("Can't create the Demodulator thread!");
-            delete t_Demod;
+        if (threadQueueDemod) {
             delete threadQueueDemod;
-            t_Demod = NULL;
-            threadQueueDemod = NULL;
-            return false;
         }
-        t_Demod->SetPriority(80);
+        if (demodulatorThread) {
+            delete demodulatorThread;
+        }
 
-        return true;
+        threadQueueDemod = new DemodulatorThreadInputQueue;
+        demodulatorThread = new DemodulatorThread(threadQueueDemod, &params);
+
+        t_Demod = new std::thread(&DemodulatorThread::threadMain, demodulatorThread);
     }
+
 };
 
 class DemodulatorMgr {
@@ -59,7 +51,7 @@ public:
     }
 
     DemodulatorInstance *newThread(wxEvtHandler* pParent) {
-        DemodulatorInstance *newDemod = new DemodulatorInstance(pParent);
+        DemodulatorInstance *newDemod = new DemodulatorInstance;
         demods.push_back(newDemod);
         return newDemod;
     }

src/demod/DemodulatorThread.cpp

@@ -2,9 +2,8 @@
 #include "CubicSDRDefs.h"
 #include <vector>
 
-
-DemodulatorThread::DemodulatorThread(DemodulatorThreadQueue* pQueue, DemodulatorThreadParameters *params_in, int id) :
-        wxThread(wxTHREAD_DETACHED), m_pQueue(pQueue), m_ID(id) {
+DemodulatorThread::DemodulatorThread(DemodulatorThreadInputQueue* pQueue, DemodulatorThreadParameters *params_in) :
+        m_pQueue(pQueue), visOutQueue(NULL) {
 
     DemodulatorThreadParameters defaultParams;
     if (!params_in) {
@@ -51,20 +50,16 @@ DemodulatorThread::DemodulatorThread(DemodulatorThreadQueue* pQueue, Demodulator
 }
 
 DemodulatorThread::~DemodulatorThread() {
-
+    std::cout << std::endl << "Demodulator Thread Done." << std::endl << std::endl;
 }
 
-wxThread::ExitCode DemodulatorThread::Entry() {
+void DemodulatorThread::threadMain() {
 
-    while (!TestDestroy()) {
+    while (1) {
+        DemodulatorThreadIQData inp;
+        m_pQueue->pop(inp);
 
-        if (m_pQueue->stackSize()) {
-
-            while (m_pQueue->stackSize()) {
-                DemodulatorThreadTask task = m_pQueue->pop(); // pop a task from the queue. this will block the worker thread if queue is empty
-                switch (task.m_cmd) {
-                case DemodulatorThreadTask::DEMOD_THREAD_DATA:
-                    std::vector<signed char> *data = &task.data->data;
+        std::vector<signed char> *data = &inp.data;
         if (data->size()) {
             liquid_float_complex filtered_input[BUF_SIZE / 2];
 
@@ -134,8 +129,18 @@ wxThread::ExitCode DemodulatorThread::Entry() {
                 newBuffer[i * 2 + 1] = y.real;
             }
 
+            AudioThreadInput ati;
+            ati.data = newBuffer;
 
-                        if (!TestDestroy()) {
+            if (params.audioInputQueue != NULL) {
+                params.audioInputQueue->push(ati);
+            }
+
+            if (visOutQueue != NULL) {
+                visOutQueue->push(ati);
+            }
+
+            /*if (!TestDestroy()) {
              DemodulatorThreadAudioData *audioOut = new DemodulatorThreadAudioData(task.data->frequency, params.audioSampleRate, newBuffer);
 
              m_pQueue->sendAudioData(DemodulatorThreadTask::DEMOD_THREAD_AUDIO_DATA, audioOut);
@@ -148,20 +153,9 @@ wxThread::ExitCode DemodulatorThread::Entry() {
              //                                audio_task.data = new AudioThreadData(task.data->frequency, params.audioSampleRate, newBuffer);
              //                                params.audioQueue->addTask(audio_task, AudioThreadQueue::AUDIO_PRIORITY_HIGHEST);
              }
-                        }
-
-                        delete task.data;
-                    }
-                    break;
-                }
-            }
-        } else {
-            this->Yield();
-            this->Sleep(1);
-        }
-    }
-    std::cout << std::endl << "Demodulator Thread Done." << std::endl << std::endl;
-
-    return (wxThread::ExitCode) 0;
+             }*/
+
+        }
+    }
 }
 

src/demod/DemodulatorThread.h

@@ -10,20 +10,100 @@
 
 #include "wx/thread.h"
 
-#include "DemodulatorThreadQueue.h"
 #include "liquid/liquid.h"
 #include "AudioThread.h"
+#include "ThreadQueue.h"
+#include "CubicSDRDefs.h"
 
-class DemodulatorThread: public wxThread {
+enum DemodulatorType {
+    DEMOD_TYPE_NULL, DEMOD_TYPE_AM, DEMOD_TYPE_FM, DEMOD_TYPE_LSB, DEMOD_TYPE_USB, DEMOD_TYPE_WFM
+};
+
+class DemodulatorThreadIQData {
+public:
+    unsigned int frequency;
+    unsigned int bandwidth;
+    std::vector<signed char> data;
+
+    DemodulatorThreadIQData() :
+            frequency(0), bandwidth(0) {
+
+    }
+
+    DemodulatorThreadIQData(unsigned int bandwidth, unsigned int frequency, std::vector<signed char> data) :
+            data(data), frequency(frequency), bandwidth(bandwidth) {
+
+    }
+
+    ~DemodulatorThreadIQData() {
+
+    }
+};
+
+class DemodulatorThreadAudioData {
+public:
+    unsigned int frequency;
+    unsigned int sampleRate;
+    unsigned char channels;
+
+    std::vector<float> data;
+
+    DemodulatorThreadAudioData() :
+            sampleRate(0), frequency(0), channels(0) {
+
+    }
+
+    DemodulatorThreadAudioData(unsigned int frequency, unsigned int sampleRate, std::vector<float> data) :
+            data(data), sampleRate(sampleRate), frequency(frequency), channels(1) {
+
+    }
+
+    ~DemodulatorThreadAudioData() {
+
+    }
+};
+
+class DemodulatorThreadParameters {
+public:
+    unsigned int inputRate;
+    unsigned int inputResampleRate; // set equal to disable second stage re-sampling?
+    unsigned int demodResampleRate;
+    unsigned int filterFrequency;
+    unsigned int audioSampleRate;
+    AudioThreadInputQueue *audioInputQueue;
+
+    DemodulatorType demodType;
+
+    DemodulatorThreadParameters() :
+            audioInputQueue(NULL), inputRate(SRATE), inputResampleRate(200000), demodResampleRate(100000), audioSampleRate(48000), filterFrequency(
+                    32000), demodType(DEMOD_TYPE_WFM) {
+
+    }
+
+    ~DemodulatorThreadParameters() {
+
+    }
+};
+
+typedef ThreadQueue<DemodulatorThreadIQData> DemodulatorThreadInputQueue;
+typedef ThreadQueue<AudioThreadInput> DemodulatorThreadOutputQueue;
+
+class DemodulatorThread {
 public:
 
-    DemodulatorThread(DemodulatorThreadQueue* pQueue, DemodulatorThreadParameters *params, int id = 0);
+    DemodulatorThread(DemodulatorThreadInputQueue* pQueue, DemodulatorThreadParameters *params);
     ~DemodulatorThread();
 
+    void threadMain();
+
+    void setVisualOutputQueue(DemodulatorThreadOutputQueue *tQueue) {
+        visOutQueue = tQueue;
+        visOutQueue->set_max_num_items(1);
+    }
+
 protected:
-    virtual ExitCode Entry();
-    DemodulatorThreadQueue* m_pQueue;
-    int m_ID;
+    DemodulatorThreadInputQueue* m_pQueue;
+    DemodulatorThreadOutputQueue* visOutQueue;
 
     firfilt_crcf fir_filter;
     firfilt_crcf fir_audio_filter;

src/demod/DemodulatorThreadQueue.cpp

@@ -1,44 +0,0 @@
-#include "DemodulatorThreadQueue.h"
-
-#include "wx/wxprec.h"
-
-#ifndef WX_PRECOMP
-#include "wx/wx.h"
-#endif
-
-DemodulatorThreadQueue::DemodulatorThreadQueue(wxEvtHandler* pParent) :
-        m_pParent(pParent) {
-}
-
-void DemodulatorThreadQueue::addTask(const DemodulatorThreadTask& task, const DEMOD_PRIORITY& priority) {
-    wxMutexLocker lock(m_MutexQueue);
-    m_Tasks.insert(std::make_pair(priority, task));
-    m_QueueCount.Post();
-}
-
-DemodulatorThreadTask DemodulatorThreadQueue::pop() {
-    DemodulatorThreadTask element;
-    m_QueueCount.Wait();
-    m_MutexQueue.Lock();
-    element = (m_Tasks.begin())->second;
-    m_Tasks.erase(m_Tasks.begin());
-    m_MutexQueue.Unlock();
-    return element;
-}
-
-void DemodulatorThreadQueue::sendAudioData(const DemodulatorThreadTask::DEMOD_THREAD_COMMAND& cmd, DemodulatorThreadAudioData *data) {
-    wxCommandEvent evt(wxEVT_THREAD, cmd);
-
-    evt.SetClientData(data);
-
-    m_pParent->AddPendingEvent(evt);
-}
-
-size_t DemodulatorThreadQueue::stackSize() {
-    wxMutexLocker lock(m_MutexQueue);
-    return m_Tasks.size();
-}
-
-wxEvtHandler* DemodulatorThreadQueue::getHandler() {
-    return m_pParent;
-}

src/demod/DemodulatorThreadQueue.h

@@ -1,28 +0,0 @@
-#pragma once
-
-#include <map>
-#include "DemodulatorThreadTask.h"
-
-#include "wx/event.h"
-
-class DemodulatorThreadQueue {
-public:
-    enum DEMOD_PRIORITY {
-        DEMOD_PRIORITY_HIGHEST, DEMOD_PRIORITY_HIGHER, DEMOD_PRIORITY_NORMAL, DEMOD_PRIORITY_BELOW_NORMAL, DEMOD_PRIORITY_LOW, DEMOD_PRIORITY_IDLE
-    };
-    DemodulatorThreadQueue(wxEvtHandler* pParent);
-
-    void addTask(const DemodulatorThreadTask& task, const DEMOD_PRIORITY& priority = DEMOD_PRIORITY_NORMAL);
-    void sendAudioData(const DemodulatorThreadTask::DEMOD_THREAD_COMMAND& cmd, DemodulatorThreadAudioData *data);
-
-    DemodulatorThreadTask pop();
-    size_t stackSize();
-
-    wxEvtHandler* getHandler();
-
-private:
-    wxEvtHandler* m_pParent;
-    std::multimap<DEMOD_PRIORITY, DemodulatorThreadTask> m_Tasks;
-    wxMutex m_MutexQueue;
-    wxSemaphore m_QueueCount;
-};

src/demod/DemodulatorThreadTask.cpp

@@ -1,2 +0,0 @@
-#include "DemodulatorThreadTask.h"
-

src/demod/DemodulatorThreadTask.h

@@ -1,91 +0,0 @@
-#pragma once
-
-#include <vector>
-#include "wx/defs.h"
-#include "wx/string.h"
-#include "wx/object.h"
-#include "CubicSDRDefs.h"
-#include "AudioThread.h"
-
-enum DemodulatorType {
-    DEMOD_TYPE_NULL, DEMOD_TYPE_AM, DEMOD_TYPE_FM, DEMOD_TYPE_LSB, DEMOD_TYPE_USB, DEMOD_TYPE_WFM
-};
-
-class DemodulatorThreadIQData: public wxObject {
-public:
-    unsigned int frequency;
-    unsigned int bandwidth;
-    std::vector<signed char> data;
-
-    DemodulatorThreadIQData(unsigned int bandwidth, unsigned int frequency, std::vector<signed char> data) :
-            data(data), frequency(frequency), bandwidth(bandwidth) {
-
-    }
-
-    ~DemodulatorThreadIQData() {
-
-    }
-};
-
-class DemodulatorThreadAudioData: public wxObject {
-public:
-    unsigned int frequency;
-    unsigned int sampleRate;
-    unsigned char channels;
-
-    std::vector<float> data;
-
-    DemodulatorThreadAudioData(unsigned int frequency, unsigned int sampleRate, std::vector<float> data) :
-            data(data), sampleRate(sampleRate), frequency(frequency), channels(1) {
-
-    }
-
-    ~DemodulatorThreadAudioData() {
-
-    }
-};
-
-class DemodulatorThreadParameters: public wxObject {
-public:
-    unsigned int inputRate;
-    unsigned int inputResampleRate; // set equal to disable second stage re-sampling?
-    unsigned int demodResampleRate;
-    unsigned int filterFrequency;
-    unsigned int audioSampleRate;
-    AudioThreadInputQueue *audioInputQueue;
-
-    DemodulatorType demodType;
-
-    DemodulatorThreadParameters() :
-        audioInputQueue(NULL), inputRate(SRATE), inputResampleRate(200000), demodResampleRate(100000), audioSampleRate(48000), filterFrequency(32000), demodType(DEMOD_TYPE_WFM) {
-
-    }
-
-    ~DemodulatorThreadParameters() {
-
-    }
-};
-
-class DemodulatorThreadTask {
-public:
-    enum DEMOD_THREAD_COMMAND {
-        DEMOD_THREAD_EXIT = wxID_EXIT,
-        DEMOD_THREAD_NULL = wxID_HIGHEST + 100,
-        DEMOD_THREAD_STARTED,
-        DEMOD_THREAD_PROCESS,
-        DEMOD_THREAD_ERROR,
-        DEMOD_THREAD_DATA,
-        DEMOD_THREAD_AUDIO_DATA
-    };
-
-    DemodulatorThreadTask() :
-            m_cmd(DEMOD_THREAD_NULL), data(NULL) {
-    }
-    DemodulatorThreadTask(DEMOD_THREAD_COMMAND cmd) :
-            m_cmd(cmd), data(NULL) {
-    }
-
-    DEMOD_THREAD_COMMAND m_cmd;
-
-    DemodulatorThreadIQData *data;
-};

src/sdr/SDRThread.cpp

@@ -170,11 +170,15 @@ void SDRThread::threadMain() {
         }
 
         if (demodulators.size()) {
+            DemodulatorThreadIQData demodDataOut;
+            demodDataOut.frequency = frequency;
+            demodDataOut.bandwidth = bandwidth;
+            demodDataOut.data = new_buffer;
+
             for (int i = 0, iMax = demodulators.size(); i < iMax; i++) {
-                DemodulatorThreadQueue *demodQueue = demodulators[i];
-                DemodulatorThreadTask demod_task = DemodulatorThreadTask(DemodulatorThreadTask::DEMOD_THREAD_DATA);
-                demod_task.data = new DemodulatorThreadIQData(bandwidth, frequency, new_buffer);
-                demodQueue->addTask(demod_task, DemodulatorThreadQueue::DEMOD_PRIORITY_HIGHEST);
+                DemodulatorInstance *demod = demodulators[i];
+                DemodulatorThreadInputQueue *demodQueue = demod->threadQueueDemod;
+                demodQueue->push(demodDataOut);
             }
         }
 

src/sdr/SDRThread.h

@@ -9,9 +9,8 @@
 
 #include "wx/thread.h"
 
-#include "DemodulatorThreadQueue.h"
-#include "DemodulatorMgr.h"
 #include "ThreadQueue.h"
+#include "DemodulatorMgr.h"
 
 class SDRThreadCommand {
 public:
@@ -65,8 +64,8 @@ public:
 
     int enumerate_rtl();
 
-    void bindDemodulator(DemodulatorInstance &demod) {
-        demodulators.push_back(demod.threadQueueDemod);
+    void bindDemodulator(DemodulatorInstance *demod) {
+        demodulators.push_back(demod);
     }
 
     void threadMain();
@@ -85,5 +84,5 @@ protected:
     SDRThreadIQDataQueue* iqDataOutQueue;
     SDRThreadIQDataQueue* iqVisualQueue;
 
-    std::vector<DemodulatorThreadQueue *> demodulators;
+    std::vector<DemodulatorInstance *> demodulators;
 };