Deep redesign: phase #2

CMakeLists.txt

@@ -111,6 +111,7 @@ set(sdrbase_SOURCES
 
 	sdrbase/util/message.cpp
 	sdrbase/util/messagequeue.cpp
+	sdrbase/util/syncmessenger.cpp
 	#sdrbase/util/miniz.cpp
 	sdrbase/util/simpleserializer.cpp
 	sdrbase/util/spinlock.cpp
@@ -185,6 +186,7 @@ set(sdrbase_HEADERS
 	include/util/export.h
 	include/util/message.h
 	include/util/messagequeue.h
+	include/util/syncmessenger.h
 	#include/util/miniz.h
 	include/util/simpleserializer.h
 	include/util/spinlock.h

Readme.md

@@ -131,6 +131,7 @@ Done since the fork
   - File sample source plugin (recording reader) not working
   - Make the DSP engine global static
   - Fixed startup initialization sequence. New initialization phase in DSP engine and new ready state
+  - Message queuing and handling redesign
     
 =====
 To Do

include-gpl/dsp/channelizer.h

@@ -12,15 +12,16 @@ class SDRANGELOVE_API Channelizer : public SampleSink {
 	Q_OBJECT
 public:
 	Channelizer(SampleSink* sampleSink);
-	~Channelizer();
+	virtual ~Channelizer();
 
 	void configure(MessageQueue* messageQueue, int sampleRate, int centerFrequency);
 	int getInputSampleRate() const { return m_inputSampleRate; }
 
-	void feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly);
+	virtual bool init(const Message& cmd);
-	void start();
+	virtual void feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly);
-	void stop();
+	virtual void start();
-	bool handleMessage(Message* cmd);
+	virtual void stop();
+	virtual bool handleMessage(const Message& cmd);
 
 protected:
 	struct FilterStage {

include-gpl/dsp/dspcommands.h

@@ -133,144 +133,115 @@ class SDRANGELOVE_API DSPConfigureSpectrumVis : public Message {
 	MESSAGE_CLASS_DECLARATION
 
 public:
-	int getFFTSize() const { return m_fftSize; }
-	int getOverlapPercent() const { return m_overlapPercent; }
-	FFTWindow::Function getWindow() const { return m_window; }
-
-	static DSPConfigureSpectrumVis* create(int fftSize, int overlapPercent, FFTWindow::Function window)
-	{
-		return new DSPConfigureSpectrumVis(fftSize, overlapPercent, window);
-	}
-
-private:
-	int m_fftSize;
-	int m_overlapPercent;
-	FFTWindow::Function m_window;
-
 	DSPConfigureSpectrumVis(int fftSize, int overlapPercent, FFTWindow::Function window) :
 		Message(),
 		m_fftSize(fftSize),
 		m_overlapPercent(overlapPercent),
 		m_window(window)
 	{ }
+
+	int getFFTSize() const { return m_fftSize; }
+	int getOverlapPercent() const { return m_overlapPercent; }
+	FFTWindow::Function getWindow() const { return m_window; }
+
+private:
+	int m_fftSize;
+	int m_overlapPercent;
+	FFTWindow::Function m_window;
 };
 
 class SDRANGELOVE_API DSPConfigureCorrection : public Message {
 	MESSAGE_CLASS_DECLARATION
 
 public:
-	bool getDCOffsetCorrection() const { return m_dcOffsetCorrection; }
-	bool getIQImbalanceCorrection() const { return m_iqImbalanceCorrection; }
-
-	static DSPConfigureCorrection* create(bool dcOffsetCorrection, bool iqImbalanceCorrection)
-	{
-		return new DSPConfigureCorrection(dcOffsetCorrection, iqImbalanceCorrection);
-	}
-
-private:
-	bool m_dcOffsetCorrection;
-	bool m_iqImbalanceCorrection;
-
 	DSPConfigureCorrection(bool dcOffsetCorrection, bool iqImbalanceCorrection) :
 		Message(),
 		m_dcOffsetCorrection(dcOffsetCorrection),
 		m_iqImbalanceCorrection(iqImbalanceCorrection)
 	{ }
+
+	bool getDCOffsetCorrection() const { return m_dcOffsetCorrection; }
+	bool getIQImbalanceCorrection() const { return m_iqImbalanceCorrection; }
+
+private:
+	bool m_dcOffsetCorrection;
+	bool m_iqImbalanceCorrection;
+
 };
 
 class SDRANGELOVE_API DSPEngineReport : public Message {
 	MESSAGE_CLASS_DECLARATION
 
 public:
-	int getSampleRate() const { return m_sampleRate; }
-	quint64 getCenterFrequency() const { return m_centerFrequency; }
-
-	static DSPEngineReport* create(int sampleRate, quint64 centerFrequency)
-	{
-		return new DSPEngineReport(sampleRate, centerFrequency);
-	}
-
-private:
-	int m_sampleRate;
-	quint64 m_centerFrequency;
-
 	DSPEngineReport(int sampleRate, quint64 centerFrequency) :
 		Message(),
 		m_sampleRate(sampleRate),
 		m_centerFrequency(centerFrequency)
 	{ }
+
+	int getSampleRate() const { return m_sampleRate; }
+	quint64 getCenterFrequency() const { return m_centerFrequency; }
+
+private:
+	int m_sampleRate;
+	quint64 m_centerFrequency;
 };
 
 class SDRANGELOVE_API DSPConfigureScopeVis : public Message {
 	MESSAGE_CLASS_DECLARATION
 
 public:
-	int getTriggerChannel() const { return m_triggerChannel; }
-	Real getTriggerLevelHigh() const { return m_triggerLevelHigh; }
-	Real getTriggerLevelLow() const { return m_triggerLevelLow; }
-
-	static DSPConfigureScopeVis* create(int triggerChannel, Real triggerLevelHigh, Real triggerLevelLow)
-	{
-		return new DSPConfigureScopeVis(triggerChannel, triggerLevelHigh, triggerLevelLow);
-	}
-
-private:
-	int m_triggerChannel;
-	Real m_triggerLevelHigh;
-	Real m_triggerLevelLow;
-
 	DSPConfigureScopeVis(int triggerChannel, Real triggerLevelHigh, Real triggerLevelLow) :
 		Message(),
 		m_triggerChannel(triggerChannel),
 		m_triggerLevelHigh(triggerLevelHigh),
 		m_triggerLevelLow(triggerLevelLow)
 	{ }
+
+	int getTriggerChannel() const { return m_triggerChannel; }
+	Real getTriggerLevelHigh() const { return m_triggerLevelHigh; }
+	Real getTriggerLevelLow() const { return m_triggerLevelLow; }
+
+private:
+	int m_triggerChannel;
+	Real m_triggerLevelHigh;
+	Real m_triggerLevelLow;
 };
 
 class SDRANGELOVE_API DSPSignalNotification : public Message {
 	MESSAGE_CLASS_DECLARATION
 
 public:
-	int getSampleRate() const { return m_sampleRate; }
-	qint64 getFrequencyOffset() const { return m_frequencyOffset; }
-
-	static DSPSignalNotification* create(int sampleRate, quint64 frequencyOffset)
-	{
-		return new DSPSignalNotification(sampleRate, frequencyOffset);
-	}
-
-private:
-	int m_sampleRate;
-	qint64 m_frequencyOffset;
-
 	DSPSignalNotification(int samplerate, qint64 frequencyOffset) :
 		Message(),
 		m_sampleRate(samplerate),
 		m_frequencyOffset(frequencyOffset)
 	{ }
+
+	int getSampleRate() const { return m_sampleRate; }
+	qint64 getFrequencyOffset() const { return m_frequencyOffset; }
+
+private:
+	int m_sampleRate;
+	qint64 m_frequencyOffset;
 };
 
 class SDRANGELOVE_API DSPConfigureChannelizer : public Message {
 	MESSAGE_CLASS_DECLARATION
 
 public:
-	int getSampleRate() const { return m_sampleRate; }
-	int getCenterFrequency() const { return m_centerFrequency; }
-
-	static DSPConfigureChannelizer* create(int sampleRate, int centerFrequency)
-	{
-		return new DSPConfigureChannelizer(sampleRate, centerFrequency);
-	}
-
-private:
-	int m_sampleRate;
-	int m_centerFrequency;
-
 	DSPConfigureChannelizer(int sampleRate, int centerFrequency) :
 		Message(),
 		m_sampleRate(sampleRate),
 		m_centerFrequency(centerFrequency)
 	{ }
+
+	int getSampleRate() const { return m_sampleRate; }
+	int getCenterFrequency() const { return m_centerFrequency; }
+
+private:
+	int m_sampleRate;
+	int m_centerFrequency;
 };
 
 #endif // INCLUDE_DSPCOMMANDS_H

include-gpl/dsp/dspengine.h

@@ -27,6 +27,7 @@
 #include "dsp/samplefifo.h"
 #include "audio/audiooutput.h"
 #include "util/messagequeue.h"
+#include "util/syncmessenger.h"
 #include "util/export.h"
 
 class SampleSource;
@@ -78,6 +79,7 @@ public:
 private:
 	MessageQueue m_inputMessageQueue;  //<! Input message queue. Post here.
 	MessageQueue m_outputMessageQueue; //<! Output message queue. Listen here.
+	SyncMessenger m_syncMessenger;     //!< Used to process messages synchronously with the thread
 
 	State m_state;
 
@@ -113,11 +115,11 @@ private:
 	State gotoError(const QString& errorMsg); //!< Go to an error state
 
 	void handleSetSource(SampleSource* source); //!< Manage source setting
-	bool distributeMessage(Message* message); // FIXME: remove ?
 
 private slots:
 	void handleData(); //!< Handle data when samples from source FIFO are ready to be processed
 	void handleInputMessages(); //!< Handle input message queue
+	void handleSynchronousMessages(Message *message); //!< Handle synchronous messages with the thread
 };
 
 #endif // INCLUDE_DSPENGINE_H

include-gpl/dsp/filesink.h

@@ -26,19 +26,17 @@ public:
     
     quint64 getByteCount() const { return m_byteCount; }
 
-	void configure(MessageQueue* msgQueue, const std::string& filename, int sampleRate, quint64 centerFrequency);
+	void configure(MessageQueue* msgQueue, const std::string& filename);
 
-	virtual bool init(Message* cmd);
+	virtual bool init(const Message& cmd);
 	virtual void feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly);
 	virtual void start();
 	virtual void stop();
