Event-driven reactor class implementing the "event loop" pattern. More...

#include <tsReactor.h>

Inheritance diagram for ts::Reactor:

[legend]

Collaboration diagram for ts::Reactor:

[legend]

Public Member Functions
	Reactor (Report *report)
	Constructor.

	Reactor (ReporterBase *delegate)
	Constructor.

virtual	~Reactor () override
	Destructor.

int	addExitReference ()
	Coordinate future exitEventLoop().

bool	cancelAndWaitAsynchronousIO (EventId id, NonBlockingDevice::IOSB &iosb, bool silent=false)
	Cancel one specific pending asynchronous I/O.

bool	cancelAsynchronousIO (EventId id, bool silent=false)
	Cancel all pending asynchronous I/O on a system file descriptor or handle.

bool	cancelProcessTermination (EventId id, bool silent=false)
	Cancel a process termination event.

bool	cancelTimer (EventId id, bool silent=false)
	Cancel a timer.

bool	close (bool silent=false)
	Close the Reactor.

bool	deleteAsynchronousIO (EventId id, bool silent=false)
	Delete a notification of asynchronous I/O.

bool	deleteEvent (EventId id, bool silent=false)
	Delete a user event.

bool	deleteReadNotify (EventId id, bool silent=false)
	Delete a notification of read-ready or read-completion.

bool	deleteWriteNotify (EventId id, bool silent=false)
	Delete a notification of write-ready or write-completion.

void	exitEventLoop (bool success=true)
	Exit processEventLoop() as soon as possible.

int	freeExitReference (bool success=true)
	Coordinated exitEventLoop().

bool	isActiveEvent (EventId id)
	Check if an event (user-defined, timer, I/O) is still active in the reactor.

bool	isOpen () const
	Check if the Reactor is open.

bool	muteReport (bool mute)
	Temporarily mute the associated report.

EventId	newAsynchronousIO (ReactorHandlerInterface *handler, SysSocketType sock)
	Add in the reactor a notification of asynchronous I/O on a system file descriptor or handle.

EventId	newEvent (ReactorHandlerInterface *handler)
	Add a user event in the reactor.

EventId	newProcessHandleTermination (ReactorHandlerInterface *handler, SysHandleType process_handle)
	Add a process termination event in the reactor, using a process handle.

EventId	newProcessIdTermination (ReactorHandlerInterface *handler, SysProcessIdType pid)
	Add a process termination event in the reactor, using a process id.

EventId	newReadNotify (ReactorHandlerInterface *handler, SysSocketType sock)
	Add in the reactor a notification of read-ready on a system file descriptor.

template<class Rep , class Period >
EventId	newTimer (ReactorHandlerInterface *handler, cn::duration< Rep, Period > duration, bool repeat)
	Add a timer in the reactor.

EventId	newWriteNotify (ReactorHandlerInterface *handler, SysSocketType sock)
	Add in the reactor a notification of write-ready or read-completion on a system file descriptor.

bool	open ()
	Open and initialize the Reactor.

bool	processEventLoop ()
	Process events until exit is requested.

virtual Report &	report () const override
	Access the Report which is associated with this object.

Report *	setReport (Report *report)
	Associate this object with another Report to log errors.

ReporterBase *	setReport (ReporterBase *delegate)
	Associate this object with another ReporterBase to log errors.

bool	signalBroadcastEvent (int error_code=SYS_SUCCESS, const ObjectPtr &user_data=nullptr)
	Signal a broadcast event in the reactor.

bool	signalEvent (EventId id)
	Signal a user event in the reactor.

template<class... Args>
void	trace (const UChar *fmt, Args &&... args)
	Issue a low-level trace message if environment variable TS_REACTOR_TRACE is defined with a non-empty value.

Static Public Member Functions
static int	SilentLevel (bool silent, int default_severity=Severity::Error)
	Compute a log severity level from a "silent" parameter.

Detailed Description

Event-driven reactor class implementing the "event loop" pattern.

A reactor is a single-threaded design pattern based on an "event loop". The application classes register handlers to be called when "events" occur in the future. The application-defined handlers typically start other background tasks (timers, I/O) and register other handlers to be called when these tasks complete.

Timers, process termination, and user-defined events (which can be triggered from other tasks) are directly handled by the class Reactor. Other types of features such as message queues, worker threads, and input/output are handled in "reactive classes" which execute on top of the Reactor.

