new buffer thingy

core/src/dsp/buffer.h

@@ -214,4 +214,112 @@ namespace dsp {
         std::condition_variable canReadVar;
         std::condition_variable canWriteVar;
     };
+
+#define TEST_BUFFER_SIZE    32
+
+    template <class T>
+    class SampleFrameBuffer : public generic_block<SampleFrameBuffer<T>> {
+    public:
+        SampleFrameBuffer() {}
+
+        SampleFrameBuffer(stream<T>* in) { init(in); }
+
+        ~SampleFrameBuffer() {
+            stop();
+            out.stopWriter();
+            stopWorker = true;
+            cnd.notify_all();
+            if (readWorkerThread.joinable()) { readWorkerThread.join(); }
+        }
+
+        void init(stream<T>* in) {
+            _in = in;
+
+            for (int i = 0; i < TEST_BUFFER_SIZE; i++) {
+                buffers[i] = new T[STREAM_BUFFER_SIZE];
+            }
+
+            generic_block<SampleFrameBuffer<T>>::registerInput(in);
+            generic_block<SampleFrameBuffer<T>>::registerOutput(&out);
+
+            readWorkerThread = std::thread(&SampleFrameBuffer::worker, this);
+        }
+
+        void setInput(stream<T>* in) {
+            std::lock_guard<std::mutex> lck(generic_block<SampleFrameBuffer<T>>::ctrlMtx);
+            generic_block<SampleFrameBuffer<T>>::tempStop();
+            generic_block<SampleFrameBuffer<T>>::unregisterInput(_in);
+            _in = in;
+            generic_block<SampleFrameBuffer<T>>::registerInput(_in);
+            generic_block<SampleFrameBuffer<T>>::tempStart();
+        }
+
+        int run() {
+            // Wait for data
+            int count = _in->read();
+            if (count < 0) { return -1; }
+
+            if (bypass) {
+                memcpy(out.writeBuf, _in->readBuf, count * sizeof(T));
+                _in->flush();
+                if (!out.swap(count)) { return -1; }
+                return count;
+            }
+
+            // Push it on the ring buffer
+            {
+                std::lock_guard<std::mutex> lck(bufMtx);
+                memcpy(buffers[writeCur], _in->readBuf, count * sizeof(T));
+                uintptr_t ptr = (uintptr_t)buffers[writeCur];
+                sizes[writeCur] = count;
+                writeCur++;
+                writeCur = ((writeCur) % TEST_BUFFER_SIZE);
+
+                if (((writeCur - readCur + TEST_BUFFER_SIZE) % TEST_BUFFER_SIZE) >= (TEST_BUFFER_SIZE-2)) {
+                    spdlog::warn("Overflow");
+                }
+            }
+            cnd.notify_all();
+            _in->flush();
+            return count;
+        }
+
+        void worker() {
+            while (true) {
+                // Wait for data
+                std::unique_lock lck(bufMtx);
+                cnd.wait(lck, [this](){ return (((writeCur - readCur + TEST_BUFFER_SIZE) % TEST_BUFFER_SIZE) > 0) || stopWorker; });
+                if (stopWorker) { break; }
+
+                // Write one to output buffer and unlock in preparation to swap buffers
+                int count = sizes[readCur];
+                memcpy(out.writeBuf, buffers[readCur], count * sizeof(T));
+                readCur++;
+                readCur = ((readCur) % TEST_BUFFER_SIZE);
+                lck.unlock();
+
+                // Swap
+                if (!out.swap(count)) { break; }
+            } 
+        }
+
+        stream<T> out;
+
+        int writeCur = 0;
+        int readCur = 0;
+
+        bool bypass = false;
+
+    private:
+        stream<T>* _in;
+
+        std::thread readWorkerThread;
+        std::mutex bufMtx;
+        std::condition_variable cnd;
+        T* buffers[TEST_BUFFER_SIZE];
+        int sizes[TEST_BUFFER_SIZE];
+        
+        bool stopWorker = false;
+
+    };
 };

core/src/gui/main_window.cpp

@@ -459,6 +459,10 @@ void MainWindow::draw() {
             ImGui::Checkbox("Show demo window", &demoWindow);
             ImGui::Text("ImGui version: %s", ImGui::GetVersion());
 
+            ImGui::Checkbox("Bypass buffering", &sigpath::signalPath.inputBuffer.bypass);
+
+            ImGui::Text("Buffering: %d", (sigpath::signalPath.inputBuffer.writeCur - sigpath::signalPath.inputBuffer.readCur + 20) % 20);
+            
             if (ImGui::Button("Test Bug")) {
                 spdlog::error("Will this make the software crash?");
             }

core/src/signal_path/dsp.cpp

@@ -10,7 +10,9 @@ void SignalPath::init(uint64_t sampleRate, int fftRate, int fftSize, dsp::stream
     this->fftSize = fftSize;
     inputBlockSize = sampleRate / 200.0f;
 
-    split.init(input);
+    // split.init(input);
+    inputBuffer.init(input);
+    split.init(&inputBuffer.out);
 
     reshape.init(&fftStream, fftSize, (sampleRate / fftRate) - fftSize);
     split.bindStream(&fftStream);
@@ -45,12 +47,14 @@ double SignalPath::getSampleRate() {
 }
 
 void SignalPath::start() {
+    inputBuffer.start();
     split.start();
     reshape.start();
     fftHandlerSink.start();
 }
 
 void SignalPath::stop() {
+    inputBuffer.stop();
     split.stop();
     reshape.stop();
     fftHandlerSink.stop();
@@ -83,7 +87,8 @@ void SignalPath::removeVFO(std::string name) {
 }
 
 void SignalPath::setInput(dsp::stream<dsp::complex_t>* input) {
-    split.setInput(input);
+    // split.setInput(input);
+    inputBuffer.setInput(input);
 }
 
 void SignalPath::bindIQStream(dsp::stream<dsp::complex_t>* stream) {

core/src/signal_path/dsp.h

@@ -21,6 +21,8 @@ public:
     void startFFT();
     void stopFFT();
 
+    dsp::SampleFrameBuffer<dsp::complex_t> inputBuffer;
+
 private:
     struct VFO_t {
         dsp::stream<dsp::complex_t>* inputStream;