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