I/O multiplexing

There are two distinct reactive I/O models, with different implementations:

Non-blocking I/O (UNIX systems)
- kqueue (macOS, FreeBSD)
- epoll (Linux)
Asynchronous I/O (Windows)
- I/O completion ports (aka IOCP)

See the documentation of class NonBlockingDevice for a detailed explanation of the differences.

The class Reactor encapsulates the various implementations and proposes a portable interface. However, while it is possible to unify the various types of non-blocking I/O (kqueue and epoll) in one single interface, it is impossible to unify non-blocking I/O and asynchronous I/O into the same interface. The way they shall be used, as well as the way the data buffers are managed, are too different. Therefore, the class Reactor exposes interfaces for both models. The application shall check the current I/O model using the "consteval" static methods UseNonBlockingIO() and UseAsynchronousIO() and then adopt the correct strategy.

In practice, the I/O multiplexing features of the class Reactor are not used by applications. They are used in a few specialized "reactive I/O" classes such as ReactiveUDPSocket or ReactiveTCPConnection. These classes are implemented on top of Reactor and have fully portable and homogeneous interfaces.

Synchronisation

The Reactor shall be used from one single thread. Unless specified otherwise, all methods shall be invoked from the thread of the reactor, where the event loop is run. All handlers are invoked in the context of the thread which invoked processEventLoop(), except worker handlers which are invoked in the context of a worker thread. Therefore, worker handlers are not allowed to call methods of the reactor.

Constructor & Destructor Documentation

◆ Reactor() [1/2]

ts::Reactor::Reactor ( Report * report )

explicit

Constructor.

Parameters

[in] report Where to report errors. The report object must remain valid as long as this object exists or setReport() is used with another Report object. If report is null, log messages are discarded.

◆ Reactor() [2/2]

ts::Reactor::Reactor ( ReporterBase * delegate )

explicit

Constructor.

Parameters

[in] delegate Use the report of another ReporterBase. If delegate is null, log messages are discarded.

◆ ~Reactor()

virtual ts::Reactor::~Reactor ( )

overridevirtual

Destructor.

All resources, such as pending timers or user events, are deleted.

Member Function Documentation

◆ open()

bool ts::Reactor::open ( )

Open and initialize the Reactor.

Must be invoked before running the event loop or registering events.

Returns: True on success, false on error.

◆ close()

bool ts::Reactor::close ( bool silent = false )

Close the Reactor.

Must not be called from within the event loop. Automatically done in the destructor.

Parameters

[in] silent If true, do not report errors through the logger.

Returns: True on success, false on error.

◆ isOpen()

bool ts::Reactor::isOpen ( ) const

inline

Check if the Reactor is open.

Returns: True if the Reactor is open, false otherwise.

◆ isActiveEvent()

bool ts::Reactor::isActiveEvent ( EventId id )

Check if an event (user-defined, timer, I/O) is still active in the reactor.

Parameters

[in] id Envent id to check.

Returns: True if id is still an active event, false if it has been deleted, completed, canceled.

◆ processEventLoop()

bool ts::Reactor::processEventLoop ( )

Process events until exit is requested.

Returns: The status from exitEventLoop(). If exitEventLoop() is invoked several times, the returned value is false if exitEventLoop() has been called at least once with success being false.

◆ exitEventLoop()

void ts::Reactor::exitEventLoop ( bool success = true )

Exit processEventLoop() as soon as possible.

This method is typically invoked from a handler. If it is invoked before processEventLoop(), then processEventLoop() exits immediately. The "exit request" condition is reset only when processEventLoop() returns.

Parameters

[in] success The value that processEventLoop() shall return.

◆ addExitReference()

int ts::Reactor::addExitReference ( )

Coordinate future exitEventLoop().

When several participants share the same reactor, we may want to exit the event loop when all participants are properly terminated. In that case, it is difficult for a participant to decide how to call exitEventLoop(). To coordinate a proper exitEventLoop() when all participants are terminated, each of them calls addExitReference() once on initialization, either inside the event loop or even before entering the event loop. When a participant considers itself as terminated, it calls freeExitReference(). When the number of references falls to zero, exitEventLoop() is automatically called.

Returns: The number of current references. This is informational only.

See also: freeExitReference(); exitEventLoop()

