#include <SyncServer.h>

Inheritance diagram for nidas::dynld::raf::SyncServer:

[legend]

Public Types
enum	runStatus { RUN_CANCELED = -1 , RUN_OK = 0 , NOT_RUNNING = 1 , RUN_EXCEPTION = 2 }
	Values that can be returned by run method. More...

enum	SchedPolicy { NU_THREAD_OTHER =SCHED_OTHER , NU_THREAD_FIFO =SCHED_FIFO , NU_THREAD_RR =SCHED_RR }

Public Member Functions
	SyncServer ()

	~SyncServer ()

void	openStream ()
	Open the data file input stream and read the nidas header, but do not parse the project.

void	init ()
	Call this method to parse the project, setup sample tags, preload calibrations using the time of the first sample, setup sample streams, and init the sensors.

int	run ()
	This method implements the Runnable interface for Threads, but it can also be called synchronously when run from main().

virtual void	interrupt ()
	Interrupt this thread.

void	read (bool once=false)

void	setSorterLengthSeconds (float sorter_secs)

void	setRawSorterLengthSeconds (float sorter_secs)

std::string	getXMLFileName ()
	Return the current XML filename setting.

void	setXMLFileName (const std::string &name)

void	resetAddress (nidas::util::SocketAddress *addr)

void	addSampleClient (SampleClient *client)
	Specify a SampleClient instance to receive the sync samples instead of writing the sync samples to an output stream.

void	setDataFileNames (const std::list< std::string > &dataFileNames)

void	setStopSignal (StopSignal *stop_signal)
	Set a callback function which will be called when the SyncServer reaches EOF on its input stream or stops on an error.

void	getTimeWindow (nidas::util::UTime start, nidas::util::UTime end)

void	setTimeWindow (nidas::util::UTime start, nidas::util::UTime end)

void	connect (SampleOutput *output) throw ()
	Implementation of SampleConnectionRequester::connect().

void	disconnect (SampleOutput *output) throw ()
	Implementation of SampleConnectionRequester::disconnect().

virtual void	start ()
	Start the thread running, meaning execute the run method in a separate thread.

virtual int	join ()
	The calling thread joins this thread, waiting until the thread finishes, which means either that the run() method returned, the thread called pthread_exit() or the thread was cancelled.

virtual void	kill (int sig)
	Send a signal to this thread.

virtual void	cancel ()
	Cancel this thread.

virtual bool	isInterrupted () const
	Return true when this thread has been interrupted.

virtual bool	isRunning () const
	Is this thread running?

virtual bool	isJoined () const
	Has this thread been joined?

virtual bool	isDetached () const
	Is this a detached thread.

bool	isCancelEnabled () const
	Return true if the cancel state of this thread is PTHREAD_CANCEL_ENABLE.

bool	isCancelDeferred () const
	Return true if the cancel type of this thread is PTHREAD_CANCEL_DEFERRED.

const std::string &	getName () const throw ()
	Return the name of this thread.

const std::string &	getFullName () throw ()
	Return a name with a bunch of descriptive fields, specifying whether it is detached, the real-time priority, etc.

bool	setRealTimeRoundRobinPriority (int val)

bool	setRealTimeFIFOPriority (int val)

bool	setNonRealTimePriority ()

void	setThreadScheduler (enum SchedPolicy policy, int priority)

void	blockSignal (int)
	Block a signal in this thread.

void	unblockSignal (int)
	Install a signal handler and unblock the signal.

Static Public Member Functions
static Thread *	currentThread ()
	Return the thread object for the current thread.

static pthread_t	currentThreadId ()

static Thread *	lookupThread (pthread_t id)

static const std::string &	currentName ()
	Convenience routine to return the name for the current thread, or a string indicating that the name of the thread is unknown.

static std::string	getPolicyString (int policy)
	Convenience function to return a string for the given scheduler policy: "Non-RT", "RT:FIFO", "RT:RR" or "RT:Unknown".

static int	test (int argc, char **argv)
	a test method.

Static Public Attributes
static const int	DEFAULT_PORT = 30001

static const float	SORTER_LENGTH_SECS = 5.0

static const float	RAW_SORTER_LENGTH_SECS = 5.0

Protected Member Functions
void	setCancelEnabled (bool val)
	Set the cancel state for this thread - false means cancel requests are ignored.

void	setCancelDeferred (bool val)
	Set the cancel type for this thread.

pthread_t	getId () const

void	testCancel () const
	Check if we have been cancelled.

virtual bool	amInterrupted () const
	Call testCancel, and return true when this thread has been interrupted.

Private Member Functions
void	initProject ()

void	initSensors (SampleInputStream &sis)

void	stop ()

void	signalStop ()

void	handleSample (nidas::core::Sample *sample)

	SyncServer (const SyncServer &)

SyncServer &	operator= (const SyncServer &)

virtual void	signalHandler (int, siginfo_t *)
	Signal handler function for this thread.

virtual int	pRun ()

void	setThreadSchedulerNolock (enum SchedPolicy policy, int priority)
	Set the scheduler policy and priority for this thread, then call setThreadSchedulerNolock() to effect the settings if the thread is running.

void	setThreadSchedulerNolock ()
	If the thread is running, set the scheduling policy according to the current policy and priority attributes in this Thread.

void	makeFullName ()

void	registerThread ()
	Register this current thread with a static registry of threads by id.

void	unregisterThread ()

void	unsetId ()

Static Private Member Functions
static void *	thr_run (void *me)

static void *	thr_run_detached (void *me)

static void	thr_cleanup (void *me)

static void	thr_cleanup_delete (void *me)

static void	thr_add_sig (int sig)

static void	sigAction (int sig, siginfo_t si, void vptr)

Private Attributes
SamplePipeline	_pipeline

SyncRecordGenerator	_syncGen

RawSampleInputStream *	_inputStream

SampleOutputStream *	_outputStream

std::string	_xmlFileName

std::list< std::string >	_dataFileNames

