Starting work on threading model, incomplete

Includes experimental averaging of spectrum

src/AppFrame.cpp

@@ -42,12 +42,51 @@ AppFrame::AppFrame() :
     Centre();
     Show();
 
+
+    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;
+    }
+
 //    static const int attribs[] = { WX_GL_RGBA, WX_GL_DOUBLEBUFFER, 0 };
 //    wxLogStatus("Double-buffered display %s supported", wxGLCanvas::IsDisplaySupported(attribs) ? "is" : "not");
 //    ShowFullScreen(true);
 }
 
 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);
 }

src/AppFrame.h

@@ -15,6 +15,9 @@ private:
     void OnIdle(wxIdleEvent& event);
 
     TestGLCanvas *canvas;
+    SDRThread *t_SDR;
+    IQBufferThread *t_IQBuffer;
+    wxCriticalSection m_pThreadCS;
     
 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

@@ -21,15 +21,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/PrimaryGLContext.cpp

@@ -156,8 +156,14 @@ 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++) {
@@ -169,31 +175,29 @@ void TestGLCanvas::setData(std::vector<signed char> *data) {
                 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++) {
+          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];
+          }
         }
 
         for (int i = 0, iMax = FFT_SIZE; i < iMax; i++) {
-            points[i * 2 + 1] = (result[i] - fft_floor) / (fft_ceil - fft_floor);
+            points[i * 2 + 1] = fft_result_maa[i] / fft_ceil;
             points[i * 2] = ((double) i / (double) iMax);
         }
     }

src/PrimaryGLContext.h

@@ -31,6 +31,11 @@ public:
     fftw_complex *in, *out[2];
     fftw_plan plan[2];
 
+    std::vector<float> fft_result;
+    std::vector<float> fft_result_ma;
+    std::vector<float> fft_result_maa;
+    
+
 private:
     void OnPaint(wxPaintEvent& event);
     void OnKeyDown(wxKeyEvent& event);

src/SDRThread.cpp

@@ -5,10 +5,9 @@
 
 //wxDEFINE_EVENT(wxEVT_COMMAND_SDRThread_INPUT, wxThreadEvent);
 
