Concurrency::_Join_node< _Type, _Destination_type, _Jtype > Class Template Reference

Defines a block allowing sources of distinct types to be joined. Join node is a single-target, multi-source ordered propagator block More...

#include <agents.h>

Inheritance diagram for Concurrency::_Join_node< _Type, _Destination_type, _Jtype >:

Public Member Functions
	_Join_node ()
	Constructs a join within the default scheduler, and places it on any schedule group of the scheduler's choosing. More...
	_Join_node (Scheduler &_PScheduler)
	Constructs a join within the specified scheduler, and places it on any schedule group of the scheduler's choosing. More...
	_Join_node (ScheduleGroup &_PScheduleGroup)
	Constructs a join within the specified schedule group. The scheduler is implied by the schedule group. More...
	~_Join_node ()
	Cleans up any resources that may have been created by the join. More...
Public Member Functions inherited from Concurrency::propagator_block< single_link_registry< ITarget< _Destination_type > >, multi_link_registry< ISource< size_t > > >
	propagator_block ()
	Constructs a propagator_block object. More...
virtual	~propagator_block ()
	Destroys a propagator_block object. More...
virtual message_status	propagate (_Inout_opt_ message< _Source_type > *_PMessage, _Inout_opt_ ISource< _Source_type > *_PSource)
	Asynchronously passes a message from a source block to this target block. More...
virtual message_status	send (_Inout_ message< _Source_type > *_PMessage, _Inout_ ISource< _Source_type > *_PSource)
	Synchronously initiates a message to this block. Called by an ISource block. When this function completes, the message will already have propagated into the block. More...
Public Member Functions inherited from Concurrency::source_block< single_link_registry< ITarget< _Destination_type > >, _MessageProcessorType >
	source_block ()
	Constructs a source_block object. More...
virtual	~source_block ()
	Destroys the source_block object. More...
virtual void	link_target (_Inout_ ITarget< _Target_type > *_PTarget)
	Links a target block to this source_block object. More...
virtual void	unlink_target (_Inout_ ITarget< _Target_type > *_PTarget)
	Unlinks a target block from this source_block object. More...
virtual void	unlink_targets ()
	Unlinks all target blocks from this source_block object. More...
virtual message< _Target_type > *	accept (runtime_object_identity _MsgId, _Inout_ ITarget< _Target_type > *_PTarget)
	Accepts a message that was offered by this source_block object, transferring ownership to the caller. More...
virtual bool	reserve (runtime_object_identity _MsgId, _Inout_ ITarget< _Target_type > *_PTarget)
	Reserves a message previously offered by this source_block object. More...
virtual message< _Target_type > *	consume (runtime_object_identity _MsgId, _Inout_ ITarget< _Target_type > *_PTarget)
	Consumes a message previously offered by this source_block object and successfully reserved by the target, transferring ownership to the caller. More...
virtual void	release (runtime_object_identity _MsgId, _Inout_ ITarget< _Target_type > *_PTarget)
	Releases a previous successful message reservation. More...
virtual void	acquire_ref (_Inout_ ITarget< _Target_type > *)
	Acquires a reference count on this source_block object, to prevent deletion. More...
virtual void	release_ref (_Inout_ ITarget< _Target_type > *_PTarget)
	Releases a reference count on this source_block object. More...
Public Member Functions inherited from Concurrency::ISource< single_link_registry< ITarget< _Destination_type > >::type::type >
virtual	~ISource ()
	Destroys the ISource object. More...
virtual void	link_target (_Inout_ ITarget< single_link_registry< ITarget< _Destination_type > >::type::type > *_PTarget)=0
	When overridden in a derived class, links a target block to this ISource block. More...
virtual void	unlink_target (_Inout_ ITarget< single_link_registry< ITarget< _Destination_type > >::type::type > *_PTarget)=0
	When overridden in a derived class, unlinks a target block from this ISource block, if found to be previously linked. More...
