Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType > Class Template Reference

The source_block class is an abstract base class for source-only blocks. The class provides basic link management functionality as well as common error checks. More...

#include <agents.h>

Inheritance diagram for Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >:

Public Types
typedef _TargetLinkRegistry::type::type	_Target_type
	The payload type of messages handled by this source_block. More...
typedef _TargetLinkRegistry::iterator	target_iterator
	The iterator to walk the connected targets. More...
Public Types inherited from Concurrency::ISource< _TargetLinkRegistry::type::type >
typedef _TargetLinkRegistry::type::type	source_type
	A type alias for _Type . More...

Public Member Functions
	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< _TargetLinkRegistry::type::type >
virtual	~ISource ()
	Destroys the ISource object. More...

Protected Member Functions
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 message< _Target_type > *	accept_message (runtime_object_identity _MsgId)=0
	When overridden in a derived class, accepts an offered message by the source. Message blocks should override this method to validate the _MsgId and return a message. More...
virtual bool	reserve_message (runtime_object_identity _MsgId)=0
	When overridden in a derived class, reserves a message previously offered by this source_block object. More...
virtual message< _Target_type > *	consume_message (runtime_object_identity _MsgId)=0
	When overridden in a derived class, consumes a message that was previously reserved. More...
virtual void	release_message (runtime_object_identity _MsgId)=0
	When overridden in a derived class, releases a previous message reservation. More...
virtual void	resume_propagation ()=0
	When overridden in a derived class, resumes propagation after a reservation has been released. 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...
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< _TargetLinkRegistry::type::type >
void	_Invoke_link_source (ITarget< _TargetLinkRegistry::type::type > *_PLinkFrom)
	Links this source to a target. More...
void	_Invoke_unlink_source (ITarget< _TargetLinkRegistry::type::type > *_PUnlinkFrom)
	Unlinks this source from a target. More...

Protected Attributes
ITarget< _Target_type > *	_M_pReservedFor
	Connected target that is holding a reservation More...
runtime_object_identity	_M_reservedId
	Reserved message ID More...
_TargetLinkRegistry	_M_connectedTargets
	Connected targets More...
_MessageProcessorType	_M_messageProcessor
	Processor used for asynchronous message handling More...

Private Member Functions
void	_Handle_message (message< _Target_type > *_PMessage)
	Private methods. More...
void	_Process_message (message< _Target_type > *_PMessage)
void	_Propagate_message ()
void	_Wait_on_ref (long _RefCount=0)

Private Attributes
::Concurrency::details::_ReentrantPPLLock	_M_internalLock
	Internal lock used for the following synchronization: More...
volatile long	_M_referenceCount

Detailed Description

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>
class Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >

The source_block class is an abstract base class for source-only blocks. The class provides basic link management functionality as well as common error checks.

Template Parameters

_TargetLinkRegistry	Link registry to be used for holding the target links.
_MessageProcessorType	Processor type for message processing.

Message blocks should derive from this block to take advantage of link management and synchronization provided by this class.

See also

ISource Class

Member Typedef Documentation

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

typedef _TargetLinkRegistry::type::type Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::_Target_type

The payload type of messages handled by this source_block.

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

typedef _TargetLinkRegistry::iterator Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::target_iterator

The iterator to walk the connected targets.

Constructor & Destructor Documentation

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::source_block ( )

inline

