Concurrency::ordered_message_processor< _Type > Class Template Reference

An ordered_message_processor is a message_processor that allows message blocks to process messages in the order they were received. More...

#include <agents.h>

Inheritance diagram for Concurrency::ordered_message_processor< _Type >:

Public Types
typedef std::tr1::function< void(message< _Type > *)>	_Handler_method
	The signature of the callback method invoked while processing messages. More...
typedef std::tr1::function< void(void)>	_Propagator_method
	The signature of the callback method invoked while propagating messages. More...
typedef _Type	type
	A type alias for _Type . More...
Public Types inherited from Concurrency::message_processor< _Type >
typedef _Type	type
	A type alias for _Type . More...

Public Member Functions
	ordered_message_processor ()
	Constructs an ordered_message_processor object. More...
virtual	~ordered_message_processor ()
	Destroys the ordered_message_processor object. More...
void	initialize (_Inout_opt_ Scheduler *_PScheduler, _Inout_opt_ ScheduleGroup *_PScheduleGroup, _Handler_method const &_Handler)
	Initializes the ordered_message_processor object with the appropriate callback function, scheduler and schedule group. More...
virtual void	initialize_batched_processing (_Handler_method const &_Processor, _Propagator_method const &_Propagator)
	Initialize batched message processing More...
virtual void	sync_send (_Inout_opt_ message< _Type > *_Msg)
	Synchronously queues up messages and starts a processing task, if this has not been done already. More...
virtual void	async_send (_Inout_opt_ message< _Type > *_Msg)
	Asynchronously queues up messages and starts a processing task, if this has not been done already. More...
virtual void	wait ()
	A processor-specific spin wait used in destructors of message blocks to make sure that all asynchronous processing tasks have time to finish before destroying the block. More...

Protected Member Functions
virtual void	process_incoming_message ()
	The processing function that is called asynchronously. It dequeues messages and begins processing them. More...

Private Member Functions
void	_Clear_queued_messages ()
void	_Sync_send_helper (message< _Type > *_Msg)
long	_Process_message_helper ()
void	_Invoke_handler (long _Count)
void	_Invoke_handler (message< _Type > *_Msg)

Private Attributes
concurrent_queue< message< _Type > * >	_M_queuedMessages
	A queue of the messages More...
::Concurrency::details::_NonReentrantPPLLock	_M_asyncSendLock
	A lock to use for queueing incoming messages. More...
volatile long	_M_queuedDataCount
	A count of the current number of messages to process. Used as a flag to see if a new process message task needs to be created. More...
Scheduler *	_M_pScheduler
	The scheduler to process messages on More...
ScheduleGroup *	_M_pScheduleGroup
	The schedule group to process messages on More...
volatile long	_M_stopProcessing
	A flag set in the destructor of a block to cease processing of new messages. This is required to guarantee that _M_queuedDataCount will get to 0 eventually. More...
volatile long	_M_lwtCount
	A counter to indicate the number of outstanding LWTs More...
_Handler_method	_M_handler
	A message handler object which exposes the callback to be invoked More...
_Handler_method	_M_processor
	A message processing object which exposes the callback to be invoked More...
_Propagator_method	_M_propagator
	A message propagating object which exposes the callback to be invoked More...

Additional Inherited Members
Static Protected Member Functions inherited from Concurrency::message_processor< _Type >
static void __cdecl	_Process_incoming_message_wrapper (void *_Data)
	Wrapper for process_incoming_message suitable for use as a argument to CreateThread and other similar methods. More...

Detailed Description

template<class _Type>
class Concurrency::ordered_message_processor< _Type >

An ordered_message_processor is a message_processor that allows message blocks to process messages in the order they were received.

Template Parameters

_Type

The payload type of messages handled by the processor.

Member Typedef Documentation

template<class _Type >

typedef std::tr1::function<void(message<_Type> *)> Concurrency::ordered_message_processor< _Type >::_Handler_method

The signature of the callback method invoked while processing messages.

template<class _Type >

typedef std::tr1::function<void(void)> Concurrency::ordered_message_processor< _Type >::_Propagator_method