nidas::util::auto_ptr< nidas::util::SocketAddress >	_address

float	_sorterLengthSecs

float	_rawSorterLengthSecs

SampleClient *	_sampleClient

StopSignal *	_stop_signal

nidas::core::Sample *	_firstSample

dsm_time_t	_startTime

dsm_time_t	_startWindow

dsm_time_t	_endWindow

Mutex	_mutex
	Mutex for accessing _id.

std::string	_name

std::string	_fullname

pthread_t	_id

bool	_running

std::atomic< bool >	_interrupted

bool	_cancel_enabled

bool	_cancel_deferred

Exception *	_exception
	Exception thrown by run method.

bool	_detached

SchedPolicy	_policy

int	_priority

sigset_t	_blockedSignals

sigset_t	_unblockedSignals

Static Private Attributes
static std::string	unknownName = std::string("unknown")

Member Enumeration Documentation

◆ runStatus

enum nidas::util::Thread::runStatus

inherited

Values that can be returned by run method.

User can define other values greater than RUN_EXCEPTION. These values are then returned by int join(). Note that PTHREAD_CANCELLED is -1

Enumerator
RUN_CANCELED
RUN_OK
NOT_RUNNING
RUN_EXCEPTION

◆ SchedPolicy

enum nidas::util::Thread::SchedPolicy

inherited

Enumerator
NU_THREAD_OTHER
NU_THREAD_FIFO
NU_THREAD_RR

Constructor & Destructor Documentation

◆ SyncServer() [1/2]

SyncServer::SyncServer ( )

◆ ~SyncServer()

SyncServer::~SyncServer ( )

References _firstSample, _inputStream, _outputStream, _stop_signal, DLOG, and nidas::core::Sample::freeReference().

◆ SyncServer() [2/2]

nidas::dynld::raf::SyncServer::SyncServer ( const SyncServer & )

private

Member Function Documentation

◆ addSampleClient()

void nidas::dynld::raf::SyncServer::addSampleClient ( SampleClient * client )

inline

Specify a SampleClient instance to receive the sync samples instead of writing the sync samples to an output stream.

References _sampleClient, and nidas::core::getSample().

◆ amInterrupted()

bool Runnable::amInterrupted ( ) const

protectedvirtualinherited

Call testCancel, and return true when this thread has been interrupted.

This is protected because it should only called within the run method of the thread.

References nidas::util::Runnable::isInterrupted(), and nidas::util::Runnable::testCancel().

Referenced by nidas::core::Looper::run(), nidas::core::SampleOutputRequestThread::run(), and nidas::core::DSMServerStat::run().

◆ blockSignal()

void Thread::blockSignal ( int sig )

inherited

Block a signal in this thread.

This method is usually called before this Thread has started. If this Thread is currently running, then this method is only effective if called from this Thread, i.e. from its own run() method.

Because SIGINT, SIGTERM and SIGHUP are typically caught in the main thread, they are blocked by default in a Thread. Call unblockSignal(sig) if you want to catch them in a Thread.

References nidas::util::Thread::_blockedSignals, nidas::util::Thread::_unblockedSignals, nidas::util::Thread::currentThread(), nidas::util::Thread::isRunning(), and nidas::util::Thread::thr_add_sig().

Referenced by nidas::dynld::UDPSampleOutput::ConnectionMonitor::ConnectionMonitor(), nidas::core::ServerSocket::ConnectionThread::ConnectionThread(), nidas::core::DerivedDataReader::DerivedDataReader(), nidas::util::McSocketListener::McSocketListener(), nidas::util::McSocketMulticaster< SocketTT >::McSocketMulticaster(), nidas::core::SensorHandler::SensorHandler(), nidas::core::SensorOpener::SensorOpener(), nidas::dynld::RawSampleService::Worker::Worker(), nidas::core::XmlRpcThread::XmlRpcThread(), and nidas::dynld::UDPSampleOutput::XMLSocketListener::XMLSocketListener().

◆ cancel()

void Thread::cancel ( )

virtualinherited

Cancel this thread.

Exceptions

Exception

References nidas::util::Thread::_id, nidas::util::Thread::_mutex, nidas::util::Exception::errnoToString(), and nidas::util::Thread::getName().

◆ connect()

void SyncServer::connect	(	SampleOutput *	output	)
throw	(	)

virtual

Implementation of SampleConnectionRequester::connect().

Does nothing.

Implements nidas::core::SampleConnectionRequester.

◆ currentName()

static const std::string & nidas::util::Thread::currentName ( )

inlinestaticinherited

Convenience routine to return the name for the current thread, or a string indicating that the name of the thread is unknown.

References nidas::util::Thread::currentThread(), and nidas::util::Thread::unknownName.

Referenced by nidas::util::Logger::msg().

◆ currentThread()

Thread * Thread::currentThread ( )

staticinherited

Return the thread object for the current thread.

Returns null if not found.

References nidas::util::Thread::currentThreadId(), and nidas::util::Thread::lookupThread().

Referenced by nidas::util::Thread::blockSignal(), nidas::util::Thread::currentName(), nidas::util::Thread::sigAction(), and nidas::util::Thread::unblockSignal().

◆ currentThreadId()

pthread_t Thread::currentThreadId ( )

staticinherited

Referenced by nidas::util::Thread::currentThread(), and nidas::util::Thread::sigAction().

◆ disconnect()

void SyncServer::disconnect	(	SampleOutput *	output	)
throw	(	)

virtual

Implementation of SampleConnectionRequester::disconnect().

If client has disconnected, interrupt the sample loop and exit.

Implements nidas::core::SampleConnectionRequester.

◆ getFullName()

const std::string & Thread::getFullName	(		)
throw	(	)

inherited

Return a name with a bunch of descriptive fields, specifying whether it is detached, the real-time priority, etc.

References nidas::util::Thread::_fullname.

Referenced by nidas::util::Thread::pRun().

◆ getId()

pthread_t nidas::util::Thread::getId ( ) const

