SourceXtractor::MultithreadedMeasurement Class Reference

#include <MultithreadedMeasurement.h>

Inheritance diagram for SourceXtractor::MultithreadedMeasurement:

Collaboration diagram for SourceXtractor::MultithreadedMeasurement:

Public Types
using	SourceToRowConverter = std::function< Euclid::Table::Row(const SourceInterface &)>

Public Member Functions
	MultithreadedMeasurement (SourceToRowConverter source_to_row, const std::shared_ptr< Euclid::ThreadPool > &thread_pool, unsigned max_queue_size)
	~MultithreadedMeasurement () override
void	receiveSource (std::unique_ptr< SourceGroupInterface > source_group) override
void	receiveProcessSignal (const ProcessSourcesEvent &event) override
void	startThreads () override
void	stopThreads () override
void	synchronizeThreads () override
Public Member Functions inherited from SourceXtractor::Measurement
	~Measurement () override=default
Public Member Functions inherited from SourceXtractor::PipelineReceiver< SourceGroupInterface >
virtual	~PipelineReceiver ()=default
Public Member Functions inherited from SourceXtractor::PipelineEmitter< SourceGroupInterface >
	~PipelineEmitter () override=default
void	setNextStage (std::shared_ptr< PipelineReceiver< SourceGroupInterface >> next)
Public Member Functions inherited from SourceXtractor::Observable< SourceGroupInterface >
virtual	~Observable ()=default
	Destructor. More...
virtual void	addObserver (std::shared_ptr< Observer< SourceGroupInterface >> observer)
	Adds an Observer that will be notified when notify Observers is called. More...
virtual void	removeObserver (std::shared_ptr< Observer< SourceGroupInterface >> observer)
	Removes a previously added Observer from the list of Observers to notify. More...

Private Member Functions
void	outputThreadLoop ()

Static Private Member Functions
static void	outputThreadStatic (MultithreadedMeasurement *measurement)

Private Attributes
SourceToRowConverter	m_source_to_row
std::shared_ptr < Euclid::ThreadPool >	m_thread_pool
std::unique_ptr< std::thread >	m_output_thread
int	m_group_counter
std::atomic_bool	m_input_done
std::atomic_bool	m_abort_raised
std::condition_variable	m_new_output
std::list< std::pair< int, std::unique_ptr < SourceGroupInterface > > >	m_output_queue
std::mutex	m_output_queue_mutex
Euclid::Semaphore	m_semaphore

Additional Inherited Members
Protected Member Functions inherited from SourceXtractor::PipelineEmitter< SourceGroupInterface >
void	sendSource (std::unique_ptr< SourceGroupInterface > source) const
void	sendProcessSignal (const ProcessSourcesEvent &event) const
Protected Member Functions inherited from SourceXtractor::Observable< SourceGroupInterface >
void	notifyObservers (const SourceGroupInterface &message) const

Detailed Description

Definition at line 38 of file MultithreadedMeasurement.h.

Member Typedef Documentation

using SourceXtractor::MultithreadedMeasurement::SourceToRowConverter = std::function<Euclid::Table::Row(const SourceInterface&)>

Definition at line 41 of file MultithreadedMeasurement.h.

Constructor & Destructor Documentation

SourceXtractor::MultithreadedMeasurement::MultithreadedMeasurement ( SourceToRowConverter  source_to_row, const std::shared_ptr< Euclid::ThreadPool > &  thread_pool, unsigned  max_queue_size  )

inline

Definition at line 42 of file MultithreadedMeasurement.h.

MultithreadedMeasurement::~MultithreadedMeasurement ( )

override

Definition at line 36 of file MultithreadedMeasurement.cpp.

References std::thread::join(), std::thread::joinable(), and m_output_thread.

Here is the call graph for this function:

Member Function Documentation

void MultithreadedMeasurement::outputThreadLoop ( )

private

Definition at line 119 of file MultithreadedMeasurement.cpp.

References m_input_done, m_new_output, m_output_queue, m_output_queue_mutex, m_thread_pool, std::move(), SourceXtractor::PipelineEmitter< SourceGroupInterface >::sendSource(), and std::condition_variable::wait_for().

Referenced by outputThreadStatic().

Here is the call graph for this function:

void MultithreadedMeasurement::outputThreadStatic ( MultithreadedMeasurement *  measurement )

staticprivate

Definition at line 103 of file MultithreadedMeasurement.cpp.

References e, SourceXtractor::logger, m_abort_raised, outputThreadLoop(), and Elements::Exception::what().

