Helper class used in multi-type non-greedy join blocks Ordered node is a single-target, single-source ordered propagator block More...

#include <agents.h>

Inheritance diagram for Concurrency::_Non_greedy_node< _Type >:

Public Member Functions
	_Non_greedy_node (ISource< _Type > _PSource, size_t _Index, ITarget< size_t > _PTarget=NULL)
	Constructs a _Non_greedy_node within the default scheduler, and places it on any schedule group of the scheduler's choosing. More...

	_Non_greedy_node (ISource< _Type > _PSource, size_t _Index, ITarget< size_t > _PTarget, filter_method const &_Filter)
	Constructs a _Non_greedy_node within the default scheduler, and places it on any schedule group of the scheduler's choosing. More...

	_Non_greedy_node (Scheduler &_PScheduler, ISource< _Type > _PSource, size_t _Index, ITarget< size_t > _PTarget=NULL)
	Constructs a _Non_greedy_node within the specified scheduler, and places it on any schedule group of the scheduler's choosing. More...

	_Non_greedy_node (Scheduler &_PScheduler, ISource< _Type > _PSource, size_t _Index, ITarget< size_t > _PTarget, filter_method const &_Filter)
	Constructs a _Non_greedy_node within the specified scheduler, and places it on any schedule group of the scheduler's choosing. More...

	_Non_greedy_node (ScheduleGroup &_PScheduleGroup, ISource< _Type > _PSource, size_t _Index, ITarget< size_t > _PTarget=NULL)
	Constructs a _Non_greedy_node within the specified schedule group. The scheduler is implied by the schedule group. More...

	_Non_greedy_node (ScheduleGroup &_PScheduleGroup, ISource< _Type > _PSource, size_t _Index, ITarget< size_t > _PTarget, filter_method const &_Filter)
	Constructs a _Non_greedy_node within the specified schedule group. The scheduler is implied by the schedule group. More...

	~_Non_greedy_node ()
	Cleans up any resources that may have been created by the _Order_node. More...

void	_Reset ()
	Resets the _Order_node and prepares it for the next propagation More...

bool	_Reserve_received_message ()
	Called for a non_greedy type join block in order to reserve the message in this join block More...

void	_Consume_received_message ()
	Called for a non_greedy type join block in order to consume the message in this join block that has been reserved More...

bool	_Release_received_message ()
	Called for a non_greedy type join block release a reservation on this block More...

Public Member Functions inherited from Concurrency::_Order_node_base< _Type >
	_Order_node_base ()
	Constructs a _Order_node_base within the default scheduler, and places it on any schedule group of the scheduler's choosing. More...

	~_Order_node_base ()
	Cleans up any resources that may have been created by the _Order_node. More...

bool	has_value () const
	Checks whether this block has been initialized yet. More...

_Type const &	value ()
	Gets a reference to the current payload of the message being stored. More...

virtual bool	reserve_message (runtime_object_identity)
	Reserves a message previously offered by the source. More...

virtual message< size_t > *	consume_message (runtime_object_identity)
	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)
	Releases a previous message reservation. More...

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