virtual message< single_link_registry< ITarget< _Destination_type > >::type::type > *	accept (runtime_object_identity _MsgId, _Inout_ ITarget< single_link_registry< ITarget< _Destination_type > >::type::type > *_PTarget)=0
	When overridden in a derived class, accepts a message that was offered by this ISource block, transferring ownership to the caller. More...
virtual bool	reserve (runtime_object_identity _MsgId, _Inout_ ITarget< single_link_registry< ITarget< _Destination_type > >::type::type > *_PTarget)=0
	When overridden in a derived class, reserves a message previously offered by this ISource block. More...
virtual message< single_link_registry< ITarget< _Destination_type > >::type::type > *	consume (runtime_object_identity _MsgId, _Inout_ ITarget< single_link_registry< ITarget< _Destination_type > >::type::type > *_PTarget)=0
	When overridden in a derived class, consumes a message previously offered by this ISource block and successfully reserved by the target, transferring ownership to the caller. More...
virtual void	release (runtime_object_identity _MsgId, _Inout_ ITarget< single_link_registry< ITarget< _Destination_type > >::type::type > *_PTarget)=0
	When overridden in a derived class, releases a previous successful message reservation. More...
virtual void	acquire_ref (_Inout_ ITarget< single_link_registry< ITarget< _Destination_type > >::type::type > *_PTarget)=0
	When overridden in a derived class, acquires a reference count on this ISource block, to prevent deletion. More...
virtual void	release_ref (_Inout_ ITarget< single_link_registry< ITarget< _Destination_type > >::type::type > *_PTarget)=0
	When overridden in a derived class, releases a reference count on this ISource block. More...
Public Member Functions inherited from Concurrency::ITarget< multi_link_registry< ISource< size_t > >::type::source_type >
virtual	~ITarget ()
	Destroys the ITarget object. More...
virtual message_status	propagate (_Inout_opt_ message< multi_link_registry< ISource< size_t > >::type::source_type > *_PMessage, _Inout_opt_ ISource< multi_link_registry< ISource< size_t > >::type::source_type > *_PSource)=0
	When overridden in a derived class, asynchronously passes a message from a source block to this target block. More...
virtual message_status	send (_Inout_ message< multi_link_registry< ISource< size_t > >::type::source_type > *_PMessage, _Inout_ ISource< multi_link_registry< ISource< size_t > >::type::source_type > *_PSource)=0
	When overridden in a derived class, synchronously passes a message to the target block. More...
virtual bool	supports_anonymous_source ()
	When overridden in a derived class, returns true or false depending on whether the message block accepts messages offered by a source that is not linked to it. If the overridden method returns true, the target cannot postpone an offered message, as consumption of a postponed message at a later time requires the source to be identified in its source link registry. More...

Protected Member Functions
virtual message_status	propagate_message (message< size_t > *_PMessage, ISource< size_t > *_PSource)
	Asynchronously passes a message from an ISource block to this ITarget block. It is invoked by the propagate method, when called by a source block. More...
virtual message< _Destination_type > *	accept_message (runtime_object_identity _MsgId)
	Accepts an offered message by the source, transferring ownership to the caller. More...
virtual bool	reserve_message (runtime_object_identity _MsgId)
	Reserves a message previously offered by the source. More...
virtual message< _Destination_type > *	consume_message (runtime_object_identity _MsgId)
	Consumes a message previously offered by the source and reserved by the target, transferring ownership to the caller. More...
virtual void	release_message (runtime_object_identity _MsgId)
	Releases a previous message reservation. More...
virtual void	resume_propagation ()
	Resumes propagation after a reservation has been released More...
virtual void	link_target_notification (_Inout_ ITarget< _Destination_type > *)
	Notification that a target was linked to this source. More...
virtual void	propagate_to_any_targets (_Inout_opt_ message< _Destination_type > *)
	Takes the message and propagates it to all the targets of this join block. More...
Protected Member Functions inherited from Concurrency::propagator_block< single_link_registry< ITarget< _Destination_type > >, multi_link_registry< ISource< size_t > > >
virtual message_status	propagate_message (_Inout_ message< _Source_type > *_PMessage, _Inout_ ISource< _Source_type > *_PSource)=0
	When overridden in a derived class, this method asynchronously passes a message from an ISource block to this propagator_block object. It is invoked by the propagate method, when called by a source block. More...