inlineprotectedinherited

References nidas::util::Thread::_id, and nidas::util::Thread::_mutex.

Referenced by nidas::util::Thread::isJoined(), nidas::util::Thread::join(), and nidas::core::Looper::removeClient().

◆ getName()

const std::string & Thread::getName	(		)	const
throw	(	)

inherited

Return the name of this thread.

References nidas::util::Thread::_name.

Referenced by nidas::util::Thread::cancel(), nidas::core::SensorHandler::PolledDSMSensor::checkTimeout(), nidas::core::SensorHandler::PolledDSMSensor::close(), nidas::core::SensorHandler::PolledDSMSensor::handlePollEvents(), nidas::core::SampleSorter::heapDecrement(), nidas::core::XmlRpcThread::interrupt(), nidas::core::SamplePipeline::join(), nidas::util::Thread::join(), nidas::util::Thread::kill(), nidas::core::FsMountWorkerThread::run(), nidas::core::SampleSorter::run(), nidas::util::ThreadJoiner::run(), nidas::util::Thread::setThreadSchedulerNolock(), nidas::util::Thread::start(), nidas::core::SampleSorter::~SampleSorter(), and nidas::util::Thread::~Thread().

◆ getPolicyString()

string Thread::getPolicyString ( int policy )

staticinherited

Convenience function to return a string for the given scheduler policy: "Non-RT", "RT:FIFO", "RT:RR" or "RT:Unknown".

References nidas::util::Thread::NU_THREAD_FIFO, nidas::util::Thread::NU_THREAD_OTHER, and nidas::util::Thread::NU_THREAD_RR.

Referenced by nidas::util::Thread::makeFullName().

◆ getTimeWindow()

void SyncServer::getTimeWindow	(	nidas::util::UTime *	start,
		nidas::util::UTime *	end )

References _endWindow, _startWindow, and nidas::util::Thread::start().

◆ getXMLFileName()

std::string nidas::dynld::raf::SyncServer::getXMLFileName ( )

inline

Return the current XML filename setting.

If called after calling openStream() and before setting it explicitly with setXMLFileName(), then the return value is the XML filename from the stream header.

References _xmlFileName.

◆ handleSample()

void SyncServer::handleSample ( nidas::core::Sample * sample )

private

References _endWindow, _inputStream, _startWindow, and nidas::dynld::SampleInputStream::distribute().

Referenced by read().

◆ init()

void SyncServer::init ( )

Call this method to parse the project, setup sample tags, preload calibrations using the time of the first sample, setup sample streams, and init the sensors.

It should be called before calling start() on the thread or before calling run() directly.

Between calling this method and entering run(), no samples have been processed, so the calibrations can be replaced and will take effect when processing starts.

Exceptions

nidas::util::Exception

References _address, _inputStream, _outputStream, _pipeline, _rawSorterLengthSecs, _sampleClient, _sorterLengthSecs, _startTime, _syncGen, nidas::dynld::raf::SyncRecordGenerator::addSampleClient(), nidas::core::IOChannel::connect(), nidas::dynld::raf::SyncRecordGenerator::connect(), nidas::core::SamplePipeline::connect(), nidas::dynld::raf::SyncRecordGenerator::connectSource(), DLOG, nidas::core::SamplePipeline::getProcessedSampleSource(), nidas::core::getSample(), nidas::dynld::raf::SyncRecordGenerator::init(), initProject(), initSensors(), openStream(), nidas::core::SamplePipeline::setProcHeapMax(), nidas::core::SamplePipeline::setProcSorterLength(), nidas::core::SamplePipeline::setRawHeapMax(), nidas::core::SamplePipeline::setRawSorterLength(), nidas::core::SamplePipeline::setRealTime(), and nidas::core::SampleOutputBase::setReconnectDelaySecs().

Referenced by main().

◆ initProject()

void SyncServer::initProject ( )

private

References _xmlFileName, DLOG, and nidas::core::Project::getInstance().

Referenced by init().

◆ initSensors()

void SyncServer::initSensors ( SampleInputStream & sis )

private

References nidas::core::DSMSensor::addSampleTag(), nidas::core::Project::getInstance(), nidas::core::DSMSensor::getRawSampleTag(), nidas::core::getSample(), and nidas::core::DSMSensor::init().

Referenced by init().

◆ interrupt()

void SyncServer::interrupt ( )

virtual

Interrupt this thread.

This sets a boolean which can be tested with isInterrupted(). It is up to the run() implementation to return when interrupted. This is a "soft" request to terminate the thread.

What follows is a discussion of when to use cancel(), kill(), or interrupt() to terminate a thread.

If you can consistently check the state of isInterrupted() in the run method, and return if it is true, at a time interval which is an acceptable amount of time to wait for the thread to terminate, then using interrupt() should work well.

If the run method does I/O, then things are usually a bit more complicated.

If all I/O is guaranteed to finish quickly, which is a rare situation, or is done with a timeout, such as using select/poll with a timeout before every read/write, then one could still use interrupt() as above, and the thread will terminate within the timeout period.

If the thread does blocking I/O operations, and the I/O might block for a period of time longer than you're willing to wait for the thread to terminate, then you need to use kill(sig) or cancel().

If a signal is sent to the thread with kill(sig) while the thread is blocking on an I/O operation, the I/O will immediately return with an errno of EINTR, and one could then return from the run method, after any necessary cleanup.

However there is a possibility that the signal could be missed. In order to make sure you receive a signal, you must block it, so that any received signals are held as pending, and then atomically unblock it with pselect/ppoll/epoll_pwait before performing the I/O operation.

If you do not use I/O timeouts, or kill(sig) with pselect/ppoll/epoll_pwait, or other tricks such as writing to a pipe that is watched with select/poll in the run method, then using cancel() is the only way to guarantee that your thread will terminate in an acceptable amount of time.