-	virtual bool handleMessage(Message* message);
+	virtual bool handleMessage(const Message& message);
     void startRecording();
     void stopRecording();
     static void readHeader(std::ifstream& samplefile, Header& header);
 
-    MessageQueue *getMessageQueue() { return m_messageQueue; }
-
 private:
 	class MsgConfigureFileSink : public Message {
 		MESSAGE_CLASS_DECLARATION
@@ -67,9 +65,8 @@ private:
     bool m_recordStart;
     std::ofstream m_sampleFile;
     quint64 m_byteCount;
-    MessageQueue m_messageQueue;
 
-	void handleConfigure(const std::string& fileName, int sampleRate, quint64 centerFrequency);
+	void handleConfigure(const std::string& fileName);
     void writeHeader();
 };
 

include-gpl/dsp/nullsink.h

@@ -4,17 +4,19 @@
 #include "dsp/samplesink.h"
 #include "util/export.h"
 
-class MessageQueue;
+class Message;
 
 class SDRANGELOVE_API NullSink : public SampleSink {
 public:
 
 	NullSink();
+	virtual ~NullSink();
 
-	void feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly);
+	virtual bool init(const Message& cmd);
-	void start();
+	virtual void feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly);
-	void stop();
+	virtual void start();
-	bool handleMessage(Message* message);
+	virtual void stop();
+	virtual bool handleMessage(const Message& message);
 };
 
 #endif // INCLUDE_NULLSINK_H

include-gpl/dsp/scopevis.h

@@ -23,6 +23,7 @@ public:
 	static const uint m_traceChunkSize;
 
 	ScopeVis(GLScope* glScope = NULL);
+	virtual ~ScopeVis();
 
 	void configure(MessageQueue* msgQueue, 
         TriggerChannel triggerChannel, 
@@ -34,11 +35,11 @@ public:
         uint traceSize);
 	void setOneShot(bool oneShot);
 
-	void feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly);
+	virtual bool init(const Message& cmd);
-	void start();
+	virtual void feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly);
-	void stop();
+	virtual void start();
-	bool handleMessageKeep(Message* message);
+	virtual void stop();
-	bool handleMessage(Message* message);
+	virtual bool handleMessage(const Message& message);
 
 	void setSampleRate(int sampleRate);
 	int getSampleRate() const { return m_sampleRate; }

include-gpl/dsp/spectrumscopecombovis.h

@@ -6,17 +6,19 @@
 #include "dsp/scopevis.h"
 #include "util/export.h"
 
-class MessageQueue;
+class Message;
 
 class SDRANGELOVE_API SpectrumScopeComboVis : public SampleSink {
 public:
 
 	SpectrumScopeComboVis(SpectrumVis* spectrumVis, ScopeVis* scopeVis);
+	virtual ~SpectrumScopeComboVis();
 
-	void feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly);
+	virtual bool init(const Message& cmd);
-	void start();
+	virtual void feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly);
-	void stop();
+	virtual void start();
-	bool handleMessage(Message* message);
+	virtual void stop();
+	virtual bool handleMessage(const Message& message);
 
 private:
 	SpectrumVis* m_spectrumVis;

include-gpl/dsp/spectrumvis.h

@@ -12,16 +12,16 @@ class MessageQueue;
 class SDRANGELOVE_API SpectrumVis : public SampleSink {
 public:
 	SpectrumVis(GLSpectrum* glSpectrum = NULL);
-	~SpectrumVis();
+	virtual ~SpectrumVis();
 
 	void configure(MessageQueue* msgQueue, int fftSize, int overlapPercent, FFTWindow::Function window);
 
-	void feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly);
+	virtual bool init(const Message& cmd);
+	virtual void feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly);
 	void feedTriggered(SampleVector::const_iterator triggerPoint, SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly);
-	void start();
+	virtual void start();
-	void stop();
+	virtual void stop();
-	bool handleMessageKeep(Message* message);
+	virtual bool handleMessage(const Message& message);
-	bool handleMessage(Message* message);
 
 private:
 	FFTEngine* m_fft;
@@ -30,11 +30,11 @@ private:
 	std::vector<Complex> m_fftBuffer;
 	std::vector<Real> m_logPowerSpectrum;
 
-	size_t m_fftSize;
+	std::size_t m_fftSize;
-	size_t m_overlapPercent;
+	std::size_t m_overlapPercent;
-	size_t m_overlapSize;
+	std::size_t m_overlapSize;
-	size_t m_refillSize;
+	std::size_t m_refillSize;
-	size_t m_fftBufferFill;
+	std::size_t m_fftBufferFill;
 	bool m_needMoreSamples;
 
 	GLSpectrum* m_glSpectrum;

include/dsp/samplesink.h

@@ -4,6 +4,7 @@
 #include <QObject>
 #include "dsptypes.h"
 #include "util/export.h"
+#include "util/messagequeue.h"
 
 class Message;
 
@@ -12,11 +13,18 @@ public:
 	SampleSink();
 	virtual ~SampleSink();
 
-	virtual bool init(Message* cmd) = 0;
+	virtual bool init(const Message& cmd) = 0;
 	virtual void feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly) = 0;
 	virtual void start() = 0;
 	virtual void stop() = 0;
-	virtual bool handleMessage(Message* cmd) = 0; //!< Handle message immediately or submit it to a queue
+	virtual bool handleMessage(const Message& cmd) = 0; //!< Processing of a message. Returns true if message has actually been processed
+
+	MessageQueue *getInputMessageQueue() { return &m_inputMessageQueue; }
+	MessageQueue *getOutputMessageQueue() { return &m_outputMessageQueue; }
+
+protected:
+	MessageQueue m_inputMessageQueue;
+	MessageQueue m_outputMessageQueue;
 };
 
 #endif // INCLUDE_SAMPLESINK_H

include/dsp/samplesource/samplesource.h

@@ -1,6 +1,6 @@
 ///////////////////////////////////////////////////////////////////////////////////
-// Copyright (C) 2012 maintech GmbH, Otto-Hahn-Str. 15, 97204 Hoechberg, Germany //
+// Copyright (C) 2015 F4EXB                                                      //
-// written by Christian Daniel                                                   //
+// written by Edouard Griffiths                                                  //
 //                                                                               //
 // This program is free software; you can redistribute it and/or modify          //
 // it under the terms of the GNU General Public License as published by          //
