Mixer: gain stages, telemetry arena, port changes between ticks

examples/Dataflow/MiniDAW.cpp

@@ -28,6 +28,7 @@ struct tick
   {
     proto.setup_buffers(st);
     (*this)(st.frames, st.seconds);
+    proto.apply_main_gain(st);
   }
 
   void operator()(unsigned long frameCount, double seconds)

src/ossia/audio/audio_engine.cpp

@@ -10,6 +10,8 @@
 #include <ossia/audio/sdl_protocol.hpp>
 #include <ossia/detail/logger.hpp>
 
+#include <chrono>
+#include <cmath>
 #include <thread>
 
 namespace ossia
@@ -87,6 +89,42 @@ void audio_engine::sync()
   }
 }
 
+void audio_engine::run_between_ticks(const std::function<void()>& f)
+{
+  if(!running())
+  {
+    f();
+    return;
+  }
+
+  m_betweenDone.store(false, std::memory_order_relaxed);
+  m_between.store(&f, std::memory_order_release);
+
+  int buffer_ms = 8;
+  if(effective_sample_rate > 0)
+    buffer_ms = std::max(
+        1, int(std::ceil(effective_buffer_size * 1000. / effective_sample_rate)));
+
+  // A few buffers' worth: a callback that does not come by then is not coming.
+  const auto deadline
+      = std::chrono::steady_clock::now() + std::chrono::milliseconds(8 * buffer_ms + 50);
+  while(!m_betweenDone.load(std::memory_order_acquire))
+  {
+    if(std::chrono::steady_clock::now() > deadline)
+    {
+      // Taken back before the callback took it: it is safe to run it here.
+      if(m_between.exchange(nullptr, std::memory_order_acq_rel) == &f)
+      {
+        ossia::logger().warn("Audio engine: no callback, running the change directly");
+        f();
+        return;
+      }
+      // The callback took it and is running it: it will be done shortly.
+    }
+    std::this_thread::sleep_for(std::chrono::microseconds(100));
+  }
+}
+
 void audio_engine::gc()
 {
   // try to make deallocations happen in the main thread as far as possible
@@ -108,6 +146,12 @@ void audio_engine::set_tick(audio_engine::fun_type&& t)
 
 void audio_engine::load_audio_tick()
 {
+  if(auto f = m_between.exchange(nullptr, std::memory_order_acq_rel))
+  {
+    (*f)();
+    m_betweenDone.store(true, std::memory_order_release);
+  }
+
   fun_type tick;
   while(tick_funlist.try_dequeue(tick))
   {

src/ossia/audio/audio_engine.hpp

@@ -6,6 +6,7 @@
 #include <smallfun.hpp>
 
 #include <atomic>
+#include <functional>
 
 #include <ossia-config.hpp>
 
@@ -24,6 +25,14 @@ class OSSIA_EXPORT audio_engine
   void gc();
   void sync();
 
+  //! Runs f on the audio thread at the start of the next callback, before
+  //! its tick, and returns once it ran: f may change what the tick reads, such
+  //! as the audio protocol's ports. The audio does not stop; the caller waits
+  //! for about one buffer. f must neither allocate nor free, nor block.
+  //! When no callback comes, f runs on the calling thread. Never call it from
+  //! the audio thread.
+  void run_between_ticks(const std::function<void()>& f);
+
   using fun_type = smallfun::function<void(const ossia::audio_tick_state&), 256>;
   void set_tick(fun_type&& t);
   void load_audio_tick();
@@ -52,6 +61,10 @@ class OSSIA_EXPORT audio_engine
   std::shared_ptr<audio_engine> self;
   std::atomic_int64_t request{-1};
   std::atomic_int64_t reply{-1};
+
+  // What run_between_ticks hands to the callback, and whether it ran.
+  std::atomic<const std::function<void()>*> m_between{};
+  std::atomic_bool m_betweenDone{};
 };
 
 OSSIA_EXPORT

src/ossia/audio/audio_parameter.cpp

@@ -3,8 +3,11 @@
 #include "audio_protocol.hpp"
 
 #include <ossia/dataflow/execution_state.hpp>
+#include <ossia/dataflow/telemetry.hpp>
 #include <ossia/network/common/complex_type.hpp>
 
+#include <array>
+
 namespace ossia
 {
 
@@ -36,19 +39,41 @@ void audio_parameter::clone_value(audio_vector& res_vec) const
   {
     auto& src = audio[chan];
     auto& res = res_vec[chan];
+    const double g = read_gain(chan);
 
     const std::size_t N = src.size();
 
     if(res.size() < N)
       res.resize(N);
 
     for(std::size_t i = 0; i < N; i++)
-      res[i] += double(src[i]);
+      res[i] += double(src[i]) * g;
+  }
+}
+
+int audio_parameter::hardware_channel(std::size_t chan) const noexcept
+{
+  return int(chan);
+}
+
+double audio_parameter::read_gain(std::size_t chan) const noexcept
+{
+  double g = stage == gain_stage::pull ? gain() : 1.;
+  if(upstream)
+    g *= upstream->gain();
+  if(hardware_inputs)
+  {
+    const int c = hardware_channel(chan);
+    if(c >= 0 && c < int(hardware_inputs->size()))
+      if(auto trim = (*hardware_inputs)[c]; trim && trim != this)
+        g *= trim->gain();
   }
+  return g;
 }
 
 void audio_parameter::push_value(const audio_port& port)
 {
+  const double g = stage == gain_stage::push ? gain() : 1.;
   auto min_chan = std::min(port.channels(), (std::size_t)audio.size());
   for(std::size_t chan = 0; chan < min_chan; chan++)
   {
@@ -57,7 +82,7 @@ void audio_parameter::push_value(const audio_port& port)
     const auto N = std::min(src.size(), (std::size_t)dst.size());
     for(std::size_t i = 0; i < N; i++)
     {
-      dst[i] += float(src[i] * m_gain);
+      dst[i] += float(src[i] * g);
     }
   }
 }
@@ -83,19 +108,16 @@ net::parameter_base& audio_parameter::push_value()
 
 value audio_parameter::value() const
 {
-  return m_gain;
+  return gain();
 }
 
 ossia::value audio_parameter::set_value(const ossia::value& v)
 {
   auto flt = ossia::convert<float>(v);
   auto vol = ossia::clamp(flt, 0.f, 1.f);
-  if(m_gain != vol)
-  {
-    m_gain = vol;
+  if(m_gain.exchange(vol, std::memory_order_relaxed) != vol)
     send(vol);
-  }
-  return m_gain;
+  return vol;
 }
 
 ossia::value audio_parameter::set_value(ossia::value&& v)
@@ -153,8 +175,21 @@ virtual_audio_parameter::virtual_audio_parameter(int num_channels, net::node_bas
   proto.register_parameter(*this);
 }
 
+auto virtual_audio_parameter::make_channels(int num_channels) const -> channels
+{
+  const auto bs = m_audio_data.empty() ? std::size_t(512) : m_audio_data.front().size();
+  channels c;
+  c.data.resize(std::max(num_channels, 0));
+  for(auto& d : c.data)
+    d.resize(bs);
+  for(auto& d : c.data)
+    c.spans.push_back(d);
+  return c;
+}
+
 void virtual_audio_parameter::push_value(const audio_port& port)
 {
+  const double g = gain();
   auto min_chan = std::min(port.channels(), (std::size_t)audio.size());
   for(std::size_t chan = 0; chan < min_chan; chan++)
   {
@@ -164,9 +199,18 @@ void virtual_audio_parameter::push_value(const audio_port& port)
     for(std::size_t i = 0; i < N; i++)
     {
       // Important: here we must not mix
-      dst[i] = float(src[i] * m_gain);
+      dst[i] = float(src[i] * g);
     }
   }
+
+  if(auto m = meter.load(std::memory_order_acquire))
+  {
+    std::array<const float*, ossia::telemetry::max_meter_channels> chans;
+    const auto n = std::min(audio.size(), chans.size());
+    for(std::size_t c = 0; c < n; c++)
+      chans[c] = audio[c].data();
+    m->pending.accumulate(chans.data(), int(n), n > 0 ? audio[0].size() : 0);
+  }
 }
 
 virtual_audio_parameter::~virtual_audio_parameter()
@@ -182,10 +226,16 @@ mapped_audio_parameter::mapped_audio_parameter(
     , mapping(std::move(m))
     , is_output{output}
 {
+  stage = output ? gain_stage::push : gain_stage::pull;
   auto& proto = static_cast<ossia::audio_protocol&>(n.get_device().get_protocol());
   proto.register_parameter(*this);
 }
 
+int mapped_audio_parameter::hardware_channel(std::size_t chan) const noexcept
+{
+  return !is_output && chan < mapping.size() ? mapping[chan] : -1;
+}
+
 mapped_audio_parameter::~mapped_audio_parameter()
 {
   auto& proto

src/ossia/audio/audio_parameter.hpp

@@ -7,17 +7,46 @@
 #include <ossia/network/base/parameter.hpp>
 #include <ossia/network/value/value.hpp>
 
+#include <atomic>
+#include <vector>
+
+namespace ossia::telemetry
+{
+struct meter_tap;
+}
+
 namespace ossia
 {
 class OSSIA_EXPORT audio_parameter : public ossia::net::parameter_base
 {
 
 public:
   ossia::small_vector<std::span<float>, 8> audio;
-  double m_gain{1.};
+
+  //! Where gain() is applied, so that it is applied exactly once.
+  enum class gain_stage : uint8_t
+  {
+    push,     //!< on what the graph writes (outputs, virtual ports)
+    pull,     //!< on what the graph reads (inputs)
+    external, //!< by whoever owns the buffers, after the sum (/out/main, /out/N)
+  } stage{gain_stage::push};
+
+  //! A parameter whose gain also applies to this one: /in/main for the
+  //! inputs.
+  const audio_parameter* upstream{};
+
+  const std::vector<audio_parameter*>* hardware_inputs{};
+  [[nodiscard]] virtual int hardware_channel(std::size_t chan) const noexcept;
+  [[nodiscard]] double read_gain(std::size_t chan) const noexcept;
 
   explicit audio_parameter(ossia::net::node_base& n);
 
+  //! Linear gain, readable from any thread.
+  [[nodiscard]] float gain() const noexcept
+  {
+    return m_gain.load(std::memory_order_relaxed);
+  }
+
   virtual ~audio_parameter();
 
   void clone_value(audio_vector& res) const;
@@ -38,6 +67,9 @@ class OSSIA_EXPORT audio_parameter : public ossia::net::parameter_base
   net::parameter_base& set_domain(const domain&) override;
   bounding_mode get_bounding() const noexcept override;
   net::parameter_base& set_bounding(bounding_mode) override;
+
+protected:
+  std::atomic<float> m_gain{1.f};
 };
 
 class OSSIA_EXPORT virtual_audio_parameter final : public audio_parameter
@@ -48,6 +80,29 @@ class OSSIA_EXPORT virtual_audio_parameter final : public audio_parameter
 public:
   virtual_audio_parameter(int num_channels, ossia::net::node_base& n);
 
+  //! Buffers for another channel count, made off the audio thread.
+  struct channels
+  {
+    std::vector<ossia::float_vector> data;
+    ossia::small_vector<std::span<float>, 8> spans;
+  };
+  [[nodiscard]] channels make_channels(int num_channels) const;
+  //! Takes buffers made by make_channels, between two ticks; `c` gets the
+  //! previous ones, to be freed off the audio thread.
+  void swap_channels(channels& c) noexcept
+  {
+    std::swap(m_audio_data, c.data);
+    std::swap(audio, c.spans);
+  }
+
+  //! Changes the channel count; only while the audio callback does not run.
+  void set_channels(int num_channels)
+  {
+    const auto bs = m_audio_data.empty() ? 0 : m_audio_data.front().size();
+    m_audio_data.resize(std::max(num_channels, 0));
+    set_buffer_size(int(bs));
+  }
+
   void set_buffer_size(int bs)
   {
     const auto chan = m_audio_data.size();
@@ -61,6 +116,11 @@ class OSSIA_EXPORT virtual_audio_parameter final : public audio_parameter
 
   void push_value(const audio_port& port) override;
 
+  //! Meters what the graph writes here. Set by the interface thread; the
+  //! telemetry arena owns the tap and keeps it alive until the audio thread
+  //! can no longer be writing to it.
+  std::atomic<ossia::telemetry::meter_tap*> meter{};
+
   virtual ~virtual_audio_parameter();
 };
 
@@ -72,6 +132,8 @@ class OSSIA_EXPORT mapped_audio_parameter final : public audio_parameter
   bool is_output{false};
   mapped_audio_parameter(bool output, audio_mapping m, ossia::net::node_base& n);
 
+  [[nodiscard]] int hardware_channel(std::size_t chan) const noexcept override;
+
   virtual ~mapped_audio_parameter();
 };
 }