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          176 : OksPipeline::Worker::setJob(OksJob* job)
      20              : {
      21          176 :   boost::mutex::scoped_lock lock(m_mutex);
      22          176 :   if (!m_running && m_idle) {
      23          133 :     m_job = job;
      24          133 :     m_idle = false;
      25          133 :     m_condition.notify_one();
      26          133 :     return true;
      27              :   }
      28              :   return false;
      29          176 : }
      30              : 
      31              : void
      32        11520 : OksPipeline::Worker::run()
      33              : {
      34        11520 :   {
      35        11520 :     boost::mutex::scoped_lock lock(m_mutex);
      36        11520 :     m_pipeline.m_barrier.wait();
      37        11499 :   }
      38        22642 :   while (true) {
      39        34140 :     {
      40        34140 :       boost::mutex::scoped_lock lock(m_mutex);
      41        34063 :       m_running = true;
      42        34063 :     }
      43        34406 :     while (!m_idle) {
      44          189 :       m_job->run();
      45          133 :       delete m_job;
      46          133 :       m_idle = !m_pipeline.getJob(m_job);
      47              :     }
      48        34217 :     {
      49        34217 :       boost::mutex::scoped_lock lock(m_mutex);
      50        34079 :       m_running = false;
      51        34079 :       if (m_shutdown)
      52              :         break;
      53        22583 :       if (m_stop) {
      54        11135 :         m_pipeline.m_barrier.wait();
      55        11505 :         m_stop = false;
      56              :       }
      57        22953 :       if (m_idle)
      58        22971 :         m_condition.wait(lock);
      59        34288 :     }
      60              :   }
      61        11496 : }
      62              : 
      63              : void
      64        11520 : OksPipeline::Worker::shutdown()
      65              : {
      66        11520 :   boost::mutex::scoped_lock lock(m_mutex);
      67        11520 :   m_shutdown = true;
      68        11520 :   m_condition.notify_one();
      69        11520 : }
      70              : 
      71              : void
      72        11520 : OksPipeline::Worker::stop()
      73              : {
      74        11520 :   boost::mutex::scoped_lock lock(m_mutex);
      75        11520 :   m_stop = true;
      76        11520 :   m_condition.notify_one();
      77        11520 : }
      78              : 
      79           90 : OksPipeline::OksPipeline(size_t size)
      80           90 :   : m_barrier(size + 1)
      81              : {
      82        11610 :   for (size_t i = 0; i < size; ++i) {
      83        11520 :     m_workers.push_back(WorkerPtr(new Worker(*this)));
      84        11520 :     m_pool.create_thread(boost::bind(&Worker::run, m_workers.back().get()));
      85              :   }
      86           90 :   m_barrier.wait();
      87           90 : }
      88              : 
      89           90 : OksPipeline::~OksPipeline()
      90              : {
      91           90 :   waitForCompletion();
      92        11610 :   for (size_t i = 0; i < m_workers.size(); ++i) {
      93        11520 :     m_workers[i]->shutdown();
      94              :   }
      95           90 :   m_pool.join_all();
      96           90 : }
      97              : 
      98              : void
      99          133 : OksPipeline::addJob(OksJob* job)
     100              : {
     101          176 :   for (size_t i = 0; i < m_workers.size(); ++i) {
     102          176 :     if (m_workers[i]->setJob(job)) {
     103          133 :       return;
     104              :     }
     105              :   }
     106            0 :   boost::mutex::scoped_lock lock(m_mutex);
     107            0 :   m_jobs.push(job);
     108            0 : }
     109              : 
     110              : void
     111           90 : OksPipeline::waitForCompletion()
     112              : {
     113           90 :   {
     114           90 :     boost::mutex::scoped_lock lock(m_mutex);
     115           90 :     while (!m_jobs.empty()) {
     116            0 :       m_condition.wait(lock);
     117              :     }
     118           90 :   }
     119        11610 :   for (size_t i = 0; i < m_workers.size(); ++i) {
     120        11520 :     m_workers[i]->stop();
     121              :   }
     122           90 :   m_barrier.wait();
     123           90 : }
     124              : 
     125              : bool
     126          133 : OksPipeline::getJob(OksJob*& job)
     127              : {
     128          133 :   boost::mutex::scoped_lock lock(m_mutex);
     129          133 :   if (!m_jobs.empty()) {
     130            0 :     job = m_jobs.front();
     131            0 :     m_jobs.pop();
     132            0 :     m_condition.notify_one();
     133            0 :     return true;
     134              :   }
     135              :   return false;
     136          133 : }
     137              : 
     138              : } // namespace oks
     139              : } // namespace dunedaq