A join messaging block is a single-target, multi-source, ordered propagator_block which combines together messages of type _Type from each of its sources. More...

#include <agents.h>

Inheritance diagram for Concurrency::join< _Type, _Jtype >:

Classes
struct	_MessageArray

struct	_SavedMessageIdArray

Public Types
typedef std::vector< _Type >	_OutputType

Public Types inherited from Concurrency::propagator_block< single_link_registry< ITarget< std::vector< _Type > > >, multi_link_registry< ISource< _Type > > >
typedef multi_link_registry< ISource< _Type > >::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< _Type > > >	_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< std::vector< _Type > > >, _MessageProcessorType >
typedef single_link_registry< ITarget< std::vector< _Type > > >::type::type	_Target_type
	The payload type of messages handled by this `source_block`. More...

typedef single_link_registry< ITarget< std::vector< _Type > > >::iterator	target_iterator
	The iterator to walk the connected targets. More...

Public Types inherited from Concurrency::ISource< single_link_registry< ITarget< std::vector< _Type > > >::type::type >
typedef single_link_registry< ITarget< std::vector< _Type > > >::type::type	source_type
	A type alias for _Type . More...

Public Types inherited from Concurrency::ITarget< multi_link_registry< ISource< _Type > >::type::source_type >
typedef multi_link_registry< ISource< _Type > >::type::source_type	type
	A type alias for _Type . More...

