Input stream and output stream both running. Added lockfree boost dependency to git modules, removed carma build as it is not required and every time again downloads Armadillo. Added functions to enable / disable all channels at once. Fixed a bug with RtAudio input streams. Fixed a bug in StreamMgr leading to segfaults (how to: use std::move ;)).

.gitmodules

@@ -14,7 +14,7 @@
 	path = third_party/tomlplusplus
 	url = https://github.com/marzer/tomlplusplus
 [submodule "third_party/lockfree"]
-	path = third_party/lockfree
+	path = third_party/boost/lockfree
 	url = https://github.com/boostorg/lockfree
 [submodule "third_party/carma"]
 	path = third_party/carma
@@ -22,3 +22,9 @@
 [submodule "third_party/thread-pool"]
 	path = third_party/thread-pool
 	url = https://github.com/bshoshany/thread-pool
+[submodule "third_party/rtaudio"]
+	path = third_party/rtaudio
+	url = https://github.com/thestk/rtaudio
+[submodule "third_party/boost/core"]
+	path = third_party/boost/core
+	url = https://github.com/boostorg/core

CMakeLists.txt

@@ -80,7 +80,6 @@ set(CMAKE_C_FLAGS_RELEASE  "-O3 -flto -mfpmath=sse -march=x86-64 -mtune=native \
 
 # ############################# End compilation flags
 include_directories(/usr/lib/python3.10/site-packages/numpy/core/include)
-add_subdirectory(third_party/carma)
 
 if(LASP_FFT_BACKEND STREQUAL "FFTW")
     find_library(fftw3 REQUIRED NAMES fftw fftw3)

src/lasp/CMakeLists.txt

@@ -5,14 +5,15 @@ add_definitions(-DARMA_DONT_USE_WRAPPER)
 
 configure_file(lasp_config.h.in lasp_config.h)
 include_directories(${CMAKE_CURRENT_BINARY_DIR})
-include_directories(SYSTEM ../../third_party/carma/include)
-include_directories(SYSTEM
-    ../../third_party/carma/extern/armadillo-code/include)
 include_directories(SYSTEM
     ../../third_party/carma/extern/pybind11/include)
+include_directories(SYSTEM
+    ../../third_party/carma/extern/armadillo-code/include)
+include_directories(SYSTEM ../../third_party/carma/include)
 
 include_directories(../../third_party/DebugTrace-cpp/include)
-include_directories(../../third_party/lockfreeThreadsafe/include)
+include_directories(../../third_party/boost/core/include)
+include_directories(../../third_party/boost/lockfree/include)
 include_directories(../../third_party/gsl-lite/include)
 include_directories(../../third_party/tomlplusplus/include)
 include_directories(../../third_party/thread-pool)

src/lasp/device/lasp_daq.cpp

@@ -2,8 +2,6 @@
 #include "debugtrace.hpp"
 #include "lasp_daqconfig.h"
 
-DEBUGTRACE_VARIABLES;
-
 #include "lasp_daq.h"
 #if LASP_HAS_ULDAQ == 1
 #include "lasp_uldaq.h"
@@ -57,6 +55,7 @@ Daq::Daq(const DeviceInfo &devinfo, const DaqConfiguration &config)
   }
 
 double Daq::samplerate() const {
+  DEBUGTRACE_ENTER;
   return availableSampleRates.at(sampleRateIndex);
 }
 

src/lasp/device/lasp_daqconfig.cpp

@@ -1,4 +1,4 @@
-#define DEBUGTRACE_ENABLED
+/* #define DEBUGTRACE_ENABLED */
 #include "debugtrace.hpp"
 
 #include "lasp_daqconfig.h"

src/lasp/device/lasp_daqconfig.h

@@ -315,6 +315,28 @@ public:
    * enabled.
    */
   int getLowestOutChannel() const;
+
+  /**
+   * @brief Set all input channels to enabled / disabled state.
+   *
+   * @param val true if enabled, false if disabled.
+   */
+  void setAllInputEnabled(bool val) {
+    for(auto& ch: inchannel_config) {
+      ch.enabled = val;
+    }
+  }
+
+  /**
+   * @brief Set all output channels to enabled / disabled state.
+   *
+   * @param val true if enabled, false if disabled.
+   */
+  void setAllOutputEnabled(bool val) {
+    for(auto& ch: outchannel_config) {
+      ch.enabled = val;
+    }
+  }
 };
 /**
  * @}

src/lasp/device/lasp_daqdata.cpp

@@ -1,6 +1,7 @@
-#include "lasp_daqdata.h"
+/* #define DEBUGTRACE_ENABLED */
 #include "debugtrace.hpp"
 #include <cassert>