-SDRThread::SDRThread(AppFrame *frame) :
+SDRThread::SDRThread(SDRThreadQueue* pQueue, int id=0) :
-        wxThread(wxTHREAD_DETACHED) {
+        wxThread(wxTHREAD_DETACHED), m_pQueue(pQueue), m_ID(id) {
     dev = NULL;
-    this->frame = frame;
 }
 SDRThread::~SDRThread() {
 
@@ -112,12 +111,34 @@ wxThread::ExitCode SDRThread::Entry() {
     std::cout << "Sampling..";
     while (!TestDestroy()) {
 
+      if (m_pQueue->Stacksize()) {
+        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_EXIT: // thread should exit
+          Sleep(1000); // wait a while
+          throw SDRThreadTask::SDR_THREAD_EXIT; // confirm exit command
+        case SDRThreadTask::SDR_THREAD_JOB: // process a standard task
+          Sleep(2000);
+          m_pQueue->Report(SDRThreadTask::SDR_THREAD_JOB, wxString::Format(wxT("Task #%s done."), task.m_Arg.c_str()), m_ID); // report successful completion
+          break;
+        case SDRThreadTask::SDR_THREAD_JOBERR: // process a task that terminates with an error
+          m_pQueue->Report(SDRThreadTask::SDR_THREAD_JOB, wxString::Format(wxT("Task #%s errorneous."), task.m_Arg.c_str()), m_ID);
+          Sleep(1000);
+          throw SDRThreadTask::SDR_THREAD_EXIT; // report exit of worker thread
+          break;
+        case SDRThreadTask::SDR_THREAD_NULL: // dummy command
+        default: break; // default
+        } // switch(task.m_cmd)
+      }
+      }
+
         rtlsdr_read_sync(dev, buf, BUF_SIZE, &n_read);
         // move around
-        long freq = 98000000+(20000000)*sin(seconds/50.0);
+        // long freq = 98000000+(20000000)*sin(seconds/50.0);
-        rtlsdr_set_center_freq(dev, freq);
+        // rtlsdr_set_center_freq(dev, freq);
-        
+        // std::cout << "Frequency: " << freq << std::endl;
-        std::cout << "Frequency: " << freq << std::endl;
 
         if (!TestDestroy()) {
             std::vector<signed char> *new_buffer = new std::vector<signed char>();
@@ -130,13 +151,13 @@ wxThread::ExitCode SDRThread::Entry() {
             seconds += time_slice;
 
             // std::cout << "Time Slice: " << time_slice << std::endl;
-            if (!TestDestroy()) {
+            // if (!TestDestroy()) {
-                wxThreadEvent event(wxEVT_THREAD, EVENT_SDR_INPUT);
+                // wxThreadEvent event(wxEVT_THREAD, EVENT_SDR_INPUT);
-                event.SetPayload(new_buffer);
+                // event.SetPayload(new_buffer);
-                wxQueueEvent(frame, event.Clone());
+                // wxQueueEvent(frame, event.Clone());
-            } else {
+            // } else {
                 delete new_buffer;
-            }
+            // }
         }
     }
     std::cout << std::endl << "Done." << std::endl << std::endl;

src/SDRThread.h

@@ -10,6 +10,7 @@
 #include "wx/thread.h"
 
 #include "AppFrame.h"
+#include "SDRThreadQueue.h"
 
 // declare a new type of event, to be used by our SDRThread class:
 //wxDECLARE_EVENT(wxEVT_COMMAND_SDRThread_COMPLETED, wxThreadEvent);
@@ -24,13 +25,14 @@ class SDRThread: public wxThread {
 public:
 	rtlsdr_dev_t *dev;
 
-	SDRThread(AppFrame *appframe);
+	SDRThread(SDRThreadQueue* pQueue, int id=0);
 	~SDRThread();
 
 	void enumerate_rtl();
 
 protected:
 	virtual ExitCode Entry();
-	AppFrame *frame;
 	uint32_t sample_rate;
+  SDRThreadQueue* m_pQueue;
+  int m_ID;
 };

src/SDRThreadQueue.h

@@ -0,0 +1,60 @@
+#pragma once
+
+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, 
+   };
+
+   SDRThreadTask() : m_cmd(eID_THREAD_NULL) {}
+   SDRThreadTask(SDR_COMMAND cmd, const wxString& arg) : m_cmd(cmd), m_Arg(arg) {}
+   SDR_COMMAND m_cmd; 
+   wxString m_Arg;
+ };
+ 
+ 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) : m_pParent(pParent) {}
+     void AddTask(const SDRThreadTask& task, const SDR_PRIORITY& priority=SDR_PRIORITY_NORMAL)
+     {
+       wxMutexLocker lock(m_MutexQueue);
+       m_Tasks.insert(std::make_pair(priority, task));
+       m_QueueCount.Post();
+     }
+     SDRThreadTask 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 Report(const SDRThreadTask::SDR_COMMAND& cmd, const wxString& sArg=wxEmptyString, int iArg=0)
+     {
+       wxCommandEvent evt(wxEVT_THREAD, cmd);
+       evt.SetString(sArg);
+       evt.SetInt(iArg);
+       m_pParent->AddPendingEvent(evt);
+     }
+     size_t Stacksize()
+     {
+       wxMutexLocker lock(m_MutexQueue);
+       return m_Tasks.size();
+     }
+
+   private:
+     wxEvtHandler* m_pParent;
+     std::multimap<tPRIORITY, SDRThreadTask> m_Tasks; 
+     wxMutex m_MutexQueue;
+     wxSemaphore m_QueueCount;
+   };

src/WorkerThread.cpp