Referenced by startThreads().

Here is the call graph for this function:

void MultithreadedMeasurement::receiveProcessSignal ( const ProcessSourcesEvent &  event )

overridevirtual

Receive a signal to process sources when grouping

Parameters

event

Implements SourceXtractor::PipelineReceiver< SourceGroupInterface >.

Definition at line 141 of file MultithreadedMeasurement.cpp.

References SourceXtractor::PipelineEmitter< SourceGroupInterface >::sendProcessSignal().

Here is the call graph for this function:

void MultithreadedMeasurement::receiveSource ( std::unique_ptr< SourceGroupInterface >  source )

overridevirtual

Receive a source from the previous stage. The receiver owns the object.

Parameters

source

Implements SourceXtractor::PipelineReceiver< SourceGroupInterface >.

Definition at line 76 of file MultithreadedMeasurement.cpp.

References m_group_counter, m_new_output, m_output_queue, m_output_queue_mutex, m_source_to_row, m_thread_pool, std::move(), and std::condition_variable::notify_one().

Here is the call graph for this function:

void MultithreadedMeasurement::startThreads ( )

overridevirtual

Implements SourceXtractor::Measurement.

Definition at line 42 of file MultithreadedMeasurement.cpp.

References m_output_thread, and outputThreadStatic().

Here is the call graph for this function:

void MultithreadedMeasurement::stopThreads ( )

overridevirtual

Implements SourceXtractor::Measurement.

Definition at line 46 of file MultithreadedMeasurement.cpp.

References std::thread::join(), SourceXtractor::logger, m_input_done, m_output_thread, and m_thread_pool.

Here is the call graph for this function:

void MultithreadedMeasurement::synchronizeThreads ( )

overridevirtual

Implements SourceXtractor::Measurement.

Definition at line 53 of file MultithreadedMeasurement.cpp.

References SourceXtractor::logger, m_output_queue, m_output_queue_mutex, m_thread_pool, and std::this_thread::sleep_for().

Here is the call graph for this function:

Member Data Documentation

std::atomic_bool SourceXtractor::MultithreadedMeasurement::m_abort_raised

private

Definition at line 67 of file MultithreadedMeasurement.h.

Referenced by outputThreadStatic().

int SourceXtractor::MultithreadedMeasurement::m_group_counter

private

Definition at line 66 of file MultithreadedMeasurement.h.

Referenced by receiveSource().

std::atomic_bool SourceXtractor::MultithreadedMeasurement::m_input_done

private

Definition at line 67 of file MultithreadedMeasurement.h.

Referenced by outputThreadLoop(), and stopThreads().

std::condition_variable SourceXtractor::MultithreadedMeasurement::m_new_output

private

Definition at line 69 of file MultithreadedMeasurement.h.

Referenced by outputThreadLoop(), and receiveSource().

std::list<std::pair<int, std::unique_ptr<SourceGroupInterface> > > SourceXtractor::MultithreadedMeasurement::m_output_queue

private

Definition at line 70 of file MultithreadedMeasurement.h.

Referenced by outputThreadLoop(), receiveSource(), and synchronizeThreads().

std::mutex SourceXtractor::MultithreadedMeasurement::m_output_queue_mutex

private

Definition at line 71 of file MultithreadedMeasurement.h.

Referenced by outputThreadLoop(), receiveSource(), and synchronizeThreads().

std::unique_ptr<std::thread> SourceXtractor::MultithreadedMeasurement::m_output_thread

private

Definition at line 64 of file MultithreadedMeasurement.h.

Referenced by startThreads(), stopThreads(), and ~MultithreadedMeasurement().

Euclid::Semaphore SourceXtractor::MultithreadedMeasurement::m_semaphore

private

Definition at line 72 of file MultithreadedMeasurement.h.

SourceToRowConverter SourceXtractor::MultithreadedMeasurement::m_source_to_row

private

Definition at line 62 of file MultithreadedMeasurement.h.

Referenced by receiveSource().

std::shared_ptr<Euclid::ThreadPool> SourceXtractor::MultithreadedMeasurement::m_thread_pool

private

Definition at line 63 of file MultithreadedMeasurement.h.

Referenced by outputThreadLoop(), receiveSource(), stopThreads(), and synchronizeThreads().

---

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

• SEImplementation/SEImplementation/Measurement/MultithreadedMeasurement.h
• SEImplementation/src/lib/Measurement/MultithreadedMeasurement.cpp