Experimenting with max speed waterfall (no dropped FFTs :)

- FFT Detail from zoomed-in views is greatly improved with the
additional data

src/process/SpectrumVisualProcessor.cpp

@@ -14,6 +14,7 @@ SpectrumVisualProcessor::SpectrumVisualProcessor() : lastInputBandwidth(0), last
     
     fft_ceil_ma = fft_ceil_maa = 100.0;
     fft_floor_ma = fft_floor_maa = 0.0;
+    desiredInputSize = 0;
 }
 
 SpectrumVisualProcessor::~SpectrumVisualProcessor() {
@@ -45,8 +46,13 @@ long SpectrumVisualProcessor::getBandwidth() {
     return bandwidth.load();
 }
 
+int SpectrumVisualProcessor::getDesiredInputSize() {
+    return desiredInputSize;
+}
+
 void SpectrumVisualProcessor::setup(int fftSize_in) {
     fftSize = fftSize_in;
+    desiredInputSize = fftSize;
     
     if (fftwInput) {
         free(fftwInput);
@@ -103,6 +109,8 @@ void SpectrumVisualProcessor::process() {
             
             int desired_input_size = fftSize / resamplerRatio;
             
+            this->desiredInputSize = desired_input_size;
+            
             if (iqData->data.size() < desired_input_size) {
                 //                std::cout << "fft underflow, desired: " << desired_input_size << " actual:" << input->data.size() << std::endl;
                 desired_input_size = iqData->data.size();
@@ -280,5 +288,7 @@ void SpectrumVisualProcessor::process() {
         
         distribute(output);
     }
-    
+ 
+    iqData->decRefCount();
 }
+

src/process/SpectrumVisualProcessor.h

@@ -26,6 +26,8 @@ public:
     void setBandwidth(long bandwidth_in);
     long getBandwidth();
     
+    int getDesiredInputSize();
+    
     void setup(int fftSize);
     
 protected:
@@ -59,4 +61,39 @@ private:
     
     std::vector<liquid_float_complex> shiftBuffer;
     std::vector<liquid_float_complex> resampleBuffer;
+    int desiredInputSize;
+};
+
+
+class FFTDataDistributor : public VisualProcessor<DemodulatorThreadIQData, DemodulatorThreadIQData> {
+public:
+    void setFFTSize(int fftSize) {
+        this->fftSize = fftSize;
+    }
+    
+protected:
+    void process() {
+        while (!input->empty()) {
+            if (!isAnyOutputEmpty()) {
+                return;
+            }
+            DemodulatorThreadIQData *inp;
+            input->pop(inp);
+            if (inp) {
+                if (inp->data.size() >= fftSize) {
+                    for (int i = 0, iMax = inp->data.size()-fftSize; i < iMax; i += fftSize) {
+                        DemodulatorThreadIQData *outp = outputBuffers.getBuffer();
+                        outp->frequency = inp->frequency;
+                        outp->sampleRate = inp->sampleRate;
+                        outp->data.assign(inp->data.begin()+i,inp->data.begin()+i+fftSize);
+                        distribute(outp);
+                    }
+                }
+                inp->decRefCount();
+            }
+        }
+    }
+    
+    ReBuffer<DemodulatorThreadIQData> outputBuffers;
+    int fftSize;
 };

src/process/VisualProcessor.h

@@ -11,6 +11,28 @@ public:
 	virtual ~VisualProcessor() {
 
 	}
+    
+    bool isInputEmpty() {
+        return input->empty();
+    }
+    
+    bool isOutputEmpty() {
+        for (outputs_i = outputs.begin(); outputs_i != outputs.end(); outputs_i++) {
+            if ((*outputs_i)->full()) {
+                return false;
+            }
+        }
+        return true;
+    }
+    
+    bool isAnyOutputEmpty() {
+        for (outputs_i = outputs.begin(); outputs_i != outputs.end(); outputs_i++) {
+            if (!(*outputs_i)->full()) {
+                return true;
+            }
+        }
+        return false;
+    }
 
     void setInput(ThreadQueue<InputDataType *> *vis_in) {
         busy_update.lock();
@@ -60,24 +82,6 @@ protected:
         	}
         }
     }
-    
-    bool isOutputEmpty() {
-        for (outputs_i = outputs.begin(); outputs_i != outputs.end(); outputs_i++) {
-            if (!(*outputs_i)->empty()) {
-            	return false;
-            }
-        }
-        return true;
-    }
-
-    bool isAnyOutputEmpty() {
-        for (outputs_i = outputs.begin(); outputs_i != outputs.end(); outputs_i++) {
-            if ((*outputs_i)->empty()) {
-            	return true;
-            }
-        }
-        return false;
-    }
 
     ThreadQueue<InputDataType *> *input;
     std::vector<ThreadQueue<OutputDataType *> *> outputs;
@@ -90,10 +94,10 @@ template<class OutputDataType = ReferenceCounter>
 class VisualDataDistributor : public VisualProcessor<OutputDataType, OutputDataType> {
 protected:
     void process() {
-        if (!VisualProcessor<OutputDataType, OutputDataType>::isOutputEmpty()) {
-            return;
-        }
         while (!VisualProcessor<OutputDataType, OutputDataType>::input->empty()) {
+            if (!VisualProcessor<OutputDataType, OutputDataType>::isAnyOutputEmpty()) {
+                return;
+            }
         	OutputDataType *inp;
         	VisualProcessor<OutputDataType, OutputDataType>::input->pop(inp);
             if (inp) {