pipeline.cpp

Go to the documentation of this file.

// DUNE DAQ modification notice:
// This file has been modified from the original ATLAS oks source for the DUNE DAQ project.
// Fork baseline commit: oks-08-03-04 (2022-04-14).
// Renamed since fork: no.

//
//    copy from IPC
//
 
#include "oks/pipeline.hpp"
#include <boost/bind/bind.hpp>
 
namespace dunedaq {
namespace oks {
 
bool
OksPipeline::Worker::setJob(OksJob* job)
{
  boost::mutex::scoped_lock lock(m_mutex);
  if (!m_running && m_idle) {
    m_job = job;
    m_idle = false;
    m_condition.notify_one();
    return true;
  }
  return false;
}
 
void
OksPipeline::Worker::run()
{
  {
    boost::mutex::scoped_lock lock(m_mutex);
    m_pipeline.m_barrier.wait();
  }
  while (true) {
    {
      boost::mutex::scoped_lock lock(m_mutex);
      m_running = true;
    }
    while (!m_idle) {
      m_job->run();
      delete m_job;
      m_idle = !m_pipeline.getJob(m_job);
    }
    {
      boost::mutex::scoped_lock lock(m_mutex);
      m_running = false;
      if (m_shutdown)
        break;
      if (m_stop) {
        m_pipeline.m_barrier.wait();
        m_stop = false;
      }
      if (m_idle)
        m_condition.wait(lock);
    }
  }
}
 
void
OksPipeline::Worker::shutdown()
{
  boost::mutex::scoped_lock lock(m_mutex);
  m_shutdown = true;
  m_condition.notify_one();
}
 
void
OksPipeline::Worker::stop()
{
  boost::mutex::scoped_lock lock(m_mutex);
  m_stop = true;
  m_condition.notify_one();
}
 
OksPipeline::OksPipeline(size_t size)
  : m_barrier(size + 1)
{
  for (size_t i = 0; i < size; ++i) {
    m_workers.push_back(WorkerPtr(new Worker(*this)));
    m_pool.create_thread(boost::bind(&Worker::run, m_workers.back().get()));
  }
  m_barrier.wait();
}
 
OksPipeline::~OksPipeline()
{
  waitForCompletion();
  for (size_t i = 0; i < m_workers.size(); ++i) {
    m_workers[i]->shutdown();
  }
  m_pool.join_all();
}
 
void
OksPipeline::addJob(OksJob* job)
{
  for (size_t i = 0; i < m_workers.size(); ++i) {
    if (m_workers[i]->setJob(job)) {
      return;
    }
  }
  boost::mutex::scoped_lock lock(m_mutex);
  m_jobs.push(job);
}
 
void
OksPipeline::waitForCompletion()
{
  {
    boost::mutex::scoped_lock lock(m_mutex);
    while (!m_jobs.empty()) {
      m_condition.wait(lock);
    }
  }
  for (size_t i = 0; i < m_workers.size(); ++i) {
    m_workers[i]->stop();
  }
  m_barrier.wait();
}
 
bool
OksPipeline::getJob(OksJob*& job)
{
  boost::mutex::scoped_lock lock(m_mutex);
  if (!m_jobs.empty()) {
    job = m_jobs.front();
    m_jobs.pop();
    m_condition.notify_one();
    return true;
  }
  return false;
}
 
} // namespace oks
} // namespace dunedaq