All NIDAS Threads support deferred cancelation. Immediate asynchronous cancellation is not fully supported (and is very hard to get right). Deferred cancelation means that cancellation is delayed until the thread next calls a system function that is a cancellation point. At that point the thread run method will simply terminate without any return value from the system function, and then execute any cleanup methods that may have been registered with pthread_cleanup_push.

A list of cancellation points is provided in the pthreads(7) man page. Cancellation points are typically I/O operations, waits or sleeps.

This immediate thread termination can be a problem if there is a possibility that your objects could be left in a bad state, such as with a mutex locked, though it isn't generally a good practice to hold mutexes during a time-consuming I/O operation, wait or sleep.

One should check the run method to see if the state of the objects is OK if execution stops at any of the cancellation points. Note that logging a message, or writing to cerr is a cancellation point. Typically there is error/exception handling associated with an I/O operation. Ensuring the state is OK upon a cancellation is similar to preparation for a fatal I/O error that requires a return of the run method. One can use setCancelEnabled() to defer cancellation.

One can use pthread_cleanup_push and pthread_cleanup_pop to register cleanup routines that are called when a thread is cancelled if special handling is required.

Reimplemented from nidas::util::Thread.

References _pipeline, DLOG, nidas::core::SamplePipeline::interrupt(), and signalStop().

Referenced by interrupt_sync_server().

◆ isCancelDeferred()

bool Thread::isCancelDeferred ( ) const

inherited

Return true if the cancel type of this thread is PTHREAD_CANCEL_DEFERRED.

References nidas::util::Thread::_cancel_deferred.

◆ isCancelEnabled()

bool Thread::isCancelEnabled ( ) const

inherited

Return true if the cancel state of this thread is PTHREAD_CANCEL_ENABLE.

References nidas::util::Thread::_cancel_enabled.

◆ isDetached()

virtual bool nidas::util::Thread::isDetached ( ) const

inlinevirtualinherited

Is this a detached thread.

References nidas::util::Thread::_detached.

◆ isInterrupted()

virtual bool nidas::util::Thread::isInterrupted ( ) const

inlinevirtualinherited

◆ isJoined()

virtual bool nidas::util::Thread::isJoined ( ) const

inlinevirtualinherited

Has this thread been joined?

References nidas::util::Thread::getId().

Referenced by nidas::core::FsMount::cancel(), nidas::core::SensorHandler::join(), nidas::core::SampleBuffer::~SampleBuffer(), nidas::core::SampleSorter::~SampleSorter(), and nidas::util::Thread::~Thread().

◆ isRunning()

virtual bool nidas::util::Thread::isRunning ( ) const

inlinevirtualinherited

Is this thread running?

References nidas::util::Thread::_mutex, and nidas::util::Thread::_running.

Referenced by nidas::util::Thread::blockSignal(), nidas::core::FsMount::cancel(), nidas::dynld::UDPSampleOutput::XMLSocketListener::checkWorkers(), nidas::dynld::UDPSampleOutput::close(), nidas::core::SamplePipeline::join(), nidas::core::DSMServerApp::killStatusThread(), nidas::core::Looper::removeClient(), nidas::core::ServerSocket::requestConnection(), nidas::core::SensorHandler::run(), nidas::core::Looper::setupClientMaps(), nidas::util::Thread::unblockSignal(), nidas::core::SampleBuffer::~SampleBuffer(), nidas::core::SampleSorter::~SampleSorter(), nidas::util::Thread::~Thread(), and nidas::dynld::UDPSampleOutput::~UDPSampleOutput().

◆ join()

int Thread::join ( )

virtualinherited

The calling thread joins this thread, waiting until the thread finishes, which means either that the run() method returned, the thread called pthread_exit() or the thread was cancelled.

The return value is the int return value of the run method, or PTHREAD_CANCELED (-1). If the run method threw an Exception, it will be caught and then re-thrown by join.

Exceptions

Exception

Reimplemented in nidas::core::SensorHandler.

References nidas::util::Thread::_exception, nidas::util::Thread::_id, nidas::util::Thread::_mutex, nidas::util::Exception::errnoToString(), nidas::util::Thread::getId(), nidas::util::Thread::getName(), nidas::util::Thread::RUN_EXCEPTION, and nidas::util::Thread::RUN_OK.

Referenced by nidas::dynld::UDPSampleOutput::XMLSocketListener::checkWorkers(), nidas::dynld::UDPSampleOutput::close(), nidas::dynld::UDPSampleOutput::XMLSocketListener::fireWorkers(), nidas::core::SamplePipeline::join(), nidas::core::SensorHandler::join(), nidas::core::DSMEngine::joinDataThreads(), nidas::core::DSMServerApp::killStatusThread(), nidas::core::DSMEngine::killXmlRpcThread(), nidas::core::DSMServerApp::killXmlRpcThread(), nidas::core::Looper::removeClient(), nidas::util::ThreadJoiner::run(), nidas::core::SampleBuffer::~SampleBuffer(), nidas::core::SampleSorter::~SampleSorter(), nidas::core::ServerSocket::~ServerSocket(), and nidas::dynld::UDPSampleOutput::~UDPSampleOutput().

◆ kill()

void Thread::kill ( int sig )

virtualinherited

Send a signal to this thread.

Exceptions

Exception

References nidas::util::Thread::_id, nidas::util::Thread::_mutex, nidas::util::Exception::errnoToString(), and nidas::util::Thread::getName().

Referenced by nidas::core::FsMount::cancel(), nidas::core::DerivedDataReader::interrupt(), nidas::core::SensorHandler::interrupt(), nidas::core::SensorOpener::interrupt(), nidas::core::XmlRpcThread::interrupt(), nidas::util::McSocketListener::interrupt(), nidas::core::DSMServerApp::killStatusThread(), nidas::core::SensorHandler::scheduleAdd(), nidas::core::SensorHandler::scheduleClose(), nidas::core::SensorHandler::scheduleClose(), nidas::core::SensorHandler::scheduleReopen(), and nidas::core::SensorHandler::sensorIsOpen().