◆ freeExitReference()

int ts::Reactor::freeExitReference ( bool success = true )

Coordinated exitEventLoop().

This function shall be called once for each addExitReference(). It decrements the reference counter of participants.

Parameters

[in] success The value that processEventLoop() shall return. Each participant sets its own value. If at least one participant sets success to false, processEventLoop() will return false. If all participants set success to true, processEventLoop() will return true.

Returns: The number of remaining references. This is informational only. If the returned value is zero or negative, the application knows that exitEventLoop() has been called.

See also: addExitReference(); exitEventLoop()

◆ newTimer()

template<class Rep , class Period >

EventId ts::Reactor::newTimer	(	ReactorHandlerInterface *	handler,
		cn::duration< Rep, Period >	duration,
		bool	repeat
	)

inline

Add a timer in the reactor.

If the timer is one-shot (the default), the timer is automatically deleted after its expiration, after the handler is invoked. If the timer is repeated, it shall be explicitly canceled if needed.

Parameters

[in]	handler	Address of a handler to call when the timer expires. Return an error if set as `nullptr`.
[in]	duration	Duration of the timer. Must be strictly positive.
[in]	repeat	If true, the timer is repeated at regular intervals, every duration, until canceled.

Returns: The identity of the timer. Invalid in case of error.

◆ cancelTimer()

bool ts::Reactor::cancelTimer	(	EventId	id,
		bool	silent = `false`
	)

Cancel a timer.

Parameters

[in]	id	Timer to cancel.
[in]	silent	If true, do not report errors through the logger.

Returns: True on success, false on error.

◆ newEvent()

EventId ts::Reactor::newEvent ( ReactorHandlerInterface * handler )

Add a user event in the reactor.

Parameters

[in] handler Address of a handler to call when the event is signalled. Return an error if set as nullptr.

Returns: The identity of the user event. Invalid in case of error.

◆ signalEvent()

bool ts::Reactor::signalEvent ( EventId id )

Signal a user event in the reactor.

As an exception to the single-thread-reactor rule, this method can be invoked from any thread.

Parameters

[in] id Event to signal.

Returns: True on success, false on error.

◆ deleteEvent()

bool ts::Reactor::deleteEvent	(	EventId	id,
		bool	silent = `false`
	)

Delete a user event.

Parameters

[in]	id	Event to delete.
[in]	silent	If true, do not report errors through the logger.

Returns: True on success, false on error.

◆ signalBroadcastEvent()

bool ts::Reactor::signalBroadcastEvent	(	int	error_code = `SYS_SUCCESS`,
		const ObjectPtr &	user_data = `nullptr`
	)

Signal a broadcast event in the reactor.

A broadcast event is sent to all currently registered handlers in the reactor. If the same handler is registered for several events (e.g. two timers, one user event and one I/O), the handler is called only once per broadcast event.

A broadcast event is typically used to wake all objects which are waiting for something and signal to them that the application is terminating, for instance, and they should cleanup and close.

Parameters

[in]	error_code	Application-specific error code, passed to the handler.
[in]	user_data	Application-specific user-data shared pointer, passed to the handler.

Returns: True on success, false on error.

◆ newProcessIdTermination()

EventId ts::Reactor::newProcessIdTermination	(	ReactorHandlerInterface *	handler,
		SysProcessIdType	pid
	)

Add a process termination event in the reactor, using a process id.

Parameters

[in]	handler	Address of a handler to call when the process terminates. Return an error if set as `nullptr`. The handler is invoked when the process is no longer there. It is possible that the process was already terminated for a while. It is even possible that the process never really started. In all cases, the handler is invoked when the watched process doesn't exist, whether it is terminated or has never existed.
[in]	pid	Process id to watch. If that process id is unknown, it is assumed that the process is already terminated and the handler will be called as soon as possible.

Returns: The identity of the reactor event. Invalid in case of error.

◆ newProcessHandleTermination()

EventId ts::Reactor::newProcessHandleTermination	(	ReactorHandlerInterface *	handler,
		SysHandleType	process_handle
	)

Add a process termination event in the reactor, using a process handle.

Note: Only Windows identifies a process by handle, in addition to process id. On other operating systems, this method always reports an error.

Parameters

