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 33 of file pipeline.hpp.

Member Typedef Documentation

WorkerPtr

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

private

Definition at line 75 of file pipeline.hpp.

Constructor & Destructor Documentation

OksPipeline()

dunedaq::oks::OksPipeline::OksPipeline

(

size_t

size

)

Definition at line 81 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 92 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 103 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 134 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 117 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 78 of file pipeline.hpp.

Member Data Documentation

m_barrier

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

private

Definition at line 85 of file pipeline.hpp.

m_condition

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

private

Definition at line 84 of file pipeline.hpp.

m_jobs

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

private

Definition at line 88 of file pipeline.hpp.

m_mutex

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

private

Definition at line 83 of file pipeline.hpp.

m_pool

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

private

Definition at line 86 of file pipeline.hpp.

m_workers

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

private

Definition at line 87 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