Concurrency::_Worker_proxy< _Index_type > Class Template Reference

#include <ppl.h>

Public Member Functions
	_Worker_proxy (_Worker_proxy *_PParent_worker=NULL)
	~_Worker_proxy ()
bool	_Receive_range (_Range< _Index_type > *_Helper_range)
bool	_Send_range (_Range< _Index_type > *_Worker_range)
void	_Enable_intrusive_steal (_Range< _Index_type > *_Worker_range)
void	_Disable_intrusive_steal ()
bool	_Is_helper_registered ()
bool	_Is_done ()
void	_Set_done ()
void	_Set_tree_done ()
bool	_Is_beacon_signaled ()
bool	_Verify_beacon_cancellation ()

Private Member Functions
void	_Wait_on_intrusive_steal ()
void	_NotifyCancel ()
void	_Propagate_cancel ()
_Worker_proxy const &	operator= (_Worker_proxy const &)

Private Attributes
_Range< _Index_type > *volatile	_M_pHelper_range
_Worker_proxy *	_M_pParent_worker
::Concurrency::details::_Cancellation_beacon	_M_beacon
::Concurrency::details::_Context	_M_context
volatile long	_M_completion_count
_Range< _Index_type > *volatile	_M_pWorker_range
volatile long	_M_stop_iterating

Static Private Attributes
static const long	_Tree_Complete = 2

Constructor & Destructor Documentation

template<typename _Index_type>

Concurrency::_Worker_proxy< _Index_type >::_Worker_proxy ( _Worker_proxy< _Index_type > *  _PParent_worker = NULL )

inline

                                                         :
      _M_pHelper_range(NULL), _M_pParent_worker(_PParent_worker), _M_pWorker_range(NULL), _M_completion_count(0), _M_stop_iterating(0)
    {
        _M_context = ::Concurrency::details::_Context::_CurrentContext();
    }

template<typename _Index_type>

Concurrency::_Worker_proxy< _Index_type >::~_Worker_proxy ( )

inline

    {
        // Make the check to avoid doing extra work in the non-exceptional cases
        if (_M_completion_count != _Tree_Complete)
        {
            // On exception, notify our parent so it breaks out of its loop.
            _Propagate_cancel();

            // On exception, we need to set _M_completion_count to ensure that the helper breaks out of its spin wait.
            _Set_done();
        }
    }

Member Function Documentation

template<typename _Index_type>

void Concurrency::_Worker_proxy< _Index_type >::_Disable_intrusive_steal ( )

inline

    {
        _M_pWorker_range = NULL;
        _Wait_on_intrusive_steal();
    }

template<typename _Index_type>

void Concurrency::_Worker_proxy< _Index_type >::_Enable_intrusive_steal ( _Range< _Index_type > *  _Worker_range )

inline

    {
        _M_pWorker_range = _Worker_range;
    }

template<typename _Index_type>

bool Concurrency::_Worker_proxy< _Index_type >::_Is_beacon_signaled ( )

inline

    {
        return _M_beacon._Is_signaled();
    }

template<typename _Index_type>

bool Concurrency::_Worker_proxy< _Index_type >::_Is_done ( )

inline

    {
        return (_M_completion_count != 0);
    }

template<typename _Index_type>

bool Concurrency::_Worker_proxy< _Index_type >::_Is_helper_registered ( )

inline

    {
        return (_M_pHelper_range != NULL);
    }

template<typename _Index_type>

void Concurrency::_Worker_proxy< _Index_type >::_NotifyCancel ( )

inlineprivate

    {
        _M_beacon._Raise();
    }

template<typename _Index_type>

void Concurrency::_Worker_proxy< _Index_type >::_Propagate_cancel ( )

inlineprivate

    {
        if (_M_pParent_worker != NULL)
        {
            _M_pParent_worker->_NotifyCancel();
        }
    }

template<typename _Index_type>

bool Concurrency::_Worker_proxy< _Index_type >::_Receive_range ( _Range< _Index_type > *  _Helper_range )