The signature of the callback method invoked while propagating messages.

template<class _Type >

typedef _Type Concurrency::ordered_message_processor< _Type >::type

A type alias for _Type .

Constructor & Destructor Documentation

template<class _Type >

Concurrency::ordered_message_processor< _Type >::ordered_message_processor ( )

inline

Constructs an ordered_message_processor object.

This ordered_message_processor will not schedule asynchronous or synchronous handlers until the initialize function is called.

                                :
      _M_queuedDataCount(0),
      _M_stopProcessing(1),
      _M_lwtCount(0),
      _M_pScheduler(NULL),
      _M_pScheduleGroup(NULL),
      _M_handler(nullptr),
      _M_processor(nullptr),
      _M_propagator(nullptr)
    {
    }

template<class _Type >

virtual Concurrency::ordered_message_processor< _Type >::~ordered_message_processor ( )

inlinevirtual

Destroys the ordered_message_processor object.

Waits for all outstanding asynchronous operations before destroying the processor.

    {
        wait();
    }

Member Function Documentation

template<class _Type >

void Concurrency::ordered_message_processor< _Type >::_Clear_queued_messages ( )

inlineprivate

    {
        message<_Type> * _Msg = NULL;
        while (_M_queuedMessages.try_pop(_Msg))
        {
            delete _Msg;
        }
    }

template<class _Type >

void Concurrency::ordered_message_processor< _Type >::_Invoke_handler ( long  _Count )

inlineprivate

    {
        // Process _Count number of messages
        for(int _I = 0; _I < _Count; _I++)
        {
            message<_Type> * _Msg = NULL;
            _M_queuedMessages.try_pop(_Msg);
            if (_M_processor == NULL)
            {
                // If a processor function does not exist, the message processor is using single
                // message processing rather than batched processing.  There should also be no
                // propagator function defined in this case.
                _CONCRT_ASSERT(_M_propagator == NULL);
                _M_handler(_Msg);
            }
            else
            {
                // Use the batched message processing function
                _M_processor(_Msg);
            }
        }

        // Call the handler which propagates the message(s)
        if (_M_propagator != NULL)
        {
            _M_propagator();
        }
    }

template<class _Type >

void Concurrency::ordered_message_processor< _Type >::_Invoke_handler ( message< _Type > *  _Msg )

inlineprivate

    {
        if (_M_processor == NULL)
        {
            // If a processor function does not exist, the message processor is using single
            // message processing rather than batched processing.  There should also be no
            // propagator function defined in this case.
            _CONCRT_ASSERT(_M_propagator == NULL);
            _M_handler(_Msg);
        }
        else
        {
            // Use the batched message processing function
            _M_processor(_Msg);

            // Call the handler which propagates the message(s)
            if (_M_propagator != NULL)
            {
                _M_propagator();
            }
        }
    }

template<class _Type >

long Concurrency::ordered_message_processor< _Type >::_Process_message_helper ( )

inlineprivate

    {
        _NR_lock _Lock(_M_asyncSendLock);

        long _Messages_processed = 0;

        // Do batched processing of messages
        // Read off the number of messages to process in this iteration by snapping a count
        volatile long _Count = _M_queuedDataCount;
        bool _StopProcessing = false;

        // This count could be 0 if there was both a synchronous and asynchronous
        // send occuring.  One of them could have sent all of the messages for the other
        while (_Count > 0)
        {
            // Process _Count number of messages
            _Invoke_handler(_Count);
            _Messages_processed += _Count;

            // Subtract the count and see if there are new things to process
            volatile long _Orig = _InterlockedExchangeAdd((volatile long *) &_M_queuedDataCount, -_Count);
            _CONCRT_ASSERT(_Orig >= _Count);
            if (_Orig == _Count)
            {
                // Because _Count did not change, we processed everything there is to process
                break;
            }

            if (_StopProcessing)
            {
                break;
            }

            // After reading the flag process the currently queued messages
            // Any messages received after we observe this flag (to be set) will not
            // be processed.
            _StopProcessing = (_M_stopProcessing == 0) ? false : true;

            // Snap the count and try to process more
            _Count = _M_queuedDataCount;
        }

        return _Messages_processed;
    }