@@ -21,42 +21,32 @@
 #include <QtGlobal>
 #include "dsp/samplefifo.h"
 #include "util/message.h"
+#include "util/messagequeue.h"
 #include "util/export.h"
 
-class PluginGUI;
-class MessageQueue;
-
 class SDRANGELOVE_API SampleSource {
 public:
-	struct SDRANGELOVE_API GeneralSettings {
+	SampleSource();
-		quint64 m_centerFrequency;
-
-		GeneralSettings();
-		void resetToDefaults();
-		QByteArray serialize() const;
-		bool deserialize(const QByteArray& data);
-	};
-
-	SampleSource(MessageQueue* guiMessageQueue);
 	virtual ~SampleSource();
 
-	virtual void init(Message* cmd) = 0;
+	virtual void init(const Message& cmd) = 0;
-	virtual bool startInput(int device) = 0;
+	virtual bool start(int device) = 0;
-	virtual void stopInput() = 0;
+	virtual void stop() = 0;
 
 	virtual const QString& getDeviceDescription() const = 0;
 	virtual int getSampleRate() const = 0;
 	virtual quint64 getCenterFrequency() const = 0;
 
-	virtual bool handleMessage(Message* message) = 0;
+	virtual bool handleMessage(const Message& message) = 0;
-
+	
-	SampleFifo* getSampleFifo() { return &m_sampleFifo; }
+	MessageQueue *getInputMessageQueue() { return &m_inputMessageQueue; }
+	MessageQueue *getOutputMessageQueue() { return &m_outputMessageQueue; }
+    SampleFifo* getSampleFifo() { return &m_sampleFifo; }
 
 protected:
-	SampleFifo m_sampleFifo;
+    SampleFifo m_sampleFifo;
-	MessageQueue* m_guiMessageQueue;
+	MessageQueue m_inputMessageQueue;
-
+	MessageQueue m_outputMessageQueue;
-	GeneralSettings m_generalSettings;
 };
 
 #endif // INCLUDE_SAMPLESOURCE_H

include/util/message.h

@@ -1,14 +1,26 @@
+///////////////////////////////////////////////////////////////////////////////////
+// Copyright (C) 2015 F4EXB                                                      //
+// written by Edouard Griffiths                                                  //
+//                                                                               //
+// This program is free software; you can redistribute it and/or modify          //
+// it under the terms of the GNU General Public License as published by          //
+// the Free Software Foundation as version 3 of the License, or                  //
+//                                                                               //
+// This program is distributed in the hope that it will be useful,               //
+// but WITHOUT ANY WARRANTY; without even the implied warranty of                //
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the                  //
+// GNU General Public License V3 for more details.                               //
+//                                                                               //
+// You should have received a copy of the GNU General Public License             //
+// along with this program. If not, see <http://www.gnu.org/licenses/>.          //
+///////////////////////////////////////////////////////////////////////////////////
+
 #ifndef INCLUDE_MESSAGE_H
 #define INCLUDE_MESSAGE_H
 
 #include <stdlib.h>
-#include <QAtomicInt>
 #include "util/export.h"
 
-class MessageQueue;
-class QWaitCondition;
-class QMutex;
-
 class SDRANGELOVE_API Message {
 public:
 	Message();
@@ -16,33 +28,22 @@ public:
 
 	virtual const char* getIdentifier() const;
 	virtual bool matchIdentifier(const char* identifier) const;
-	static bool match(Message* message);
+	static bool match(const Message* message);
 
 	void* getDestination() const { return m_destination; }
-
+	void setDestination(void *destination) { m_destination = destination; }
-	void submit(MessageQueue* queue, void* destination = NULL);
-	int execute(MessageQueue* queue, void* destination = NULL);
-
-	void completed(int result = 0);
 
 protected:
 	// addressing
 	static const char* m_identifier;
 	void* m_destination;
-
-	// stuff for synchronous messages
-	bool m_synchronous;
-	QWaitCondition* m_waitCondition;
-	QMutex* m_mutex;
-	QAtomicInt m_complete;
-	int m_result;
 };
 
 #define MESSAGE_CLASS_DECLARATION \
 	public: \
 		const char* getIdentifier() const; \
 		bool matchIdentifier(const char* identifier) const; \
-		static bool match(Message* message); \
+		static bool match(const Message* message); \
 	protected: \
 		static const char* m_identifier; \
 	private:
@@ -53,6 +54,6 @@ protected:
 	bool Name::matchIdentifier(const char* identifier) const {\
 		return (m_identifier == identifier) ? true : BaseClass::matchIdentifier(identifier); \
 	} \
-	bool Name::match(Message* message) { return message->matchIdentifier(m_identifier); }
+	bool Name::match(const Message* message) { return message->matchIdentifier(m_identifier); }
 
 #endif // INCLUDE_MESSAGE_H

include/util/messagequeue.h

@@ -1,3 +1,20 @@
+///////////////////////////////////////////////////////////////////////////////////
+// Copyright (C) 2015 F4EXB                                                      //
+// written by Edouard Griffiths                                                  //
+//                                                                               //
+// This program is free software; you can redistribute it and/or modify          //
+// it under the terms of the GNU General Public License as published by          //
+// the Free Software Foundation as version 3 of the License, or                  //
+//                                                                               //
+// This program is distributed in the hope that it will be useful,               //
+// but WITHOUT ANY WARRANTY; without even the implied warranty of                //
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the                  //
+// GNU General Public License V3 for more details.                               //
+//                                                                               //
+// You should have received a copy of the GNU General Public License             //
+// along with this program. If not, see <http://www.gnu.org/licenses/>.          //
+///////////////////////////////////////////////////////////////////////////////////
+
 #ifndef INCLUDE_MESSAGEQUEUE_H
 #define INCLUDE_MESSAGEQUEUE_H
 
@@ -15,11 +32,11 @@ public:
 	MessageQueue(QObject* parent = NULL);
 	~MessageQueue();
 
-	void submit(Message* message);
+	void push(Message* message, bool emitSignal = true);  //!< Push message onto queue
-	Message* accept();
+	Message* pop(); //!< Pop message from queue
 
-	int countPending();
+	int size(); //!< Returns queue size
-	void clear();
+	void clear(); //!< Empty queue
 
 signals:
 	void messageEnqueued();

include/util/syncmessenger.h

@@ -0,0 +1,54 @@
+///////////////////////////////////////////////////////////////////////////////////
+// Copyright (C) 2015 F4EXB                                                      //
+// written by Edouard Griffiths                                                  //
+//                                                                               //
+// This program is free software; you can redistribute it and/or modify          //
+// it under the terms of the GNU General Public License as published by          //
+// the Free Software Foundation as version 3 of the License, or                  //
+//                                                                               //
+// This program is distributed in the hope that it will be useful,               //
+// but WITHOUT ANY WARRANTY; without even the implied warranty of                //
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the                  //
+// GNU General Public License V3 for more details.                               //
+//                                                                               //
+// You should have received a copy of the GNU General Public License             //
+// along with this program. If not, see <http://www.gnu.org/licenses/>.          //
+///////////////////////////////////////////////////////////////////////////////////
+
+#ifndef INCLUDE_UTIL_SYNCMESSENGER_H_
+#define INCLUDE_UTIL_SYNCMESSENGER_H_
+
+#include <QObject>
+#include <QWaitCondition>
+#include <QMutex>
+#include <QAtomicInt>
+#include "util/export.h"
+
+class Message;
+
+/**
+ * This class is responsible of managing the synchronous processing of a message across threads
+ */
+class SDRANGELOVE_API SyncMessenger : public QObject {
+	Q_OBJECT
+
+public:
+	SyncMessenger();
+	~SyncMessenger();
+
+	int sendWait(Message *message, unsigned long msPollTime = 100); //!< Send message and waits for its process completion
+	void done(int result = 0); //!< Processing of the message is complete
+
+signals:
+	void messageSent(Message *message);
+
+protected:
+	QWaitCondition m_waitCondition;
+	QMutex m_mutex;
+	QAtomicInt m_complete;
+	int m_result;
+};
+
+
+
+#endif /* INCLUDE_UTIL_SYNCMESSENGER_H_ */