+#include "lasp_daqdata.h"
 
 DEBUGTRACE_VARIABLES;
 

src/lasp/device/lasp_rtaudiodaq.cpp

@@ -17,6 +17,7 @@ using std::vector;
 DEBUGTRACE_VARIABLES;
 
 void fillRtAudioDeviceInfo(vector<DeviceInfo> &devinfolist) {
+  DEBUGTRACE_ENTER;
 
   vector<RtAudio::Api> apis;
   RtAudio::getCompiledApi(apis);
@@ -106,8 +107,9 @@ void fillRtAudioDeviceInfo(vector<DeviceInfo> &devinfolist) {
   }
 }
 
-static int mycallback(void *outputBuffer, void *inputBuffer, unsigned int nFrames,
-    double streamTime, RtAudioStreamStatus status, void *userData);
+static int mycallback(void *outputBuffer, void *inputBuffer,
+                      unsigned int nFrames, double streamTime,
+                      RtAudioStreamStatus status, void *userData);
 
 static void myerrorcallback(RtAudioError::Type, const string &errorText);
 
@@ -124,7 +126,7 @@ class RtAudioDaq : public Daq {
 
   std::atomic<StreamStatus> _streamStatus{};
 
-  public:
+public:
   RtAudioDaq(const DeviceInfo &devinfo, const DaqConfiguration &config)
       : Daq(devinfo, config),
         rtaudio(static_cast<RtAudio::Api>(devinfo.api.api_specific_subcode)),
@@ -208,7 +210,7 @@ class RtAudioDaq : public Daq {
   }
 
   virtual void start(InDaqCallback inCallback,
-      OutDaqCallback outCallback) override {
+                     OutDaqCallback outCallback) override final {
 
     DEBUGTRACE_ENTER;
 
@@ -226,19 +228,20 @@ class RtAudioDaq : public Daq {
 
     if (inCallback) {
       _incallback = inCallback;
-      if (neninchannels()==0) {
+      if (neninchannels() == 0) {
         throw runtime_error(
             "Input callback given, but stream does not provide input data");
       }
     }
     if (outCallback) {
       _outcallback = outCallback;
-      if (nenoutchannels()==0) {
+      if (nenoutchannels() == 0) {
         throw runtime_error(
             "Output callback given, but stream does not provide output data");
       }
     }
 
+    // Start the stream. Throws on error.
     rtaudio.startStream();
 
     // If we are here, we are running without errors.
@@ -247,9 +250,9 @@ class RtAudioDaq : public Daq {
     _streamStatus = status;
   }
 
-  StreamStatus getStreamStatus() const override { return _streamStatus; }
+  StreamStatus getStreamStatus() const override final { return _streamStatus; }
 
-  void stop() override {
+  void stop() override final {
     DEBUGTRACE_ENTER;
     if (getStreamStatus().runningOK()) {
       rtaudio.stopStream();
@@ -262,10 +265,9 @@ class RtAudioDaq : public Daq {
 
   int streamCallback(void *outputBuffer, void *inputBuffer,
                      unsigned int nFrames, double streamTime,
-
                      RtAudioStreamStatus status) {
 
-    /* DEBUGTRACE_ENTER; */
+    DEBUGTRACE_ENTER;
 
     using se = StreamStatus::StreamError;
 
@@ -305,19 +307,21 @@ class RtAudioDaq : public Daq {
     }
 
     if (inputBuffer) {
+      assert(_incallback);
       std::vector<uint8_t *> ptrs;
       ptrs.reserve(neninchannels);
-      /* DaqData(neninchannels_inc_mon, nFramesPerBlock, dtype); */
+
+      us i = 0;
       for (int ch = getLowestInChannel(); ch <= getHighestInChannel(); ch++) {
         if (inchannel_config.at(ch).enabled) {
-          ptrs.push_back(&static_cast<uint8_t *>(
-                inputBuffer)[sw * ninchannels * ch * nFramesPerBlock]);
+          ptrs.push_back(static_cast<uint8_t *>(inputBuffer) +
+                         sw * i * nFramesPerBlock);
+          i++;
         }
       }
       DaqData d{neninchannels, nFramesPerBlock, dtype};
       d.copyInFromRaw(ptrs);
 
-      assert(_incallback);
       bool ret = _incallback(d);
       if (!ret) {
         stopWithError(se::noError);
@@ -326,20 +330,24 @@ class RtAudioDaq : public Daq {
     }
 
     if (outputBuffer) {
+      assert(_outcallback);
       std::vector<uint8_t *> ptrs;
       ptrs.reserve(nenoutchannels);
 
       /* outCallback */
-      for (int ch = 0; ch <= getHighestOutChannel(); ch++) {
-        /* DEBUGTRACE_PRINT(outchannel_config.at(ch).enabled); */
+      us i = 0;
+      for (int ch = getLowestOutChannel(); ch <= getHighestOutChannel(); ch++) {
+
         if (outchannel_config.at(ch).enabled) {
-          ptrs.push_back(&(static_cast<uint8_t *>(
-                outputBuffer)[sw * nenoutchannels * ch * nFramesPerBlock]));
+
+          ptrs.push_back(static_cast<uint8_t *>(outputBuffer) +
+                         sw * i * nFramesPerBlock);
+
+          i++;
         }
       }
       DaqData d{nenoutchannels, nFramesPerBlock, dtype};
 
-      assert(_outcallback);
       bool ret = _outcallback(d);
       if (!ret) {
         stopWithError(se::noError);

src/lasp/device/lasp_streammgr.cpp

@@ -57,7 +57,14 @@ void StreamMgr::checkRightThread() const {
 
 void StreamMgr::rescanDAQDevices(bool background,
                                  std::function<void()> callback) {
+  DEBUGTRACE_ENTER;
+
   checkRightThread();
+  if(!_devices_mtx.try_lock()) {
+    throw rte("A background DAQ device scan is probably already running");
+  }
+  _devices_mtx.unlock();
+
   if (_inputStream || _outputStream) {
     throw rte("Rescanning DAQ devices only possible when no stream is running");
   }
@@ -70,6 +77,7 @@ void StreamMgr::rescanDAQDevices(bool background,
   }
 }
 void StreamMgr::rescanDAQDevices_impl(std::function<void()> callback) {
+  DEBUGTRACE_ENTER;
   std::scoped_lock lck(_devices_mtx);
   _devices = DeviceInfo::getDeviceInfo();
   if (callback) {
@@ -79,6 +87,7 @@ void StreamMgr::rescanDAQDevices_impl(std::function<void()> callback) {
 bool StreamMgr::inCallback(const DaqData &data) {
 
   /* DEBUGTRACE_ENTER; */
+
   std::scoped_lock<std::mutex> lck(_inDataHandler_mtx);
 
   for (auto &handler : _inDataHandlers) {
@@ -122,8 +131,6 @@ template <typename T> bool fillData(DaqData &data, const vd &signal) {
   /* DEBUGTRACE_ENTER; */
   assert(data.nframes == signal.size());
 
-  /* cerr << "SFSG: data.nframes:" << data.nframes << endl; */
-  /* cerr << "SFSG: data.nchannels:" << data.nchannels << endl; */
   T *res = reinterpret_cast<T *>(data.raw_ptr());
   if (std::is_floating_point<T>()) {
     for (us ch = 0; ch < data.nchannels; ch++) {
@@ -266,7 +273,7 @@ void StreamMgr::startStream(const DaqConfiguration &config) {
 
   if (isInput) {
     _inputStream = std::move(daq);
-    if (daq->duplexMode()) {
+    if (_inputStream->duplexMode()) {
       assert(!_outputStream);
     }
   } else {

src/lasp/device/lasp_streammgr.h

@@ -13,13 +13,13 @@ class StreamMgr;
 
 class InDataHandler {
 
-protected:
+  protected:
     StreamMgr &_mgr;
 #if LASP_DEBUG == 1
     std::atomic<bool> stopCalled{false};
 #endif
 
-public:
+  public:
     virtual ~InDataHandler();
 
     /**
@@ -101,7 +101,7 @@ class StreamMgr {
   // Singleton, no public destructor
   ~StreamMgr();
 
-public:
+  public:
   enum class StreamType : us {
     /**
      * @brief Input stream
@@ -136,7 +136,8 @@ public:
 
   /**
    * @brief Triggers a background scan of the DAQ devices, which updates the
-   * internally stored list of devices.
+   * internally stored list of devices. Throws a runtime error when a
+   * background thread is already scanning for devices.
    *
    * @param background Perform searching for DAQ devices in the background. If
    * set to true, the function returns immediately.
@@ -163,6 +164,17 @@ public:
   bool isStreamRunningOK(const StreamType type) const {
     return getStreamStatus(type).runningOK();
   }
+  bool isStreamRunning(const StreamType type) const {
+    switch(type) {
+      case(StreamType::input): 
+        return bool(_inputStream);
+        break;
+      case(StreamType::output): 
+        return bool(_outputStream);
+        break;
+    }
+    return false;
+  }
 
   /**
    * @brief Get the streamstatus object corresponding to a given stream.
@@ -205,7 +217,7 @@ public:
    */
   void setSiggen(std::shared_ptr<Siggen> s);
 
-private:
+  private:
   bool inCallback(const DaqData &data);
   bool outCallback(DaqData &data);
 

src/lasp/pybind11/lasp_daq.cpp

@@ -12,7 +12,7 @@ void init_daq(py::module &m) {
   py::class_<Daq, DaqConfiguration, DeviceInfo> daq(m, "Daq");
 
   /// Daq::StreamStatus
-  py::class_<Daq::StreamStatus> ss(m, "StreamStatus");
+  py::class_<Daq::StreamStatus> ss(daq, "StreamStatus");
   ss.def("error", &Daq::StreamStatus::error);
   ss.def("runningOK", &Daq::StreamStatus::runningOK);
   ss.def_readonly("isRunning", &Daq::StreamStatus::isRunning);
@@ -29,6 +29,8 @@ void init_daq(py::module &m) {
       .value("apiSpecificError",
              Daq::StreamStatus::StreamError::apiSpecificError);
 
+  ss.def("errorMsg", &Daq::StreamStatus::errorMsg);
+
   /// Daq
   daq.def("neninchannels", &Daq::neninchannels);
   daq.def("nenoutchannels", &Daq::nenoutchannels);

src/lasp/pybind11/lasp_daqconfig.cpp

@@ -91,4 +91,6 @@ void init_daqconfiguration(py::module &m) {
                           &DaqConfiguration::inchannel_config);
   daqconfig.def_readwrite("outchannel_config",
                           &DaqConfiguration::outchannel_config);
+  daqconfig.def("setAllInputEnabled", &DaqConfiguration::setAllInputEnabled);
+  daqconfig.def("setAllOutputEnabled", &DaqConfiguration::setAllOutputEnabled);
 }

src/lasp/pybind11/lasp_dsp_pybind.cpp

@@ -56,7 +56,7 @@ void init_dsp(py::module &m) {
   });
   
   /// BiquadBank
-  py::class_<BiquadBank> bqb(m, "BiquadBank");
+  py::class_<BiquadBank, std::shared_ptr<BiquadBank>> bqb(m, "BiquadBank");
   bqb.def(py::init<const dmat&,const vd*>());
   bqb.def("setGains",&BiquadBank::setGains);
   bqb.def("filter",&BiquadBank::filter);

src/lasp/pybind11/lasp_streammgr.cpp

@@ -29,4 +29,5 @@ void init_streammgr(py::module &m) {
   smgr.def("getDeviceInfo", &StreamMgr::getDeviceInfo);
   smgr.def("getStreamStatus", &StreamMgr::getStreamStatus);
   smgr.def("isStreamRunningOK", &StreamMgr::isStreamRunningOK);
+  smgr.def("isStreamRunning", &StreamMgr::isStreamRunning);
 }

third_party/boost/core

@@ -0,0 +1 @@
+Subproject commit b407b5d87df30f75ca501c1b6f2930c0913d4ca7

third_party/rtaudio

@@ -0,0 +1 @@
+Subproject commit a3e9aa621e65a0744f125e9740d057a4d088e0e9