SamplePipeline sorts samples that are coming from one or more inputs. More...

#include <SamplePipeline.h>

Inheritance diagram for nidas::core::SamplePipeline:

Public Member Functions
	SamplePipeline ()

virtual	~SamplePipeline ()

std::string	getName () const

SampleSource *	getRawSampleSource ()
	Several objects in NIDAS can be both a SampleSource of raw Samples and processed Samples.

SampleSource *	getProcessedSampleSource ()
	Several objects in NIDAS can be both a SampleSource of raw Samples and processed Samples.

void	connect (SampleSource *src) throw ()
	Add an input to be merged and sorted.

void	disconnect (SampleSource *src) throw ()
	Remove a SampleSource from the SamplePipeline.

void	addSampleTag (const SampleTag *tag) throw ()
	Add a SampleTag to this SampleSource.

void	removeSampleTag (const SampleTag *tag) throw ()

std::list< const SampleTag * >	getSampleTags () const
	What SampleTags am I a SampleSource for?

SampleTagIterator	getSampleTagIterator () const

void	addSampleClient (SampleClient *client) throw ()
	Add a SampleClient that wants all processed samples.

void	removeSampleClient (SampleClient *client) throw ()
	Remove a SampleClient from this SampleSource.

void	addSampleClientForTag (SampleClient , const SampleTag ) throw ()
	Add a SampleClient that wants samples which have been merged from various inputs, sorted, processed through a certain DSMSensor, and then re-sorted again.

void	removeSampleClientForTag (SampleClient , const SampleTag ) throw ()
	Remove a SampleClient for a given SampleTag from this SampleSource.

int	getClientCount () const throw ()
	How many SampleClients are currently in my list.

void	flush () throw ()
	Purge samples from the SampleSorters in this pipeline.

void	interrupt ()
	Interrupt the SampleSorters in this pipeline.

void	join () throw ()
	Join the SampleSorters in this pipeline.

void	setRealTime (bool val)

bool	getRealTime () const

const SampleStats &	getSampleStats () const

size_t	getSorterNumRawSamples () const
	Number of raw samples currently in the sorter.

size_t	getSorterNumProcSamples () const
	Number of processed samples currently in the sorter.

size_t	getSorterNumRawBytes () const
	Current size in bytes of the raw sample sorter.

size_t	getSorterNumProcBytes () const
	Current size in bytes of the processed sample sorter.

size_t	getSorterNumRawBytesMax () const
	Current size in bytes of the raw sample sorter.

size_t	getSorterNumProcBytesMax () const
	Current size in bytes of the processed sample sorter.

size_t	getNumDiscardedRawSamples () const
	Number of raw samples discarded because sorter was getting too big.

size_t	getNumDiscardedProcSamples () const
	Number of processed samples discarded because sorter was getting too big.

size_t	getNumFutureRawSamples () const
	Number of raw samples discarded because their timetags were in the future.

size_t	getNumFutureProcSamples () const
	Number of processed samples discarded because their timetags were in the future.

void	setRawSorterLength (float val)
	Set length of raw SampleSorter, in seconds.

float	getRawSorterLength () const

void	setProcSorterLength (float val)
	Set length of processed SampleSorter, in seconds.

float	getProcSorterLength () const

void	setRawHeapMax (size_t val)
	Set the maximum amount of heap memory to use for sorting samples.

size_t	getRawHeapMax () const

void	setProcHeapMax (size_t val)
	Set the maximum amount of heap memory to use for sorting samples.

size_t	getProcHeapMax () const

void	setHeapBlock (bool val)

bool	getHeapBlock () const

void	setKeepStats (bool val)

void	setRawLateSampleCacheSize (unsigned int val)
	Get the size of the late sample cache in the raw sample sorter.

unsigned int	getRawLateSampleCacheSize () const
	Cache this number of samples with potentially anomalous, late time tags in the raw sample sorter.

void	setProcLateSampleCacheSize (unsigned int val)
	Get the size of the late sample cache in the processed sample sorter.

unsigned int	getProcLateSampleCacheSize () const
	Cache this number of samples with potentially anomalous, late time tags in the processed sample sorter.

