comicsansg/CubicSDR
comicsansg
/
CubicSDR
mirror of https://github.com/cjcliffe/CubicSDR.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #16 from cjcliffe/more-demod-threads-less-work-per-thread 

Split up demodulator workload, tweaked OSX threads
This commit is contained in:
Charles J. Cliffe 2014-12-16 21:10:47 -05:00
parent 76c68cc4fa 912ef930a2
commit a78651687a
16 changed files with 661 additions and 452 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CMakeLists.txt
View File

@ -125,6 +125,7 @@ SET (cubicsdr_sources
src/AppFrame.cpp
src/sdr/SDRThread.cpp
src/sdr/SDRPostThread.cpp
src/demod/DemodulatorPreThread.cpp
src/demod/DemodulatorThread.cpp
src/demod/DemodulatorWorkerThread.cpp
src/demod/DemodulatorInstance.cpp
@ -151,6 +152,7 @@ SET (cubicsdr_headers
src/AppFrame.h
src/sdr/SDRThread.h
src/sdr/SDRPostThread.h
src/demod/DemodulatorPreThread.h
src/demod/DemodulatorThread.h
src/demod/DemodulatorWorkerThread.h
src/demod/DemodulatorInstance.h

6
src/AppFrame.cpp
View File

@ -93,6 +93,10 @@ void AppFrame::OnThread(wxCommandEvent& event) {
void AppFrame::OnIdle(wxIdleEvent& event) {
bool work_done = false;
//#ifdef __APPLE__
// std::this_thread::sleep_for(std::chrono::milliseconds(4));
// std::this_thread::yield();
//#endif
if (!wxGetApp().getIQVisualQueue()->empty()) {
SDRThreadIQData iqData;
wxGetApp().getIQVisualQueue()->pop(iqData);
@ -127,6 +131,6 @@ void AppFrame::OnIdle(wxIdleEvent& event) {
}
if (!work_done) {
event.Skip();
event.Skip();
}
}

8
src/CubicSDRDefs.h
View File

@ -1,12 +1,8 @@
#pragma once
#ifdef __APPLE__
#define BUF_SIZE (16384*3)
#else
#define BUF_SIZE (16384*4)
#endif
#define SRATE 2000000
#define SRATE 2500000
#define FFT_SIZE 2048
#define DEFAULT_FREQ 98900000
#define AUDIO_FREQUENCY 48000
#define AUDIO_FREQUENCY 44100

12
src/audio/AudioThread.cpp
View File

@ -44,7 +44,14 @@ static int audioCallback(void *outputBuffer, void *inputBuffer, unsigned int nBu
}
void AudioThread::threadMain() {
std::cout << "Audio thread initializing.." << std::endl;
#ifdef __APPLE__
pthread_t tID = pthread_self(); // ID of this thread
int priority = sched_get_priority_min( SCHED_RR );
sched_param prio = { priority }; // scheduling priority of thread
pthread_setschedparam( tID, SCHED_RR, &prio );
#endif
std::cout << "Audio thread initializing.." << std::endl;
if (dac.getDeviceCount() < 1) {
std::cout << "No audio devices found!" << std::endl;
@ -59,7 +66,8 @@ void AudioThread::threadMain() {
unsigned int bufferFrames = 256;
RtAudio::StreamOptions opts;
// opts.flags = RTAUDIO_SCHEDULE_REALTIME | RTAUDIO_MINIMIZE_LATENCY;
opts.flags = RTAUDIO_SCHEDULE_REALTIME;
// | RTAUDIO_MINIMIZE_LATENCY;
// opts.flags = RTAUDIO_MINIMIZE_LATENCY;
opts.streamName = "CubicSDR Audio Output";
// opts.priority = sched_get_priority_max(SCHED_FIFO);

43
src/demod/DemodDefs.h
View File

@ -1,6 +1,8 @@
#pragma once
#include "ThreadQueue.h"
#include "CubicSDRDefs.h"
#include "liquid/liquid.h"
enum DemodulatorType {
DEMOD_TYPE_NULL,
@ -17,6 +19,7 @@ public:
DEMOD_THREAD_CMD_NULL,
DEMOD_THREAD_CMD_SET_BANDWIDTH,
DEMOD_THREAD_CMD_SET_FREQUENCY,
DEMOD_THREAD_CMD_DEMOD_PREPROCESS_TERMINATED,
DEMOD_THREAD_CMD_DEMOD_TERMINATED,
DEMOD_THREAD_CMD_AUDIO_TERMINATED
};
@ -58,6 +61,24 @@ public:
}
};
class DemodulatorThreadPostIQData {
public:
std::vector<liquid_float_complex> data;
float audio_resample_ratio;
msresamp_crcf audio_resampler;
float resample_ratio;
msresamp_crcf resampler;
DemodulatorThreadPostIQData(): audio_resample_ratio(0), audio_resampler(NULL) {
}
~DemodulatorThreadPostIQData() {
}
};
class DemodulatorThreadAudioData {
public:
unsigned int frequency;
@ -84,4 +105,26 @@ public:
};
typedef ThreadQueue<DemodulatorThreadIQData> DemodulatorThreadInputQueue;
typedef ThreadQueue<DemodulatorThreadPostIQData> DemodulatorThreadPostInputQueue;
typedef ThreadQueue<DemodulatorThreadCommand> DemodulatorThreadCommandQueue;
class DemodulatorThreadParameters {
public:
unsigned int frequency;
unsigned int inputRate;
unsigned int bandwidth; // set equal to disable second stage re-sampling?
unsigned int audioSampleRate;
DemodulatorType demodType;
DemodulatorThreadParameters() :
frequency(0), inputRate(SRATE), bandwidth(200000), audioSampleRate(
AUDIO_FREQUENCY), demodType(DEMOD_TYPE_FM) {
}
~DemodulatorThreadParameters() {
}
};

146
src/demod/DemodulatorInstance.cpp
View File

@ -1,87 +1,139 @@
#include "DemodulatorInstance.h"
DemodulatorInstance::DemodulatorInstance() :
t_Demod(NULL), t_Audio(NULL), threadQueueDemod(NULL), demodulatorThread(NULL), terminated(false), audioTerminated(false), demodTerminated(
false) {
t_Demod(NULL), t_PreDemod(NULL), t_Audio(NULL), threadQueueDemod(NULL), demodulatorThread(NULL), terminated(false), audioTerminated(false), demodTerminated(
false), preDemodTerminated(false) {
label = new std::string("Unnamed");
threadQueueDemod = new DemodulatorThreadInputQueue;
threadQueueCommand = new DemodulatorThreadCommandQueue;
threadQueueNotify = new DemodulatorThreadCommandQueue;
demodulatorThread = new DemodulatorThread(threadQueueDemod, threadQueueNotify);
demodulatorThread->setCommandQueue(threadQueueCommand);
audioInputQueue = new AudioThreadInputQueue;
audioThread = new AudioThread(audioInputQueue, threadQueueNotify);
demodulatorThread->setAudioInputQueue(audioInputQueue);
label = new std::string("Unnamed");
threadQueueDemod = new DemodulatorThreadInputQueue;
threadQueuePostDemod = new DemodulatorThreadPostInputQueue;
threadQueueCommand = new DemodulatorThreadCommandQueue;
threadQueueNotify = new DemodulatorThreadCommandQueue;
demodulatorPreThread = new DemodulatorPreThread(threadQueueDemod, threadQueuePostDemod, threadQueueNotify);
demodulatorPreThread->setCommandQueue(threadQueueCommand);
demodulatorThread = new DemodulatorThread(threadQueuePostDemod, threadQueueNotify);
audioInputQueue = new AudioThreadInputQueue;
audioThread = new AudioThread(audioInputQueue, threadQueueNotify);
demodulatorThread->setAudioInputQueue(audioInputQueue);
}
DemodulatorInstance::~DemodulatorInstance() {
delete audioThread;
delete demodulatorThread;
delete audioThread;
delete demodulatorThread;
delete audioInputQueue;
delete threadQueueDemod;
delete audioInputQueue;
delete threadQueueDemod;
}
void DemodulatorInstance::setVisualOutputQueue(DemodulatorThreadOutputQueue *tQueue) {
demodulatorThread->setVisualOutputQueue(tQueue);
demodulatorThread->setVisualOutputQueue(tQueue);
}
void DemodulatorInstance::run() {
t_Audio = new std::thread(&AudioThread::threadMain, audioThread);
t_Audio = new std::thread(&AudioThread::threadMain, audioThread);
#ifdef __APPLE__ // Already using pthreads, might as well do some custom init..
pthread_attr_t attr;
size_t size;
pthread_attr_t attr;
size_t size;
pthread_attr_init(&attr);
pthread_attr_setstacksize(&attr, 2048000);
pthread_attr_getstacksize(&attr, &size);
pthread_create(&t_Demod, &attr, &DemodulatorThread::pthread_helper, demodulatorThread);
pthread_attr_destroy(&attr);
pthread_attr_init(&attr);
pthread_attr_setstacksize(&attr, 2048000);
pthread_attr_getstacksize(&attr, &size);
pthread_attr_setschedpolicy(&attr, SCHED_RR);
pthread_create(&t_PreDemod, &attr, &DemodulatorPreThread::pthread_helper, demodulatorPreThread);
pthread_attr_destroy(&attr);
std::cout << "Initialized demodulator stack size of " << size << std::endl;
pthread_attr_init(&attr);
pthread_attr_setstacksize(&attr, 2048000);
pthread_attr_getstacksize(&attr, &size);
pthread_attr_setschedpolicy(&attr, SCHED_RR);
pthread_create(&t_Demod, &attr, &DemodulatorThread::pthread_helper, demodulatorThread);
pthread_attr_destroy(&attr);
std::cout << "Initialized demodulator stack size of " << size << std::endl;
#else
t_Demod = new std::thread(&DemodulatorThread::threadMain, demodulatorThread);
t_PreDemod = new std::thread(&DemodulatorPreThread::threadMain, demodulatorPreThread);
t_Demod = new std::thread(&DemodulatorThread::threadMain, demodulatorThread);
#endif
}
void DemodulatorInstance::updateLabel(int freq) {
std::stringstream newLabel;
newLabel.precision(3);
newLabel << std::fixed << ((float) freq / 1000000.0);
setLabel(newLabel.str());
std::stringstream newLabel;
newLabel.precision(3);
newLabel << std::fixed << ((float) freq / 1000000.0);
setLabel(newLabel.str());
}
DemodulatorThreadCommandQueue *DemodulatorInstance::getCommandQueue() {
return threadQueueCommand;
return threadQueueCommand;
}
DemodulatorThreadParameters &DemodulatorInstance::getParams() {
return demodulatorThread->getParams();
return demodulatorPreThread->getParams();
}
void DemodulatorInstance::terminate() {
std::cout << "Terminating demodulator thread.." << std::endl;
demodulatorThread->terminate();
//#ifdef __APPLE__
// pthread_join(t_Demod,NULL);
//#else
//#endif
std::cout << "Terminating demodulator audio thread.." << std::endl;
audioThread->terminate();
std::cout << "Terminating demodulator preprocessor thread.." << std::endl;
demodulatorPreThread->terminate();
std::cout << "Terminating demodulator thread.." << std::endl;
demodulatorThread->terminate();
std::cout << "Terminating demodulator audio thread.." << std::endl;
audioThread->terminate();
}
std::string DemodulatorInstance::getLabel() {
return *(label.load());
return *(label.load());
}
void DemodulatorInstance::setLabel(std::string labelStr) {
std::string *newLabel = new std::string;
newLabel->append(labelStr);
std::string *oldLabel;
oldLabel = label;
label = newLabel;
delete oldLabel;
std::string *newLabel = new std::string;
newLabel->append(labelStr);
std::string *oldLabel;
oldLabel = label;
label = newLabel;
delete oldLabel;
}
bool DemodulatorInstance::isTerminated() {
while (!threadQueueNotify->empty()) {
DemodulatorThreadCommand cmd;
threadQueueNotify->pop(cmd);
switch (cmd.cmd) {
case DemodulatorThreadCommand::DEMOD_THREAD_CMD_AUDIO_TERMINATED:
audioThread = NULL;
t_Audio->join();
audioTerminated = true;
break;
case DemodulatorThreadCommand::DEMOD_THREAD_CMD_DEMOD_TERMINATED:
demodulatorThread = NULL;
#ifdef __APPLE__
pthread_join(t_Demod, NULL);
#else
t_Demod->join();
#endif
demodTerminated = true;
break;
case DemodulatorThreadCommand::DEMOD_THREAD_CMD_DEMOD_PREPROCESS_TERMINATED:
demodulatorPreThread = NULL;
#ifdef __APPLE__
pthread_join(t_PreDemod, NULL);
#else
t_PreDemod->join();
#endif
preDemodTerminated = true;
break;
default:
break;
}
}
terminated = audioTerminated && demodTerminated && preDemodTerminated;
return terminated;
}

6
src/demod/DemodulatorInstance.h
View File

@ -5,17 +5,22 @@
#include <thread>
#include "DemodulatorThread.h"
#include "DemodulatorPreThread.h"
class DemodulatorInstance {
public:
DemodulatorThreadInputQueue* threadQueueDemod;
DemodulatorThreadPostInputQueue* threadQueuePostDemod;
DemodulatorThreadCommandQueue* threadQueueCommand;
DemodulatorThreadCommandQueue* threadQueueNotify;
DemodulatorPreThread *demodulatorPreThread;
DemodulatorThread *demodulatorThread;
#ifdef __APPLE__
pthread_t t_PreDemod;
pthread_t t_Demod;
#else
std::thread *t_PreDemod;
std::thread *t_Demod;
#endif
@ -44,6 +49,7 @@ private:
bool terminated;
bool demodTerminated;
bool audioTerminated;
bool preDemodTerminated;
};

193
src/demod/DemodulatorMgr.cpp
View File

@ -5,169 +5,138 @@
#include <string>
#include <sstream>
DemodulatorMgr::DemodulatorMgr() :
activeDemodulator(NULL), lastActiveDemodulator(NULL), activeVisualDemodulator(NULL) {
activeDemodulator(NULL), lastActiveDemodulator(NULL), activeVisualDemodulator(NULL) {
}
DemodulatorMgr::~DemodulatorMgr() {
terminateAll();
terminateAll();
}
DemodulatorInstance *DemodulatorMgr::newThread() {
DemodulatorInstance *newDemod = new DemodulatorInstance;
DemodulatorInstance *newDemod = new DemodulatorInstance;
demods.push_back(newDemod);
demods.push_back(newDemod);
std::stringstream label;
label << demods.size();
newDemod->setLabel(label.str());
std::stringstream label;
label << demods.size();
newDemod->setLabel(label.str());
return newDemod;
return newDemod;
}
void DemodulatorMgr::terminateAll() {
while (demods.size()) {
DemodulatorInstance *d = demods.back();
deleteThread(d);
}
while (demods.size()) {
DemodulatorInstance *d = demods.back();
deleteThread(d);
}
}
std::vector<DemodulatorInstance *> &DemodulatorMgr::getDemodulators() {
return demods;
return demods;
}
void DemodulatorMgr::deleteThread(DemodulatorInstance *demod) {
std::vector<DemodulatorInstance *>::iterator i;
std::vector<DemodulatorInstance *>::iterator i;
i = std::find(demods.begin(), demods.end(), demod);
i = std::find(demods.begin(), demods.end(), demod);
if (activeDemodulator == demod) {
activeDemodulator = NULL;
}
if (lastActiveDemodulator == demod) {
lastActiveDemodulator = NULL;
}
if (activeVisualDemodulator == demod) {
activeVisualDemodulator = NULL;
}
if (activeDemodulator == demod) {
activeDemodulator = NULL;
}
if (lastActiveDemodulator == demod) {
lastActiveDemodulator = NULL;
}
if (activeVisualDemodulator == demod) {
activeVisualDemodulator = NULL;
}
if (i != demods.end()) {
demods.erase(i);
demod->terminate();
}
if (i != demods.end()) {
demods.erase(i);
demod->terminate();
}
demods_deleted.push_back(demod);
demods_deleted.push_back(demod);
garbageCollect();
garbageCollect();
}
std::vector<DemodulatorInstance *> *DemodulatorMgr::getDemodulatorsAt(int freq, int bandwidth) {
std::vector<DemodulatorInstance *> *foundDemods = new std::vector<DemodulatorInstance *>();
std::vector<DemodulatorInstance *> *foundDemods = new std::vector<DemodulatorInstance *>();
for (int i = 0, iMax = demods.size(); i < iMax; i++) {
DemodulatorInstance *testDemod = demods[i];
for (int i = 0, iMax = demods.size(); i < iMax; i++) {
DemodulatorInstance *testDemod = demods[i];
int freqTest = testDemod->getParams().frequency;
int bandwidthTest = testDemod->getParams().bandwidth;
int halfBandwidthTest = bandwidthTest / 2;
int freqTest = testDemod->getParams().frequency;
int bandwidthTest = testDemod->getParams().bandwidth;
int halfBandwidthTest = bandwidthTest / 2;
int halfBuffer = bandwidth / 2;
int halfBuffer = bandwidth / 2;
if ((freq <= (freqTest + halfBandwidthTest + halfBuffer)) && (freq >= (freqTest - halfBandwidthTest - halfBuffer))) {
foundDemods->push_back(testDemod);
}
}
if ((freq <= (freqTest + halfBandwidthTest + halfBuffer)) && (freq >= (freqTest - halfBandwidthTest - halfBuffer))) {
foundDemods->push_back(testDemod);
}
}
return foundDemods;
return foundDemods;
}
void DemodulatorMgr::setActiveDemodulator(DemodulatorInstance *demod, bool temporary) {
if (!temporary) {
if (activeDemodulator != NULL) {
lastActiveDemodulator = activeDemodulator;
} else {
lastActiveDemodulator = demod;
}
}
if (!temporary) {
if (activeDemodulator != NULL) {
lastActiveDemodulator = activeDemodulator;
} else {
lastActiveDemodulator = demod;
}
}
if (activeVisualDemodulator) {
activeVisualDemodulator->setVisualOutputQueue(NULL);
}
if (demod) {
demod->setVisualOutputQueue(wxGetApp().getAudioVisualQueue());
activeVisualDemodulator = demod;
} else {
DemodulatorInstance *last = getLastActiveDemodulator();
if (last) {
last->setVisualOutputQueue(wxGetApp().getAudioVisualQueue());
}
activeVisualDemodulator = last;
}
if (activeVisualDemodulator) {
activeVisualDemodulator->setVisualOutputQueue(NULL);
}
if (demod) {
demod->setVisualOutputQueue(wxGetApp().getAudioVisualQueue());
activeVisualDemodulator = demod;
} else {
DemodulatorInstance *last = getLastActiveDemodulator();
if (last) {
last->setVisualOutputQueue(wxGetApp().getAudioVisualQueue());
}
activeVisualDemodulator = last;
}
activeDemodulator = demod;
activeDemodulator = demod;
garbageCollect();
garbageCollect();
}
DemodulatorInstance *DemodulatorMgr::getActiveDemodulator() {
return activeDemodulator;
return activeDemodulator;
}
DemodulatorInstance *DemodulatorMgr::getLastActiveDemodulator() {
if (std::find(demods.begin(), demods.end(), lastActiveDemodulator) == demods.end()) {
lastActiveDemodulator = activeDemodulator;
}
if (std::find(demods.begin(), demods.end(), lastActiveDemodulator) == demods.end()) {
lastActiveDemodulator = activeDemodulator;
}
return lastActiveDemodulator;
return lastActiveDemodulator;
}
void DemodulatorMgr::garbageCollect() {
if (demods_deleted.size()) {
std::vector<DemodulatorInstance *>::iterator i;
if (demods_deleted.size()) {
std::vector<DemodulatorInstance *>::iterator i;
for (i = demods_deleted.begin(); i != demods_deleted.end(); i++) {
if ((*i)->isTerminated()) {
DemodulatorInstance *deleted = (*i);
demods_deleted.erase(i);
for (i = demods_deleted.begin(); i != demods_deleted.end(); i++) {
if ((*i)->isTerminated()) {
DemodulatorInstance *deleted = (*i);
demods_deleted.erase(i);
std::cout << "Garbage collected demodulator instance " << deleted->getLabel() << std::endl;
std::cout << "Garbage collected demodulator instance " << deleted->getLabel() << std::endl;
delete deleted;
return;
}
}
}
}
bool DemodulatorInstance::isTerminated() {
while (!threadQueueNotify->empty()) {
DemodulatorThreadCommand cmd;
threadQueueNotify->pop(cmd);
switch (cmd.cmd) {
case DemodulatorThreadCommand::DEMOD_THREAD_CMD_AUDIO_TERMINATED:
audioThread = NULL;
t_Audio->join();
audioTerminated = true;
break;
case DemodulatorThreadCommand::DEMOD_THREAD_CMD_DEMOD_TERMINATED:
demodulatorThread = NULL;
#ifdef __APPLE__
pthread_join(t_Demod,NULL);
#else
t_Demod->join();
#endif
demodTerminated = true;
break;
default:
break;
}
}
terminated = audioTerminated && demodTerminated;
return terminated;
delete deleted;
return;
}
}
}
}

240
src/demod/DemodulatorPreThread.cpp Normal file
View File

@ -0,0 +1,240 @@
#include "CubicSDRDefs.h"
#include <vector>
#ifdef __APPLE__
#include <pthread.h>
#endif
#include "DemodulatorPreThread.h"
DemodulatorPreThread::DemodulatorPreThread(DemodulatorThreadInputQueue* pQueueIn, DemodulatorThreadPostInputQueue* pQueueOut, DemodulatorThreadCommandQueue* threadQueueNotify) :
inputQueue(pQueueIn), postInputQueue(pQueueOut), terminated(false), initialized(false), audio_resampler(NULL), resample_ratio(1), audio_resample_ratio(
1), resampler(NULL), commandQueue(NULL), fir_filter(NULL), audioInputQueue(NULL), threadQueueNotify(threadQueueNotify) {
float kf = 0.5; // modulation factor
fdem = freqdem_create(kf);
// freqdem_print(fdem);
nco_shift = nco_crcf_create(LIQUID_VCO);
shift_freq = 0;
workerQueue = new DemodulatorThreadWorkerCommandQueue;
workerResults = new DemodulatorThreadWorkerResultQueue;
workerThread = new DemodulatorWorkerThread(workerQueue, workerResults);
t_Worker = new std::thread(&DemodulatorWorkerThread::threadMain, workerThread);
}
void DemodulatorPreThread::initialize() {
initialized = false;
resample_ratio = (float) (params.bandwidth) / (float) params.inputRate;
audio_resample_ratio = (float) (params.audioSampleRate) / (float) params.bandwidth;
float fc = 0.5 * ((double) params.bandwidth / (double) params.inputRate); // filter cutoff frequency
if (fc <= 0) {
fc = 0;
}
if (fc >= 0.5) {
fc = 0.5;
}
float ft = 0.05f; // filter transition
float As = 60.0f; // stop-band attenuation [dB]
float mu = 0.0f; // fractional timing offset
// estimate required filter length and generate filter
unsigned int h_len = estimate_req_filter_len(ft, As);
float h[h_len];
liquid_firdes_kaiser(h_len, fc, As, mu, h);
if (fir_filter) {
firfilt_crcf_recreate(fir_filter, h, h_len);
} else {
fir_filter = firfilt_crcf_create(h, h_len);
}
// create multi-stage arbitrary resampler object
if (resampler) {
msresamp_crcf_destroy(resampler);
}
resampler = msresamp_crcf_create(resample_ratio, As);
// msresamp_crcf_print(resampler);
if (audio_resampler) {
msresamp_crcf_destroy(audio_resampler);
}
audio_resampler = msresamp_crcf_create(audio_resample_ratio, As);
// msresamp_crcf_print(audio_resampler);
initialized = true;
// std::cout << "inputResampleRate " << params.bandwidth << std::endl;
last_params = params;
}
DemodulatorPreThread::~DemodulatorPreThread() {
delete workerThread;
delete workerQueue;
delete workerResults;
}
#ifdef __APPLE__
void *DemodulatorPreThread::threadMain() {
#else
void DemodulatorPreThread::threadMain() {
#endif
#ifdef __APPLE__
pthread_t tID = pthread_self(); // ID of this thread
int priority = sched_get_priority_min( SCHED_RR );
sched_param prio = { priority }; // scheduling priority of thread
pthread_setschedparam( tID, SCHED_RR, &prio );
#endif
if (!initialized) {
initialize();
}
std::cout << "Demodulator preprocessor thread started.." << std::endl;
while (!terminated) {
DemodulatorThreadIQData inp;
inputQueue->pop(inp);
bool bandwidthChanged = false;
DemodulatorThreadParameters bandwidthParams = params;
if (!commandQueue->empty()) {
bool paramsChanged = false;
while (!commandQueue->empty()) {
DemodulatorThreadCommand command;
commandQueue->pop(command);
switch (command.cmd) {
case DemodulatorThreadCommand::DEMOD_THREAD_CMD_SET_BANDWIDTH:
if (command.int_value < 3000) {
command.int_value = 3000;
}
if (command.int_value > SRATE) {
command.int_value = SRATE;
}
bandwidthParams.bandwidth = command.int_value;
bandwidthChanged = true;
break;
case DemodulatorThreadCommand::DEMOD_THREAD_CMD_SET_FREQUENCY:
params.frequency = command.int_value;
break;
}
}
if (bandwidthChanged) {
DemodulatorWorkerThreadCommand command(DemodulatorWorkerThreadCommand::DEMOD_WORKER_THREAD_CMD_BUILD_FILTERS);
command.audioSampleRate = bandwidthParams.audioSampleRate;
command.bandwidth = bandwidthParams.bandwidth;
command.frequency = bandwidthParams.frequency;
command.inputRate = bandwidthParams.inputRate;
workerQueue->push(command);
}
}
if (!initialized) {
continue;
}
// Requested frequency is not center, shift it into the center!
if (inp.frequency != params.frequency) {
if ((params.frequency - inp.frequency) != shift_freq) {
shift_freq = params.frequency - inp.frequency;
if (abs(shift_freq) <= (int) ((float) (SRATE / 2) * 1.5)) {
nco_crcf_set_frequency(nco_shift, (2.0 * M_PI) * (((float) abs(shift_freq)) / ((float) SRATE)));
}
}
}
if (abs(shift_freq) > (int) ((float) (SRATE / 2) * 1.5)) {
continue;
}
std::vector<signed char> *data = &inp.data;
if (data->size()) {
int bufSize = data->size() / 2;
liquid_float_complex in_buf_data[bufSize];
liquid_float_complex out_buf_data[bufSize];
liquid_float_complex *in_buf = in_buf_data;
liquid_float_complex *out_buf = out_buf_data;
liquid_float_complex *temp_buf = NULL;
for (int i = 0; i < bufSize; i++) {
in_buf[i].real = (float) (*data)[i * 2] / 127.0f;
in_buf[i].imag = (float) (*data)[i * 2 + 1] / 127.0f;
}
if (shift_freq != 0) {
if (shift_freq < 0) {
nco_crcf_mix_block_up(nco_shift, in_buf, out_buf, bufSize);
} else {
nco_crcf_mix_block_down(nco_shift, in_buf, out_buf, bufSize);
}
temp_buf = in_buf;
in_buf = out_buf;
out_buf = temp_buf;
}
DemodulatorThreadPostIQData resamp;
resamp.data.resize(bufSize);
firfilt_crcf_execute_block(fir_filter, in_buf, bufSize, &resamp.data[0]);
resamp.audio_resample_ratio = audio_resample_ratio;
resamp.audio_resampler = audio_resampler;
resamp.resample_ratio = resample_ratio;
resamp.resampler = resampler;
postInputQueue->push(resamp);
}
if (!workerResults->empty()) {
while (!workerResults->empty()) {
DemodulatorWorkerThreadResult result;
workerResults->pop(result);
switch (result.cmd) {
case DemodulatorWorkerThreadResult::DEMOD_WORKER_THREAD_RESULT_FILTERS:
firfilt_crcf_destroy(fir_filter);
// msresamp_crcf_destroy(resampler);
// msresamp_crcf_destroy(audio_resampler);
fir_filter = result.fir_filter;
resampler = result.resampler;
audio_resampler = result.audio_resampler;
resample_ratio = result.resample_ratio;
audio_resample_ratio = result.audio_resample_ratio;
params.audioSampleRate = result.audioSampleRate;
params.bandwidth = result.bandwidth;
params.inputRate = result.inputRate;
break;
}
}
}
}
std::cout << "Demodulator preprocessor thread done." << std::endl;
DemodulatorThreadCommand tCmd(DemodulatorThreadCommand::DEMOD_THREAD_CMD_DEMOD_PREPROCESS_TERMINATED);
tCmd.context = this;
threadQueueNotify->push(tCmd);
}
void DemodulatorPreThread::terminate() {
terminated = true;
DemodulatorThreadIQData inp; // push dummy to nudge queue
inputQueue->push(inp);
workerThread->terminate();
}

73
src/demod/DemodulatorPreThread.h Normal file
View File

@ -0,0 +1,73 @@
#pragma once
#include <queue>
#include <vector>
#include "CubicSDRDefs.h"
#include "DemodDefs.h"
#include "DemodulatorWorkerThread.h"
class DemodulatorPreThread {
public:
DemodulatorPreThread(DemodulatorThreadInputQueue* pQueueIn, DemodulatorThreadPostInputQueue* pQueueOut, DemodulatorThreadCommandQueue* threadQueueNotify);
~DemodulatorPreThread();
#ifdef __APPLE__
void *threadMain();
#else
void threadMain();
#endif
void setCommandQueue(DemodulatorThreadCommandQueue *tQueue) {
commandQueue = tQueue;
}
void setAudioInputQueue(AudioThreadInputQueue *tQueue) {
audioInputQueue = tQueue;
}
DemodulatorThreadParameters &getParams() {
return params;
}
void initialize();
void terminate();
#ifdef __APPLE__
static void *pthread_helper(void *context) {
return ((DemodulatorPreThread *) context)->threadMain();
}
#endif
protected:
DemodulatorThreadInputQueue* inputQueue;
DemodulatorThreadPostInputQueue* postInputQueue;
DemodulatorThreadCommandQueue* commandQueue;
AudioThreadInputQueue *audioInputQueue;
firfilt_crcf fir_filter;
msresamp_crcf resampler;
float resample_ratio;
msresamp_crcf audio_resampler;
float audio_resample_ratio;
DemodulatorThreadParameters params;
DemodulatorThreadParameters last_params;
freqdem fdem;
nco_crcf nco_shift;
int shift_freq;
std::atomic<bool> terminated;
std::atomic<bool> initialized;
DemodulatorWorkerThread *workerThread;
std::thread *t_Worker;
DemodulatorThreadWorkerCommandQueue *workerQueue;
DemodulatorThreadWorkerResultQueue *workerResults;
DemodulatorThreadCommandQueue* threadQueueNotify;
};

297
src/demod/DemodulatorThread.cpp
View File

@ -6,256 +6,113 @@
#include <pthread.h>
#endif
DemodulatorThread::DemodulatorThread(DemodulatorThreadInputQueue* pQueue, DemodulatorThreadCommandQueue* threadQueueNotify) :
inputQueue(pQueue), visOutQueue(NULL), terminated(false), initialized(false), audio_resampler(NULL), resample_ratio(1), audio_resample_ratio(
1), resampler(NULL), commandQueue(NULL), fir_filter(NULL), audioInputQueue(NULL), threadQueueNotify(threadQueueNotify) {
DemodulatorThread::DemodulatorThread(DemodulatorThreadPostInputQueue* pQueue, DemodulatorThreadCommandQueue* threadQueueNotify) :
postInputQueue(pQueue), visOutQueue(NULL), terminated(false), audioInputQueue(NULL), threadQueueNotify(threadQueueNotify) {
float kf = 0.5; // modulation factor
fdem = freqdem_create(kf);
float kf = 0.5; // modulation factor
fdem = freqdem_create(kf);
// freqdem_print(fdem);
nco_shift = nco_crcf_create(LIQUID_VCO);
shift_freq = 0;
workerQueue = new DemodulatorThreadWorkerCommandQueue;
workerResults = new DemodulatorThreadWorkerResultQueue;
workerThread = new DemodulatorWorkerThread(workerQueue, workerResults);
t_Worker = new std::thread(&DemodulatorWorkerThread::threadMain, workerThread);
}
void DemodulatorThread::initialize() {
initialized = false;
resample_ratio = (float) (params.bandwidth) / (float) params.inputRate;
audio_resample_ratio = (float) (params.audioSampleRate) / (float) params.bandwidth;
float fc = 0.5 * ((double) params.bandwidth / (double) params.inputRate); // filter cutoff frequency
if (fc <= 0) {
fc = 0;
}
if (fc >= 0.5) {
fc = 0.5;
}
float ft = 0.05f; // filter transition
float As = 60.0f; // stop-band attenuation [dB]
float mu = 0.0f; // fractional timing offset
// estimate required filter length and generate filter
unsigned int h_len = estimate_req_filter_len(ft, As);
float h[h_len];
liquid_firdes_kaiser(h_len, fc, As, mu, h);
if (fir_filter) {
firfilt_crcf_recreate(fir_filter, h, h_len);
} else {
fir_filter = firfilt_crcf_create(h, h_len);
}
// create multi-stage arbitrary resampler object
if (resampler) {
msresamp_crcf_destroy(resampler);
}
resampler = msresamp_crcf_create(resample_ratio, As);
// msresamp_crcf_print(resampler);
if (audio_resampler) {
msresamp_crcf_destroy(audio_resampler);
}
audio_resampler = msresamp_crcf_create(audio_resample_ratio, As);
// msresamp_crcf_print(audio_resampler);
initialized = true;
// std::cout << "inputResampleRate " << params.bandwidth << std::endl;
last_params = params;
}
DemodulatorThread::~DemodulatorThread() {
delete workerThread;
delete workerQueue;
delete workerResults;
}
#ifdef __APPLE__
void *DemodulatorThread::threadMain() {
#else
void DemodulatorThread::threadMain() {
void DemodulatorThread::threadMain() {
#endif
#ifdef __APPLE__
pthread_t tID = pthread_self(); // ID of this thread
int priority = sched_get_priority_min( SCHED_RR );
sched_param prio = { priority }; // scheduling priority of thread
pthread_setschedparam( tID, SCHED_RR, &prio );
#endif
if (!initialized) {
initialize();
}
msresamp_crcf audio_resampler = NULL;
msresamp_crcf resampler = NULL;
liquid_float_complex *in_buf = new liquid_float_complex[BUF_SIZE / 2];
liquid_float_complex *out_buf = new liquid_float_complex[BUF_SIZE / 2];
std::cout << "Demodulator thread started.." << std::endl;
while (!terminated) {
DemodulatorThreadPostIQData inp;
postInputQueue->pop(inp);
std::cout << "Demodulator thread started.." << std::endl;
while (!terminated) {
DemodulatorThreadIQData inp;
inputQueue->pop(inp);
int bufSize = inp.data.size();
bool bandwidthChanged = false;
DemodulatorThreadParameters bandwidthParams = params;
if (!bufSize) {
continue;
}
if (!commandQueue->empty()) {
bool paramsChanged = false;
while (!commandQueue->empty()) {
DemodulatorThreadCommand command;
commandQueue->pop(command);
switch (command.cmd) {
case DemodulatorThreadCommand::DEMOD_THREAD_CMD_SET_BANDWIDTH:
if (command.int_value < 3000) {
command.int_value = 3000;
}
if (command.int_value > SRATE) {
command.int_value = SRATE;
}
bandwidthParams.bandwidth = command.int_value;
bandwidthChanged = true;
break;
case DemodulatorThreadCommand::DEMOD_THREAD_CMD_SET_FREQUENCY:
params.frequency = command.int_value;
break;
}
}
if (resampler == NULL) {
resampler = inp.resampler;
audio_resampler = inp.audio_resampler;
} else if (resampler != inp.resampler) {
msresamp_crcf_destroy(resampler);
msresamp_crcf_destroy(audio_resampler);
resampler = inp.resampler;
audio_resampler = inp.audio_resampler;
}
if (bandwidthChanged) {
DemodulatorWorkerThreadCommand command(DemodulatorWorkerThreadCommand::DEMOD_WORKER_THREAD_CMD_BUILD_FILTERS);
command.audioSampleRate = bandwidthParams.audioSampleRate;
command.bandwidth = bandwidthParams.bandwidth;
command.frequency = bandwidthParams.frequency;
command.inputRate = bandwidthParams.inputRate;
int out_size = ceil((float) (bufSize) * inp.resample_ratio);
liquid_float_complex resampled_data[out_size];
unsigned int num_written;
msresamp_crcf_execute(resampler, &inp.data[0], bufSize, resampled_data, &num_written);
float audio_resample_ratio = inp.audio_resample_ratio;
workerQueue->push(command);
}
}
float demod_output[num_written];
if (!initialized) {
continue;
}
freqdem_demodulate_block(fdem, resampled_data, num_written, demod_output);
// Requested frequency is not center, shift it into the center!
if (inp.frequency != params.frequency) {
if ((params.frequency - inp.frequency) != shift_freq) {
shift_freq = params.frequency - inp.frequency;
if (abs(shift_freq) <= (int) ((float) (SRATE / 2) * 1.5)) {
nco_crcf_set_frequency(nco_shift, (2.0 * M_PI) * (((float) abs(shift_freq)) / ((float) SRATE)));
}
}
}
liquid_float_complex demod_audio_data[num_written];
if (abs(shift_freq) > (int) ((float) (SRATE / 2) * 1.5)) {
continue;
}
for (int i = 0; i < num_written; i++) {
demod_audio_data[i].real = demod_output[i];
demod_audio_data[i].imag = 0;
}
std::vector<signed char> *data = &inp.data;
if (data->size()) {
liquid_float_complex *temp_buf;
int audio_out_size = ceil((float) (num_written) * audio_resample_ratio);
liquid_float_complex resampled_audio_output[audio_out_size];
for (int i = 0; i < BUF_SIZE / 2; i++) {
in_buf[i].real = (float) (*data)[i * 2] / 127.0f;
in_buf[i].imag = (float) (*data)[i * 2 + 1] / 127.0f;
}
unsigned int num_audio_written;
msresamp_crcf_execute(audio_resampler, demod_audio_data, num_written, resampled_audio_output, &num_audio_written);
if (shift_freq != 0) {
if (shift_freq < 0) {
nco_crcf_mix_block_up(nco_shift, in_buf, out_buf, BUF_SIZE / 2);
} else {
nco_crcf_mix_block_down(nco_shift, in_buf, out_buf, BUF_SIZE / 2);
}
temp_buf = in_buf;
in_buf = out_buf;
out_buf = temp_buf;
}
std::vector<float> newBuffer;
newBuffer.resize(num_audio_written * 2);
for (int i = 0; i < num_audio_written; i++) {
liquid_float_complex y = resampled_audio_output[i];
firfilt_crcf_execute_block(fir_filter, in_buf, BUF_SIZE / 2, out_buf);
newBuffer[i * 2] = y.real;
newBuffer[i * 2 + 1] = y.real;
}
int out_size = ceil((float) (BUF_SIZE / 2) * resample_ratio);
AudioThreadInput ati;
ati.data = newBuffer;
liquid_float_complex resampled_output[out_size];
float demod_output[out_size];
if (audioInputQueue != NULL) {
audioInputQueue->push(ati);
}
unsigned int num_written; // number of values written to buffer
msresamp_crcf_execute(resampler, out_buf, (BUF_SIZE / 2), resampled_output, &num_written);
if (visOutQueue != NULL) {
visOutQueue->push(ati);
}
}
freqdem_demodulate_block(fdem, resampled_output, num_written, demod_output);
if (resampler != NULL) {
msresamp_crcf_destroy(resampler);
}
if (audio_resampler != NULL) {
msresamp_crcf_destroy(audio_resampler);
}
for (int i = 0; i < num_written; i++) {
resampled_output[i].real = demod_output[i];
resampled_output[i].imag = 0;
}
int audio_out_size = ceil((float) (num_written) * audio_resample_ratio);
liquid_float_complex resampled_audio_output[audio_out_size];
unsigned int num_audio_written;
msresamp_crcf_execute(audio_resampler, resampled_output, num_written, resampled_audio_output, &num_audio_written);
std::vector<float> newBuffer;
newBuffer.resize(num_audio_written * 2);
for (int i = 0; i < num_audio_written; i++) {
liquid_float_complex y = resampled_audio_output[i];
newBuffer[i * 2] = y.real;
newBuffer[i * 2 + 1] = y.real;
}
AudioThreadInput ati;
ati.data = newBuffer;
if (audioInputQueue != NULL) {
audioInputQueue->push(ati);
}
if (visOutQueue != NULL) {
visOutQueue->push(ati);
}
}
if (!workerResults->empty()) {
while (!workerResults->empty()) {
DemodulatorWorkerThreadResult result;
workerResults->pop(result);
switch (result.cmd) {
case DemodulatorWorkerThreadResult::DEMOD_WORKER_THREAD_RESULT_FILTERS:
firfilt_crcf_destroy(fir_filter);
msresamp_crcf_destroy(resampler);
msresamp_crcf_destroy(audio_resampler);
fir_filter = result.fir_filter;
resampler = result.resampler;
audio_resampler = result.audio_resampler;
resample_ratio = result.resample_ratio;
audio_resample_ratio = result.audio_resample_ratio;
params.audioSampleRate = result.audioSampleRate;
params.bandwidth = result.bandwidth;
params.inputRate = result.inputRate;
break;
}
}
}
}
delete in_buf;
delete out_buf;
std::cout << "Demodulator thread done." << std::endl;
DemodulatorThreadCommand tCmd(DemodulatorThreadCommand::DEMOD_THREAD_CMD_DEMOD_TERMINATED);
tCmd.context = this;
threadQueueNotify->push(tCmd);
std::cout << "Demodulator thread done." << std::endl;
DemodulatorThreadCommand tCmd(DemodulatorThreadCommand::DEMOD_THREAD_CMD_DEMOD_TERMINATED);
tCmd.context = this;
threadQueueNotify->push(tCmd);
}
void DemodulatorThread::terminate() {
terminated = true;
DemodulatorThreadIQData inp; // push dummy to nudge queue
inputQueue->push(inp);
workerThread->terminate();
terminated = true;
DemodulatorThreadPostIQData inp; // push dummy to nudge queue
postInputQueue->push(inp);
}

63
src/demod/DemodulatorThread.h
View File

@ -2,45 +2,16 @@
#include <queue>
#include <vector>
#include "wx/wxprec.h"
#ifndef WX_PRECOMP
#include "wx/wx.h"
#endif
#include "wx/thread.h"
#include "liquid/liquid.h"
#include "CubicSDRDefs.h"
#include "DemodulatorWorkerThread.h"
#include "DemodDefs.h"
class DemodulatorThreadParameters {
public:
unsigned int frequency;
unsigned int inputRate;
unsigned int bandwidth; // set equal to disable second stage re-sampling?
unsigned int audioSampleRate;
DemodulatorType demodType;
DemodulatorThreadParameters() :
frequency(0), inputRate(SRATE), bandwidth(200000), audioSampleRate(
AUDIO_FREQUENCY), demodType(DEMOD_TYPE_FM) {
}
~DemodulatorThreadParameters() {
}
};
#include "AudioThread.h"
typedef ThreadQueue<AudioThreadInput> DemodulatorThreadOutputQueue;
class DemodulatorThread {
public:
DemodulatorThread(DemodulatorThreadInputQueue* pQueue, DemodulatorThreadCommandQueue* threadQueueNotify);
DemodulatorThread(DemodulatorThreadPostInputQueue* pQueueIn, DemodulatorThreadCommandQueue* threadQueueNotify);
~DemodulatorThread();
#ifdef __APPLE__
@ -53,18 +24,10 @@ public:
visOutQueue = tQueue;
}
void setCommandQueue(DemodulatorThreadCommandQueue *tQueue) {
commandQueue = tQueue;
}
void setAudioInputQueue(AudioThreadInputQueue *tQueue) {
audioInputQueue = tQueue;
}
DemodulatorThreadParameters &getParams() {
return params;
}
void initialize();
void terminate();
@ -76,33 +39,13 @@ public:
#endif
protected:
DemodulatorThreadInputQueue* inputQueue;
DemodulatorThreadPostInputQueue* postInputQueue;
DemodulatorThreadOutputQueue* visOutQueue;
DemodulatorThreadCommandQueue* commandQueue;
AudioThreadInputQueue *audioInputQueue;
firfilt_crcf fir_filter;
msresamp_crcf resampler;
float resample_ratio;
msresamp_crcf audio_resampler;
float audio_resample_ratio;
DemodulatorThreadParameters params;
DemodulatorThreadParameters last_params;
freqdem fdem;
nco_crcf nco_shift;
int shift_freq;
std::atomic<bool> terminated;
std::atomic<bool> initialized;
DemodulatorWorkerThread *workerThread;
std::thread *t_Worker;
DemodulatorThreadWorkerCommandQueue *workerQueue;
DemodulatorThreadWorkerResultQueue *workerResults;
DemodulatorThreadCommandQueue* threadQueueNotify;
};

7
src/sdr/SDRPostThread.cpp
View File

@ -42,6 +42,13 @@ void SDRPostThread::threadMain() {
int n_read;
double seconds = 0.0;
//#ifdef __APPLE__
// pthread_t tID = pthread_self(); // ID of this thread
// int priority = sched_get_priority_min( SCHED_RR );
// sched_param prio = { priority }; // scheduling priority of thread
// pthread_setschedparam( tID, SCHED_RR, &prio );
//#endif
dcFilter = iirfilt_crcf_create_dc_blocker(0.0005);
liquid_float_complex x, y;

6
src/sdr/SDRThread.cpp
View File

@ -89,6 +89,12 @@ int SDRThread::enumerate_rtl() {
}
void SDRThread::threadMain() {
//#ifdef __APPLE__
// pthread_t tID = pthread_self(); // ID of this thread
// int priority = sched_get_priority_min( SCHED_RR );
// sched_param prio = { priority }; // scheduling priority of thread
// pthread_setschedparam( tID, SCHED_RR, &prio );
//#endif
std::cout << "SDR thread initializing.." << std::endl;

5
src/visual/PrimaryGLContext.cpp
View File

@ -54,7 +54,10 @@ void PrimaryGLContext::CheckGLError() {
PrimaryGLContext::PrimaryGLContext(wxGLCanvas *canvas, wxGLContext *sharedContext) :
wxGLContext(canvas, sharedContext) {
SetCurrent(*canvas);
// Pre-load fonts
for (int i = 0; i < GLFONT_MAX; i++) {
getFont((GLFontSize)i);
}
CheckGLError();
}

6
src/visual/WaterfallCanvas.cpp
View File

@ -444,7 +444,7 @@ void WaterfallCanvas::mouseReleased(wxMouseEvent& event) {
demod->run();
wxGetApp().bindDemodulator(demod);
wxGetApp().getDemodMgr().setActiveDemodulator(demod);
wxGetApp().getDemodMgr().setActiveDemodulator(demod,false);
}
if (demod == NULL) {
@ -474,7 +474,7 @@ void WaterfallCanvas::mouseReleased(wxMouseEvent& event) {
int freq = center_freq - (int) (0.5 * (float) SRATE) + (int) ((float) pos * (float) SRATE);
wxGetApp().getDemodMgr().setActiveDemodulator(wxGetApp().getDemodMgr().getActiveDemodulator(), false);
nextDragState = WF_DRAG_NONE;
nextDragState = WF_DRAG_FREQUENCY;
}
} else if (dragState == WF_DRAG_RANGE) {
float width = mTracker.getOriginDeltaMouseX();
@ -503,7 +503,7 @@ void WaterfallCanvas::mouseReleased(wxMouseEvent& event) {
demod->run();
wxGetApp().bindDemodulator(demod);
wxGetApp().getDemodMgr().setActiveDemodulator(demod);
wxGetApp().getDemodMgr().setActiveDemodulator(demod,false);
}
if (demod == NULL) {