Merge pull request #1 from cjcliffe/experimental-threading

Thread task queue and basic frequency tuning support for SDRThread

CMakeLists.txt

@@ -78,6 +78,8 @@ SET (cubicsdr_sources
 	src/IQBufferThread.cpp
 	src/PrimaryGLContext.cpp
 	src/AppFrame.cpp
+	src/SDRThreadQueue.cpp
+	src/SDRThreadTask.cpp
 )
 
 SET (cubicsdr_headers
@@ -87,6 +89,8 @@ SET (cubicsdr_headers
 	src/PrimaryGLContext.h
 	src/AppFrame.h
 	src/CubicSDRDefs.h
+	src/SDRThreadQueue.h
+	src/SDRThreadTask.h
 )
 #configure_files(${PROJECT_SOURCE_DIR}/shaders ${PROJECT_BINARY_DIR}/shaders COPYONLY)
 #configure_files(${PROJECT_SOURCE_DIR}/png ${PROJECT_BINARY_DIR}/png COPYONLY)

src/AppFrame.cpp

@@ -21,7 +21,7 @@ EVT_IDLE(AppFrame::OnIdle)
 wxEND_EVENT_TABLE()
 
 AppFrame::AppFrame() :
-        wxFrame(NULL, wxID_ANY, wxT("CubicSDR")) {
+        wxFrame(NULL, wxID_ANY, wxT("CubicSDR")), frequency(DEFAULT_FREQ) {
 
     canvas = new TestGLCanvas(this, NULL);
 
@@ -42,12 +42,58 @@ AppFrame::AppFrame() :
     Centre();
     Show();
 
+    m_pQueue = new SDRThreadQueue(this);
+
+    t_SDR = new SDRThread(m_pQueue);
+    if (t_SDR->Run() != wxTHREAD_NO_ERROR) {
+        wxLogError
+        ("Can't create the thread!");
+        delete t_SDR;
+        t_SDR = NULL;
+    }
+
+//    t_IQBuffer = new IQBufferThread(this);
+//    if (t_IQBuffer->Run() != wxTHREAD_NO_ERROR) {
+//        wxLogError
+//        ("Can't create the thread!");
+//        delete t_IQBuffer;
+    t_IQBuffer = NULL;
+//    }
+
 //    static const int attribs[] = { WX_GL_RGBA, WX_GL_DOUBLEBUFFER, 0 };
 //    wxLogStatus("Double-buffered display %s supported", wxGLCanvas::IsDisplaySupported(attribs) ? "is" : "not");
 //    ShowFullScreen(true);
 }
 
+AppFrame::~AppFrame() {
+    delete t_SDR;
+//    delete t_IQBuffer;
+    delete m_pQueue;
+}
+
 void AppFrame::OnClose(wxCommandEvent& WXUNUSED(event)) {
+
+    {
+        wxCriticalSectionLocker enter(m_pThreadCS);
+        if (t_SDR) {
+            wxMessageOutputDebug().Printf("CubicSDR: deleting thread");
+            if (t_SDR->Delete() != wxTHREAD_NO_ERROR) {
+                wxLogError
+                ("Can't delete the thread!");
+            }
+        }
+    }
+
+    {
+        wxCriticalSectionLocker enter(m_pThreadCS);
+        if (t_IQBuffer) {
+            wxMessageOutputDebug().Printf("CubicSDR: deleting thread");
+            if (t_IQBuffer->Delete() != wxTHREAD_NO_ERROR) {
+                wxLogError
+                ("Can't delete the thread!");
+            }
+        }
+    }
     // true is to force the frame to close
     Close(true);
 }
@@ -70,3 +116,14 @@ void AppFrame::OnIdle(wxIdleEvent& event) {
 
     event.Skip();
 }
+
+void AppFrame::setFrequency(unsigned int freq) {
+    frequency = freq;
+    SDRThreadTask task = SDRThreadTask(SDRThreadTask::SDR_THREAD_TUNING);
+    task.setUInt(freq);
+    m_pQueue->addTask(task, SDRThreadQueue::SDR_PRIORITY_HIGHEST);
+}
+
+int AppFrame::getFrequency() {
+    return frequency;
+}

src/AppFrame.h

@@ -2,19 +2,29 @@
 
 #include "wx/frame.h"
 #include "PrimaryGLContext.h"
+#include "SDRThread.h"
 
 // Define a new frame type
 class AppFrame: public wxFrame {
 public:
     AppFrame();
+    ~AppFrame();
     void OnEventInput(wxThreadEvent& event);
 
+    void setFrequency(unsigned int freq);
+    int getFrequency();
+
 private:
     void OnClose(wxCommandEvent& event);
     void OnNewWindow(wxCommandEvent& event);
     void OnIdle(wxIdleEvent& event);
 
     TestGLCanvas *canvas;
+    SDRThread *t_SDR;
+    IQBufferThread *t_IQBuffer;
+    wxCriticalSection m_pThreadCS;
+    SDRThreadQueue* m_pQueue;
+    unsigned int frequency;
 
 wxDECLARE_EVENT_TABLE();
 };

src/CubicSDR.cpp

@@ -21,51 +21,12 @@ bool CubicSDR::OnInit() {
 
     AppFrame *appframe = new AppFrame();
 
-    t_SDR = new SDRThread(appframe);
-    if (t_SDR->Run() != wxTHREAD_NO_ERROR) {
-        wxLogError
-        ("Can't create the thread!");
-        delete t_SDR;
-        t_SDR = NULL;
-    }
-
-    t_IQBuffer = new IQBufferThread(this);
-    if (t_IQBuffer->Run() != wxTHREAD_NO_ERROR) {
-        wxLogError
-        ("Can't create the thread!");
-        delete t_IQBuffer;
-        t_IQBuffer = NULL;
-    }
-
     return true;
 }
 
 int CubicSDR::OnExit() {
     delete m_glContext;
 
-    {
-        wxCriticalSectionLocker enter(m_pThreadCS);
-        if (t_SDR) {
-            wxMessageOutputDebug().Printf("CubicSDR: deleting thread");
-            if (t_SDR->Delete() != wxTHREAD_NO_ERROR) {
-                wxLogError
-                ("Can't delete the thread!");
-            }
-        }
-    }
-
-    {
-        wxCriticalSectionLocker enter(m_pThreadCS);
-        if (t_IQBuffer) {
-            wxMessageOutputDebug().Printf("CubicSDR: deleting thread");
-            if (t_IQBuffer->Delete() != wxTHREAD_NO_ERROR) {
-                wxLogError
-                ("Can't delete the thread!");
-            }
-        }
-    }
-    wxThread::This()->Sleep(1);
-
 //	while (1) {
 //		{ wxCriticalSectionLocker enter(m_pThreadCS);
 //			if (!m_pThread)

src/CubicSDR.h

@@ -4,8 +4,6 @@
 //WX_GL_MAJOR_VERSION 3
 //WX_GL_MINOR_VERSION 2
 
-#include "SDRThread.h"
-#include "IQBufferThread.h"
 #include "wx/glcanvas.h"
 #include "PrimaryGLContext.h"
 
@@ -13,7 +11,6 @@ class CubicSDR: public wxApp {
 public:
     CubicSDR() {
         m_glContext = NULL;
-        t_SDR = NULL;
     }
 
     PrimaryGLContext &GetContext(wxGLCanvas *canvas);
@@ -21,15 +18,9 @@ public:
     virtual bool OnInit();
     virtual int OnExit();
 
-    void OnEventInput(wxEvent& event) {
-        std::cout << "event !" << std::endl;
-    }
-
 private:
     PrimaryGLContext *m_glContext;
-    SDRThread *t_SDR;
+
-    IQBufferThread *t_IQBuffer;
-    wxCriticalSection m_pThreadCS;
 };
 
 DECLARE_APP(CubicSDR)

src/CubicSDRDefs.h

@@ -4,3 +4,5 @@
 #define SRATE 2500000
 #define FFT_SIZE 8192
 
+#define DEFAULT_FREQ 105700000
+

src/PrimaryGLContext.cpp

@@ -12,6 +12,7 @@
 
 #include "CubicSDR.h"
 #include "CubicSDRDefs.h"
+#include "AppFrame.h"
 #include <algorithm>
 
 wxString glGetwxString(GLenum name) {
@@ -67,6 +68,8 @@ void PrimaryGLContext::Plot(std::vector<float> &points) {
     glMatrixMode(GL_MODELVIEW);
     glLoadIdentity();
 
+//    glEnable(GL_LINE_SMOOTH);
+
     glPushMatrix();
     glTranslatef(-1.0f, -0.9f, 0.0f);
     glScalef(2.0f, 1.8f, 1.0f);
@@ -96,7 +99,7 @@ wxEND_EVENT_TABLE()
 
 TestGLCanvas::TestGLCanvas(wxWindow *parent, int *attribList) :
         wxGLCanvas(parent, wxID_ANY, attribList, wxDefaultPosition, wxDefaultSize,
-        wxFULL_REPAINT_ON_RESIZE) {
+        wxFULL_REPAINT_ON_RESIZE), parent(parent) {
 
     int in_block_size = BUF_SIZE / 2;
     int out_block_size = FFT_SIZE;
@@ -106,6 +109,8 @@ TestGLCanvas::TestGLCanvas(wxWindow *parent, int *attribList) :
     out[1] = (fftw_complex*) fftw_malloc(sizeof(fftw_complex) * out_block_size);
     plan[0] = fftw_plan_dft_1d(out_block_size, in, out[0], FFTW_BACKWARD, FFTW_MEASURE);
     plan[1] = fftw_plan_dft_1d(out_block_size, in, out[1], FFTW_FORWARD, FFTW_MEASURE);
+
+    fft_ceil_ma = fft_ceil_maa = 1.0;
 }
 
 void TestGLCanvas::OnPaint(wxPaintEvent& WXUNUSED(event)) {
@@ -123,10 +128,17 @@ void TestGLCanvas::OnPaint(wxPaintEvent& WXUNUSED(event)) {
 void TestGLCanvas::OnKeyDown(wxKeyEvent& event) {
     float angle = 5.0;
 
+    unsigned int freq;
     switch (event.GetKeyCode()) {
     case WXK_RIGHT:
+        freq = ((AppFrame*) parent)->getFrequency();
+        freq += 100000;
+        ((AppFrame*) parent)->setFrequency(freq);
         break;
     case WXK_LEFT:
+        freq = ((AppFrame*) parent)->getFrequency();
+        freq -= 100000;
+        ((AppFrame*) parent)->setFrequency(freq);
         break;
     case WXK_DOWN:
         break;
@@ -156,44 +168,51 @@ void TestGLCanvas::setData(std::vector<signed char> *data) {
         fftw_execute(plan[0]);
         fftw_execute(plan[1]);
 
-        double result[FFT_SIZE];
+        double fft_ceil = 0;
-        double fft_floor, fft_ceil;
+        // fft_floor, 
+
+        if (fft_result.size() < FFT_SIZE) {
+            fft_result.resize(FFT_SIZE);
+            fft_result_ma.resize(FFT_SIZE);
+            fft_result_maa.resize(FFT_SIZE);
+        }
 
         for (int j = 0; j < 2; j++) {
             for (int i = 0, iMax = FFT_SIZE / 2; i < iMax; i++) {
-                double a = out[j][j?i:((iMax-1)-i)][0];
+                double a = out[j][j ? i : ((iMax - 1) - i)][0];
-                double b = out[j][j?i:((iMax-1)-i)][1];
+                double b = out[j][j ? i : ((iMax - 1) - i)][1];
                 double c = sqrt(a * a + b * b);
 
-                double x = out[j?0:1][j?((FFT_SIZE-1)-i):((FFT_SIZE/2)+i)][0];
+                double x = out[j ? 0 : 1][j ? ((FFT_SIZE - 1) - i) : ((FFT_SIZE / 2) + i)][0];
-                double y = out[j?0:1][j?((FFT_SIZE-1)-i):((FFT_SIZE/2)+i)][1];
+                double y = out[j ? 0 : 1][j ? ((FFT_SIZE - 1) - i) : ((FFT_SIZE / 2) + i)][1];
                 double z = sqrt(x * x + y * y);
 
-                if (i == 1) {
+                double r = (c < z) ? c : z;
-                    fft_floor = fft_ceil = c;
-                } else if (i < FFT_SIZE - 1) {
-                    if (c < fft_floor) {
-                        fft_floor = c;
-                    }
-                    if (c > fft_ceil) {
-                        fft_ceil = c;
-                    }
-                }
 
                 if (!j) {
-                    result[i] = (c<z)?c:z;
+                    fft_result[i] = r;
                 } else {
-                    result[(FFT_SIZE/2) + i] = (c<z)?c:z;
+                    fft_result[(FFT_SIZE / 2) + i] = r;
                 }
             }
         }
 
-        if (fft_ceil - fft_floor < 10.0) {
+        float time_slice = (float) SRATE / (float) (BUF_SIZE / 2);
-            fft_ceil = fft_floor + 10.0;
-        }
 
         for (int i = 0, iMax = FFT_SIZE; i < iMax; i++) {
-            points[i * 2 + 1] = (result[i] - fft_floor) / (fft_ceil - fft_floor);
+            fft_result_maa[i] += (fft_result_ma[i] - fft_result_maa[i]) * 0.65;
+            fft_result_ma[i] += (fft_result[i] - fft_result_ma[i]) * 0.65;
+
+            if (fft_result_maa[i] > fft_ceil) {
+                fft_ceil = fft_result_maa[i];
+            }
+        }
+
+        fft_ceil_ma = fft_ceil_ma + (fft_ceil - fft_ceil_ma) * 0.05;
+        fft_ceil_maa = fft_ceil_maa + (fft_ceil - fft_ceil_maa) * 0.05;
+
+        for (int i = 0, iMax = FFT_SIZE; i < iMax; i++) {
+            points[i * 2 + 1] = fft_result_maa[i] / fft_ceil_maa;
             points[i * 2] = ((double) i / (double) iMax);
         }
     }

src/PrimaryGLContext.h

@@ -7,8 +7,7 @@
 #include "CubicSDRDefs.h"
 #include "fftw3.h"
 
-class PrimaryGLContext : public wxGLContext
+class PrimaryGLContext: public wxGLContext {
-{
 public:
     PrimaryGLContext(wxGLCanvas *canvas);
 
@@ -19,23 +18,29 @@ private:
     GLuint m_textures[6];
 };
 
-class TestGLCanvas : public wxGLCanvas
+class TestGLCanvas: public wxGLCanvas {
-{
 public:
     TestGLCanvas(wxWindow *parent, int *attribList = NULL);
 
     void setData(std::vector<signed char> *data);
 
-    std::vector<float> points;
-
-    fftw_complex *in, *out[2];
-    fftw_plan plan[2];
-
 private:
     void OnPaint(wxPaintEvent& event);
     void OnKeyDown(wxKeyEvent& event);
 
     void OnIdle(wxIdleEvent &event);
 
-    wxDECLARE_EVENT_TABLE();
+    wxWindow *parent;
+    std::vector<float> points;
+
+    fftw_complex *in, *out[2];
+    fftw_plan plan[2];
+
+    float fft_ceil_ma, fft_ceil_maa;
+
+    std::vector<float> fft_result;
+    std::vector<float> fft_result_ma;
+    std::vector<float> fft_result_maa;
+
+wxDECLARE_EVENT_TABLE();
 };

src/SDRThread.cpp

@@ -2,13 +2,12 @@
 #include "CubicSDRDefs.h"
 #include <vector>
 
-
 //wxDEFINE_EVENT(wxEVT_COMMAND_SDRThread_INPUT, wxThreadEvent);
 
-SDRThread::SDRThread(AppFrame *frame) :
+SDRThread::SDRThread(SDRThreadQueue* pQueue, int id) :
-        wxThread(wxTHREAD_DETACHED) {
+        wxThread(wxTHREAD_DETACHED), m_pQueue(pQueue), m_ID(id) {
     dev = NULL;
-    this->frame = frame;
+    sample_rate = SRATE;
 }
 SDRThread::~SDRThread() {
 
@@ -85,11 +84,10 @@ void SDRThread::enumerate_rtl() {
 wxThread::ExitCode SDRThread::Entry() {
     signed char *buf = (signed char *) malloc(BUF_SIZE);
 
-
     int use_my_dev = 1;
     int dev_count = rtlsdr_get_device_count();
 
-    if (use_my_dev > dev_count-1) {
+    if (use_my_dev > dev_count - 1) {
         use_my_dev = 0;
     }
 
@@ -97,7 +95,7 @@ wxThread::ExitCode SDRThread::Entry() {
 
     rtlsdr_open(&dev, use_my_dev);
     rtlsdr_set_sample_rate(dev, SRATE);
-    rtlsdr_set_center_freq(dev, 105700000);
+    rtlsdr_set_center_freq(dev, DEFAULT_FREQ);
     rtlsdr_set_agc_mode(dev, 1);
     rtlsdr_set_offset_tuning(dev, 1);
     rtlsdr_reset_buffer(dev);
@@ -112,12 +110,27 @@ wxThread::ExitCode SDRThread::Entry() {
     std::cout << "Sampling..";
     while (!TestDestroy()) {
 
+        if (m_pQueue->stackSize()) {
+            bool freq_changed = false;
+            float new_freq;
+
+            while (m_pQueue->stackSize()) {
+                SDRThreadTask 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 SDRThreadTask::SDR_THREAD_TUNING:
+                    std::cout << "Set frequency: " << task.getUInt() << std::endl;
+                    freq_changed = true;
+                    new_freq = task.getUInt();
+                    break;
+                }
+            }
+
+            if (freq_changed) {
+                rtlsdr_set_center_freq(dev, new_freq);
+            }
+        }
+
         rtlsdr_read_sync(dev, buf, BUF_SIZE, &n_read);
-        // move around
-        long freq = 98000000+(20000000)*sin(seconds/50.0);
-        rtlsdr_set_center_freq(dev, freq);
-        
-        std::cout << "Frequency: " << freq << std::endl;
 
         if (!TestDestroy()) {
             std::vector<signed char> *new_buffer = new std::vector<signed char>();
@@ -126,14 +139,14 @@ wxThread::ExitCode SDRThread::Entry() {
                 new_buffer->push_back(buf[i] - 127);
             }
 
-            double time_slice = (double)n_read/(double)sample_rate;
+            double time_slice = (double) n_read / (double) sample_rate;
             seconds += time_slice;
 
             // std::cout << "Time Slice: " << time_slice << std::endl;
             if (!TestDestroy()) {
                 wxThreadEvent event(wxEVT_THREAD, EVENT_SDR_INPUT);
                 event.SetPayload(new_buffer);
-                wxQueueEvent(frame, event.Clone());
+                wxQueueEvent(m_pQueue->getHandler(), event.Clone());
             } else {
                 delete new_buffer;
             }

src/SDRThread.h

@@ -9,7 +9,9 @@
 
 #include "wx/thread.h"
 
-#include "AppFrame.h"
+#include "SDRThread.h"
+#include "IQBufferThread.h"
+#include "SDRThreadQueue.h"
 
 // declare a new type of event, to be used by our SDRThread class:
 //wxDECLARE_EVENT(wxEVT_COMMAND_SDRThread_COMPLETED, wxThreadEvent);
@@ -17,20 +19,21 @@
 //wxDECLARE_EVENT(wxEVT_COMMAND_SDRThread_INPUT, wxThreadEvent);
 
 enum {
-    EVENT_SDR_INPUT = wxID_HIGHEST+1
+    EVENT_SDR_INPUT = wxID_HIGHEST + 1
 };
 
 class SDRThread: public wxThread {
 public:
-	rtlsdr_dev_t *dev;
+    rtlsdr_dev_t *dev;
 
-	SDRThread(AppFrame *appframe);
+    SDRThread(SDRThreadQueue* pQueue, int id = 0);
-	~SDRThread();
+    ~SDRThread();
 
-	void enumerate_rtl();
+    void enumerate_rtl();
 
 protected:
-	virtual ExitCode Entry();
+    virtual ExitCode Entry();
-	AppFrame *frame;
+    uint32_t sample_rate;
-	uint32_t sample_rate;
+    SDRThreadQueue* m_pQueue;
+    int m_ID;
 };

src/SDRThreadQueue.cpp

@@ -0,0 +1,43 @@
+#include "SDRThreadQueue.h"
+
+#include "wx/wxprec.h"
+
+#ifndef WX_PRECOMP
+#include "wx/wx.h"
+#endif
+
+SDRThreadQueue::SDRThreadQueue(wxEvtHandler* pParent) :
+        m_pParent(pParent) {
+}
+
+void SDRThreadQueue::addTask(const SDRThreadTask& task, const SDR_PRIORITY& priority) {
+    wxMutexLocker lock(m_MutexQueue);
+    m_Tasks.insert(std::make_pair(priority, task));
+    m_QueueCount.Post();
+}
+
+SDRThreadTask SDRThreadQueue::pop() {
+    SDRThreadTask element;
+    m_QueueCount.Wait();
+    m_MutexQueue.Lock();
+    element = (m_Tasks.begin())->second;
+    m_Tasks.erase(m_Tasks.begin());
+    m_MutexQueue.Unlock();
+    return element;
+}
+
+void SDRThreadQueue::report(const SDRThreadTask::SDR_COMMAND& cmd, const wxString& sArg, int iArg) {
+    wxCommandEvent evt(wxEVT_THREAD, cmd);
+    evt.SetString(sArg);
+    evt.SetInt(iArg);
+    m_pParent->AddPendingEvent(evt);
+}
+
+size_t SDRThreadQueue::stackSize() {
+    wxMutexLocker lock(m_MutexQueue);
+    return m_Tasks.size();
+}
+
+wxEvtHandler* SDRThreadQueue::getHandler() {
+    return m_pParent;
+}

src/SDRThreadQueue.h

@@ -0,0 +1,28 @@
+#pragma once
+
+#include <map>
+#include "SDRThreadTask.h"
+
+#include "wx/event.h"
+
+class SDRThreadQueue {
+public:
+    enum SDR_PRIORITY {
+        SDR_PRIORITY_HIGHEST, SDR_PRIORITY_HIGHER, SDR_PRIORITY_NORMAL, SDR_PRIORITY_BELOW_NORMAL, SDR_PRIORITY_LOW, SDR_PRIORITY_IDLE
+    };
+    SDRThreadQueue(wxEvtHandler* pParent);
+
+    void addTask(const SDRThreadTask& task, const SDR_PRIORITY& priority = SDR_PRIORITY_NORMAL);
+    void report(const SDRThreadTask::SDR_COMMAND& cmd, const wxString& sArg = wxEmptyString, int iArg = 0);
+
+    SDRThreadTask pop();
+    size_t stackSize();
+
+    wxEvtHandler* getHandler();
+
+private:
+    wxEvtHandler* m_pParent;
+    std::multimap<SDR_PRIORITY, SDRThreadTask> m_Tasks;
+    wxMutex m_MutexQueue;
+    wxSemaphore m_QueueCount;
+};

src/SDRThreadTask.cpp

@@ -0,0 +1,9 @@
+#include "SDRThreadTask.h"
+
+void SDRThreadTask::setUInt(unsigned int i) {
+    arg_int = i;
+}
+
+unsigned int SDRThreadTask::getUInt() {
+    return arg_int;
+}

src/SDRThreadTask.h

@@ -0,0 +1,24 @@
+#pragma once
+
+#include "wx/defs.h"
+#include "wx/string.h"
+
+class SDRThreadTask {
+public:
+    enum SDR_COMMAND {
+        SDR_THREAD_EXIT = wxID_EXIT, SDR_THREAD_NULL = wxID_HIGHEST + 1, SDR_THREAD_STARTED, SDR_THREAD_PROCESS, SDR_THREAD_ERROR, SDR_THREAD_TUNING
+    };
+
+    SDRThreadTask() :
+            m_cmd(SDR_THREAD_NULL) {
+    }
+    SDRThreadTask(SDR_COMMAND cmd) :
+            m_cmd(cmd) {
+    }
+
+    void setUInt(unsigned int i);
+    unsigned int getUInt();
+
+    SDR_COMMAND m_cmd;
+    unsigned int arg_int;
+};