sdrbase/dsp/channelizer.cpp

@@ -20,10 +20,26 @@ Channelizer::~Channelizer()
 	freeFilterChain();
 }
 
+bool Channelizer::init(const Message& cmd)
+{
+	if (DSPSignalNotification::match(&cmd))
+	{
+		DSPSignalNotification* notif = (DSPSignalNotification*) &cmd;
+		m_inputSampleRate = notif->getSampleRate();
+		qDebug() << "FileSink::init: DSPSignalNotification: m_inputSampleRate: " << m_inputSampleRate;
+		emit inputSampleRateChanged();
+		return true;
+	}
+	else
+	{
+		return false;
+	}
+}
+
 void Channelizer::configure(MessageQueue* messageQueue, int sampleRate, int centerFrequency)
 {
-	Message* cmd = DSPConfigureChannelizer::create(sampleRate, centerFrequency);
+	Message* cmd = new DSPConfigureChannelizer(sampleRate, centerFrequency);
-	cmd->submit(messageQueue, this);
+	messageQueue->push(cmd);
 }
 
 void Channelizer::feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly)
@@ -64,52 +80,58 @@ void Channelizer::stop()
 		m_sampleSink->stop();
 }
 
-bool Channelizer::handleMessage(Message* cmd)
+bool Channelizer::handleMessage(const Message& cmd)
 {
-	qDebug() << "Channelizer::handleMessage: " << cmd->getIdentifier();
+	qDebug() << "Channelizer::handleMessage: " << cmd.getIdentifier();
 
-	if (DSPSignalNotification::match(cmd))
+	/*
+	if (DSPSignalNotification::match(&cmd))
 	{
-		DSPSignalNotification* signal = (DSPSignalNotification*)cmd;
+		DSPSignalNotification* notif = (DSPSignalNotification*) &cmd;
-		m_inputSampleRate = signal->getSampleRate();
+		m_inputSampleRate = notif->getSampleRate();
 		qDebug() << "Channelizer::handleMessage: DSPSignalNotification: m_inputSampleRate: " << m_inputSampleRate;
 		applyConfiguration();
-		cmd->completed();
+
-		if(m_sampleSink != NULL)
+		delete cmd;
+
+		if (m_sampleSink != NULL)
 		{
-			signal = DSPSignalNotification::create(m_currentOutputSampleRate, m_currentCenterFrequency);
+			DSPSignalNotification notif(m_currentOutputSampleRate, m_currentCenterFrequency);
-			if(!m_sampleSink->handleMessage(signal)) {
+			m_sampleSink->handleMessage(notif))
-				signal->completed();
-			}
 		}
 
 		emit inputSampleRateChanged();
 		return true;
 	}
-	else if (DSPConfigureChannelizer::match(cmd))
+	else*/
+	if (DSPConfigureChannelizer::match(&cmd))
 	{
-		DSPConfigureChannelizer* chan = (DSPConfigureChannelizer*)cmd;
+		DSPConfigureChannelizer* chan = (DSPConfigureChannelizer*) &cmd;
 		m_requestedOutputSampleRate = chan->getSampleRate();
 		m_requestedCenterFrequency = chan->getCenterFrequency();
+
 		qDebug() << "Channelizer::handleMessage: DSPConfigureChannelizer:"
 				<< " m_requestedOutputSampleRate: " << m_requestedOutputSampleRate
 				<< " m_requestedCenterFrequency: " << m_requestedCenterFrequency;
+
 		applyConfiguration();
-		cmd->completed();
+
-		if(m_sampleSink != NULL)
+		if (m_sampleSink != NULL)
 		{
-			DSPSignalNotification* signal = DSPSignalNotification::create(m_currentOutputSampleRate, m_currentCenterFrequency);
+			DSPSignalNotification notif(m_currentOutputSampleRate, m_currentCenterFrequency);
-			if(!m_sampleSink->handleMessage(signal)) {
+			m_sampleSink->handleMessage(notif);
-				signal->completed();
-			}
 		}
+
 		return true;
 	}
 	else
 	{
-		if(m_sampleSink != NULL) {
+		if (m_sampleSink != NULL)
+		{
 			return m_sampleSink->handleMessage(cmd);
-		} else {
+		}
+		else
+		{
 			return false;
 		}
 	}

sdrbase/dsp/dspengine.cpp

@@ -58,6 +58,7 @@ void DSPEngine::run()
 	qDebug() << "DSPEngine::run";
 
 	connect(&m_inputMessageQueue, SIGNAL(messageEnqueued()), this, SLOT(handleInputMessages()), Qt::QueuedConnection);
+	connect(&m_syncMessenger, SIGNAL(messageSent(Message*)), this, SLOT(handleSynchronousMessages(Message*)), Qt::QueuedConnection);
 
 	m_state = StIdle;
 
@@ -70,34 +71,34 @@ void DSPEngine::start()
 	qDebug() << "DSPEngine::start";
 	DSPPing cmd;
 	QThread::start();
-	cmd.execute(&m_inputMessageQueue);
+	m_syncMessenger.sendWait(&cmd);
 }
 
 void DSPEngine::stop()
 {
 	qDebug() << "DSPEngine::stop";
 	DSPExit cmd;
-	cmd.execute(&m_inputMessageQueue);
+	m_syncMessenger.sendWait(&cmd);
 }
 
 bool DSPEngine::initAcquisition()
 {
 	DSPAcquisitionInit cmd;
 
-	return cmd.execute(&m_inputMessageQueue) == StReady;
+	return 	m_syncMessenger.sendWait(&cmd) == StReady;
 }
 
 bool DSPEngine::startAcquisition()
 {
 	DSPAcquisitionStart cmd;
 
-	return cmd.execute(&m_inputMessageQueue) == StRunning;
+	return m_syncMessenger.sendWait(&cmd) == StRunning;
 }
 
 void DSPEngine::stopAcquistion()
 {
 	DSPAcquisitionStop cmd;
-	cmd.execute(&m_inputMessageQueue);
+	m_syncMessenger.sendWait(&cmd);
 
 	if(m_dcOffsetCorrection)
 	{
@@ -108,51 +109,51 @@ void DSPEngine::stopAcquistion()
 void DSPEngine::setSource(SampleSource* source)
 {
 	DSPSetSource cmd(source);
-	cmd.execute(&m_inputMessageQueue);
+	m_syncMessenger.sendWait(&cmd);
 }
 
 void DSPEngine::addSink(SampleSink* sink)
 {
 	qDebug() << "DSPEngine::addSink: " << sink->objectName().toStdString().c_str();
 	DSPAddSink cmd(sink);
-	cmd.execute(&m_inputMessageQueue);
+	m_syncMessenger.sendWait(&cmd);
 }
 
 void DSPEngine::removeSink(SampleSink* sink)
 {
 	DSPRemoveSink cmd(sink);
-	cmd.execute(&m_inputMessageQueue);
+	m_syncMessenger.sendWait(&cmd);
 }
 
 void DSPEngine::addAudioSink(AudioFifo* audioFifo)
 {
 	DSPAddAudioSink cmd(audioFifo);
-	cmd.execute(&m_inputMessageQueue);
+	m_syncMessenger.sendWait(&cmd);
 }
 
 void DSPEngine::removeAudioSink(AudioFifo* audioFifo)
 {
 	DSPRemoveAudioSink cmd(audioFifo);
-	cmd.execute(&m_inputMessageQueue);
+	m_syncMessenger.sendWait(&cmd);
 }
 
 void DSPEngine::configureCorrections(bool dcOffsetCorrection, bool iqImbalanceCorrection)
 {
-	Message* cmd = DSPConfigureCorrection::create(dcOffsetCorrection, iqImbalanceCorrection);
+	DSPConfigureCorrection* cmd = new DSPConfigureCorrection(dcOffsetCorrection, iqImbalanceCorrection);
-	cmd->submit(&m_inputMessageQueue);
+	m_inputMessageQueue.push(cmd);
 }
 
 QString DSPEngine::errorMessage()
 {
 	DSPGetErrorMessage cmd;
-	cmd.execute(&m_inputMessageQueue);
+	m_syncMessenger.sendWait(&cmd);
 	return cmd.getErrorMessage();
 }
 
 QString DSPEngine::sourceDeviceDescription()
 {
 	DSPGetSourceDeviceDescription cmd;
-	cmd.execute(&m_inputMessageQueue);
+	m_syncMessenger.sendWait(&cmd);
 	return cmd.getDeviceDescription();
 }
 
@@ -229,17 +230,17 @@ void DSPEngine::imbalance(SampleVector::iterator begin, SampleVector::iterator e
 void DSPEngine::work()
 {
 	SampleFifo* sampleFifo = m_sampleSource->getSampleFifo();
-	size_t samplesDone = 0;
+	std::size_t samplesDone = 0;
 	bool positiveOnly = false;
 
-	while ((sampleFifo->fill() > 0) && (m_inputMessageQueue.countPending() == 0) && (samplesDone < m_sampleRate))
+	while ((sampleFifo->fill() > 0) && (m_inputMessageQueue.size() == 0) && (samplesDone < m_sampleRate))
 	{
 		SampleVector::iterator part1begin;
 		SampleVector::iterator part1end;
 		SampleVector::iterator part2begin;
 		SampleVector::iterator part2end;
 
-		size_t count = sampleFifo->readBegin(sampleFifo->fill(), &part1begin, &part1end, &part2begin, &part2end);
+		std::size_t count = sampleFifo->readBegin(sampleFifo->fill(), &part1begin, &part1end, &part2begin, &part2end);
 
 		// first part of FIFO data
 		if (part1begin != part1end)
@@ -278,7 +279,7 @@ void DSPEngine::work()
 		}
 
 		// adjust FIFO pointers
-		sampleFifo->readCommit(count);
+		sampleFifo->readCommit((unsigned int) count);
 		samplesDone += count;
 	}
 }
@@ -314,7 +315,7 @@ DSPEngine::State DSPEngine::gotoIdle()
 		(*it)->stop();
 	}
 
-	m_sampleSource->stopInput();
+	m_sampleSource->stop();
 	m_deviceDescription.clear();
 	m_audioSink.stop();
 	m_sampleRate = 0;
@@ -362,14 +363,14 @@ DSPEngine::State DSPEngine::gotoInit()
 	for(SampleSinks::const_iterator it = m_sampleSinks.begin(); it != m_sampleSinks.end(); it++)
 	{
 		qDebug() << "  - initializing " << (*it)->objectName().toStdString().c_str();
-		DSPSignalNotification* notif = DSPSignalNotification::create(m_sampleRate, m_centerFrequency);
+		DSPSignalNotification notif(m_sampleRate, m_centerFrequency);
 		(*it)->init(notif);
 	}
 
 	// pass sample rate to main window
 
-	Message* rep = DSPEngineReport::create(m_sampleRate, m_centerFrequency);
+	DSPEngineReport* rep = new DSPEngineReport(m_sampleRate, m_centerFrequency);
-	rep->submit(&m_outputMessageQueue);
+	m_outputMessageQueue.push(rep);
 
 	return StReady;
 }
@@ -402,7 +403,7 @@ DSPEngine::State DSPEngine::gotoRunning()
 
 	// Start everything
 
-	if(!m_sampleSource->startInput(0))
+	if(!m_sampleSource->start(0))
 	{
 		return gotoError("Could not start sample source");
 	}
@@ -415,7 +416,7 @@ DSPEngine::State DSPEngine::gotoRunning()
 		(*it)->start();
 	}
 