virtual message_status	send_message (_Inout_ message< _Source_type > *, _Inout_ ISource< _Source_type > *)
	When overridden in a derived class, this method synchronously passes a message from an ISource block to this propagator_block object. It is invoked by the send method, when called by a source block. More...
virtual void	link_source (_Inout_ ISource< _Source_type > *_PSource)
	Links a specified source block to this propagator_block object. More...
virtual void	unlink_source (_Inout_ ISource< _Source_type > *_PSource)
	Unlinks a specified source block from this propagator_block object. More...
virtual void	unlink_sources ()
	Unlinks all source blocks from this propagator_block object. More...
virtual void	process_input_messages (_Inout_ message< _Target_type > *_PMessage)
	Process input messages. This is only useful for propagator blocks, which derive from source_block More...
void	initialize_source_and_target (_Inout_opt_ Scheduler *_PScheduler=NULL, _Inout_opt_ ScheduleGroup *_PScheduleGroup=NULL)
	Initializes the base object. Specifically, the message_processor object needs to be initialized. More...
void	register_filter (filter_method const &_Filter)
	Registers a filter method that will be invoked on every received message. More...
void	decline_incoming_messages ()
	Indicates to the block that new messages should be declined. More...
void	remove_network_links ()
	Removes all the source and target network links from this propagator_block object. More...
Protected Member Functions inherited from Concurrency::source_block< single_link_registry< ITarget< _Destination_type > >, _MessageProcessorType >
virtual void	link_target_notification (_Inout_ ITarget< _Target_type > *)
	A callback that notifies that a new target has been linked to this source_block object. More...
virtual void	unlink_target_notification (_Inout_ ITarget< _Target_type > *_PTarget)
	A callback that notifies that a target has been unlinked from this source_block object. More...
virtual void	propagate_output_messages ()
	Propagate messages to targets. More...
virtual void	propagate_to_any_targets (_Inout_opt_ message< _Target_type > *_PMessage)
	When overridden in a derived class, propagates the given message to any or all of the linked targets. This is the main propagation routine for message blocks. More...
void	initialize_source (_Inout_opt_ Scheduler *_PScheduler=NULL, _Inout_opt_ ScheduleGroup *_PScheduleGroup=NULL)
	Initializes the message_propagator within this source_block. More...
void	enable_batched_processing ()
	Enables batched processing for this block. More...
virtual void	sync_send (_Inout_opt_ message< _Target_type > *_Msg)
	Synchronously queues up messages and starts a propagation task, if this has not been done already. More...
virtual void	async_send (_Inout_opt_ message< _Target_type > *_Msg)
	Asynchronously queues up messages and starts a propagation task, if this has not been done already More...
void	wait_for_outstanding_async_sends ()
	Waits for all asynchronous propagations to complete. This propagator-specific spin wait is used in destructors of message blocks to make sure that all asynchronous propagations have time to finish before destroying the block. More...
void	remove_targets ()
	Removes all target links for this source block. This should be called from the destructor. More...
Protected Member Functions inherited from Concurrency::ISource< single_link_registry< ITarget< _Destination_type > >::type::type >
void	_Invoke_link_source (ITarget< single_link_registry< ITarget< _Destination_type > >::type::type > *_PLinkFrom)
	Links this source to a target. More...
void	_Invoke_unlink_source (ITarget< single_link_registry< ITarget< _Destination_type > >::type::type > *_PUnlinkFrom)
	Unlinks this source from a target. More...
Protected Member Functions inherited from Concurrency::ITarget< multi_link_registry< ISource< size_t > >::type::source_type >
virtual void	link_source (_Inout_ ISource< multi_link_registry< ISource< size_t > >::type::source_type > *_PSource)=0
	When overridden in a derived class, links a specified source block to this ITarget block. More...