◆ lookupThread()

Thread * Thread::lookupThread ( pthread_t id )

staticinherited

Referenced by nidas::util::Thread::currentThread(), and nidas::util::LogContextState::threadName().

◆ makeFullName()

void Thread::makeFullName ( )

privateinherited

References nidas::util::Thread::_cancel_deferred, nidas::util::Thread::_cancel_enabled, nidas::util::Thread::_detached, nidas::util::Thread::_fullname, nidas::util::Thread::_id, nidas::util::Thread::_name, and nidas::util::Thread::getPolicyString().

Referenced by nidas::util::Thread::start().

◆ openStream()

void SyncServer::openStream ( )

Open the data file input stream and read the nidas header, but do not parse the project.

References _dataFileNames, _firstSample, _inputStream, _startTime, _xmlFileName, DLOG, nidas::core::SampleInputHeader::getConfigName(), nidas::core::FileSet::getFileSet(), nidas::core::Sample::getTimeTag(), and nidas::dynld::SampleInputStream::setMaxDsmId().

Referenced by init().

◆ operator=()

SyncServer & nidas::dynld::raf::SyncServer::operator= ( const SyncServer & )

private

◆ pRun()

int Thread::pRun ( )

privatevirtualinherited

References nidas::util::Thread::_blockedSignals, nidas::util::Thread::_exception, nidas::util::Thread::_mutex, nidas::util::Thread::_unblockedSignals, nidas::util::Exception::clone(), nidas::util::Thread::getFullName(), ILOG, nidas::util::Mutex::lock(), nidas::util::Runnable::run(), nidas::util::Thread::RUN_EXCEPTION, nidas::util::Thread::setThreadSchedulerNolock(), and nidas::util::Mutex::unlock().

◆ read()

void SyncServer::read ( bool once = false )

Exceptions

nidas::util::IOException

References _firstSample, _inputStream, _pipeline, DLOG, nidas::core::SamplePipeline::flush(), nidas::core::getSample(), handleSample(), nidas::util::Thread::isInterrupted(), and stop().

Referenced by nidas::dynld::raf::SyncRecordReader::read(), and run().

◆ registerThread()

void Thread::registerThread ( )

privateinherited

Register this current thread with a static registry of threads by id.

Thread::registerThread (new Thread ("Main"));

References nidas::util::Thread::_id.

Referenced by nidas::util::Thread::start().

◆ resetAddress()

void nidas::dynld::raf::SyncServer::resetAddress ( nidas::util::SocketAddress * addr )

inline

References _address.

Referenced by parseRunstring().

◆ run()

int SyncServer::run ( )

virtual

This method implements the Runnable interface for Threads, but it can also be called synchronously when run from main().

Implements nidas::util::Runnable.

References DLOG, nidas::core::getSample(), read(), stop(), and nidas::core::XMLImplementation::terminate().

Referenced by main().

◆ setCancelDeferred()

void Thread::setCancelDeferred ( bool val )

protectedinherited

Set the cancel type for this thread.

true means cancel requests are deferred until the next cancellation point. false means they occur instantly. This is protected, it should be called only from a thread's own run method. See the pthreads(7) man page for a list of the cancellation points.

Note: non-deferred canceling is difficult to get right. It has not been tested with this class, and is not recommended.

References nidas::util::Thread::_cancel_deferred.

◆ setCancelEnabled()

void Thread::setCancelEnabled ( bool val )

protectedinherited

Set the cancel state for this thread - false means cancel requests are ignored.

See pthread_setcancelstate. This is protected, it should be called only from a thread's own run method.

References nidas::util::Thread::_cancel_enabled.

◆ setDataFileNames()

void nidas::dynld::raf::SyncServer::setDataFileNames ( const std::list< std::string > & dataFileNames )

inline

References _dataFileNames.

Referenced by parseRunstring().

◆ setNonRealTimePriority()

bool Thread::setNonRealTimePriority ( )

inherited

Exceptions

Exception

References nidas::util::Thread::NU_THREAD_OTHER, and nidas::util::Thread::setThreadScheduler().

◆ setRawSorterLengthSeconds()

void nidas::dynld::raf::SyncServer::setRawSorterLengthSeconds ( float sorter_secs )

inline

References _rawSorterLengthSecs, and nidas::core::getSample().

Referenced by parseRunstring().

◆ setRealTimeFIFOPriority()

bool Thread::setRealTimeFIFOPriority ( int val )

inherited

Exceptions

Exception

References nidas::util::Thread::NU_THREAD_FIFO, and nidas::util::Thread::setThreadScheduler().

Referenced by nidas::core::DSMEngine::openSensors(), nidas::core::SamplePipeline::procinit(), and nidas::core::SamplePipeline::rawinit().

◆ setRealTimeRoundRobinPriority()

bool Thread::setRealTimeRoundRobinPriority ( int val )

inherited

Exceptions

Exception

References nidas::util::Thread::NU_THREAD_RR, and nidas::util::Thread::setThreadScheduler().

◆ setSorterLengthSeconds()

void nidas::dynld::raf::SyncServer::setSorterLengthSeconds ( float sorter_secs )

inline

References _sorterLengthSecs, and nidas::core::getSample().

Referenced by parseRunstring().

◆ setStopSignal()

void nidas::dynld::raf::SyncServer::setStopSignal ( StopSignal * stop_signal )

inline

Set a callback function which will be called when the SyncServer reaches EOF on its input stream or stops on an error.

Since the pipeline is flushed when the input is closed, no more sync samples will be sent after this function is called. It is called from the thread running the run() method.

References _stop_signal, and nidas::core::getSample().

◆ setThreadScheduler()

void Thread::setThreadScheduler	(	enum SchedPolicy	policy,
		int	priority )

inherited

Exceptions

Exception

References nidas::util::Thread::_mutex, and nidas::util::Thread::setThreadSchedulerNolock().