Constructs a source_block object.

                   :
      _M_pReservedFor(NULL),
      _M_reservedId(-1),
      _M_referenceCount(0)
    {
        _Trace_agents(AGENTS_EVENT_CREATE,
            ::Concurrency::details::_Trace_agents_get_id(this),
            ::Concurrency::details::_Trace_agents_get_id(&_M_messageProcessor));
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::~source_block ( )

inlinevirtual

Destroys the source_block object.

    {
        // All targets should have been unlinked
        _CONCRT_ASSERT(_M_connectedTargets.count() == 0);

        _Trace_agents(AGENTS_EVENT_DESTROY, ::Concurrency::details::_Trace_agents_get_id(this));
    }

Member Function Documentation

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::_Handle_message ( message< _Target_type > *  _PMessage )

inlineprivate

Private methods.

    {
        // Hold a lock to synchronize with unlink targets
        _R_lock _Lock(_M_internalLock);
        propagate_to_any_targets(_PMessage);
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::_Process_message ( message< _Target_type > *  _PMessage )

inlineprivate

    {
        // Don't need a lock to process the message
        process_input_messages(_PMessage);
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::_Propagate_message ( )

inlineprivate

    {
        // Hold a lock to synchronize with unlink targets
        _R_lock _Lock(_M_internalLock);
        propagate_output_messages();
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::_Wait_on_ref ( long  _RefCount = 0 )

inlineprivate

    {
        ::Concurrency::details::_SpinWaitBackoffNone spinWait;
        while(_M_referenceCount != _RefCount)
        {
            spinWait._SpinOnce();
        }
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual message<_Target_type>* Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::accept ( runtime_object_identity  _MsgId, _Inout_ ITarget< _Target_type > *  _PTarget  )

inlinevirtual

Accepts a message that was offered by this source_block object, transferring ownership to the caller.

Parameters

_MsgId	The runtime_object_identity of the offered message object.
_PTarget	A pointer to the target block that is calling the accept method.

Returns

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

The method throws an invalid_argument exception if the parameter _PTarget is NULL.

The accept method is called by a target while a message is being offered by this ISource block. The message pointer returned may be different from the one passed into the propagate method of the ITarget block, if this source decides to make a copy of the message.

Implements Concurrency::ISource< _TargetLinkRegistry::type::type >.

    {
        if (_PTarget == NULL)
        {
            throw std::invalid_argument("_PTarget");
        }

        // Assert if the target is not connected
        _CONCRT_ASSERT(_M_connectedTargets.contains(_PTarget));

        return accept_message(_MsgId);
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual message<_Target_type>* Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::accept_message ( runtime_object_identity  _MsgId )

protectedpure virtual

When overridden in a derived class, accepts an offered message by the source. Message blocks should override this method to validate the _MsgId and return a message.

Parameters

_MsgId

The runtime object identity of the message object.

Returns

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

To transfer ownership, the original message pointer should be returned. To maintain ownership, a copy of message payload needs to be made and returned.

Implemented in Concurrency::_Join_node< _Type, _Destination_type, _Jtype >, Concurrency::_Non_greedy_node< _Type >, Concurrency::_Greedy_node< _Type >, Concurrency::_Reserving_node< _Type >, Concurrency::join< _Type, _Jtype >, Concurrency::single_assignment< _Type >, Concurrency::single_assignment< size_t >, Concurrency::timer< _Type >, Concurrency::transformer< _Input, _Output >, Concurrency::overwrite_buffer< _Type >, Concurrency::overwrite_buffer< agent_status >, and Concurrency::unbounded_buffer< _Type >.

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::acquire_ref ( _Inout_ ITarget< _Target_type > *  )

inlinevirtual

Acquires a reference count on this source_block object, to prevent deletion.

Parameters

_PTarget

A pointer to the target block that is calling this method.

This method is called by an ITarget object that is being linked to this source during the link_target method.

Implements Concurrency::ISource< _TargetLinkRegistry::type::type >.

    {
        _InterlockedIncrement(&_M_referenceCount);
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::async_send ( _Inout_opt_ message< _Target_type > *  _Msg )

inlineprotectedvirtual

Asynchronously queues up messages and starts a propagation task, if this has not been done already

Parameters

_Msg	A pointer to a message object to asynchronously send.

    {
        _M_messageProcessor.async_send(_Msg);
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual message<_Target_type>* Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::consume ( runtime_object_identity  _MsgId, _Inout_ ITarget< _Target_type > *  _PTarget  )

inlinevirtual

Consumes a message previously offered by this source_block object and successfully reserved by the target, transferring ownership to the caller.

Parameters

_MsgId	The runtime_object_identity of the reserved message object.
_PTarget	A pointer to the target block that is calling the consume method.

Returns

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

The method throws an invalid_argument exception if the parameter _PTarget is NULL.

The method throws a bad_target exception if the parameter _PTarget does not represent the target that called reserve.

The consume method is similar to accept, but must always be preceded by a call to reserve that returned true.

Implements Concurrency::ISource< _TargetLinkRegistry::type::type >.

    {
        _R_lock _Lock(_M_internalLock);

        if (_PTarget == NULL)
        {
            throw std::invalid_argument("_PTarget");
        }

        if (_M_pReservedFor == NULL || _PTarget != _M_pReservedFor)
        {
            throw bad_target();
        }

        message<_Target_type> * _Msg = consume_message(_MsgId);

        if (_Msg != NULL)
        {
            // Clear the reservation
            // _M_pReservedId is intentionally not reset so that it can assist in debugging
            _M_pReservedFor = NULL;

            // Reservation is assumed to block propagation. Notify that propagation can now be resumed
            resume_propagation();
        }

        return _Msg;
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual message<_Target_type>* Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::consume_message ( runtime_object_identity  _MsgId )

protectedpure virtual

When overridden in a derived class, consumes a message that was previously reserved.

Parameters

_MsgId

The runtime_object_identity of the message object being consumed.

Returns

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

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

Implemented in Concurrency::_Join_node< _Type, _Destination_type, _Jtype >, Concurrency::_Order_node_base< _Type >, Concurrency::join< _Type, _Jtype >, Concurrency::single_assignment< _Type >, Concurrency::single_assignment< size_t >, Concurrency::timer< _Type >, Concurrency::transformer< _Input, _Output >, Concurrency::overwrite_buffer< _Type >, Concurrency::overwrite_buffer< agent_status >, and Concurrency::unbounded_buffer< _Type >.

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::enable_batched_processing ( )

inlineprotected

Enables batched processing for this block.

    {
        // Register callbacks for CRT110 batched processing
        _M_messageProcessor.initialize_batched_processing(
            // Processing function used by CRT110
            [this](message<_Target_type> * _PMessage)
            {
                // Handle message through new process_input_message to use CRT110 batch processing
                this->process_input_messages(_PMessage);
            },
            [this](void)
            {
                this->_Propagate_message();
            });
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::initialize_source ( _Inout_opt_ Scheduler *  _PScheduler = NULL, _Inout_opt_ ScheduleGroup *  _PScheduleGroup = NULL  )

inlineprotected

Initializes the message_propagator within this source_block.

Parameters

_PScheduler	The scheduler to be used for scheduling tasks.
_PScheduleGroup	The schedule group to be used for scheduling tasks.

See also

Scheduler Class, ScheduleGroup Class

    {
        // Register a callback
        _M_messageProcessor.initialize(_PScheduler, _PScheduleGroup,
            [this](message<_Target_type> * _PMessage)
            {
                this->_Handle_message(_PMessage);
            });
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::link_target ( _Inout_ ITarget< _Target_type > *  _PTarget )

inlinevirtual

Links a target block to this source_block object.

Parameters

_PTarget

A pointer to an ITarget block to link to this source_block object.

The method throws an invalid_argument exception if the parameter _PTarget is NULL.

Implements Concurrency::ISource< _TargetLinkRegistry::type::type >.

    {
        _R_lock _Lock(_M_internalLock);

        if (_PTarget == NULL)
        {
            throw std::invalid_argument("_PTarget");
        }

        _M_connectedTargets.add(_PTarget);
        _Invoke_link_source(_PTarget);
        link_target_notification(_PTarget);
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::link_target_notification ( _Inout_ ITarget< _Target_type > *  )

inlineprotectedvirtual

A callback that notifies that a new target has been linked to this source_block object.

Parameters

_PTarget

The ITarget block that was linked.

    {
        // By default, we restart propagation if there is no pending resrvation
        if (_M_pReservedFor == NULL)
        {
            propagate_to_any_targets(NULL);
        }
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::process_input_messages ( _Inout_ message< _Target_type > *  _PMessage )

inlineprotectedvirtual

Process input messages. This is only useful for propagator blocks, which derive from source_block

Reimplemented in Concurrency::propagator_block< _TargetLinkRegistry, _SourceLinkRegistry, _MessageProcessorType >, Concurrency::propagator_block< multi_link_registry< ITarget< size_t > >, multi_link_registry< ISource< size_t > > >, Concurrency::propagator_block< multi_link_registry< ITarget< _Type > >, multi_link_registry< ISource< _Type > > >, Concurrency::propagator_block< single_link_registry< ITarget< std::vector< _Type > > >, multi_link_registry< ISource< _Type > > >, Concurrency::propagator_block< single_link_registry< ITarget< _Destination_type > >, multi_link_registry< ISource< size_t > > >, Concurrency::propagator_block< single_link_registry< ITarget< size_t > >, multi_link_registry< ISource< _Type > > >, Concurrency::propagator_block< multi_link_registry< ITarget< agent_status > >, multi_link_registry< ISource< agent_status > > >, and Concurrency::propagator_block< single_link_registry< ITarget< _Output > >, multi_link_registry< ISource< _Input > > >.

    {
        // source_blocks do not need to process anything
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::propagate_output_messages ( )

inlineprotectedvirtual

Propagate messages to targets.

Reimplemented in Concurrency::unbounded_buffer< _Type >.

    {
        throw invalid_operation("To use batched processing, you must override propagate_output_messages in the message block.");
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::propagate_to_any_targets ( _Inout_opt_ message< _Target_type > *  _PMessage )

inlineprotectedvirtual

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.

Parameters

_PMessage

A pointer to the message that is to be propagated.

    {
        throw invalid_operation("To use ordered message processing, you must override propagate_to_any_targets in the message block.");
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::release ( runtime_object_identity  _MsgId, _Inout_ ITarget< _Target_type > *  _PTarget  )

inlinevirtual

Releases a previous successful message reservation.

Parameters

_MsgId	The runtime_object_identity of the reserved message object.
_PTarget	A pointer to the target block that is calling the release method.

The method throws an invalid_argument exception if the parameter _PTarget is NULL.

The method throws a bad_target exception if the parameter _PTarget does not represent the target that called reserve.

Implements Concurrency::ISource< _TargetLinkRegistry::type::type >.

    {
        _R_lock _Lock(_M_internalLock);

        if (_PTarget == NULL)
        {
            throw std::invalid_argument("_PTarget");
        }

        if (_PTarget != _M_pReservedFor)
        {
            throw bad_target();
        }

        release_message(_MsgId);

        // Clear the reservation
        // _M_pReservedId is intentionally not reset so that it can assist in debugging
        _M_pReservedFor = NULL;

        // Reservation is assumed to block propagation. Notify that propagation can now be resumed
        resume_propagation();
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::release_message ( runtime_object_identity  _MsgId )

protectedpure virtual

When overridden in a derived class, releases a previous message reservation.

Parameters

_MsgId

The runtime_object_identity of the message object being released.

Implemented in Concurrency::_Join_node< _Type, _Destination_type, _Jtype >, Concurrency::_Order_node_base< _Type >, Concurrency::join< _Type, _Jtype >, Concurrency::single_assignment< _Type >, Concurrency::single_assignment< size_t >, Concurrency::timer< _Type >, Concurrency::transformer< _Input, _Output >, Concurrency::overwrite_buffer< _Type >, Concurrency::overwrite_buffer< agent_status >, and Concurrency::unbounded_buffer< _Type >.

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::release_ref ( _Inout_ ITarget< _Target_type > *  _PTarget )

inlinevirtual

Releases a reference count on this source_block object.

Parameters

_PTarget

A pointer to the target block that is calling this method.

This method is called by an ITarget object that is being unlinked from this source. The source block is allowed to release any resources reserved for the target block.

Implements Concurrency::ISource< _TargetLinkRegistry::type::type >.

    {
        if (_PTarget != NULL)
        {
            _R_lock _Lock(_M_internalLock);

            // We assume that each target would keep a single reference on its source, so
            // we call unlink target notification on every release. Otherwise, we would be
            // required to keep a reference count per target.
            // Note: unlink_target_notification can check the value of this _PTarget pointer, but
            // must not dereference it, as it may have already been deleted.
            unlink_target_notification(_PTarget);
        }

        _InterlockedDecrement(&_M_referenceCount);

        // It is *unsafe* to touch the "this" pointer after decrementing the reference count
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::remove_targets ( )

inlineprotected

Removes all target links for this source block. This should be called from the destructor.

    {
        // Wait for outstanding propagation to complete.
        wait_for_outstanding_async_sends();

        unlink_targets();

        _Wait_on_ref();
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual bool Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::reserve ( runtime_object_identity  _MsgId, _Inout_ ITarget< _Target_type > *  _PTarget  )

inlinevirtual

Reserves a message previously offered by this source_block object.

Parameters

_MsgId	The runtime_object_identity of the offered message object.
_PTarget	A pointer to the target block that is calling the reserve method.

Returns

true if the message was successfully reserved, false otherwise. Reservations can fail for many reasons, including: the message was already reserved or accepted by another target, the source could deny reservations, and so forth.

The method throws an invalid_argument exception if the parameter _PTarget is NULL.

After you call reserve, if it succeeds, you must call either consume or release in order to take or give up possession of the message, respectively.

Implements Concurrency::ISource< _TargetLinkRegistry::type::type >.

    {
        _R_lock _Lock(_M_internalLock);

        if (_PTarget == NULL)
        {
            throw std::invalid_argument("_PTarget");
        }

        if ( _M_pReservedFor != NULL)
        {
            // Someone else is holding the reservation
            return false;
        }

        if (!reserve_message(_MsgId))
        {
            // Failed to reserve the msg ID
            return false;
        }

        // Save the reserving target and the msg ID
        _M_pReservedFor = _PTarget;
        _M_reservedId = _MsgId;

        return true;
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual bool Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::reserve_message ( runtime_object_identity  _MsgId )

protectedpure virtual

When overridden in a derived class, reserves a message previously offered by this source_block object.

Parameters

_MsgId

The runtime_object_identity of the message object being reserved.

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.

Implemented in Concurrency::_Join_node< _Type, _Destination_type, _Jtype >, Concurrency::_Order_node_base< _Type >, Concurrency::join< _Type, _Jtype >, Concurrency::single_assignment< _Type >, Concurrency::single_assignment< size_t >, Concurrency::timer< _Type >, Concurrency::transformer< _Input, _Output >, Concurrency::overwrite_buffer< _Type >, Concurrency::overwrite_buffer< agent_status >, and Concurrency::unbounded_buffer< _Type >.

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::resume_propagation ( )

protectedpure virtual

When overridden in a derived class, resumes propagation after a reservation has been released.

Implemented in Concurrency::_Join_node< _Type, _Destination_type, _Jtype >, Concurrency::_Order_node_base< _Type >, Concurrency::join< _Type, _Jtype >, Concurrency::single_assignment< _Type >, Concurrency::single_assignment< size_t >, Concurrency::timer< _Type >, Concurrency::transformer< _Input, _Output >, Concurrency::overwrite_buffer< _Type >, Concurrency::overwrite_buffer< agent_status >, and Concurrency::unbounded_buffer< _Type >.

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::sync_send ( _Inout_opt_ message< _Target_type > *  _Msg )

inlineprotectedvirtual

Synchronously queues up messages and starts a propagation task, if this has not been done already.

Parameters

_Msg	A pointer to a message object to synchronously send.

    {
        // Caller shall not be holding any locks when calling this routine
        _M_messageProcessor.sync_send(_Msg);
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::unlink_target ( _Inout_ ITarget< _Target_type > *  _PTarget )

inlinevirtual

Unlinks a target block from this source_block object.

Parameters

_PTarget

A pointer to an ITarget block to unlink from this source_block object.

The method throws an invalid_argument exception if the parameter _PTarget is NULL.

Implements Concurrency::ISource< _TargetLinkRegistry::type::type >.

    {
        _R_lock _Lock(_M_internalLock);

        if (_PTarget == NULL)
        {
            throw std::invalid_argument("_PTarget");
        }

        if (_M_connectedTargets.remove(_PTarget))
        {
            // We were able to remove the target from our list.
            // Inform the target to unlink from us
            _Invoke_unlink_source(_PTarget);
        }
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::unlink_target_notification ( _Inout_ ITarget< _Target_type > *  _PTarget )

inlineprotectedvirtual

A callback that notifies that a target has been unlinked from this source_block object.

Parameters

_PTarget

The ITarget block that was unlinked.

    {
        // At this point, the target has already been disconnected from the
        // source.  It is safe to check the value of this pointer, but not
        // safe to dereference it, as it may have already been deleted.

        // If the target being unlinked is the one holding the reservation,
        // release the reservation
        if (_M_pReservedFor == _PTarget)
        {
            release(_M_reservedId, _PTarget);
        }
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

virtual void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::unlink_targets ( )

inlinevirtual

Unlinks all target blocks from this source_block object.

Implements Concurrency::ISource< _TargetLinkRegistry::type::type >.

    {
        _R_lock _Lock(_M_internalLock);

        for (target_iterator _Iter = _M_connectedTargets.begin(); *_Iter != NULL; ++_Iter)
        {
            ITarget<_Target_type> * _PTarget = *_Iter;
            _CONCRT_ASSERT(_PTarget != NULL);

            unlink_target(_PTarget);
        }

        // All the targets should be unlinked.
        _CONCRT_ASSERT(_M_connectedTargets.count() == 0);
    }

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

void Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::wait_for_outstanding_async_sends ( )

inlineprotected

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.

    {
        _M_messageProcessor.wait();
    }

Member Data Documentation

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

_TargetLinkRegistry Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::_M_connectedTargets

protected

Connected targets

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

::Concurrency::details::_ReentrantPPLLock Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::_M_internalLock

private

Internal lock used for the following synchronization:

1. Synchronize between link and unlink target
2. Synchronize between propagate_to_any_targets and unlink_target
3. Synchronize between reserve and consume/release

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

_MessageProcessorType Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::_M_messageProcessor

protected

Processor used for asynchronous message handling

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

ITarget<_Target_type>* Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::_M_pReservedFor

protected

Connected target that is holding a reservation

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

volatile long Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::_M_referenceCount

private

template<class _TargetLinkRegistry, class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>

runtime_object_identity Concurrency::source_block< _TargetLinkRegistry, _MessageProcessorType >::_M_reservedId

protected

Reserved message ID

---

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

• VS2013/inc/agents.h