#include <agents.h>

Inheritance diagram for Concurrency::details::_SyncOriginator< _Type >:

Public Types
typedef single_link_registry< ITarget< _Type > >	_Target_registry

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

Public Member Functions
	_SyncOriginator ()

virtual	~_SyncOriginator ()

virtual void	unlink_target (ITarget< _Type > *_PTarget)

virtual void	unlink_targets ()
	When overridden in a derived class, unlinks all target blocks from this `ISource` block. More...

virtual message< _Type > *	accept (runtime_object_identity _MsgId, ITarget< _Type > *_PTarget)

virtual bool	reserve (runtime_object_identity _MsgId, ITarget< _Type > *_PTarget)

virtual message< _Type > *	consume (runtime_object_identity _MsgId, ITarget< _Type > *_PTarget)

virtual void	release (runtime_object_identity _MsgId, ITarget< _Type > *_PTarget)

virtual void	acquire_ref (_Inout_ ITarget< _Type > *)
	When overridden in a derived class, acquires a reference count on this `ISource` block, to prevent deletion. More...

virtual void	release_ref (_Inout_ ITarget< _Type > *)
	When overridden in a derived class, releases a reference count on this `ISource` block. More...

Public Member Functions inherited from Concurrency::ISource< _Type >
virtual	~ISource ()
	Destroys the `ISource` object. More...

virtual void	link_target (_Inout_ ITarget< _Type > *_PTarget)=0
	When overridden in a derived class, links a target block to this `ISource` block. More...