Private Member Functions
void	rawinit ()

void	procinit ()

	SamplePipeline (const SamplePipeline &x)
	No copying.

SamplePipeline &	operator= (const SamplePipeline &x)
	No assignment.

Private Attributes
std::string	_name

nidas::util::Mutex	_rawMutex

SampleThread *	_rawSorter

nidas::util::Mutex	_procMutex

SampleThread *	_procSorter

std::list< const SampleTag * >	_sampleTags

std::list< const DSMConfig * >	_dsmConfigs

bool	_realTime

float	_rawSorterLength
	seconds

float	_procSorterLength
	seconds

size_t	_rawHeapMax

size_t	_procHeapMax

bool	_heapBlock

bool	_keepStats

unsigned int	_rawLateSampleCacheSize

unsigned int	_procLateSampleCacheSize

Detailed Description

SamplePipeline sorts samples that are coming from one or more inputs.

SamplePipeline makes use of two SampleSorters, one called rawSorter and procSorter.

rawSorter is a client of one or more inputs (the text arrows show the sample flow):

input -—v input --> _rawSorter input -—^

After sorting the samples, rawSorter passes them on to the two types of SampleClients that have registered with SamplePipeline. SampleClients that have registered with SamplePipeline::addSampleClient will receive their raw samples directly from rawSorter.

rawSorter -> sampleClients

SampleClients that have registered with SamplePipeline::addProcessedSampleClient will receive their samples indirectly:

rawSorter -> sensor -> procSorter -> processedSampleClients

rawSorter provides sorting of the samples from the various inputs.

procSorter provides sorting of the processed samples. DSMSensors are apt to create processed samples with different time-tags than the input raw samples, therefore they need to be sorted again.

Multiple threads can be passing samples to the sorters. Thread exclusion is enforced when passing the samples to the SampleClient::receive() methods from either sorter, so the SampleClient::receive() methods don't have to worry about being re-entrant.

Constructor & Destructor Documentation

◆ SamplePipeline() [1/2]

SamplePipeline::SamplePipeline ( )

◆ ~SamplePipeline()

SamplePipeline::~SamplePipeline ( )

virtual

References _procMutex, _procSorter, _rawMutex, _rawSorter, nidas::util::Mutex::lock(), and nidas::util::Mutex::unlock().

◆ SamplePipeline() [2/2]

nidas::core::SamplePipeline::SamplePipeline ( const SamplePipeline & x )

private

No copying.

Member Function Documentation

◆ addSampleClient()

void SamplePipeline::addSampleClient	(	SampleClient *	client	)
throw	(	)

virtual

Add a SampleClient that wants all processed samples.

Implementation of SampleSource::addSampleClient().

Implements nidas::core::SampleSource.

References nidas::core::DSMSensor::addSampleClient(), nidas::core::SampleTag::getDSMSensor(), nidas::core::DSMSensor::getName(), nidas::core::DSMSensor::getRawSampleTag(), nidas::core::getSample(), and VLOG.

◆ addSampleClientForTag()

void SamplePipeline::addSampleClientForTag	(	SampleClient *	client,
		const SampleTag *	stag )
throw	(	)

virtual

Add a SampleClient that wants samples which have been merged from various inputs, sorted, processed through a certain DSMSensor, and then re-sorted again.

Implementation of SampleSource::addSampleClient().

Implements nidas::core::SampleSource.

References nidas::core::DSMSensor::addSampleClient(), nidas::core::DSMSensor::addSampleClientForTag(), and nidas::core::DSMSensor::getRawSampleTag().

◆ addSampleTag()

void nidas::core::SamplePipeline::addSampleTag	(	const SampleTag *		)
throw	(	)

inlinevirtual

Add a SampleTag to this SampleSource.

This SampleSource does not own the SampleTag.

Exceptions

nidas::util::InvalidParameterException

Implements nidas::core::SampleSource.

References _procSorter, and nidas::core::SampleSource::addSampleTag().

◆ connect()

void SamplePipeline::connect	(	SampleSource *	src	)
throw	(	)