Referenced by nidas::dynld::RawSampleService::connect(), nidas::util::Thread::setNonRealTimePriority(), nidas::util::Thread::setRealTimeFIFOPriority(), and nidas::util::Thread::setRealTimeRoundRobinPriority().

◆ setThreadSchedulerNolock() [1/2]

void Thread::setThreadSchedulerNolock ( )

privateinherited

If the thread is running, set the scheduling policy according to the current policy and priority attributes in this Thread.

Warn but continue if realtime priority cannot be set, otherwise raise Exception.

Exceptions

Exception

References nidas::util::Thread::_id, nidas::util::Thread::_policy, nidas::util::Thread::_priority, ELOG, nidas::util::Exception::errnoToString(), nidas::util::Thread::getName(), nidas::util::Thread::NU_THREAD_FIFO, nidas::util::Thread::NU_THREAD_RR, and WLOG.

Referenced by nidas::util::Thread::pRun(), nidas::util::Thread::setThreadScheduler(), and nidas::util::Thread::setThreadSchedulerNolock().

◆ setThreadSchedulerNolock() [2/2]

void Thread::setThreadSchedulerNolock	(	enum SchedPolicy	policy,
		int	priority )

privateinherited

Set the scheduler policy and priority for this thread, then call setThreadSchedulerNolock() to effect the settings if the thread is running.

Exceptions

Exception

References nidas::util::Thread::_policy, nidas::util::Thread::_priority, and nidas::util::Thread::setThreadSchedulerNolock().

◆ setTimeWindow()

void SyncServer::setTimeWindow	(	nidas::util::UTime	start,
		nidas::util::UTime	end )

References _endWindow, _startWindow, DLOG, nidas::util::Thread::start(), and nidas::util::UTime::toUsecs().

◆ setXMLFileName()

void nidas::dynld::raf::SyncServer::setXMLFileName ( const std::string & name )

inline

References _xmlFileName.

Referenced by parseRunstring().

◆ sigAction()

void Thread::sigAction	(	int	sig,
		siginfo_t *	si,
		void *	vptr )

staticprivateinherited

References nidas::util::Thread::currentThread(), nidas::util::Thread::currentThreadId(), and DLOG.

Referenced by nidas::util::Thread::thr_add_sig().

◆ signalHandler()

virtual void nidas::util::Thread::signalHandler	(	int	,
		siginfo_t *	)

inlineprivatevirtualinherited

Signal handler function for this thread.

The default handler just sets _interrupt to true, so that amInterrupted() or isInterrupted() will return true. Derived classes can override this method for custom signal handling. However, derived classes are limited in what they can do in their signal handler. Specifically, from the pthread_cond_signal man page:

"It is not safe to use the pthread_cond_signal() function in a signal handler that is invoked asynchronously."

Therefore do not call Cond::signal() from a signal handler.

Reimplemented in nidas::core::SensorHandler.

References nidas::util::Thread::_interrupted.

◆ signalStop()

void SyncServer::signalStop ( )

private

References _stop_signal, DLOG, and nidas::dynld::raf::StopSignal::stop().

Referenced by interrupt(), and stop().

◆ start()

void Thread::start ( )

virtualinherited

Start the thread running, meaning execute the run method in a separate thread.

Exceptions

Exception

References nidas::util::Thread::_detached, nidas::util::Thread::_id, nidas::util::Thread::_interrupted, nidas::util::Thread::_mutex, nidas::util::Thread::_running, nidas::util::Exception::errnoToString(), nidas::util::Thread::getName(), nidas::util::Thread::makeFullName(), nidas::util::Thread::registerThread(), nidas::util::Runnable::run(), nidas::util::Thread::thr_run(), and nidas::util::Thread::thr_run_detached().

Referenced by nidas::dynld::RawSampleService::connect(), nidas::dynld::XMLConfigAllService::connected(), nidas::dynld::XMLConfigService::connected(), getTimeWindow(), main(), nidas::core::FsMount::mount(), nidas::core::DSMEngine::openSensors(), nidas::core::SamplePipeline::procinit(), nidas::core::SamplePipeline::rawinit(), nidas::core::Socket::requestConnection(), nidas::core::ServerSocket::requestConnection(), nidas::core::DSMEngine::run(), nidas::core::SensorHandler::run(), nidas::dynld::UDPSampleOutput::XMLSocketListener::run(), setTimeWindow(), nidas::core::Looper::setupClientMaps(), nidas::core::DSMServerApp::startStatusThread(), nidas::core::DSMEngine::startXmlRpcThread(), and nidas::core::DSMServerApp::startXmlRpcThread().

◆ stop()

◆ test()

int Thread::test	(	int	argc,
		char **	argv )

staticinherited

a test method.

◆ testCancel()

void nidas::util::Runnable::testCancel ( ) const

inlineprotectedinherited

Check if we have been cancelled.

Calls pthread_testcancel. This is protected since it only checks the current thread - i.e. it must be called within the run method. Since it just calls pthread_testcancel, it is a cancellation point.

Referenced by nidas::util::Runnable::amInterrupted(), and TestSource::run().

◆ thr_add_sig()

void Thread::thr_add_sig ( int sig )

staticprivateinherited

References nidas::util::Thread::sigAction().

Referenced by nidas::util::Thread::blockSignal(), and nidas::util::Thread::unblockSignal().

◆ thr_cleanup()

void Thread::thr_cleanup ( void * me )

staticprivateinherited

References ILOG.

Referenced by nidas::util::Thread::thr_run().

◆ thr_cleanup_delete()

void Thread::thr_cleanup_delete ( void * me )

staticprivateinherited

References nidas::util::Thread::_mutex, and nidas::util::Mutex::lock().

Referenced by nidas::util::Thread::thr_run_detached().

◆ thr_run()

void * Thread::thr_run ( void * me )

staticprivateinherited

References nidas::util::Thread::thr_cleanup().

Referenced by nidas::util::Thread::start().

◆ thr_run_detached()

void * Thread::thr_run_detached ( void * me )

staticprivateinherited