virtual void	unlink_source (_Inout_ ISource< multi_link_registry< ISource< size_t > >::type::source_type > *_PSource)=0
	When overridden in a derived class, unlinks a specified source block from this ITarget block. More...

Private Types
typedef single_link_registry< ITarget< _Destination_type > >	_TargetLinkRegistry
typedef multi_link_registry< ISource< size_t > >	_SourceLinkRegistry

Private Member Functions
template<int _Index>
bool	_Try_consume_source_messages (_Destination_type &_Destination_tuple, ISource< size_t > **_Sources)
	Tries to reserve from all sources. If successful, it will consume all the messages More...
template<>
bool	_Try_consume_source_messages (_Destination_type &, ISource< size_t > **)
	Provides a sentinel template specialization for _Try_consume_source_messages recursive template expansion. More...
bool	_Non_greedy_acquire_messages ()
	Tries to acquire all of the messages from the _Non_greedy_nodes. Each node has already indicated that it has received a message that it can try to reserve. This function starts the reservation and consume process. More...
void	_Propagate_priority_order (::Concurrency::details::_Queue< message< _Target_type >> &_MessageBuffer)
	Propagate messages in priority order More...
message< _Destination_type > *	_Create_send_message ()
	Called when all the source messaging blocks have received their messages. The payloads are copied into local tuple and then packaged into a message to be propagated: _M_pSendMessage. More...
void	_Delete_stored_messages ()
	Deletes all messages currently stored in this message block. Should be called by the destructor to ensure any messages propagated in are cleaned up. More...
template<int _Index>
void	_Populate_destination_tuple (_Destination_type &_Destination_tuple, ISource< size_t > **_Sources)
	Copies payloads from all sources to destination tuple. More...
template<>
void	_Populate_destination_tuple (_Destination_type &, ISource< size_t > **)
	Provides a sentinel template specialization for _Populate_destination_tuple recursive template expansion. More...
	_Join_node (const _Join_node &_Join)
_Join_node const &	operator= (_Join_node const &)

Private Attributes
_Type	_M_sourceTuple
volatile long	_M_counter
::Concurrency::details::_Queue< message< _Destination_type > >	_M_messageBuffer

Additional Inherited Members
Public Types inherited from Concurrency::propagator_block< single_link_registry< ITarget< _Destination_type > >, multi_link_registry< ISource< size_t > > >
typedef multi_link_registry< ISource< size_t > >::type::source_type	_Source_type
	The type of the payload for the incoming message to this propagator_block. More...
typedef source_link_manager< multi_link_registry< ISource< size_t > > >	_SourceLinkManager
	The type of the source_link_manager this propagator_block. More...
typedef _SourceLinkManager::iterator	source_iterator
	The type of the iterator for the source_link_manager for this propagator_block. More...
Public Types inherited from Concurrency::source_block< single_link_registry< ITarget< _Destination_type > >, _MessageProcessorType >
typedef single_link_registry< ITarget< _Destination_type > >::type::type	_Target_type
	The payload type of messages handled by this source_block. More...
typedef single_link_registry< ITarget< _Destination_type > >::iterator	target_iterator
	The iterator to walk the connected targets. More...
Public Types inherited from Concurrency::ISource< single_link_registry< ITarget< _Destination_type > >::type::type >
typedef single_link_registry< ITarget< _Destination_type > >::type::type	source_type
	A type alias for _Type . More...
Public Types inherited from Concurrency::ITarget< multi_link_registry< ISource< size_t > >::type::source_type >
typedef multi_link_registry< ISource< size_t > >::type::source_type	type
	A type alias for _Type . More...
typedef std::function< bool(multi_link_registry< ISource< size_t > >::type::source_typeconst &)>	filter_method
	The signature of any method used by the block that returns a bool value to determine whether an offered message should be accepted. More...
Protected Attributes inherited from Concurrency::propagator_block< single_link_registry< ITarget< _Destination_type > >, multi_link_registry< ISource< size_t > > >
_SourceLinkManager	_M_connectedSources
	The container for all the sources connected to this block. More...
filter_method *	_M_pFilter
	The filter function which determines whether offered messages should be accepted. More...