[in]	handler	Address of a handler to call when the process terminates. Return an error if set as `nullptr`. The handler is invoked when the process is no longer there. It is possible that the process was already terminated for a while. It is even possible that the process never really started. In all cases, the handler is invoked when the watched process doesn't exist, whether it is terminated or has never existed.
[in]	process_handle	Process handle to watch.

Returns: The identity of the reactor event. Invalid in case of error.

◆ cancelProcessTermination()

bool ts::Reactor::cancelProcessTermination	(	EventId	id,
		bool	silent = `false`
	)

Cancel a process termination event.

Parameters

[in]	id	Event to delete.
[in]	silent	If true, do not report errors through the logger.

Returns: True on success, false on error.

◆ newAsynchronousIO()

EventId ts::Reactor::newAsynchronousIO	(	ReactorHandlerInterface *	handler,
		SysSocketType	sock
	)

Add in the reactor a notification of asynchronous I/O on a system file descriptor or handle.

This method is normally never used in applications. It is used only by "reactive I/O classes", in the asynchronous I/O model.

Parameters

[in]	handler	Address of a handler to call when any asynchronous I/O completes on the specified system handle. Return an error if set as `nullptr`.
[in]	sock	A system-specific file descriptor or handle. This can be a socket or something else.

Returns: The identity of the asynchronous I/O. Invalid in case of error.

◆ cancelAsynchronousIO()

bool ts::Reactor::cancelAsynchronousIO	(	EventId	id,
		bool	silent = `false`
	)

Cancel all pending asynchronous I/O on a system file descriptor or handle.

This method is normally never used in applications. It is used only by "reactive I/O classes", in the asynchronous I/O model.

Parameters

[in]	id	Asynchronous I/O id, as previously returned by newAsynchronousIO().
[in]	silent	If true, do not report errors through the logger.

Returns: True on success, false on error.

IMPORTANT: After canceling all asynchronous I/O, the application shall wait for the reception of all I/O completions (presumably with an error status) before releasing the data buffers.

See also: NonBlockingDevice

◆ cancelAndWaitAsynchronousIO()

bool ts::Reactor::cancelAndWaitAsynchronousIO	(	EventId	id,
		NonBlockingDevice::IOSB &	iosb,
		bool	silent = `false`
	)

Cancel one specific pending asynchronous I/O.

This method is normally never used in applications. It is used only by "reactive I/O classes", in the asynchronous I/O model. Warning: This is a blocking call. It shall be used in case of trouble only.

Parameters

[in]	id	Asynchronous I/O id, as previously returned by newAsynchronousIO().
[in,out]	iosb	The asynchronous I/O status block.
[in]	silent	If true, do not report errors through the logger.

Returns: True on success, false on error.

◆ deleteAsynchronousIO()

bool ts::Reactor::deleteAsynchronousIO	(	EventId	id,
		bool	silent = `false`
	)

Delete a notification of asynchronous I/O.

This method is normally never used in applications. It is used only by "reactive I/O classes", in the asynchronous I/O model.

Parameters

[in]	id	Asynchronous I/O id, as previously returned by newAsynchronousIO().
[in]	silent	If true, do not report errors through the logger.

Returns: True on success, false on error.

◆ newReadNotify()

EventId ts::Reactor::newReadNotify	(	ReactorHandlerInterface *	handler,
		SysSocketType	sock
	)

Add in the reactor a notification of read-ready on a system file descriptor.

This method is normally never used in applications. It is used only by "reactive I/O classes", in the non-blocking I/O model.

Parameters

[in]	handler	Address of a handler to call when the operation is ready. Return an error if set as `nullptr`.
[in]	sock	A system-specific file descriptor or handle. This can be a socket or something else.

Returns: The identity of the user event. Invalid in case of error.

◆ deleteReadNotify()

bool ts::Reactor::deleteReadNotify	(	EventId	id,
		bool	silent = `false`
	)

Delete a notification of read-ready or read-completion.

This method is normally never used in applications. It is used only by "reactive I/O classes", in the non-blocking I/O model.

Parameters

[in]	id	Event to delete.
[in]	silent	If true, do not report errors through the logger.

Returns: True on success, false on error.

◆ newWriteNotify()

EventId ts::Reactor::newWriteNotify	(	ReactorHandlerInterface *	handler,
		SysSocketType	sock
	)

Add in the reactor a notification of write-ready or read-completion on a system file descriptor.