template<class _Type >

void Concurrency::ordered_message_processor< _Type >::_Sync_send_helper ( message< _Type > *  _Msg )

inlineprivate

    {
        _NR_lock _Lock(_M_asyncSendLock);

        // Message block destructors sets the _M_stopProcessing flag to stop
        // processing any more messages. This is required to guarantee
        // that the destructor's wait_for_async_sends will complete
        if (_M_stopProcessing == 0)
        {
            if (_M_queuedDataCount > 0)
            {
                long _Count = _InterlockedExchange((volatile long *) &_M_queuedDataCount, 0);
                _Invoke_handler(_Count);
            }

            _Invoke_handler(_Msg);
        }
        else
        {
            // Destructor is running. Do not process the message
            // Delete the msg, if any.
            if (_Msg != NULL)
            {
                delete _Msg;
            }
        }

    }

template<class _Type >

virtual void Concurrency::ordered_message_processor< _Type >::async_send ( _Inout_opt_ message< _Type > *  _Msg )

inlinevirtual

Asynchronously queues up messages and starts a processing task, if this has not been done already.

Parameters

_Msg	A pointer to a message.

Implements Concurrency::message_processor< _Type >.

    {
        if (_M_handler == NULL)
        {
            throw invalid_operation("async_send called without registering a callback");
        }

        //
        // If there is a message to send, enqueue it in the processing queue.
        // async_send can be sent a NULL message if the block wishes to reprocess
        // the messages that are in its queue.  For example, an unbounded_buffer
        // that has its head node released after reservation.
        //
        if (_Msg != NULL)
        {
            _M_queuedMessages.push(_Msg);
        }

        if (_InterlockedIncrement(&_M_queuedDataCount) == 1)
        {
            // Indicate that an LWT is in progress. This will cause the
            // destructor to block.
            _InterlockedIncrement(&_M_lwtCount);

            if (_M_stopProcessing == 0)
            {
                _CONCRT_ASSERT(_M_lwtCount > 0);

                _Trace_agents(AGENTS_EVENT_SCHEDULE, ::Concurrency::details::_Trace_agents_get_id(this));

                TaskProc _Proc = &::Concurrency::ordered_message_processor<_Type>::_Process_incoming_message_wrapper;
#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP
                if (_M_pScheduleGroup != NULL)
                {
                    _M_pScheduleGroup->ScheduleTask(_Proc, this);
                }
                else if (_M_pScheduler != NULL)
                {
                    _M_pScheduler->ScheduleTask(_Proc, this);
                }
                else
                {
#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */
                    ::Concurrency::details::_CurrentScheduler::_ScheduleTask(_Proc, this);
#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP
                }
#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */

                // The LWT will decrement _M_lwtCount.
                return;
            }

            // If we get here then no task was scheduled. Decrement LWT count to reflect this fact
            _InterlockedDecrement(&_M_lwtCount);
        }
    }

template<class _Type >

void Concurrency::ordered_message_processor< _Type >::initialize ( _Inout_opt_ Scheduler *  _PScheduler, _Inout_opt_ ScheduleGroup *  _PScheduleGroup, _Handler_method const &  _Handler  )

inline

Initializes the ordered_message_processor object with the appropriate callback function, scheduler and schedule group.

Parameters

_PScheduler	A pointer to the scheduler to be used for scheduling light-weight tasks.
_PScheduleGroup	A pointer to the schedule group to be used for scheduling light-weight tasks.
_Handler	The handler functor invoked during callback.

See also

Scheduler Class, ScheduleGroup Class

    {
        _M_pScheduler = _PScheduler;
        _M_pScheduleGroup = _PScheduleGroup;
        _M_handler = _Handler;
        _M_stopProcessing = 0;
    }

template<class _Type >

virtual void Concurrency::ordered_message_processor< _Type >::initialize_batched_processing ( _Handler_method const &  _Processor, _Propagator_method const &  _Propagator  )

inlinevirtual

Initialize batched message processing

Parameters