volatile bool	_M_fDeclineMessages
	A bool that is set to indicate that all messages should be declined in preparation for deleting the block More...
Protected Attributes inherited from Concurrency::source_block< single_link_registry< ITarget< _Destination_type > >, _MessageProcessorType >
ITarget< _Target_type > *	_M_pReservedFor
	Connected target that is holding a reservation More...
runtime_object_identity	_M_reservedId
	Reserved message ID More...
single_link_registry< ITarget< _Destination_type > >	_M_connectedTargets
	Connected targets More...
_MessageProcessorType	_M_messageProcessor
	Processor used for asynchronous message handling More...

Detailed Description

template<typename _Type, typename _Destination_type, join_type _Jtype>
class Concurrency::_Join_node< _Type, _Destination_type, _Jtype >

Defines a block allowing sources of distinct types to be joined. Join node is a single-target, multi-source ordered propagator block

Template Parameters

_Type	The payload tuple type
_Jtype	The kind of join this is, either 'greedy' or 'non-greedy'

Member Typedef Documentation

template<typename _Type, typename _Destination_type, join_type _Jtype>

typedef multi_link_registry<ISource<size_t> > Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_SourceLinkRegistry

private

template<typename _Type, typename _Destination_type, join_type _Jtype>

typedef single_link_registry<ITarget<_Destination_type> > Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_TargetLinkRegistry

private

Constructor & Destructor Documentation

template<typename _Type, typename _Destination_type, join_type _Jtype>

Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_Join_node ( )

inline

