#include <bulkDataNTArrayThread.h>

Inheritance diagram for AcsBulkdata::BulkDataNTArrayThread:

Collaboration diagram for AcsBulkdata::BulkDataNTArrayThread:

Public Member Functions
	BulkDataNTArrayThread (const ACE_CString &name, const std::string &streamName, const std::string &sendFlowName, const std::string &qosLib, const double &throttling, const double &sendTimeout, const double &ACKTimeout)
	~BulkDataNTArrayThread ()
void	run ()
ACE_CString	getName () const
bool	addDataEvent (const uint8_t *buffer, const size_t size)
void	startSequence ()
void	stopSequence ()
Private Types
enum	SequenceEndStatus { SequenceEndStatus_OK, SequenceEndStatus_EXCEPTION, SequenceEndStatus_TIMEOUT, SequenceEndStatus_STOPPED }
Private Member Functions
	BulkDataNTArrayThread (const BulkDataNTArrayThread &toCopy)
BulkDataNTArrayThread &	operator= (const BulkDataNTArrayThread &)
virtual void	exit ()
SequenceEndStatus	handleSequenceLoop ()
void	waitForDataEvent (const ACS::TimeInterval to, std::list< std::pair< uint8_t *, size_t > > &out)
void	deleteFrontEndBuffer (const ACS::TimeInterval _to, std::list< std::pair< uint8_t *, size_t > > &out)
void	relyDataToStreamer (std::list< std::pair< uint8_t *, size_t > > &data)
void	abort (const std::string &reason)
Private Attributes
ACE_CString	name
std::string	m_streamName
std::string	m_sendFlowName
std::string	m_qosLib
StreamerThread *	m_Streamer
std::list< std::pair< uint8_t *, size_t > >	m_frontEndBuffer
pthread_mutex_t	m_eventListMutex
pthread_mutex_t	m_accessMutex
pthread_cond_t	m_condition
pthread_cond_t	m_newDataCondition
bool	m_addDataEventLogFlag
pthread_cond_t	m_stopCondition
bool	m_sequenceStopFlag
bool	sequenceAlreadyRunningFlag
double	m_throttling
double	m_sendTimeout
double	m_ACKTimeout
Static Private Attributes
static const ACS::TimeInterval	m_accessTimeout = 50000000LLU

Detailed Description

This class handles incoming data from cdp nodes and its final packeting into an SDM document to be sent to the Archive. It also takes care of transmitting related data to the data-capture component associated with the on going sub-scan. This class spawns a thread that takes care of the on-the-fly actions for handling the data and each instance of this class is associate to an specific sub-array.

Member Enumeration Documentation

enum AcsBulkdata::BulkDataNTArrayThread::SequenceEndStatus [private]

End of sequence status as returned by the loop handler method.

Enumerator:

SequenceEndStatus_OK
SequenceEndStatus_EXCEPTION
SequenceEndStatus_TIMEOUT
SequenceEndStatus_STOPPED

Constructor & Destructor Documentation

AcsBulkdata::BulkDataNTArrayThread::BulkDataNTArrayThread	(	const ACE_CString &	name,
		const std::string &	streamName,
		const std::string &	sendFlowName,
		const std::string &	qosLib,
		const double &	throttling,
		const double &	sendTimeout,
		const double &	ACKTimeout
	)

constructor. TODO: Add ACK Timeout send timeout, throttling

AcsBulkdata::BulkDataNTArrayThread::~BulkDataNTArrayThread ( )

destructor.

AcsBulkdata::BulkDataNTArrayThread::BulkDataNTArrayThread ( const BulkDataNTArrayThread & toCopy ) [private]

Member Function Documentation

void AcsBulkdata::BulkDataNTArrayThread::abort ( const std::string & reason ) [private]

Abort sequence by aborting streaming

bool AcsBulkdata::BulkDataNTArrayThread::addDataEvent	(	const uint8_t *	buffer,
		const size_t	size
	)

add a data event to the thread handler's queue for deferred processing.

Parameters:

data

event data structure received from the data source (normally a cdp node)

Returns:: true on successful insertion in the queue.

void AcsBulkdata::BulkDataNTArrayThread::deleteFrontEndBuffer	(	const ACS::TimeInterval	_to,
		std::list< std::pair< uint8_t *, size_t > > &	out
	)			`[private]`

Remove available items from front-end-buffer and return their sizes and memory addresses. The caller now owns those memory addresses.

virtual void AcsBulkdata::BulkDataNTArrayThread::exit ( ) [inline, private, virtual]

signals the thread to exit it's main running loop. This is a re-implementation of the base method. We need to do this as to broadcast the condition variable and forcing that way the thread to recheck its running status.

Reimplemented from ACS::ThreadBase.

References m_condition.

