score-plugin-avnd/Crousti/Executor.hpp

#pragma once
 
#include <Process/Execution/ProcessComponent.hpp>
#include <Process/ExecutionContext.hpp>
#include <Process/ExecutionSetup.hpp>
 
#include <Explorer/DocumentPlugin/DeviceDocumentPlugin.hpp>
 
#include <Crousti/CpuAnalysisNode.hpp>
#include <Crousti/CpuFilterNode.hpp>
#include <Crousti/ExecutorPortSetup.hpp>
#include <Crousti/ExecutorUpdateControlValueInUi.hpp>
#include <Crousti/File.hpp>
#include <Crousti/GpuComputeNode.hpp>
#include <Crousti/GpuNode.hpp>
#include <Crousti/MessageBus.hpp>
#include <Crousti/Metadatas.hpp>
#include <Crousti/ProcessModel.hpp>
 
#include <score/tools/Bind.hpp>
#include <ossia/detail/flat_set.hpp>
#include <ossia/dataflow/exec_state_facade.hpp>
#include <ossia/dataflow/graph/graph_interface.hpp>
#include <ossia/dataflow/graph_edge.hpp>
#include <ossia/dataflow/node_process.hpp>
#include <ossia/network/context.hpp>
 
#include <ossia-qt/invoke.hpp>
 
#include <QGuiApplication>
 
#include <flicks.h>
 
#if SCORE_PLUGIN_GFX
#include <Crousti/GpuNode.hpp>
#include <Gfx/GfxApplicationPlugin.hpp>
#endif
 
#include <Scenario/Settings/ScenarioSettingsModel.hpp>
 
#include <score/tools/ThreadPool.hpp>
 
#include <ossia/detail/type_if.hpp>
 
#include <QTimer>
 
#include <atomic>
 
#include <avnd/binding/ossia/data_node.hpp>
#include <avnd/binding/ossia/mono_audio_node.hpp>
#include <avnd/binding/ossia/node.hpp>
#include <avnd/binding/ossia/ossia_audio_node.hpp>
#include <avnd/binding/ossia/ossia_dynamic_audio_node.hpp>
#include <avnd/binding/ossia/poly_audio_node.hpp>
#include <avnd/concepts/dynamic_items.hpp>
#include <avnd/concepts/temporality.hpp>
#include <avnd/concepts/ui.hpp>
#include <avnd/concepts/worker.hpp>
#include <Crousti/WorkerBinding.hpp>
 
#include <QPointer>
 
#include <atomic>
 
namespace oscr
{
[[nodiscard]] static ossia::value
gpuControlValue(
    const Process::ControlInlet& ctrl, const ossia::value& v,
    const score::DocumentContext& ctx)
{
  if(qobject_cast<const Process::FileChooserBase*>(&ctrl)
     || qobject_cast<const Process::FolderChooser*>(&ctrl))
    return resolvePathValue(v, ctx);
  return v;
}
 
template <typename Node>
struct node_with_worker : node_with_ui_wakeups<Node>
{
  using node_with_ui_wakeups<Node>::node_with_ui_wakeups;
 
  Execution::ExecutionCommandQueue worker_results;
 
  void
  run(const ossia::token_request& tk, ossia::exec_state_facade st) noexcept override
  {
    Execution::ExecutionCommand cmd;
    if(worker_results.try_dequeue(cmd))
    {
      const auto [start, frames] = st.timings(tk);
      this->start_frame_for_this_tick = start;
      this->frame_count_for_this_tick = frames;
      do
      {
        cmd();
      } while(worker_results.try_dequeue(cmd));
    }
 
    node_with_ui_wakeups<Node>::run(tk, st);
  }
};
 
template <typename Node>
struct exec_worker_delivery
{
  std::weak_ptr<node_with_worker<Node>> node_wp;
  exec_worker_delivery begin() const noexcept { return *this; }
 
  template <typename F>
  void operator()(F&& apply)
  {
    ossia::qt::run_async(
        qApp, [node_wp = std::move(node_wp), apply = std::move(apply)]() mutable {
      if(std::shared_ptr n = node_wp.lock())
        n->worker_results.enqueue(std::move(apply));
    });
  }
};
 
template <typename Node>
using exec_base_node_t = std::conditional_t<
    avnd::has_worker<Node>, node_with_worker<Node>, node_with_ui_wakeups<Node>>;
 
template <typename Node>
concept stops_with_extra_tick = avnd::tag_process_exec<Node>
                                && avnd::midi_output_introspection<Node>::size > 0
                                && requires(Node& n) { n.stop(); };
 
template <typename Node>
struct node_with_deferred_stop : exec_base_node_t<Node>
{
  using exec_base_node_t<Node>::exec_base_node_t;
 
  std::atomic_bool stop_pending{};
 
  ~node_with_deferred_stop()
  {
    if(stop_pending.exchange(false))
      this->impl.effect.stop();
  }
 
