Line data    Source code

       1              : // DUNE DAQ modification notice:
       2              : // This file has been modified from the original ATLAS oks source for the DUNE DAQ project.
       3              : // Fork baseline commit: oks-08-03-04 (2022-04-14).
       4              : // Renamed since fork: no.
       5              : 
       6              : ////////////////////////////////////////////////////////////////////////////
       7              : //
       8              : //    copy from IPC
       9              : //
      10              : ////////////////////////////////////////////////////////////////////////////
      11              : 
      12              : #include "oks/pipeline.hpp"
      13              : #include <boost/bind/bind.hpp>
      14              : 
      15              : namespace dunedaq {
      16              : namespace oks {
      17              : 
      18              : bool
      19          168 : OksPipeline::Worker::setJob( OksJob * job )
      20              : {
      21          168 :     boost::mutex::scoped_lock lock( m_mutex );
      22          168 :     if ( !m_running && m_idle )
      23              :     {
      24          125 :         m_job = job;
      25          125 :         m_idle = false;
      26          125 :         m_condition.notify_one();
      27          125 :         return true;
      28              :     }
      29              :     return false;
      30          168 : }
      31              : 
      32              : void
      33        10496 : OksPipeline::Worker::run( )
      34              : {
      35        10496 :     {
      36        10496 :         boost::mutex::scoped_lock lock( m_mutex );
      37        10496 :         m_pipeline.m_barrier.wait();
      38        10481 :     }
      39        20745 :     while ( true )
      40              :     {
      41        31212 :         {
      42        31212 :             boost::mutex::scoped_lock lock( m_mutex );
      43        31113 :             m_running = true;
      44        31113 :         }
      45        31388 :         while ( !m_idle ) {
      46          187 :             m_job->run();
      47          125 :             delete m_job;
      48          125 :             m_idle = !m_pipeline.getJob( m_job );
      49              :         }
      50        31201 :         {
      51        31201 :             boost::mutex::scoped_lock lock( m_mutex );
      52        31133 :             m_running = false;
      53        31133 :             if ( m_shutdown ) 
      54              :                 break;
      55        20652 :             if ( m_stop ) {
      56        10205 :                 m_pipeline.m_barrier.wait();
      57        10482 :                 m_stop = false;
      58              :             }
      59        20929 :             if ( m_idle )
      60        20943 :                 m_condition.wait( lock ); 
      61        31331 :         }
      62              :     }
      63        10475 : }
      64              : 
      65              : void
      66        10496 : OksPipeline::Worker::shutdown( )
      67              : {
      68        10496 :     boost::mutex::scoped_lock lock( m_mutex );
      69        10496 :     m_shutdown = true;
      70        10496 :     m_condition.notify_one( );
      71        10496 : }
      72              : 
      73              : void
      74        10496 : OksPipeline::Worker::stop( )
      75              : {
      76        10496 :     boost::mutex::scoped_lock lock( m_mutex );
      77        10496 :     m_stop = true;
      78        10496 :     m_condition.notify_one( );
      79        10496 : }
      80              : 
      81           82 : OksPipeline::OksPipeline( size_t size )
      82           82 :   : m_barrier( size + 1 )
      83              : {
      84        10578 :     for ( size_t i = 0; i < size; ++i )
      85              :     {
      86        10496 :         m_workers.push_back( WorkerPtr( new Worker( *this ) ) );
      87        10496 :         m_pool.create_thread( boost::bind( &Worker::run, m_workers.back().get() ) );
      88              :     }
      89           82 :     m_barrier.wait();
      90           82 : }
      91              :  
      92           82 : OksPipeline::~OksPipeline()
      93              : {
      94           82 :     waitForCompletion();
      95        10578 :     for ( size_t i = 0; i < m_workers.size(); ++i )
      96              :     {
      97        10496 :         m_workers[i] -> shutdown();
      98              :     }
      99           82 :     m_pool.join_all();
     100           82 : }
     101              : 
     102              : void
     103          125 : OksPipeline::addJob( OksJob * job )
     104              : {
     105          168 :     for ( size_t i = 0; i < m_workers.size(); ++i )
     106              :     {
     107          168 :         if ( m_workers[i] -> setJob( job ) )
     108              :         {
     109          125 :             return;
     110              :         }
     111              :     }
     112            0 :     boost::mutex::scoped_lock lock( m_mutex );
     113            0 :     m_jobs.push( job );
     114            0 : }
     115              : 
     116              : void
     117           82 : OksPipeline::waitForCompletion()
     118              : {
     119           82 :     {
     120           82 :         boost::mutex::scoped_lock lock( m_mutex );
     121           82 :         while ( !m_jobs.empty() )
     122              :         {
     123            0 :             m_condition.wait( lock );
     124              :         }
     125           82 :     }
     126        10578 :     for ( size_t i = 0; i < m_workers.size(); ++i )
     127              :     {
     128        10496 :         m_workers[i] -> stop();
     129              :     }
     130           82 :     m_barrier.wait();
     131           82 : }
     132              : 
     133              : bool
     134          125 : OksPipeline::getJob( OksJob* & job )
     135              : {
     136          125 :     boost::mutex::scoped_lock lock( m_mutex );
     137          125 :     if ( !m_jobs.empty() )
     138              :     {
     139            0 :         job = m_jobs.front();
     140            0 :         m_jobs.pop();
     141            0 :         m_condition.notify_one( );
     142            0 :         return true;
     143              :     }
     144              :     return false;
     145          125 : }
     146              : 
     147              : } // namespace oks
     148              : } // namespace dunedaq