References nidas::util::Thread::thr_cleanup_delete().

Referenced by nidas::util::Thread::start().

◆ unblockSignal()

void Thread::unblockSignal ( int sig )

inherited

Install a signal handler and unblock the signal.

The signal handler will log a message about the receipt of the signal at severity LOG_INFO using the nidas::util::Logger. Then, if the signal handler is being invoked from a registered Thread, the virtual method signalHandler() for that Thread will be called.

The signal handler is installed with the sigaction() system call, and will be the action for the given signal in all threads, including the main() thread. If other threads do not wish to take action on a given signal, they should call blockSignal(sig). Or they can define their own signalHandler() method.

After installing the signal handler, the signal is added to those that are unblocked for the thread, or if the Thread is not yet running, the signal will be unblocked in the thread once it runs.

As with blockSignal(), this method is typically called on this Thread before it has started. If this Thread has started, then the signal will only be unblocked if the method is called from this Thread, i.e. from its own run() method.

To install a signal handler, and then block the signal so that it is held as pending until it is later unblocked, typically with pselect(), or sigwaitinfo(), do:

void Thread::run() 
{
    // get the existing signal mask
    sigset_t sigmask;
    pthread_sigmask(SIG_BLOCK,NULL,&sigmask);
    // remove SIGUSR1 from the mask passed to pselect
    sigdelset(&sigmask,SIGUSR1);
    
    for (;;) {
        pselect(nfd,&readfds,0,0,0,&sigmask);
        ...
    }
}
...
thread.unblockSignal(SIGUSR1);
thread.blockSignal(SIGUSR1);
thread.start();
...
try {
    if (thread.isRunning()) {
        thread.kill(SIGUSR1);
        thread.join()
    }
}

References nidas::util::Thread::_blockedSignals, nidas::util::Thread::_unblockedSignals, nidas::util::Thread::currentThread(), nidas::util::Thread::isRunning(), and nidas::util::Thread::thr_add_sig().

Referenced by nidas::core::DerivedDataReader::DerivedDataReader(), nidas::core::FsMountWorkerThread::FsMountWorkerThread(), nidas::core::StatusListener::StatusListener(), nidas::core::StatusThread::StatusThread(), nidas::dynld::XMLConfigService::Worker::Worker(), nidas::core::XmlRpcThread::XmlRpcThread(), and nidas::dynld::UDPSampleOutput::XMLSocketListener::XMLSocketListener().

◆ unregisterThread()

void Thread::unregisterThread ( )

privateinherited

References nidas::util::Thread::_id.

◆ unsetId()

void nidas::util::Thread::unsetId ( )

inlineprivateinherited

References nidas::util::Thread::_id, and nidas::util::Thread::_mutex.

Member Data Documentation

◆ _address

nidas::util::auto_ptr<nidas::util::SocketAddress> nidas::dynld::raf::SyncServer::_address

private

Referenced by init(), and resetAddress().

◆ _blockedSignals

sigset_t nidas::util::Thread::_blockedSignals

privateinherited

Referenced by nidas::util::Thread::blockSignal(), nidas::util::Thread::pRun(), nidas::util::Thread::Thread(), and nidas::util::Thread::unblockSignal().

◆ _cancel_deferred

bool nidas::util::Thread::_cancel_deferred

privateinherited

Referenced by nidas::util::Thread::isCancelDeferred(), nidas::util::Thread::makeFullName(), and nidas::util::Thread::setCancelDeferred().

◆ _cancel_enabled

bool nidas::util::Thread::_cancel_enabled

privateinherited

Referenced by nidas::util::Thread::isCancelEnabled(), nidas::util::Thread::makeFullName(), and nidas::util::Thread::setCancelEnabled().

◆ _dataFileNames

std::list<std::string> nidas::dynld::raf::SyncServer::_dataFileNames

private

Referenced by openStream(), and setDataFileNames().

◆ _detached

bool nidas::util::Thread::_detached

privateinherited

Referenced by nidas::util::Thread::isDetached(), nidas::util::Thread::makeFullName(), nidas::util::Thread::start(), and nidas::util::Thread::~Thread().

◆ _endWindow

dsm_time_t nidas::dynld::raf::SyncServer::_endWindow

private

Referenced by getTimeWindow(), handleSample(), and setTimeWindow().

◆ _exception

Exception* nidas::util::Thread::_exception

privateinherited

Exception thrown by run method.

Referenced by nidas::util::Thread::join(), nidas::util::Thread::pRun(), and nidas::util::Thread::~Thread().

◆ _firstSample

nidas::core::Sample* nidas::dynld::raf::SyncServer::_firstSample

private

Referenced by openStream(), read(), and ~SyncServer().

◆ _fullname

std::string nidas::util::Thread::_fullname

privateinherited

Referenced by nidas::util::Thread::getFullName(), and nidas::util::Thread::makeFullName().

◆ _id

pthread_t nidas::util::Thread::_id

privateinherited

Referenced by nidas::util::Thread::cancel(), nidas::util::Thread::getId(), nidas::util::Thread::join(), nidas::util::Thread::kill(), nidas::util::Thread::makeFullName(), nidas::util::Thread::registerThread(), nidas::util::Thread::setThreadSchedulerNolock(), nidas::util::Thread::start(), nidas::util::Thread::unregisterThread(), and nidas::util::Thread::unsetId().

◆ _inputStream

RawSampleInputStream* nidas::dynld::raf::SyncServer::_inputStream

private

Referenced by handleSample(), init(), openStream(), read(), stop(), and ~SyncServer().

◆ _interrupted

std::atomic<bool> nidas::util::Thread::_interrupted

privateinherited

Referenced by nidas::util::Thread::interrupt(), nidas::util::Thread::isInterrupted(), nidas::util::Thread::signalHandler(), and nidas::util::Thread::start().

◆ _mutex

Mutex nidas::util::Thread::_mutex

mutableprivateinherited

Mutex for accessing _id.