ACE_CString AcsBulkdata::BulkDataNTArrayThread::getName ( void ) const [inline]

Get name of the thread.

Returns:: name of the thread

Reimplemented from ACS::ThreadBase.

SequenceEndStatus AcsBulkdata::BulkDataNTArrayThread::handleSequenceLoop ( ) [private]

Method that implements a loop to receive data for every sub-scan in a sequence until all those sub-scans have completed.

BulkDataNTArrayThread& AcsBulkdata::BulkDataNTArrayThread::operator= ( const BulkDataNTArrayThread & ) [private]

void AcsBulkdata::BulkDataNTArrayThread::relyDataToStreamer ( std::list< std::pair< uint8_t *, size_t > > & data ) [private]

void AcsBulkdata::BulkDataNTArrayThread::run ( ) [virtual]

thread's work function which handles data produced by sub-scans.

Reimplemented from ACS::Thread.

void AcsBulkdata::BulkDataNTArrayThread::startSequence ( )

start a sub-scan sequence.

Exceptions:

BDNTEx

void AcsBulkdata::BulkDataNTArrayThread::stopSequence ( )

stop a sub-scan sequence.

Exceptions:

BDNTEx

void AcsBulkdata::BulkDataNTArrayThread::waitForDataEvent	(	const ACS::TimeInterval	to,
		std::list< std::pair< uint8_t *, size_t > > &	out
	)			`[private]`

Wait for new data in the front end buffer. Before blocking on the event list semaphore the access mutex is unlocked and after getting the semaphore or timing out on the semaphore the mutex is reacquired. If new data is signaled during the alloted timeout then the function return true.

Member Data Documentation

pthread_mutex_t AcsBulkdata::BulkDataNTArrayThread::m_accessMutex [private]

Mutex used for synchronizing the configuration, startup and stopping of sequences. It must be a recursive mutex, for example, clean calls stop.

const ACS::TimeInterval AcsBulkdata::BulkDataNTArrayThread::m_accessTimeout = 50000000LLU [static, private]

Timeout for guarded access operations.

double AcsBulkdata::BulkDataNTArrayThread::m_ACKTimeout [private]

bool AcsBulkdata::BulkDataNTArrayThread::m_addDataEventLogFlag [private]

In case of error while ingesting new data to the event's help (within the AV callback handler) report the error as a log, but do that only once for one given sub-scan (avoid logging cascade.)

pthread_cond_t AcsBulkdata::BulkDataNTArrayThread::m_condition [private]

condition variable object for running a sequence. The thread blocks until requested to terminate or start a sequence, that is, the condition is signaled by either a command to terminate the thread or by a request to start a sequence.

Referenced by exit().

pthread_mutex_t AcsBulkdata::BulkDataNTArrayThread::m_eventListMutex [private]

Guard event lists with this mutex.

std::list< std::pair<uint8_t *, size_t> > AcsBulkdata::BulkDataNTArrayThread::m_frontEndBuffer [private]

list for data events from nodes.

pthread_cond_t AcsBulkdata::BulkDataNTArrayThread::m_newDataCondition [private]

If an error has occurred while handling data from nodes then this variable will set to error. Condition variable use to signal array thread that new data has been delivered into the front-end-buffer.

std::string AcsBulkdata::BulkDataNTArrayThread::m_qosLib [private]

std::string AcsBulkdata::BulkDataNTArrayThread::m_sendFlowName [private]

double AcsBulkdata::BulkDataNTArrayThread::m_sendTimeout [private]

bool AcsBulkdata::BulkDataNTArrayThread::m_sequenceStopFlag [private]

flag used to synchronize the stopping of a sequence. The stopSequence method sets the stop flag and then it waits for the thread to acknowledge by signaling this condition variable and checking that the flag was actually reset by the thread.

pthread_cond_t AcsBulkdata::BulkDataNTArrayThread::m_stopCondition [private]

condition variable object for stopping a sequence. The stop sequence method waits on this variable for feedback from thread after it has actually stopped. The method itself raises the stop flag that's check by the thread. The associated mutex is the one declared above as m_accessMutex. Note: seems to be convenient to use a conditional variable for this purpose given that the 'wait' call helps releasing the access mutex that needs to be released any how.

StreamerThread* AcsBulkdata::BulkDataNTArrayThread::m_Streamer [private]

Memory heap reference to request/release memory. BDF streamers. Each sequence utilize just one streamer per data type.

std::string AcsBulkdata::BulkDataNTArrayThread::m_streamName [private]

double AcsBulkdata::BulkDataNTArrayThread::m_throttling [private]

ACE_CString AcsBulkdata::BulkDataNTArrayThread::name [private]

bool AcsBulkdata::BulkDataNTArrayThread::sequenceAlreadyRunningFlag [private]

flag used to know whether a sequence is already running

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