-	qDebug() << "  - input message queue pending: " << m_inputMessageQueue.countPending();
+	qDebug() << "  - input message queue pending: " << m_inputMessageQueue.size();
 
 	return StRunning;
 }
@@ -448,33 +449,6 @@ void DSPEngine::handleSetSource(SampleSource* source)
 	}
 }
 
-bool DSPEngine::distributeMessage(Message* message)
-{
-	if (m_sampleSource != 0)
-	{
-		if ((message->getDestination() == 0) || (message->getDestination() == m_sampleSource))
-		{
-			if (m_sampleSource->handleMessage(message))
-			{
-				return true;
-			}
-		}
-	}
-
-	for (SampleSinks::const_iterator it = m_sampleSinks.begin(); it != m_sampleSinks.end(); it++)
-	{
-		if ((message->getDestination() == NULL) || (message->getDestination() == *it))
-		{
-			if ((*it)->handleMessage(message))
-			{
-				return true;
-			}
-		}
-	}
-
-	return false;
-}
-
 void DSPEngine::handleData()
 {
 	if(m_state == StRunning)
@@ -483,90 +457,92 @@ void DSPEngine::handleData()
 	}
 }
 
+void DSPEngine::handleSynchronousMessages(Message *message)
+{
+	if (DSPExit::match(message))
+	{
+		gotoIdle();
+		m_state = StNotStarted;
+		exit();
+		m_syncMessenger.done(m_state);
+	}
+	else if (DSPAcquisitionInit::match(message))
+	{
+		m_state = gotoIdle();
+
+		if(m_state == StIdle) {
+			m_state = gotoInit(); // State goes ready if init is performed
+		}
+
+		m_syncMessenger.done(m_state);
+	}
+	else if (DSPAcquisitionStart::match(message))
+	{
+		if(m_state == StReady) {
+			m_state = gotoRunning();
+		}
+
+		m_syncMessenger.done(m_state);
+	}
+	else if (DSPAcquisitionStop::match(message))
+	{
+		m_state = gotoIdle();
+		m_syncMessenger.done(m_state);
+	}
+	else if (DSPGetSourceDeviceDescription::match(message))
+	{
+		((DSPGetSourceDeviceDescription*)message)->setDeviceDescription(m_deviceDescription);
+		m_syncMessenger.done(m_state);
+	}
+	else if (DSPGetErrorMessage::match(message))
+	{
+		((DSPGetErrorMessage*)message)->setErrorMessage(m_errorMessage);
+		m_syncMessenger.done(m_state);
+	}
+	else if (DSPSetSource::match(message)) {
+		handleSetSource(((DSPSetSource*)message)->getSampleSource());
+		m_syncMessenger.done(m_state);
+	}
+	else if (DSPAddSink::match(message))
+	{
+		SampleSink* sink = ((DSPAddSink*)message)->getSampleSink();
+		m_sampleSinks.push_back(sink);
+		m_syncMessenger.done(m_state);
+	}
+	else if (DSPRemoveSink::match(message))
+	{
+		SampleSink* sink = ((DSPRemoveSink*)message)->getSampleSink();
+
+		if(m_state == StRunning) {
+			sink->stop();
+		}
+
+		m_sampleSinks.remove(sink);
+		m_syncMessenger.done(m_state);
+	}
+	else if (DSPAddAudioSink::match(message))
+	{
+		m_audioSink.addFifo(((DSPAddAudioSink*)message)->getAudioFifo());
+		m_syncMessenger.done(m_state);
+	}
+	else if (DSPRemoveAudioSink::match(message))
+	{
+		m_audioSink.removeFifo(((DSPRemoveAudioSink*)message)->getAudioFifo());
+		m_syncMessenger.done(m_state);
+	}
+
+	m_syncMessenger.done(m_state);
+}
+
 void DSPEngine::handleInputMessages()
 {
 	Message* message;
 
-	while ((message = m_inputMessageQueue.accept()) != NULL)
+	while ((message = m_inputMessageQueue.pop()) != 0)
 	{
 		qDebug("DSPEngine::handleInputMessages: Message: %s", message->getIdentifier());
 
-		if (DSPPing::match(message))
+		if (DSPConfigureCorrection::match(message))
-		{
-			message->completed(m_state);
-		}
-		else if (DSPExit::match(message))
-		{
-			gotoIdle();
-			m_state = StNotStarted;
-			exit();
-			message->completed(m_state);
-		}
-		else if (DSPAcquisitionInit::match(message))
-		{
-			m_state = gotoIdle();
-
-			if(m_state == StIdle) {
-				m_state = gotoInit(); // State goes ready if init is performed
-			}
-
-			message->completed(m_state);
-		}
-		else if (DSPAcquisitionStart::match(message))
-		{
-			if(m_state == StReady) {
-				m_state = gotoRunning();
-			}
-
-			message->completed(m_state);
-		}
-		else if (DSPAcquisitionStop::match(message))
-		{
-			m_state = gotoIdle();
-			message->completed(m_state);
-		}
-		else if (DSPGetSourceDeviceDescription::match(message))
-		{
-			((DSPGetSourceDeviceDescription*)message)->setDeviceDescription(m_deviceDescription);
-			message->completed();
-		}
-		else if (DSPGetErrorMessage::match(message))
-		{
-			((DSPGetErrorMessage*)message)->setErrorMessage(m_errorMessage);
-			message->completed();
-		}
-		else if (DSPSetSource::match(message)) {
-			handleSetSource(((DSPSetSource*)message)->getSampleSource());
-			message->completed();
-		}
-		else if (DSPAddSink::match(message))
-		{
-			SampleSink* sink = ((DSPAddSink*)message)->getSampleSink();
-			m_sampleSinks.push_back(sink);
-			message->completed();
-		}
-		else if (DSPRemoveSink::match(message))
-		{
-			SampleSink* sink = ((DSPAddSink*)message)->getSampleSink();
-
-			if(m_state == StRunning) {
-				sink->stop();
-			}
-
-			m_sampleSinks.remove(sink);
-			message->completed();
-		}
-		else if (DSPAddAudioSink::match(message))
-		{
-			m_audioSink.addFifo(((DSPAddAudioSink*)message)->getAudioFifo());
-			message->completed();
-		}
-		else if (DSPRemoveAudioSink::match(message))
-		{
-			m_audioSink.removeFifo(((DSPAddAudioSink*)message)->getAudioFifo());
-			message->completed();
-		}
-		else if (DSPConfigureCorrection::match(message))
 		{
 			DSPConfigureCorrection* conf = (DSPConfigureCorrection*)message;
 			m_iqImbalanceCorrection = conf->getIQImbalanceCorrection();
@@ -585,20 +561,8 @@ void DSPEngine::handleInputMessages()
 				m_qRange = 1 << 16;
 				m_imbalance = 65536;
 			}
+		}
 
-			message->completed();
+		delete message;
-		}
-		else
-		{
-			if (DSPSignalNotification::match(message))
-			{
-				DSPSignalNotification *conf = (DSPSignalNotification*)message;
-				qDebug() << "  (" << conf->getSampleRate() << "," << conf->getFrequencyOffset() << ")";
-			}
-			if (!distributeMessage(message))
-			{
-				message->completed();
-			}
-		}
 	}
 }