Add an input to be merged and sorted.

SamplePipeline does not own the SampleSource.

References GET_DSM_ID, GET_SPS_ID, nidas::core::SampleTag::getDSMSensor(), nidas::core::SampleTag::getId(), nidas::core::DSMSensor::getName(), nidas::core::SampleSource::getRawSampleSource(), nidas::core::getSample(), nidas::core::SampleTag::getSampleId(), nidas::core::DSMSensor::getSampleTagIterator(), and LOG_VERBOSE.

Referenced by nidas::dynld::raf::SyncServer::init(), and nidas::core::DSMEngine::openSensors().

◆ disconnect()

void SamplePipeline::disconnect	(	SampleSource *	src	)
throw	(	)

Remove a SampleSource from the SamplePipeline.

References nidas::core::SampleTag::getDSMSensor(), nidas::core::SampleSource::getRawSampleSource(), nidas::core::getSample(), and nidas::core::DSMSensor::getSampleTagIterator().

Referenced by nidas::dynld::raf::SyncServer::stop().

◆ flush()

void nidas::core::SamplePipeline::flush	(		)
throw	(	)

inlinevirtual

Purge samples from the SampleSorters in this pipeline.

This call will block, until both sorters are empty.

Implements nidas::core::SampleSource.

References _procSorter, _rawSorter, and nidas::core::SampleThread::flush().

Referenced by nidas::core::DSMEngine::interrupt(), nidas::dynld::RawSampleService::interrupt(), and nidas::dynld::raf::SyncServer::read().

◆ getClientCount()

int nidas::core::SamplePipeline::getClientCount	(		)	const
throw	(	)

inlinevirtual

How many SampleClients are currently in my list.

Implements nidas::core::SampleSource.

References _procSorter, and nidas::core::SampleSource::getClientCount().

◆ getHeapBlock()

bool nidas::core::SamplePipeline::getHeapBlock ( ) const

inline

References _heapBlock.

Referenced by procinit(), and rawinit().

◆ getName()

std::string nidas::core::SamplePipeline::getName ( ) const

inline

References _name.

Referenced by nidas::dynld::RawSampleService::schedule().

◆ getNumDiscardedProcSamples()

size_t nidas::core::SamplePipeline::getNumDiscardedProcSamples ( ) const

inline

Number of processed samples discarded because sorter was getting too big.

References _procSorter, and nidas::core::SampleThread::getNumDiscardedSamples().

Referenced by nidas::dynld::RawSampleService::printStatus().

◆ getNumDiscardedRawSamples()

size_t nidas::core::SamplePipeline::getNumDiscardedRawSamples ( ) const

inline

Number of raw samples discarded because sorter was getting too big.

References _rawSorter, and nidas::core::SampleThread::getNumDiscardedSamples().

Referenced by nidas::dynld::RawSampleService::printStatus().

◆ getNumFutureProcSamples()

size_t nidas::core::SamplePipeline::getNumFutureProcSamples ( ) const

inline

Number of processed samples discarded because their timetags were in the future.

References _procSorter, and nidas::core::SampleThread::getNumFutureSamples().

Referenced by nidas::dynld::RawSampleService::printStatus().

◆ getNumFutureRawSamples()

size_t nidas::core::SamplePipeline::getNumFutureRawSamples ( ) const

inline

Number of raw samples discarded because their timetags were in the future.

References _rawSorter, and nidas::core::SampleThread::getNumFutureSamples().

Referenced by nidas::dynld::RawSampleService::printStatus().

◆ getProcessedSampleSource()

SampleSource * nidas::core::SamplePipeline::getProcessedSampleSource ( )

inlinevirtual

Several objects in NIDAS can be both a SampleSource of raw Samples and processed Samples.

SampleClients use this method to get a pointer to whatever sample source they are interested in. Derived classes can return NULL if they are not a SampleSource of processed samples.

Implements nidas::core::SampleSource.

References procinit().