  void
  run(const ossia::token_request& tk, ossia::exec_state_facade st) noexcept override
  {
    if(tk.date == tk.prev_date && stop_pending.exchange(false))
      this->impl.effect.stop();
    exec_base_node_t<Node>::run(tk, st);
  }
};
 
template <typename Node>
using exec_node_t = std::conditional_t<
    stops_with_extra_tick<Node>, node_with_deferred_stop<Node>, exec_base_node_t<Node>>;
 
template <typename Node>
class CustomNodeProcess : public ossia::node_process
{
  using node_process::node_process;
  void start() override
  {
    node_process::start();
    auto& n = static_cast<safe_node<Node>&>(*node);
    if_possible(n.impl.effect.start());
  }
  void pause() override
  {
    auto& n = static_cast<safe_node<Node>&>(*node);
    if_possible(n.impl.effect.pause());
    node_process::pause();
  }
 
  void resume() override
  {
    node_process::resume();
    auto& n = static_cast<safe_node<Node>&>(*node);
    if_possible(n.impl.effect.resume());
  }
 
  void stop() override
  {
    auto& n = static_cast<safe_node<Node>&>(*node);
    if constexpr(stops_with_extra_tick<Node>)
    {
      // A stopped node is not ticked again: one more, empty tick sends the
      // note-offs, and stops the object at its start.
      static_cast<node_with_deferred_stop<Node>&>(*node).stop_pending = true;
      node_process::stop();
      n.request(ossia::token_request{});
    }
    else
    {
      if_possible(n.impl.effect.stop());
      node_process::stop();
    }
  }
 
  void offset_impl(ossia::time_value date) override
  {
    node_process::offset_impl(date);
    auto& n = static_cast<safe_node<Node>&>(*node);
    util::flicks f{date.impl};
    if_possible(n.impl.effect.transport(f));
  }
 