Constructs a join within the default scheduler, and places it on any schedule group of the scheduler's choosing.

                 : _M_counter(std::tuple_size<_Destination_type>::value)
    {
        this->initialize_source_and_target();
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_Join_node ( Scheduler &  _PScheduler )

inline

Constructs a join within the specified scheduler, and places it on any schedule group of the scheduler's choosing.

Parameters

_PScheduler

A reference to a scheduler instance.

                                       : _M_counter(std::tuple_size<_Destination_type>::value)
    {
        this->initialize_source_and_target(&_PScheduler);
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_Join_node ( ScheduleGroup &  _PScheduleGroup )

inline

Constructs a join within the specified schedule group. The scheduler is implied by the schedule group.

Parameters

_PScheduleGroup

A reference to a schedule group.

                                               : _M_counter(std::tuple_size<_Destination_type>::value)
    {
        this->initialize_source_and_target(NULL, &_PScheduleGroup);
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::~_Join_node ( )

inline

Cleans up any resources that may have been created by the join.

    {
        // Remove all links
        this->remove_network_links();

        // Clean up any messages left in this message block
        _Delete_stored_messages();
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_Join_node ( const _Join_node< _Type, _Destination_type, _Jtype > &  _Join )

private

Member Function Documentation

template<typename _Type, typename _Destination_type, join_type _Jtype>

message<_Destination_type>* Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_Create_send_message ( )

inlineprivate

Called when all the source messaging blocks have received their messages. The payloads are copied into local tuple and then packaged into a message to be propagated: _M_pSendMessage.

    {
        _Destination_type _Destination_tuple;

        // Populate the sources buffer
        ISource<size_t> * _Sources[std::tuple_size<_Type>::value];
        size_t _Index = 0;

        // Get an iterator which will keep a reference on the connected sources
        source_iterator _Iter = this->_M_connectedSources.begin();

        while (*_Iter != NULL)
        {
            ISource<size_t> * _PSource = *_Iter;

            if (_PSource == NULL)
            {
                // One of the sources disconnected
                break;
            }

            // Avoid buffer overrun
            if (_Index >= std::tuple_size<_Type>::value)
            {
                // More sources that we expect
                break;
            }

            _Sources[_Index] = *_Iter;
            _Index++;
            ++_Iter;
        }

        // The order nodes should not have unlinked while the join node is
        // active.
        if (_Index != std::tuple_size<_Type>::value)
        {
            // On debug build assert to help debugging
            _CONCRT_ASSERT(_Index == std::tuple_size<_Type>::value);
            return NULL;
        }

        _Populate_destination_tuple<0>(_Destination_tuple, _Sources);

        return new message<_Destination_type>(_Destination_tuple);
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

void Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_Delete_stored_messages ( )

inlineprivate

Deletes all messages currently stored in this message block. Should be called by the destructor to ensure any messages propagated in are cleaned up.

    {
        // Delete any messages remaining in the output queue
        for (;;)
        {
            message<_Destination_type> * _Msg = _M_messageBuffer._Dequeue();
            if (_Msg == NULL)
            {
                break;
            }
            delete _Msg;
        }
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

bool Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_Non_greedy_acquire_messages ( )

inlineprivate

Tries to acquire all of the messages from the _Non_greedy_nodes. Each node has already indicated that it has received a message that it can try to reserve. This function starts the reservation and consume process.

Returns

A bool indicating whether the reserve/consume of all messages succeeded.

    {
        _Destination_type _Destination_tuple;

        // Populate the sources buffer
        ISource<size_t> * _Sources[std::tuple_size<_Type>::value];
        size_t _Index = 0;

        // Get an iterator which will keep a reference on the connected sources
        source_iterator _Iter = this->_M_connectedSources.begin();

        while (*_Iter != NULL)
        {
            ISource<size_t> * _PSource = *_Iter;

            if (_PSource == NULL)
            {
                // One of the sources disconnected
                break;
            }

            if (_Index >= std::tuple_size<_Type>::value)
            {
                // More sources that we expect
                break;
            }

            _Sources[_Index] = _PSource;
            _Index++;
            ++_Iter;
        }

        // The order nodes should not have unlinked while the join node is
        // active.

        if (_Index != std::tuple_size<_Type>::value)
        {
            // On debug build assert to help debugging
            _CONCRT_ASSERT(_Index == std::tuple_size<_Type>::value);
            return false;
        }

        bool _IsAcquireSuccessful = _Try_consume_source_messages<0>(_Destination_tuple, _Sources);

        return _IsAcquireSuccessful;
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

template<int _Index>

void Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_Populate_destination_tuple ( _Destination_type &  _Destination_tuple, ISource< size_t > **  _Sources  )

inlineprivate

Copies payloads from all sources to destination tuple.

    {
        typedef _Order_node_base<typename std::remove_pointer_t<std::tuple_element_t<_Index, _Type>>::source_type> _Order_node_base_source_type;
        _Order_node_base_source_type * _Node = static_cast<_Order_node_base_source_type *>(_Sources[_Index]);

        std::get<_Index>(_Destination_tuple) = _Node->value();
        _Node->_Reset();

        _Populate_destination_tuple<_Index + 1>(_Destination_tuple, _Sources);
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

template<>

void Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_Populate_destination_tuple ( _Destination_type &  , ISource< size_t > **    )

inlineprivate

Provides a sentinel template specialization for _Populate_destination_tuple recursive template expansion.

    {
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

void Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_Propagate_priority_order ( ::Concurrency::details::_Queue< message< _Target_type >> &  _MessageBuffer )

inlineprivate

Propagate messages in priority order

Parameters

_MessageBuffer

Reference to a message queue with messages to be propagated

    {
        message<_Target_type> * _Msg = _MessageBuffer._Peek();

        // If someone has reserved the _Head message, don't propagate anymore
        if (this->_M_pReservedFor != NULL)
        {
            return;
        }

        while (_Msg != NULL)
        {
            message_status _Status = declined;

            // Always start from the first target that linked
            for (target_iterator _Iter = this->_M_connectedTargets.begin(); *_Iter != NULL; ++_Iter)
            {
                ITarget<_Target_type> * _PTarget = *_Iter;
                _Status = _PTarget->propagate(_Msg, this);

                // Ownership of message changed. Do not propagate this
                // message to any other target.
                if (_Status == accepted)
                {
                    break;
                }

                // If the target just propagated to reserved this message, stop
                // propagating it to others
                if (this->_M_pReservedFor != NULL)
                {
                    break;
                }
            }

            // If status is anything other than accepted, then the head message
            // was not propagated out.  Thus, nothing after it in the queue can
            // be propagated out.  Cease propagation.
            if (_Status != accepted)
            {
                break;
            }

            // Get the next message
            _Msg = _MessageBuffer._Peek();
        }
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

template<int _Index>

bool Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_Try_consume_source_messages ( _Destination_type &  _Destination_tuple, ISource< size_t > **  _Sources  )

inlineprivate

Tries to reserve from all sources. If successful, it will consume all the messages

Returns

A bool indicating whether the consumption attempt worked.

Template Parameters

_Index

The highest-number index of the join's sources

    {
        typedef _Non_greedy_node<typename std::remove_pointer_t<std::tuple_element_t<_Index, _Type>>::source_type> _Non_greedy_node_source_type;
        _Non_greedy_node_source_type * _Node = static_cast<_Non_greedy_node_source_type *>(_Sources[_Index]);

        // Increment the counter once for each reservation
        _InterlockedIncrement(&_M_counter);

        if (_Node->_Reserve_received_message())
        {
            bool _Ret_val = _Try_consume_source_messages<_Index + 1>(_Destination_tuple, _Sources);

            if (_Ret_val)
            {
                _Node->_Consume_received_message();
            }
            else
            {
                if (_Node->_Release_received_message())
                {
                    // If _Release_received_message() restored the ID, decrement the count for that
                    // restoration
                    if (_InterlockedDecrement(&_M_counter) == 0)
                    {
                        this->async_send(NULL);
                    }
                }
            }

            return _Ret_val;
        }

        return false;
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

template<>

bool Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_Try_consume_source_messages ( _Destination_type &  , ISource< size_t > **    )

inlineprivate

Provides a sentinel template specialization for _Try_consume_source_messages recursive template expansion.

Returns

A bool indicating whether the consumption attempt worked.

    {
        return true;
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

virtual message<_Destination_type>* Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::accept_message ( runtime_object_identity  _MsgId )

inlineprotectedvirtual

Accepts an offered message by the source, transferring ownership to the caller.

Parameters

_MsgId

The runtime object identity of the message.

Returns

A pointer to the message that the caller now has ownership of.

Implements Concurrency::source_block< single_link_registry< ITarget< _Destination_type > >, _MessageProcessorType >.

    {
        //
        // Peek at the head message in the message buffer.  If the IDs match
        // dequeue and transfer ownership
        //
        message<_Destination_type> * _Msg = NULL;

        if (_M_messageBuffer._Is_head(_MsgId))
        {
            _Msg = _M_messageBuffer._Dequeue();
        }

        return _Msg;
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

virtual message<_Destination_type>* Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::consume_message ( runtime_object_identity  _MsgId )

inlineprotectedvirtual

Consumes a message previously offered by the source and reserved by the target, transferring ownership to the caller.

Parameters

_MsgId

The runtime object identity of the message.

Returns

A pointer to the message that the caller now has ownership of.

consume_message is similar to accept, but is always preceded by a call to reserve.

Implements Concurrency::source_block< single_link_registry< ITarget< _Destination_type > >, _MessageProcessorType >.

    {
        // By default, accept the message
        return accept_message(_MsgId);
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

virtual void Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::link_target_notification ( _Inout_ ITarget< _Destination_type > *  )

inlineprotectedvirtual

Notification that a target was linked to this source.

Parameters

_PTarget

A pointer to the newly linked target.

    {
        // There is only a single target.
        _Propagate_priority_order(_M_messageBuffer);
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

_Join_node const& Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::operator= ( _Join_node< _Type, _Destination_type, _Jtype > const &  )

private

template<typename _Type, typename _Destination_type, join_type _Jtype>

virtual message_status Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::propagate_message ( message< size_t > *  _PMessage, ISource< size_t > *  _PSource  )

inlineprotectedvirtual

Asynchronously passes a message from an ISource block to this ITarget block. It is invoked by the propagate method, when called by a source block.

Parameters

_PMessage	A pointer to the message object.
_PSource	A pointer to the source block offering the message.

Returns

A message_status indication of what the target decided to do with the message.

    {
        // This join block is connected to the _Order_node sources, which know not to send
        // any more messages until join propagates them further. That is why join can
        // always accept the incoming messages.

        _PMessage = _PSource->accept(_PMessage->msg_id(), this);

        //
        // Source block created an int message only to notify join that the real
        // payload is available. There is no need to keep this message around.
        //
        _CONCRT_ASSERT(_PMessage != NULL);
        delete _PMessage;

        long _Ret_val = _InterlockedDecrement(&_M_counter);

        _CONCRT_ASSERT(_Ret_val >= 0);

        if (_Ret_val == 0)
        {
            //
            // All source messages are now received so join can propagate them further
            //
            this->async_send(NULL);
        }

        return accepted;
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

virtual void Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::propagate_to_any_targets ( _Inout_opt_ message< _Destination_type > *  )

inlineprotectedvirtual

Takes the message and propagates it to all the targets of this join block.

Parameters

_PMessage

A pointer to a new message.

This function packages source payloads into a tuple message and immediately sends it to the targets.

    {
        message<_Destination_type> * _Msg = NULL;

        if (_M_counter == 0)
        {
            bool fIsNonGreedy = (_Jtype == non_greedy);

            if (fIsNonGreedy)
            {
                if (!_Non_greedy_acquire_messages())
                {
                    return;
                }
            }

            if (!fIsNonGreedy)
            {
                // Because a greedy join has captured all input, we can reset
                // the counter to the total number of inputs
                _InterlockedExchange(&_M_counter, std::tuple_size<_Destination_type>::value);
            }

            _Msg = _Create_send_message();
        }

        if (_Msg != NULL)
        {
            _M_messageBuffer._Enqueue(_Msg);

            if (!_M_messageBuffer._Is_head(_Msg->msg_id()))
            {
                // another message is at the head of the outbound message queue and blocked
                // simply return
                return;
            }
        }

        _Propagate_priority_order(_M_messageBuffer);
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

virtual void Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::release_message ( runtime_object_identity  _MsgId )

inlineprotectedvirtual

Releases a previous message reservation.

Parameters

_MsgId

The runtime object identity of the message.

Implements Concurrency::source_block< single_link_registry< ITarget< _Destination_type > >, _MessageProcessorType >.

    {
        // The head message is the one reserved.
        if (!_M_messageBuffer._Is_head(_MsgId))
        {
            throw message_not_found();
        }
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

virtual bool Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::reserve_message ( runtime_object_identity  _MsgId )

inlineprotectedvirtual

Reserves a message previously offered by the source.

Parameters

_MsgId

The runtime object identity of the message.

Returns

A bool indicating whether the reservation worked or not.

After reserve is called, if it returns true, either consume or release must be called to either take or release ownership of the message.

Implements Concurrency::source_block< single_link_registry< ITarget< _Destination_type > >, _MessageProcessorType >.

    {
        // Allow reservation if this is the head message
        return _M_messageBuffer._Is_head(_MsgId);
    }

template<typename _Type, typename _Destination_type, join_type _Jtype>

virtual void Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::resume_propagation ( )

inlineprotectedvirtual

Resumes propagation after a reservation has been released

Implements Concurrency::source_block< single_link_registry< ITarget< _Destination_type > >, _MessageProcessorType >.

    {
        // If there are any messages in the buffer, propagate them out
        if (_M_messageBuffer._Count() > 0)
        {
            this->async_send(NULL);
        }
    }

Member Data Documentation

template<typename _Type, typename _Destination_type, join_type _Jtype>

volatile long Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_M_counter

private

template<typename _Type, typename _Destination_type, join_type _Jtype>

::Concurrency::details::_Queue<message<_Destination_type> > Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_M_messageBuffer

private

template<typename _Type, typename _Destination_type, join_type _Jtype>

_Type Concurrency::_Join_node< _Type, _Destination_type, _Jtype >::_M_sourceTuple

private

---

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

• VS2015/inc/agents.h