@@ -0,0 +1,154 @@
+
+ 
+ 
+   class WorkerThread : public wxThread
+    {
+    public:
+      WorkerThread(SDRThreadQueue* pQueue, int id=0) : m_pQueue(pQueue), m_ID(id) { assert(pQueue); wxThread::Create(); }
+ 
+    private:
+      SDRThreadQueue* m_pQueue;
+      int m_ID;
+ 
+      virtual wxThread::ExitCode Entry()
+      {
+        Sleep(1000); // sleep a while to simulate some time-consuming init procedure
+        SDRThreadTask::SDR_COMMAND iErr;
+        m_pQueue->Report(SDRThreadTask::SDR_THREAD_STARTED, wxEmptyString, m_ID); // tell main thread that worker thread has successfully started
+        try { while(true) OnTask(); } // this is the main loop: process tasks until a task handler throws
+        catch(SDRThreadTask::SDR_COMMAND& i) { m_pQueue->Report(iErr=i, wxEmptyString, m_ID); } // catch return value from error condition
+        return (wxThread::ExitCode)iErr; // and return exit code
+      } // virtual wxThread::ExitCode Entry()
+ 
+      virtual void OnTask()
+      {
+        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_EXIT: // thread should exit
+          Sleep(1000); // wait a while
+          throw SDRThreadTask::SDR_THREAD_EXIT; // confirm exit command
+        case SDRThreadTask::SDR_THREAD_JOB: // process a standard task
+          Sleep(2000);
+          m_pQueue->Report(SDRThreadTask::SDR_THREAD_JOB, wxString::Format(wxT("Task #%s done."), task.m_Arg.c_str()), m_ID); // report successful completion
+          break;
+        case SDRThreadTask::SDR_THREAD_JOBERR: // process a task that terminates with an error
+          m_pQueue->Report(SDRThreadTask::SDR_THREAD_JOB, wxString::Format(wxT("Task #%s errorneous."), task.m_Arg.c_str()), m_ID);
+          Sleep(1000);
+          throw SDRThreadTask::SDR_THREAD_EXIT; // report exit of worker thread
+          break;
+        case SDRThreadTask::SDR_THREAD_NULL: // dummy command
+        default: break; // default
+        } // switch(task.m_cmd)
+      } // virtual void OnTask()
+    }; // class WorkerThread : public wxThread
+ 
+ 
+    // ----------------------------------------------------------------------------
+    // main frame
+    // ----------------------------------------------------------------------------
+    class MyFrame : public wxFrame
+    {
+      enum { eQUIT=wxID_CLOSE, eSTART_THREAD=wxID_HIGHEST+100 };
+      DECLARE_DYNAMIC_CLASS(MyFrame)
+    public:
+      MyFrame(const wxString& title) : wxFrame(NULL, wxID_ANY, title), m_pQueue(NULL)
+      {
+        wxMenu *fileMenu=new wxMenu;
+        fileMenu->Append(eSTART_THREAD,   _T("&Start a thread\tAlt-S"),     _T("Starts one worker thread"));
+        fileMenu->Append(SDRThreadTask::SDR_THREAD_EXIT,     _T("Sto&p a thread\tAlt-P"), _T("Stops one worker thread"));
+        fileMenu->Append(SDRThreadTask::SDR_THREAD_JOB,      _T("&Add task to thread\tAlt-A"),  _T("Adds a task to the worker thread"));
+        fileMenu->Append(SDRThreadTask::SDR_THREAD_JOBERR,   _T("Add &errorneous task to thread\tAlt-E"), _T("Adds am errorneous task to the worker thread"));
+        fileMenu->Append(eQUIT,      _T("E&xit\tAlt-X"),  _T("Exit this program"));
+        wxMenuBar *menuBar=new wxMenuBar();
+        menuBar->Append(fileMenu, _T("&Options"));
+        SetMenuBar(menuBar);
+        EnableThreadControls(false); // disable thread controls since worker thread isn't running yet
+        fileMenu->Enable(eSTART_THREAD, true); // starting threads should always be possible
+        CreateStatusBar();
+        SetStatusText(_T("Worker thread sample"));
+        m_pQueue=new SDRThreadQueue(this);
+      } // MyFrame(const wxString& title)
+      ~MyFrame() { delete m_pQueue; }
+      void EnableThreadControls(bool bEnable)
+      {
+        wxMenu* pMenu=GetMenuBar()->GetMenu(0);
+        static const int MENUIDS[]={eSTART_THREAD, SDRThreadTask::SDR_THREAD_EXIT, SDRThreadTask::SDR_THREAD_JOB, SDRThreadTask::SDR_THREAD_JOBERR};
+        for(int i=0; i<WXSIZEOF(MENUIDS); pMenu->Enable(MENUIDS[i++], bEnable));
+      }
+      void OnStart(wxCommandEvent& WXUNUSED(event)) // start one worker thread
+      {
+        int id=m_Threads.empty()?1:m_Threads.back()+1;
+        m_Threads.push_back(id);
+        WorkerThread* pThread=new WorkerThread(m_pQueue, id); // create a new worker thread, increment thread counter (this implies, thread will start OK)
+        pThread->Run();
+        SetStatusText(wxString::Format(wxT("[%i]: Thread started."), id));
+      }
+      void OnStop(wxCommandEvent& WXUNUSED(event)) // stop one worker thread
+      {
+        if(m_Threads.empty()) { EnableThreadControls(false); Destroy(); return; } // no thread(s) running: frame can be destroyed right away
+        m_pQueue->AddTask(SDRThreadTask(SDRThreadTask::SDR_THREAD_EXIT, wxEmptyString), SDRThreadQueue::eHIGHEST); // add SDR_THREAD_EXIT notification with highest priority to bypass other running tasks
+        SetStatusText(_T("Stopping thread..."));
+      }
+      void OnTask(wxCommandEvent& event) // handler for launching a task for worker thread(s)
+      {
+        int iTask=rand();
+        m_pQueue->AddTask(SDRThreadTask((SDRThreadTask::SDR_COMMAND)event.GetId(), wxString::Format(wxT("%u"), iTask)));
+        SetStatusText(wxString::Format(wxT("Task #%i started."), iTask)); // just set the status text
+      }
+      void OnThread(wxCommandEvent& event) // handler for thread notifications
+      {
+        switch(event.GetId())
+        {
+        case SDRThreadTask::SDR_THREAD_JOB:
+          SetStatusText(wxString::Format(wxT("[%i]: %s"), event.GetInt(), event.GetString().c_str())); // progress display
+          break; 
+        case SDRThreadTask::SDR_THREAD_EXIT:
+          SetStatusText(wxString::Format(wxT("[%i]: Stopped."), event.GetInt()));
+          m_Threads.remove(event.GetInt()); // thread has exited: remove thread ID from list
+          if(m_Threads.empty()) { EnableThreadControls(false); Destroy(); } // destroy main window if no more threads
+          break;
+        case SDRThreadTask::SDR_THREAD_STARTED:
+          SetStatusText(wxString::Format(wxT("[%i]: Ready."), event.GetInt()));
+          EnableThreadControls(true); // at least one thread successfully started: enable controls
+          break; 
+        default: event.Skip();
+        }
+      }
+      void OnQuit(wxCommandEvent& WXUNUSED(event))
+      {
+        if(m_Threads.empty()) { Destroy(); return; } // no thread(s) running - exit right away
+        for(size_t t=0; t<m_Threads.size(); ++t) m_pQueue->AddTask(SDRThreadTask(SDRThreadTask::SDR_THREAD_EXIT, wxEmptyString), SDRThreadQueue::eHIGHEST); } // send all running threads the "EXIT" signal
+      void OnClose(wxCloseEvent& WXUNUSED(event))  { wxCommandEvent e; OnQuit(e); } // just run OnQuit() which will terminate worker threads and destroy the main frame
+    private:
+      MyFrame() : wxFrame() {}
+      SDRThreadQueue* m_pQueue;
+      std::list<int> m_Threads;
+      DECLARE_EVENT_TABLE()
+    }; // class MyFrame : public wxFrame
+    IMPLEMENT_DYNAMIC_CLASS(MyFrame, wxFrame)
+    BEGIN_EVENT_TABLE(MyFrame, wxFrame)
+    EVT_MENU(SDRThreadTask::SDR_THREAD_JOB, MyFrame::OnTask)
+    EVT_MENU(SDRThreadTask::SDR_THREAD_JOBERR, MyFrame::OnTask)
+    EVT_MENU(eSTART_THREAD, MyFrame::OnStart)
+    EVT_MENU(SDRThreadTask::SDR_THREAD_EXIT, MyFrame::OnStop)
+    EVT_COMMAND(wxID_ANY, wxEVT_THREAD, MyFrame::OnThread)
+    EVT_MENU(eQUIT, MyFrame::OnQuit)
+    EVT_CLOSE(MyFrame::OnClose)
+    END_EVENT_TABLE()
+ 
+    // ----------------------------------------------------------------------------
+    // the application
+    // ----------------------------------------------------------------------------
+    class MyApp : public wxApp
+    {
+    public:
+      virtual bool OnInit()
+      {
+        if(!wxApp::OnInit()) return false;
+        MyFrame *frame=new MyFrame(_T("Minimal wxWidgets App"));
+        frame->Show(true);
+        return true;
+      }
+    }; // class MyApp : public wxApp
+    IMPLEMENT_APP(MyApp)