This method is normally never used in applications. It is used only by "reactive I/O classes", in the non-blocking I/O model.

Parameters

[in]	handler	Address of a handler to call when the operation is ready. Return an error if set as `nullptr`.
[in]	sock	A system-specific file descriptor or handle. This can be a socket or something else.

Returns: The identity of the user event. Invalid in case of error.

◆ deleteWriteNotify()

bool ts::Reactor::deleteWriteNotify	(	EventId	id,
		bool	silent = `false`
	)

Delete a notification of write-ready or write-completion.

This method is normally never used in applications. It is used only by "reactive I/O classes", in the non-blocking I/O model.

Parameters

[in]	id	Event to delete.
[in]	silent	If true, do not report errors through the logger.

Returns: True on success, false on error.

◆ trace()

template<class... Args>

void ts::Reactor::trace	(	const UChar *	fmt,
		Args &&...	args
	)

inline

Issue a low-level trace message if environment variable TS_REACTOR_TRACE is defined with a non-empty value.

This is typically use to troubleshoot the internals of a Reactor on a given platform.

Parameters

[in]	fmt	Format string with embedded '%' sequences.
[in]	args	List of arguments to substitute in the format string.

See also: UString::format()

◆ report()

virtual Report & ts::ReporterBase::report ( ) const

overridevirtualinherited

Access the Report which is associated with this object.

Can be called from another thread only if the Report object is thread-safe.

Returns: A reference to the associated report.

Implements ts::ReporterInterface.

◆ setReport() [1/2]

Report * ts::ReporterBase::setReport ( Report * report )

inherited

Associate this object with another Report to log errors.

Parameters

[in] report Where to report errors. The report object must remain valid as long as this object exists or setReport() is used with another Report object. If report is null, log messages are discarded.

Returns: The address of the previous Report object or a null pointer if there was none.

◆ setReport() [2/2]

ReporterBase * ts::ReporterBase::setReport ( ReporterBase * delegate )

inherited

Associate this object with another ReporterBase to log errors.

Parameters

[in] delegate Use the report of another ReporterBase. If delegate is null, the previous explicit Report is used..

Returns: The address of the previous ReporterBase object or a null pointer if there was none.

◆ muteReport()

bool ts::ReporterBase::muteReport ( bool mute )

inherited

Temporarily mute the associated report.

Parameters

[in] mute It true, report() will return a null report (log messages are discarded), until muteReport() is invoked again with mute set to false.

Returns: Previous state of the mute field.

◆ SilentLevel()

static int ts::ReporterBase::SilentLevel	(	bool	silent,
		int	default_severity = `Severity::Error`
	)

inlinestaticinherited

Compute a log severity level from a "silent" parameter.

Some subclass methods have a "silent" parameter to avoid reporting errors which may be insignificant, typically when closing a device after an error, in which case the close operation may produce other errors if the previous error left the device in an inconsistent state. While those errors should not be displayed as errors, we still display them at debug level.

Parameters

[in]	silent	If true, do not report errors, report debug messages instead.
[in]	default_severity	Default severity, in non-silent mode (error by default).

Returns: Error when silent is false, Debug otherwise.

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

tsReactor.h

Public Member Functions

Static Public Member Functions

Detailed Description

I/O multiplexing

Synchronisation

Constructor & Destructor Documentation

◆ Reactor() [1/2]

◆ Reactor() [2/2]

◆ ~Reactor()

Member Function Documentation

◆ open()

◆ close()

◆ isOpen()

◆ isActiveEvent()

◆ processEventLoop()

◆ exitEventLoop()

◆ addExitReference()

◆ freeExitReference()

◆ newTimer()

◆ cancelTimer()

◆ newEvent()

◆ signalEvent()

◆ deleteEvent()

◆ signalBroadcastEvent()

◆ newProcessIdTermination()

◆ newProcessHandleTermination()

◆ cancelProcessTermination()

◆ newAsynchronousIO()

◆ cancelAsynchronousIO()

◆ cancelAndWaitAsynchronousIO()

◆ deleteAsynchronousIO()

◆ newReadNotify()

◆ deleteReadNotify()

◆ newWriteNotify()

◆ deleteWriteNotify()

◆ trace()

◆ report()

◆ setReport() [1/2]

◆ setReport() [2/2]

◆ muteReport()

◆ SilentLevel()