diff options
Diffstat (limited to 'src/libs')
28 files changed, 533 insertions, 174 deletions
diff --git a/src/libs/engine/AudioDriver.hpp b/src/libs/engine/AudioDriver.hpp index bc7d1ee9..51373532 100644 --- a/src/libs/engine/AudioDriver.hpp +++ b/src/libs/engine/AudioDriver.hpp @@ -49,6 +49,8 @@ public: virtual SampleCount buffer_size() const = 0; virtual SampleCount sample_rate() const = 0; virtual SampleCount frame_time() const = 0; + + virtual bool is_realtime() const = 0; }; diff --git a/src/libs/engine/Connection.cpp b/src/libs/engine/Connection.cpp index 3a402948..637da891 100644 --- a/src/libs/engine/Connection.cpp +++ b/src/libs/engine/Connection.cpp @@ -50,7 +50,7 @@ Connection::Connection(Port* src_port, Port* dst_port) if (_must_mix) _local_buffer = BufferFactory::create(dst_port->type(), dst_port->buffer(0)->size()); - cerr << src_port->path() << " -> " << dst_port->path() << " must mix: " << _must_mix << endl; + //cerr << src_port->path() << " -> " << dst_port->path() << " must mix: " << _must_mix << endl; } diff --git a/src/libs/engine/Engine.cpp b/src/libs/engine/Engine.cpp index af4eca27..f904ce2a 100644 --- a/src/libs/engine/Engine.cpp +++ b/src/libs/engine/Engine.cpp @@ -38,6 +38,7 @@ #include "EnablePatchEvent.hpp" #include "OSCEngineReceiver.hpp" #include "PostProcessor.hpp" +#include "ProcessSlave.hpp" #ifdef HAVE_JACK_MIDI #include "JackMidiDriver.hpp" #endif @@ -50,21 +51,23 @@ namespace Ingen { Engine::Engine(Ingen::Shared::World* world) -: _world(world), - _midi_driver(NULL), - _osc_driver(NULL), - _maid(new Raul::Maid(maid_queue_size)), - _post_processor(new PostProcessor(/**_maid, */post_processor_queue_size)), - _broadcaster(new ClientBroadcaster()), - _object_store(new ObjectStore()), - _node_factory(new NodeFactory(world)), -/*#ifdef HAVE_LASH - _lash_driver(new LashDriver()), -#else */ -// _lash_driver(NULL), -//#endif - _quit_flag(false), - _activated(false) + : _world(world) + , _midi_driver(NULL) + , _osc_driver(NULL) + , _maid(new Raul::Maid(maid_queue_size)) + , _post_processor(new PostProcessor(/**_maid, */post_processor_queue_size)) + , _broadcaster(new ClientBroadcaster()) + , _object_store(new ObjectStore()) + , _node_factory(new NodeFactory(world)) +#if 0 +#ifdef HAVE_LASH + , _lash_driver(new LashDriver()) +#else + , _lash_driver(NULL) +#endif +#endif + , _quit_flag(false) + , _activated(false) { } @@ -72,6 +75,11 @@ Engine::Engine(Ingen::Shared::World* world) Engine::~Engine() { deactivate(); + + for (size_t i=0; i < _process_slaves.size(); ++i) + delete _process_slaves[i]; + + _process_slaves.clear(); for (ObjectStore::Objects::const_iterator i = _object_store->objects().begin(); i != _object_store->objects().end(); ++i) { @@ -198,7 +206,7 @@ Engine::set_event_source(SharedPtr<EventSource> source) bool -Engine::activate() +Engine::activate(size_t parallelism) { if (_activated) return false; @@ -216,17 +224,23 @@ Engine::activate() // Create root patch - Patch* root_patch = new Patch("", 1, NULL, + Patch* root_patch = new Patch(*this, "", 1, NULL, _audio_driver->sample_rate(), _audio_driver->buffer_size(), 1); root_patch->activate(); _object_store->add(root_patch); - root_patch->process_order(root_patch->build_process_order()); - root_patch->enable(); + root_patch->compiled_patch(root_patch->compile()); assert(_audio_driver->root_patch() == NULL); _audio_driver->set_root_patch(root_patch); _audio_driver->activate(); + + _process_slaves.clear(); + _process_slaves.reserve(parallelism); + for (size_t i=0; i < parallelism - 1; ++i) + _process_slaves.push_back(new ProcessSlave(_audio_driver->is_realtime())); + + root_patch->enable(); //_post_processor->start(); diff --git a/src/libs/engine/Engine.hpp b/src/libs/engine/Engine.hpp index 562a2fa3..d4f3c972 100644 --- a/src/libs/engine/Engine.hpp +++ b/src/libs/engine/Engine.hpp @@ -19,11 +19,11 @@ #define ENGINE_H #include CONFIG_H_PATH -#include "module/module.h" - #include <cassert> +#include <vector> #include <boost/utility.hpp> #include <raul/SharedPtr.hpp> +#include "module/module.h" #include "DataType.hpp" template<typename T> class Queue; @@ -46,6 +46,7 @@ class QueuedEvent; class QueuedEngineInterface; class LashDriver; class Driver; +class ProcessSlave; /** The main class for the Engine. @@ -74,9 +75,7 @@ public: virtual SharedPtr<QueuedEngineInterface> new_queued_interface(); - //virtual void set_event_source(SharedPtr<EventSource>); - - virtual bool activate(); + virtual bool activate(size_t parallelism); virtual void deactivate(); virtual bool activated() { return _activated; } @@ -96,8 +95,12 @@ public: Driver* driver(DataType type); Ingen::Shared::World* world() { return _world; } + + typedef std::vector<ProcessSlave*> ProcessSlaves; + inline const ProcessSlaves& process_slaves() const { return _process_slaves; } private: + ProcessSlaves _process_slaves; Ingen::Shared::World* _world; diff --git a/src/libs/engine/JackAudioDriver.hpp b/src/libs/engine/JackAudioDriver.hpp index 583d2c62..9b195b13 100644 --- a/src/libs/engine/JackAudioDriver.hpp +++ b/src/libs/engine/JackAudioDriver.hpp @@ -98,7 +98,7 @@ public: inline const jack_position_t* position() { return &_position; } inline const jack_transport_state_t transport_state() { return _transport_state; } - bool is_realtime() { return jack_is_realtime(_client); } + bool is_realtime() const { return jack_is_realtime(_client); } jack_client_t* jack_client() const { return _client; } SampleCount buffer_size() const { return _buffer_size; } diff --git a/src/libs/engine/Makefile.am b/src/libs/engine/Makefile.am index 7f87c479..8d082d50 100644 --- a/src/libs/engine/Makefile.am +++ b/src/libs/engine/Makefile.am @@ -30,6 +30,7 @@ libingen_engine_la_SOURCES = \ ClientBroadcaster.hpp \ Connection.cpp \ Connection.hpp \ + CompiledPatch.hpp \ DataType.hpp \ Driver.hpp \ DuplexPort.cpp \ @@ -83,6 +84,8 @@ libingen_engine_la_SOURCES = \ Port.hpp \ PostProcessor.cpp \ PostProcessor.hpp \ + ProcessSlave.hpp \ + ProcessSlave.cpp \ QueuedEngineInterface.cpp \ QueuedEngineInterface.hpp \ QueuedEvent.hpp \ diff --git a/src/libs/engine/Node.hpp b/src/libs/engine/Node.hpp index e6c5316b..feb9e9ef 100644 --- a/src/libs/engine/Node.hpp +++ b/src/libs/engine/Node.hpp @@ -62,6 +62,37 @@ public: virtual void deactivate() = 0; virtual bool activated() = 0; + /** Parallelism: Reset flags for start of a new cycle. + */ + virtual void reset_input_ready() = 0; + + /** Parallelism: Claim this node (to wait on its input). + * Only one thread will ever take this lock on a particular Node. + * \return true if lock was aquired, false otherwise + */ + virtual bool process_lock() = 0; + + /** Parallelism: Unclaim this node (let someone else wait on its input). + * Only a thread which successfully called process_lock may call this. + */ + virtual void process_unlock() = 0; + + /** Parallelism: Wait for signal that input is ready. + * Only a thread which successfully called process_lock may call this. + */ + virtual void wait_for_input(size_t num_providers) = 0; + + /** Parallelism: Signal that input is ready. Realtime safe. + * Calling this will wake up the thread which blocked on wait_for_input + * if there is one, and otherwise cause it to return true the next call. + * \return true if lock was aquired and input is ready, false otherwise + */ + virtual void signal_input_ready() = 0; + + /** Parallelism: Return the number of providers that have signalled. + */ + virtual unsigned n_inputs_ready() const = 0; + /** Run the node for @a nframes input/output. * * @a start and @a end are transport times: end is not redundant in the case diff --git a/src/libs/engine/NodeBase.cpp b/src/libs/engine/NodeBase.cpp index b7066a0c..b61783b3 100644 --- a/src/libs/engine/NodeBase.cpp +++ b/src/libs/engine/NodeBase.cpp @@ -41,6 +41,9 @@ NodeBase::NodeBase(const Plugin* plugin, const string& name, uint32_t poly, Patc _buffer_size(buffer_size), _activated(false), _traversed(false), + _input_ready(1), + _process_lock(0), + _n_inputs_ready(0), _ports(NULL), _providers(new Raul::List<Node*>()), _dependants(new Raul::List<Node*>()) @@ -88,6 +91,55 @@ NodeBase::set_buffer_size(size_t size) for (size_t i=0; i < _ports->size(); ++i) _ports->at(i)->set_buffer_size(size); } + + +void +NodeBase::reset_input_ready() +{ + //cout << path() << " RESET" << endl; + _n_inputs_ready = 0; + _process_lock = 0; + _input_ready.reset(0); +} + + +bool +NodeBase::process_lock() +{ + return _process_lock.compare_and_exchange(0, 1); +} + + +void +NodeBase::process_unlock() +{ + _process_lock = 0; +} + + +void +NodeBase::wait_for_input(size_t num_providers) +{ + assert(_process_lock.get() == 1); + + while ((unsigned)_n_inputs_ready.get() < num_providers) { + //cout << path() << " WAITING " << _n_inputs_ready.get() << endl; + _input_ready.wait(); + //cout << path() << " CAUGHT SIGNAL" << endl; + //++_n_inputs_ready; + } + + //cout << path() << " READY" << endl; +} + + +void +NodeBase::signal_input_ready() +{ + //cout << path() << " SIGNAL" << endl; + ++_n_inputs_ready; + _input_ready.post(); +} /** Prepare to run a cycle (in the audio thread) @@ -108,7 +160,8 @@ void NodeBase::post_process(SampleCount nframes, FrameTime start, FrameTime end) { assert(_activated); - // Prepare any output ports for reading (MIDI) + + /* Write output ports */ for (size_t i=0; i < _ports->size(); ++i) _ports->at(i)->post_process(nframes, start, end); } diff --git a/src/libs/engine/NodeBase.hpp b/src/libs/engine/NodeBase.hpp index 68a60068..668f615c 100644 --- a/src/libs/engine/NodeBase.hpp +++ b/src/libs/engine/NodeBase.hpp @@ -21,6 +21,8 @@ #include "types.hpp" #include <string> #include <cstdlib> +#include <raul/Semaphore.hpp> +#include <raul/AtomicInt.hpp> #include "Node.hpp" using std::string; @@ -52,10 +54,16 @@ public: virtual void activate(); virtual void deactivate(); bool activated() { return _activated; } + + virtual void reset_input_ready(); + virtual bool process_lock(); + virtual void process_unlock(); + virtual void wait_for_input(size_t num_providers); + virtual unsigned n_inputs_ready() const { return _n_inputs_ready.get(); } - virtual void post_process(SampleCount nframes, FrameTime start, FrameTime end); - virtual void process(SampleCount nframes, FrameTime start, FrameTime end) = 0; virtual void pre_process(SampleCount nframes, FrameTime start, FrameTime end); + virtual void process(SampleCount nframes, FrameTime start, FrameTime end) = 0; + virtual void post_process(SampleCount nframes, FrameTime start, FrameTime end); virtual void set_port_buffer(uint32_t voice, uint32_t port_num, Buffer* buf) {} @@ -70,18 +78,24 @@ public: const Raul::Array<Port*>& ports() const { return *_ports; } - virtual Raul::List<Node*>* providers() { return _providers; } - virtual void providers(Raul::List<Node*>* l) { _providers = l; } + /* These are NOT to be used in the audio thread! + * The providers and dependants in CompiledNode are for that + */ + + virtual Raul::List<Node*>* providers() { return _providers; } + virtual void providers(Raul::List<Node*>* l) { _providers = l; } - virtual Raul::List<Node*>* dependants() { return _dependants; } - virtual void dependants(Raul::List<Node*>* l) { _dependants = l; } + virtual Raul::List<Node*>* dependants() { return _dependants; } + virtual void dependants(Raul::List<Node*>* l) { _dependants = l; } virtual const Plugin* plugin() const { return _plugin; } /** A node's parent is always a patch, so static cast should be safe */ Patch* parent_patch() const { return (Patch*)_parent; } -protected: +protected: + virtual void signal_input_ready(); + const Plugin* _plugin; uint32_t _poly; @@ -89,10 +103,13 @@ protected: size_t _buffer_size; bool _activated; - bool _traversed; ///< Flag for process order algorithm - Raul::Array<Port*>* _ports; ///< Access in audio thread only - Raul::List<Node*>* _providers; ///< Nodes connected to this one's input ports - Raul::List<Node*>* _dependants; ///< Nodes this one's output ports are connected to + bool _traversed; ///< Flag for process order algorithm + Raul::Semaphore _input_ready; ///< Parallelism: input ready signal + Raul::AtomicInt _process_lock; ///< Parallelism: Waiting on inputs 'lock' + Raul::AtomicInt _n_inputs_ready; ///< Parallelism: # input ready signals this cycle + Raul::Array<Port*>* _ports; ///< Access in audio thread only + Raul::List<Node*>* _providers; ///< Nodes connected to this one's input ports + Raul::List<Node*>* _dependants; ///< Nodes this one's output ports are connected to }; diff --git a/src/libs/engine/NodeFactory.cpp b/src/libs/engine/NodeFactory.cpp index dfc8c194..c2432813 100644 --- a/src/libs/engine/NodeFactory.cpp +++ b/src/libs/engine/NodeFactory.cpp @@ -58,7 +58,7 @@ NodeFactory::NodeFactory(Ingen::Shared::World* world) // Add builtin plugin types to _internal_plugins list // FIXME: ewwww, definitely a better way to do this! - Patch* parent = new Patch("dummy", 1, NULL, 1, 1, 1); + Patch* parent = new Patch(*world->local_engine, "dummy", 1, NULL, 1, 1, 1); Node* n = NULL; n = new MidiNoteNode("foo", 1, parent, 1, 1); diff --git a/src/libs/engine/Patch.cpp b/src/libs/engine/Patch.cpp index c9c0bee6..ad1459e8 100644 --- a/src/libs/engine/Patch.cpp +++ b/src/libs/engine/Patch.cpp @@ -25,16 +25,19 @@ #include "Port.hpp" #include "Connection.hpp" #include "DuplexPort.hpp" +#include "Engine.hpp" +#include "ProcessSlave.hpp" -using std::cerr; using std::cout; using std::endl; +using namespace std; namespace Ingen { -Patch::Patch(const string& path, uint32_t poly, Patch* parent, SampleRate srate, size_t buffer_size, uint32_t internal_poly) +Patch::Patch(Engine& engine, const string& path, uint32_t poly, Patch* parent, SampleRate srate, size_t buffer_size, uint32_t internal_poly) : NodeBase(new Plugin(Plugin::Patch, "ingen:patch"), path, poly, parent, srate, buffer_size), + _engine(engine), _internal_poly(internal_poly), - _process_order(NULL), + _compiled_patch(NULL), _process(false) { assert(internal_poly >= 1); @@ -62,7 +65,7 @@ Patch::~Patch() delete _nodes.erase(i); } - delete _process_order; + delete _compiled_patch; } @@ -107,30 +110,94 @@ Patch::disable() /** Run the patch for the specified number of frames. * - * Calls all Nodes in the order _process_order specifies. + * Calls all Nodes in (roughly, if parallel) the order _compiled_patch specifies. */ void Patch::process(SampleCount nframes, FrameTime start, FrameTime end) { - if (_process_order == NULL || !_process) + if (_compiled_patch == NULL || _compiled_patch->size() == 0 || !_process) return; - - // FIXME: This is far too slow, too much iteration/conditionals every cycle + CompiledPatch* const cp = _compiled_patch; + + /* Prepare input ports */ + // This breaks MIDI input, somehow (?) //for (Raul::List<Port*>::iterator i = _input_ports.begin(); i != _input_ports.end(); ++i) // (*i)->pre_process(nframes, start, end); for (Raul::List<Port*>::iterator i = _output_ports.begin(); i != _output_ports.end(); ++i) (*i)->pre_process(nframes, start, end); - // Run all nodes (consume input ports) - for (size_t i=0; i < _process_order->size(); ++i) { - // Could be a gap due to a node removal event (see RemoveNodeEvent.cpp) - // Yes, this is ugly - if (_process_order->at(i)) - _process_order->at(i)->process(nframes, start, end); + + /* Start p-1 slaves */ + + size_t n_slaves = _engine.process_slaves().size(); + + if (n_slaves >= cp->size()) + n_slaves = cp->size()-1; + + if (n_slaves > 0) { + for (size_t i=0; i < cp->size(); ++i) + (*cp)[i].node()->reset_input_ready(); + + for (size_t i=0; i < n_slaves; ++i) + _engine.process_slaves()[i]->whip(cp, i+1, nframes, start, end); } + + /* Process ourself until everything is done + * This is analogous to ProcessSlave::_whipped(), but this is the master + * (i.e. what the main Jack process thread calls). Where ProcessSlave + * waits on input, this just skips the node and tries the next, to avoid + * waiting in the Jack thread which pisses Jack off. + */ + + size_t index = 0; + //size_t run_count = 0; + size_t num_finished = 0; // Number of consecutive finished nodes hit + + while (num_finished < cp->size()) { + + CompiledNode& n = (*cp)[index]; + + if (n.node()->process_lock()) { + if (n.node()->n_inputs_ready() == n.n_providers()) { + //cout << "************ Main running " << n.node()->path() << " at index " << index << endl; + n.node()->process(nframes, start, end); + + //cerr << n.node()->path() << " @ " << &n << " dependants: " << n.dependants().size() << endl; + + /* Signal dependants their input is ready */ + for (size_t i=0; i < n.dependants().size(); ++i) + n.dependants()[i]->signal_input_ready(); + + //++run_count; + ++num_finished; + } else { + n.node()->process_unlock(); + num_finished = 0; + } + } else { + if (n.node()->n_inputs_ready() == n.n_providers()) + ++num_finished; + else + num_finished = 0; + } + + index = (index + 1) % cp->size(); + } + + /* Tell slaves we're done in case we beat them, and pray they're + * really done by the start of next cycle. + * FIXME: This probably breaks (race) at extremely small nframes. + */ + for (size_t i=0; i < n_slaves; ++i) + _engine.process_slaves()[i]->finish(); + + //cout << "Main Thread ran \t" << run_count << " nodes this cycle." << endl; + + /* Write output ports */ + for (Raul::List<Port*>::iterator i = _input_ports.begin(); i != _input_ports.end(); ++i) (*i)->post_process(nframes, start, end); for (Raul::List<Port*>::iterator i = _output_ports.begin(); i != _output_ports.end(); ++i) @@ -292,63 +359,41 @@ Patch::build_ports_array() const * * Not realtime safe. */ -Raul::Array<Node*>* -Patch::build_process_order() const +CompiledPatch* +Patch::compile() const { assert(ThreadManager::current_thread_id() == THREAD_PRE_PROCESS); //cerr << "*********** Building process order for " << path() << endl; - Raul::Array<Node*>* const process_order = new Raul::Array<Node*>(_nodes.size(), NULL); + CompiledPatch* const compiled_patch = new CompiledPatch();//_nodes.size()); // FIXME: tweak algorithm so it just ends up like this and save the cost of iteration? for (Raul::List<Node*>::const_iterator i = _nodes.begin(); i != _nodes.end(); ++i) (*i)->traversed(false); - // Traverse backwards starting at outputs - //for (Raul::List<Port*>::const_iterator p = _output_ports.begin(); p != _output_ports.end(); ++p) { - - /*const Port* const port = (*p); - for (Raul::List<Connection*>::const_iterator c = port->connections().begin(); - c != port->connections().end(); ++c) { - const Connection* const connection = (*c); - assert(connection->dst_port() == port); - assert(connection->src_port()); - assert(connection->src_port()->parent_node()); - build_process_order_recursive(connection->src_port()->parent_node(), process_order); - }*/ - //} - for (Raul::List<Node*>::const_iterator i = _nodes.begin(); i != _nodes.end(); ++i) { Node* const node = (*i); // Either a sink or connected to our output ports: if ( ( ! node->traversed()) && node->dependants()->size() == 0) - build_process_order_recursive(node, process_order); + compile_recursive(node, compiled_patch); } - - // Add any (disjoint) nodes that weren't hit by the traversal - // FIXME: this shouldn't be necessary - /*for (Raul::List<Node*>::const_iterator i = _nodes.begin(); i != _nodes.end(); ++i) { - Node* const node = (*i); - if ( ! node->traversed()) { - process_order->push_back(*i); - node->traversed(true); - cerr << "********** APPENDED DISJOINT NODE " << node->path() << endl; - } - }*/ - - /* - cerr << "----------------------------------------\n"; + + /*cerr << "----------------------------------------\n"; for (size_t i=0; i < process_order->size(); ++i) { assert(process_order->at(i)); cerr << process_order->at(i)->path() << endl; } - cerr << "----------------------------------------\n"; - */ + cerr << "----------------------------------------\n";*/ + + assert(compiled_patch->size() == _nodes.size()); - assert(process_order->size() == _nodes.size()); +#ifndef NDEBUG + for (size_t i=0; i < compiled_patch->size(); ++i) + assert(compiled_patch->at(i).node()); +#endif - return process_order; + return compiled_patch; } diff --git a/src/libs/engine/Patch.hpp b/src/libs/engine/Patch.hpp index 0b911541..7780c60b 100644 --- a/src/libs/engine/Patch.hpp +++ b/src/libs/engine/Patch.hpp @@ -24,6 +24,7 @@ #include "NodeBase.hpp" #include "Plugin.hpp" #include "DataType.hpp" +#include "CompiledPatch.hpp" using std::string; @@ -32,6 +33,8 @@ template <typename T> class Array; namespace Ingen { class Connection; +class Engine; +class CompiledPatch; /** A group of nodes in a graph, possibly polyphonic. @@ -45,7 +48,7 @@ class Connection; class Patch : public NodeBase { public: - Patch(const string& name, uint32_t poly, Patch* parent, SampleRate srate, size_t buffer_size, uint32_t local_poly); + Patch(Engine& engine, const string& name, uint32_t poly, Patch* parent, SampleRate srate, size_t buffer_size, uint32_t local_poly); virtual ~Patch(); void activate(); @@ -76,13 +79,13 @@ public: void add_connection(Raul::ListNode<Connection*>* c) { _connections.push_back(c); } Raul::ListNode<Connection*>* remove_connection(const Port* src_port, const Port* dst_port); - Raul::Array<Node*>* process_order() { return _process_order; } - void process_order(Raul::Array<Node*>* po) { _process_order = po; } + CompiledPatch* compiled_patch() { return _compiled_patch; } + void compiled_patch(CompiledPatch* cp) { _compiled_patch = cp; } Raul::Array<Port*>* external_ports() { return _ports; } void external_ports(Raul::Array<Port*>* pa) { _ports = pa; } - Raul::Array<Node*>* build_process_order() const; + CompiledPatch* compile() const; Raul::Array<Port*>* build_ports_array() const; /** Whether to run this patch's DSP bits in the audio thread */ @@ -93,32 +96,35 @@ public: uint32_t internal_poly() const { return _internal_poly; } private: - inline void build_process_order_recursive(Node* n, Raul::Array<Node*>* order) const; + inline void compile_recursive(Node* n, CompiledPatch* output) const; + Engine& _engine; uint32_t _internal_poly; - Raul::Array<Node*>* _process_order; ///< Accessed in audio thread only - Raul::List<Connection*> _connections; ///< Accessed in audio thread only - Raul::List<Port*> _input_ports; ///< Accessed in preprocessing thread only - Raul::List<Port*> _output_ports; ///< Accessed in preprocessing thread only - Raul::List<Node*> _nodes; ///< Accessed in preprocessing thread only + CompiledPatch* _compiled_patch; ///< Accessed in audio thread only + Raul::List<Connection*> _connections; ///< Accessed in audio thread only + Raul::List<Port*> _input_ports; ///< Accessed in preprocessing thread only + Raul::List<Port*> _output_ports; ///< Accessed in preprocessing thread only + Raul::List<Node*> _nodes; ///< Accessed in preprocessing thread only bool _process; }; -/** Private helper for build_process_order */ +/** Private helper for compile */ inline void -Patch::build_process_order_recursive(Node* n, Raul::Array<Node*>* order) const +Patch::compile_recursive(Node* n, CompiledPatch* output) const { - if (n == NULL || n->traversed()) return; + if (n == NULL || n->traversed()) + return; + n->traversed(true); - assert(order != NULL); + assert(output != NULL); for (Raul::List<Node*>::iterator i = n->providers()->begin(); i != n->providers()->end(); ++i) if ( ! (*i)->traversed() ) - build_process_order_recursive((*i), order); + compile_recursive((*i), output); - order->push_back(n); + output->push_back(CompiledNode(n, n->providers()->size(), n->dependants())); } diff --git a/src/libs/engine/ProcessSlave.cpp b/src/libs/engine/ProcessSlave.cpp new file mode 100644 index 00000000..fba09da4 --- /dev/null +++ b/src/libs/engine/ProcessSlave.cpp @@ -0,0 +1,83 @@ +/* This file is part of Ingen. + * Copyright (C) 2007 Dave Robillard <http://drobilla.net> + * + * Ingen is free software; you can redistribute it and/or modify it under the + * terms of the GNU General Public License as published by the Free Software + * Foundation; either version 2 of the License, or (at your option) any later + * version. + * + * Ingen is distributed in the hope that it will be useful, but WITHOUT ANY + * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS + * FOR A PARTICULAR PURPOSE. See the GNU General Public License for details. + * + * You should have received a copy of the GNU General Public License along + * with this program; if not, write to the Free Software Foundation, Inc., + * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#include <iostream> +#include "ProcessSlave.hpp" +#include "Node.hpp" +#include "CompiledPatch.hpp" + +using namespace std; + +namespace Ingen { + + +size_t ProcessSlave::_next_id = 0; + + +void +ProcessSlave::_whipped() +{ + assert(_compiled_patch); + CompiledPatch* const cp = _compiled_patch; + + /* Iterate over all nodes attempting to run immediately or block then run, + * until we've been through the entire array without getting a lock, + * and thus are finished this cycle. + */ + + //size_t run_count = 0; + size_t num_finished = 0; // Number of consecutive finished nodes hit + + while (_state == STATE_RUNNING) { + + CompiledNode& n = (*cp)[_index]; + + if (n.node()->process_lock()) { + + n.node()->wait_for_input(n.n_providers()); + + //cout << "************ Thread " << _id << " running " + // << n.node()->path() << " at index " << _index << endl; + n.node()->process(_nframes, _start, _end); + + //cerr << n.node()->path() << " @ " << &n << " dependants: " << n.dependants().size() << endl; + + /* Signal dependants their input is ready */ + for (size_t i=0; i < n.dependants().size(); ++i) + n.dependants()[i]->signal_input_ready(); + + //++run_count; + num_finished = 1; + } else { + ++num_finished; + } + + _index = (_index + 1) % cp->size(); + + if (num_finished >= cp->size()) + break; + } + + _index = 0; + _compiled_patch = NULL; + _state = STATE_FINISHED; + + //cout << "Thread " << _id << " ran \t" << run_count << " nodes this cycle." << endl; +} + + +} // namespace Ingen diff --git a/src/libs/engine/ProcessSlave.hpp b/src/libs/engine/ProcessSlave.hpp new file mode 100644 index 00000000..a5d2d9cc --- /dev/null +++ b/src/libs/engine/ProcessSlave.hpp @@ -0,0 +1,97 @@ +/* This file is part of Ingen. + * Copyright (C) 2007 Dave Robillard <http://drobilla.net> + * + * Ingen is free software; you can redistribute it and/or modify it under the + * terms of the GNU General Public License as published by the Free Software + * Foundation; either version 2 of the License, or (at your option) any later + * version. + * + * Ingen is distributed in the hope that it will be useful, but WITHOUT ANY + * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS + * FOR A PARTICULAR PURPOSE. See the GNU General Public License for details. + * + * You should have received a copy of the GNU General Public License along + * with this program; if not, write to the Free Software Foundation, Inc., + * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#ifndef PROCESS_SLAVE_HPP +#define PROCESS_SLAVE_HPP + +#include CONFIG_H_PATH + +#include <sstream> +#include <raul/Slave.hpp> +#include <raul/Array.hpp> +#include <raul/AtomicInt.hpp> +#include "types.hpp" + +namespace Ingen { + +class Node; +class CompiledPatch; + + +class ProcessSlave : protected Raul::Slave { +public: + ProcessSlave(bool realtime) + : _id(_next_id++), _state(STATE_FINISHED), _index(0), _compiled_patch(NULL) + { + std::stringstream ss; + ss << "Process Slave "; + ss << _id; + set_name(ss.str()); + + start(); + + if (realtime) + set_scheduling(SCHED_FIFO, 40); + } + + ~ProcessSlave() { + stop(); + } + + inline void whip(CompiledPatch* compiled_patch, size_t start_index, + SampleCount nframes, FrameTime start, FrameTime end) { + assert(_state == STATE_FINISHED); + _nframes = nframes; + _start = start; + _end = end; + _index = start_index; + _compiled_patch = compiled_patch; + _state = STATE_RUNNING; + Raul::Slave::whip(); + } + + inline void finish() { + while (_state.get() != STATE_FINISHED) + _state.compare_and_exchange(STATE_RUNNING, STATE_FINISH_SIGNALLED); + } + + size_t id() const { return _id; } + +private: + + void _whipped(); + + static size_t _next_id; + + static const int STATE_RUNNING = 0; + static const int STATE_FINISH_SIGNALLED = 1; + static const int STATE_FINISHED = 2; + + size_t _id; + Raul::AtomicInt _state; + SampleCount _nframes; + FrameTime _start; + FrameTime _end; + size_t _index; + CompiledPatch* _compiled_patch; +}; + + +} // namespace Ingen + +#endif // PROCESS_SLAVE_HPP + diff --git a/src/libs/engine/events/AddNodeEvent.cpp b/src/libs/engine/events/AddNodeEvent.cpp index e692a8ee..3b2b01cf 100644 --- a/src/libs/engine/events/AddNodeEvent.cpp +++ b/src/libs/engine/events/AddNodeEvent.cpp @@ -42,7 +42,7 @@ AddNodeEvent::AddNodeEvent(Engine& engine, SharedPtr<Responder> responder, Sampl _poly(poly), _patch(NULL), _node(NULL), - _process_order(NULL), + _compiled_patch(NULL), _node_already_exists(false) { } @@ -62,7 +62,7 @@ AddNodeEvent::AddNodeEvent(Engine& engine, SharedPtr<Responder> responder, Sampl _poly(poly), _patch(NULL), _node(NULL), - _process_order(NULL), + _compiled_patch(NULL), _node_already_exists(false) { } @@ -101,7 +101,7 @@ AddNodeEvent::pre_process() // FIXME: not really necessary to build process order since it's not connected, // just append to the list if (_patch->enabled()) - _process_order = _patch->build_process_order(); + _compiled_patch = _patch->compile(); } } QueuedEvent::pre_process(); @@ -114,9 +114,9 @@ AddNodeEvent::execute(SampleCount nframes, FrameTime start, FrameTime end) QueuedEvent::execute(nframes, start, end); if (_node != NULL) { - if (_patch->process_order() != NULL) - _engine.maid()->push(_patch->process_order()); - _patch->process_order(_process_order); + if (_patch->compiled_patch() != NULL) + _engine.maid()->push(_patch->compiled_patch()); + _patch->compiled_patch(_compiled_patch); } } diff --git a/src/libs/engine/events/AddNodeEvent.hpp b/src/libs/engine/events/AddNodeEvent.hpp index 28df137c..eaf48807 100644 --- a/src/libs/engine/events/AddNodeEvent.hpp +++ b/src/libs/engine/events/AddNodeEvent.hpp @@ -31,6 +31,7 @@ namespace Ingen { class Patch; class Node; class Plugin; +class CompiledPatch; /** An event to load a Node and insert it into a Patch. @@ -62,17 +63,17 @@ public: void post_process(); private: - string _patch_name; - Raul::Path _path; - string _plugin_uri; ///< If nonempty then type, library, label, are ignored - string _plugin_type; - string _plugin_lib; - string _plugin_label; - bool _poly; - Patch* _patch; - Node* _node; - Raul::Array<Node*>* _process_order; ///< Patch's new process order - bool _node_already_exists; + string _patch_name; + Raul::Path _path; + string _plugin_uri; ///< If nonempty then type, library, label, are ignored + string _plugin_type; + string _plugin_lib; + string _plugin_label; + bool _poly; + Patch* _patch; + Node* _node; + CompiledPatch* _compiled_patch; ///< Patch's new process order + bool _node_already_exists; }; diff --git a/src/libs/engine/events/ClearPatchEvent.cpp b/src/libs/engine/events/ClearPatchEvent.cpp index 2c991cf7..5a914901 100644 --- a/src/libs/engine/events/ClearPatchEvent.cpp +++ b/src/libs/engine/events/ClearPatchEvent.cpp @@ -72,9 +72,9 @@ ClearPatchEvent::execute(SampleCount nframes, FrameTime start, FrameTime end) //for (Raul::List<Node*>::const_iterator i = _patch->nodes().begin(); i != _patch->nodes().end(); ++i) // (*i)->remove_from_patch(); - if (_patch->process_order() != NULL) { - _engine.maid()->push(_patch->process_order()); - _patch->process_order(NULL); + if (_patch->compiled_patch() != NULL) { + _engine.maid()->push(_patch->compiled_patch()); + _patch->compiled_patch(NULL); } } } diff --git a/src/libs/engine/events/ConnectionEvent.cpp b/src/libs/engine/events/ConnectionEvent.cpp index a8c5dc13..69bb2645 100644 --- a/src/libs/engine/events/ConnectionEvent.cpp +++ b/src/libs/engine/events/ConnectionEvent.cpp @@ -41,7 +41,7 @@ ConnectionEvent::ConnectionEvent(Engine& engine, SharedPtr<Responder> responder, _patch(NULL), _src_port(NULL), _dst_port(NULL), - _process_order(NULL), + _compiled_patch(NULL), _connection(NULL), _patch_listnode(NULL), _port_listnode(NULL), @@ -138,7 +138,7 @@ ConnectionEvent::pre_process() } if (_patch->enabled()) - _process_order = _patch->build_process_order(); + _compiled_patch = _patch->compile(); QueuedEvent::pre_process(); } @@ -153,9 +153,9 @@ ConnectionEvent::execute(SampleCount nframes, FrameTime start, FrameTime end) // These must be inserted here, since they're actually used by the audio thread _dst_input_port->add_connection(_port_listnode); _patch->add_connection(_patch_listnode); - if (_patch->process_order() != NULL) - _engine.maid()->push(_patch->process_order()); - _patch->process_order(_process_order); + if (_patch->compiled_patch() != NULL) + _engine.maid()->push(_patch->compiled_patch()); + _patch->compiled_patch(_compiled_patch); } } @@ -170,7 +170,6 @@ ConnectionEvent::post_process() // FIXME: better error messages string msg = "Unable to make connection "; msg.append(_src_port_path + " -> " + _dst_port_path); - cerr << "CONNECTION ERROR " << (unsigned)_error << endl; _responder->respond_error(msg); } } diff --git a/src/libs/engine/events/ConnectionEvent.hpp b/src/libs/engine/events/ConnectionEvent.hpp index 1efcee1b..202fef01 100644 --- a/src/libs/engine/events/ConnectionEvent.hpp +++ b/src/libs/engine/events/ConnectionEvent.hpp @@ -39,6 +39,7 @@ class Port; class Connection; class InputPort; class OutputPort; +class CompiledPatch; /** Make a Connection between two Ports. @@ -75,7 +76,7 @@ private: OutputPort* _src_output_port; InputPort* _dst_input_port; - Raul::Array<Node*>* _process_order; ///< New process order for Patch + CompiledPatch* _compiled_patch; ///< New process order for Patch Connection* _connection; Raul::ListNode<Connection*>* _patch_listnode; diff --git a/src/libs/engine/events/CreatePatchEvent.cpp b/src/libs/engine/events/CreatePatchEvent.cpp index 6c0cd0e8..365b1a02 100644 --- a/src/libs/engine/events/CreatePatchEvent.cpp +++ b/src/libs/engine/events/CreatePatchEvent.cpp @@ -36,7 +36,7 @@ CreatePatchEvent::CreatePatchEvent(Engine& engine, SharedPtr<Responder> responde _path(path), _patch(NULL), _parent(NULL), - _process_order(NULL), + _compiled_patch(NULL), _poly(poly), _error(NO_ERROR) { @@ -69,13 +69,13 @@ CreatePatchEvent::pre_process() if (_parent != NULL && _poly > 1 && _poly == static_cast<int>(_parent->internal_poly())) poly = _poly; - _patch = new Patch(_path.name(), poly, _parent, _engine.audio_driver()->sample_rate(), _engine.audio_driver()->buffer_size(), _poly); + _patch = new Patch(_engine, _path.name(), poly, _parent, _engine.audio_driver()->sample_rate(), _engine.audio_driver()->buffer_size(), _poly); if (_parent != NULL) { _parent->add_node(new Raul::ListNode<Node*>(_patch)); if (_parent->enabled()) - _process_order = _parent->build_process_order(); + _compiled_patch = _parent->compile(); } _patch->activate(); @@ -102,9 +102,9 @@ CreatePatchEvent::execute(SampleCount nframes, FrameTime start, FrameTime end) assert(_parent != NULL); assert(_path != "/"); - if (_parent->process_order() != NULL) - _engine.maid()->push(_parent->process_order()); - _parent->process_order(_process_order); + if (_parent->compiled_patch() != NULL) + _engine.maid()->push(_parent->compiled_patch()); + _parent->compiled_patch(_compiled_patch); } } } diff --git a/src/libs/engine/events/CreatePatchEvent.hpp b/src/libs/engine/events/CreatePatchEvent.hpp index 1e45b353..1eb33ed4 100644 --- a/src/libs/engine/events/CreatePatchEvent.hpp +++ b/src/libs/engine/events/CreatePatchEvent.hpp @@ -32,6 +32,7 @@ namespace Ingen { class Patch; class Node; class Plugin; +class CompiledPatch; /** Creates a new Patch. @@ -53,8 +54,7 @@ private: Raul::Path _path; Patch* _patch; Patch* _parent; - Raul::Array<Node*>* _process_order; - TreeNode<Node*>* _patch_treenode; + CompiledPatch* _compiled_patch; int _poly; ErrorType _error; }; diff --git a/src/libs/engine/events/DestroyEvent.cpp b/src/libs/engine/events/DestroyEvent.cpp index a30bb3ae..91823a79 100644 --- a/src/libs/engine/events/DestroyEvent.cpp +++ b/src/libs/engine/events/DestroyEvent.cpp @@ -47,7 +47,7 @@ DestroyEvent::DestroyEvent(Engine& engine, SharedPtr<Responder> responder, Frame _patch_node_listnode(NULL), _patch_port_listnode(NULL), _ports_array(NULL), - _process_order(NULL), + _compiled_patch(NULL), _disconnect_node_event(NULL), _disconnect_port_event(NULL) { @@ -65,9 +65,8 @@ DestroyEvent::DestroyEvent(Engine& engine, SharedPtr<Responder> responder, Frame _driver_port(NULL), _patch_node_listnode(NULL), _patch_port_listnode(NULL), - _store_treenode(NULL), _ports_array(NULL), - _process_order(NULL), + _compiled_patch(NULL), _disconnect_node_event(NULL), _disconnect_port_event(NULL) { @@ -114,17 +113,19 @@ DestroyEvent::pre_process() if (_node->parent_patch()->enabled()) { // FIXME: is this called multiple times? - _process_order = _node->parent_patch()->build_process_order(); + _compiled_patch = _node->parent_patch()->compile(); // Remove node to be removed from the process order so it isn't executed by // Patch::run and can safely be destroyed - //for (size_t i=0; i < _process_order->size(); ++i) - // if (_process_order->at(i) == _node) - // _process_order->at(i) = NULL; // ew, gap + //for (size_t i=0; i < _compiled_patch->size(); ++i) + // if (_compiled_patch->at(i) == _node) + // _compiled_patch->at(i) = NULL; // ew, gap #ifdef DEBUG // Be sure node is removed from process order, so it can be destroyed - for (size_t i=0; i < _process_order->size(); ++i) - assert(_process_order->at(i) != _node); + for (size_t i=0; i < _compiled_patch->size(); ++i) { + assert(_compiled_patch->at(i).node() != _node); + // FIXME: check providers/dependants too + } #endif } } @@ -141,7 +142,7 @@ DestroyEvent::pre_process() if (_port->parent_patch()->enabled()) { // FIXME: is this called multiple times? - _process_order = _port->parent_patch()->build_process_order(); + _compiled_patch = _port->parent_patch()->compile(); _ports_array = _port->parent_patch()->build_ports_array(); assert(_ports_array->size() == _port->parent_patch()->num_ports()); } @@ -164,9 +165,9 @@ DestroyEvent::execute(SampleCount nframes, FrameTime start, FrameTime end) if (_disconnect_node_event) _disconnect_node_event->execute(nframes, start, end); - if (_node->parent_patch()->process_order()) - _engine.maid()->push(_node->parent_patch()->process_order()); - _node->parent_patch()->process_order(_process_order); + if (_node->parent_patch()->compiled_patch()) + _engine.maid()->push(_node->parent_patch()->compiled_patch()); + _node->parent_patch()->compiled_patch(_compiled_patch); } else if (_patch_port_listnode) { assert(_port); @@ -174,10 +175,10 @@ DestroyEvent::execute(SampleCount nframes, FrameTime start, FrameTime end) if (_disconnect_port_event) _disconnect_port_event->execute(nframes, start, end); - if (_port->parent_patch()->process_order()) - _engine.maid()->push(_port->parent_patch()->process_order()); + if (_port->parent_patch()->compiled_patch()) + _engine.maid()->push(_port->parent_patch()->compiled_patch()); - _port->parent_patch()->process_order(_process_order); + _port->parent_patch()->compiled_patch(_compiled_patch); if (_port->parent_patch()->external_ports()) _engine.maid()->push(_port->parent_patch()->external_ports()); diff --git a/src/libs/engine/events/DestroyEvent.hpp b/src/libs/engine/events/DestroyEvent.hpp index a858ee0b..c401ff31 100644 --- a/src/libs/engine/events/DestroyEvent.hpp +++ b/src/libs/engine/events/DestroyEvent.hpp @@ -41,6 +41,7 @@ class DriverPort; class Plugin; class DisconnectNodeEvent; class DisconnectPortEvent; +class CompiledPatch; /** An event to remove and delete a Node. @@ -68,7 +69,7 @@ private: Raul::ListNode<Node*>* _patch_node_listnode; Raul::ListNode<Port*>* _patch_port_listnode; Raul::Array<Port*>* _ports_array; ///< New (external) ports array for Patch - Raul::Array<Node*>* _process_order; ///< Patch's new process order + CompiledPatch* _compiled_patch; ///< Patch's new process order DisconnectNodeEvent* _disconnect_node_event; DisconnectPortEvent* _disconnect_port_event; }; diff --git a/src/libs/engine/events/DisconnectionEvent.cpp b/src/libs/engine/events/DisconnectionEvent.cpp index 794399b3..67c3b5fb 100644 --- a/src/libs/engine/events/DisconnectionEvent.cpp +++ b/src/libs/engine/events/DisconnectionEvent.cpp @@ -44,7 +44,7 @@ DisconnectionEvent::DisconnectionEvent(Engine& engine, SharedPtr<Responder> resp _src_port(NULL), _dst_port(NULL), _lookup(true), - _process_order(NULL), + _compiled_patch(NULL), _error(NO_ERROR) { } @@ -58,7 +58,7 @@ DisconnectionEvent::DisconnectionEvent(Engine& engine, SharedPtr<Responder> resp _src_port(src_port), _dst_port(dst_port), _lookup(false), - _process_order(NULL), + _compiled_patch(NULL), _error(NO_ERROR) { // FIXME: These break for patch ports.. is that ok? @@ -145,7 +145,7 @@ DisconnectionEvent::pre_process() } if (_patch->enabled()) - _process_order = _patch->build_process_order(); + _compiled_patch = _patch->compile(); QueuedEvent::pre_process(); } @@ -172,9 +172,9 @@ DisconnectionEvent::execute(SampleCount nframes, FrameTime start, FrameTime end) _engine.maid()->push(patch_connection); _engine.maid()->push(port_connection->elem()); - if (_patch->process_order() != NULL) - _engine.maid()->push(_patch->process_order()); - _patch->process_order(_process_order); + if (_patch->compiled_patch() != NULL) + _engine.maid()->push(_patch->compiled_patch()); + _patch->compiled_patch(_compiled_patch); } else { _error = CONNECTION_NOT_FOUND; } diff --git a/src/libs/engine/events/DisconnectionEvent.hpp b/src/libs/engine/events/DisconnectionEvent.hpp index 5a30b7f1..7109c645 100644 --- a/src/libs/engine/events/DisconnectionEvent.hpp +++ b/src/libs/engine/events/DisconnectionEvent.hpp @@ -39,6 +39,7 @@ class Port; class Connection; class InputPort; class OutputPort; +class CompiledPatch; /** Make a Connection between two Ports. @@ -78,7 +79,7 @@ private: bool _lookup; - Raul::Array<Node*>* _process_order; ///< New process order for Patch + CompiledPatch* _compiled_patch; ///< New process order for Patch ErrorType _error; }; diff --git a/src/libs/engine/events/EnablePatchEvent.cpp b/src/libs/engine/events/EnablePatchEvent.cpp index ed39ae04..46cc5cc6 100644 --- a/src/libs/engine/events/EnablePatchEvent.cpp +++ b/src/libs/engine/events/EnablePatchEvent.cpp @@ -30,7 +30,7 @@ EnablePatchEvent::EnablePatchEvent(Engine& engine, SharedPtr<Responder> responde : QueuedEvent(engine, responder, timestamp), _patch_path(patch_path), _patch(NULL), - _process_order(NULL) + _compiled_patch(NULL) { } @@ -44,8 +44,8 @@ EnablePatchEvent::pre_process() /* Any event that requires a new process order will set the patch's * process order to NULL if it is executed when the patch is not * active. So, if the PO is NULL, calculate it here */ - if (_patch->process_order() == NULL) - _process_order = _patch->build_process_order(); + if (_patch->compiled_patch() == NULL) + _compiled_patch = _patch->compile(); } QueuedEvent::pre_process(); @@ -60,8 +60,8 @@ EnablePatchEvent::execute(SampleCount nframes, FrameTime start, FrameTime end) if (_patch != NULL) { _patch->enable(); - if (_patch->process_order() == NULL) - _patch->process_order(_process_order); + if (_patch->compiled_patch() == NULL) + _patch->compiled_patch(_compiled_patch); } } diff --git a/src/libs/engine/events/EnablePatchEvent.hpp b/src/libs/engine/events/EnablePatchEvent.hpp index d187594b..2bacd17d 100644 --- a/src/libs/engine/events/EnablePatchEvent.hpp +++ b/src/libs/engine/events/EnablePatchEvent.hpp @@ -29,6 +29,7 @@ namespace Ingen { class Patch; class Node; +class CompiledPatch; /** Enables a patch's DSP processing. @@ -45,9 +46,9 @@ public: void post_process(); private: - string _patch_path; - Patch* _patch; - Raul::Array<Node*>* _process_order; // Patch's new process order + string _patch_path; + Patch* _patch; + CompiledPatch* _compiled_patch; // Patch's new process order }; diff --git a/src/libs/gui/ConnectWindow.cpp b/src/libs/gui/ConnectWindow.cpp index 1504176b..333d0454 100644 --- a/src/libs/gui/ConnectWindow.cpp +++ b/src/libs/gui/ConnectWindow.cpp @@ -112,7 +112,7 @@ ConnectWindow::start(SharedPtr<Ingen::Engine> engine, SharedPtr<Shared::EngineIn if (engine) { - engine->activate(); + engine->activate(1); // FIXME Glib::signal_timeout().connect( sigc::mem_fun(engine.get(), &Ingen::Engine::main_iteration), 1000); @@ -252,7 +252,7 @@ ConnectWindow::connect() App::instance().attach(engine_interface, client); - _engine->activate(); + _engine->activate(1); // FIXME Glib::signal_timeout().connect( sigc::mem_fun(_engine.get(), &Ingen::Engine::main_iteration), 1000); |