_Processor	The processor functor invoked during callback.
_Propagator	The propagator functor invoked during callback.

    {
        _M_processor = _Processor;
        _M_propagator = _Propagator;
    }

template<class _Type >

virtual void Concurrency::ordered_message_processor< _Type >::process_incoming_message ( )

inlineprotectedvirtual

The processing function that is called asynchronously. It dequeues messages and begins processing them.

Implements Concurrency::message_processor< _Type >.

    {
        _Trace_agents(AGENTS_EVENT_START, ::Concurrency::details::_Trace_agents_get_id(this));
        long _Count = _Process_message_helper();
        _Trace_agents(AGENTS_EVENT_END, ::Concurrency::details::_Trace_agents_get_id(this), _Count);

        // Indicate that an LWT completed
        _InterlockedDecrement(&_M_lwtCount);

        // Do not access any members here. If the count goes to
        // 0 as a result of the above decrement, the object
        // could be immediately deleted.
    }

template<class _Type >

virtual void Concurrency::ordered_message_processor< _Type >::sync_send ( _Inout_opt_ message< _Type > *  _Msg )

inlinevirtual

Synchronously queues up messages and starts a processing task, if this has not been done already.

Parameters

_Msg	A pointer to a message.

Implements Concurrency::message_processor< _Type >.

    {
        if (_M_handler == NULL)
        {
            throw invalid_operation("sync_send called without registering a callback");
        }

        _Sync_send_helper(_Msg);
    }

template<class _Type >

virtual void Concurrency::ordered_message_processor< _Type >::wait ( )

inlinevirtual

A processor-specific spin wait used in destructors of message blocks to make sure that all asynchronous processing tasks have time to finish before destroying the block.

Implements Concurrency::message_processor< _Type >.

    {
        // Cease processing of any new messages
        _InterlockedIncrement(&_M_stopProcessing);

        // This spin makes sure all previously initiated message processings
        // will still process correctly.  As soon as this count reaches zero, we can
        // procede with the message block destructor.
        ::Concurrency::details::_SpinWaitBackoffNone spinWait(::Concurrency::details::_Context::_Yield);
        while(_M_lwtCount != 0)
        {
            spinWait._SpinOnce();
        }

        // Synchronize with sync_send
        {
            _NR_lock _Lock(_M_asyncSendLock);
            _Clear_queued_messages();
        }

    }

Member Data Documentation

template<class _Type >

::Concurrency::details::_NonReentrantPPLLock Concurrency::ordered_message_processor< _Type >::_M_asyncSendLock

private

A lock to use for queueing incoming messages.

template<class _Type >

_Handler_method Concurrency::ordered_message_processor< _Type >::_M_handler

private

A message handler object which exposes the callback to be invoked

template<class _Type >

volatile long Concurrency::ordered_message_processor< _Type >::_M_lwtCount

private

A counter to indicate the number of outstanding LWTs

template<class _Type >

_Handler_method Concurrency::ordered_message_processor< _Type >::_M_processor

private

A message processing object which exposes the callback to be invoked

template<class _Type >

_Propagator_method Concurrency::ordered_message_processor< _Type >::_M_propagator

private

A message propagating object which exposes the callback to be invoked

template<class _Type >

ScheduleGroup* Concurrency::ordered_message_processor< _Type >::_M_pScheduleGroup

private

The schedule group to process messages on

template<class _Type >

Scheduler* Concurrency::ordered_message_processor< _Type >::_M_pScheduler

private

The scheduler to process messages on

template<class _Type >

volatile long Concurrency::ordered_message_processor< _Type >::_M_queuedDataCount

private

A count of the current number of messages to process. Used as a flag to see if a new process message task needs to be created.

template<class _Type >

concurrent_queue<message<_Type> *> Concurrency::ordered_message_processor< _Type >::_M_queuedMessages

private

A queue of the messages

template<class _Type >

volatile long Concurrency::ordered_message_processor< _Type >::_M_stopProcessing

private

A flag set in the destructor of a block to cease processing of new messages. This is required to guarantee that _M_queuedDataCount will get to 0 eventually.

---

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

• VS2013/inc/agents.h