finally fixed the waterfall7

core/src/dsp/routing.h

@@ -1,6 +1,8 @@
 #pragma once
 #include <dsp/block.h>
+#include <dsp/buffer.h>
 #include <cstring>
+#include <numeric>
 #include <spdlog/spdlog.h>
 
 namespace dsp {
@@ -62,18 +64,22 @@ namespace dsp {
 
     };
 
+
+    // NOTE: I'm not proud of this, it's BAD and just taken from the previous DSP, but it works...
     template <class T>
     class Reshaper : public generic_block<Reshaper<T>> {
     public:
         Reshaper() {}
 
-        Reshaper(stream<T>* in) { init(in); }
+        Reshaper(stream<T>* in, int keep, int skip) { init(in, keep, skip); }
 
         ~Reshaper() { generic_block<Reshaper<T>>::stop(); }
 
-        void init(stream<T>* in) {
+        void init(stream<T>* in, int keep, int skip) {
             _in = in;
-            buffer = (T*)volk_malloc(STREAM_BUFFER_SIZE * sizeof(T), volk_get_alignment());
+            _keep = keep;
+            _skip = skip;
+            ringBuf.init(keep * 2);
             generic_block<Reshaper<T>>::registerInput(_in);
             generic_block<Reshaper<T>>::registerOutput(&out);
         }
@@ -87,19 +93,92 @@ namespace dsp {
             generic_block<Reshaper<T>>::tempStart();
         }
 
+        void setKeep(int keep) {
+            std::lock_guard<std::mutex> lck(generic_block<Reshaper<T>>::ctrlMtx);
+            generic_block<Reshaper<T>>::tempStop();
+            generic_block<Reshaper<T>>::unregisterInput(_in);
+            _keep = keep;
+            generic_block<Reshaper<T>>::registerInput(_in);
+            generic_block<Reshaper<T>>::tempStart();
+        }
+
+        void setSkip(int skip) {
+            std::lock_guard<std::mutex> lck(generic_block<Reshaper<T>>::ctrlMtx);
+            generic_block<Reshaper<T>>::tempStop();
+            generic_block<Reshaper<T>>::unregisterInput(_in);
+            _skip = skip;
+            generic_block<Reshaper<T>>::registerInput(_in);
+            generic_block<Reshaper<T>>::tempStart();
+        }
+
         int run() {
             int count = _in->read();
+            if (count < 0) { return -1; }
+            ringBuf.write(_in->data, count);
             _in->flush();
-            return count;
         }
 
         stream<T> out;
 
     private:
+        void doStart() {
+            workThread = std::thread(&Reshaper<T>::loop, this);
+            bufferWorkerThread = std::thread(&Reshaper<T>::bufferWorker, this);
+        }
+
+        void loop() {
+            while (run() >= 0);
+        }
+
+        void doStop() {
+            _in->stopReader();
+            ringBuf.stopReader();
+            out.stopWriter();
+            ringBuf.stopWriter();
+
+            if (workThread.joinable()) {
+                workThread.join();
+            }
+            if (bufferWorkerThread.joinable()) {
+                bufferWorkerThread.join();
+            }
+
+            _in->clearReadStop();
+            ringBuf.clearReadStop();
+            out.clearWriteStop();
+            ringBuf.clearWriteStop();
+        }
+
+        void bufferWorker() {
+            complex_t* buf = new complex_t[_keep];
+            bool delay = _skip < 0;
+
+            int readCount = std::min<int>(_keep + _skip, _keep);
+            int skip = std::max<int>(_skip, 0);
+            int delaySize = (-_skip) * sizeof(complex_t);
+
+            complex_t* start = &buf[std::max<int>(-_skip, 0)];
+            complex_t* delayStart = &buf[_keep + _skip];
+
+            while (true) {
+                if (delay) {
+                    memmove(buf, delayStart, delaySize);
+                }
+                if (ringBuf.readAndSkip(start, readCount, skip) < 0) { break; };
+                if (out.aquire() < 0) { break; }
+                memcpy(out.data, buf, _keep * sizeof(complex_t));
+                out.write(_keep);
+            }
+            delete[] buf;
+        }
+
         stream<T>* _in;
-        T* buffer;
         int _outBlockSize;
-        int readCount;
+        RingBuffer<T> ringBuf;
+        std::thread bufferWorkerThread;
+        std::thread workThread;
+        int _keep, _skip;
+        
 
     };
 }