Referenced by nidas::util::Thread::cancel(), nidas::util::Thread::getId(), nidas::util::Thread::interrupt(), nidas::util::Thread::isInterrupted(), nidas::util::Thread::isRunning(), nidas::util::Thread::join(), nidas::util::Thread::kill(), nidas::util::Thread::pRun(), nidas::util::Thread::setThreadScheduler(), nidas::util::Thread::start(), nidas::util::Thread::thr_cleanup_delete(), and nidas::util::Thread::unsetId().

◆ _name

std::string nidas::util::Thread::_name

privateinherited

Referenced by nidas::util::Thread::getName(), and nidas::util::Thread::makeFullName().

◆ _outputStream

SampleOutputStream* nidas::dynld::raf::SyncServer::_outputStream

private

Referenced by init(), stop(), and ~SyncServer().

◆ _pipeline

SamplePipeline nidas::dynld::raf::SyncServer::_pipeline

private

Referenced by init(), interrupt(), read(), and stop().

◆ _policy

SchedPolicy nidas::util::Thread::_policy

privateinherited

Referenced by nidas::util::Thread::setThreadSchedulerNolock(), and nidas::util::Thread::setThreadSchedulerNolock().

◆ _priority

int nidas::util::Thread::_priority

privateinherited

Referenced by nidas::util::Thread::setThreadSchedulerNolock(), and nidas::util::Thread::setThreadSchedulerNolock().

◆ _rawSorterLengthSecs

float nidas::dynld::raf::SyncServer::_rawSorterLengthSecs

private

Referenced by init(), and setRawSorterLengthSeconds().

◆ _running

bool nidas::util::Thread::_running

privateinherited

Referenced by nidas::util::Thread::isRunning(), and nidas::util::Thread::start().

◆ _sampleClient

SampleClient* nidas::dynld::raf::SyncServer::_sampleClient

private

Referenced by addSampleClient(), init(), and stop().

◆ _sorterLengthSecs

float nidas::dynld::raf::SyncServer::_sorterLengthSecs

private

Referenced by init(), and setSorterLengthSeconds().

◆ _startTime

dsm_time_t nidas::dynld::raf::SyncServer::_startTime

private

Referenced by init(), and openStream().

◆ _startWindow

dsm_time_t nidas::dynld::raf::SyncServer::_startWindow

private

Referenced by getTimeWindow(), handleSample(), and setTimeWindow().

◆ _stop_signal

StopSignal* nidas::dynld::raf::SyncServer::_stop_signal

private

Referenced by setStopSignal(), signalStop(), and ~SyncServer().

◆ _syncGen

SyncRecordGenerator nidas::dynld::raf::SyncServer::_syncGen

private

Referenced by init(), and stop().

◆ _unblockedSignals

sigset_t nidas::util::Thread::_unblockedSignals

privateinherited

Referenced by nidas::util::Thread::blockSignal(), nidas::util::Thread::pRun(), nidas::util::Thread::Thread(), and nidas::util::Thread::unblockSignal().

◆ _xmlFileName

std::string nidas::dynld::raf::SyncServer::_xmlFileName

private

Referenced by getXMLFileName(), initProject(), openStream(), and setXMLFileName().

◆ DEFAULT_PORT

const int nidas::dynld::raf::SyncServer::DEFAULT_PORT = 30001

static

Referenced by usage().

◆ RAW_SORTER_LENGTH_SECS

const float SyncServer::RAW_SORTER_LENGTH_SECS = 5.0

static

Referenced by usage().

◆ SORTER_LENGTH_SECS

const float SyncServer::SORTER_LENGTH_SECS = 5.0

static

Referenced by usage().

◆ unknownName

std::string Thread::unknownName = std::string("unknown")

staticprivateinherited

Referenced by nidas::util::Thread::currentName().

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

nidas/dynld/raf/SyncServer.h
nidas/dynld/raf/SyncServer.cc

Public Types

Public Member Functions

Static Public Member Functions

Static Public Attributes

Protected Member Functions

Private Member Functions

Static Private Member Functions

Private Attributes

Static Private Attributes

Member Enumeration Documentation

◆ runStatus

◆ SchedPolicy

Constructor & Destructor Documentation

◆ SyncServer() [1/2]

◆ ~SyncServer()

◆ SyncServer() [2/2]

Member Function Documentation

◆ addSampleClient()

◆ amInterrupted()

◆ blockSignal()

◆ cancel()

◆ connect()

◆ currentName()

◆ currentThread()

◆ currentThreadId()

◆ disconnect()

◆ getFullName()

◆ getId()

◆ getName()

◆ getPolicyString()

◆ getTimeWindow()

◆ getXMLFileName()

◆ handleSample()

◆ init()

◆ initProject()

◆ initSensors()

◆ interrupt()

◆ isCancelDeferred()

◆ isCancelEnabled()

◆ isDetached()

◆ isInterrupted()

◆ isJoined()

◆ isRunning()

◆ join()

◆ kill()

◆ lookupThread()

◆ makeFullName()

◆ openStream()

◆ operator=()

◆ pRun()

◆ read()

◆ registerThread()

◆ resetAddress()

◆ run()

◆ setCancelDeferred()

◆ setCancelEnabled()

◆ setDataFileNames()

◆ setNonRealTimePriority()

◆ setRawSorterLengthSeconds()

◆ setRealTimeFIFOPriority()

◆ setRealTimeRoundRobinPriority()

◆ setSorterLengthSeconds()

◆ setStopSignal()

◆ setThreadScheduler()

◆ setThreadSchedulerNolock() [1/2]

◆ setThreadSchedulerNolock() [2/2]

◆ setTimeWindow()

◆ setXMLFileName()

◆ sigAction()

◆ signalHandler()

◆ signalStop()

◆ start()

◆ stop()

◆ test()

◆ testCancel()

◆ thr_add_sig()

◆ thr_cleanup()

◆ thr_cleanup_delete()

◆ thr_run()

◆ thr_run_detached()