Loading...
Searching...
No Matches
score-plugin-avnd/Crousti/Executor.hpp
1#pragma once
2
3#include <Process/Execution/ProcessComponent.hpp>
4#include <Process/ExecutionContext.hpp>
5#include <Process/ExecutionSetup.hpp>
6
7#include <Explorer/DocumentPlugin/DeviceDocumentPlugin.hpp>
8
9#include <Crousti/CpuAnalysisNode.hpp>
10#include <Crousti/CpuFilterNode.hpp>
11#include <Crousti/ExecutorPortSetup.hpp>
12#include <Crousti/ExecutorUpdateControlValueInUi.hpp>
13#include <Crousti/File.hpp>
14#include <Crousti/GpuComputeNode.hpp>
15#include <Crousti/GpuNode.hpp>
16#include <Crousti/MessageBus.hpp>
17#include <Crousti/Metadatas.hpp>
18#include <Crousti/ProcessModel.hpp>
19
20#include <score/tools/Bind.hpp>
21#include <ossia/detail/flat_set.hpp>
22#include <ossia/dataflow/exec_state_facade.hpp>
23#include <ossia/dataflow/graph/graph_interface.hpp>
24#include <ossia/dataflow/graph_edge.hpp>
25#include <ossia/dataflow/node_process.hpp>
26#include <ossia/network/context.hpp>
27
28#include <ossia-qt/invoke.hpp>
29
30#include <QGuiApplication>
31
32#include <flicks.h>
33
34#if SCORE_PLUGIN_GFX
35#include <Crousti/GpuNode.hpp>
36#include <Gfx/GfxApplicationPlugin.hpp>
37#endif
38
39#include <Scenario/Settings/ScenarioSettingsModel.hpp>
40
41#include <score/tools/ThreadPool.hpp>
42
43#include <ossia/detail/type_if.hpp>
44
45#include <QTimer>
46
47#include <avnd/binding/ossia/data_node.hpp>
48#include <avnd/binding/ossia/mono_audio_node.hpp>
49#include <avnd/binding/ossia/node.hpp>
50#include <avnd/binding/ossia/ossia_audio_node.hpp>
51#include <avnd/binding/ossia/ossia_dynamic_audio_node.hpp>
52#include <avnd/binding/ossia/poly_audio_node.hpp>
53#include <avnd/concepts/dynamic_items.hpp>
54#include <avnd/concepts/temporality.hpp>
55#include <avnd/concepts/ui.hpp>
56#include <avnd/concepts/worker.hpp>
57#include <Crousti/WorkerBinding.hpp>
58
59#include <QPointer>
60
61namespace oscr
62{
69[[nodiscard]] static ossia::value
70gpuControlValue(
71 const Process::ControlInlet& ctrl, const ossia::value& v,
72 const score::DocumentContext& ctx)
73{
74 if(qobject_cast<const Process::FileChooserBase*>(&ctrl)
75 || qobject_cast<const Process::FolderChooser*>(&ctrl))
76 return resolvePathValue(v, ctx);
77 return v;
78}
79
93template <typename Node>
94struct node_with_worker : safe_node<Node>
95{
96 using safe_node<Node>::safe_node;
97
101
102 void
103 run(const ossia::token_request& tk, ossia::exec_state_facade st) noexcept override
104 {
105 Execution::ExecutionCommand cmd;
106 if(worker_results.try_dequeue(cmd))
107 {
108 const auto [start, frames] = st.timings(tk);
109 this->start_frame_for_this_tick = start;
110 this->frame_count_for_this_tick = frames;
111 do
112 {
113 cmd();
114 } while(worker_results.try_dequeue(cmd));
115 }
116
117 safe_node<Node>::run(tk, st);
118 }
119};
120
125template <typename Node>
127{
128 std::weak_ptr<node_with_worker<Node>> node_wp;
129 exec_worker_delivery begin() const noexcept { return *this; }
130
131 template <typename F>
132 void operator()(F&& apply)
133 {
134 ossia::qt::run_async(
135 qApp, [node_wp = std::move(node_wp), apply = std::move(apply)]() mutable {
136 if(std::shared_ptr n = node_wp.lock())
137 n->worker_results.enqueue(std::move(apply));
138 });
139 }
140};
141
144template <typename Node>
146 = std::conditional_t<avnd::has_worker<Node>, node_with_worker<Node>, safe_node<Node>>;
147
148template <typename Node>
149class CustomNodeProcess : public ossia::node_process
150{
151 using node_process::node_process;
152 void start() override
153 {
154 node_process::start();
155 auto& n = static_cast<safe_node<Node>&>(*node);
156 if_possible(n.impl.effect.start());
157 }
158 void pause() override
159 {
160 auto& n = static_cast<safe_node<Node>&>(*node);
161 if_possible(n.impl.effect.pause());
162 node_process::pause();
163 }
164
165 void resume() override
166 {
167 node_process::resume();
168 auto& n = static_cast<safe_node<Node>&>(*node);
169 if_possible(n.impl.effect.resume());
170 }
171
172 void stop() override
173 {
174 auto& n = static_cast<safe_node<Node>&>(*node);
175 if_possible(n.impl.effect.stop());
176 node_process::stop();
177 }
178
179 void offset_impl(ossia::time_value date) override
180 {
181 node_process::offset_impl(date);
182 auto& n = static_cast<safe_node<Node>&>(*node);
183 util::flicks f{date.impl};
184 if_possible(n.impl.effect.transport(f));
185 }
186
187 void transport_impl(ossia::time_value date) override
188 {
189 node_process::transport_impl(date);
190 auto& n = static_cast<safe_node<Node>&>(*node);
191
192 util::flicks f{date.impl};
193 if_possible(n.impl.effect.transport(f));
194 }
195};
196
197template <typename Node>
198class Executor final
199 : public Execution::ProcessComponent_T<ProcessModel<Node>, ossia::node_process>
200 , public dynamic_ports_component_data<Node>
201{
202public:
203 static Q_DECL_RELAXED_CONSTEXPR UuidKey<score::Component> static_key() noexcept
204 {
205 return uuid_from_string<Node>();
206 }
207
208 UuidKey<score::Component> key() const noexcept final override { return static_key(); }
209
210 bool key_match(UuidKey<score::Component> other) const noexcept final override
211 {
212 return static_key() == other || Execution::ProcessComponent::base_key_match(other);
213 }
214
215#if defined(SCORE_PLUGIN_GFX)
216 [[no_unique_address]] ossia::type_if<int, is_gpu<Node>> node_id = score::gfx::invalid_node_index;
217#endif
218
219 Executor(ProcessModel<Node>& element, const ::Execution::Context& ctx, QObject* p)
221 element, ctx, "Executor::ProcessModel<Info>", p}
222 {
223 if constexpr(is_gpu<Node>)
224 {
225#if SCORE_PLUGIN_GFX
226 setup_gpu(element, ctx, p);
227#endif
228 }
229 else
230 {
231 setup_cpu(element, ctx, p);
232 }
233
234 if constexpr(avnd::tag_process_exec<Node>)
235 {
236 this->m_ossia_process = std::make_shared<CustomNodeProcess<Node>>(this->node);
237 }
238 else
239 {
240 this->m_ossia_process = std::make_shared<ossia::node_process>(this->node);
241 }
242 }
243
244 void
245 setup_cpu(ProcessModel<Node>& element, const ::Execution::Context& ctx, QObject* p)
246 {
247 auto& net_ctx
248 = *ctx.doc.findPlugin<Explorer::DeviceDocumentPlugin>()->networkContext();
249 const auto id
250 = std::hash<ObjectPath>{}(Path<Process::ProcessModel>{element}.unsafePath());
251
252 auto st = ossia::exec_state_facade{ctx.execState.get()};
253 std::shared_ptr<safe_node<Node>> ptr;
254 auto node = new exec_node_t<Node>{st.bufferSize(), (double)st.sampleRate(), id};
255 node->root_inputs().reserve(element.inlets().size());
256 node->root_outputs().reserve(element.outlets().size());
257
258 node->prepare(*ctx.execState.get()); // Preparation of the ossia side
259
260 if_possible(node->impl.effect.ossia_state = st);
261 if_possible(node->impl.effect.io_context = &net_ctx.context);
262 if_possible(node->impl.effect.ossia_document_context = &ctx.doc);
263 ptr.reset(node);
264 this->node = ptr;
265
266 if constexpr(requires { ptr->impl.effect; })
267 if constexpr(std::is_same_v<std::decay_t<decltype(ptr->impl.effect)>, Node>)
268 {
269 connect_message_bus(element, ctx, ptr->impl.effect);
270 connect_dynamic_items(element, ptr->impl.effect);
271 }
272 connect_worker(ptr);
273
274 node->dynamic_ports = element.dynamic_ports;
275 node->finish_init();
276
277 connect_controls(element, ctx, ptr);
278 update_controls(ptr);
279
280 // To call prepare() after evertyhing is ready
281 node->audio_configuration_changed(st);
282
283 if constexpr(oscr::has_dynamic_ports<Node>)
284 {
285 QObject::connect(
286 &element, &Process::ProcessModel::inletsChanged, this,
288 QObject::connect(
289 &element, &Process::ProcessModel::outletsChanged, this,
291
292 this->m_oldInlets = element.inlets();
293 this->m_oldOutlets = element.outlets();
294 }
295 }
296
297 void
298 setup_gpu(ProcessModel<Node>& element, const ::Execution::Context& ctx, QObject* p)
299 {
300#if SCORE_PLUGIN_GFX
301 // FIXME net context for gpu node ?
302 const int64_t id
303 = std::hash<ObjectPath>{}(Path<Process::ProcessModel>{element}.unsafePath());
304
305 auto& gfx_exec = ctx.doc.plugin<Gfx::DocumentPlugin>().exec;
306
307 // Create the executor in the audio thread
308 struct named_exec_node final : Gfx::gfx_exec_node
309 {
310 using Gfx::gfx_exec_node::gfx_exec_node;
311 std::string label() const noexcept override
312 {
313 return std::string(avnd::get_name<Node>());
314 }
315 };
316
317 auto node = std::make_shared<named_exec_node>(gfx_exec);
318 node->prepare(*ctx.execState);
319
320 this->node = node;
321
322 // Create the controls, inputs outputs etc.
323 std::size_t i = 0;
324
325 for(auto& ctl : element.inlets())
326 {
327 if(auto ctrl = qobject_cast<Process::ControlInlet*>(ctl))
328 {
329 auto& p = node->add_control();
330 p->value = gpuControlValue(*ctrl, ctrl->value(), ctx.doc);
331 // An impulse is only ever sent when it is triggered: the object's
332 // one-shot inputs start empty, like setup_control_for_exec leaves them
333 // on the CPU path.
334 p->changed = p->value.get_type() != ossia::val_type::IMPULSE;
335
336 // What the control holds and the range it takes, so that a cable into
337 // it converts and rescales. The ISF executor sets its controls up the
338 // same way.
339 ctrl->setupExecution(*node->root_inputs().back(), this);
340
341 QObject::connect(
342 ctrl, &Process::ControlInlet::valueChanged, this,
343 [con = Gfx::con_unvalidated{ctx, i, 0, node}, ctrl,
344 &doc = ctx.doc](const ossia::value& v) mutable {
345 con(gpuControlValue(*ctrl, v, doc));
346 });
347 i++;
348 }
349 else if(auto ctrl = qobject_cast<Process::ValueInlet*>(ctl))
350 {
351 auto& p = node->add_control();
352 p->changed = false;
353 i++;
354 }
355 else if(auto ctrl = qobject_cast<Process::AudioInlet*>(ctl))
356 {
357 node->add_audio();
358 }
359 else if(auto ctrl = qobject_cast<Gfx::TextureInlet*>(ctl))
360 {
361 ossia::texture_inlet& inl = *node->add_texture();
362 ctrl->setupExecution(inl, this);
363 }
364 else if(auto ctrl = qobject_cast<Gfx::GeometryInlet*>(ctl))
365 {
366 ossia::geometry_inlet& inl = *node->add_geometry();
367 ctrl->setupExecution(inl, this);
368 }
369 }
370
371 // FIXME refactor this with other GFX processes
372 for(auto* outlet : element.outlets())
373 {
374 if(auto ctrl = qobject_cast<Process::ControlOutlet*>(outlet))
375 {
376 node->add_control_out();
377 }
378 else if(auto ctrl = qobject_cast<Process::ValueOutlet*>(outlet))
379 {
380 node->add_control_out();
381 }
382 else if(auto out = qobject_cast<Gfx::TextureOutlet*>(outlet))
383 {
384 node->add_texture_out();
385 out->nodeId = node_id;
386 }
387 else if(auto out = qobject_cast<Gfx::GeometryOutlet*>(outlet))
388 {
389 node->add_geometry_out();
390 }
391 }
392
393 // Create the GPU node
394 std::weak_ptr qex_ptr = std::shared_ptr<Execution::ExecutionCommandQueue>(
395 ctx.alias.lock(), &ctx.executionQueue);
396 std::unique_ptr<score::gfx::Node> ptr;
397 if constexpr(GpuGraphicsNode2<Node>)
398 {
399 auto gpu_node = new CustomGpuNode<Node>(qex_ptr, node->control_outs, id, ctx.doc);
400 ptr.reset(gpu_node);
401 }
402 else if constexpr(GpuComputeNode2<Node>)
403 {
404 auto gpu_node = new GpuComputeNode<Node>(qex_ptr, node->control_outs, id, ctx.doc);
405 ptr.reset(gpu_node);
406 }
407 else if constexpr(GpuNode<Node>)
408 {
409 auto gpu_node
410 = new GfxNode<Node>(element, qex_ptr, node->control_outs, id, ctx.doc);
411 ptr.reset(gpu_node);
412 }
413
414 i = 0;
415 for(auto& ctl : element.inlets())
416 {
417 if(auto ctrl = qobject_cast<Gfx::TextureInlet*>(ctl))
418 {
419 ossia::texture_inlet& inl
420 = static_cast<ossia::texture_inlet&>(*node->root_inputs()[i]);
421 ptr->process(i, inl.data); // Setup render_target_spec
422 }
423 i++;
424 }
425 node->id = gfx_exec.ui->register_node(std::move(ptr));
426 node_id = node->id;
427#endif
428 }
429
443 {
444 if constexpr(oscr::has_dynamic_ports<Node> && !is_gpu<Node>)
445 {
446 using T = Node;
447 auto n = std::dynamic_pointer_cast<safe_node<Node>>(this->node);
448 if(!n)
449 return;
450
451 auto& element = this->process();
452 const Execution::Context& ctx = this->system();
453 Execution::SetupContext& setup = ctx.setup;
454
455 // The ports the exec node is currently set up for. Those that are not
456 // in the model anymore are about to be deleted (they are still alive
457 // here: ProcessModel deletes them after inletsChanged()).
458 const Process::Inlets old_inlets = std::move(this->m_oldInlets);
459 const Process::Outlets old_outlets = std::move(this->m_oldOutlets);
460 const Process::Inlets& new_inlets = element.inlets();
461 const Process::Outlets& new_outlets = element.outlets();
462
463 for(auto port : old_inlets)
464 if(!ossia::contains(new_inlets, port))
465 this->m_connectedControls.erase(port);
466
467 // The cables at either end of this process that are wired up in the
468 // graph: their edges point to the exec ports that are being replaced.
469 // A cable that is on the ports but not wired yet (restored by a command
470 // that announces it after inletsChanged()) is left to that announcement.
471 ossia::small_vector<Process::Cable*, 8> cables;
472 auto collect_cables = [&](const auto& ports) {
473 for(auto port : ports)
474 for(auto& cbl : port->cables())
475 if(auto c = cbl.try_find(ctx.doc))
476 if(setup.m_cables.find(c->id()) != setup.m_cables.end())
477 if(!ossia::contains(cables, c))
478 cables.push_back(c);
479 };
480 collect_cables(old_inlets);
481 collect_cables(old_outlets);
482 collect_cables(new_inlets);
483 collect_cables(new_outlets);
484
485 // Build the new set of exec ports.
486 struct port_storage
487 {
488 ossia::inlets new_inls_buffer;
489 ossia::outlets new_outls_buffer;
490 inlet_reload_storage<T> reload_inlet;
491 outlet_reload_storage<T> reload_outlet;
492 port_storage(std::size_t ins, std::size_t outs)
493 {
494 new_inls_buffer.reserve(ins);
495 new_outls_buffer.reserve(outs);
496 }
497 };
498 auto port_st = std::make_shared<port_storage>(new_inlets.size(), new_outlets.size());
499 auto& inbuf = port_st->new_inls_buffer;
500 auto& outbuf = port_st->new_outls_buffer;
501
502 // Same order as safe_node_base::reinit() + initialize_all_ports():
503 // first the ports that do not come from a field of the object
504 // (audio / value argument of the processor, message inlets), then the fields.
505 n->audio_ports.init(inbuf, outbuf);
506 n->arg_value_ports.init(inbuf, outbuf);
507 n->message_ports.init(inbuf);
508
509 auto& dp = element.dynamic_ports;
510 if constexpr(avnd::inputs_type<T>::size > 0)
511 {
512 using in_info = avnd::input_introspection<T>;
513 using in_type = typename avnd::inputs_type<T>::type;
514 auto& port_tuple = n->ossia_inlets.ports;
515
516 [&]<typename K, K... Index>(std::integer_sequence<K, Index...>) {
517 reload_inlets<safe_node_base_base<T>> init{*n, inbuf, dp};
518 (init(
519 avnd::field_reflection<Index, avnd::pfr::tuple_element_t<Index, in_type>>{},
520 tuplet::get<Index>(port_tuple),
521 tuplet::get<Index>(port_st->reload_inlet.ports)),
522 ...);
523 }(typename in_info::indices_n{});
524 }
525
526 if constexpr(avnd::outputs_type<T>::size > 0)
527 {
528 using out_info = avnd::output_introspection<T>;
529 using out_type = typename avnd::outputs_type<T>::type;
530 auto& port_tuple = n->ossia_outlets.ports;
531
532 [&]<typename K, K... Index>(std::integer_sequence<K, Index...>) {
533 reload_outlets<safe_node_base_base<T>> init{*n, outbuf, dp};
534 (init(
535 avnd::field_reflection<
536 Index, avnd::pfr::tuple_element_t<Index, out_type>>{},
537 tuplet::get<Index>(port_tuple),
538 tuplet::get<Index>(port_st->reload_outlet.ports)),
539 ...);
540 }(typename out_info::indices_n{});
541 }
542
543 // The exec node must line up with the model port for port, otherwise
544 // cables and addresses would land on the wrong exec port.
545 SCORE_SOFT_ASSERT(inbuf.size() == new_inlets.size());
546 SCORE_SOFT_ASSERT(outbuf.size() == new_outlets.size());
547
548 Execution::Transaction commands{ctx};
549
550 // 1. The edges of the cables, before their ports go away
551 for(auto c : cables)
552 setup.removeCable(*c, commands);
553
554 // 2. Old ports out of the execution state and of the setup maps
555 setup.unregister_node_soft(old_inlets, old_outlets, this->node, commands);
556
557 // 3. Swap the ports in
558 commands.push_back(
559 [node = n, dp, port_st, gcq = ctx.weakGCQueue(),
560 wg = std::weak_ptr{ctx.execGraph}]() mutable {
561 OSSIA_ENSURE_CURRENT_THREAD_KIND(ossia::thread_type::Audio);
562
563 // Nothing may keep an edge to a port that is about to be freed: the
564 // cables were disconnected above, whatever else is still there (the
565 // interval's audio propagation, or an edge nobody tracks) is dropped.
566 // Ports that stay keep their edges.
567 if(auto g = wg.lock())
568 {
569 const auto& inbuf = port_st->new_inls_buffer;
570 const auto& outbuf = port_st->new_outls_buffer;
571 for(auto* p : node->root_inputs())
572 {
573 if(ossia::contains(inbuf, p))
574 continue;
575 const auto edges = p->sources;
576 for(auto e : edges)
577 g->disconnect(e);
578 }
579 for(auto* p : node->root_outputs())
580 {
581 if(ossia::contains(outbuf, p))
582 continue;
583 const auto edges = p->targets;
584 for(auto e : edges)
585 g->disconnect(e);
586 }
587 }
588
589 node->dynamic_ports = dp;
590 node->reload_all_ports(port_st->reload_inlet, port_st->reload_outlet);
591
592 auto& inbuf = port_st->new_inls_buffer;
593 auto& outbuf = port_st->new_outls_buffer;
594 node->root_inputs().assign(inbuf.begin(), inbuf.end());
595 node->root_outputs().assign(outbuf.begin(), outbuf.end());
596
597 // Size the object's own port vectors right away: they are otherwise
598 // only resized on the next tick, and a value from the UI for a new
599 // port may arrive before that.
600 if constexpr(avnd::dynamic_ports_input_introspection<T>::size > 0)
601 {
602 for(auto state : node->impl.full_state())
603 {
604 avnd::dynamic_ports_input_introspection<T>::for_all_n2(
605 state.inputs, [&]<std::size_t Idx>(
606 auto& field, auto, avnd::field_index<Idx> idx) {
607 field.ports.resize(std::max(0, node->dynamic_ports.num_in_ports(idx)));
608 });
609 }
610 }
611 if constexpr(avnd::dynamic_ports_output_introspection<T>::size > 0)
612 {
613 for(auto state : node->impl.full_state())
614 {
615 avnd::dynamic_ports_output_introspection<T>::for_all_n2(
616 state.outputs, [&]<std::size_t Idx>(
617 auto& field, auto, avnd::field_index<Idx> idx) {
618 field.ports.resize(std::max(0, node->dynamic_ports.num_out_ports(idx)));
619 });
620 }
621 }
622
623 // port_st now holds the previous dynamic ports: free them in the main thread
624 if(auto q = gcq.lock())
625 q->enqueue(Execution::gc(std::move(port_st)));
626 });
627
628 // 4. New ports in
629 setup.proc_map[this->node.get()] = &element;
630 const std::size_t n_in = std::min(new_inlets.size(), inbuf.size());
631 for(std::size_t i = 0; i < n_in; i++)
632 setup.register_inlet(*new_inlets[i], inbuf[i], this->node, commands);
633
634 const std::size_t n_out = std::min(new_outlets.size(), outbuf.size());
635 for(std::size_t i = 0; i < n_out; i++)
636 setup.register_outlet(*new_outlets[i], outbuf[i], this->node, commands);
637
638 // 5. Cables against the new ports; then whoever keeps other edges to
639 // this node (the interval's audio propagation) syncs them
640 for(auto c : cables)
641 setup.connectCable(*c, commands);
642 this->portsReplaced(&commands);
643
644 commands.run_all();
645
646 this->m_oldInlets = new_inlets;
647 this->m_oldOutlets = new_outlets;
648
649 // UI -> exec for the control inlets that appeared
650 using dynamic_ports_port_type = avnd::dynamic_ports_input_introspection<Node>;
651 if constexpr(dynamic_ports_port_type::size > 0)
652 {
653 avnd::effect_container<Node>& eff = n->impl;
654 for(auto state : eff.full_state())
655 {
656 dynamic_ports_port_type::for_all_n2(
657 state.inputs,
658 dispatch_control_reconnect<Node>{element, ctx, n, *this, this});
659 }
660 }
661 }
662 }
663
664 void connect_controls(
665 ProcessModel<Node>& element, const ::Execution::Context& ctx,
666 std::shared_ptr<safe_node<Node>>& ptr)
667 {
668 using dynamic_ports_port_type = avnd::dynamic_ports_input_introspection<Node>;
669 using control_inputs_type = avnd::control_input_introspection<Node>;
670 using curve_inputs_type = avnd::curve_input_introspection<Node>;
671 using soundfile_inputs_type = avnd::soundfile_input_introspection<Node>;
672 using midifile_inputs_type = avnd::midifile_input_introspection<Node>;
673 using raw_file_inputs_type = avnd::raw_file_input_introspection<Node>;
674 using control_outputs_type = avnd::control_output_introspection<Node>;
675
676 // UI controls to engine
677 safe_node<Node>& node = *ptr;
678 avnd::effect_container<Node>& eff = node.impl;
679
680 // Initialize all the controls in the node with the current value.
681 // And update the node when the UI changes
682
683 if constexpr(dynamic_ports_port_type::size > 0)
684 {
685 for(auto state : eff.full_state())
686 {
687 dynamic_ports_port_type::for_all_n2(
688 state.inputs, dispatch_control_setup<Node>{element, ctx, ptr, *this, this});
689 }
690 }
691 if constexpr(control_inputs_type::size > 0)
692 {
693 for(auto state : eff.full_state())
694 {
695 control_inputs_type::for_all_n2(
696 state.inputs, dispatch_control_setup<Node>{element, ctx, ptr, *this, this});
697 }
698 }
699 if constexpr(curve_inputs_type::size > 0)
700 {
701 for(auto state : eff.full_state())
702 {
703 curve_inputs_type::for_all_n2(
704 state.inputs, dispatch_control_setup<Node>{element, ctx, ptr, *this, this});
705 }
706 }
707 if constexpr(soundfile_inputs_type::size > 0)
708 {
709 soundfile_inputs_type::for_all_n2(
710 avnd::get_inputs<Node>(eff),
711 dispatch_control_setup<Node>{element, ctx, ptr, *this, this});
712
713 setup_soundfile_task_pool(element, ctx, ptr);
714 }
715 if constexpr(midifile_inputs_type::size > 0)
716 {
717 midifile_inputs_type::for_all_n2(
718 avnd::get_inputs<Node>(eff),
719 dispatch_control_setup<Node>{element, ctx, ptr, *this, this});
720 }
721 if constexpr(raw_file_inputs_type::size > 0)
722 {
723 raw_file_inputs_type::for_all_n2(
724 avnd::get_inputs<Node>(eff),
725 dispatch_control_setup<Node>{element, ctx, ptr, *this, this});
726 }
727
728 // Engine to ui controls
729 if constexpr(control_inputs_type::size > 0 || control_outputs_type::size > 0)
730 {
731 auto& settings = score::AppContext().settings<Scenario::Settings::Model>();
732
733 // The node enqueues its control outputs for the UI every tick; with no
734 // reader installed below, that queue grows without bound and allocates in
735 // the audio callback. Tell it not to bother.
736 if constexpr(control_outputs_type::size > 0)
737 ptr->control.notify_ui.store(
738 settings.getExecutionUpdate(), std::memory_order_relaxed);
739
740 if(settings.getExecutionUpdate())
741 {
742 // Update the value in the UI
743 std::weak_ptr<safe_node<Node>> weak_node = ptr;
744 update_control_value_in_ui<Node> timer_action{weak_node, &element};
745 timer_action();
746
747 con(ctx.doc.coarseUpdateTimer, &QTimer::timeout, this,
748 [timer_action = std::move(timer_action)] { timer_action(); },
749 Qt::QueuedConnection);
750 }
751 }
752 }
753
754 void setup_soundfile_task_pool(
755 ProcessModel<Node>& element, const ::Execution::Context& ctx,
756 std::shared_ptr<safe_node<Node>>& ptr)
757 {
758 safe_node<Node>& node = *ptr;
759
760 using soundfile_inputs_type = avnd::soundfile_input_introspection<Node>;
761
762 auto& tq = score::TaskPool::instance();
763 node.soundfiles.load_request
764 = [&tq, p = std::weak_ptr{ptr}, &ctx](std::string& str, int idx) {
765 auto eff_ptr = p.lock();
766 if(!eff_ptr)
767 return;
768 tq.post([eff_ptr = std::move(eff_ptr), filename = str, &ctx, idx]() mutable {
769 if(auto file = loadSoundfile(filename, ctx.doc, ctx.execState))
770 {
771 // Sadly we have to go back to the main thread as execution queue is SPSC
772 QMetaObject::invokeMethod(
773 QCoreApplication::instance(),
774 [&ctx, file = std::move(file), eff_ptr = std::move(eff_ptr), idx = idx] {
775 ctx.executionQueue.enqueue(
776 [sf = std::move(file), p = std::weak_ptr{eff_ptr}, idx]() mutable {
777 auto eff_ptr = p.lock();
778 if(!eff_ptr)
779 return;
780
781 avnd::effect_container<Node>& eff = eff_ptr->impl;
782 soundfile_inputs_type::for_nth_mapped_n2(
783 avnd::get_inputs<Node>(eff), idx,
784 [&]<std::size_t NField, std::size_t N>(
785 auto& field, avnd::predicate_index<N> p,
786 avnd::field_index<NField> f) {
787 sf = eff_ptr->soundfile_loaded(sf, p, f);
788 });
789 });
790 });
791 }
792 });
793 };
794 }
795
796 void connect_dynamic_items(ProcessModel<Node>& element, Node& eff)
797 {
798 // type-only inputs: get_inputs would hard-error
799 if constexpr(
800 avnd::inputs_is_type<Node>
801 || avnd::control_input_introspection<Node>::size == 0)
802 return;
803 else
804 avnd::control_input_introspection<Node>::for_all_n2(
805 avnd::get_inputs<Node>(eff),
806 [&element]<std::size_t Idx, typename F>(
807 F& field, auto pred_index, avnd::field_index<Idx>) {
808 if constexpr(avnd::dynamic_items_parameter<F>)
809 {
810 auto ports = element.avnd_input_idx_to_model_ports(Idx);
811 if(ports.size() != 1)
812 return;
813 if(auto combo = qobject_cast<Process::ComboBox*>(ports[0]))
814 {
815 field.update_items = [p = QPointer<Process::ComboBox>{combo}](
816 std::vector<std::string> items) {
817 std::vector<std::pair<QString, ossia::value>> alts;
818 alts.reserve(items.size());
819 for(std::size_t i = 0; i < items.size(); i++)
820 alts.emplace_back(QString::fromStdString(items[i]), (int)i);
821 QMetaObject::invokeMethod(
822 qApp,
823 [p, alts = std::move(alts)]() mutable {
824 if(p)
825 p->setAlternatives(std::move(alts));
826 },
827 Qt::QueuedConnection);
828 };
829 }
830 }
831 });
832 }
833
834 void connect_message_bus(
835 ProcessModel<Node>& element, const ::Execution::Context& ctx, Node& eff)
836 {
837 // Custom UI messages to engine
838 if constexpr(avnd::has_gui_to_processor_bus<Node>)
839 {
840 element.from_ui = [qex_ptr = weak_exec, &eff](QByteArray b) {
841 auto qex = qex_ptr.lock();
842 if(!qex)
843 return;
844
845 qex->enqueue([mess = std::move(b), &eff]() mutable {
846 using refl = avnd::function_reflection<&Node::process_message>;
847 static_assert(refl::count <= 1);
848
849 if constexpr(refl::count == 0)
850 {
851 // no arguments, just call it
852 eff.process_message();
853 }
854 else if constexpr(refl::count == 1)
855 {
856 using arg_type = avnd::first_argument<&Node::process_message>;
857 std::decay_t<arg_type> arg;
858 MessageBusReader reader{mess};
859 reader(arg);
860 eff.process_message(std::move(arg));
861 }
862 });
863 };
864 }
865
866 if constexpr(avnd::has_processor_to_gui_bus<Node>)
867 {
868 if constexpr(requires { eff.send_message = [](auto&&) { }; })
869 {
870 eff.send_message = [proc = QPointer{&this->process()},
871 qed_ptr = weak_edit]<typename T>(T&& b) mutable {
872 auto qed = qed_ptr.lock();
873 if(!qed)
874 return;
875 if constexpr(
876 sizeof(QPointer<QObject>) + sizeof(b)
877 < Execution::ExecutionCommand::max_storage)
878 {
879 qed->enqueue([proc, bb = std::move(b)]() mutable {
880 if(proc && proc->to_ui)
881 MessageBusSender{proc->to_ui}(std::move(bb));
882 });
883 }
884 else
885 {
886 qed->enqueue(
887 [proc, bb = std::make_unique<std::decay_t<T>>(std::move(b))]() mutable {
888 if(proc && proc->to_ui)
889 MessageBusSender{proc->to_ui}(*std::move(bb));
890 });
891 }
892 };
893 }
894 else if constexpr(requires { eff.send_message = []() { }; })
895 {
896 eff.send_message
897 = [proc = QPointer{&this->process()}, qed_ptr = weak_edit]() mutable {
898 if(!proc)
899 return;
900 auto qed = qed_ptr.lock();
901 if(!qed)
902 return;
903
904 qed->enqueue([proc]() mutable {
905 if(proc && proc->to_ui)
906 MessageBusSender{proc->to_ui}();
907 });
908 };
909 }
910 }
911 }
912
913 void connect_worker(const std::shared_ptr<safe_node<Node>>& node_ptr)
914 {
915 if constexpr(avnd::has_worker<Node>)
916 {
917 avnd::effect_container<Node>& eff = node_ptr->impl;
918
919 // Initialize the thread pool beforehand
920 score::TaskPool::instance();
921
922 // An object with a worker is always given the node which delivers the
923 // results at the beginning of its tick (exec_node_t).
924 const std::shared_ptr<node_with_worker<Node>> self{
925 node_ptr, static_cast<node_with_worker<Node>*>(node_ptr.get())};
926
927 for(auto& e : eff.effects())
929 e, std::weak_ptr{std::shared_ptr<Node>(node_ptr, &e)},
930 exec_worker_delivery<Node>{self});
931 }
932 }
933
934 // Update everything
935 void update_controls(std::shared_ptr<safe_node<Node>>& ptr)
936 {
937 avnd::effect_container<Node>& eff = ptr->impl;
938 {
939 for(auto state : eff.full_state())
940 {
941 avnd::input_introspection<Node>::for_all(state.inputs, [&](auto& field) {
942 avnd::update_if_holding_value(state, field);
943 });
944 }
945 }
946 }
947
948 void cleanup() override
949 {
950 if constexpr(requires { this->process().from_ui; })
951 {
952 this->process().from_ui = [](QByteArray arr) {};
953 }
954 // FIXME cleanup eff.effect.send_message too ?
955
956#if SCORE_PLUGIN_GFX
957 if constexpr(is_gpu<Node>)
958 {
959 // FIXME this must move in the Node dtor. See video_node
960 auto& gfx_exec = this->system().doc.template plugin<Gfx::DocumentPlugin>().exec;
961 if(node_id >= 0)
962 {
963 gfx_exec.ui->unregister_node(node_id);
964 node_id = score::gfx::invalid_node_index;
965 }
966 }
967
968 // FIXME refactor this with other GFX processes
969 for(auto* outlet : this->process().outlets())
970 {
971 if(auto out = qobject_cast<Gfx::TextureOutlet*>(outlet))
972 {
973 out->nodeId = -1;
974 }
975 }
976#endif
977 ::Execution::ProcessComponent::cleanup();
978 }
979
980 ~Executor() { }
981};
982}
Definition GfxApplicationPlugin.hpp:13
Definition GfxExecNode.hpp:40
The Path class is a typesafe wrapper around ObjectPath.
Definition Path.hpp:52
Definition Port.hpp:242
Definition ScenarioSettingsModel.hpp:21
Definition UuidKey.hpp:345
Definition score-plugin-avnd/Crousti/Executor.hpp:150
Definition score-plugin-avnd/Crousti/Executor.hpp:201
void recompute_ports()
Definition score-plugin-avnd/Crousti/Executor.hpp:442
Definition score-plugin-avnd/Crousti/ProcessModel.hpp:88
Definition Metadatas.hpp:44
Definition Metadatas.hpp:36
Definition Metadatas.hpp:16
Definition Metadatas.hpp:51
TreeNode< DeviceExplorerNode > Node
Definition DeviceNode.hpp:74
Definition Controls.hpp:27
std::conditional_t< avnd::has_worker< Node >, node_with_worker< Node >, safe_node< Node > > exec_node_t
Definition score-plugin-avnd/Crousti/Executor.hpp:146
void bind_worker(Object &obj, std::weak_ptr< Object > obj_wp, Delivery deliver)
Definition WorkerBinding.hpp:31
STL namespace.
Definition ExecutionContext.hpp:201
Definition ExecutionContext.hpp:73
Definition Process/Execution/ProcessComponent.hpp:101
Definition ExecutionSetup.hpp:38
Definition ExecutionTransaction.hpp:21
Definition GfxExecNode.hpp:134
Definition PortForward.hpp:23
Definition PortForward.hpp:27
Definition ExecutorPortSetup.hpp:563
Definition ExecutorPortSetup.hpp:515
Definition ExecutorPortSetup.hpp:16
Definition score-plugin-avnd/Crousti/Executor.hpp:127
Definition score-plugin-avnd/Crousti/Executor.hpp:95
Execution::ExecutionCommandQueue worker_results
Definition score-plugin-avnd/Crousti/Executor.hpp:100
T & settings() const
Access a specific Settings model instance.
Definition ApplicationContext.hpp:41
Definition DocumentContext.hpp:18