typedef std::function< bool(multi_link_registry< ISource< _Type > >::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...

Public Member Functions
	join (size_t _NumInputs)
	Constructs a `join` messaging block. More...

	join (size_t _NumInputs, filter_method const &_Filter)
	Constructs a `join` messaging block. More...

	~join ()
	Destroys the `join` block. More...

Public Member Functions inherited from Concurrency::propagator_block< single_link_registry< ITarget< std::vector< _Type > > >, multi_link_registry< ISource< _Type > > >
	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< std::vector< _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< std::vector< _Type > > >::type::type >
virtual	~ISource ()
	Destroys the `ISource` object. More...

virtual void	link_target (_Inout_ ITarget< single_link_registry< ITarget< std::vector< _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< std::vector< _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< std::vector< _Type > > >::type::type > *	accept (runtime_object_identity _MsgId, _Inout_ ITarget< single_link_registry< ITarget< std::vector< _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< std::vector< _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< std::vector< _Type > > >::type::type > *	consume (runtime_object_identity _MsgId, _Inout_ ITarget< single_link_registry< ITarget< std::vector< _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< std::vector< _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< std::vector< _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< std::vector< _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< _Type > >::type::source_type >
virtual	~ITarget ()
	Destroys the `ITarget` object. More...

virtual message_status	propagate (_Inout_opt_ message< multi_link_registry< ISource< _Type > >::type::source_type > _PMessage, _Inout_opt_ ISource< multi_link_registry< ISource< _Type > >::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< _Type > >::type::source_type > _PMessage, _Inout_ ISource< multi_link_registry< ISource< _Type > >::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
message_status	propagate_message (_Inout_ message< _Type > _PMessage, _Inout_ ISource< _Type > _PSource)
	Asynchronously passes a message from an `ISource` block to this `join` messaging block. It is invoked by the `propagate` method, when called by a source block. More...

virtual message< _OutputType > *	accept_message (runtime_object_identity _MsgId)
	Accepts a message that was offered by this `join` messaging block, transferring ownership to the caller. More...

virtual bool	reserve_message (runtime_object_identity _MsgId)
	Reserves a message previously offered by this `join` messaging block. More...

virtual message< _OutputType > *	consume_message (runtime_object_identity _MsgId)
	Consumes a message previously offered by the `join` messaging block 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< std::vector< _Type >> *)
	A callback that notifies that a new target has been linked to this `join` messaging block. More...

void	propagate_to_any_targets (_Inout_opt_ message< _OutputType > *)
	Constructs an output message containing an input message from each source when they have all propagated a message. Sends this output message out to each of its targets. More...

Protected Member Functions inherited from Concurrency::propagator_block< single_link_registry< ITarget< std::vector< _Type > > >, multi_link_registry< ISource< _Type > > >
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 > *)
	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< std::vector< _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< std::vector< _Type > > >::type::type >
void	_Invoke_link_source (ITarget< single_link_registry< ITarget< std::vector< _Type > > >::type::type > *_PLinkFrom)
	Links this source to a target. More...

void	_Invoke_unlink_source (ITarget< single_link_registry< ITarget< std::vector< _Type > > >::type::type > *_PUnlinkFrom)
	Unlinks this source from a target. More...

Protected Member Functions inherited from Concurrency::ITarget< multi_link_registry< ISource< _Type > >::type::source_type >
virtual void	link_source (_Inout_ ISource< multi_link_registry< ISource< _Type > >::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< _Type > >::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< std::vector< _Type > > >	_TargetLinkRegistry

typedef multi_link_registry< ISource< _Type > >	_SourceLinkRegistry

Private Member Functions
void	_Propagate_priority_order (::Concurrency::details::_Queue< message< _Target_type >> &_MessageBuffer)
	Propagate messages in priority order. More...

message< std::vector< _Type > > *__cdecl	_Create_new_message ()
	Constructs a new message from the data output. More...

void	_Initialize (size_t _NumInputs, Scheduler _PScheduler=NULL, ScheduleGroup _PScheduleGroup=NULL)
	Initializes the `join` messaging block. 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...

Private Attributes
volatile size_t	_M_messagesRemaining

_MessageArray	_M_messageArray

_SavedMessageIdArray	_M_savedMessageIdArray

runtime_object_identity *	_M_savedIdBuffer

::Concurrency::details::_NonReentrantPPLLock	_M_propagationLock

::Concurrency::details::_Queue< message< std::vector< _Type > > >	_M_messageBuffer

Additional Inherited Members
Protected Attributes inherited from Concurrency::propagator_block< single_link_registry< ITarget< std::vector< _Type > > >, multi_link_registry< ISource< _Type > > >
_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< std::vector< _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< std::vector< _Type > > >	_M_connectedTargets
	Connected targets More...

_MessageProcessorType	_M_messageProcessor
	Processor used for asynchronous message handling More...

Detailed Description

template<class _Type, join_type _Jtype = non_greedy>
class Concurrency::join< _Type, _Jtype >

A join messaging block is a single-target, multi-source, ordered propagator_block which combines together messages of type _Type from each of its sources.

Template Parameters

_Type	The payload type of the messages joined and propagated by the block.
_Jtype	The kind of `join` block this is, either `greedy` or `non_greedy`

For more information, see Asynchronous Message Blocks.

See also: choice Class, multitype_join Class, join_type Enumeration

Member Typedef Documentation

template<class _Type , join_type _Jtype = non_greedy>

typedef std::vector<_Type> Concurrency::join< _Type, _Jtype >::_OutputType

template<class _Type , join_type _Jtype = non_greedy>

typedef multi_link_registry<ISource<_Type> > Concurrency::join< _Type, _Jtype >::_SourceLinkRegistry

private

template<class _Type , join_type _Jtype = non_greedy>

typedef single_link_registry<ITarget<std::vector<_Type> > > Concurrency::join< _Type, _Jtype >::_TargetLinkRegistry

private

Constructor & Destructor Documentation

template<class _Type , join_type _Jtype = non_greedy>

Concurrency::join< _Type, _Jtype >::join ( size_t _NumInputs )

inline

Constructs a join messaging block.

Parameters

_NumInputs The number of inputs this join block will be allowed.

The runtime uses the default scheduler if you do not specify the _PScheduler or _PScheduleGroup parameters.

The type filter_method is a functor with signature bool (_Type const &) which is invoked by this join messaging block to determine whether or not it should accept an offered message.

See also: Scheduler Class, ScheduleGroup Class

         : _M_messageArray(_NumInputs),
           _M_savedMessageIdArray(_NumInputs)
     {
         _Initialize(_NumInputs);
     }

template<class _Type , join_type _Jtype = non_greedy>

Concurrency::join< _Type, _Jtype >::join	(	size_t	_NumInputs,
		filter_method const &	_Filter
	)

inline

Constructs a join messaging block.

Parameters

_NumInputs	The number of inputs this `join` block will be allowed.
_Filter	A filter function which determines whether offered messages should be accepted.

The runtime uses the default scheduler if you do not specify the _PScheduler or _PScheduleGroup parameters.

The type filter_method is a functor with signature bool (_Type const &) which is invoked by this join messaging block to determine whether or not it should accept an offered message.

See also: Scheduler Class, ScheduleGroup Class

         : _M_messageArray(_NumInputs),
           _M_savedMessageIdArray(_NumInputs)
     {
         _Initialize(_NumInputs);
         register_filter(_Filter);
     }

template<class _Type , join_type _Jtype = non_greedy>

Concurrency::join< _Type, _Jtype >::~join ( )

inline

Destroys the join block.

     {
         // Remove all links that are targets of this join
         this->remove_network_links();
 
         // Clean up any messages left in this message block
         _Delete_stored_messages();
 
         delete [] _M_savedIdBuffer;
     }

Member Function Documentation

template<class _Type , join_type _Jtype = non_greedy>

message<std::vector<_Type> >* __cdecl Concurrency::join< _Type, _Jtype >::_Create_new_message ( )

inlineprivate

Constructs a new message from the data output.

Returns: The created message (NULL if creation failed)

     {
         bool fIsNonGreedy = (_Jtype == non_greedy);
 
         // If this is a non-greedy join, check each source and try to consume their message
         if (fIsNonGreedy)
         {
 
             // The iterator _Iter below will ensure that it is safe to touch
             // non-NULL source pointers. Take a snapshot.
             std::vector<ISource<_Type> *> _Sources;
             source_iterator _Iter = this->_M_connectedSources.begin();
 
             while (*_Iter != NULL)
             {
                 ISource<_Type> * _PSource = *_Iter;
 
                 if (_PSource == NULL)
                 {
                     break;
                 }
 
                 _Sources.push_back(_PSource);
                 ++_Iter;
             }
 
             if (_Sources.size() != _M_messageArray._M_count)
             {
                 // Some of the sources were unlinked. The join is broken
                 return NULL;
             }
 
             // First, try and reserve all the messages.  If a reservation fails,
             // then release any reservations that had been made.
             for (size_t i = 0; i < _M_savedMessageIdArray._M_count; i++)
             {
                 // Snap the current saved ID into a buffer.  This value can be changing behind the scenes from
                 // other source->propagate(msg, this) calls, but if so, that just means the reserve below will
                 // fail.
                 _InterlockedIncrementSizeT(&_M_messagesRemaining);
                 _M_savedIdBuffer[i] = _InterlockedExchange((volatile long *) &_M_savedMessageIdArray._M_savedIds[i], -1);
 
                 _CONCRT_ASSERT(_M_savedIdBuffer[i] != -1);
 
                 if (!_Sources[i]->reserve(_M_savedIdBuffer[i], this))
                 {
                     // A reservation failed, release all reservations made up until
                     // this block, and wait for another message to arrive on this link
                     for (size_t j = 0; j < i; j++)
                     {
                         _Sources[j]->release(_M_savedIdBuffer[j], this);
                         if (_InterlockedCompareExchange((volatile long *) &_M_savedMessageIdArray._M_savedIds[j], _M_savedIdBuffer[j], -1) == -1)
                         {
                             if (_InterlockedDecrementSizeT(&_M_messagesRemaining) == 0)
                             {
                                 this->async_send(NULL);
                             }
                         }
                     }
 
                     // Return NULL to indicate that the create failed
                     return NULL;
                 }
             }
 
             // Because everything has been reserved, consume all the messages.
             // This is guaranteed to return true.
             for (size_t i = 0; i < _M_messageArray._M_count; i++)
             {
                 _M_messageArray._M_messages[i] = _Sources[i]->consume(_M_savedIdBuffer[i], this);
                 _M_savedIdBuffer[i] = -1;
             }
         }
 
         if (!fIsNonGreedy)
         {
             // Reinitialize how many messages are being waited for.
             // This is safe because all messages have been received, thus no new async_sends for
             // greedy joins can be called.
             _M_messagesRemaining = _M_messageArray._M_count;
         }
 
         std::vector<_Type> _OutputVector;
         for (size_t i = 0; i < _M_messageArray._M_count; i++)
         {
             _CONCRT_ASSERT(_M_messageArray._M_messages[i] != NULL);
             _OutputVector.push_back(_M_messageArray._M_messages[i]->payload);
 
             delete _M_messageArray._M_messages[i];
             if (fIsNonGreedy)
             {
                 _M_messageArray._M_messages[i] = NULL;
             }
         }
         return (new message<std::vector<_Type>>(_OutputVector));
     }

template<class _Type , join_type _Jtype = non_greedy>

void Concurrency::join< _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.

     {
         // Input messages for this message block are in the base-class input buffer
         // All messages in that buffer are guaranteed to have moved to the output
         // buffer because the destructor first waits for all async sends to finish
         // before reaching this point
 
         // Delete any messages remaining in the output queue
         for (;;)
         {
             message<std::vector<_Type>> * _Msg = _M_messageBuffer._Dequeue();
             if (_Msg == NULL)
             {
                 break;
             }
             delete _Msg;
         }
     }

template<class _Type , join_type _Jtype = non_greedy>

void Concurrency::join< _Type, _Jtype >::_Initialize	(	size_t	_NumInputs,
		Scheduler *	_PScheduler = `NULL`,
		ScheduleGroup *	_PScheduleGroup = `NULL`
	)

inlineprivate

Initializes the join messaging block.

Parameters

_NumInputs	The number of inputs.
_PScheduler	The scheduler onto which the task to propagate the `join` block's message will be scheduled. If unspecified, the `join` messaging block uses the default scheduler.
_PScheduleGroup	The schedule group into which the task to propagate the `join` block's message will be scheduled. The scheduler used is implied by the schedule group. If unspecified, the `join` uses a schedule group of the scheduler's choosing.

     {
         this->initialize_source_and_target(_PScheduler, _PScheduleGroup);
 
         this->_M_connectedSources.set_bound(_NumInputs);
         _M_messagesRemaining = _NumInputs;
 
         bool fIsNonGreedy = (_Jtype == non_greedy);
 
         if (fIsNonGreedy)
         {
             // Non greedy joins need a buffer to snap off saved message IDs to.
             _M_savedIdBuffer = new runtime_object_identity[_NumInputs];
             memset(_M_savedIdBuffer, -1, sizeof(runtime_object_identity) * _NumInputs);
         }
         else
         {
             _M_savedIdBuffer = NULL;
         }
     }

template<class _Type , join_type _Jtype = non_greedy>

void Concurrency::join< _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<class _Type , join_type _Jtype = non_greedy>

virtual message<_OutputType>* Concurrency::join< _Type, _Jtype >::accept_message ( runtime_object_identity _MsgId )

inlineprotectedvirtual

Accepts a message that was offered by this join messaging block, transferring ownership to the caller.

Parameters

_MsgId The runtime_object_identity of the offered message object.

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

Implements Concurrency::source_block< single_link_registry< ITarget< std::vector< _Type > > >, _MessageProcessorType >.

     {
         //
         // Peek at the head message in the message buffer.  If the IDs match
         // dequeue and transfer ownership
         //
         message<_OutputType> * _Msg = NULL;
 
         if (_M_messageBuffer._Is_head(_MsgId))
         {
             _Msg = _M_messageBuffer._Dequeue();
         }
 
         return _Msg;
     }

template<class _Type , join_type _Jtype = non_greedy>

virtual message<_OutputType>* Concurrency::join< _Type, _Jtype >::consume_message ( runtime_object_identity _MsgId )

inlineprotectedvirtual

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

Parameters

_MsgId The runtime_object_identity of the message object being consumed.

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

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

Implements Concurrency::source_block< single_link_registry< ITarget< std::vector< _Type > > >, _MessageProcessorType >.

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

template<class _Type , join_type _Jtype = non_greedy>

virtual void Concurrency::join< _Type, _Jtype >::link_target_notification ( _Inout_ ITarget< std::vector< _Type >> * )

inlineprotectedvirtual

A callback that notifies that a new target has been linked to this join messaging block.

Parameters

_PTarget A pointer to the newly linked target.

     {
         // If the message queue is blocked due to reservation
         // there is no need to do any message propagation
         if (this->_M_pReservedFor != NULL)
         {
             return;
         }
 
         _Propagate_priority_order(_M_messageBuffer);
     }

template<class _Type , join_type _Jtype = non_greedy>

message_status Concurrency::join< _Type, _Jtype >::propagate_message	(	_Inout_ message< _Type > *	_PMessage,
		_Inout_ ISource< _Type > *	_PSource
	)

inlineprotected

Asynchronously passes a message from an ISource block to this join messaging 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.

     {
         // It is important that calls to propagate do *not* take the same lock on the
         // internal structure that is used by Consume and the LWT.  Doing so could
         // result in a deadlock with the Consume call.
 
         message_status _Ret_val = accepted;
 
         //
         // Find the slot index of this source
         //
         size_t _Slot = 0;
         bool _Found = false;
         for (source_iterator _Iter = this->_M_connectedSources.begin(); *_Iter != NULL; ++_Iter)
         {
             if (*_Iter == _PSource)
             {
                 _Found = true;
                 break;
             }
 
             _Slot++;
         }
 
         if (!_Found)
         {
             // If this source was not found in the array, this is not a connected source
             // decline the message
             return declined;
         }
 
         _CONCRT_ASSERT(_Slot < _M_messageArray._M_count);
 
         bool fIsGreedy = (_Jtype == greedy);
 
         if (fIsGreedy)
         {
             //
             // Greedy type joins immediately accept the message.
             //
             {
                 _NR_lock lockHolder(_M_propagationLock);
                 if (_M_messageArray._M_messages[_Slot] != NULL)
                 {
                     _M_savedMessageIdArray._M_savedIds[_Slot] = _PMessage->msg_id();
                     _Ret_val = postponed;
                 }
             }
 
             if (_Ret_val != postponed)
             {
                 _M_messageArray._M_messages[_Slot] = _PSource->accept(_PMessage->msg_id(), this);
 
                 if (_M_messageArray._M_messages[_Slot] != NULL)
                 {
                     if (_InterlockedDecrementSizeT(&_M_messagesRemaining) == 0)
                     {
                         // If messages have arrived on all links, start a propagation
                         // of the current message
                         this->async_send(NULL);
                     }
                 }
                 else
                 {
                     _Ret_val = missed;
                 }
             }
         }
         else
         {
             //
             // Non-greedy type joins save the message IDs until they have all arrived
             //
 
             if (_InterlockedExchange((volatile long *) &_M_savedMessageIdArray._M_savedIds[_Slot], _PMessage->msg_id()) == -1)
             {
                 // Decrement the message remaining count if this thread is switching
                 // the saved ID from -1 to a valid value.
                 if (_InterlockedDecrementSizeT(&_M_messagesRemaining) == 0)
                 {
                     this->async_send(NULL);
                 }
             }
 
             // Always return postponed.  This message will be consumed
             // in the LWT
             _Ret_val = postponed;
         }
 
         return _Ret_val;
     }

template<class _Type , join_type _Jtype = non_greedy>

void Concurrency::join< _Type, _Jtype >::propagate_to_any_targets ( _Inout_opt_ message< _OutputType > * )

inlineprotected

Constructs an output message containing an input message from each source when they have all propagated a message. Sends this output message out to each of its targets.

     {
         message<_OutputType> * _Msg = NULL;
         // Create a new message from the input sources
         // If messagesRemaining == 0, we have a new message to create.  Otherwise, this is coming from
         // a consume or release from the target.  In that case we don't want to create a new message.
         if (_M_messagesRemaining == 0)
         {
             // A greedy join can immediately create the message, a non-greedy
             // join must try and consume all the messages it has postponed
             _Msg = _Create_new_message();
         }
 
         if (_Msg == NULL)
         {
             // Create message failed.  This happens in non_greedy joins when the
             // reserve/consumption of a postponed message failed.
             _Propagate_priority_order(_M_messageBuffer);
             return;
         }
 
         bool fIsGreedy = (_Jtype == greedy);
 
         // For a greedy join, reset the number of messages remaining
         // Check to see if multiple messages have been passed in on any of the links,
         // and postponed. If so, try and reserve/consume them now
         if (fIsGreedy)
         {
             // Look at the saved IDs and reserve/consume any that have passed in while
             // this join was waiting to complete
             _CONCRT_ASSERT(_M_messageArray._M_count == _M_savedMessageIdArray._M_count);
 
             for (size_t i = 0; i < _M_messageArray._M_count; i++)
             {
                 for(;;)
                 {
                     runtime_object_identity _Saved_id;
                     // Grab the current saved ID value.  This value could be changing from based on any
                     // calls of source->propagate(this).  If the message ID is different than what is snapped
                     // here, that means, the reserve below must fail.  This is because reserve is trying
                     // to get the same source lock the propagate(this) call must be holding.
                     {
                         _NR_lock lockHolder(_M_propagationLock);
 
                         _CONCRT_ASSERT(_M_messageArray._M_messages[i] != NULL);
 
                         _Saved_id = _M_savedMessageIdArray._M_savedIds[i];
 
                         if (_Saved_id == -1)
                         {
                             _M_messageArray._M_messages[i] = NULL;
                             break;
                         }
                         else
                         {
                             _M_savedMessageIdArray._M_savedIds[i] = -1;
                         }
                     }
 
                     if (_Saved_id != -1)
                     {
                         source_iterator _Iter = this->_M_connectedSources.begin();
 
                         ISource<_Type> * _PSource = _Iter[i];
                         if ((_PSource != NULL) && _PSource->reserve(_Saved_id, this))
                         {
                             _M_messageArray._M_messages[i] = _PSource->consume(_Saved_id, this);
                             _InterlockedDecrementSizeT(&_M_messagesRemaining);
                             break;
                         }
                     }
                 }
             }
 
             // If messages have all been received, async_send again, this will start the
             // LWT up to create a new message
             if (_M_messagesRemaining == 0)
             {
                 this->async_send(NULL);
             }
         }
 
         // Add the new message to the outbound queue
         _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<class _Type , join_type _Jtype = non_greedy>

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

inlineprotectedvirtual

Releases a previous message reservation.

Parameters

_MsgId The runtime_object_identity of the message object being released.

Implements Concurrency::source_block< single_link_registry< ITarget< std::vector< _Type > > >, _MessageProcessorType >.

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

template<class _Type , join_type _Jtype = non_greedy>

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

inlineprotectedvirtual

Reserves a message previously offered by this join messaging block.

Parameters

_MsgId The runtime_object_identity of the offered message object.

Returns: true if the message was successfully reserved, false otherwise.

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< std::vector< _Type > > >, _MessageProcessorType >.

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

template<class _Type , join_type _Jtype = non_greedy>

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

inlineprotectedvirtual

Resumes propagation after a reservation has been released.

Implements Concurrency::source_block< single_link_registry< ITarget< std::vector< _Type > > >, _MessageProcessorType >.

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