dunedaq::oks::OksPipeline Class Reference

#include <pipeline.hpp>

Classes
struct	Worker

Public Member Functions
	OksPipeline (size_t size)
	~OksPipeline ()
void	waitForCompletion ()
void	addJob (OksJob *job)

Private Types
typedef std::shared_ptr< Worker >	WorkerPtr

Private Member Functions
bool	getJob (OksJob *&job)

Private Attributes
boost::mutex	m_mutex
boost::condition	m_condition
boost::barrier	m_barrier
boost::thread_group	m_pool
std::vector< WorkerPtr >	m_workers
std::queue< OksJob * >	m_jobs

Friends
struct	Worker

Detailed Description

Definition at line 31 of file pipeline.hpp.

Member Typedef Documentation

WorkerPtr

typedef std::shared_ptr<Worker> dunedaq::oks::OksPipeline::WorkerPtr

private

Definition at line 74 of file pipeline.hpp.

Constructor & Destructor Documentation

OksPipeline()

dunedaq::oks::OksPipeline::OksPipeline

(

size_t

size

)

Definition at line 79 of file pipeline.cpp.

  : 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()

dunedaq::oks::OksPipeline::~OksPipeline

(

)

Definition at line 89 of file pipeline.cpp.

{
  waitForCompletion();
  for (size_t i = 0; i < m_workers.size(); ++i) {
    m_workers[i]->shutdown();
  }
  m_pool.join_all();
}

Member Function Documentation

addJob()

void dunedaq::oks::OksPipeline::addJob

(

OksJob *

job

)

Definition at line 99 of file pipeline.cpp.

{
  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);
}

getJob()

bool dunedaq::oks::OksPipeline::getJob ( OksJob *& job )

private

Definition at line 126 of file pipeline.cpp.

{
  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;
}

waitForCompletion()

void dunedaq::oks::OksPipeline::waitForCompletion

(

)

Definition at line 111 of file pipeline.cpp.

{
  {
    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();
}

Friends And Related Symbol Documentation

Worker

friend struct Worker

friend

Definition at line 77 of file pipeline.hpp.

Member Data Documentation

m_barrier

boost::barrier dunedaq::oks::OksPipeline::m_barrier

private

Definition at line 84 of file pipeline.hpp.

m_condition

boost::condition dunedaq::oks::OksPipeline::m_condition

private

Definition at line 83 of file pipeline.hpp.

m_jobs

std::queue<OksJob*> dunedaq::oks::OksPipeline::m_jobs

private

Definition at line 87 of file pipeline.hpp.

m_mutex

boost::mutex dunedaq::oks::OksPipeline::m_mutex

private

Definition at line 82 of file pipeline.hpp.

m_pool

boost::thread_group dunedaq::oks::OksPipeline::m_pool

private

Definition at line 85 of file pipeline.hpp.

m_workers

std::vector<WorkerPtr> dunedaq::oks::OksPipeline::m_workers

private

Definition at line 86 of file pipeline.hpp.

---

The documentation for this class was generated from the following files:

• /github/workspace/dunedaq/sourcecode/oks/include/oks/pipeline.hpp
• /github/workspace/dunedaq/sourcecode/oks/src/pipeline.cpp