sdrbase/dsp/filesink.cpp

@@ -1,4 +1,5 @@
 #include "dsp/filesink.h"
+#include "dsp/dspcommands.h"
 #include "util/simpleserializer.h"
 #include "util/messagequeue.h"
 
@@ -23,23 +24,26 @@ FileSink::~FileSink()
     stopRecording();
 }
 
-void FileSink::configure(MessageQueue* msgQueue, const std::string& filename, int sampleRate, quint64 centerFrequency)
+void FileSink::configure(MessageQueue* msgQueue, const std::string& filename)
 {
-	Message* cmd = MsgConfigureFileSink::create(filename, sampleRate, centerFrequency);
+	Message* cmd = MsgConfigureFileSink::create(filename);
-	cmd->submit(msgQueue, this);
+	msgQueue->push(cmd);
 }
 
-bool FileSink::init(Message* cmd)
+bool FileSink::init(const Message& cmd)
 {
-	if (DSPSignalNotification::match(cmd))
+	if (DSPSignalNotification::match(&cmd))
 	{
-		DSPSignalNotification* notif = (DSPSignalNotification*) cmd;
+		DSPSignalNotification* notif = (DSPSignalNotification*) &cmd;
 		m_sampleRate = notif->getSampleRate();
 		m_centerFrequency = notif->getFrequencyOffset();
 		qDebug() << "FileSink::init: DSPSignalNotification: m_inputSampleRate: " << m_sampleRate;
-		cmd->completed();
 		return true;
 	}
+	else
+	{
+		return false;
+	}
 }
 
 void FileSink::feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly)
@@ -93,17 +97,14 @@ void FileSink::stopRecording()
     }
 }
 
-bool FileSink::handleMessage(Message* message)
+bool FileSink::handleMessage(const Message& message)
 {
-    if (MsgConfigureFileSink::match(message))
+    if (MsgConfigureFileSink::match(&message))
     {
-        MsgConfigureFileSink* conf = (MsgConfigureFileSink*) message;
+        MsgConfigureFileSink* conf = (MsgConfigureFileSink*) &message;
-        handleConfigure(conf->getFileName(), conf->getSampleRate(), conf->getCenterFrequency());
+        handleConfigure(conf->getFileName());
         qDebug() << "FileSink::handleMessage:"
-        		<< " fileName: " << m_fileName.c_str()
+        		<< " fileName: " << m_fileName.c_str();
-				<< " sampleRate: " << m_sampleRate
-				<< " centerFrequency: " << m_centerFrequency;
-        message->completed();
         return true;
     }
     else
@@ -112,16 +113,14 @@ bool FileSink::handleMessage(Message* message)
     }
 }
 
-void FileSink::handleConfigure(const std::string& fileName, int sampleRate, quint64 centerFrequency)
+void FileSink::handleConfigure(const std::string& fileName)
 {
-    if ((fileName != m_fileName) || (m_sampleRate != sampleRate) || (m_centerFrequency != centerFrequency))
+    if (fileName != m_fileName)
     {
         stopRecording();
     }
     
 	m_fileName = fileName;
-    m_sampleRate = sampleRate;
-    m_centerFrequency = centerFrequency;
 }
 
 void FileSink::writeHeader()

sdrbase/dsp/nullsink.cpp

@@ -7,6 +7,15 @@ NullSink::NullSink()
 	setObjectName("NullSink");
 }
 
+NullSink::~NullSink()
+{
+}
+
+bool NullSink::init(const Message& message)
+{
+	return false;
+}
+
 void NullSink::feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly)
 {
 }
@@ -19,8 +28,7 @@ void NullSink::stop()
 {
 }
 
-bool NullSink::handleMessage(Message* message)
+bool NullSink::handleMessage(const Message& message)
 {
-	message->completed();
+	return false;
-	return true;
 }

sdrbase/dsp/samplesink.cpp

@@ -7,65 +7,3 @@ SampleSink::SampleSink()
 SampleSink::~SampleSink()
 {
 }
-
-bool SampleSink::executeMessage(Message *cmd)
-{
-	return handleMessage(cmd);
-}
-
-#if 0
-#include "samplesink.h"
-
-SampleSink::SampleSink() :
-	m_sinkBuffer(),
-	m_sinkBufferFill(0)
-{
-}
-
-void SampleSink::feed(SampleVector::const_iterator begin, SampleVector::const_iterator end)
-{
-	size_t wus = workUnitSize();
-
-	// make sure our buffer is big enough for at least one work unit
-	if(m_sinkBuffer.size() < wus)
-		m_sinkBuffer.resize(wus);
-
-	while(begin < end) {
-		// if the buffer contains something, keep filling it until it contains one complete work unit
-		if((m_sinkBufferFill > 0) && (m_sinkBufferFill < wus)) {
-			// check number if missing samples, but don't copy more than we have
-			size_t len = wus - m_sinkBufferFill;
-			if(len > (size_t)(end - begin))
-				len = end - begin;
-			// copy
-			std::copy(begin, begin + len, m_sinkBuffer.begin() + m_sinkBufferFill);
-			// adjust pointers
-			m_sinkBufferFill += len;
-			begin += len;
-		}
-
-		// if one complete work unit is in the buffer, feed it to the worker
-		if(m_sinkBufferFill >= wus) {
-			size_t done = 0;
-			while(m_sinkBufferFill - done >= wus)
-				done += work(m_sinkBuffer.begin() + done, m_sinkBuffer.begin() + done + wus);
-			// now copy the remaining data to the front of the buffer (should be zero)
-			if(m_sinkBufferFill - done > 0) {
-				qDebug("error in SampleSink buffer management");
-				std::copy(m_sinkBuffer.begin() + done, m_sinkBuffer.begin() + m_sinkBufferFill, m_sinkBuffer.begin());
-			}
-			m_sinkBufferFill -= done;
-		}
-
-		// if no remainder from last run is buffered and we have at least one work unit left, feed the data to the worker
-		if(m_sinkBufferFill == 0) {
-			while((size_t)(end - begin) > wus)
-				begin += work(begin, begin + wus);
-			// copy any remaining data to the buffer
-			std::copy(begin, end, m_sinkBuffer.begin());
-			m_sinkBufferFill = end - begin;
-			begin += m_sinkBufferFill;
-		}
-	}
-}
-#endif