inline

    {
        // If the worker already finished, then there is no work left for the helper
        if (_M_completion_count)
        {
            return false;
        }

        _CONCRT_ASSERT(_Helper_range != NULL);

        // There are two special values for _M_current_iteration that are not valid: one is the
        // initial value of the working class which it will never share, and the other is
        // the last exclusive iteration of the working class, which has no work to be done.
        // We use the former value so that we can understand worker's response.
        _Index_type _Cached_first_iteration = _Helper_range->_M_current_iteration;

        // Following operation is not done via interlocked operation because it does not have to.
        // Helper lazily registers that it would like to help the worker, but it allows for some
        // time to elapse before that information has made it over to the worker. The idea
        // is not to disturb the worker if it is not necessary. It is possible to add interlocked
        // operation in the future if the time spent in the busy wait loop is too big.
        _CONCRT_ASSERT(_M_pHelper_range == NULL);
        _M_pHelper_range = _Helper_range;

        ::Concurrency::details::_SpinWaitBackoffNone spinWait(::Concurrency::details::_Context::_Yield);

        // If the worker is done, it will flush the store buffer and signal the helper by
        // changing _M_current_iteration in the helper's range.
        while ((_Helper_range->_M_current_iteration == _Cached_first_iteration) && !_M_completion_count)
        {
            if ((_M_pWorker_range != NULL) && _M_context._IsSynchronouslyBlocked())
            {
                // Attempt to steal the entire range from the worker if it is synchronously blocked.

                // Make sure that worker makes no forward progress while helper is attempting to
                // steal its range. If worker does get unblocked, simply back off in the helper.
                // Note that there could be another helper running if a range has already been
                // sent to us.
                long _Stop_iterating = _InterlockedIncrement(&_M_stop_iterating);
                _CONCRT_ASSERT(_Stop_iterating > 0);

                // We need to make a local copy as the pointer could be changed by the worker.
                _Range<_Index_type> * _Worker_range = _M_pWorker_range;

                // The order of comparison needs to be preserved. If the parent is blocked, then
                // it cannot send a range (because _M_stop_iterating is already set). If it sent a range
                // before being synchronously blocked, then we are no longer the helper. Refrain
                // from intrusively stealing the range.
                if ((_Worker_range != NULL) && _M_context._IsSynchronouslyBlocked()
                    && (_Helper_range->_M_current_iteration == _Cached_first_iteration) && !_M_completion_count)
                {
                    _CONCRT_ASSERT(_M_pHelper_range == _Helper_range);

                    _M_pHelper_range = NULL;
                    _Worker_range->_Steal_range(_Helper_range);

                    _CONCRT_ASSERT(_Helper_range->_M_current_iteration != _Cached_first_iteration);
                }

                // At this point, worker is either:
                //
                // a) no longer blocked so range will come to the helper naturally, or
                // b) out of iterations because helper stole all of it
                _Stop_iterating = _InterlockedDecrement(&_M_stop_iterating);
                _CONCRT_ASSERT(_Stop_iterating >= 0);
            }
            else
            {
                // If there is no work received in a full spin, then start yielding the context
                spinWait._SpinOnce();
            }
        }

        // If the initial iteration is the same as the original first iteration then the
        // worker class is sending the signal that it does not need any help.
        if (_Helper_range->_M_current_iteration == _Cached_first_iteration)
        {
            return false;
        }

        return (_Helper_range->_Number_of_iterations() > 0);
    }

template<typename _Index_type>

bool Concurrency::_Worker_proxy< _Index_type >::_Send_range ( _Range< _Index_type > *  _Worker_range )

inline

    {
        // Worker range shall not be available for stealing at this time.
        _CONCRT_ASSERT(_M_pWorker_range == NULL);

        // Helper shall be registered.
        _CONCRT_ASSERT(_M_pHelper_range != NULL);

        // Send the range
        _Worker_range->_Send_range(_M_pHelper_range);

        // Notify the helper. The fence ensures that the prior updates are visible.
        _InterlockedExchangePointer(reinterpret_cast<void * volatile *>(&_M_pHelper_range), NULL);

        // The current iteration should still be left
        _CONCRT_ASSERT(_Worker_range->_Number_of_iterations() >= 1);

        // Indicate if we need another helper
        return (_Worker_range->_Number_of_iterations() > 1);
    }

template<typename _Index_type>

void Concurrency::_Worker_proxy< _Index_type >::_Set_done ( )

inline

    {
        // Let the helper know that this class is done with work and flush the store buffer. This operation
        // ensures that any buffered store to helper range in _Send_range is flushed and
        // available in _Receive_range (so there will be no lost ranges).
        _InterlockedExchange(&_M_completion_count, 1);
    }

template<typename _Index_type>

void Concurrency::_Worker_proxy< _Index_type >::_Set_tree_done ( )

inline

    {
        // Make sure that **WE** know when our destructor hits that the entire tree is complete.
        _M_completion_count = _Tree_Complete;
    }

template<typename _Index_type>

bool Concurrency::_Worker_proxy< _Index_type >::_Verify_beacon_cancellation ( )

inline

    {
        return _M_beacon._Confirm_cancel();
    }

template<typename _Index_type>

void Concurrency::_Worker_proxy< _Index_type >::_Wait_on_intrusive_steal ( )

inlineprivate

    {
        // This code is used to synchronize with helper in case of worker cooperative blocking.
        if (_M_stop_iterating != 0)
        {
            ::Concurrency::details::_SpinWaitBackoffNone spinWait;

            while (_M_stop_iterating != 0)
            {
                spinWait._SpinOnce();
            }
        }
    }

template<typename _Index_type>

_Worker_proxy const& Concurrency::_Worker_proxy< _Index_type >::operator= ( _Worker_proxy< _Index_type > const &  )

private

Member Data Documentation

template<typename _Index_type>

::Concurrency::details::_Cancellation_beacon Concurrency::_Worker_proxy< _Index_type >::_M_beacon

private

template<typename _Index_type>

volatile long Concurrency::_Worker_proxy< _Index_type >::_M_completion_count

private

template<typename _Index_type>

::Concurrency::details::_Context Concurrency::_Worker_proxy< _Index_type >::_M_context

private

template<typename _Index_type>

_Range<_Index_type>* volatile Concurrency::_Worker_proxy< _Index_type >::_M_pHelper_range

private

template<typename _Index_type>

_Worker_proxy* Concurrency::_Worker_proxy< _Index_type >::_M_pParent_worker

private

template<typename _Index_type>

_Range<_Index_type>* volatile Concurrency::_Worker_proxy< _Index_type >::_M_pWorker_range

private

template<typename _Index_type>

volatile long Concurrency::_Worker_proxy< _Index_type >::_M_stop_iterating

private

template<typename _Index_type>

const long Concurrency::_Worker_proxy< _Index_type >::_Tree_Complete = 2

staticprivate

---

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

• VS2017/inc/ppl.h