Concurrency::details::_Aggregator< _Operation_type, _Handler_type > Class Template Reference

An aggregator for collecting operations coming from multiple sources and executing them serially on a single thread. _Operation_type must be derived from _Aggregated_operation. The parameter _Handler_type is a functor that will be passed the list of operations and is expected to handle each operation appropriately, setting the status of each operation to non-zero. More...

#include <concurrent_priority_queue.h>

Public Member Functions
	_Aggregator ()
	~_Aggregator ()
void	_Initialize_handler (_Handler_type _Handler)
void	_Execute (_Operation_type *_Op)
	Place operation in list and either handle list or wait for operation to complete. More...

Private Member Functions
void	_Start_handle_operations ()

Private Attributes
_Operation_type *volatile	_M_pPending_operations
volatile long	_M_handler_busy
_Handler_type	_M_handle_operations

Detailed Description

template<typename _Operation_type, typename _Handler_type>
class Concurrency::details::_Aggregator< _Operation_type, _Handler_type >

An aggregator for collecting operations coming from multiple sources and executing them serially on a single thread. _Operation_type must be derived from _Aggregated_operation. The parameter _Handler_type is a functor that will be passed the list of operations and is expected to handle each operation appropriately, setting the status of each operation to non-zero.

Constructor & Destructor Documentation

template<typename _Operation_type, typename _Handler_type>

Concurrency::details::_Aggregator< _Operation_type, _Handler_type >::_Aggregator ( )

inline

                      : _M_handler_busy(0)
        {
            _M_pPending_operations = NULL;
        }

template<typename _Operation_type, typename _Handler_type>

Concurrency::details::_Aggregator< _Operation_type, _Handler_type >::~_Aggregator ( )

inline

        {
        }

Member Function Documentation

template<typename _Operation_type, typename _Handler_type>

void Concurrency::details::_Aggregator< _Operation_type, _Handler_type >::_Execute ( _Operation_type *  _Op )

inline

Place operation in list and either handle list or wait for operation to complete.

        {
            _Operation_type *_Res;

            // Insert the operation in the queue
            do
            {
                _Op->_M_pNext = _Res = _M_pPending_operations;
            } while (_InterlockedCompareExchangePointer(reinterpret_cast<void *volatile *>(&_M_pPending_operations), _Op, _Res) != _Res);

            if (_Res == NULL)
            {
                // This item was the first one in the list; this thread will handle the operations. Note that by the time the
                // thread pops the list of operations to handle, there may be several operations queued up.
                _Start_handle_operations();
                _CONCRT_ASSERT(_Op->_M_status != 0);
            }
            else
            {
                // This item was not the first one on the list; a different thread is going to handle this operation, wait for
                // the result to be ready.
                ::Concurrency::details::_SpinWaitBackoffNone _SpinWait;
                while (_Op->_M_status == 0)
                {
                    _SpinWait._SpinOnce();
                }
            }
        }

template<typename _Operation_type, typename _Handler_type>

void Concurrency::details::_Aggregator< _Operation_type, _Handler_type >::_Initialize_handler ( _Handler_type  _Handler )

inline

        {
            _M_handle_operations = _Handler;
        }

template<typename _Operation_type, typename _Handler_type>

void Concurrency::details::_Aggregator< _Operation_type, _Handler_type >::_Start_handle_operations ( )

inlineprivate

        {
            _Operation_type * _POp_list;

            // Acquire the _M_handler_busy flag. Only one thread can possibly spin here at a time
            ::Concurrency::details::_SpinWaitBackoffNone _SpinWait;
            while (_M_handler_busy == 1)
            {
                _SpinWait._SpinOnce();
            }

            long _OldVal = _InterlockedExchange(&_M_handler_busy, 1);
            _CONCRT_ASSERT(_OldVal == 0);

            // Get the list of pending operations
            _POp_list = reinterpret_cast<_Operation_type *>(_InterlockedExchangePointer(reinterpret_cast<void * volatile *>(&_M_pPending_operations), NULL));

            // Handle all the operations
            _M_handle_operations(_POp_list);

            // Release the handler
            _OldVal = _InterlockedExchange(&_M_handler_busy, 0);
            _CONCRT_ASSERT(_OldVal == 1);
        }

Member Data Documentation

template<typename _Operation_type, typename _Handler_type>

_Handler_type Concurrency::details::_Aggregator< _Operation_type, _Handler_type >::_M_handle_operations

private

template<typename _Operation_type, typename _Handler_type>

volatile long Concurrency::details::_Aggregator< _Operation_type, _Handler_type >::_M_handler_busy

private

template<typename _Operation_type, typename _Handler_type>

_Operation_type* volatile Concurrency::details::_Aggregator< _Operation_type, _Handler_type >::_M_pPending_operations

private

---

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

• VS2017/inc/concurrent_priority_queue.h