Referenced by nidas::core::DSMEngine::closeOutputs(), nidas::core::DSMEngine::connectOutputs(), nidas::dynld::raf::SyncServer::init(), nidas::dynld::RawSampleService::printStatus(), nidas::dynld::RawSampleService::schedule(), and nidas::dynld::raf::SyncServer::stop().

◆ getProcHeapMax()

size_t nidas::core::SamplePipeline::getProcHeapMax ( ) const

inline

References _procHeapMax.

Referenced by procinit().

◆ getProcLateSampleCacheSize()

unsigned int nidas::core::SamplePipeline::getProcLateSampleCacheSize ( ) const

inline

Cache this number of samples with potentially anomalous, late time tags in the processed sample sorter.

See SampleSorter::setLateSampleCacheSize(val).

References _procLateSampleCacheSize.

Referenced by procinit().

◆ getProcSorterLength()

float nidas::core::SamplePipeline::getProcSorterLength ( ) const

inline

References _procSorterLength.

Referenced by procinit().

◆ getRawHeapMax()

size_t nidas::core::SamplePipeline::getRawHeapMax ( ) const

inline

References _rawHeapMax.

Referenced by rawinit().

◆ getRawLateSampleCacheSize()

unsigned int nidas::core::SamplePipeline::getRawLateSampleCacheSize ( ) const

inline

Cache this number of samples with potentially anomalous, late time tags in the raw sample sorter.

See SampleSorter::setLateSampleCacheSize(val).

References _rawLateSampleCacheSize.

Referenced by rawinit().

◆ getRawSampleSource()

SampleSource * nidas::core::SamplePipeline::getRawSampleSource ( )

inlinevirtual

Several objects in NIDAS can be both a SampleSource of raw Samples and processed Samples.

SampleClients use this method to get a pointer to whatever sample source they are interested in. Derived classes can return NULL if they are not a SampleSource of raw samples.

Implements nidas::core::SampleSource.

References _rawSorter, and rawinit().

Referenced by nidas::core::DSMEngine::closeOutputs(), nidas::core::DSMEngine::connectOutputs(), nidas::dynld::RawSampleService::printClock(), nidas::dynld::RawSampleService::printStatus(), and nidas::dynld::RawSampleService::schedule().

◆ getRawSorterLength()

float nidas::core::SamplePipeline::getRawSorterLength ( ) const

inline

References _rawSorterLength.

Referenced by rawinit().

◆ getRealTime()

bool nidas::core::SamplePipeline::getRealTime ( ) const

inline

References _realTime.

Referenced by procinit(), and rawinit().

◆ getSampleStats()

const SampleStats & nidas::core::SamplePipeline::getSampleStats ( ) const

inlinevirtual

Implements nidas::core::SampleSource.

References _procSorter, and nidas::core::SampleSource::getSampleStats().

◆ getSampleTagIterator()

SampleTagIterator nidas::core::SamplePipeline::getSampleTagIterator ( ) const

inlinevirtual

Implements nidas::core::SampleSource.

References _procSorter, and nidas::core::SampleSource::getSampleTagIterator().

◆ getSampleTags()

std::list< const SampleTag * > nidas::core::SamplePipeline::getSampleTags ( ) const

inlinevirtual

What SampleTags am I a SampleSource for?

Implements nidas::core::SampleSource.

References _procSorter, and nidas::core::SampleSource::getSampleTags().

◆ getSorterNumProcBytes()

size_t nidas::core::SamplePipeline::getSorterNumProcBytes ( ) const

inline

Current size in bytes of the processed sample sorter.

References _procSorter, and nidas::core::SampleThread::getHeapSize().

Referenced by nidas::dynld::RawSampleService::printStatus().

◆ getSorterNumProcBytesMax()

size_t nidas::core::SamplePipeline::getSorterNumProcBytesMax ( ) const

inline

Current size in bytes of the processed sample sorter.

References _procSorter, and nidas::core::SampleThread::getHeapMax().

Referenced by nidas::dynld::RawSampleService::printStatus().

◆ getSorterNumProcSamples()

size_t nidas::core::SamplePipeline::getSorterNumProcSamples ( ) const

inline

Number of processed samples currently in the sorter.

References _procSorter, and nidas::core::SampleThread::size().

