libs/ardour/graph.cc

   1 /*
   2   Copyright (C) 2010 Paul Davis
   3   Author: Torben Hohn
   4
   5   This program is free software; you can redistribute it and/or modify
   6   it under the terms of the GNU General Public License as published by
   7   the Free Software Foundation; either version 2 of the License, or
   8   (at your option) any later version.
   9
  10   This program is distributed in the hope that it will be useful,
  11   but WITHOUT ANY WARRANTY; without even the implied warranty of
  12   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13   GNU General Public License for more details.
  14
  15   You should have received a copy of the GNU General Public License
  16   along with this program; if not, write to the Free Software
  17   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  18
  19 */
  20 #include <stdio.h>
  21 #include <cmath>
  22
  23 #include "pbd/compose.h"
  24 #include "pbd/debug_rt_alloc.h"
  25 #include "pbd/pthread_utils.h"
  26
  27 #include "ardour/debug.h"
  28 #include "ardour/graph.h"
  29 #include "ardour/types.h"
  30 #include "ardour/session.h"
  31 #include "ardour/route.h"
  32 #include "ardour/process_thread.h"
  33 #include "ardour/audioengine.h"
  34
  35 #include "i18n.h"
  36
  37 using namespace ARDOUR;
  38 using namespace PBD;
  39 using namespace std;
  40
  41 #ifdef DEBUG_RT_ALLOC
  42 static Graph* graph = 0;
  43
  44 extern "C" {
  45
  46 int alloc_allowed ()
  47 {
  48         return !graph->in_process_thread ();
  49 }
  50
  51 }
  52 #endif
  53
  54 Graph::Graph (Session & session)
  55         : SessionHandleRef (session)
  56         , _threads_active (false)
  57         , _execution_sem ("graph_execution", 0)
  58         , _callback_start_sem ("graph_start", 0)
  59         , _callback_done_sem ("graph_done", 0)
  60         , _cleanup_sem ("graph_cleanup", 0)
  61 {
  62         pthread_mutex_init( &_trigger_mutex, NULL);
  63
  64         /* XXX: rather hacky `fix' to stop _trigger_queue.push_back() allocating
  65            memory in the RT thread.
  66         */
  67         _trigger_queue.reserve (8192);
  68
  69         _execution_tokens = 0;
  70
  71         _current_chain = 0;
  72         _pending_chain = 0;
  73         _setup_chain   = 1;
  74         _graph_empty = true;
  75
  76
  77         ARDOUR::AudioEngine::instance()->Running.connect_same_thread (engine_connections, boost::bind (&Graph::reset_thread_list, this));
  78         ARDOUR::AudioEngine::instance()->Stopped.connect_same_thread (engine_connections, boost::bind (&Graph::engine_stopped, this));
  79         ARDOUR::AudioEngine::instance()->Halted.connect_same_thread (engine_connections, boost::bind (&Graph::engine_stopped, this));
  80
  81         reset_thread_list ();
  82
  83 #ifdef DEBUG_RT_ALLOC
  84         graph = this;
  85         pbd_alloc_allowed = &::alloc_allowed;
  86 #endif
  87 }
  88
  89 void
  90 Graph::engine_stopped ()
  91 {
  92         if (AudioEngine::instance()->process_thread_count() != 0) {
  93                 drop_threads ();
  94         }
  95 }
  96
  97 /** Set up threads for running the graph */
  98 void
  99 Graph::reset_thread_list ()
 100 {
 101         uint32_t num_threads = how_many_dsp_threads ();
 102
 103         /* For now, we shouldn't be using the graph code if we only have 1 DSP thread */
 104         assert (num_threads > 1);
 105
 106         /* don't bother doing anything here if we already have the right
 107            number of threads.
 108         */
 109
 110         if (AudioEngine::instance()->process_thread_count() == num_threads) {
 111                 return;
 112         }
 113
 114         Glib::Threads::Mutex::Lock lm (_session.engine().process_lock());
 115
 116         if (AudioEngine::instance()->process_thread_count() != 0) {
 117                 drop_threads ();
 118         }
 119
 120         if (AudioEngine::instance()->create_process_thread (boost::bind (&Graph::main_thread, this)) != 0) {
 121                 throw failed_constructor ();
 122         }
 123
 124         for (uint32_t i = 1; i < num_threads; ++i) {
 125                 if (AudioEngine::instance()->create_process_thread (boost::bind (&Graph::helper_thread, this))) {
 126                         throw failed_constructor ();
 127                 }
 128         }
 129         _threads_active = true;
 130 }
 131
 132 void
 133 Graph::session_going_away()
 134 {
 135         drop_threads ();
 136
 137         // now drop all references on the nodes.
 138         _nodes_rt[0].clear();
 139         _nodes_rt[1].clear();
 140         _init_trigger_list[0].clear();
 141         _init_trigger_list[1].clear();
 142         _trigger_queue.clear();
 143 }
 144
 145 void
 146 Graph::drop_threads ()
 147 {
 148         _threads_active = false;
 149
 150         uint32_t thread_count = AudioEngine::instance()->process_thread_count ();
 151
 152         for (unsigned int i=0; i < thread_count; i++) {
 153                 _execution_sem.signal ();
 154         }
 155
 156         _callback_start_sem.signal ();
 157
 158         AudioEngine::instance()->join_process_threads ();
 159
 160         _execution_tokens = 0;
 161 }
 162
 163 void
 164 Graph::clear_other_chain ()
 165 {
 166         Glib::Threads::Mutex::Lock ls (_swap_mutex);
 167
 168         while (1) {
 169                 if (_setup_chain != _pending_chain) {
 170
 171                         for (node_list_t::iterator ni=_nodes_rt[_setup_chain].begin(); ni!=_nodes_rt[_setup_chain].end(); ni++) {
 172                                 (*ni)->_activation_set[_setup_chain].clear();
 173                         }
 174
 175                         _nodes_rt[_setup_chain].clear ();
 176                         _init_trigger_list[_setup_chain].clear ();
 177                         break;
 178                 }
 179                 /* setup chain == pending chain - we have
 180                    to wait till this is no longer true.
 181                 */
 182                 _cleanup_cond.wait (_swap_mutex);
 183         }
 184 }
 185
 186 void
 187 Graph::prep()
 188 {
 189         node_list_t::iterator i;
 190         int chain;
 191
 192         if (_swap_mutex.trylock()) {
 193                 // we got the swap mutex.
 194                 if (_current_chain != _pending_chain)
 195                 {
 196                         // printf ("chain swap ! %d -> %d\n", _current_chain, _pending_chain);
 197                         _setup_chain = _current_chain;
 198                         _current_chain = _pending_chain;
 199                         _cleanup_cond.signal ();
 200                 }
 201                 _swap_mutex.unlock ();
 202         }
 203
 204         chain = _current_chain;
 205
 206         _graph_empty = true;
 207         for (i=_nodes_rt[chain].begin(); i!=_nodes_rt[chain].end(); i++) {
 208                 (*i)->prep( chain);
 209                 _graph_empty = false;
 210         }
 211         _finished_refcount = _init_finished_refcount[chain];
 212
 213         /* Trigger the initial nodes for processing, which are the ones at the `input' end */
 214         pthread_mutex_lock (&_trigger_mutex);
 215         for (i=_init_trigger_list[chain].begin(); i!=_init_trigger_list[chain].end(); i++) {
 216                 /* don't use ::trigger here, as we have already locked the mutex */
 217                 _trigger_queue.push_back (i->get ());
 218         }
 219         pthread_mutex_unlock (&_trigger_mutex);
 220 }
 221
 222 void
 223 Graph::trigger (GraphNode* n)
 224 {
 225         pthread_mutex_lock (&_trigger_mutex);
 226         _trigger_queue.push_back (n);
 227         pthread_mutex_unlock (&_trigger_mutex);
 228 }
 229
 230 /** Called when a node at the `output' end of the chain (ie one that has no-one to feed)
 231  *  is finished.
 232  */
 233 void
 234 Graph::dec_ref()
 235 {
 236         if (g_atomic_int_dec_and_test (const_cast<gint*> (&_finished_refcount))) {
 237
 238                 /* We have run all the nodes that are at the `output' end of
 239                    the graph, so there is nothing more to do this time around.
 240                 */
 241
 242                 restart_cycle ();
 243         }
 244 }
 245
 246 void
 247 Graph::restart_cycle()
 248 {
 249         // we are through. wakeup our caller.
 250
 251   again:
 252         _callback_done_sem.signal ();
 253
 254         /* Block until the a process callback triggers us */
 255         _callback_start_sem.wait();
 256
 257         if (!_threads_active) {
 258                 return;
 259         }
 260
 261         prep ();
 262
 263         if (_graph_empty) {
 264                 goto again;
 265         }
 266
 267         // returning will restart the cycle.
 268         // starting with waking up the others.
 269 }
 270
 271 /** Rechain our stuff using a list of routes (which can be in any order) and
 272  *  a directed graph of their interconnections, which is guaranteed to be
 273  *  acyclic.
 274  */
 275
 276 void
 277 Graph::rechain (boost::shared_ptr<RouteList> routelist, GraphEdges const & edges)
 278 {
 279         Glib::Threads::Mutex::Lock ls (_swap_mutex);
 280
 281         int chain = _setup_chain;
 282         DEBUG_TRACE (DEBUG::Graph, string_compose ("============== setup %1\n", chain));
 283
 284         /* This will become the number of nodes that do not feed any other node;
 285            once we have processed this number of those nodes, we have finished.
 286         */
 287         _init_finished_refcount[chain] = 0;
 288
 289         /* This will become a list of nodes that are not fed by another node, ie
 290            those at the `input' end.
 291         */
 292         _init_trigger_list[chain].clear();
 293
 294         _nodes_rt[chain].clear();
 295
 296         /* Clear things out, and make _nodes_rt[chain] a copy of routelist */
 297         for (RouteList::iterator ri=routelist->begin(); ri!=routelist->end(); ri++) {
 298                 (*ri)->_init_refcount[chain] = 0;
 299                 (*ri)->_activation_set[chain].clear();
 300                 _nodes_rt[chain].push_back (*ri);
 301         }
 302
 303         // now add refs for the connections.
 304
 305         for (node_list_t::iterator ni = _nodes_rt[chain].begin(); ni != _nodes_rt[chain].end(); ni++) {
 306
 307                 boost::shared_ptr<Route> r = boost::dynamic_pointer_cast<Route> (*ni);
 308
 309                 /* The routes that are directly fed by r */
 310                 set<GraphVertex> fed_from_r = edges.from (r);
 311
 312                 /* Hence whether r has an output */
 313                 bool const has_output = !fed_from_r.empty ();
 314
 315                 /* Set up r's activation set */
 316                 for (set<GraphVertex>::iterator i = fed_from_r.begin(); i != fed_from_r.end(); ++i) {
 317                         r->_activation_set[chain].insert (*i);
 318                 }
 319
 320                 /* r has an input if there are some incoming edges to r in the graph */
 321                 bool const has_input = !edges.has_none_to (r);
 322
 323                 /* Increment the refcount of any route that we directly feed */
 324                 for (node_set_t::iterator ai = r->_activation_set[chain].begin(); ai != r->_activation_set[chain].end(); ai++) {
 325                         (*ai)->_init_refcount[chain] += 1;
 326                 }
 327
 328                 if (!has_input) {
 329                         /* no input, so this node needs to be triggered initially to get things going */
 330                         _init_trigger_list[chain].push_back (*ni);
 331                 }
 332
 333                 if (!has_output) {
 334                         /* no output, so this is one of the nodes that we can count off to decide
 335                            if we've finished
 336                         */
 337                         _init_finished_refcount[chain] += 1;
 338                 }
 339         }
 340
 341         _pending_chain = chain;
 342         dump(chain);
 343 }
 344
 345 /** Called by both the main thread and all helpers.
 346  *  @return true to quit, false to carry on.
 347  */
 348 bool
 349 Graph::run_one()
 350 {
 351         GraphNode* to_run;
 352
 353         pthread_mutex_lock (&_trigger_mutex);
 354         if (_trigger_queue.size()) {
 355                 to_run = _trigger_queue.back();
 356                 _trigger_queue.pop_back();
 357         } else {
 358                 to_run = 0;
 359         }
 360
 361         /* the number of threads that are asleep */
 362         int et = _execution_tokens;
 363         /* the number of nodes that need to be run */
 364         int ts = _trigger_queue.size();
 365
 366         /* hence how many threads to wake up */
 367         int wakeup = min (et, ts);
 368         /* update the number of threads that will still be sleeping */
 369         _execution_tokens -= wakeup;
 370
 371         DEBUG_TRACE(DEBUG::ProcessThreads, string_compose ("%1 signals %2\n", pthread_name(), wakeup));
 372
 373         for (int i = 0; i < wakeup; i++) {
 374                 _execution_sem.signal ();
 375         }
 376
 377         while (to_run == 0) {
 378                 _execution_tokens += 1;
 379                 pthread_mutex_unlock (&_trigger_mutex);
 380                 DEBUG_TRACE (DEBUG::ProcessThreads, string_compose ("%1 goes to sleep\n", pthread_name()));
 381                 _execution_sem.wait ();
 382                 if (!_threads_active) {
 383                         return true;
 384                 }
 385                 DEBUG_TRACE (DEBUG::ProcessThreads, string_compose ("%1 is awake\n", pthread_name()));
 386                 pthread_mutex_lock (&_trigger_mutex);
 387                 if (_trigger_queue.size()) {
 388                         to_run = _trigger_queue.back();
 389                         _trigger_queue.pop_back();
 390                 }
 391         }
 392         pthread_mutex_unlock (&_trigger_mutex);
 393
 394         to_run->process();
 395         to_run->finish (_current_chain);
 396
 397         DEBUG_TRACE(DEBUG::ProcessThreads, string_compose ("%1 has finished run_one()\n", pthread_name()));
 398
 399         return false;
 400 }
 401
 402 void
 403 Graph::helper_thread()
 404 {
 405         suspend_rt_malloc_checks ();
 406         ProcessThread* pt = new ProcessThread ();
 407         resume_rt_malloc_checks ();
 408
 409         pt->get_buffers();
 410
 411         while(1) {
 412                 if (run_one()) {
 413                         break;
 414                 }
 415         }
 416
 417         pt->drop_buffers();
 418 }
 419
 420 /** Here's the main graph thread */
 421 void
 422 Graph::main_thread()
 423 {
 424         suspend_rt_malloc_checks ();
 425         ProcessThread* pt = new ProcessThread ();
 426         resume_rt_malloc_checks ();
 427
 428         pt->get_buffers();
 429
 430   again:
 431         _callback_start_sem.wait ();
 432
 433         DEBUG_TRACE(DEBUG::ProcessThreads, "main thread is awake\n");
 434
 435         if (!_threads_active) {
 436                 return;
 437         }
 438
 439         prep ();
 440
 441         if (_graph_empty && _threads_active) {
 442                 _callback_done_sem.signal ();
 443                 DEBUG_TRACE(DEBUG::ProcessThreads, "main thread sees graph done, goes back to sleep\n");
 444                 goto again;
 445         }
 446
 447         /* This loop will run forever */
 448         while (1) {
 449                 DEBUG_TRACE(DEBUG::ProcessThreads, "main thread runs one graph node\n");
 450                 if (run_one()) {
 451                         break;
 452                 }
 453         }
 454
 455         pt->drop_buffers();
 456 }
 457
 458 void
 459 Graph::dump (int chain)
 460 {
 461 #ifndef NDEBUG
 462         node_list_t::iterator ni;
 463         node_set_t::iterator ai;
 464
 465         chain = _pending_chain;
 466
 467         DEBUG_TRACE (DEBUG::Graph, "--------------------------------------------Graph dump:\n");
 468         for (ni=_nodes_rt[chain].begin(); ni!=_nodes_rt[chain].end(); ni++) {
 469                 boost::shared_ptr<Route> rp = boost::dynamic_pointer_cast<Route>( *ni);
 470                 DEBUG_TRACE (DEBUG::Graph, string_compose ("GraphNode: %1  refcount: %2\n", rp->name().c_str(), (*ni)->_init_refcount[chain]));
 471                 for (ai=(*ni)->_activation_set[chain].begin(); ai!=(*ni)->_activation_set[chain].end(); ai++) {
 472                         DEBUG_TRACE (DEBUG::Graph, string_compose ("  triggers: %1\n", boost::dynamic_pointer_cast<Route>(*ai)->name().c_str()));
 473                 }
 474         }
 475
 476         DEBUG_TRACE (DEBUG::Graph, "------------- trigger list:\n");
 477         for (ni=_init_trigger_list[chain].begin(); ni!=_init_trigger_list[chain].end(); ni++) {
 478                 DEBUG_TRACE (DEBUG::Graph, string_compose ("GraphNode: %1  refcount: %2\n", boost::dynamic_pointer_cast<Route>(*ni)->name().c_str(), (*ni)->_init_refcount[chain]));
 479         }
 480
 481         DEBUG_TRACE (DEBUG::Graph, string_compose ("final activation refcount: %1\n", _init_finished_refcount[chain]));
 482 #endif
 483 }
 484
 485 int
 486 Graph::silent_process_routes (pframes_t nframes, framepos_t start_frame, framepos_t end_frame, bool& need_butler)
 487 {
 488         if (!_threads_active) return 0;
 489
 490         _process_nframes = nframes;
 491         _process_start_frame = start_frame;
 492         _process_end_frame = end_frame;
 493
 494         _process_silent = true;
 495         _process_noroll = false;
 496         _process_retval = 0;
 497         _process_need_butler = false;
 498
 499         if (!_graph_empty) {
 500                 DEBUG_TRACE(DEBUG::ProcessThreads, "wake graph for silent process\n");
 501                 _callback_start_sem.signal ();
 502                 _callback_done_sem.wait ();
 503         }
 504
 505         need_butler = _process_need_butler;
 506
 507         return _process_retval;
 508 }
 509
 510 int
 511 Graph::process_routes (pframes_t nframes, framepos_t start_frame, framepos_t end_frame, int declick, bool& need_butler)
 512 {
 513         DEBUG_TRACE (DEBUG::ProcessThreads, string_compose ("graph execution from %1 to %2 = %3\n", start_frame, end_frame, nframes));
 514
 515         if (!_threads_active) return 0;
 516
 517         _process_nframes = nframes;
 518         _process_start_frame = start_frame;
 519         _process_end_frame = end_frame;
 520         _process_declick = declick;
 521
 522         _process_silent = false;
 523         _process_noroll = false;
 524         _process_retval = 0;
 525         _process_need_butler = false;
 526
 527         DEBUG_TRACE(DEBUG::ProcessThreads, "wake graph for non-silent process\n");
 528         _callback_start_sem.signal ();
 529         _callback_done_sem.wait ();
 530
 531         DEBUG_TRACE (DEBUG::ProcessThreads, "graph execution complete\n");
 532
 533         need_butler = _process_need_butler;
 534
 535         return _process_retval;
 536 }
 537
 538 int
 539 Graph::routes_no_roll (pframes_t nframes, framepos_t start_frame, framepos_t end_frame,
 540                        bool non_rt_pending, int declick)
 541 {
 542         DEBUG_TRACE (DEBUG::ProcessThreads, string_compose ("no-roll graph execution from %1 to %2 = %3\n", start_frame, end_frame, nframes));
 543
 544         if (!_threads_active) return 0;
 545
 546         _process_nframes = nframes;
 547         _process_start_frame = start_frame;
 548         _process_end_frame = end_frame;
 549         _process_declick = declick;
 550         _process_non_rt_pending = non_rt_pending;
 551
 552         _process_silent = false;
 553         _process_noroll = true;
 554         _process_retval = 0;
 555         _process_need_butler = false;
 556
 557         DEBUG_TRACE(DEBUG::ProcessThreads, "wake graph for no-roll process\n");
 558         _callback_start_sem.signal ();
 559         _callback_done_sem.wait ();
 560
 561         return _process_retval;
 562 }
 563 void
 564 Graph::process_one_route (Route* route)
 565 {
 566         bool need_butler = false;
 567         int retval;
 568
 569         assert (route);
 570
 571         DEBUG_TRACE (DEBUG::ProcessThreads, string_compose ("%1 runs route %2\n", pthread_name(), route->name()));
 572
 573         if (_process_silent) {
 574                 retval = route->silent_roll (_process_nframes, _process_start_frame, _process_end_frame, need_butler);
 575         } else if (_process_noroll) {
 576                 route->set_pending_declick (_process_declick);
 577                 retval = route->no_roll (_process_nframes, _process_start_frame, _process_end_frame, _process_non_rt_pending);
 578         } else {
 579                 route->set_pending_declick (_process_declick);
 580                 retval = route->roll (_process_nframes, _process_start_frame, _process_end_frame, _process_declick, need_butler);
 581         }
 582
 583         if (retval) {
 584                 _process_retval = retval;
 585         }
 586
 587         if (need_butler) {
 588                 _process_need_butler = true;
 589         }
 590 }
 591
 592 bool
 593 Graph::in_process_thread () const
 594 {
 595         return AudioEngine::instance()->in_process_thread ();
 596 }