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

Make ReBuffer refcount management automatic, using std::shared_ptr: 

- No longer need to call setRefCount() / decRefCount()
- Restore old ThreadBlockingQueue using std::deque, for std::shared_ptr correct behaviour.
This commit is contained in:
vsonnier 2017-05-24 12:24:36 +02:00
parent 8776908340
commit 4b323e9863
32 changed files with 289 additions and 409 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/IOThread.cpp
View File

@ -4,9 +4,6 @@
#include "IOThread.h"
#include <typeinfo>
std::mutex ReBufferGC::g_mutex;
std::set<ReferenceCounter *> ReBufferGC::garbage;
#define SPIN_WAIT_SLEEP_MS 5
IOThread::IOThread() {

216
src/IOThread.h
View File

@ -5,13 +5,14 @@
#include <mutex>
#include <atomic>
#include <deque>
#include <vector>
#include <map>
#include <set>
#include <string>
#include <iostream>
#include <thread>
#include <memory>
#include <climits>
#include "ThreadBlockingQueue.h"
#include "Timer.h"
@ -23,163 +24,122 @@ struct map_string_less : public std::binary_function<std::string,std::string,boo
}
};
class ReferenceCounter {
template <typename PtrType>
class ReBufferAge {
public:
PtrType ptr;
int age;
//default constructor, initialized with refcont 1, sounds very natural
ReferenceCounter() {
refCount = 1;
}
// void setIndex(int idx) {
// std::lock_guard < std::recursive_mutex > lock(m_mutex);
// index = idx;
// }
// int getIndex() {
// std::lock_guard < std::recursive_mutex > lock(m_mutex);
// return index;
// }
void setRefCount(int rc) {
std::lock_guard < std::recursive_mutex > lock(m_mutex);
refCount = rc;
}
void decRefCount() {
std::lock_guard < std::recursive_mutex > lock(m_mutex);
refCount--;
}
int getRefCount() {
std::lock_guard < std::recursive_mutex > lock(m_mutex);
return refCount;
}
// Access to the own mutex protecting the ReferenceCounter, i.e the monitor of the class
std::recursive_mutex& getMonitor() const {
return m_mutex;
}
protected:
//this is a basic mutex for all ReferenceCounter derivatives operations INCLUDING the counter itself for consistency !
mutable std::recursive_mutex m_mutex;
private:
int refCount;
// int index;
virtual ~ReBufferAge() {};
};
#define REBUFFER_GC_LIMIT 100
#define REBUFFER_WARNING_THRESHOLD 150
class ReBufferGC {
public:
static void garbageCollect() {
std::lock_guard < std::mutex > lock(g_mutex);
std::deque<ReferenceCounter *> garbageRemoval;
for (typename std::set<ReferenceCounter *>::iterator i = garbage.begin(); i != garbage.end(); i++) {
if ((*i)->getRefCount() <= 0) {
garbageRemoval.push_back(*i);
}
else {
// std::cout << "Garbage in queue buffer idx #" << (*i)->getIndex() << ", " << (*i)->getRefCount() << " usage(s)" << std::endl;
std::cout << "Garbage in queue buffer with " << (*i)->getRefCount() << " usage(s)" << std::endl;
}
}
if ( garbageRemoval.size() ) {
std::cout << "Garbage collecting " << garbageRemoval.size() << " ReBuffer(s)" << std::endl;
while (!garbageRemoval.empty()) {
ReferenceCounter *ref = garbageRemoval.back();
garbageRemoval.pop_back();
garbage.erase(ref);
delete ref;
}
}
}
static void addGarbage(ReferenceCounter *ref) {
std::lock_guard < std::mutex > lock(g_mutex);
garbage.insert(ref);
}
private:
static std::mutex g_mutex;
static std::set<ReferenceCounter *> garbage;
};
template<class BufferType = ReferenceCounter>
template<typename BufferType>
class ReBuffer {
typedef typename std::shared_ptr<BufferType> ReBufferPtr;
public:
ReBuffer(std::string bufferId) : bufferId(bufferId) {
// indexCounter.store(0);
}
BufferType *getBuffer() {
/// Return a new ReBuffer_ptr usable by the application.
ReBufferPtr getBuffer() {
std::lock_guard < std::mutex > lock(m_mutex);
BufferType* buf = nullptr;
for (outputBuffersI = outputBuffers.begin(); outputBuffersI != outputBuffers.end(); outputBuffersI++) {
if (buf == nullptr && (*outputBuffersI)->getRefCount() <= 0) {
buf = (*outputBuffersI);
buf->setRefCount(1);
} else if ((*outputBuffersI)->getRefCount() <= 0) {
(*outputBuffersI)->decRefCount();
// iterate the ReBuffer_ptr list: if the std::shared_ptr count == 1, it means
//it is only referenced in outputBuffers itself, so available for re-use.
//else if the std::shared_ptr count <= 1, make it age.
//else the ReBuffer_ptr is in use, don't use it.
ReBufferPtr buf = nullptr;
outputBuffersI it = outputBuffers.begin();
while (it != outputBuffers.end()) {
long use = it->ptr.use_count();
//1. If we encounter a shared_ptr with a use count of 0, this
//is a bug since it is supposed to be at least 1, because it is referenced here.
//in this case, purge it from here and trace.
if (use == 0) {
it = outputBuffers.erase(it);
std::cout << "Warning: in ReBuffer '" << bufferId << "' count '" << outputBuffers.size() << "', found 1 dangling buffer !" << std::endl << std::flush;
}
else if (use == 1) {
if (buf == nullptr) {
it->age = 1; //select this one.
buf = it->ptr;
//std::cout << "**" << std::flush;
it++;
}
else {
//make the other unused buffers age
it->age--;
it++;
}
}
}
if (buf != nullptr) {
if (outputBuffers.back()->getRefCount() < -REBUFFER_GC_LIMIT) {
BufferType *ref = outputBuffers.back();
else {
it++;
}
} //end while
//2.1 Garbage collect the oldest (last element) if it aged too much, and return the buffer
if (buf != nullptr) {
if (outputBuffers.back().age < -REBUFFER_GC_LIMIT) {
//by the nature of the shared_ptr, memory will ne deallocated automatically.
outputBuffers.pop_back();
delete ref;
//std::cout << "--" << std::flush;
}
// buf->setIndex(indexCounter++);
return buf;
}
#define REBUFFER_WARNING_THRESHOLD 100
if (outputBuffers.size() > REBUFFER_WARNING_THRESHOLD) {
std::cout << "Warning: ReBuffer '" << bufferId << "' count '" << outputBuffers.size() << "' exceeds threshold of '" << REBUFFER_WARNING_THRESHOLD << "'" << std::endl;
}
//by default created with refcount = 1
buf = new BufferType();
// buf->setIndex(indexCounter++);
outputBuffers.push_back(buf);
return buf;
//3.We need to allocate a new buffer.
ReBufferAge < ReBufferPtr > newBuffer;
//careful here: newBuffer.ptr is already constructed, so we need to set "in place" its
//ownership to a (new BufferType()).
newBuffer.ptr.reset(new BufferType());
newBuffer.age = 1;
outputBuffers.push_back(newBuffer);
//std::cout << "++" << std::flush;
return newBuffer.ptr;
}
/// Purge the cache.
void purge() {
std::lock_guard < std::mutex > lock(m_mutex);
// if (bufferId == "DemodulatorThreadBuffers") {
// std::cout << "'" << bufferId << "' purging.. total indexes: " << indexCounter.load() << std::endl;
// }
while (!outputBuffers.empty()) {
BufferType *ref = outputBuffers.front();
outputBuffers.pop_front();
if (ref->getRefCount() <= 0) {
delete ref;
} else {
// Something isn't done with it yet; throw it on the pile.. keep this as a bug indicator for now..
std::cout << "'" << bufferId << "' pushed garbage.." << std::endl;
ReBufferGC::addGarbage(ref);
}
}
// since outputBuffers are full std::shared_ptr,
//purging if will effectively loose the local reference,
// so the std::shared_ptr will naturally be deallocated
//when their time comes.
outputBuffers.clear();
}
private:
private:
//name of the buffer cache kind
std::string bufferId;
std::deque<BufferType*> outputBuffers;
typename std::deque<BufferType*>::iterator outputBuffersI;
//the ReBuffer cache
std::vector< ReBufferAge < ReBufferPtr > > outputBuffers;
typedef typename std::vector< ReBufferAge < ReBufferPtr > >::iterator outputBuffersI;
//mutex protecting access to outputBuffers.
mutable std::mutex m_mutex;
// std::atomic_int indexCounter;
};

33
src/audio/AudioThread.cpp
View File

@ -123,7 +123,7 @@ static int audioCallback(void *outputBuffer, void * /* inputBuffer */, unsigned
if (srcmix->currentInput->sampleRate == src->getSampleRate()) {
break;
}
srcmix->currentInput->decRefCount();
}
srcmix->currentInput = nullptr;
} //end while
@ -140,7 +140,7 @@ static int audioCallback(void *outputBuffer, void * /* inputBuffer */, unsigned
if (!srcmix->inputQueue->empty()) {
srcmix->audioQueuePtr = 0;
if (srcmix->currentInput) {
srcmix->currentInput->decRefCount();
srcmix->currentInput = nullptr;
}
@ -160,7 +160,7 @@ static int audioCallback(void *outputBuffer, void * /* inputBuffer */, unsigned
if (srcmix->audioQueuePtr >= srcmix->currentInput->data.size()) {
srcmix->audioQueuePtr = 0;
if (srcmix->currentInput) {
srcmix->currentInput->decRefCount();
srcmix->currentInput = nullptr;
}
@ -187,7 +187,7 @@ static int audioCallback(void *outputBuffer, void * /* inputBuffer */, unsigned
if (srcmix->audioQueuePtr >= srcmix->currentInput->data.size()) {
srcmix->audioQueuePtr = 0;
if (srcmix->currentInput) {
srcmix->currentInput->decRefCount();
srcmix->currentInput = nullptr;
}
@ -444,20 +444,13 @@ void AudioThread::run() {
std::lock_guard<std::recursive_mutex> lock(m_mutex);
// Drain any remaining inputs, with a non-blocking pop
AudioThreadInput *ref;
while (inputQueue && inputQueue->try_pop(ref)) {
if (ref) {
ref->decRefCount();
}
} //end while
if (inputQueue != nullptr) {
inputQueue->flush();
}
//Nullify currentInput...
if (currentInput) {
currentInput->setRefCount(0);
currentInput = nullptr;
}
currentInput = nullptr;
//Stop
if (deviceController[parameters.deviceId] != this) {
deviceController[parameters.deviceId]->removeThread(this);
@ -494,7 +487,6 @@ void AudioThread::setActive(bool state) {
std::lock_guard<std::recursive_mutex> lock(m_mutex);
AudioThreadInput *dummy;
if (state && !active && inputQueue) {
deviceController[parameters.deviceId]->bindThread(this);
} else if (!state && active) {
@ -504,12 +496,7 @@ void AudioThread::setActive(bool state) {
// Activity state changing, clear any inputs
if(inputQueue) {
while (inputQueue->try_pop(dummy)) { // flush queue, non-blocking pop
if (dummy) {
dummy->decRefCount();
}
}
inputQueue->flush();
}
active = state;
}

12
src/audio/AudioThread.h
View File

@ -8,12 +8,12 @@
#include <map>
#include <string>
#include <atomic>
#include <memory>
#include "ThreadBlockingQueue.h"
#include "RtAudio.h"
#include "DemodDefs.h"
class AudioThreadInput: public ReferenceCounter {
class AudioThreadInput {
public:
long long frequency;
int inputRate;
@ -29,10 +29,12 @@ public:
}
~AudioThreadInput() {
std::lock_guard < std::recursive_mutex > lock(m_mutex);
}
};
typedef std::shared_ptr<AudioThreadInput> AudioThreadInputPtr;
class AudioThreadCommand {
public:
enum AudioThreadCommandEnum {
@ -47,12 +49,12 @@ public:
int int_value;
};
typedef ThreadBlockingQueue<AudioThreadInput *> AudioThreadInputQueue;
typedef ThreadBlockingQueue<AudioThreadInputPtr> AudioThreadInputQueue;
typedef ThreadBlockingQueue<AudioThreadCommand> AudioThreadCommandQueue;
class AudioThread : public IOThread {
public:
AudioThreadInput *currentInput;
AudioThreadInputPtr currentInput;
AudioThreadInputQueue *inputQueue;
std::atomic_uint audioQueuePtr;
std::atomic_uint underflowCount;

21
src/demod/DemodDefs.h
View File

@ -9,6 +9,7 @@
#include <vector>
#include <atomic>
#include <mutex>
#include <memory>
#include "IOThread.h"
@ -29,7 +30,7 @@ public:
std::string demodType;
};
class DemodulatorThreadIQData: public ReferenceCounter {
class DemodulatorThreadIQData {
public:
long long frequency;
long long sampleRate;
@ -48,7 +49,7 @@ public:
return *this;
}
~DemodulatorThreadIQData() {
virtual ~DemodulatorThreadIQData() {
}
};
@ -56,7 +57,7 @@ public:
class Modem;
class ModemKit;
class DemodulatorThreadPostIQData: public ReferenceCounter {
class DemodulatorThreadPostIQData {
public:
std::vector<liquid_float_complex> data;
@ -71,13 +72,13 @@ public:
}
~DemodulatorThreadPostIQData() {
std::lock_guard < std::recursive_mutex > lock(m_mutex);
virtual ~DemodulatorThreadPostIQData() {
}
};
class DemodulatorThreadAudioData: public ReferenceCounter {
class DemodulatorThreadAudioData {
public:
long long frequency;
unsigned int sampleRate;
@ -95,11 +96,13 @@ public:
}
~DemodulatorThreadAudioData() {
virtual ~DemodulatorThreadAudioData() {
}
};
typedef std::shared_ptr<DemodulatorThreadIQData> DemodulatorThreadIQDataPtr;
typedef std::shared_ptr<DemodulatorThreadPostIQData> DemodulatorThreadPostIQDataPtr;
typedef ThreadBlockingQueue<DemodulatorThreadIQData *> DemodulatorThreadInputQueue;
typedef ThreadBlockingQueue<DemodulatorThreadPostIQData *> DemodulatorThreadPostInputQueue;
typedef ThreadBlockingQueue< DemodulatorThreadIQDataPtr > DemodulatorThreadInputQueue;
typedef ThreadBlockingQueue< DemodulatorThreadPostIQDataPtr > DemodulatorThreadPostInputQueue;
typedef ThreadBlockingQueue<DemodulatorThreadControlCommand> DemodulatorThreadControlCommandQueue;

28
src/demod/DemodulatorMgr.cpp
View File

@ -228,7 +228,6 @@ void DemodulatorMgr::setActiveDemodulator(DemodulatorInstance *demod, bool tempo
} else {
std::lock_guard < std::recursive_mutex > lock(demods_busy);
garbageCollect();
ReBufferGC::garbageCollect();
}
if (activeVisualDemodulator.load()) {
@ -283,23 +282,24 @@ DemodulatorInstance *DemodulatorMgr::getLastDemodulatorWith(const std::string& t
//Private internal method, no need to protect it with demods_busy
void DemodulatorMgr::garbageCollect() {
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);
std::vector<DemodulatorInstance *>::iterator it = demods_deleted.begin();
std::cout << "Garbage collected demodulator instance " << deleted->getLabel() << std::endl;
while (it != demods_deleted.end()) {
delete deleted;
return;
}
if ((*it)->isTerminated()) {
DemodulatorInstance *deleted = (*it);
delete deleted;
it = demods_deleted.erase(it);
std::cout << "Garbage collected demodulator instance '" << deleted->getLabel() << "'... " << std::endl << std::flush;
}
}
else {
it++;
}
} //end while
}
void DemodulatorMgr::updateLastState() {

17
src/demod/DemodulatorPreThread.cpp
View File

@ -71,7 +71,7 @@ void DemodulatorPreThread::run() {
t_Worker = new std::thread(&DemodulatorWorkerThread::threadMain, workerThread);
while (!stopping) {
DemodulatorThreadIQData *inp;
DemodulatorThreadIQDataPtr inp;
iqInputQueue->pop(inp);
@ -157,7 +157,7 @@ void DemodulatorPreThread::run() {
}
if (cModem && cModemKit && abs(shiftFrequency) > (int) ((double) (inp->sampleRate / 2) * 1.5)) {
inp->decRefCount();
continue;
}
@ -192,7 +192,7 @@ void DemodulatorPreThread::run() {
out_buf = temp_buf;
}
DemodulatorThreadPostIQData *resamp = buffers.getBuffer();
DemodulatorThreadPostIQDataPtr resamp = buffers.getBuffer();
size_t out_size = ceil((double) (bufSize) * iqResampleRatio) + 512;
@ -218,8 +218,6 @@ void DemodulatorPreThread::run() {
iqOutputQueue->push(resamp);
}
inp->decRefCount();
DemodulatorWorkerThreadResult result;
//process all worker results until
while (!stopping && workerResults->try_pop(result)) {
@ -277,11 +275,8 @@ void DemodulatorPreThread::run() {
}
} //end while stopping
DemodulatorThreadPostIQData *tmp;
while (iqOutputQueue->try_pop(tmp)) {
tmp->decRefCount();
}
iqOutputQueue->flush();
buffers.purge();
}
@ -348,7 +343,7 @@ int DemodulatorPreThread::getAudioSampleRate() {
void DemodulatorPreThread::terminate() {
IOThread::terminate();
DemodulatorThreadIQData *inp = new DemodulatorThreadIQData; // push dummy to nudge queue
DemodulatorThreadIQDataPtr inp(new DemodulatorThreadIQData); // push dummy to nudge queue
//VSO: blocking push :
iqInputQueue->push(inp);

2
src/demod/DemodulatorPreThread.h
View File

@ -17,7 +17,7 @@ class DemodulatorPreThread : public IOThread {
public:
DemodulatorPreThread(DemodulatorInstance *parent);
~DemodulatorPreThread();
virtual ~DemodulatorPreThread();
virtual void run();

46
src/demod/DemodulatorThread.cpp
View File

@ -79,15 +79,14 @@ void DemodulatorThread::run() {
ModemIQData modemData;
while (!stopping) {
DemodulatorThreadPostIQData *inp;
DemodulatorThreadPostIQDataPtr inp;
iqInputQueue->pop(inp);
// std::lock_guard < std::mutex > lock(inp->m_mutex);
size_t bufSize = inp->data.size();
if (!bufSize) {
inp->decRefCount();
continue;
}
@ -104,7 +103,7 @@ void DemodulatorThread::run() {
}
if (!cModem || !cModemKit) {
inp->decRefCount();
continue;
}
@ -115,7 +114,7 @@ void DemodulatorThread::run() {
modemData.sampleRate = inp->sampleRate;
modemData.data.assign(inputData->begin(), inputData->end());
AudioThreadInput *ati = nullptr;
AudioThreadInputPtr ati = nullptr;
ModemAnalog *modemAnalog = (cModem->getType() == "analog")?((ModemAnalog *)cModem):nullptr;
ModemDigital *modemDigital = (cModem->getType() == "digital")?((ModemDigital *)cModem):nullptr;
@ -133,7 +132,7 @@ void DemodulatorThread::run() {
ati->data.resize(0);
}
cModem->demodulate(cModemKit, &modemData, ati);
cModem->demodulate(cModemKit, &modemData, ati.get());
double currentSignalLevel = 0;
double sampleTime = double(inp->data.size()) / double(inp->sampleRate);
@ -225,7 +224,6 @@ void DemodulatorThread::run() {
}
}
} else if (ati) {
ati->setRefCount(0);
ati = nullptr;
}
@ -238,7 +236,7 @@ void DemodulatorThread::run() {
}
if ((ati || modemDigital) && localAudioVisOutputQueue != nullptr && localAudioVisOutputQueue->empty()) {
AudioThreadInput *ati_vis = new AudioThreadInput;
AudioThreadInputPtr ati_vis(new AudioThreadInput);
ati_vis->sampleRate = inp->sampleRate;
ati_vis->inputRate = inp->sampleRate;
@ -246,7 +244,7 @@ void DemodulatorThread::run() {
size_t num_vis = DEMOD_VIS_SIZE;
if (modemDigital) {
if (ati) { // TODO: handle digital modems with audio output
ati->setRefCount(0);
ati = nullptr;
}
ati_vis->data.resize(inputData->size());
@ -300,7 +298,7 @@ void DemodulatorThread::run() {
if (!localAudioVisOutputQueue->try_push(ati_vis)) {
//non-blocking push needed for audio vis out
ati_vis->setRefCount(0);
std::cout << "DemodulatorThread::run() cannot push ati_vis into localAudioVisOutputQueue, is full !" << std::endl;
std::this_thread::yield();
}
@ -310,12 +308,10 @@ void DemodulatorThread::run() {
if (!muted.load() && (!wxGetApp().getSoloMode() || (demodInstance == wxGetApp().getDemodMgr().getLastActiveDemodulator()))) {
//non-blocking push needed for audio out
if (!audioOutputQueue->try_push(ati)) {
ati->decRefCount();
std::cout << "DemodulatorThread::run() cannot push ati into audioOutputQueue, is full !" << std::endl;
std::this_thread::yield();
}
} else {
ati->setRefCount(0);
}
}
@ -335,28 +331,12 @@ void DemodulatorThread::run() {
break;
}
}
inp->decRefCount();
}
// end while !stopping
// Purge any unused inputs, with a non-blocking pop
DemodulatorThreadPostIQData *ref;
while (iqInputQueue->try_pop(ref)) {
if (ref) { // May have other consumers; just decrement
ref->decRefCount();
}
}
AudioThreadInput *ref_audio;
while (audioOutputQueue->try_pop(ref_audio)) {
if (ref_audio) { // Originated here; set RefCount to 0
ref_audio->setRefCount(0);
}
}
iqInputQueue->flush();
audioOutputQueue->flush();
outputBuffers.purge();
@ -365,7 +345,7 @@ void DemodulatorThread::run() {
void DemodulatorThread::terminate() {
IOThread::terminate();
DemodulatorThreadPostIQData *inp = new DemodulatorThreadPostIQData; // push dummy to nudge queue
DemodulatorThreadPostIQDataPtr inp(new DemodulatorThreadPostIQData); // push dummy to nudge queue
//VSO: blocking push
iqInputQueue->push(inp);

4
src/demod/DemodulatorThread.h
View File

@ -10,7 +10,7 @@
#include "AudioThread.h"
#include "Modem.h"
typedef ThreadBlockingQueue<AudioThreadInput *> DemodulatorThreadOutputQueue;
typedef ThreadBlockingQueue<AudioThreadInputPtr> DemodulatorThreadOutputQueue;
#define DEMOD_VIS_SIZE 2048
#define DEMOD_SIGNAL_MIN -30
@ -22,7 +22,7 @@ class DemodulatorThread : public IOThread {
public:
DemodulatorThread(DemodulatorInstance *parent);
~DemodulatorThread();
virtual ~DemodulatorThread();
void onBindOutput(std::string name, ThreadQueueBase *threadQueue);

2
src/demod/DemodulatorWorkerThread.h
View File

@ -76,7 +76,7 @@ class DemodulatorWorkerThread : public IOThread {
public:
DemodulatorWorkerThread();
~DemodulatorWorkerThread();
virtual ~DemodulatorWorkerThread();
virtual void run();

9
src/modules/modem/Modem.h
View File

@ -8,6 +8,7 @@
#include "AudioThread.h"
#include <cmath>
#include <atomic>
#include <memory>
#define MIN_BANDWIDTH 500
@ -25,7 +26,7 @@ public:
int audioSampleRate;
};
class ModemIQData: public ReferenceCounter {
class ModemIQData {
public:
std::vector<liquid_float_complex> data;
long long sampleRate;
@ -34,11 +35,13 @@ public:
}
~ModemIQData() {
std::lock_guard < std::recursive_mutex > lock(m_mutex);
virtual ~ModemIQData() {
}
};
typedef std::shared_ptr<ModemIQData> ModemIQDataPtr;
// Copy of SoapySDR::Range, original comments
class ModemRange
{

4
src/modules/modem/analog/ModemAM.cpp
View File

@ -24,13 +24,13 @@ int ModemAM::getDefaultSampleRate() {
return 6000;
}
void ModemAM::demodulate(ModemKit *kit, ModemIQData *input, AudioThreadInput *audioOut) {
void ModemAM::demodulate(ModemKit *kit, ModemIQData *input, AudioThreadInput* audioOut) {
ModemKitAnalog *amkit = (ModemKitAnalog *)kit;
initOutputBuffers(amkit,input);
if (!bufSize) {
input->decRefCount();
return;
}

2
src/modules/modem/analog/ModemDSB.cpp
View File

@ -30,7 +30,7 @@ void ModemDSB::demodulate(ModemKit *kit, ModemIQData *input, AudioThreadInput *a
initOutputBuffers(amkit, input);
if (!bufSize) {
input->decRefCount();
return;
}

2
src/modules/modem/analog/ModemFM.cpp
View File

@ -29,7 +29,7 @@ void ModemFM::demodulate(ModemKit *kit, ModemIQData *input, AudioThreadInput *au
initOutputBuffers(fmkit, input);
if (!bufSize) {
input->decRefCount();
return;
}

2
src/modules/modem/analog/ModemIQ.cpp
View File

@ -42,7 +42,7 @@ void ModemIQ::demodulate(ModemKit * /* kit */, ModemIQData *input, AudioThreadIn
size_t bufSize = input->data.size();
if (!bufSize) {
input->decRefCount();
return;
}

2
src/modules/modem/analog/ModemLSB.cpp
View File

@ -46,7 +46,7 @@ void ModemLSB::demodulate(ModemKit *kit, ModemIQData *input, AudioThreadInput *a
initOutputBuffers(akit,input);
if (!bufSize) {
input->decRefCount();
return;
}

2
src/modules/modem/analog/ModemNBFM.cpp
View File

@ -29,7 +29,7 @@ void ModemNBFM::demodulate(ModemKit *kit, ModemIQData *input, AudioThreadInput *
initOutputBuffers(fmkit, input);
if (!bufSize) {
input->decRefCount();
return;
}

2
src/modules/modem/analog/ModemUSB.cpp
View File

@ -46,7 +46,7 @@ void ModemUSB::demodulate(ModemKit *kit, ModemIQData *input, AudioThreadInput *a
initOutputBuffers(akit,input);
if (!bufSize) {
input->decRefCount();
return;
}

7
src/process/FFTDataDistributor.cpp
View File

@ -28,7 +28,7 @@ void FFTDataDistributor::process() {
if (!isAnyOutputEmpty()) {
return;
}
DemodulatorThreadIQData *inp;
DemodulatorThreadIQDataPtr inp;
input->pop(inp);
if (inp) {
@ -73,7 +73,7 @@ void FFTDataDistributor::process() {
memcpy(&inputBuffer.data[bufferOffset+bufferedItems],&inp->data[0], nbSamplesToAdd *sizeof(liquid_float_complex));
bufferedItems += nbSamplesToAdd;
//
inp->decRefCount();
} else {
//empty inp, wait for another.
continue;
@ -105,7 +105,8 @@ void FFTDataDistributor::process() {
if (lineRateAccum >= 1.0) {
//each i represents a FFT computation
DemodulatorThreadIQData *outp = outputBuffers.getBuffer();
DemodulatorThreadIQDataPtr outp = outputBuffers.getBuffer();
outp->frequency = inputBuffer.frequency;
outp->sampleRate = inputBuffer.sampleRate;
outp->data.assign(inputBuffer.data.begin()+bufferOffset+i,

13
src/process/ScopeVisualProcessor.cpp
View File

@ -47,7 +47,7 @@ void ScopeVisualProcessor::process() {
if (!isOutputEmpty()) {
return;
}
AudioThreadInput *audioInputData;
AudioThreadInputPtr audioInputData;
if (input->try_pop(audioInputData)) {
@ -56,11 +56,12 @@ void ScopeVisualProcessor::process() {
}
size_t i, iMax = audioInputData->data.size();
if (!iMax) {
delete audioInputData; //->decRefCount();
//discard audioInputData.
audioInputData = nullptr;
return;
}
ScopeRenderData *renderData = NULL;
ScopeRenderDataPtr renderData = nullptr;
if (scopeEnabled) {
iMax = audioInputData->data.size();
@ -150,7 +151,7 @@ void ScopeVisualProcessor::process() {
renderData->inputRate = audioInputData->inputRate;
renderData->sampleRate = audioInputData->sampleRate;
delete audioInputData; //->decRefCount();
audioInputData = nullptr; //->decRefCount();
double fft_ceil = 0, fft_floor = 1;
@ -212,8 +213,6 @@ void ScopeVisualProcessor::process() {
renderData->spectrum = true;
distribute(renderData);
} else {
delete audioInputData; //->decRefCount();
}
}
} //end if try_pop()
}

11
src/process/ScopeVisualProcessor.h
View File

@ -6,8 +6,9 @@
#include "VisualProcessor.h"
#include "AudioThread.h"
#include "ScopePanel.h"
#include <memory>
class ScopeRenderData: public ReferenceCounter {
class ScopeRenderData {
public:
std::vector<float> waveform_points;
ScopePanel::ScopeMode mode = ScopePanel::SCOPE_MODE_Y;
@ -17,9 +18,15 @@ public:
bool spectrum;
int fft_size;
double fft_floor, fft_ceil;
virtual ~ScopeRenderData() {
}
};
typedef ThreadBlockingQueue<ScopeRenderData *> ScopeRenderDataQueue;
typedef std::shared_ptr<ScopeRenderData> ScopeRenderDataPtr;
typedef ThreadBlockingQueue<ScopeRenderDataPtr> ScopeRenderDataQueue;
class ScopeVisualProcessor : public VisualProcessor<AudioThreadInput, ScopeRenderData> {
public:

16
src/process/SpectrumVisualProcessor.cpp
View File

@ -192,7 +192,7 @@ void SpectrumVisualProcessor::process() {
fftSizeChanged.store(false);
}
DemodulatorThreadIQData *iqData;
DemodulatorThreadIQDataPtr iqData;
input->pop(iqData);
@ -200,14 +200,8 @@ void SpectrumVisualProcessor::process() {
return;
}
//Start by locking concurrent access to iqData
std::lock_guard < std::recursive_mutex > lock(iqData->getMonitor());
//then get the busy_lock
std::lock_guard < std::mutex > busy_lock(busy_run);
bool doPeak = peakHold.load() && (peakReset.load() == 0);
@ -246,7 +240,6 @@ void SpectrumVisualProcessor::process() {
if (is_view.load()) {
if (!iqData->sampleRate) {
iqData->decRefCount();
return;
}
@ -387,7 +380,7 @@ void SpectrumVisualProcessor::process() {
}
if (execute) {
SpectrumVisualData *output = outputBuffers.getBuffer();
SpectrumVisualDataPtr output = outputBuffers.getBuffer();
if (output->spectrum_points.size() != fftSize * 2) {
output->spectrum_points.resize(fftSize * 2);
@ -597,10 +590,7 @@ void SpectrumVisualProcessor::process() {
distribute(output);
}
}
iqData->decRefCount();
}
lastView = is_view.load();
}

8
src/process/SpectrumVisualProcessor.h
View File

@ -6,20 +6,24 @@
#include "VisualProcessor.h"
#include "DemodDefs.h"
#include <cmath>
#include <memory>
#define SPECTRUM_VZM 2
#define PEAK_RESET_COUNT 30
class SpectrumVisualData : public ReferenceCounter {
class SpectrumVisualData {
public:
std::vector<float> spectrum_points;
std::vector<float> spectrum_hold_points;
double fft_ceiling, fft_floor;
long long centerFreq;
int bandwidth;
virtual ~SpectrumVisualData() {};
};
typedef ThreadBlockingQueue<SpectrumVisualData *> SpectrumVisualDataQueue;
typedef std::shared_ptr<SpectrumVisualData> SpectrumVisualDataPtr;
typedef ThreadBlockingQueue<SpectrumVisualDataPtr> SpectrumVisualDataQueue;
class SpectrumVisualProcessor : public VisualProcessor<DemodulatorThreadIQData, SpectrumVisualData> {
public:

46
src/process/VisualProcessor.h
View File

@ -9,11 +9,16 @@
#include <algorithm>
#include <vector>
template<typename InputDataType = ReferenceCounter, typename OutputDataType = ReferenceCounter>
template<typename InputDataType, typename OutputDataType>
class VisualProcessor {
//
typedef ThreadBlockingQueue<InputDataType*> VisualInputQueueType;
typedef ThreadBlockingQueue<OutputDataType*> VisualOutputQueueType;
typedef std::shared_ptr<InputDataType> InputDataTypePtr;
typedef std::shared_ptr<OutputDataType> OutputDataTypePtr;
typedef ThreadBlockingQueue<InputDataTypePtr> VisualInputQueueType;
typedef ThreadBlockingQueue<OutputDataTypePtr> VisualOutputQueueType;
typedef typename std::vector< VisualOutputQueueType *>::iterator outputs_i;
public:
virtual ~VisualProcessor() {
@ -96,22 +101,19 @@ protected:
//available outputs, previously set by attachOutput().
//* \param[in] timeout The number of microseconds to wait to push an item in each one of the outputs, 0(default) means indefinite wait.
//* \param[in] errorMessage an error message written on std::cout in case pf push timeout.
void distribute(OutputDataType *item, std::uint64_t timeout = BLOCKING_INFINITE_TIMEOUT, const char* errorMessage = "") {
void distribute(OutputDataTypePtr item, std::uint64_t timeout = BLOCKING_INFINITE_TIMEOUT, const char* errorMessage = "") {
std::lock_guard < std::recursive_mutex > busy_lock(busy_update);
//We will try to distribute 'output' among all 'outputs',
//so 'output' will a-priori be shared among all 'outputs' so set its ref count to this
//amount.
item->setRefCount((int)outputs.size());
//so 'output' will a-priori be shared among all 'outputs'.
for (outputs_i it = outputs.begin(); it != outputs.end(); it++) {
//if 'output' failed to be given to an outputs_i, dec its ref count accordingly.
//blocking push, with a timeout
//'output' can fail to be given to an outputs_i,
//using a blocking push, with a timeout
if (!(*it)->push(item, timeout, errorMessage)) {
item->decRefCount();
//TODO : trace ?
}
}
// Now 'item' refcount matches the times 'item' has been successfully distributed,
//i.e shared among the outputs.
}
//the incoming data queue
@ -127,7 +129,7 @@ protected:
//Specialization much like VisualDataReDistributor, except
//the input (pointer) is directly re-dispatched
//to outputs, so that all output indeed SHARE the same instance.
template<typename OutputDataType = ReferenceCounter>
template<typename OutputDataType>
class VisualDataDistributor : public VisualProcessor<OutputDataType, OutputDataType> {
protected:
virtual void process() {
@ -136,18 +138,14 @@ protected:
if (!VisualProcessor<OutputDataType, OutputDataType>::isAnyOutputEmpty()) {
if (inp) {
inp->decRefCount();
//nothing
}
return;
}
if (inp) {
int previousRefCount = inp->getRefCount();
VisualProcessor<OutputDataType, OutputDataType>::distribute(inp);
//inp is now shared through the distribute(), which overwrite the previous ref count,
//so increment it properly.
int distributeRefCount = inp->getRefCount();
inp->setRefCount(previousRefCount + distributeRefCount);
//inp is now shared through the distribute() call.
}
}
}
@ -155,7 +153,7 @@ protected:
//specialization class which process() take an input item and re-dispatch
//A COPY to every outputs, without further processing. This is a 1-to-n dispatcher.
template<typename OutputDataType = ReferenceCounter>
template<typename OutputDataType>
class VisualDataReDistributor : public VisualProcessor<OutputDataType, OutputDataType> {
protected:
virtual void process() {
@ -164,15 +162,17 @@ protected:
if (!VisualProcessor<OutputDataType, OutputDataType>::isAnyOutputEmpty()) {
if (inp) {
inp->decRefCount();
//nothing
}
return;
}
if (inp) {
OutputDataType *outp = buffers.getBuffer();
OutputDataTypePtr outp = buffers.getBuffer();
//'deep copy of the contents
(*outp) = (*inp);
inp->decRefCount();
VisualProcessor<OutputDataType, OutputDataType>::distribute(outp);
}
}

30
src/sdr/SDRPostThread.cpp
View File

@ -183,7 +183,7 @@ void SDRPostThread::run() {
iqActiveDemodVisualQueue = static_cast<DemodulatorThreadInputQueue*>(getOutputQueue("IQActiveDemodVisualDataOutput"));
while (!stopping) {
SDRThreadIQData *data_in;
SDRThreadIQDataPtr data_in;
iqDataInQueue->pop(data_in);
// std::lock_guard < std::mutex > lock(data_in->m_mutex);
@ -192,14 +192,14 @@ void SDRPostThread::run() {
if (data_in && data_in->data.size()) {
if(data_in->numChannels > 1) {
runPFBCH(data_in);
runPFBCH(data_in.get());
} else {
runSingleCH(data_in);
runSingleCH(data_in.get());
}
}
if (data_in) {
data_in->decRefCount();
//nothing
}
bool doUpdate = false;
@ -217,20 +217,19 @@ void SDRPostThread::run() {
} //end while
//Be safe, remove as many elements as possible
DemodulatorThreadIQData *visualDataDummy;
while (iqVisualQueue && iqVisualQueue->try_pop(visualDataDummy)) {
visualDataDummy->decRefCount();
if (iqVisualQueue) {
iqVisualQueue->flush();
}
// buffers.purge();
// visualDataBuffers.purge();
buffers.purge();
visualDataBuffers.purge();
// std::cout << "SDR post-processing thread done." << std::endl;
}
void SDRPostThread::terminate() {
IOThread::terminate();
SDRThreadIQData *dummy = new SDRThreadIQData;
SDRThreadIQDataPtr dummy(new SDRThreadIQData);
//VSO: blocking push
iqDataInQueue->push(dummy);
}
@ -278,8 +277,8 @@ void SDRPostThread::runSingleCH(SDRThreadIQData *data_in) {
}
if (refCount) {
DemodulatorThreadIQData *demodDataOut = buffers.getBuffer();
demodDataOut->setRefCount(refCount);
DemodulatorThreadIQDataPtr demodDataOut = buffers.getBuffer();
demodDataOut->frequency = frequency;
demodDataOut->sampleRate = sampleRate;
@ -333,7 +332,7 @@ void SDRPostThread::runPFBCH(SDRThreadIQData *data_in) {
}
if (iqDataOutQueue != NULL && !iqDataOutQueue->full()) {
DemodulatorThreadIQData *iqDataOut = visualDataBuffers.getBuffer();
DemodulatorThreadIQDataPtr iqDataOut = visualDataBuffers.getBuffer();
bool doVis = false;
@ -341,8 +340,6 @@ void SDRPostThread::runPFBCH(SDRThreadIQData *data_in) {
doVis = true;
}
iqDataOut->setRefCount(1 + (doVis?1:0));
iqDataOut->frequency = data_in->frequency;
iqDataOut->sampleRate = data_in->sampleRate;
iqDataOut->data.assign(data_in->data.begin(), data_in->data.begin() + dataSize);
@ -407,8 +404,7 @@ void SDRPostThread::runPFBCH(SDRThreadIQData *data_in) {
continue;
}
DemodulatorThreadIQData *demodDataOut = buffers.getBuffer();
demodDataOut->setRefCount(demodChannelActive[i] + doDemodVis);
DemodulatorThreadIQDataPtr demodDataOut = buffers.getBuffer();
demodDataOut->frequency = chanCenters[i];
demodDataOut->sampleRate = chanBw;

14
src/sdr/SoapySDRThread.cpp
View File

@ -207,17 +207,17 @@ void SDRThread::readStream(SDRThreadIQDataQueue* iqDataOutQueue) {
//TODO: Add in doc the need to reduce SoapySDR device buffer length (if available) to restore higher fps.
//0. Retreive a new batch
SDRThreadIQData *dataOut = buffers.getBuffer();
SDRThreadIQDataPtr dataOut = buffers.getBuffer();
//resize to the target size immedialetly, to minimize later reallocs:
assureBufferMinSize(dataOut, nElems);
assureBufferMinSize(dataOut.get(), nElems);
//1.If overflow occured on the previous readStream(), transfer it in dataOut directly.
if (numOverflow > 0) {
int n_overflow = std::min(numOverflow, nElems);
//safety
assureBufferMinSize(dataOut, n_overflow);
assureBufferMinSize(dataOut.get(), n_overflow);
::memcpy(&dataOut->data[0], &overflowBuffer.data[0], n_overflow * sizeof(liquid_float_complex));
n_read = n_overflow;
@ -266,7 +266,7 @@ void SDRThread::readStream(SDRThreadIQDataQueue* iqDataOutQueue) {
float *pp = (float *)buffs[0];
//safety
assureBufferMinSize(dataOut, n_read + n_requested);
assureBufferMinSize(dataOut.get(), n_read + n_requested);
if (iq_swap.load()) {
for (int i = 0; i < n_requested; i++) {
@ -315,7 +315,7 @@ void SDRThread::readStream(SDRThreadIQDataQueue* iqDataOutQueue) {
float *pp = (float *)buffs[0];
//safety
assureBufferMinSize(dataOut, n_read + n_stream_read);
assureBufferMinSize(dataOut.get(), n_read + n_stream_read);
if (iq_swap.load()) {
for (int i = 0; i < n_stream_read; i++) {
@ -349,8 +349,7 @@ void SDRThread::readStream(SDRThreadIQDataQueue* iqDataOutQueue) {
if (!iqDataOutQueue->try_push(dataOut)) {
//The rest of the system saturates,
//finally the push didn't suceeded, recycle dataOut immediatly.
dataOut->setRefCount(0);
//finally the push didn't suceeded.
std::cout << "SDRThread::readStream(): 3.2 iqDataOutQueue output queue is full, discard processing of the batch..." << std::endl;
@ -359,7 +358,6 @@ void SDRThread::readStream(SDRThreadIQDataQueue* iqDataOutQueue) {
}
}
else {
dataOut->setRefCount(0);
std::cout << "SDRThread::readStream(): 3.1 iqDataOutQueue output queue is full, discard processing of the batch..." << std::endl;
//saturation, let a chance to the other threads to consume the existing samples
std::this_thread::yield();

10
src/sdr/SoapySDRThread.h
View File

@ -4,7 +4,7 @@
#pragma once
#include <atomic>
#include <memory>
#include "ThreadBlockingQueue.h"
#include "DemodulatorMgr.h"
#include "SDRDeviceInfo.h"
@ -17,7 +17,7 @@
#include <stddef.h>
class SDRThreadIQData: public ReferenceCounter {
class SDRThreadIQData {
public:
long long frequency;
long long sampleRate;
@ -35,12 +35,12 @@ public:
}
~SDRThreadIQData() {
virtual ~SDRThreadIQData() {
}
};
typedef ThreadBlockingQueue<SDRThreadIQData *> SDRThreadIQDataQueue;
typedef std::shared_ptr<SDRThreadIQData> SDRThreadIQDataPtr;
typedef ThreadBlockingQueue<SDRThreadIQDataPtr> SDRThreadIQDataQueue;
class SDRThread : public IOThread {
private:

112
src/util/ThreadBlockingQueue.h
View File

@ -3,7 +3,7 @@
#pragma once
#include <vector>
#include <deque>
#include <mutex>
#include <thread>
#include <cstdint>
@ -29,24 +29,22 @@ class ThreadQueueBase {
template<typename T>
class ThreadBlockingQueue : public ThreadQueueBase {
typedef typename std::vector<T>::value_type value_type;
typedef typename std::vector<T>::size_type size_type;
typedef typename std::deque<T>::value_type value_type;
typedef typename std::deque<T>::size_type size_type;
public:
/*! Create safe blocking queue. */
ThreadBlockingQueue() {
//at least 1 (== Exchanger)
m_circular_buffer.resize(MIN_ITEM_NB);
m_max_num_items = MIN_ITEM_NB;
};
//Copy constructor
ThreadBlockingQueue(const ThreadBlockingQueue& sq) {
std::lock_guard < std::mutex > lock(sq.m_mutex);
m_circular_buffer = sq.m_circular_buffer;
m_head = sq.m_head;
m_tail = sq.m_tail;
m_size = sq.m_size;
m_queue = sq.m_queue;
m_max_num_items = sq.m_max_num_items;
}
/*! Destroy safe queue. */
@ -62,11 +60,10 @@ public:
void set_max_num_items(unsigned int max_num_items) {
std::lock_guard < std::mutex > lock(m_mutex);
if (max_num_items > (unsigned int)m_circular_buffer.size()) {
if (max_num_items > m_max_num_items) {
//Only raise the existing max size, never reduce it
//for simplification sake at runtime.
m_circular_buffer.resize(max_num_items);
//m_head and m_tail stays valid for the new size.
m_max_num_items = max_num_items;
m_cond_not_full.notify_all();
}
}
@ -86,14 +83,14 @@ public:
if (timeout == BLOCKING_INFINITE_TIMEOUT) {
m_cond_not_full.wait(lock, [this]() // Lambda funct
{
return m_size < m_circular_buffer.size();
return m_queue.size() < m_max_num_items;
});
} else if (timeout <= NON_BLOCKING_TIMEOUT && m_size >= m_circular_buffer.size()) {
} else if (timeout <= NON_BLOCKING_TIMEOUT && m_queue.size() >= m_max_num_items) {
// if the value is below a threshold, consider it is a try_push()
return false;
}
else if (false == m_cond_not_full.wait_for(lock, std::chrono::microseconds(timeout),
[this]() { return m_size < m_circular_buffer.size(); })) {
[this]() { return m_queue.size() < m_max_num_items; })) {
std::thread::id currentThreadId = std::this_thread::get_id();
std::cout << "WARNING: Thread 0x" << std::hex << currentThreadId << std::dec <<
" (" << currentThreadId << ") executing {" << typeid(*this).name() << "}.push() has failed with timeout > " <<
@ -101,11 +98,7 @@ public:
return false;
}
//m_tail is already the next valid place an item can be put
m_circular_buffer[m_tail] = item;
m_tail = nextIndex(m_tail, (int)m_circular_buffer.size());
m_size++;
m_queue.push_back(item);
m_cond_not_empty.notify_all();
return true;
}
@ -118,14 +111,11 @@ public:
bool try_push(const value_type& item) {
std::lock_guard < std::mutex > lock(m_mutex);
if (m_size >= m_circular_buffer.size()) {
if (m_queue.size() >= m_max_num_items) {
return false;
}
//m_tail is already the next valid place an item can be put
m_circular_buffer[m_tail] = item;
m_tail = nextIndex(m_tail, (int)m_circular_buffer.size());
m_size++;
m_queue.push_back(item);
m_cond_not_empty.notify_all();
return true;
}
@ -142,14 +132,14 @@ public:
if (timeout == BLOCKING_INFINITE_TIMEOUT) {
m_cond_not_empty.wait(lock, [this]() // Lambda funct
{
return m_size > 0;
return !m_queue.empty();
});
} else if (timeout <= NON_BLOCKING_TIMEOUT && m_size == 0) {
} else if (timeout <= NON_BLOCKING_TIMEOUT && m_queue.empty()) {
// if the value is below a threshold, consider it is try_pop()
return false;
}
else if (false == m_cond_not_empty.wait_for(lock, std::chrono::microseconds(timeout),
[this]() { return m_size > 0; })) {
[this]() { return !m_queue.empty(); })) {
std::thread::id currentThreadId = std::this_thread::get_id();
std::cout << "WARNING: Thread 0x" << std::hex << currentThreadId << std::dec <<
" (" << currentThreadId << ") executing {" << typeid(*this).name() << "}.pop() has failed with timeout > " <<
@ -157,9 +147,8 @@ public:
return false;
}
item = m_circular_buffer[m_head];
m_head = nextIndex(m_head, (int)m_circular_buffer.size());
m_size--;
item = m_queue.front();
m_queue.pop_front();
m_cond_not_full.notify_all();
return true;
}
@ -172,13 +161,12 @@ public:
bool try_pop(value_type& item) {
std::lock_guard < std::mutex > lock(m_mutex);
if (m_size == 0) {
if (m_queue.empty()) {
return false;
}
item = m_circular_buffer[m_head];
m_head = nextIndex(m_head, (int)m_circular_buffer.size());
m_size--;
item = m_queue.front();
m_queue.pop_front();
m_cond_not_full.notify_all();
return true;
}
@ -190,7 +178,7 @@ public:
*/
size_type size() const {
std::lock_guard < std::mutex > lock(m_mutex);
return m_size;
return m_queue.size();
}
/**
@ -199,7 +187,7 @@ public:
*/
bool empty() const {
std::lock_guard < std::mutex > lock(m_mutex);
return m_size == 0;
return m_queue.empty();
}
/**
@ -208,7 +196,7 @@ public:
*/
bool full() const {
std::lock_guard < std::mutex > lock(m_mutex);
return (m_size >= m_circular_buffer.size());
return (m_queue.size() >= m_max_num_items);
}
/**
@ -216,9 +204,7 @@ public:
*/
void flush() {
std::lock_guard < std::mutex > lock(m_mutex);
m_head = 0;
m_tail = 0;
m_size = 0;
m_queue.clear();
m_cond_not_full.notify_all();
}
@ -230,25 +216,22 @@ public:
if (this != &sq) {
std::lock_guard < std::mutex > lock1(m_mutex);
std::lock_guard < std::mutex > lock2(sq.m_mutex);
m_circular_buffer.swap(sq.m_circular_buffer);
m_queue.swap(sq.m_queue);
std::swap(m_max_num_items, sq.m_max_num_items);
std::swap(m_head, sq.m_head);
std::swap(m_tail, sq.m_tail);
std::swap(m_size, sq.m_size);
if (m_size > 0) {
if (!m_queue.empty()) {
m_cond_not_empty.notify_all();
}
if (sq.m_size > 0) {
if (!sq.m_queue.empty()) {
sq.m_cond_not_empty.notify_all();
}
if (m_size < m_circular_buffer.size()) {
if (!m_queue.full()) {
m_cond_not_full.notify_all();
}
if (sq.m_size < sq.m_circular_buffer.size()) {
if (!sq.m_queue.full()) {
sq.m_cond_not_full.notify_all();
}
}
@ -260,17 +243,14 @@ public:
std::lock_guard < std::mutex > lock1(m_mutex);
std::lock_guard < std::mutex > lock2(sq.m_mutex);
m_circular_buffer = sq.m_circular_buffer;
m_queue = sq.m_queue;
m_max_num_items = sq.m_max_num_items;
m_head = sq.m_head;
m_tail = sq.m_tail;
m_size = sq.m_size;
if (m_size > 0) {
if (!m_queue.empty()) {
m_cond_not_empty.notify_all();
}
if (m_size < m_circular_buffer.size()) {
if (!m_queue.full()) {
m_cond_not_full.notify_all();
}
}
@ -278,27 +258,13 @@ public:
}
private:
/// use a circular buffer structure to prevent allocations / reallocations (fixed array + modulo)
std::vector<T> m_circular_buffer;
/**
* The 'head' index of the element at the head of the deque, 'tail'
* the next (valid !) index at which an element can be pushed.
* m_head == m_tail means empty.
*/
int m_head = 0, m_tail = 0;
//hold the current number of elements.
size_type m_size = 0;
//
inline int nextIndex(int index, int mod) const {
return (index + 1 == mod) ? 0 : index + 1;
}
//TODO: use a circular buffer structure ? (fixed array + modulo)
std::deque<T> m_queue;
mutable std::mutex m_mutex;
std::condition_variable m_cond_not_empty;
std::condition_variable m_cond_not_full;
size_t m_max_num_items = MIN_ITEM_NB;
};
/*! Swaps the contents of two ThreadBlockingQueue objects. (external operator) */

7
src/visual/ScopeCanvas.cpp
View File

@ -104,7 +104,7 @@ void ScopeCanvas::OnPaint(wxPaintEvent& WXUNUSED(event)) {
wxPaintDC dc(this);
const wxSize ClientSize = GetClientSize();
ScopeRenderData *avData;
ScopeRenderDataPtr avData;
while (inputData.try_pop(avData)) {
@ -113,7 +113,7 @@ void ScopeCanvas::OnPaint(wxPaintEvent& WXUNUSED(event)) {
if (avData->waveform_points.size()) {
scopePanel.setPoints(avData->waveform_points);
}
avData->decRefCount();
} else {
if (avData->waveform_points.size()) {
spectrumPanel.setPoints(avData->waveform_points);
@ -124,8 +124,7 @@ void ScopeCanvas::OnPaint(wxPaintEvent& WXUNUSED(event)) {
spectrumPanel.setFFTSize(avData->fft_size);
spectrumPanel.setShowDb(showDb);
}
avData->decRefCount();
}
}

3
src/visual/SpectrumCanvas.cpp
View File

@ -54,7 +54,7 @@ void SpectrumCanvas::OnPaint(wxPaintEvent& WXUNUSED(event)) {
wxPaintDC dc(this);
const wxSize ClientSize = GetClientSize();
SpectrumVisualData *vData;
SpectrumVisualDataPtr vData;
if (visualDataQueue.try_pop(vData)) {
if (vData) {
@ -62,7 +62,6 @@ void SpectrumCanvas::OnPaint(wxPaintEvent& WXUNUSED(event)) {
spectrumPanel.setPeakPoints(vData->spectrum_hold_points);
spectrumPanel.setFloorValue(vData->fft_floor);
spectrumPanel.setCeilValue(vData->fft_ceiling);
vData->decRefCount();
}
}

12
src/visual/WaterfallCanvas.cpp
View File

@ -97,7 +97,7 @@ void WaterfallCanvas::processInputQueue() {
if (linesPerSecond) {
if (lpsIndex >= targetVis) {
while (lpsIndex >= targetVis) {
SpectrumVisualData *vData;
SpectrumVisualDataPtr vData;
if (visualDataQueue.try_pop(vData)) {
@ -106,7 +106,7 @@ void WaterfallCanvas::processInputQueue() {
waterfallPanel.setPoints(vData->spectrum_points);
}
waterfallPanel.step();
vData->decRefCount();
updated = true;
}
lpsIndex-=targetVis;
@ -915,13 +915,7 @@ void WaterfallCanvas::setLinesPerSecond(int lps) {
linesPerSecond = lps;
//empty all
SpectrumVisualData *vData;
while (visualDataQueue.try_pop(vData)) {
if (vData) {
vData->decRefCount();
}
}
visualDataQueue.flush();
}
void WaterfallCanvas::setMinBandwidth(int min) {