Referenced by nidas::dynld::RawSampleService::printStatus().

◆ getSorterNumRawBytes()

size_t nidas::core::SamplePipeline::getSorterNumRawBytes ( ) const

inline

Current size in bytes of the raw sample sorter.

References _rawSorter, and nidas::core::SampleThread::getHeapSize().

Referenced by nidas::dynld::RawSampleService::printStatus().

◆ getSorterNumRawBytesMax()

size_t nidas::core::SamplePipeline::getSorterNumRawBytesMax ( ) const

inline

Current size in bytes of the raw sample sorter.

References _rawSorter, and nidas::core::SampleThread::getHeapMax().

Referenced by nidas::dynld::RawSampleService::printStatus().

◆ getSorterNumRawSamples()

size_t nidas::core::SamplePipeline::getSorterNumRawSamples ( ) const

inline

Number of raw samples currently in the sorter.

References _rawSorter, and nidas::core::SampleThread::size().

Referenced by nidas::dynld::RawSampleService::printStatus().

◆ interrupt()

void SamplePipeline::interrupt ( )

Interrupt the SampleSorters in this pipeline.

References _procMutex, _procSorter, _rawMutex, _rawSorter, nidas::util::Thread::interrupt(), nidas::util::Mutex::lock(), and nidas::util::Mutex::unlock().

Referenced by nidas::core::DSMEngine::interrupt(), nidas::dynld::raf::SyncServer::interrupt(), nidas::dynld::RawSampleService::interrupt(), and nidas::dynld::raf::SyncServer::stop().

◆ join()

void SamplePipeline::join	(		)
throw	(	)

Join the SampleSorters in this pipeline.

References _procMutex, _procSorter, _rawMutex, _rawSorter, nidas::util::Thread::getName(), nidas::util::Thread::interrupt(), nidas::util::Thread::isRunning(), nidas::util::Thread::join(), nidas::util::Mutex::lock(), nidas::util::Mutex::unlock(), and WLOG.

Referenced by nidas::core::DSMEngine::joinDataThreads(), nidas::dynld::raf::SyncServer::stop(), and nidas::dynld::RawSampleService::~RawSampleService().

◆ operator=()

SamplePipeline & nidas::core::SamplePipeline::operator= ( const SamplePipeline & x )

private

No assignment.

◆ procinit()

void SamplePipeline::procinit ( )

private

◆ rawinit()

void SamplePipeline::rawinit ( )

private

◆ removeSampleClient()

void SamplePipeline::removeSampleClient	(	SampleClient *	c	)
throw	(	)

virtual

Remove a SampleClient from this SampleSource.

Exceptions

)

Implements nidas::core::SampleSource.

References nidas::core::SampleTag::getDSMSensor(), nidas::core::DSMSensor::getRawSampleTag(), nidas::core::getSample(), and nidas::core::DSMSensor::removeSampleClient().

◆ removeSampleClientForTag()

void SamplePipeline::removeSampleClientForTag	(	SampleClient *	c,
		const SampleTag *	)
throw	(	)

virtual

Remove a SampleClient for a given SampleTag from this SampleSource.

The pointer to the SampleClient must remain valid, until after it is removed.

Exceptions

)

Implements nidas::core::SampleSource.

References nidas::core::DSMSensor::getRawSampleTag(), nidas::core::getSample(), nidas::core::DSMSensor::removeSampleClient(), and nidas::core::DSMSensor::removeSampleClientForTag().

◆ removeSampleTag()

void nidas::core::SamplePipeline::removeSampleTag	(	const SampleTag *		)
throw	(	)

inlinevirtual

Exceptions

)

Implements nidas::core::SampleSource.

References _procSorter, and nidas::core::SampleSource::removeSampleTag().

◆ setHeapBlock()

void nidas::core::SamplePipeline::setHeapBlock ( bool val )

inline

Parameters

val	If true, and heapSize exceeds heapMax, then wait for heapSize to be less then heapMax, which will block any SampleSources that are inserting samples into the raw buffer. If false, then discard any samples that are received while heapSize exceeds heapMax.