virtual void	unlink_target (_Inout_ ITarget< _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< _Type > *	accept (runtime_object_identity _MsgId, _Inout_ ITarget< _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< _Type > *_PTarget)=0
	When overridden in a derived class, reserves a message previously offered by this `ISource` block. More...

virtual message< _Type > *	consume (runtime_object_identity _MsgId, _Inout_ ITarget< _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< _Type > *_PTarget)=0
	When overridden in a derived class, releases a previous successful message reservation. More...

Private Member Functions
bool	_internal_send (ITarget< _Type > *_PTarget, _Type const &_Value)

virtual void	link_target (ITarget< _Type > *_PTarget)

void	_Wait_for_completion ()

void	_Wait_on_ref ()

void	_Done (message_status _Status)

Private Attributes
message< _Type > *	_M_pMessage

event	_M_ev

volatile message_status	_M_fStatus

::Concurrency::details::_ReentrantPPLLock	_M_internalLock

_Target_registry	_M_connectedTargets

volatile long	_M_referenceCount

Friends
class	_Originator

Additional Inherited Members
Protected Member Functions inherited from Concurrency::ISource< _Type >
void	_Invoke_link_source (ITarget< _Type > *_PLinkFrom)
	Links this source to a target. More...

void	_Invoke_unlink_source (ITarget< _Type > *_PUnlinkFrom)
	Unlinks this source from a target. More...

Member Typedef Documentation

template<class _Type>

typedef single_link_registry<ITarget<_Type> > Concurrency::details::_SyncOriginator< _Type >::_Target_registry

Constructor & Destructor Documentation

template<class _Type>

Concurrency::details::_SyncOriginator< _Type >::_SyncOriginator ( )

inline

                           :
           _M_pMessage(NULL),
           _M_fStatus(postponed),
           _M_referenceCount(0)
         {
         }

template<class _Type>

virtual Concurrency::details::_SyncOriginator< _Type >::~_SyncOriginator ( )

inlinevirtual

         {
             unlink_targets();
 
             _Wait_on_ref();
 
             delete _M_pMessage;
         }

Member Function Documentation

template<class _Type>

void Concurrency::details::_SyncOriginator< _Type >::_Done ( message_status _Status )

inlineprivate

         {
             // postponed is not a done state
             _CONCRT_ASSERT(_Status != postponed);
 
             _M_fStatus = _Status;
             _M_ev.set();
         }

template<class _Type>

bool Concurrency::details::_SyncOriginator< _Type >::_internal_send	(	ITarget< _Type > *	_PTarget,
		_Type const &	_Value
	)

inlineprivate

         {
             // _send should only be called once.
             if (_PTarget == NULL)
             {
                 throw std::invalid_argument("_PTarget");
             }
 
             message_status _Status = declined;
             message<_Type> * _Msg = new message<_Type>(_Value);
 
             {
                 // Hold the lock to ensure that the target doesn't unlink while
                 // propagation is in progress.
                 _R_lock _Lock(_M_internalLock);
 
                 // link to the target, create a message and send it
                 link_target(_PTarget);
 
                 _CONCRT_ASSERT(_M_pMessage == NULL);
                 _M_pMessage = _Msg;
 
                 // Send the message synchronously to the target
                 _Status = _PTarget->send(_M_pMessage, this);
             }
 
             if (_Status == postponed)
             {
                 // If the target postponed the message, wait for it to
                 // be accepted/declined.
                 _Wait_for_completion();
 
                 // Procure the final status
                 _Status = _M_fStatus;
             }
 
             // status should not be postponed.
             _CONCRT_ASSERT(_Status != postponed);
 
             return (_Status == accepted);
         }

template<class _Type>

void Concurrency::details::_SyncOriginator< _Type >::_Wait_for_completion ( )

inlineprivate

         {
             // Wait for the event to be signalled
             _M_ev.wait(COOPERATIVE_TIMEOUT_INFINITE);
             _CONCRT_ASSERT(_M_fStatus != postponed);
 
         }

template<class _Type>

void Concurrency::details::_SyncOriginator< _Type >::_Wait_on_ref ( )

inlineprivate

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

template<class _Type>

virtual message<_Type>* Concurrency::details::_SyncOriginator< _Type >::accept	(	runtime_object_identity	_MsgId,
		ITarget< _Type > *	_PTarget
	)

inlinevirtual

         {
             if (_PTarget == NULL)
             {
                 return NULL;
             }
 
             if (!_M_connectedTargets.contains(_PTarget))
             {
                 return NULL;
             }
 
             if (_M_pMessage == NULL || _M_pMessage->msg_id() != _MsgId)
             {
                 return NULL;
             }
 
             // The IDs match. Actually transfer ownership of the message and
             // unlink away from the target
             message<_Type> * _Result = _M_pMessage;
 
             // The ownership of this message has changed.  Set the internal pointer to NULL
             // so it won't be deleted in the destructor
             _M_pMessage = NULL;
 
             // The message has been accepted/consumed, propagate indication that it has succeeded
             _Done(accepted);
 
             return _Result;
         }

template<class _Type>

virtual void Concurrency::details::_SyncOriginator< _Type >::acquire_ref ( _Inout_ ITarget< _Type > * _PTarget )

inlinevirtual

When overridden in a derived class, acquires a reference count on this ISource block, 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< _Type >.

         {
             _InterlockedIncrement(&_M_referenceCount);
         }

template<class _Type>

virtual message<_Type>* Concurrency::details::_SyncOriginator< _Type >::consume	(	runtime_object_identity	_MsgId,
		ITarget< _Type > *	_PTarget
	)

inlinevirtual

         {
             if (_PTarget == NULL)
             {
                 throw std::invalid_argument("_PTarget");
             }
 
             if (!_M_connectedTargets.contains(_PTarget))
             {
                 throw bad_target();
             }
 
             return accept(_MsgId, _PTarget);
         }

template<class _Type>

virtual void Concurrency::details::_SyncOriginator< _Type >::link_target ( ITarget< _Type > * _PTarget )

inlineprivatevirtual

         {
             if (_PTarget == NULL)
             {
                 throw std::invalid_argument("_PTarget");
             }
 
             _M_connectedTargets.add(_PTarget);
             this->_Invoke_link_source(_PTarget);
 
             // There should be no pending messages to propagate at this time.
             _CONCRT_ASSERT(_M_pMessage == NULL);
         }

template<class _Type>

virtual void Concurrency::details::_SyncOriginator< _Type >::release	(	runtime_object_identity	_MsgId,
		ITarget< _Type > *	_PTarget
	)

inlinevirtual

         {
             if (_PTarget == NULL)
             {
                 throw std::invalid_argument("_PTarget");
             }
 
             if (!_M_connectedTargets.contains(_PTarget))
             {
                 throw bad_target();
             }
 
             if ((_M_pMessage == NULL) || (_M_pMessage->msg_id() != _MsgId))
             {
                 throw message_not_found();
             }
 
             // If the previously reserved message is released, then propagate
             // declined  to indicate that the message was not accepted.
             _Done(declined);
         }

template<class _Type>

virtual void Concurrency::details::_SyncOriginator< _Type >::release_ref ( _Inout_ ITarget< _Type > * _PTarget )

inlinevirtual

When overridden in a derived class, releases a reference count on this ISource block.

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< _Type >.

         {
             _InterlockedDecrement(&_M_referenceCount);
         }

template<class _Type>

virtual bool Concurrency::details::_SyncOriginator< _Type >::reserve	(	runtime_object_identity	_MsgId,
		ITarget< _Type > *	_PTarget
	)

inlinevirtual

         {
             if (_PTarget == NULL)
             {
                 throw std::invalid_argument("_PTarget");
             }
 
             if (!_M_connectedTargets.contains(_PTarget))
             {
                 return false;
             }
 
             if (_M_pMessage->msg_id() != _MsgId)
             {
                 return false;
             }
 
             return true;
         }

template<class _Type>

virtual void Concurrency::details::_SyncOriginator< _Type >::unlink_target ( ITarget< _Type > * _PTarget )

inlinevirtual

         {
             if (_PTarget == NULL)
             {
                 throw std::invalid_argument("_PTarget");
             }
             {
                 // Hold the lock to ensure that the target doesn't unlink while
                 // propagation is in progress.
                 _R_lock _Lock(_M_internalLock);
                 if (_M_connectedTargets.remove(_PTarget))
                 {
                     this->_Invoke_unlink_source(_PTarget);
 
                     // Indicate that the send is complete
                     _Done(declined);
                 }
             }
         }

template<class _Type>

virtual void Concurrency::details::_SyncOriginator< _Type >::unlink_targets ( )

inlinevirtual

When overridden in a derived class, unlinks all target blocks from this ISource block.

Implements Concurrency::ISource< _Type >.

         {
             // Hold the lock to ensure that the target doesn't unlink while
             // propagation is in progress.
             _R_lock _Lock(_M_internalLock);
 
             for (typename _Target_registry::iterator _Iter = _M_connectedTargets.begin(); *_Iter != NULL; ++_Iter)
             {
                 ITarget<_Type> * _PTarget = *_Iter;
                 if (_M_connectedTargets.remove(_PTarget))
                 {
                     this->_Invoke_unlink_source(_PTarget);
                 }
             }
 
             // All targets should be unlinked
             _CONCRT_ASSERT(_M_connectedTargets.count() == 0);
 
             // Indicate that the send is complete
             _Done(declined);
         }