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