sdrbase/dsp/scopevis.cpp

@@ -32,6 +32,10 @@ ScopeVis::ScopeVis(GLScope* glScope) :
 	m_traceback.resize(20*m_traceChunkSize);
 }
 
+ScopeVis::~ScopeVis()
+{
+}
+
 void ScopeVis::configure(MessageQueue* msgQueue, 
     TriggerChannel triggerChannel, 
     Real triggerLevel, 
@@ -48,7 +52,21 @@ void ScopeVis::configure(MessageQueue* msgQueue,
 			triggerPre,
 			triggerDelay,
 			traceSize);
-	cmd->submit(msgQueue, this);
+	msgQueue->push(cmd);
+}
+
+bool ScopeVis::init(const Message& cmd)
+{
+	if (DSPSignalNotification::match(&cmd))
+	{
+		DSPSignalNotification* signal = (DSPSignalNotification*) &cmd;
+		m_sampleRate = signal->getSampleRate();
+		return true;
+	}
+	else
+	{
+		return false;
+	}
 }
 
 void ScopeVis::feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly)
@@ -200,18 +218,11 @@ void ScopeVis::stop()
 {
 }
 
-bool ScopeVis::handleMessageKeep(Message* message)
+bool ScopeVis::handleMessage(const Message& message)
 {
-	if(DSPSignalNotification::match(message)) {
+	if (MsgConfigureScopeVis::match(&message))
-		DSPSignalNotification* signal = (DSPSignalNotification*)message;
+	{
-		m_sampleRate = signal->getSampleRate();
+		MsgConfigureScopeVis* conf = (MsgConfigureScopeVis*) &message;
-		/*fprintf(stderr, "ScopeVis::handleMessage : %d samples/sec, %lld Hz offset, traceSize: \n",
-				m_sampleRate,
-				signal->getFrequencyOffset(),
-				m_trace.size());*/
-		return true;
-	} else if(MsgConfigureScopeVis::match(message)) {
-		MsgConfigureScopeVis* conf = (MsgConfigureScopeVis*)message;
         m_tracebackCount = 0;
 		m_triggerState = Config;
 		m_triggerChannel = (TriggerChannel) conf->getTriggerChannel();
@@ -230,7 +241,7 @@ bool ScopeVis::handleMessageKeep(Message* message)
         if (newSize > m_traceback.size()) {  // fitting the exact required space is not a requirement for the back trace
             m_traceback.resize(newSize);
         }
-		qDebug() << "ScopeVis::handleMessageKeep:"
+		qDebug() << "ScopeVis::handleMessage:"
 				<< " m_triggerChannel: " << m_triggerChannel
 				<< " m_triggerLevel: " << m_triggerLevel
 				<< " m_triggerPositiveEdge: " << (m_triggerPositiveEdge ? "edge+" : "edge-")
@@ -239,31 +250,13 @@ bool ScopeVis::handleMessageKeep(Message* message)
 				<< " m_triggerDelay: " << m_triggerDelay
 				<< " m_traceSize: " << m_trace.size();
 		return true;
-	/*
+	}
-	} else if(DSPConfigureScopeVis::match(message)) {
+	else
-		DSPConfigureScopeVis* conf = (DSPConfigureScopeVis*)message;
+	{
-		m_triggerState = Untriggered;
-		m_triggerChannel = (TriggerChannel)conf->getTriggerChannel();
-		m_triggerLevelHigh = conf->getTriggerLevelHigh() * 32767;
-		m_triggerLevelLow = conf->getTriggerLevelLow() * 32767;
-		return true;*/
-	} else {
 		return false;
 	}
 }
 
-bool ScopeVis::handleMessage(Message* message)
-{
-	bool done = handleMessageKeep(message);
-
-	if (done)
-	{
-		message->completed();
-	}
-
-	return done;
-}
-
 void ScopeVis::setSampleRate(int sampleRate)
 {
 	m_sampleRate = sampleRate;

sdrbase/dsp/spectrumscopecombovis.cpp

@@ -9,6 +9,18 @@ SpectrumScopeComboVis::SpectrumScopeComboVis(SpectrumVis* spectrumVis, ScopeVis*
 	setObjectName("SpectrumScopeComboVis");
 }
 
+SpectrumScopeComboVis::~SpectrumScopeComboVis()
+{
+}
+
+bool SpectrumScopeComboVis::init(const Message& cmd)
+{
+	bool spectDone = m_spectrumVis->init(cmd);
+	bool scopeDone = m_scopeVis->init(cmd);
+
+	return (spectDone || scopeDone);
+}
+
 void SpectrumScopeComboVis::feed(SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly)
 {
 	m_scopeVis->feed(begin, end, false);
@@ -28,15 +40,10 @@ void SpectrumScopeComboVis::stop()
 	m_scopeVis->stop();
 }
 
-bool SpectrumScopeComboVis::handleMessage(Message* message)
+bool SpectrumScopeComboVis::handleMessage(const Message& message)
 {
-	bool spectDone = m_spectrumVis->handleMessageKeep(message);
+	bool spectDone = m_spectrumVis->handleMessage(message);
-	bool scopeDone = m_scopeVis->handleMessageKeep(message);
+	bool scopeDone = m_scopeVis->handleMessage(message);
-
-	if (spectDone || scopeDone)
-	{
-		message->completed();
-	}
 
 	return (spectDone || scopeDone);
 }

sdrbase/dsp/spectrumvis.cpp

@@ -30,10 +30,15 @@ SpectrumVis::~SpectrumVis()
 	delete m_fft;
 }
 
+bool SpectrumVis::init(const Message& cmd)
+{
+	return false;
+}
+
 void SpectrumVis::configure(MessageQueue* msgQueue, int fftSize, int overlapPercent, FFTWindow::Function window)
 {
-	Message* cmd = DSPConfigureSpectrumVis::create(fftSize, overlapPercent, window);
+	DSPConfigureSpectrumVis* cmd = new DSPConfigureSpectrumVis(fftSize, overlapPercent, window);
-	cmd->submit(msgQueue, this);
+	msgQueue->push(cmd);
 }
 
 void SpectrumVis::feedTriggered(SampleVector::const_iterator triggerPoint, SampleVector::const_iterator begin, SampleVector::const_iterator end, bool positiveOnly)
@@ -136,39 +141,39 @@ void SpectrumVis::stop()
 {
 }
 
-bool SpectrumVis::handleMessageKeep(Message* message)
+bool SpectrumVis::handleMessage(const Message& message)
 {
-	if(DSPConfigureSpectrumVis::match(message)) {
+	if(DSPConfigureSpectrumVis::match(&message))
-		DSPConfigureSpectrumVis* conf = (DSPConfigureSpectrumVis*)message;
+	{
+		DSPConfigureSpectrumVis* conf = (DSPConfigureSpectrumVis*) &message;
 		handleConfigure(conf->getFFTSize(), conf->getOverlapPercent(), conf->getWindow());
 		return true;
-	} else {
+	}
+	else
+	{
 		return false;
 	}
 }
 
-bool SpectrumVis::handleMessage(Message* message)
-{
-	bool done = handleMessageKeep(message);
-
-	if (done)
-	{
-		message->completed();
-	}
-
-	return done;
-}
-
 void SpectrumVis::handleConfigure(int fftSize, int overlapPercent, FFTWindow::Function window)
 {
-	if(fftSize > MAX_FFT_SIZE)
+	if (fftSize > MAX_FFT_SIZE)
+	{
 		fftSize = MAX_FFT_SIZE;
-	else if(fftSize < 64)
+	}
+	else if (fftSize < 64)
+	{
 		fftSize = 64;
-	if(overlapPercent > 100)
+	}
+
+	if (overlapPercent > 100)
+	{
 		m_overlapPercent = 100;
-	else if(overlapPercent < 0)
+	}
+	else if (overlapPercent < 0)
+	{
 		m_overlapPercent = 0;
+	}
 
 	m_fftSize = fftSize;
 	m_overlapPercent = overlapPercent;

sdrbase/mainwindow.cpp

@@ -319,7 +319,7 @@ void MainWindow::handleDSPMessages()
 {
 	Message* message;
 
-	while ((message = m_dspEngine->getOutputMessageQueue()->accept()) != 0)
+	while ((message = m_dspEngine->getOutputMessageQueue()->pop()) != 0)
 	{
 		qDebug("Message: %s", message->getIdentifier());
 
@@ -333,10 +333,11 @@ void MainWindow::handleDSPMessages()
 			qDebug("SampleRate:%d, CenterFrequency:%llu", rep->getSampleRate(), rep->getCenterFrequency());
 			updateCenterFreqDisplay();
 			updateSampleRate();
-			message->completed();
 			qDebug() << "MainWindow::handleMessages: m_fileSink->configure";
-			m_fileSink->configure(m_fileSink->getMessageQueue(), m_sampleFileName, m_sampleRate, m_centerFrequency);
+			m_fileSink->configure(m_fileSink->getInputMessageQueue(), m_sampleFileName);
 		}
+
+		delete message;
 	}
 }
 
@@ -344,14 +345,14 @@ void MainWindow::handleMessages()
 {
 	Message* message;
 
-	while ((message = m_messageQueue->accept()) != 0)
+	while ((message = m_messageQueue->pop()) != 0)
 	{
 		qDebug("Message: %s", message->getIdentifier());
-
 		std::cerr << "MainWindow::handleMessages: " << message->getIdentifier() << std::endl;
 
-		if (!m_pluginManager->handleMessage(message)) {
+		if (!m_pluginManager->handleMessage(message))
-			message->completed();
+		{
+			delete message;
 		}
 	}
 }

sdrbase/util/message.cpp

@@ -1,3 +1,20 @@
+///////////////////////////////////////////////////////////////////////////////////
+// Copyright (C) 2015 F4EXB                                                      //
+// written by Edouard Griffiths                                                  //
+//                                                                               //
+// This program is free software; you can redistribute it and/or modify          //
+// it under the terms of the GNU General Public License as published by          //
+// the Free Software Foundation as version 3 of the License, or                  //
+//                                                                               //
+// This program is distributed in the hope that it will be useful,               //
+// but WITHOUT ANY WARRANTY; without even the implied warranty of                //
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the                  //
+// GNU General Public License V3 for more details.                               //
+//                                                                               //
+// You should have received a copy of the GNU General Public License             //
+// along with this program. If not, see <http://www.gnu.org/licenses/>.          //
+///////////////////////////////////////////////////////////////////////////////////
+
 #include <QWaitCondition>
 #include <QMutex>
 #include "util/message.h"
@@ -6,20 +23,12 @@
 const char* Message::m_identifier = "Message";
 
 Message::Message() :
-	m_destination(NULL),
+	m_destination(0)
-	m_synchronous(false),
-	m_waitCondition(NULL),
-	m_mutex(NULL),
-	m_complete(0)
 {
 }
 
 Message::~Message()
 {
-	if(m_waitCondition != NULL)
-		delete m_waitCondition;
-	if(m_mutex != NULL)
-		delete m_mutex;
 }
 
 const char* Message::getIdentifier() const
@@ -32,48 +41,7 @@ bool Message::matchIdentifier(const char* identifier) const
 	return m_identifier == identifier;
 }
 
-bool Message::match(Message* message)
+bool Message::match(const Message* message)
 {
 	return message->matchIdentifier(m_identifier);
 }
-
-void Message::submit(MessageQueue* queue, void* destination)
-{
-	m_destination = destination;
-	m_synchronous = false;
-	queue->submit(this);
-}
-
-int Message::execute(MessageQueue* queue, void* destination)
-{
-	m_destination = destination;
-	m_synchronous = true;
-
-	if(m_waitCondition == NULL)
-		m_waitCondition = new QWaitCondition;
-	if(m_mutex == NULL)
-		m_mutex = new QMutex;
-
-	m_mutex->lock();
-	m_complete.testAndSetAcquire(0, 1);
-	queue->submit(this);
-	while(!m_complete.testAndSetAcquire(0, 1))
-		((QWaitCondition*)m_waitCondition)->wait(m_mutex, 100);
-	m_complete = 0;
-	int result = m_result;
-	m_mutex->unlock();
-	return result;
-}
-
-void Message::completed(int result)
-{
-	if(m_synchronous) {
-		m_result = result;
-		m_complete = 0;
-		if(m_waitCondition == NULL)
-			qFatal("wait condition is NULL");
-		m_waitCondition->wakeAll();
-	} else {
-		delete this;
-	}
-}

sdrbase/util/messagequeue.cpp

@@ -1,3 +1,20 @@
+///////////////////////////////////////////////////////////////////////////////////
+// Copyright (C) 2015 F4EXB                                                      //
+// written by Edouard Griffiths                                                  //
+//                                                                               //
+// This program is free software; you can redistribute it and/or modify          //
+// it under the terms of the GNU General Public License as published by          //
+// the Free Software Foundation as version 3 of the License, or                  //
+//                                                                               //
+// This program is distributed in the hope that it will be useful,               //
+// but WITHOUT ANY WARRANTY; without even the implied warranty of                //
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the                  //
+// GNU General Public License V3 for more details.                               //
+//                                                                               //
+// You should have received a copy of the GNU General Public License             //
+// along with this program. If not, see <http://www.gnu.org/licenses/>.          //
+///////////////////////////////////////////////////////////////////////////////////
+
 #include "util/messagequeue.h"
 #include "util/message.h"
 
@@ -10,29 +27,44 @@ MessageQueue::MessageQueue(QObject* parent) :
 
 MessageQueue::~MessageQueue()
 {
-	Message* cmd;
+	Message* message;
-	while((cmd = accept()) != NULL)
+
-		cmd->completed();
+	while ((message = pop()) != 0)
+	{
+		delete message;
+	}
 }
 
-void MessageQueue::submit(Message* message)
+void MessageQueue::push(Message* message, bool emitSignal)
 {
-	m_lock.lock();
+	if (message)
-	m_queue.append(message);
+	{
-	m_lock.unlock();
+		m_lock.lock();
-	emit messageEnqueued();
+		m_queue.append(message);
+		m_lock.unlock();
+	}
+
+	if (emitSignal)
+	{
+		emit messageEnqueued();
+	}
 }
 
-Message* MessageQueue::accept()
+Message* MessageQueue::pop()
 {
 	SpinlockHolder spinlockHolder(&m_lock);
 
-	if(m_queue.isEmpty())
+	if (m_queue.isEmpty())
-		return NULL;
+	{
-	else return m_queue.takeFirst();
+		return 0;
+	}
+	else
+	{
+		return m_queue.takeFirst();
+	}
 }
 
-int MessageQueue::countPending()
+int MessageQueue::size()
 {
 	SpinlockHolder spinlockHolder(&m_lock);
 

sdrbase/util/syncmessenger.cpp

@@ -0,0 +1,56 @@
+///////////////////////////////////////////////////////////////////////////////////
+// Copyright (C) 2015 F4EXB                                                      //
+// written by Edouard Griffiths                                                  //
+//                                                                               //
+// This program is free software; you can redistribute it and/or modify          //
+// it under the terms of the GNU General Public License as published by          //
+// the Free Software Foundation as version 3 of the License, or                  //
+//                                                                               //
+// This program is distributed in the hope that it will be useful,               //
+// but WITHOUT ANY WARRANTY; without even the implied warranty of                //
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the                  //
+// GNU General Public License V3 for more details.                               //
+//                                                                               //
+// You should have received a copy of the GNU General Public License             //
+// along with this program. If not, see <http://www.gnu.org/licenses/>.          //
+///////////////////////////////////////////////////////////////////////////////////
+
+#include "util/syncmessenger.h"
+#include "util/message.h"
+
+SyncMessenger::SyncMessenger() :
+	m_complete(0),
+	m_result(0)
+{}
+
+SyncMessenger::~SyncMessenger()
+{}
+
+int SyncMessenger::sendWait(Message *message, unsigned long msPollTime)
+{
+	m_mutex.lock();
+	m_complete.testAndSetAcquire(0, 1);
+
+	emit messageSent(message);
+
+	while (!m_complete.testAndSetAcquire(0, 1))
+	{
+		m_waitCondition.wait(&m_mutex, msPollTime);
+	}
+
+	m_complete = 0;
+	int result = m_result;
+	m_mutex.unlock();
+
+	return result;
+}
+
+void SyncMessenger::done(int result)
+{
+	m_result = result;
+	m_complete = 0;
+	m_waitCondition.wakeAll();
+}
+
+
+