  void transport_impl(ossia::time_value date) override
  {
    node_process::transport_impl(date);
    auto& n = static_cast<safe_node<Node>&>(*node);
 
    util::flicks f{date.impl};
    if_possible(n.impl.effect.transport(f));
  }
};
 
template <typename Node>
class Executor final
    : public Execution::ProcessComponent_T<ProcessModel<Node>, ossia::node_process>
    , public dynamic_ports_component_data<Node>
{
public:
  static Q_DECL_RELAXED_CONSTEXPR UuidKey<score::Component> static_key() noexcept
  {
    return uuid_from_string<Node>();
  }
 
  UuidKey<score::Component> key() const noexcept final override { return static_key(); }
 
  bool key_match(UuidKey<score::Component> other) const noexcept final override
  {
    return static_key() == other || Execution::ProcessComponent::base_key_match(other);
  }
 
#if defined(SCORE_PLUGIN_GFX)
  [[no_unique_address]] ossia::type_if<int, is_gpu<Node>> node_id = score::gfx::invalid_node_index;
#endif
 
  std::vector<QMetaObject::Connection> m_executionUpdate;
 
  template <typename F>
  void followExecutionUpdate(F f)
  {
    auto& settings = score::AppContext().settings<Scenario::Settings::Model>();
    f(settings.getExecutionUpdate());
    m_executionUpdate.push_back(QObject::connect(
        &settings, &Scenario::Settings::Model::ExecutionUpdateChanged, this,
        std::move(f)));
  }
 
  Executor(ProcessModel<Node>& element, const ::Execution::Context& ctx, QObject* p)
      : Execution::ProcessComponent_T<ProcessModel<Node>, ossia::node_process>{
          element, ctx, "Executor::ProcessModel<Info>", p}
  {
    if constexpr(is_gpu<Node>)
    {
#if SCORE_PLUGIN_GFX
      setup_gpu(element, ctx, p);
#endif
    }
    else
    {
      setup_cpu(element, ctx, p);
    }
 
    if constexpr(avnd::tag_process_exec<Node>)
    {
      this->m_ossia_process = std::make_shared<CustomNodeProcess<Node>>(this->node);
    }
    else
    {
      this->m_ossia_process = std::make_shared<ossia::node_process>(this->node);
    }
  }
 
  void
  setup_cpu(ProcessModel<Node>& element, const ::Execution::Context& ctx, QObject* p)
  {
    auto& net_ctx
        = *ctx.doc.findPlugin<Explorer::DeviceDocumentPlugin>()->networkContext();
    const auto id
        = std::hash<ObjectPath>{}(Path<Process::ProcessModel>{element}.unsafePath());
 
    auto st = ossia::exec_state_facade{ctx.execState.get()};
    std::shared_ptr<safe_node<Node>> ptr;
    auto node = new exec_node_t<Node>{st.setupBufferSize(), (double)st.sampleRate(), id};
    node->root_inputs().reserve(element.inlets().size());
    node->root_outputs().reserve(element.outlets().size());
 
    node->prepare(*ctx.execState.get()); // Preparation of the ossia side
 
    if_possible(node->impl.effect.ossia_state = st);
    if_possible(node->impl.effect.io_context = &net_ctx.context);
    if_possible(node->impl.effect.ossia_document_context = &ctx.doc);
    ptr.reset(node);
    this->node = ptr;
 
    if constexpr(requires { ptr->impl.effect; })
      if constexpr(std::is_same_v<std::decay_t<decltype(ptr->impl.effect)>, Node>)
      {
        connect_message_bus(element, ctx, ptr->impl.effect, node->wakeups);
        connect_dynamic_items(element, ptr->impl.effect);
        // Views follow execution only when the user lets them, as for every
        // execution-to-UI update: without a monitor they stop being viewers,
        // and the node stops writing for them
        if constexpr(requires { element.setMonitor(ptr->impl.effect.monitor); })
          followExecutionUpdate(
              [&element, monitor = ptr->impl.effect.monitor](bool enabled) {
            element.setMonitor(enabled ? monitor : nullptr);
          });
      }
    connect_worker(ptr);
 
    node->dynamic_ports = element.dynamic_ports;
    node->finish_init();
 
    connect_controls(element, ctx, ptr);
    update_controls(ptr);
 
    // To call prepare() after evertyhing is ready
    node->audio_configuration_changed(st);
 
    if constexpr(oscr::has_dynamic_ports<Node>)
    {
      QObject::connect(
          &element, &Process::ProcessModel::inletsChanged, this,
          &Executor::recompute_ports);
      QObject::connect(
          &element, &Process::ProcessModel::outletsChanged, this,
          &Executor::recompute_ports);
 
      this->m_oldInlets = element.inlets();
      this->m_oldOutlets = element.outlets();
    }
  }
 
  void
  setup_gpu(ProcessModel<Node>& element, const ::Execution::Context& ctx, QObject* p)
  {
#if SCORE_PLUGIN_GFX
    // FIXME net context for gpu node ?
    const int64_t id
        = std::hash<ObjectPath>{}(Path<Process::ProcessModel>{element}.unsafePath());
 
    // The monitor its views share with every renderer of the node: they take
    // it from the process when they are made, while the user lets execution
    // update the views
    if constexpr(requires { Node::monitor; })
    {
      using monitor_type = typename decltype(Node::monitor)::element_type;
      followExecutionUpdate(
          [&element, monitor = std::make_shared<monitor_type>()](bool enabled) {
        element.setMonitor(enabled ? monitor : nullptr);
      });
    }
 
    auto& gfx_exec = ctx.doc.plugin<Gfx::DocumentPlugin>().exec;
 
    // Create the executor in the audio thread
    struct named_exec_node final : Gfx::gfx_exec_node
    {
      using Gfx::gfx_exec_node::gfx_exec_node;
      std::string label() const noexcept override
      {
        return std::string(avnd::get_name<Node>());
      }
    };
 
    auto node = std::make_shared<named_exec_node>(gfx_exec);
    node->prepare(*ctx.execState);
 
    this->node = node;
 
    // Create the controls, inputs outputs etc.
    std::size_t i = 0;
 
    for(auto& ctl : element.inlets())
    {
      if(auto ctrl = qobject_cast<Process::ControlInlet*>(ctl))
      {
        auto& p = node->add_control();
        p->value = gpuControlValue(*ctrl, ctrl->value(), ctx.doc);
        // An impulse is only ever sent when it is triggered: the object's
        // one-shot inputs start empty, like setup_control_for_exec leaves them
        // on the CPU path.
        p->changed = p->value.get_type() != ossia::val_type::IMPULSE;
 
        // What the control holds and the range it takes, so that a cable into
        // it converts and rescales. The ISF executor sets its controls up the
        // same way.
        ctrl->setupExecution(*node->root_inputs().back(), this);
 
        QObject::connect(
            ctrl, &Process::ControlInlet::valueChanged, this,
            [con = Gfx::con_unvalidated{ctx, i, 0, node}, ctrl,
             &doc = ctx.doc](const ossia::value& v) mutable {
          con(gpuControlValue(*ctrl, v, doc));
        });
        i++;
      }
      else if(auto ctrl = qobject_cast<Process::ValueInlet*>(ctl))
      {
        auto& p = node->add_control();
        p->changed = false;
        i++;
      }
      else if(auto ctrl = qobject_cast<Process::AudioInlet*>(ctl))
      {
        node->add_audio();
      }
      else if(auto ctrl = qobject_cast<Gfx::TextureInlet*>(ctl))
      {
        ossia::texture_inlet& inl = *node->add_texture();
        ctrl->setupExecution(inl, this);
      }
      else if(auto ctrl = qobject_cast<Gfx::GeometryInlet*>(ctl))
      {
        ossia::geometry_inlet& inl = *node->add_geometry();
        ctrl->setupExecution(inl, this);
      }
    }
 
    // FIXME refactor this with other GFX processes
    for(auto* outlet : element.outlets())
    {
      if(auto ctrl = qobject_cast<Process::ControlOutlet*>(outlet))
      {
        node->add_control_out();
      }
      else if(auto ctrl = qobject_cast<Process::ValueOutlet*>(outlet))
      {
        node->add_control_out();
      }
      else if(auto out = qobject_cast<Gfx::TextureOutlet*>(outlet))
      {
        node->add_texture_out();
      }
      else if(auto out = qobject_cast<Gfx::GeometryOutlet*>(outlet))
      {
        node->add_geometry_out();
      }
    }
 
    // Create the GPU node
    std::weak_ptr qex_ptr = std::shared_ptr<Execution::ExecutionCommandQueue>(
        ctx.alias.lock(), &ctx.executionQueue);
    std::unique_ptr<score::gfx::Node> ptr;
    if constexpr(GpuGraphicsNode2<Node>)
    {
      auto gpu_node = new CustomGpuNode<Node>(qex_ptr, node->control_outs, id, ctx.doc);
      ptr.reset(gpu_node);
    }
    else if constexpr(GpuComputeNode2<Node>)
    {
      auto gpu_node = new GpuComputeNode<Node>(qex_ptr, node->control_outs, id, ctx.doc);
      ptr.reset(gpu_node);
    }
    else if constexpr(GpuNode<Node>)
    {
      auto gpu_node
          = new GfxNode<Node>(element, qex_ptr, node->control_outs, id, ctx.doc);
      ptr.reset(gpu_node);
    }
 
    i = 0;
    for(auto& ctl : element.inlets())
    {
      if(auto ctrl = qobject_cast<Gfx::TextureInlet*>(ctl))
      {
        ossia::texture_inlet& inl
            = static_cast<ossia::texture_inlet&>(*node->root_inputs()[i]);
        ptr->process(i, inl.data); // Setup render_target_spec
      }
      i++;
    }
    node->id = gfx_exec.ui->register_node(std::move(ptr));
    node_id = node->id;
 
    // Control outputs reach the model's outlets, for the views and the
    // inspector, at the UI rate
    if(!node->control_outs.empty())
    {
      std::vector<Process::ControlOutlet*> ui_outlets;
      for(auto* outlet : element.outlets())
        if(auto ctl = qobject_cast<Process::ControlOutlet*>(outlet))
          ui_outlets.push_back(ctl);
        else if(qobject_cast<Process::ValueOutlet*>(outlet))
          ui_outlets.push_back(nullptr);
      followExecutionUpdate([weak = std::weak_ptr{node}](bool enabled) {
        if(auto n = weak.lock())
          n->notify_ui.store(enabled, std::memory_order_relaxed);
      });
      con(ctx.doc.coarseUpdateTimer, &QTimer::timeout, this,
          [weak = std::weak_ptr{node}, ui_outlets = std::move(ui_outlets),
           v = ossia::value{}]() mutable {
        auto n = weak.lock();
        if(!n || !n->notify_ui.load(std::memory_order_relaxed))
          return;
        const int count = std::min(int(ui_outlets.size()), int(n->ui_control_outs.size()));
        for(int i = 0; i < count; i++)
          if(ui_outlets[i] && n->ui_control_outs[i]->latest.consume(v))
            ui_outlets[i]->setValue(v);
      }, Qt::QueuedConnection);
    }
 
    // Only known once registered: previews of the outlets connect to it
    for(auto* outlet : element.outlets())
      if(auto out = qobject_cast<Gfx::TextureOutlet*>(outlet))
        out->nodeId = node_id;
#endif
  }
 
  void recompute_ports()
  {
    if constexpr(oscr::has_dynamic_ports<Node> && !is_gpu<Node>)
    {
      using T = Node;
      auto n = std::dynamic_pointer_cast<safe_node<Node>>(this->node);
      if(!n)
        return;
 
      auto& element = this->process();
      const Execution::Context& ctx = this->system();
      Execution::SetupContext& setup = ctx.setup;
 
      // The ports the exec node is currently set up for. Those that are not
      // in the model anymore are about to be deleted (they are still alive
      // here: ProcessModel deletes them after inletsChanged()).
      const Process::Inlets old_inlets = std::move(this->m_oldInlets);
      const Process::Outlets old_outlets = std::move(this->m_oldOutlets);
      const Process::Inlets& new_inlets = element.inlets();
      const Process::Outlets& new_outlets = element.outlets();
 
      for(auto port : old_inlets)
        if(!ossia::contains(new_inlets, port))
          this->m_connectedControls.erase(port);
 
      // The cables at either end of this process that are wired up in the
      // graph: their edges point to the exec ports that are being replaced.
      // A cable that is on the ports but not wired yet (restored by a command
      // that announces it after inletsChanged()) is left to that announcement.
      ossia::small_vector<Process::Cable*, 8> cables;
      auto collect_cables = [&](const auto& ports) {
        for(auto port : ports)
          for(auto& cbl : port->cables())
            if(auto c = cbl.try_find(ctx.doc))
              if(setup.m_cables.find(c->id()) != setup.m_cables.end())
                if(!ossia::contains(cables, c))
                  cables.push_back(c);
      };
      collect_cables(old_inlets);
      collect_cables(old_outlets);
      collect_cables(new_inlets);
      collect_cables(new_outlets);
 
      // Build the new set of exec ports.
      struct port_storage
      {
        ossia::inlets new_inls_buffer;
        ossia::outlets new_outls_buffer;
        inlet_reload_storage<T> reload_inlet;
        outlet_reload_storage<T> reload_outlet;
        port_storage(std::size_t ins, std::size_t outs)
        {
          new_inls_buffer.reserve(ins);
          new_outls_buffer.reserve(outs);
        }
      };
      auto port_st = std::make_shared<port_storage>(new_inlets.size(), new_outlets.size());
      auto& inbuf = port_st->new_inls_buffer;
      auto& outbuf = port_st->new_outls_buffer;
 
      // Same order as safe_node_base::reinit() + initialize_all_ports():
      // first the ports that do not come from a field of the object
      // (audio / value argument of the processor, message inlets), then the fields.
      n->audio_ports.init(inbuf, outbuf);
      n->arg_value_ports.init(inbuf, outbuf);
      n->message_ports.init(inbuf);
 
      auto& dp = element.dynamic_ports;
      if constexpr(avnd::inputs_type<T>::size > 0)
      {
        using in_info = avnd::input_introspection<T>;
        using in_type = typename avnd::inputs_type<T>::type;
        auto& port_tuple = n->ossia_inlets.ports;
 
        [&]<typename K, K... Index>(std::integer_sequence<K, Index...>) {
          reload_inlets<safe_node_base_base<T>> init{*n, inbuf, dp};
          (init(
               avnd::field_reflection<Index, avnd::pfr::tuple_element_t<Index, in_type>>{},
               tuplet::get<Index>(port_tuple),
               tuplet::get<Index>(port_st->reload_inlet.ports)),
           ...);
        }(typename in_info::indices_n{});
      }
 
      if constexpr(avnd::outputs_type<T>::size > 0)
      {
        using out_info = avnd::output_introspection<T>;
        using out_type = typename avnd::outputs_type<T>::type;
        auto& port_tuple = n->ossia_outlets.ports;
 
        [&]<typename K, K... Index>(std::integer_sequence<K, Index...>) {
          reload_outlets<safe_node_base_base<T>> init{*n, outbuf, dp};
          (init(
               avnd::field_reflection<
                   Index, avnd::pfr::tuple_element_t<Index, out_type>>{},
               tuplet::get<Index>(port_tuple),
               tuplet::get<Index>(port_st->reload_outlet.ports)),
           ...);
        }(typename out_info::indices_n{});
      }
 
      // The exec node must line up with the model port for port, otherwise
      // cables and addresses would land on the wrong exec port.
      SCORE_SOFT_ASSERT(inbuf.size() == new_inlets.size());
      SCORE_SOFT_ASSERT(outbuf.size() == new_outlets.size());
 
      Execution::Transaction commands{ctx};
 
      // 1. The edges of the cables, before their ports go away
      for(auto c : cables)
        setup.removeCable(*c, commands);
 
      // 2. Old ports out of the execution state and of the setup maps
      setup.unregister_node_soft(old_inlets, old_outlets, this->node, commands);
 
      // 3. Swap the ports in
      commands.push_back(
          [node = n, dp, port_st, gcq = ctx.weakGCQueue(),
           wg = std::weak_ptr{ctx.execGraph}]() mutable {
        OSSIA_ENSURE_CURRENT_THREAD_KIND(ossia::thread_type::Audio);
 
        // Nothing may keep an edge to a port that is about to be freed: the
        // cables were disconnected above, whatever else is still there (the
        // interval's audio propagation, or an edge nobody tracks) is dropped.
        // Ports that stay keep their edges.
        if(auto g = wg.lock())
        {
          const auto& inbuf = port_st->new_inls_buffer;
          const auto& outbuf = port_st->new_outls_buffer;
          for(auto* p : node->root_inputs())
          {
            if(ossia::contains(inbuf, p))
              continue;
            const auto edges = p->sources;
            for(auto e : edges)
              g->disconnect(e);
          }
          for(auto* p : node->root_outputs())
          {
            if(ossia::contains(outbuf, p))
              continue;
            const auto edges = p->targets;
            for(auto e : edges)
              g->disconnect(e);
          }
        }
 
        node->dynamic_ports = dp;
        node->reload_all_ports(port_st->reload_inlet, port_st->reload_outlet);
 
        auto& inbuf = port_st->new_inls_buffer;
        auto& outbuf = port_st->new_outls_buffer;
        node->root_inputs().assign(inbuf.begin(), inbuf.end());
        node->root_outputs().assign(outbuf.begin(), outbuf.end());
 
        // Size the object's own port vectors right away: they are otherwise
        // only resized on the next tick, and a value from the UI for a new
        // port may arrive before that.
        if constexpr(avnd::dynamic_ports_input_introspection<T>::size > 0)
        {
          for(auto state : node->impl.full_state())
          {
            avnd::dynamic_ports_input_introspection<T>::for_all_n2(
                state.inputs, [&]<std::size_t Idx>(
                                  auto& field, auto, avnd::field_index<Idx> idx) {
              field.ports.resize(std::max(0, node->dynamic_ports.num_in_ports(idx)));
            });
          }
        }
        if constexpr(avnd::dynamic_ports_output_introspection<T>::size > 0)
        {
          for(auto state : node->impl.full_state())
          {
            avnd::dynamic_ports_output_introspection<T>::for_all_n2(
                state.outputs, [&]<std::size_t Idx>(
                                   auto& field, auto, avnd::field_index<Idx> idx) {
              field.ports.resize(std::max(0, node->dynamic_ports.num_out_ports(idx)));
            });
          }
        }
 
        // port_st now holds the previous dynamic ports: free them in the main thread
        if(auto q = gcq.lock())
          q->enqueue(Execution::gc(std::move(port_st)));
      });
 
      // 4. New ports in
      setup.proc_map[this->node.get()] = &element;
      const std::size_t n_in = std::min(new_inlets.size(), inbuf.size());
      for(std::size_t i = 0; i < n_in; i++)
        setup.register_inlet(*new_inlets[i], inbuf[i], this->node, commands);
 
      const std::size_t n_out = std::min(new_outlets.size(), outbuf.size());
      for(std::size_t i = 0; i < n_out; i++)
        setup.register_outlet(*new_outlets[i], outbuf[i], this->node, commands);
 
      // 5. Cables against the new ports; then whoever keeps other edges to
      //    this node (the interval's audio propagation) syncs them
      for(auto c : cables)
        setup.connectCable(*c, commands);
      this->portsReplaced(&commands);
 
      commands.run_all();
 
      this->m_oldInlets = new_inlets;
      this->m_oldOutlets = new_outlets;
 
      // UI -> exec for the control inlets that appeared
      using dynamic_ports_port_type = avnd::dynamic_ports_input_introspection<Node>;
      if constexpr(dynamic_ports_port_type::size > 0)
      {
        avnd::effect_container<Node>& eff = n->impl;
        for(auto state : eff.full_state())
        {
          dynamic_ports_port_type::for_all_n2(
              state.inputs,
              dispatch_control_reconnect<Node>{element, ctx, n, *this, this});
        }
      }
    }
  }
 
  void connect_controls(
      ProcessModel<Node>& element, const ::Execution::Context& ctx,
      std::shared_ptr<safe_node<Node>>& ptr)
  {
    using dynamic_ports_port_type = avnd::dynamic_ports_input_introspection<Node>;
    using control_inputs_type = avnd::control_input_introspection<Node>;
    using curve_inputs_type = avnd::curve_input_introspection<Node>;
    using soundfile_inputs_type = avnd::soundfile_input_introspection<Node>;
    using midifile_inputs_type = avnd::midifile_input_introspection<Node>;
    using raw_file_inputs_type = avnd::raw_file_input_introspection<Node>;
    using control_outputs_type = avnd::control_output_introspection<Node>;
 
    // UI controls to engine
    safe_node<Node>& node = *ptr;
    avnd::effect_container<Node>& eff = node.impl;
 
    // Initialize all the controls in the node with the current value.
    // And update the node when the UI changes
 
    if constexpr(dynamic_ports_port_type::size > 0)
    {
      for(auto state : eff.full_state())
      {
        dynamic_ports_port_type::for_all_n2(
            state.inputs, dispatch_control_setup<Node>{element, ctx, ptr, *this, this});
      }
    }
    if constexpr(control_inputs_type::size > 0)
    {
      for(auto state : eff.full_state())
      {
        control_inputs_type::for_all_n2(
            state.inputs, dispatch_control_setup<Node>{element, ctx, ptr, *this, this});
      }
    }
    if constexpr(curve_inputs_type::size > 0)
    {
      for(auto state : eff.full_state())
      {
        curve_inputs_type::for_all_n2(
            state.inputs, dispatch_control_setup<Node>{element, ctx, ptr, *this, this});
      }
    }
    if constexpr(soundfile_inputs_type::size > 0)
    {
      soundfile_inputs_type::for_all_n2(
          avnd::get_inputs<Node>(eff),
          dispatch_control_setup<Node>{element, ctx, ptr, *this, this});
 
      setup_soundfile_task_pool(element, ctx, ptr);
    }
    if constexpr(midifile_inputs_type::size > 0)
    {
      midifile_inputs_type::for_all_n2(
          avnd::get_inputs<Node>(eff),
          dispatch_control_setup<Node>{element, ctx, ptr, *this, this});
    }
    if constexpr(raw_file_inputs_type::size > 0)
    {
      raw_file_inputs_type::for_all_n2(
          avnd::get_inputs<Node>(eff),
          dispatch_control_setup<Node>{element, ctx, ptr, *this, this});
    }
 
    // Engine to ui controls
    if constexpr(control_inputs_type::size > 0 || control_outputs_type::size > 0)
    {
      std::weak_ptr<safe_node<Node>> weak_node = ptr;
      update_control_value_in_ui<Node> timer_action{weak_node, &element};
      followExecutionUpdate(
          [this, weak_node, timer_action, &ctx,
           timer = std::make_shared<QMetaObject::Connection>()](bool enabled) {
        // The node enqueues its control outputs for the UI every tick; with
        // no reader, that queue grows without bound and allocates in the
        // audio callback. Tell it not to bother.
        if constexpr(control_outputs_type::size > 0)
          if(auto n = weak_node.lock())
            n->control.notify_ui.store(enabled, std::memory_order_relaxed);
 
        QObject::disconnect(*timer);
        if(enabled)
        {
          timer_action();
          *timer = QObject::connect(
              &ctx.doc.coarseUpdateTimer, &QTimer::timeout, this,
              [timer_action] { timer_action(); }, Qt::QueuedConnection);
        }
      });
    }
  }
 
  void setup_soundfile_task_pool(
      ProcessModel<Node>& element, const ::Execution::Context& ctx,
      std::shared_ptr<safe_node<Node>>& ptr)
  {
    safe_node<Node>& node = *ptr;
 
    using soundfile_inputs_type = avnd::soundfile_input_introspection<Node>;
 
    auto& tq = score::TaskPool::instance();
    node.soundfiles.load_request
        = [&tq, p = std::weak_ptr{ptr}, &ctx](std::string& str, int idx) {
      auto eff_ptr = p.lock();
      if(!eff_ptr)
        return;
      tq.post([eff_ptr = std::move(eff_ptr), filename = str, &ctx, idx]() mutable {
        if(auto file = loadSoundfile(filename, ctx.doc, ctx.execState))
        {
          // Sadly we have to go back to the main thread as execution queue is SPSC
          QMetaObject::invokeMethod(
              QCoreApplication::instance(),
              [&ctx, file = std::move(file), eff_ptr = std::move(eff_ptr), idx = idx] {
            ctx.executionQueue.enqueue([sf = std::move(file), p = std::weak_ptr{eff_ptr},
                                        gcq = ctx.weakGCQueue(), idx]() mutable {
              if(auto eff_ptr = p.lock())
              {
                avnd::effect_container<Node>& eff = eff_ptr->impl;
                soundfile_inputs_type::for_nth_mapped_n2(
                    avnd::get_inputs<Node>(eff), idx,
                    [&]<std::size_t NField, std::size_t N>(
                        auto& field, avnd::predicate_index<N> p,
                        avnd::field_index<NField> f) {
                  sf = eff_ptr->soundfile_loaded(sf, p, f);
                });
              }
              // sf is now the file the port held before
              free_on_main_thread(gcq, sf);
            });
          });
        }
      });
    };
  }
 
  void connect_dynamic_items(ProcessModel<Node>& element, Node& eff)
  {
    // type-only inputs: get_inputs would hard-error
    if constexpr(
        avnd::inputs_is_type<Node>
        || avnd::control_input_introspection<Node>::size == 0)
      return;
    else
    avnd::control_input_introspection<Node>::for_all_n2(
        avnd::get_inputs<Node>(eff),
        [&element]<std::size_t Idx, typename F>(
            F& field, auto pred_index, avnd::field_index<Idx>) {
      if constexpr(avnd::dynamic_items_parameter<F>)
      {
        auto ports = element.avnd_input_idx_to_model_ports(Idx);
        if(ports.size() != 1)
          return;
        if(auto combo = qobject_cast<Process::ComboBox*>(ports[0]))
        {
          field.update_items = [p = QPointer<Process::ComboBox>{combo}](
                                   std::vector<std::string> items) {
            std::vector<std::pair<QString, ossia::value>> alts;
            alts.reserve(items.size());
            for(std::size_t i = 0; i < items.size(); i++)
              alts.emplace_back(QString::fromStdString(items[i]), (int)i);
            QMetaObject::invokeMethod(
                qApp,
                [p, alts = std::move(alts)]() mutable {
              if(p)
                p->setAlternatives(std::move(alts));
            },
                Qt::QueuedConnection);
          };
        }
      }
    });
  }
 
  void connect_message_bus(
      ProcessModel<Node>& element, const ::Execution::Context& ctx, Node& eff,
      ui_wakeups& wakeups)
  {
    // Custom UI messages to engine
    if constexpr(avnd::has_gui_to_processor_bus<Node>)
    {
      element.from_ui = ui_to_processor_bus(eff, weak_exec, weak_gc);
    }
 
    // Processor to UI messages never allocate on the processor's thread: they
    // go in an execution command when they fit, else in a ring of the node's
    // own. Both have a fixed capacity: when it is full the message is dropped,
    // and the UI catches up with the next one. The messages a ring holds are
    // not lost when the queue is full: the node wakes the UI on a later tick.
    if constexpr(avnd::has_processor_to_gui_bus<Node>)
    {
      if constexpr(requires { eff.send_message = [](auto&&) { }; })
      {
        using send_type = std::remove_cvref_t<decltype(eff.send_message)>;
        using message_type
            = std::remove_cvref_t<typename function_argument<send_type>::type>;
        if constexpr(
            sizeof(QPointer<QObject>) + sizeof(message_type)
            < Execution::ExecutionCommand::max_storage)
        {
          eff.send_message = [proc = QPointer{&this->process()},
                              qed_ptr = weak_edit]<typename T>(T&& b) mutable {
            auto qed = qed_ptr.lock();
            if(!qed)
              return;
            qed->try_enqueue([proc, bb = message_type(std::forward<T>(b))]() mutable {
              if(proc && proc->to_ui)
                MessageBusSender{proc->to_ui}(std::move(bb));
            });
          };
        }
        else
        {
          static_assert(
              std::is_default_constructible_v<message_type>,
              "a processor to UI message is stored in a preallocated ring");
          auto ring = std::make_shared<message_ring<message_type>>();
          // Sends the wake-up the ring requested; false when the queue is full.
          auto wake = [proc = QPointer{&this->process()}, qed_ptr = weak_edit, ring] {
            auto qed = qed_ptr.lock();
            if(!qed)
            {
              ring->cancel();
              return true;
            }
            if(qed->try_enqueue([proc, ring]() mutable {
              ring->drain([&](message_type&& m) {
                if(proc && proc->to_ui)
                  MessageBusSender{proc->to_ui}(std::move(m));
              });
            }))
              return true;
            ring->park();
            return false;
          };
          wakeups.add([ring, wake] {
            if(!ring->unpark() || !ring->request())
              return true;
            return wake();
          });
 
          eff.send_message = [qed_ptr = weak_edit, ring, wake,
                              wakeups = &wakeups]<typename T>(T&& b) mutable {
            if(qed_ptr.expired())
              return;
            if(!ring->push(message_type(std::forward<T>(b))))
              return;
            if(!ring->request())
              return;
            if(!wake())
              wakeups->park();
          };
        }
      }
      else if constexpr(requires { eff.send_message = []() { }; })
      {
        eff.send_message
            = [proc = QPointer{&this->process()}, qed_ptr = weak_edit]() mutable {
          if(!proc)
            return;
          auto qed = qed_ptr.lock();
          if(!qed)
            return;
 
          qed->try_enqueue([proc]() mutable {
            if(proc && proc->to_ui)
              MessageBusSender{proc->to_ui}();
          });
        };
      }
 
      // A message sent by a worker, from its thread: it may be sent once the
      // node is gone, e.g. by a job the score's stop started. It is posted to
      // the UI thread, where it reaches the views of the process if the
      // process is still there. It allocates: never from the processing
      // thread.
      if constexpr(requires { eff.send_message_from_worker = [](auto&&) { }; })
      {
        using send_type = std::remove_cvref_t<decltype(eff.send_message_from_worker)>;
        using message_type
            = std::remove_cvref_t<typename function_argument<send_type>::type>;
        eff.send_message_from_worker = [proc = QPointer{&this->process()}](message_type m) {
          ossia::qt::run_async(qApp, [proc, m = std::move(m)]() mutable {
            if(proc && proc->to_ui)
              MessageBusSender{proc->to_ui}(std::move(m));
          });
        };
      }
    }
  }
 
  void connect_worker(const std::shared_ptr<safe_node<Node>>& node_ptr)
  {
    if constexpr(avnd::has_worker<Node>)
    {
      avnd::effect_container<Node>& eff = node_ptr->impl;
 
      // Initialize the thread pool beforehand
      score::TaskPool::instance();
 
      // An object with a worker is always given the node which delivers the
      // results at the beginning of its tick (exec_node_t).
      const std::shared_ptr<node_with_worker<Node>> self{
          node_ptr, static_cast<node_with_worker<Node>*>(node_ptr.get())};
 
      for(auto& e : eff.effects())
        bind_worker(
            e, std::weak_ptr{std::shared_ptr<Node>(node_ptr, &e)},
            exec_worker_delivery<Node>{self});
    }
  }
 
  // Update everything
  void update_controls(std::shared_ptr<safe_node<Node>>& ptr)
  {
    avnd::effect_container<Node>& eff = ptr->impl;
    {
      for(auto state : eff.full_state())
      {
        avnd::input_introspection<Node>::for_all(state.inputs, [&](auto& field) {
          avnd::update_if_holding_value(state, field);
        });
      }
    }
  }
 
  void cleanup() override
  {
    if constexpr(requires { this->process().from_ui; })
    {
      this->process().from_ui = [](QByteArray arr) {};
    }
    // FIXME cleanup eff.effect.send_message too ?
    for(auto& c : m_executionUpdate)
      QObject::disconnect(c);
    m_executionUpdate.clear();
    if constexpr(requires { Node::monitor; })
      this->process().setMonitor({});
 
#if SCORE_PLUGIN_GFX
    if constexpr(is_gpu<Node>)
    {
      // FIXME this must move in the Node dtor. See video_node
      auto& gfx_exec = this->system().doc.template plugin<Gfx::DocumentPlugin>().exec;
      if(node_id >= 0)
      {
        gfx_exec.ui->unregister_node(node_id);
        node_id = score::gfx::invalid_node_index;
      }
    }
 
    // FIXME refactor this with other GFX processes
    for(auto* outlet : this->process().outlets())
    {
      if(auto out = qobject_cast<Gfx::TextureOutlet*>(outlet))
      {
        out->nodeId = -1;
      }
    }
#endif
    ::Execution::ProcessComponent::cleanup();
  }
 
  ~Executor() { }
};
}