virtual void	link_target (_Inout_ ITarget< single_link_registry< ITarget< size_t > >::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< size_t > >::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< size_t > >::type::type > *	accept (runtime_object_identity _MsgId, _Inout_ ITarget< single_link_registry< ITarget< size_t > >::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< size_t > >::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< size_t > >::type::type > *	consume (runtime_object_identity _MsgId, _Inout_ ITarget< single_link_registry< ITarget< size_t > >::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< size_t > >::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< size_t > >::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< size_t > >::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 sourse link registry. More...

Protected Member Functions
virtual message_status	propagate_message (message< _Type > _PMessage, ISource< _Type > )
	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< size_t > *	accept_message (runtime_object_identity _MsgId)
	Accept the message by making a copy of the payload. More...

virtual void	propagate_to_any_targets (_Inout_opt_ message< size_t > *)
	Takes the message and propagates it to all the targets of this _Order_node More...

Protected Member Functions inherited from Concurrency::_Order_node_base< _Type >
virtual void	resume_propagation ()
	Resumes propagation after a reservation has been released More...

virtual void	link_target_notification (_Inout_ ITarget< size_t > *)
	Notification that a target was linked to this source. More...

void	_Create_send_message ()
	Create a message that contains an index used to determine the source message More...

void	_Initialize_order_node (ISource< _Type > _PSource, size_t _Index, ITarget< size_t > _PTarget, Scheduler _PScheduler=NULL, ScheduleGroup _PScheduleGroup=NULL)
	Validate constructor arguments and fully connect this _Order_node_base. More...

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

void	_Invoke_unlink_source (ITarget< single_link_registry< ITarget< size_t > >::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 Member Functions
_Non_greedy_node const &	operator= (_Non_greedy_node const &)

	_Non_greedy_node (_Non_greedy_node const &)

Private Attributes
ISource< _Type > *	_M_pReservedSource

::Concurrency::details::_ReentrantPPLLock	_M_resetLock

runtime_object_identity	_M_savedId

runtime_object_identity	_M_reservedId

volatile bool	_M_fIsInitialized

Additional Inherited Members
Public Types inherited from Concurrency::propagator_block< single_link_registry< ITarget< size_t > >, 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< size_t > >, _MessageProcessorType >
typedef single_link_registry< ITarget< size_t > >::type::type	_Target_type
	The payload type of messages handled by this `source_block`. More...

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

Public Types inherited from Concurrency::ISource< single_link_registry< ITarget< size_t > >::type::type >
typedef single_link_registry< ITarget< size_t > >::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::tr1::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...

Protected Attributes inherited from Concurrency::_Order_node_base< _Type >
message< _Type > *	_M_pReceiveMessage

message< size_t > *	_M_pSendMessage

size_t	_M_index

Protected Attributes inherited from Concurrency::propagator_block< single_link_registry< ITarget< size_t > >, 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< size_t > >, _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< size_t > >	_M_connectedTargets
	Connected targets More...

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

Detailed Description

template<class _Type>
class Concurrency::_Non_greedy_node< _Type >

Helper class used in multi-type non-greedy join blocks Ordered node is a single-target, single-source ordered propagator block

Template Parameters

_Type The payload type

Constructor & Destructor Documentation

template<class _Type >

Concurrency::_Non_greedy_node< _Type >::_Non_greedy_node	(	ISource< _Type > *	_PSource,
		size_t	_Index,
		ITarget< size_t > *	_PTarget = `NULL`
	)

inline

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

Parameters

_PSource	The source of data passed into the node
_Index	The node's index, assigned from the outside.
_PTarget	The target to which the node will signal about having received its input data

                                                                                                   :
         _M_savedId(-1),
         _M_reservedId(-1),
         _M_pReservedSource(NULL)
     {
         _Initialize_order_node(_PSource, _Index, _PTarget);
     }

template<class _Type >

Concurrency::_Non_greedy_node< _Type >::_Non_greedy_node	(	ISource< _Type > *	_PSource,
		size_t	_Index,
		ITarget< size_t > *	_PTarget,
		filter_method const &	_Filter
	)

inline

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

Parameters

_PSource	The source of data passed into the node
_Index	The node's index, assigned from the outside.
_PTarget	The target to which the node will signal about having received its input data
_Filter	A reference to a filter function.

                                                                                                                          :
         _M_savedId(-1),
         _M_reservedId(-1),
         _M_pReservedSource(NULL)
     {
         register_filter(_Filter);
         _Initialize_order_node(_PSource, _Index, _PTarget);
     }

template<class _Type >

Concurrency::_Non_greedy_node< _Type >::_Non_greedy_node	(	Scheduler &	_PScheduler,
		ISource< _Type > *	_PSource,
		size_t	_Index,
		ITarget< size_t > *	_PTarget = `NULL`
	)

inline

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

Parameters

_PScheduler	A reference to a scheduler instance.
_PSource	The source of data passed into the node
_Index	The node's index, assigned from the outside.
_PTarget	The target to which the node will signal about having received its input data

                                                                                                                           :
         _M_savedId(-1),
         _M_reservedId(-1),
         _M_pReservedSource(NULL)
     {
         _Initialize_order_node(_PSource, _Index, _PTarget, &_PScheduler);
     }

template<class _Type >

Concurrency::_Non_greedy_node< _Type >::_Non_greedy_node	(	Scheduler &	_PScheduler,
		ISource< _Type > *	_PSource,
		size_t	_Index,
		ITarget< size_t > *	_PTarget,
		filter_method const &	_Filter
	)

inline

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

Parameters

_PScheduler	A reference to a scheduler instance.
_PSource	The source of data passed into the node
_Index	The node's index, assigned from the outside.
_PTarget	The target to which the node will signal about having received its input data
_Filter	A reference to a filter function.

                                                                                                                                                  :
         _M_savedId(-1),
         _M_reservedId(-1),
         _M_pReservedSource(NULL)
     {
         register_filter(_Filter);
         _Initialize_order_node(_PSource, _Index, _PTarget, &_PScheduler);
     }

template<class _Type >

Concurrency::_Non_greedy_node< _Type >::_Non_greedy_node	(	ScheduleGroup &	_PScheduleGroup,
		ISource< _Type > *	_PSource,
		size_t	_Index,
		ITarget< size_t > *	_PTarget = `NULL`
	)

inline

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

Parameters

_PScheduleGroup	A reference to a schedule group.
_PSource	The source of data passed into the node
_Index	The node's index, assigned from the outside.
_PTarget	The target to which the node will signal about having received its input data

                                                                                                                                   :
         _M_savedId(-1),
         _M_reservedId(-1),
         _M_pReservedSource(NULL)
     {
         _Initialize_order_node(_PSource, _Index, _PTarget, NULL, &_PScheduleGroup);
     }

template<class _Type >

Concurrency::_Non_greedy_node< _Type >::_Non_greedy_node	(	ScheduleGroup &	_PScheduleGroup,
		ISource< _Type > *	_PSource,
		size_t	_Index,
		ITarget< size_t > *	_PTarget,
		filter_method const &	_Filter
	)

inline

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

Parameters

_PScheduleGroup	A reference to a schedule group.
_PSource	The source of data passed into the node
_Index	The node's index, assigned from the outside.
_PTarget	The target to which the node will signal about having received its input data
_Filter	A reference to a filter function.

                                                                                                                                                          :
         _M_savedId(-1),
         _M_reservedId(-1),
         _M_pReservedSource(NULL)
     {
         register_filter(_Filter);
         _Initialize_order_node(_PSource, _Index, _PTarget, NULL, &_PScheduleGroup);
     }

template<class _Type >

Concurrency::_Non_greedy_node< _Type >::~_Non_greedy_node ( )

inline

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

     {
         if (_M_pReservedSource != NULL)
         {
             _M_pReservedSource = NULL;
             _M_connectedSources.release();
         }
 
         // Remove all links
         remove_network_links();
     }

template<class _Type >

Concurrency::_Non_greedy_node< _Type >::_Non_greedy_node ( _Non_greedy_node< _Type > const & )

private

Member Function Documentation

template<class _Type >

void Concurrency::_Non_greedy_node< _Type >::_Consume_received_message ( )

inline

Called for a non_greedy type join block in order to consume the message in this join block that has been reserved

     {
         if (_M_pReservedSource != NULL)
         {
             runtime_object_identity _SavedId = _M_reservedId;
             _M_pReceiveMessage = _M_pReservedSource->consume(_SavedId, this);
 
             runtime_object_identity _OldId = NULL;
             _OldId = _InterlockedExchange((volatile long *) &_M_reservedId, -1);
 
             _CONCRT_ASSERT(_OldId == _SavedId);
 
             // Release the reference on the source
             _M_pReservedSource = NULL;
             _M_connectedSources.release();
         }
     }

template<class _Type >

bool Concurrency::_Non_greedy_node< _Type >::_Release_received_message ( )

inline

Called for a non_greedy type join block release a reservation on this block

     {
         bool retVal = false;
 
         if (_M_pReservedSource != NULL)
         {
             runtime_object_identity _SavedId = _M_reservedId;
             // If the _M_savedId is still -1, then swap the succeeded one back
             _M_pReservedSource->release(_SavedId, this);
 
             if (_InterlockedCompareExchange((volatile long *) &_M_savedId, _SavedId, -1) == -1)
             {
                 retVal = true;
             }
 
             // Release the reference on the source
             _M_pReservedSource = NULL;
             _M_connectedSources.release();
         }
 
         return retVal;
     }

template<class _Type >

bool Concurrency::_Non_greedy_node< _Type >::_Reserve_received_message ( )

inline

Called for a non_greedy type join block in order to reserve the message in this join block

Returns: A bool indicating whether the reservation worked

     {
         bool _Ret_val = false;
 
         // Order node has only  a single source.
         // Obtain an iterator to the first source. It will guarantee that the reference
         // count on the source is maintained
         source_iterator _Iter = _M_connectedSources.begin();
         ISource<_Type> * _PSource = *_Iter;
 
         if (_PSource != NULL)
         {
             // CAS out the current saved ID, in order to try and reserve it
             runtime_object_identity _SavedId = _InterlockedExchange((volatile long *) &_M_savedId, -1);
 
             _Ret_val = _PSource->reserve(_SavedId, this);
             //
             // If this reserved failed, that means we need to wait for another message
             // to come in on this link.  _M_savedID was set to -1 to indicate to the _Order_node
             // that it needs to async_send when that next message comes through
             //
             // If the reserve succeeds, save away the reserved ID.  This will be use later in
             // consume
             //
             if (_Ret_val)
             {
                 _M_reservedId = _SavedId;
 
                 // Acquire a reference on the source
                 _M_connectedSources.reference();
                 _M_pReservedSource = _PSource;
             }
         }
 
         return _Ret_val;
     }

template<class _Type >

void Concurrency::_Non_greedy_node< _Type >::_Reset ( )

inlinevirtual

Resets the _Order_node and prepares it for the next propagation

_Reset is called from Populate_destination_tuple through propagate_to_any_targets() thus, it always has the internal lock held.

Implements Concurrency::_Order_node_base< _Type >.

     {
         _R_lock _Lock(_M_resetLock);
 
         delete _M_pReceiveMessage;
         _M_pReceiveMessage = NULL;
 
         delete _M_pSendMessage;
         _M_pSendMessage = NULL;
     }

template<class _Type >

virtual message<size_t>* Concurrency::_Non_greedy_node< _Type >::accept_message ( runtime_object_identity _MsgId )

inlineprotectedvirtual

Accept the message by making a copy of the payload.

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< size_t > >, _MessageProcessorType >.

     {
         // This check is to prevent spoofing and verify that the propagated message is
         // the one that is accepted at the end.
         if (_M_pSendMessage == NULL || _MsgId != _M_pSendMessage->msg_id())
         {
             return NULL;
         }
 
         //
         // Instead of returning the internal message, we return a copy of the
         // message passed in.
         //
         // Because we are returning a copy, the accept routine for a _Non_greedy_node
         // does not need to grab the internal lock.
         //
         return (new message<size_t>(_M_pSendMessage->payload));
     }

template<class _Type >

_Non_greedy_node const& Concurrency::_Non_greedy_node< _Type >::operator= ( _Non_greedy_node< _Type > const & )

private

template<class _Type >

virtual message_status Concurrency::_Non_greedy_node< _Type >::propagate_message	(	message< _Type > *	_PMessage,
		ISource< _Type > *
	)

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.

It is important that calls to propagate do not take the same lock on the internal structure that is used by Consume and the light-weight task. Doing so could result in a deadlock with the Consume call.

     {
         // Change the message ID.  If it was -1, that means an async-send needs to occur
         if (_InterlockedExchange((volatile long *) &_M_savedId, _PMessage->msg_id()) == -1)
         {
             async_send(NULL);
         }
 
         // Always return postponed.  This message will be consumed
         // in the LWT
 
         return postponed;
     }

template<class _Type >

virtual void Concurrency::_Non_greedy_node< _Type >::propagate_to_any_targets ( _Inout_opt_ message< size_t > * )

inlineprotectedvirtual

Takes the message and propagates it to all the targets of this _Order_node

Parameters

_PMessage A pointer to a new message.

This function packages its _M_index into a message and immediately sends it to the targets.

     {
         _R_lock _Lock(_M_resetLock);
 
         if (_M_pSendMessage == NULL)
         {
             _Create_send_message();
         }
 
         for (target_iterator _Iter = _M_connectedTargets.begin(); *_Iter != NULL; ++_Iter)
         {
             ITarget<size_t> * _PTarget = *_Iter;
             _PTarget->propagate(_M_pSendMessage, this);
         }
     }