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/cpus.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
  43 Graph::Graph (Session & session)
  44         : SessionHandleRef (session)
  45         , _execution_sem ("graph_execution", 0)
  46         , _callback_start_sem ("graph_start", 0)
  47         , _callback_done_sem ("graph_done", 0)
  48         , _cleanup_sem ("graph_cleanup", 0)
  49 {
  50         pthread_mutex_init( &_trigger_mutex, NULL);
  51
  52         _execution_tokens = 0;
  53
  54         _current_chain = 0;
  55         _pending_chain = 0;
  56         _setup_chain   = 1;
  57         _quit_threads = false;
  58         _graph_empty = true;
  59
  60         int num_cpu = hardware_concurrency();
  61         int num_threads = num_cpu;
  62         int pu = Config->get_processor_usage ();
  63
  64         if (pu < 0) {
  65                 /* pu is negative: use "pu" less cores for DSP than appear to be available
  66                  */
  67
  68                 if (-pu < num_threads) {
  69                         num_threads += pu;
  70                 } else {
  71                         num_threads = 1;
  72                 }
  73         } else {
  74                 /* use "pu" cores, if available
  75                  */
  76
  77                 if (pu <= num_threads) {
  78                         num_threads = pu;
  79                 }
  80         }
  81
  82         info << string_compose (_("Using %2 threads on %1 CPUs"), num_cpu, num_threads) << endmsg;
  83
  84         pthread_t a_thread;
  85
  86         if (AudioEngine::instance()->create_process_thread (boost::bind (&Graph::main_thread, this), &a_thread, 100000) == 0) {
  87                 _thread_list.push_back (a_thread);
  88         }
  89
  90         for (int i = 1; i < num_threads; ++i) {
  91                 if (AudioEngine::instance()->create_process_thread (boost::bind (&Graph::helper_thread, this), &a_thread, 100000) == 0) {
  92                         _thread_list.push_back (a_thread);
  93                 }
  94         }
  95 }
  96
  97 void
  98 Graph::session_going_away()
  99 {
 100         _quit_threads = true;
 101
 102         for (unsigned int i=0; i<_thread_list.size(); i++) {
 103                 _execution_sem.signal ();
 104         }
 105
 106         _callback_start_sem.signal ();
 107
 108         for (list<pthread_t>::iterator i = _thread_list.begin(); i != _thread_list.end(); i++) {
 109                 void* status;
 110                 pthread_join (*i, &status);
 111         }
 112
 113         // now drop all references on the nodes.
 114         _nodes_rt[0].clear();
 115         _nodes_rt[1].clear();
 116         _init_trigger_list[0].clear();
 117         _init_trigger_list[1].clear();
 118         _trigger_queue.clear();
 119 }
 120
 121 void
 122 Graph::clear_other_chain ()
 123 {
 124         Glib::Mutex::Lock ls (_swap_mutex);
 125
 126         while (1) {
 127                 if (_setup_chain != _pending_chain) {
 128
 129                         for (node_list_t::iterator ni=_nodes_rt[_setup_chain].begin(); ni!=_nodes_rt[_setup_chain].end(); ni++) {
 130                                 (*ni)->_activation_set[_setup_chain].clear();
 131                         }
 132
 133                         _nodes_rt[_setup_chain].clear ();
 134                         _init_trigger_list[_setup_chain].clear ();
 135                         break;
 136                 }
 137                 /* setup chain == pending chain - we have
 138                    to wait till this is no longer true.
 139                 */
 140                 _cleanup_cond.wait (_swap_mutex);
 141         }
 142 }
 143
 144 void
 145 Graph::prep()
 146 {
 147         node_list_t::iterator i;
 148         int chain;
 149
 150         if (_swap_mutex.trylock()) {
 151                 // we got the swap mutex.
 152                 if (_current_chain != _pending_chain)
 153                 {
 154                         // printf ("chain swap ! %d -> %d\n", _current_chain, _pending_chain);
 155                         _setup_chain = _current_chain;
 156                         _current_chain = _pending_chain;
 157                         _cleanup_cond.signal ();
 158                 }
 159                 _swap_mutex.unlock ();
 160         }
 161
 162         chain = _current_chain;
 163
 164         _graph_empty = true;
 165         for (i=_nodes_rt[chain].begin(); i!=_nodes_rt[chain].end(); i++) {
 166                 (*i)->prep( chain);
 167                 _graph_empty = false;
 168         }
 169         _finished_refcount = _init_finished_refcount[chain];
 170
 171         for (i=_init_trigger_list[chain].begin(); i!=_init_trigger_list[chain].end(); i++) {
 172                 this->trigger( i->get() );
 173         }
 174 }
 175
 176 void
 177 Graph::trigger (GraphNode* n)
 178 {
 179         pthread_mutex_lock (&_trigger_mutex);
 180         _trigger_queue.push_back( n);
 181         pthread_mutex_unlock (&_trigger_mutex);
 182 }
 183
 184 void
 185 Graph::dec_ref()
 186 {
 187         if (g_atomic_int_dec_and_test (&_finished_refcount)) {
 188
 189                 // ok... this cycle is finished now.
 190                 // we are the only thread alive.
 191
 192                 this->restart_cycle();
 193         }
 194 }
 195
 196 void
 197 Graph::restart_cycle()
 198 {
 199         //printf( "cycle_done chain: %d\n", _current_chain);
 200
 201         // we are through. wakeup our caller.
 202   again:
 203         _callback_done_sem.signal ();
 204
 205         // block until we are triggered.
 206         _callback_start_sem.wait();
 207         if (_quit_threads)
 208                 return;
 209
 210         //printf( "cycle_start\n" );
 211
 212         this->prep();
 213         if (_graph_empty)
 214                 goto again;
 215         //printf( "cycle_start chain: %d\n", _current_chain);
 216
 217         // returning will restart the cycle.
 218         //  starting with waking up the others.
 219 }
 220
 221 static bool
 222 is_feedback (boost::shared_ptr<RouteList> routelist, Route* from, boost::shared_ptr<Route> to)
 223 {
 224         for (RouteList::iterator ri=routelist->begin(); ri!=routelist->end(); ri++) {
 225                 if ((*ri).get() == from)
 226                         return false;
 227                 if ((*ri) == to)
 228                         return true;
 229         }
 230         assert(0);
 231         return false;
 232 }
 233
 234 static bool
 235 is_feedback (boost::shared_ptr<RouteList> routelist, boost::shared_ptr<Route> from, Route* to)
 236 {
 237         for (RouteList::iterator ri=routelist->begin(); ri!=routelist->end(); ri++) {
 238                 if ((*ri).get() == to)
 239                         return true;
 240                 if ((*ri) == from)
 241                         return false;
 242         }
 243         assert(0);
 244         return false;
 245 }
 246
 247 void
 248 Graph::rechain (boost::shared_ptr<RouteList> routelist)
 249 {
 250         node_list_t::iterator ni;
 251         Glib::Mutex::Lock ls (_swap_mutex);
 252
 253         int chain = _setup_chain;
 254         DEBUG_TRACE (DEBUG::Graph, string_compose ("============== setup %1\n", chain));
 255         // set all refcounts to 0;
 256
 257         _init_finished_refcount[chain] = 0;
 258         _init_trigger_list[chain].clear();
 259
 260         _nodes_rt[chain].clear();
 261
 262         for (RouteList::iterator ri=routelist->begin(); ri!=routelist->end(); ri++) {
 263                 node_ptr_t n = boost::dynamic_pointer_cast<GraphNode> (*ri);
 264
 265                 n->_init_refcount[chain] = 0;
 266                 n->_activation_set[chain].clear();
 267                 _nodes_rt[chain].push_back(n);
 268         }
 269
 270         // now add refs for the connections.
 271
 272         for (ni=_nodes_rt[chain].begin(); ni!=_nodes_rt[chain].end(); ni++) {
 273                 bool has_input  = false;
 274                 bool has_output = false;
 275
 276                 boost::shared_ptr<Route> rp = boost::dynamic_pointer_cast<Route>( *ni);
 277
 278                 for (RouteList::iterator ri=routelist->begin(); ri!=routelist->end(); ri++) {
 279                         if (rp->direct_feeds (*ri)) {
 280                                 if (is_feedback (routelist, rp.get(), *ri)) {
 281                                         continue;
 282                                 }
 283
 284                                 has_output = true;
 285                                 (*ni)->_activation_set[chain].insert (boost::dynamic_pointer_cast<GraphNode> (*ri) );
 286                         }
 287                 }
 288
 289                 for (Route::FedBy::iterator fi=rp->fed_by().begin(); fi!=rp->fed_by().end(); fi++) {
 290                         if (boost::shared_ptr<Route> r = fi->r.lock()) {
 291                                 if (!is_feedback (routelist, r, rp.get())) {
 292                                         has_input = true;
 293                                 }
 294                         }
 295                 }
 296
 297                 for (node_set_t::iterator ai=(*ni)->_activation_set[chain].begin(); ai!=(*ni)->_activation_set[chain].end(); ai++) {
 298                         (*ai)->_init_refcount[chain] += 1;
 299                 }
 300
 301                 if (!has_input)
 302                         _init_trigger_list[chain].push_back (*ni);
 303
 304                 if (!has_output)
 305                         _init_finished_refcount[chain] += 1;
 306         }
 307
 308         _pending_chain = chain;
 309         dump(chain);
 310 }
 311
 312 bool
 313 Graph::run_one()
 314 {
 315         GraphNode* to_run;
 316
 317         pthread_mutex_lock (&_trigger_mutex);
 318         if (_trigger_queue.size()) {
 319                 to_run = _trigger_queue.back();
 320                 _trigger_queue.pop_back();
 321         } else {
 322                 to_run = 0;
 323         }
 324
 325         int et = _execution_tokens;
 326         int ts = _trigger_queue.size();
 327
 328         int wakeup = min (et, ts);
 329         _execution_tokens -= wakeup;
 330
 331         for (int i=0; i<wakeup; i++ ) {
 332                 _execution_sem.signal ();
 333         }
 334
 335         while (to_run == 0) {
 336                 _execution_tokens += 1;
 337                 pthread_mutex_unlock (&_trigger_mutex);
 338                 DEBUG_TRACE (DEBUG::ProcessThreads, string_compose ("%1 goes to sleep\n", pthread_self()));
 339                 _execution_sem.wait ();
 340                 if (_quit_threads)
 341                         return true;
 342                 DEBUG_TRACE (DEBUG::ProcessThreads, string_compose ("%1 is awake\n", pthread_self()));
 343                 pthread_mutex_lock (&_trigger_mutex);
 344                 if (_trigger_queue.size()) {
 345                         to_run = _trigger_queue.back();
 346                         _trigger_queue.pop_back();
 347                 }
 348         }
 349         pthread_mutex_unlock (&_trigger_mutex);
 350
 351         to_run->process();
 352         to_run->finish (_current_chain);
 353
 354         return false;
 355 }
 356
 357 static void get_rt()
 358 {
 359         if (!jack_is_realtime (AudioEngine::instance()->jack())) {
 360                 return;
 361         }
 362
 363         int priority = jack_client_real_time_priority (AudioEngine::instance()->jack());
 364
 365         if (priority) {
 366                 struct sched_param rtparam;
 367
 368                 memset (&rtparam, 0, sizeof (rtparam));
 369                 rtparam.sched_priority = priority;
 370
 371                 pthread_setschedparam (pthread_self(), SCHED_FIFO, &rtparam);
 372         }
 373 }
 374
 375 void
 376 Graph::helper_thread()
 377 {
 378         ProcessThread *pt = new ProcessThread;
 379
 380         pt->get_buffers();
 381         get_rt();
 382
 383         while(1) {
 384                 if (run_one()) {
 385                         break;
 386                 }
 387         }
 388
 389         pt->drop_buffers();
 390 }
 391
 392 void
 393 Graph::main_thread()
 394 {
 395         ProcessThread *pt = new ProcessThread;
 396
 397         pt->get_buffers();
 398         get_rt();
 399
 400   again:
 401         _callback_start_sem.wait ();
 402         DEBUG_TRACE(DEBUG::Graph, "main thread is awake\n");
 403         this->prep();
 404
 405         if (_graph_empty && !_quit_threads) {
 406                 _callback_done_sem.signal ();
 407                 goto again;
 408         }
 409
 410         while (1) {
 411                 DEBUG_TRACE(DEBUG::Graph, "main thread runs one graph node\n");
 412                 if (run_one()) {
 413                         break;
 414                 }
 415         }
 416
 417         pt->drop_buffers();
 418 }
 419
 420 void
 421 Graph::dump (int chain)
 422 {
 423 #ifndef NDEBUG
 424         node_list_t::iterator ni;
 425         node_set_t::iterator ai;
 426
 427         chain = _pending_chain;
 428
 429         DEBUG_TRACE (DEBUG::Graph, "--------------------------------------------Graph dump:\n");
 430         for (ni=_nodes_rt[chain].begin(); ni!=_nodes_rt[chain].end(); ni++) {
 431                 boost::shared_ptr<Route> rp = boost::dynamic_pointer_cast<Route>( *ni);
 432                 DEBUG_TRACE (DEBUG::Graph, string_compose ("GraphNode: %1  refcount: %2\n", rp->name().c_str(), (*ni)->_init_refcount[chain]));
 433                 for (ai=(*ni)->_activation_set[chain].begin(); ai!=(*ni)->_activation_set[chain].end(); ai++) {
 434                         DEBUG_TRACE (DEBUG::Graph, string_compose ("  triggers: %1\n", boost::dynamic_pointer_cast<Route>(*ai)->name().c_str()));
 435                 }
 436         }
 437
 438         DEBUG_TRACE (DEBUG::Graph, "------------- trigger list:\n");
 439         for (ni=_init_trigger_list[chain].begin(); ni!=_init_trigger_list[chain].end(); ni++) {
 440                 DEBUG_TRACE (DEBUG::Graph, string_compose ("GraphNode: %1  refcount: %2\n", boost::dynamic_pointer_cast<Route>(*ni)->name().c_str(), (*ni)->_init_refcount[chain]));
 441         }
 442
 443         DEBUG_TRACE (DEBUG::Graph, string_compose ("final activation refcount: %1\n", _init_finished_refcount[chain]));
 444 #endif
 445 }
 446
 447 int
 448 Graph::silent_process_routes (nframes_t nframes, framepos_t start_frame, framepos_t end_frame,
 449                               bool can_record, bool rec_monitors_input, bool& need_butler)
 450 {
 451         _process_nframes = nframes;
 452         _process_start_frame = start_frame;
 453         _process_end_frame = end_frame;
 454         _process_can_record = can_record;
 455         _process_rec_monitors_input = rec_monitors_input;
 456
 457         _process_silent = true;
 458         _process_noroll = false;
 459         _process_retval = 0;
 460         _process_need_butler = false;
 461
 462         if (!_graph_empty) {
 463                 DEBUG_TRACE(DEBUG::Graph, "wake graph for silent process\n");
 464                 _callback_start_sem.signal ();
 465                 _callback_done_sem.wait ();
 466         }
 467
 468         need_butler = _process_need_butler;
 469
 470         return _process_retval;
 471 }
 472
 473 int
 474 Graph::process_routes (nframes_t nframes, framepos_t start_frame, framepos_t end_frame, int declick,
 475                        bool can_record, bool rec_monitors_input, bool& need_butler)
 476 {
 477         DEBUG_TRACE (DEBUG::Graph, string_compose ("graph execution from %1 to %2 = %3\n", start_frame, end_frame, nframes));
 478
 479         _process_nframes = nframes;
 480         _process_start_frame = start_frame;
 481         _process_end_frame = end_frame;
 482         _process_can_record = can_record;
 483         _process_rec_monitors_input = rec_monitors_input;
 484         _process_declick = declick;
 485
 486         _process_silent = false;
 487         _process_noroll = false;
 488         _process_retval = 0;
 489         _process_need_butler = false;
 490
 491         DEBUG_TRACE(DEBUG::Graph, "wake graph for non-silent process\n");
 492         _callback_start_sem.signal ();
 493         _callback_done_sem.wait ();
 494
 495         DEBUG_TRACE (DEBUG::Graph, "graph execution complete\n");
 496
 497         need_butler = _process_need_butler;
 498
 499         return _process_retval;
 500 }
 501
 502 int
 503 Graph::routes_no_roll (nframes_t nframes, framepos_t start_frame, framepos_t end_frame,
 504                        bool non_rt_pending, bool can_record, int declick)
 505 {
 506         DEBUG_TRACE (DEBUG::Graph, string_compose ("no-roll 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_can_record = can_record;
 512         _process_declick = declick;
 513         _process_non_rt_pending = non_rt_pending;
 514
 515         _process_silent = false;
 516         _process_noroll = true;
 517         _process_retval = 0;
 518         _process_need_butler = false;
 519
 520         DEBUG_TRACE(DEBUG::Graph, "wake graph for no-roll process\n");
 521         _callback_start_sem.signal ();
 522         _callback_done_sem.wait ();
 523
 524         return _process_retval;
 525 }
 526 void
 527 Graph::process_one_route (Route* route)
 528 {
 529         bool need_butler = false;
 530         int retval;
 531
 532         assert (route);
 533
 534         DEBUG_TRACE (DEBUG::ProcessThreads, string_compose ("%1 runs route %2\n", pthread_self(), route->name()));
 535
 536         if (_process_silent) {
 537                 retval = route->silent_roll (_process_nframes, _process_start_frame, _process_end_frame, _process_can_record, _process_rec_monitors_input, need_butler);
 538         } else if (_process_noroll) {
 539                 route->set_pending_declick (_process_declick);
 540                 retval = route->no_roll (_process_nframes, _process_start_frame, _process_end_frame, _process_non_rt_pending, _process_can_record, _process_declick);
 541         } else {
 542                 route->set_pending_declick (_process_declick);
 543                 retval = route->roll (_process_nframes, _process_start_frame, _process_end_frame, _process_declick, _process_can_record, _process_rec_monitors_input, need_butler);
 544         }
 545
 546         if (retval) {
 547                 _process_retval = retval;
 548         }
 549
 550         if (need_butler) {
 551                 _process_need_butler = true;
 552         }
 553 }
 554
 555
 556