X-Git-Url: https://main.carlh.net/gitweb/?a=blobdiff_plain;f=libs%2Fardour%2Fgraph.cc;h=d183bcd7d271e41bcfa839dd3a4c4d281111e9d1;hb=84f55440a3c6317dd2ab5d7e3d638016609e4d03;hp=cc82890af7f16fe8c4b37e388097095bed681577;hpb=3ceee5b20ebf825f8b9da0208270e28c9dbf28f9;p=ardour.git diff --git a/libs/ardour/graph.cc b/libs/ardour/graph.cc index cc82890af7..d183bcd7d2 100644 --- a/libs/ardour/graph.cc +++ b/libs/ardour/graph.cc @@ -21,7 +21,6 @@ #include #include "pbd/compose.h" -#include "pbd/cpus.h" #include "pbd/debug_rt_alloc.h" #include "ardour/debug.h" @@ -42,6 +41,7 @@ using namespace std; #ifdef DEBUG_RT_ALLOC static Graph* graph = 0; + extern "C" { int alloc_allowed () @@ -52,8 +52,9 @@ int alloc_allowed () } #endif -Graph::Graph (Session & session) - : SessionHandleRef (session) +Graph::Graph (Session & session) + : SessionHandleRef (session) + , _quit_threads (false) , _execution_sem ("graph_execution", 0) , _callback_start_sem ("graph_start", 0) , _callback_done_sem ("graph_done", 0) @@ -74,76 +75,85 @@ Graph::Graph (Session & session) _quit_threads = false; _graph_empty = true; - int num_cpu = hardware_concurrency(); - int num_threads = num_cpu; - int pu = Config->get_processor_usage (); + reset_thread_list (); - if (pu < 0) { - /* pu is negative: use "pu" less cores for DSP than appear to be available - */ +#ifdef DEBUG_RT_ALLOC + graph = this; + pbd_alloc_allowed = &::alloc_allowed; +#endif +} - if (-pu < num_threads) { - num_threads += pu; - } else { - num_threads = 1; - } - } else { - /* use "pu" cores, if available - */ +/** Set up threads for running the graph */ +void +Graph::reset_thread_list () +{ + uint32_t num_threads = how_many_dsp_threads (); - if (pu <= num_threads) { - num_threads = pu; - } + /* For now, we shouldn't be using the graph code if we only have 1 DSP thread */ + assert (num_threads > 1); + + /* don't bother doing anything here if we already have the right + number of threads. + */ + + if (AudioEngine::instance()->process_thread_count() == num_threads) { + return; } - info << string_compose (_("Using %2 threads on %1 CPUs"), num_cpu, num_threads) << endmsg; + Glib::Threads::Mutex::Lock lm (_session.engine().process_lock()); - pthread_t a_thread; + if (AudioEngine::instance()->process_thread_count() != 0) { + drop_threads (); + } - if (AudioEngine::instance()->create_process_thread (boost::bind (&Graph::main_thread, this), &a_thread, 100000) == 0) { - _thread_list.push_back (a_thread); + if (AudioEngine::instance()->create_process_thread (boost::bind (&Graph::main_thread, this)) != 0) { + throw failed_constructor (); } - for (int i = 1; i < num_threads; ++i) { - if (AudioEngine::instance()->create_process_thread (boost::bind (&Graph::helper_thread, this), &a_thread, 100000) == 0) { - _thread_list.push_back (a_thread); + for (uint32_t i = 1; i < num_threads; ++i) { + if (AudioEngine::instance()->create_process_thread (boost::bind (&Graph::helper_thread, this))) { + throw failed_constructor (); } } - -#ifdef DEBUG_RT_ALLOC - graph = this; - pbd_alloc_allowed = &::alloc_allowed; -#endif } void Graph::session_going_away() +{ + drop_threads (); + + // now drop all references on the nodes. + _nodes_rt[0].clear(); + _nodes_rt[1].clear(); + _init_trigger_list[0].clear(); + _init_trigger_list[1].clear(); + _trigger_queue.clear(); +} + +void +Graph::drop_threads () { _quit_threads = true; - for (unsigned int i=0; i<_thread_list.size(); i++) { + uint32_t thread_count = AudioEngine::instance()->process_thread_count (); + + for (unsigned int i=0; i < thread_count; i++) { _execution_sem.signal (); } _callback_start_sem.signal (); - for (list::iterator i = _thread_list.begin(); i != _thread_list.end(); i++) { - void* status; - pthread_join (*i, &status); - } + AudioEngine::instance()->join_process_threads (); - // now drop all references on the nodes. - _nodes_rt[0].clear(); - _nodes_rt[1].clear(); - _init_trigger_list[0].clear(); - _init_trigger_list[1].clear(); - _trigger_queue.clear(); + _execution_tokens = 0; + + _quit_threads = false; } void Graph::clear_other_chain () { - Glib::Mutex::Lock ls (_swap_mutex); + Glib::Threads::Mutex::Lock ls (_swap_mutex); while (1) { if (_setup_chain != _pending_chain) { @@ -159,7 +169,7 @@ Graph::clear_other_chain () /* setup chain == pending chain - we have to wait till this is no longer true. */ - _cleanup_cond.wait (_swap_mutex); + _cleanup_cond.wait (_swap_mutex); } } @@ -190,147 +200,141 @@ Graph::prep() } _finished_refcount = _init_finished_refcount[chain]; + /* Trigger the initial nodes for processing, which are the ones at the `input' end */ + pthread_mutex_lock (&_trigger_mutex); for (i=_init_trigger_list[chain].begin(); i!=_init_trigger_list[chain].end(); i++) { - this->trigger( i->get() ); + /* don't use ::trigger here, as we have already locked the mutex */ + _trigger_queue.push_back (i->get ()); } + pthread_mutex_unlock (&_trigger_mutex); } void Graph::trigger (GraphNode* n) { - pthread_mutex_lock (&_trigger_mutex); - _trigger_queue.push_back( n); - pthread_mutex_unlock (&_trigger_mutex); + pthread_mutex_lock (&_trigger_mutex); + _trigger_queue.push_back (n); + pthread_mutex_unlock (&_trigger_mutex); } +/** Called when a node at the `output' end of the chain (ie one that has no-one to feed) + * is finished. + */ void Graph::dec_ref() { - if (g_atomic_int_dec_and_test (&_finished_refcount)) { + if (g_atomic_int_dec_and_test (const_cast (&_finished_refcount))) { - // ok... this cycle is finished now. - // we are the only thread alive. - - this->restart_cycle(); + /* We have run all the nodes that are at the `output' end of + the graph, so there is nothing more to do this time around. + */ + + restart_cycle (); } } void Graph::restart_cycle() { - //printf( "cycle_done chain: %d\n", _current_chain); - // we are through. wakeup our caller. + again: _callback_done_sem.signal (); - // block until we are triggered. + /* Block until the a process callback triggers us */ _callback_start_sem.wait(); - if (_quit_threads) + + if (_quit_threads) { return; + } - //printf( "cycle_start\n" ); + prep (); - this->prep(); - if (_graph_empty) + if (_graph_empty) { goto again; - //printf( "cycle_start chain: %d\n", _current_chain); + } // returning will restart the cycle. - // starting with waking up the others. + // starting with waking up the others. } -static bool -is_feedback (boost::shared_ptr routelist, Route* from, boost::shared_ptr to) -{ - for (RouteList::iterator ri=routelist->begin(); ri!=routelist->end(); ri++) { - if ((*ri).get() == from) - return false; - if ((*ri) == to) - return true; - } - assert(0); - return false; -} - -static bool -is_feedback (boost::shared_ptr routelist, boost::shared_ptr from, Route* to) -{ - for (RouteList::iterator ri=routelist->begin(); ri!=routelist->end(); ri++) { - if ((*ri).get() == to) - return true; - if ((*ri) == from) - return false; - } - assert(0); - return false; -} +/** Rechain our stuff using a list of routes (which can be in any order) and + * a directed graph of their interconnections, which is guaranteed to be + * acyclic. + */ void -Graph::rechain (boost::shared_ptr routelist) +Graph::rechain (boost::shared_ptr routelist, GraphEdges const & edges) { - node_list_t::iterator ni; - Glib::Mutex::Lock ls (_swap_mutex); + Glib::Threads::Mutex::Lock ls (_swap_mutex); int chain = _setup_chain; DEBUG_TRACE (DEBUG::Graph, string_compose ("============== setup %1\n", chain)); - // set all refcounts to 0; + /* This will become the number of nodes that do not feed any other node; + once we have processed this number of those nodes, we have finished. + */ _init_finished_refcount[chain] = 0; + + /* This will become a list of nodes that are not fed by another node, ie + those at the `input' end. + */ _init_trigger_list[chain].clear(); _nodes_rt[chain].clear(); + /* Clear things out, and make _nodes_rt[chain] a copy of routelist */ for (RouteList::iterator ri=routelist->begin(); ri!=routelist->end(); ri++) { - node_ptr_t n = boost::dynamic_pointer_cast (*ri); - - n->_init_refcount[chain] = 0; - n->_activation_set[chain].clear(); - _nodes_rt[chain].push_back(n); + (*ri)->_init_refcount[chain] = 0; + (*ri)->_activation_set[chain].clear(); + _nodes_rt[chain].push_back (*ri); } // now add refs for the connections. - for (ni=_nodes_rt[chain].begin(); ni!=_nodes_rt[chain].end(); ni++) { - bool has_input = false; - bool has_output = false; + for (node_list_t::iterator ni = _nodes_rt[chain].begin(); ni != _nodes_rt[chain].end(); ni++) { - boost::shared_ptr rp = boost::dynamic_pointer_cast( *ni); + boost::shared_ptr r = boost::dynamic_pointer_cast (*ni); - for (RouteList::iterator ri=routelist->begin(); ri!=routelist->end(); ri++) { - if (rp->direct_feeds (*ri)) { - if (is_feedback (routelist, rp.get(), *ri)) { - continue; - } + /* The routes that are directly fed by r */ + set fed_from_r = edges.from (r); - has_output = true; - (*ni)->_activation_set[chain].insert (boost::dynamic_pointer_cast (*ri) ); - } - } + /* Hence whether r has an output */ + bool const has_output = !fed_from_r.empty (); - for (Route::FedBy::iterator fi=rp->fed_by().begin(); fi!=rp->fed_by().end(); fi++) { - if (boost::shared_ptr r = fi->r.lock()) { - if (!is_feedback (routelist, r, rp.get())) { - has_input = true; - } - } - } + /* Set up r's activation set */ + for (set::iterator i = fed_from_r.begin(); i != fed_from_r.end(); ++i) { + r->_activation_set[chain].insert (*i); + } + + /* r has an input if there are some incoming edges to r in the graph */ + bool const has_input = !edges.has_none_to (r); - for (node_set_t::iterator ai=(*ni)->_activation_set[chain].begin(); ai!=(*ni)->_activation_set[chain].end(); ai++) { + /* Increment the refcount of any route that we directly feed */ + for (node_set_t::iterator ai = r->_activation_set[chain].begin(); ai != r->_activation_set[chain].end(); ai++) { (*ai)->_init_refcount[chain] += 1; } - if (!has_input) + if (!has_input) { + /* no input, so this node needs to be triggered initially to get things going */ _init_trigger_list[chain].push_back (*ni); + } - if (!has_output) + if (!has_output) { + /* no output, so this is one of the nodes that we can count off to decide + if we've finished + */ _init_finished_refcount[chain] += 1; - } + } + } _pending_chain = chain; dump(chain); } +/** Called by both the main thread and all helpers. + * @return true to quit, false to carry on. + */ bool Graph::run_one() { @@ -344,13 +348,19 @@ Graph::run_one() to_run = 0; } + /* the number of threads that are asleep */ int et = _execution_tokens; + /* the number of nodes that need to be run */ int ts = _trigger_queue.size(); + /* hence how many threads to wake up */ int wakeup = min (et, ts); + /* update the number of threads that will still be sleeping */ _execution_tokens -= wakeup; - for (int i=0; iprocess(); to_run->finish (_current_chain); - return false; -} - -static void get_rt() -{ - if (!jack_is_realtime (AudioEngine::instance()->jack())) { - return; - } - - int priority = jack_client_real_time_priority (AudioEngine::instance()->jack()); + DEBUG_TRACE(DEBUG::ProcessThreads, string_compose ("%1 has finished run_one()\n", pthread_self())); - if (priority) { - struct sched_param rtparam; - - memset (&rtparam, 0, sizeof (rtparam)); - rtparam.sched_priority = priority; - - pthread_setschedparam (pthread_self(), SCHED_FIFO, &rtparam); - } + return false; } void @@ -402,7 +397,6 @@ Graph::helper_thread() resume_rt_malloc_checks (); pt->get_buffers(); - get_rt(); while(1) { if (run_one()) { @@ -413,6 +407,7 @@ Graph::helper_thread() pt->drop_buffers(); } +/** Here's the main graph thread */ void Graph::main_thread() { @@ -421,20 +416,27 @@ Graph::main_thread() resume_rt_malloc_checks (); pt->get_buffers(); - get_rt(); again: _callback_start_sem.wait (); - DEBUG_TRACE(DEBUG::Graph, "main thread is awake\n"); - this->prep(); + + DEBUG_TRACE(DEBUG::ProcessThreads, "main thread is awake\n"); + + if (_quit_threads) { + return; + } + + prep (); if (_graph_empty && !_quit_threads) { _callback_done_sem.signal (); + DEBUG_TRACE(DEBUG::ProcessThreads, "main thread sees graph done, goes back to sleep\n"); goto again; } + /* This loop will run forever */ while (1) { - DEBUG_TRACE(DEBUG::Graph, "main thread runs one graph node\n"); + DEBUG_TRACE(DEBUG::ProcessThreads, "main thread runs one graph node\n"); if (run_one()) { break; } @@ -471,14 +473,11 @@ Graph::dump (int chain) } int -Graph::silent_process_routes (pframes_t nframes, framepos_t start_frame, framepos_t end_frame, - bool can_record, bool rec_monitors_input, bool& need_butler) +Graph::silent_process_routes (pframes_t nframes, framepos_t start_frame, framepos_t end_frame, bool& need_butler) { _process_nframes = nframes; _process_start_frame = start_frame; _process_end_frame = end_frame; - _process_can_record = can_record; - _process_rec_monitors_input = rec_monitors_input; _process_silent = true; _process_noroll = false; @@ -486,7 +485,7 @@ Graph::silent_process_routes (pframes_t nframes, framepos_t start_frame, framepo _process_need_butler = false; if (!_graph_empty) { - DEBUG_TRACE(DEBUG::Graph, "wake graph for silent process\n"); + DEBUG_TRACE(DEBUG::ProcessThreads, "wake graph for silent process\n"); _callback_start_sem.signal (); _callback_done_sem.wait (); } @@ -497,16 +496,13 @@ Graph::silent_process_routes (pframes_t nframes, framepos_t start_frame, framepo } int -Graph::process_routes (pframes_t nframes, framepos_t start_frame, framepos_t end_frame, int declick, - bool can_record, bool rec_monitors_input, bool& need_butler) +Graph::process_routes (pframes_t nframes, framepos_t start_frame, framepos_t end_frame, int declick, bool& need_butler) { - DEBUG_TRACE (DEBUG::Graph, string_compose ("graph execution from %1 to %2 = %3\n", start_frame, end_frame, nframes)); + DEBUG_TRACE (DEBUG::ProcessThreads, string_compose ("graph execution from %1 to %2 = %3\n", start_frame, end_frame, nframes)); _process_nframes = nframes; _process_start_frame = start_frame; _process_end_frame = end_frame; - _process_can_record = can_record; - _process_rec_monitors_input = rec_monitors_input; _process_declick = declick; _process_silent = false; @@ -514,11 +510,11 @@ Graph::process_routes (pframes_t nframes, framepos_t start_frame, framepos_t end _process_retval = 0; _process_need_butler = false; - DEBUG_TRACE(DEBUG::Graph, "wake graph for non-silent process\n"); + DEBUG_TRACE(DEBUG::ProcessThreads, "wake graph for non-silent process\n"); _callback_start_sem.signal (); _callback_done_sem.wait (); - DEBUG_TRACE (DEBUG::Graph, "graph execution complete\n"); + DEBUG_TRACE (DEBUG::ProcessThreads, "graph execution complete\n"); need_butler = _process_need_butler; @@ -526,15 +522,14 @@ Graph::process_routes (pframes_t nframes, framepos_t start_frame, framepos_t end } int -Graph::routes_no_roll (pframes_t nframes, framepos_t start_frame, framepos_t end_frame, - bool non_rt_pending, bool can_record, int declick) +Graph::routes_no_roll (pframes_t nframes, framepos_t start_frame, framepos_t end_frame, + bool non_rt_pending, int declick) { - DEBUG_TRACE (DEBUG::Graph, string_compose ("no-roll graph execution from %1 to %2 = %3\n", start_frame, end_frame, nframes)); + DEBUG_TRACE (DEBUG::ProcessThreads, string_compose ("no-roll graph execution from %1 to %2 = %3\n", start_frame, end_frame, nframes)); _process_nframes = nframes; _process_start_frame = start_frame; _process_end_frame = end_frame; - _process_can_record = can_record; _process_declick = declick; _process_non_rt_pending = non_rt_pending; @@ -543,7 +538,7 @@ Graph::routes_no_roll (pframes_t nframes, framepos_t start_frame, framepos_t end _process_retval = 0; _process_need_butler = false; - DEBUG_TRACE(DEBUG::Graph, "wake graph for no-roll process\n"); + DEBUG_TRACE(DEBUG::ProcessThreads, "wake graph for no-roll process\n"); _callback_start_sem.signal (); _callback_done_sem.wait (); @@ -560,19 +555,19 @@ Graph::process_one_route (Route* route) DEBUG_TRACE (DEBUG::ProcessThreads, string_compose ("%1 runs route %2\n", pthread_self(), route->name())); if (_process_silent) { - retval = route->silent_roll (_process_nframes, _process_start_frame, _process_end_frame, _process_can_record, _process_rec_monitors_input, need_butler); + retval = route->silent_roll (_process_nframes, _process_start_frame, _process_end_frame, need_butler); } else if (_process_noroll) { route->set_pending_declick (_process_declick); - retval = route->no_roll (_process_nframes, _process_start_frame, _process_end_frame, _process_non_rt_pending, _process_can_record, _process_declick); + retval = route->no_roll (_process_nframes, _process_start_frame, _process_end_frame, _process_non_rt_pending); } else { route->set_pending_declick (_process_declick); - retval = route->roll (_process_nframes, _process_start_frame, _process_end_frame, _process_declick, _process_can_record, _process_rec_monitors_input, need_butler); + retval = route->roll (_process_nframes, _process_start_frame, _process_end_frame, _process_declick, need_butler); } if (retval) { _process_retval = retval; } - + if (need_butler) { _process_need_butler = true; } @@ -581,10 +576,5 @@ Graph::process_one_route (Route* route) bool Graph::in_process_thread () const { - list::const_iterator i = _thread_list.begin (); - while (i != _thread_list.end() && *i != pthread_self ()) { - ++i; - } - - return i != _thread_list.end (); + return AudioEngine::instance()->in_process_thread (); }