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