agents.h

Go to the documentation of this file.

/***
* ==++==
*
* Copyright (c) Microsoft Corporation.  All rights reserved.
*
* ==--==
* =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
*
* agents.h
*
* Main public header file for ConcRT's asynchronous agents layer. This is the only header file a
* C++ program must include to use asynchronous agents.
*
* The core runtime, Parallel Patterns Library (PPL), and resource manager are defined in separate header files.
* =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
****/

#pragma once

#include <crtdefs.h>
#include <concrt.h>
#include <stdexcept>
#include <functional>
#include <tuple>
#include <type_traits>
#include <vector>
#include <concurrent_queue.h>

#define _AGENTS_H

#pragma pack(push,_CRT_PACKING)
#pragma warning(push)
#pragma warning(disable: 4100) // Unreferenced formal parameter - needed for document generation
#pragma warning(disable: 4702) // Unreachable code - needed for retail version code path
// Forward declarations


namespace Concurrency
{

typedef __int32 runtime_object_identity;


typedef ::Concurrency::details::_NonReentrantPPLLock::_Scoped_lock _NR_lock;


typedef ::Concurrency::details::_ReentrantPPLLock::_Scoped_lock _R_lock;


//***************************************************************************
// Internal namespace:
//
// Concurrency::details contains definitions to support routines in the public namespaces and macros.
// Clients should not directly interact with this namespace.
//***************************************************************************

namespace details
{
    //**************************************************************************
    // Core Messaging Support:
    //**************************************************************************

    //
    // A base class to derive from that keeps unique IDs on its derived classes
    //
    class _Runtime_object : public _AllocBase
    {
    public:
        // Creates a new runtime object.
        _CRTIMP2 _Runtime_object();

        // Creates a runtime object from an identity.
        _CRTIMP2 _Runtime_object(::Concurrency::runtime_object_identity _Id);

        // Gets the runtime object identity.
        virtual ::Concurrency::runtime_object_identity _GetId() const
        {
            return _M_id;
        }

    protected:
        // The runtime object identity.
        ::Concurrency::runtime_object_identity _M_id;
    };

    // A queue used to hold the messages for the messaging blocks
    template<class _Message>
    class _Queue : public _AllocBase
    {
    protected:
        // A pointer to the head of the queue.
        _Message * _M_pHead;

        // A pointer to a pointer to the tail of the queue.
        _Message ** _M_ppTail;

        // The number of elements presently stored in the queue.
        size_t _M_count;

    public:
        typedef typename _Message type;

        // Create a Queue
        _Queue() : _M_pHead(NULL), _M_ppTail(&_M_pHead), _M_count(0)
        {
        }

        // Destroy the queue
        ~_Queue()
        {
        }

        // Returns the count of items in the queue
        size_t _Count() const
        {
            return _M_count;
        }

        // Add an item to the tail of the queue
        //
        // Returns a Boolean indicating whether the operation succeeded.
        bool _Enqueue(_Message *_Element)
        {
            _CONCRT_ASSERT(_Element->_M_pNext == NULL);
            _CONCRT_ASSERT(*_M_ppTail == NULL);

            *_M_ppTail = _Element;
            _Element->_M_pNext = NULL;
            _M_ppTail = &(_Element->_M_pNext);
            _M_count++;

            return true;
        }

        // Remove the specified element from the queue
        //
        // Returns a Boolean indicating whether the operation succeeded, that is, the message was found in the queue.
        bool _Remove(_Message * _OldElement)
        {
            bool _Result = false;

            _CONCRT_ASSERT(_OldElement != NULL);

            if (_M_pHead == _OldElement)
            {
                _M_pHead = _OldElement->_M_pNext;
                if (_M_pHead == NULL)
                {
                    _M_ppTail = &_M_pHead;
                }

                _OldElement->_M_pNext = NULL;
                _M_count--;
                _Result = true;
            }
            else
            {
                _Message * _Next = NULL;
                for (_Message * _Node = _M_pHead; _Node != NULL; _Node = _Next)
                {
                    _Next = _Node->_M_pNext;

                    if (_Node->_M_pNext == _OldElement)
                    {
                        _Node->_M_pNext = _OldElement->_M_pNext;
                        // if this is the last element of the _Queue
                        if (_Node->_M_pNext == NULL && _M_count == 1)
                        {
                            _M_ppTail = &_M_pHead;
                        }

                        _OldElement->_M_pNext = NULL;
                        _M_count--;
                        _Result = true;
                        break;
                    }
                }
            }

            return _Result;
        }

        // Dequeue an item from the head of queue
        //
        // Returns a pointer to the message found at the head of the queue.
        _Message * _Dequeue()
        {
            if (_M_pHead == NULL)
            {
                return NULL;
            }

            _Message * _Result = _M_pHead;

            _M_pHead = _Result->_M_pNext;
            if (_M_pHead == NULL)
            {
                _M_ppTail = &_M_pHead;
            }

            _Result->_M_pNext = NULL;
            _M_count--;
            return _Result;
        }

        // Return the item at the head of the queue, without dequeuing
        //
        // Returns a pointer to the message found at the head of the queue.
        _Message * _Peek()
        {
            return _M_pHead;
        }

        // Return true if the ID matches the message at the head of the queue
        bool _Is_head(runtime_object_identity _MsgId)
        {
            // Peek at the next message in the message buffer.  Use it to
            // check if the IDs match
            _Message * _Msg = _M_pHead;

            if (_Msg == NULL || _Msg->msg_id() != _MsgId)
            {
                return false;
            }

            return true;
        }
    };

    //
    // _Dynamic_array implements a container very similar to std::vector.
    // However, it exposes a reduced subset of functionality that is
    // geared towards use in network_link_registry. The array acess is not
    // thread-safe.
    //
    template<class _Type>
    class _Dynamic_array
    {
    public:

        typedef _Dynamic_array<_Type> _Myt;

        typedef _Type& reference;
        typedef _Type const& const_reference;

        //
        // Construct a dynamic array
        //
        _Dynamic_array()
        {
            _Init();
        }

        //
        // Release any resources used by dynamic array
        //
        ~_Dynamic_array()
        {
            _Clear();
        }

        //
        // Assignment operator. Copy the contents of _Right
        //
        _Myt& operator=(const _Myt& _Right)
        {
            if (this != &_Right)
            {
                // Remove all the elements
                _Clear();

                // Allocate space for the new elements
                size_t _Size = _Right._Size();
                _Grow(_Size);

                // Copy over the new elements
                for (size_t _I=0; _I < _Size; _I++)
                {
                    _Push_back(_Right[_I]);
                }
            }

            return *this;
        }

        //
        // Clear all the elements in the array
        //
        void _Clear()
        {
            if (_M_array != NULL)
            {
                delete [] _M_array;
                _Init();
            }
        }

        //
        // Add an element to the end of the array
        //
        void _Push_back(_Type const& _Element)
        {
            if (_M_index >= _M_size)
            {
                // Not enough space. Grow the array
                size_t _NewSize = (_M_index + 1) * _S_growthFactor;
                _Grow(_NewSize);
            }

            _CONCRT_ASSERT(_M_index < _M_size);
            _M_array[_M_index] = _Element;
            _M_index++;
        }

        //
        // Index operation. Retrieve an element at the specified index. No bounds check is done.
        //
        reference operator[](size_t _Pos)
        {
            _CONCRT_ASSERT(_Pos < _M_size);
            return _M_array[_Pos];
        }

        //
        // Index operation. Retrieve an element at the specified index. No bounds check is done.
        //
        const_reference operator[](size_t _Pos) const
        {
            _CONCRT_ASSERT(_Pos < _M_size);
            return _M_array[_Pos];
        }

        //
        // Returns the count of elements in the array
        //
        size_t _Size() const
        {
            return _M_index;
        }

        //
        // Swap the contents of this array with _Right
        //
        void _Swap(_Myt& _Right)
        {
            if (this != &_Right)
            {
                // Swap the details.
                _Type * _Array = _M_array;
                size_t _Index = _M_index;
                size_t _Size = _M_size;

                _M_array = _Right._M_array;
                _M_index = _Right._M_index;
                _M_size = _Right._M_size;

                _Right._M_array = _Array;
                _Right._M_index = _Index;
                _Right._M_size = _Size;
            }
        }

    private:
        //
        // Initialize the array
        //
        void _Init()
        {
            _M_array = NULL;
            _M_index = 0;
            _M_size = 0;
        }

        //
        // Grow the array to the given size. The old elements are copied over.
        //
        void _Grow(size_t _NewSize)
        {
            _CONCRT_ASSERT( _NewSize > _M_size );

            _Type * _Array = new _Type[_NewSize];

            if (_M_array != NULL)
            {
                // Copy over the elememts
                for (size_t _I = 0; _I < _M_size; _I++)
                {
                    _Array[_I] = _M_array[_I];
                }

                delete [] _M_array;
            }

            _M_array = _Array;
            _M_size = _NewSize;
        }

        // Private data members

        // Array of elements
        _Type * _M_array;

        // Index where the next element should be inserted
        size_t  _M_index;

        // Capacity of the array.
        size_t  _M_size;

        static const int _S_growthFactor = 2;
    };

    //
    // Returns an identifier for the given object that could be used
    // in an ETW trace (call to _Trace_agents)
    //
    template <class _Type>
    __int64 _Trace_agents_get_id(_Type * _PObject)
    {
        return reinterpret_cast<__int64>(_PObject);
    }

} // namespace details

//**************************************************************************
// Public Namespace:
//
// Anything in the Concurrency namespace is intended for direct client consumption.
//
//**************************************************************************

//
// Forward declarations:
//
template<class _Type> class ISource;
template<class _Type> class ITarget;

//**************************************************************************
// Network link registry
//**************************************************************************

// Forward declaration for use in the iterator
template<class _Block> class network_link_registry;


template<class _Block>
class _Network_link_iterator
{
public:

    typedef _Network_link_iterator<_Block> _Myt;
    typedef network_link_registry<_Block>  _MyContainer;

    // Element type
    typedef _Block* _EType;

    // Const iterator - iterator shall not be used to modify the links
    typedef _EType const& const_reference;
    typedef _EType const* const_pointer;

    
    _Network_link_iterator(_MyContainer * _PNetwork_link, size_t _Index) : _M_pNetwork_link(_PNetwork_link), _M_index(_Index), _M_value(NULL)
    {
        _M_pNetwork_link->_Next_index(_M_index);
    }

    
    _Network_link_iterator(_Myt const& _Right)
    {
        _M_pNetwork_link = _Right._M_pNetwork_link;
        _M_index = _Right._M_index;
    }

    
    _Myt const& operator=(_Myt const& _Right)
    {
        _M_pNetwork_link = _Right._M_pNetwork_link;
        _M_index = _Right._M_index;
        return *this;
    }

    
    const_reference operator*()
    {
        _M_value = _M_pNetwork_link->_Get_element(_M_index);
        return _M_value;
    }

    
    const_pointer operator->() const
    {
        return (&**this);
    }

    
    _Myt& operator++()
    {
        ++_M_index;
        _M_pNetwork_link->_Next_index(_M_index);
        return (*this);
    }

    
    _Myt operator++(int)
    {
        _Myt _Tmp = *this;
        ++*this;
        return (_Tmp);
    }

private:

    // Pointer to the underlying container (network link registry)
    _MyContainer * _M_pNetwork_link;

    // Current index
    size_t _M_index;

    // Current value
    _EType _M_value;
};


template<class _Block>
class network_link_registry
{
public:

    
    typedef typename _Block type;

    
    typedef _Block * _EType;

    
    typedef _EType const& const_reference;

    
    typedef _EType const* const_pointer;

    // Make the iterators friends so that they can access some of the
    // private routines such as _Get_element.
    
    friend class _Network_link_iterator<_Block>;

    
    typedef _Network_link_iterator<_Block> iterator;

    
    virtual void add(_EType _Link) = 0;

    
    virtual bool remove(_EType _Link) = 0;

    
    virtual bool contains(_EType _Link) = 0;

    
    virtual size_t count() = 0;

    
    virtual iterator begin() = 0;

protected:

    
    virtual void _Next_index(size_t& _Index) = 0;

    
    virtual _EType _Get_element(size_t _Index) const = 0;
};


template<class _Block>
class single_link_registry : public network_link_registry<_Block>
{
public:

    
    single_link_registry() : _M_connectedLink(NULL)
    {
    }

    
    virtual ~single_link_registry()
    {
        // It is an error to delete link registry with links
        // still present
        if (count() != 0)
        {
            throw invalid_operation("Deleting link registry before removing all the links");
        }
    }

    
    virtual void add(_EType _Link)
    {
        if (_Link == NULL)
        {
            return;
        }

        // Only one link can be added.
        if (_M_connectedLink != NULL)
        {
            throw invalid_link_target("_Link");
        }

        _M_connectedLink = _Link;
    }

    
    virtual bool remove(_EType _Link)
    {
        if ((_Link != NULL) && (_M_connectedLink == _Link))
        {
            _M_connectedLink = NULL;
            return true;
        }

        return false;
    }

    
    virtual bool contains(_EType _Link)
    {
        return ((_Link != NULL) && (_M_connectedLink == _Link));
    }

    
    virtual size_t count()
    {
        return (_M_connectedLink == NULL) ? 0 : 1;
    }

    
    virtual iterator begin()
    {
        return (iterator(this, 0));
    }

protected:

    
    virtual void _Next_index(size_t& _Index)
    {
        if (_M_connectedLink == NULL)
        {
            _Index++;
        }
    }

    
    virtual _EType _Get_element(size_t _Index) const
    {
        if (_Index == 0)
        {
            return _M_connectedLink;
        }

        return NULL;
    }

private:

    // A single pointer is used to hold the link
    _EType _M_connectedLink;
};


template<class _Block>
class multi_link_registry : public network_link_registry<_Block>
{
public:

    
    multi_link_registry() : _M_maxLinks(_NOT_SET)
    {
    }

    
    virtual ~multi_link_registry()
    {
        // It is an error to delete link registry with links
        // still present
        if (count() != 0)
        {
            throw invalid_operation("Deleting link registry before removing all the links");
        }
    }

    
    void set_bound(size_t _MaxLinks)
    {
        _CONCRT_ASSERT(count() == 0);
        _M_maxLinks = _MaxLinks;
    }

    
    virtual void add(_EType _Link)
    {
        if (_Link == NULL)
        {
            return;
        }

        _Add(_Link);
    }

    
    virtual bool remove(_EType _Link)
    {
        if (_Link == NULL)
        {
            return false;
        }

        return (_Remove(_Link));
    }

    
    virtual bool contains(_EType _Link)
    {
        if (_Link == NULL)
        {
            return false;
        }

        return (_Find(_Link) < _M_vector._Size());
    }

    
    virtual size_t count()
    {
        return _Count();
    }

    
    virtual iterator begin()
    {
        return (iterator(this, 0));
    }

protected:

    
    virtual void _Next_index(size_t& _Index)
    {
        size_t _Size = _M_vector._Size();
        while (_Index < _Size)
        {
            if (_M_vector[_Index] != NULL)
            {
                break;
            }

            ++_Index;
        }
    }

    
    virtual _EType _Get_element(size_t _Index) const
    {
        if (_Index < _M_vector._Size())
        {
            return _M_vector[_Index];
        }

        return NULL;
    }

private:

    
    void _Add(_EType _Link)
    {
        size_t _Size = _M_vector._Size();
        size_t _Insert_pos = 0;

        _CONCRT_ASSERT(_Link != NULL);

        // If max links is set, ensure that inserting the new
        // link will not exceed the bound.
        if ((_M_maxLinks != _NOT_SET) && ((_Size+1) > (size_t) _M_maxLinks))
        {
            throw invalid_link_target("_Link");
        }

        for (size_t _Index = 0; _Index < _Size; _Index++)
        {
            if (_M_vector[_Index] != NULL)
            {
                // We want to find the first NULL entry after all the
                // non-NULL entries.
                _Insert_pos = _Index + 1;

                // Throw if dupiclate entry is found
                if (_M_vector[_Index] == _Link)
                {
                    throw invalid_link_target("_Link");
                }
            }
        }

        if (_Insert_pos < _Size)
        {
            _M_vector[_Insert_pos] = _Link;
        }
        else
        {
            _M_vector._Push_back(_Link);
        }
    }

    
    bool _Remove(_EType _Link)
    {
        _CONCRT_ASSERT(_Link != NULL);

        for (size_t _Index = 0; _Index < _M_vector._Size(); _Index++)
        {
            if (_M_vector[_Index] == _Link)
            {
                _M_vector[_Index] = NULL;

                // If max links is set, prevent new additions to the registry
                if (_M_maxLinks != _NOT_SET && _M_maxLinks > 0)
                {
                    // Setting the bound to 0. This causes add to always throw.
                    _M_maxLinks = 0;
                }

                return true;
            }
        }

        return false;
    }


    
    virtual size_t _Find(_EType _Link)
    {
        size_t _Index = 0;
        for (_Index = 0; _Index < _M_vector._Size(); _Index++)
        {
            if (_M_vector[_Index] == _Link)
            {
                break;
            }
        }

        return _Index;
    }

    
    size_t _Count() const
    {
        size_t _Count = 0;

        for (size_t _Index = 0; _Index < _M_vector._Size(); _Index++)
        {
            if (_M_vector[_Index] != NULL)
            {
                _Count++;
            }
        }

        return _Count;
    }

    static const size_t _NOT_SET = SIZE_MAX;

    // Maximum number of links allowed.
    size_t _M_maxLinks;

    // ::Concurrency::details::_Dynamic_array is used to hold the links
    ::Concurrency::details::_Dynamic_array<_EType> _M_vector;
};

// Forward declaration for the iterator
template<class _LinkRegistry> class source_link_manager;


template<class _LinkRegistry>
class _Source_link_iterator
{
public:

    typedef typename _LinkRegistry::type  _Block;

    typedef _Source_link_iterator<_LinkRegistry> _Myt;
    typedef source_link_manager<_LinkRegistry>  _MyContainer;

    // Element type
    typedef _Block* _EType;

    // Const iterator - iterator shall not be used to modify the links
    typedef _EType const& const_reference;
    typedef _EType const* const_pointer;

    
    _Source_link_iterator(_MyContainer * _PNetwork_link, size_t _Index) : _M_pNetwork_link(_PNetwork_link), _M_index(_Index), _M_sentinel(NULL)
    {
        // Take a snapshot of the link registry. This will reference the registry.
        _M_pNetwork_link->_To_array(_M_array);
    }

    
    virtual ~_Source_link_iterator()
    {
        if (_M_pNetwork_link != NULL)
        {
            _M_pNetwork_link->release();
        }
    }
    
    _Source_link_iterator(_Myt const& _Right)
    {
        _M_pNetwork_link = _Right._M_pNetwork_link;
        _M_index = _Right._M_index;
        _M_array = _Right._M_array;

        _M_pNetwork_link->reference();
    }

    
    _Myt const& operator=(_Myt const& _Right)
    {
        _MyContainer * _OldContainer = _M_pNetwork_link;
        _CONCRT_ASSERT(_OldContainer != NULL);

        _M_pNetwork_link = _Right._M_pNetwork_link;
        _M_index = _Right._M_index;
        _M_array = _Right._M_array;

        if (_OldContainer != _M_pNetwork_link)
        {
            _OldContainer->release();
            _M_pNetwork_link->reference();
        }

        return *this;
    }

    
    const_reference operator*()
    {
        return _Get(0);
    }

    
    const_pointer operator->() const
    {
        return (&**this);
    }

    
    const_reference operator[](size_t _Pos) const
    {
        return _Get(_Pos);
    }

    
    _Myt& operator++()
    {
        ++_M_index;
        return (*this);
    }

    
    _Myt operator++(int)
    {
        _Myt _Tmp = *this;
        ++*this;
        return (_Tmp);
    }

private:

    // Get the element at the given offset.
    const_reference _Get(size_t _Pos) const
    {
        size_t _Index = _M_index + _Pos;
        if (_Index >= _M_array._Size())
        {
            return _M_sentinel;
        }

        return _M_array[_Index];
    }

    // Array to hold the snapshot of the link registry
    ::Concurrency::details::_Dynamic_array<_EType> _M_array;

    // Pointer to the underlying container (network link registry)
    _MyContainer * _M_pNetwork_link;

    // Current index
    size_t _M_index;

    // Sentinel value to return on bounds overflow
    _EType _M_sentinel;
};


template<class _LinkRegistry>
class source_link_manager
{
public:

    
    typedef _LinkRegistry type;

    
    typedef typename _LinkRegistry::type _Block;

    
    typedef std::tr1::function<void(_Block *, bool)>  _Callback_method;

    
    typedef _Block * _EType;

    
    typedef _EType const& const_reference;

    
    typedef _EType const* const_pointer;

    // Iterator
    friend class _Source_link_iterator<_LinkRegistry>;

    
    typedef _Source_link_iterator<_LinkRegistry> iterator;

    
    typedef ::Concurrency::details::_ReentrantPPLLock _LockType;

    
    typedef _LockType::_Scoped_lock _LockHolder;

    
    source_link_manager() : _M_iteratorCount(0), _M_pLinkedTarget(NULL)
    {
    }

    
    ~source_link_manager()
    {
        _CONCRT_ASSERT(_M_pendingRemove._Size() == 0);
    }

    
    void register_target_block(_Inout_ ITarget<typename _Block::source_type> * _PTarget)
    {
        _M_pLinkedTarget = _PTarget;
    }

    
    void set_bound(size_t _MaxLinks)
    {
        _M_links.set_bound(_MaxLinks);
    }

    
    void add(_EType _Link)
    {
        if (_Link == NULL)
        {
            return;
        }

        {
            _LockHolder _Lock(_M_lock);
            _M_links.add(_Link);

            // We need to add the _Link first and then invoke the
            // callback because _Add could throw.

            // As soon as the above lock is released, remove would
            // find the link that was added and could unlink it before
            // we are able to invoke the notification below. Keeping an
            // active iterator would prevent that from happening.
            _M_iteratorCount++;
        }

        // Acquire a reference on this link by the target
        _Link->acquire_ref(_M_pLinkedTarget);

        // Release the active iterator
        release();
    }

    
    bool remove(_EType _Link)
    {
        bool _Removed = false;
        _EType _RemovedLink = NULL;
        ITarget<typename _Block::source_type> * _LinkedTarget = _M_pLinkedTarget;

        if (_Link == NULL)
        {
            return false;
        }

        {
             _LockHolder _Lock(_M_lock);
             _Removed = _M_links.remove(_Link);

             if (!_Removed)
             {
                 // No change was made
                 return _Removed;
             }

             if (_M_iteratorCount == 0)
             {
                 // Set the removed link to indicate that
                 // notification callback needs to be invoked.
                 _RemovedLink = _Link;
             }
             else
             {
                 // The iterator will complete the pending operation
                 _M_pendingRemove._Push_back(_Link);
             }
        }

        // NOTE: touching "this" pointer is dangerous as soon as the above lock is released

        // Release the reference for this link
        if (_RemovedLink != NULL)
        {
            _RemovedLink->release_ref(_LinkedTarget);
        }

         return _Removed;
    }

    
    void reference()
    {
        _LockHolder _Lock(_M_lock);
        _M_iteratorCount++;
    }

    
    void release()
    {
        ITarget<typename _Block::source_type> * _LinkedTarget = _M_pLinkedTarget;
        ::Concurrency::details::_Dynamic_array<_EType> _LinksToRemove;

        {
            _LockHolder _Lock(_M_lock);
            _CONCRT_ASSERT(_M_iteratorCount > 0);
            _M_iteratorCount--;

            if (_M_iteratorCount == 0)
            {
                if (_M_pendingRemove._Size() > 0)
                {
                    // Snap the pending remove list with the lock held
                    _M_pendingRemove._Swap(_LinksToRemove);
                }
            }
        }

        // NOTE: touching "this" pointer is dangerous as soon as the above lock is released

        // Release the references
        size_t _Size = _LinksToRemove._Size();

        for (size_t _I=0; _I < _Size; _I++)
        {
            _LinksToRemove[_I]->release_ref(_LinkedTarget);
        }
    }

    
    bool contains(_EType _Link)
    {
        _LockHolder _Lock(_M_lock);
        return _M_links.contains(_Link);
    }

    
    size_t count()
    {
        _LockHolder _Lock(_M_lock);
        return _M_links.count();
    }


    
    iterator begin()
    {
        return (iterator(this, 0));
    }

private:

    // Called by the iterator. This routine takes a snapshot of the links
    // in the registry and copies it to the array provided.
    void _To_array(::Concurrency::details::_Dynamic_array<_EType>& _Array)
    {
        _LockHolder _Lock(_M_lock);
        _M_iteratorCount++;

        for(_LinkRegistry::iterator _Link = _M_links.begin(); *_Link != NULL; _Link++)
        {
            _Array._Push_back(*_Link);
        }
    }

    // Internal lock used for synchronization
    _LockType _M_lock;

    // Count to indicate that an iterator is active
    volatile long _M_iteratorCount;

    // A vector of all pending link remove operations
    ::Concurrency::details::_Dynamic_array<_EType> _M_pendingRemove;

    // Underlying link registry
    _LinkRegistry _M_links;

    // Target block holding this source link manager
    ITarget<typename _Block::source_type> * volatile _M_pLinkedTarget;
};


enum message_status
{
    
    accepted,
    
    declined,
    
    postponed,
    
    missed
};


template<class _Type>
class message : public ::Concurrency::details::_Runtime_object
{
    friend class ::Concurrency::details::_Queue<message<_Type>>;

public:
    
    message(_Type const &_P) : payload(_P), _M_pNext(NULL), _M_refCount(0) { }

    
    message(_Type const &_P, runtime_object_identity _Id)
        : ::Concurrency::details::_Runtime_object(_Id), payload(_P), _M_pNext(NULL), _M_refCount(0)
    {
    }

    
    message(message const & _Msg) : payload(_Msg.payload), _M_pNext(NULL), _M_refCount(0) { }

    
    message(_In_ message const * _Msg) : payload((_Msg == NULL) ? NULL : _Msg->payload), _M_pNext(NULL), _M_refCount(0)
    {
        if (_Msg == NULL)
        {
            throw std::invalid_argument("_Msg");
        }
    }

    
    virtual ~message() { }

    
    runtime_object_identity msg_id() const
    {
        return _M_id;
    }

    
    _Type const payload;

    
    long add_ref()
    {
        return _InterlockedIncrement(&_M_refCount);
    }

    
    long remove_ref()
    {
        return _InterlockedDecrement(&_M_refCount);
    }

    
    typedef typename _Type type;

private:
    // The intrusive next pointer used by blocks that need
    // to chain messages it's holding together
    message * _M_pNext;

    // Avoid warnings about not generating assignment operators.
    message<_Type> const &operator =(message<_Type> const &);

    // A reference count for the message
    volatile long _M_refCount;
};

//**************************************************************************
// Message processor:
//**************************************************************************


template<class _Type>
class message_processor
{
public:
    
    typedef typename _Type type;

    
    virtual void async_send(_Inout_opt_ message<_Type> * _Msg) = 0;

    
    virtual void sync_send(_Inout_opt_ message<_Type> * _Msg) = 0;

    
    virtual void wait() = 0;

protected:

    
    virtual void process_incoming_message() = 0;

    
    static void __cdecl _Process_incoming_message_wrapper(void * _Data)
    {
       message_processor<_Type> * _PMessageProcessor = (message_processor<_Type> *) _Data;
       _PMessageProcessor->process_incoming_message();
    }
};


template<class _Type>
class ordered_message_processor : public message_processor<_Type>
{
public:
    
    typedef std::tr1::function<void(message<_Type> *)>  _Handler_method;

    
    typedef std::tr1::function<void(void)>  _Propagator_method;

    
    typedef _Type type;

    
    ordered_message_processor() :
      _M_queuedDataCount(0),
      _M_stopProcessing(1),
      _M_lwtCount(0),
      _M_pScheduler(NULL),
      _M_pScheduleGroup(NULL),
      _M_handler(nullptr),
      _M_processor(nullptr),
      _M_propagator(nullptr)
    {
    }

    
    virtual ~ordered_message_processor()
    {
        wait();
    }

    
    void initialize(_Inout_opt_ Scheduler * _PScheduler, _Inout_opt_ ScheduleGroup * _PScheduleGroup, _Handler_method const& _Handler)
    {
        _M_pScheduler = _PScheduler;
        _M_pScheduleGroup = _PScheduleGroup;
        _M_handler = _Handler;
        _M_stopProcessing = 0;
    }

    virtual void initialize_batched_processing(_Handler_method const& _Processor, _Propagator_method const& _Propagator)
    {
        _M_processor = _Processor;
        _M_propagator = _Propagator;
    }

    
    virtual void sync_send(_Inout_opt_ message<_Type> * _Msg)
    {
        if (_M_handler == NULL)
        {
            throw invalid_operation("sync_send called without registering a callback");
        }

        _Sync_send_helper(_Msg);
    }

    
    virtual void async_send(_Inout_opt_ message<_Type> * _Msg)
    {
        if (_M_handler == NULL)
        {
            throw invalid_operation("async_send called without registering a callback");
        }

        //
        // If there is a message to send, enqueue it in the processing queue.
        // async_send can be sent a NULL message if the block wishes to reprocess
        // the messages that are in its queue.  For example, an unbounded_buffer
        // that has its head node released after reservation.
        //
        if (_Msg != NULL)
        {
            _M_queuedMessages.push(_Msg);
        }

        if (_InterlockedIncrement(&_M_queuedDataCount) == 1)
        {
            // Indicate that an LWT is in progress. This will cause the
            // destructor to block.
            _InterlockedIncrement(&_M_lwtCount);

            if (_M_stopProcessing == 0)
            {
                _CONCRT_ASSERT(_M_lwtCount > 0);

                _Trace_agents(AGENTS_EVENT_SCHEDULE, ::Concurrency::details::_Trace_agents_get_id(this));

                TaskProc _Proc = &::Concurrency::ordered_message_processor<_Type>::_Process_incoming_message_wrapper;
#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP
                if (_M_pScheduleGroup != NULL)
                {
                    _M_pScheduleGroup->ScheduleTask(_Proc, this);
                }
                else if (_M_pScheduler != NULL)
                {
                    _M_pScheduler->ScheduleTask(_Proc, this);
                }
                else
                {
#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */
                    ::Concurrency::details::_CurrentScheduler::_ScheduleTask(_Proc, this);
#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP
                }
#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */

                // The LWT will decrement _M_lwtCount.
                return;
            }

            // If we get here then no task was scheduled. Decrement LWT count to reflect this fact
            _InterlockedDecrement(&_M_lwtCount);
        }
    }

    
    virtual void wait()
    {
        // Cease processing of any new messages
        _InterlockedIncrement(&_M_stopProcessing);

        // This spin makes sure all previously initiated message processings
        // will still process correctly.  As soon as this count reaches zero, we can
        // procede with the message block destructor.
        ::Concurrency::details::_SpinWaitBackoffNone spinWait(::Concurrency::details::_Context::_Yield);
        while(_M_lwtCount != 0)
        {
            spinWait._SpinOnce();
        }

        // Synchronize with sync_send
        {
            _NR_lock _Lock(_M_asyncSendLock);
            _Clear_queued_messages();
        }

    }

protected:

    
    virtual void process_incoming_message()
    {
        _Trace_agents(AGENTS_EVENT_START, ::Concurrency::details::_Trace_agents_get_id(this));
        long _Count = _Process_message_helper();
        _Trace_agents(AGENTS_EVENT_END, ::Concurrency::details::_Trace_agents_get_id(this), _Count);

        // Indicate that an LWT completed
        _InterlockedDecrement(&_M_lwtCount);

        // Do not access any members here. If the count goes to
        // 0 as a result of the above decrement, the object
        // could be immediately deleted.
    }

 private:

    void _Clear_queued_messages()
    {
        message<_Type> * _Msg = NULL;
        while (_M_queuedMessages.try_pop(_Msg))
        {
            delete _Msg;
        }
    }

    void _Sync_send_helper(message<_Type> * _Msg)
    {
        _NR_lock _Lock(_M_asyncSendLock);

        // Message block destructors sets the _M_stopProcessing flag to stop
        // processing any more messages. This is required to guarantee
        // that the destructor's wait_for_async_sends will complete
        if (_M_stopProcessing == 0)
        {
            if (_M_queuedDataCount > 0)
            {
                long _Count = _InterlockedExchange((volatile long *) &_M_queuedDataCount, 0);
                _Invoke_handler(_Count);
            }

            _Invoke_handler(_Msg);
        }
        else
        {
            // Destructor is running. Do not process the message
            // Delete the msg, if any.
            if (_Msg != NULL)
            {
                delete _Msg;
            }
        }

    }

    // Helper function to dequeue and process messages to any targets
    long _Process_message_helper()
    {
        _NR_lock _Lock(_M_asyncSendLock);

        long _Messages_processed = 0;

        // Do batched processing of messages
        // Read off the number of messages to process in this iteration by snapping a count
        volatile long _Count = _M_queuedDataCount;
        bool _StopProcessing = false;

        // This count could be 0 if there was both a synchronous and asynchronous
        // send occuring.  One of them could have sent all of the messages for the other
        while (_Count > 0)
        {
            // Process _Count number of messages
            _Invoke_handler(_Count);
            _Messages_processed += _Count;

            // Subtract the count and see if there are new things to process
            volatile long _Orig = _InterlockedExchangeAdd((volatile long *) &_M_queuedDataCount, -_Count);
            _CONCRT_ASSERT(_Orig >= _Count);
            if (_Orig == _Count)
            {
                // Because _Count did not change, we processed everything there is to process
                break;
            }

            if (_StopProcessing)
            {
                break;
            }

            // After reading the flag process the currently queued messages
            // Any messages received after we observe this flag (to be set) will not
            // be processed.
            _StopProcessing = (_M_stopProcessing == 0) ? false : true;

            // Snap the count and try to process more
            _Count = _M_queuedDataCount;
        }

        return _Messages_processed;
    }

    // Invoke the handler in the message block for the given
    // count
    void _Invoke_handler(long _Count)
    {
        // Process _Count number of messages
        for(int _I = 0; _I < _Count; _I++)
        {
            message<_Type> * _Msg = NULL;
            _M_queuedMessages.try_pop(_Msg);
            if (_M_processor == NULL)
            {
                // If a processor function does not exist, the message processor is using single
                // message processing rather than batched processing.  There should also be no
                // propagator function defined in this case.
                _CONCRT_ASSERT(_M_propagator == NULL);
                _M_handler(_Msg);
            }
            else
            {
                // Use the batched message processing function
                _M_processor(_Msg);
            }
        }

        // Call the handler which propagates the message(s)
        if (_M_propagator != NULL)
        {
            _M_propagator();
        }
    }

    // Invoke the message block handler for the given message
    void _Invoke_handler(message<_Type> * _Msg)
    {
        if (_M_processor == NULL)
        {
            // If a processor function does not exist, the message processor is using single
            // message processing rather than batched processing.  There should also be no
            // propagator function defined in this case.
            _CONCRT_ASSERT(_M_propagator == NULL);
            _M_handler(_Msg);
        }
        else
        {
            // Use the batched message processing function
            _M_processor(_Msg);

            // Call the handler which propagates the message(s)
            if (_M_propagator != NULL)
            {
                _M_propagator();
            }
        }
    }

 private:
    
    concurrent_queue<message<_Type> *> _M_queuedMessages;

    
    ::Concurrency::details::_NonReentrantPPLLock _M_asyncSendLock;

    
    volatile long _M_queuedDataCount;

    
    Scheduler * _M_pScheduler;

    
    ScheduleGroup * _M_pScheduleGroup;

    
    volatile long _M_stopProcessing;

    
    volatile long _M_lwtCount;

    
    _Handler_method _M_handler;

    
    _Handler_method _M_processor;

    
    _Propagator_method _M_propagator;
};


template<class _Type>
class ITarget
{
    //
    // ISource<T> is a friend class because calls to Source->link_target()
    // and Source->unlink_target() need to call their respective
    // Target->link_source() and Target->unlink_source() on the block they are
    // linking/unlinking.  Those functions are private here because we don't
    // want users calling link_source() or unlink_source() directly.  link_source/
    // unlink_source don't call respective link_target/unlink_target because an
    // infinite loop would occur.
    //
    friend class ISource<_Type>;

public:
    
    virtual ~ITarget() {}

    // It is important that calls to propagate do *not* take the same lock on an
    // internal message structure that is used by Consume and the LWT.  Doing so could
    // result in a deadlock with the Consume call.

    
    virtual message_status propagate(_Inout_opt_ message<_Type> * _PMessage, _Inout_opt_ ISource<_Type> * _PSource) = 0;

    
    virtual message_status send(_Inout_ message<_Type> * _PMessage, _Inout_ ISource<_Type> * _PSource) = 0;

    
    virtual bool supports_anonymous_source()
    {
        return false;
    }

    
    typedef typename _Type type;

    
    typedef std::tr1::function<bool(_Type const&)> filter_method;

protected:

    
    virtual void link_source(_Inout_ ISource<_Type> * _PSource) = 0;

    
    virtual void unlink_source(_Inout_ ISource<_Type> * _PSource) = 0;

    
    virtual void unlink_sources() = 0;
};


template<class _Type>
class ISource
{
public:
    
    virtual ~ISource() {}

    
    virtual void link_target(_Inout_ ITarget<_Type> * _PTarget) = 0;

    
    virtual void unlink_target(_Inout_ ITarget<_Type> * _PTarget) = 0;

    
    virtual void unlink_targets() = 0;

    
    virtual message<_Type> * accept(runtime_object_identity _MsgId, _Inout_ ITarget<_Type> * _PTarget) = 0;

    
    virtual bool reserve(runtime_object_identity _MsgId, _Inout_ ITarget<_Type> * _PTarget) = 0;

    
    virtual message<_Type> * consume(runtime_object_identity _MsgId, _Inout_ ITarget<_Type> * _PTarget) = 0;

    
    virtual void release(runtime_object_identity _MsgId, _Inout_ ITarget<_Type> * _PTarget) = 0;

    
    virtual void acquire_ref(_Inout_ ITarget<_Type> * _PTarget) = 0;

    
    virtual void release_ref(_Inout_ ITarget<_Type> * _PTarget) = 0;

    
    typedef typename _Type source_type;

protected:
    
    void _Invoke_link_source(ITarget<_Type> * _PLinkFrom)
    {
        _PLinkFrom->link_source(this);
    }

    
    void _Invoke_unlink_source(ITarget<_Type> * _PUnlinkFrom)
    {
        _PUnlinkFrom->unlink_source(this);
    }
};

//**************************************************************************
// Target Block:
//**************************************************************************


template<class _SourceLinkRegistry,
    class _MessageProcessorType = ordered_message_processor<typename _SourceLinkRegistry::type::source_type>>
class target_block : public ITarget<typename _SourceLinkRegistry::type::source_type>
{
public:

    
    typedef typename _SourceLinkRegistry::type::source_type _Source_type;

    
    typedef source_link_manager<_SourceLinkRegistry> _SourceLinkManager;

    
    typedef typename _SourceLinkManager::iterator source_iterator;

    
    target_block() : _M_pFilter(NULL), _M_fDeclineMessages(false)
    {
        _Trace_agents(AGENTS_EVENT_CREATE,
            ::Concurrency::details::_Trace_agents_get_id(this),
            ::Concurrency::details::_Trace_agents_get_id(&_M_messageProcessor));
    }

    
    virtual ~target_block()
    {
        // All sources should have been unlinked
        _CONCRT_ASSERT(_M_connectedSources.count() == 0);
        delete _M_pFilter;

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

    
    virtual message_status propagate(_Inout_opt_ message<_Source_type> * _PMessage, _Inout_opt_ ISource<_Source_type> * _PSource)
    {
        // It is important that calls to propagate do *not* take the same lock on the
        // internal structure that is used by <c>consume</c> and the LWT.  Doing so could
        // result in a deadlock.

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

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

        if (_M_fDeclineMessages)
        {
            return declined;
        }

        if (_M_pFilter != NULL && !(*_M_pFilter)(_PMessage->payload))
        {
            return declined;
        }

        return propagate_message(_PMessage, _PSource);
    }

    
    virtual message_status send(_Inout_ message<_Source_type> * _PMessage, _Inout_ ISource<_Source_type> * _PSource)
    {
        if (_PMessage == NULL)
        {
            throw std::invalid_argument("_PMessage");
        }

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

        if (_M_fDeclineMessages)
        {
            return declined;
        }

        if (_M_pFilter != NULL && !(*_M_pFilter)(_PMessage->payload))
        {
            return declined;
        }

        return send_message(_PMessage, _PSource);
    }

protected:

    
    virtual message_status propagate_message(_Inout_ message<_Source_type> * _PMessage, _Inout_ ISource<_Source_type> * _PSource) = 0;

    
    virtual message_status send_message(_Inout_ message<_Source_type> *, _Inout_ ISource<_Source_type> *)
    {
        // By default we do not allow send()
        return declined;
    }

    
    virtual void link_source(_Inout_ ISource<_Source_type> * _PSource)
    {
        _M_connectedSources.add(_PSource);
        _Trace_agents(AGENTS_EVENT_LINK,
            ::Concurrency::details::_Trace_agents_get_id(_PSource),
            ::Concurrency::details::_Trace_agents_get_id(this));
    }

    
    virtual void unlink_source(_Inout_ ISource<_Source_type> * _PSource)
    {
        _Trace_agents(AGENTS_EVENT_UNLINK,
            ::Concurrency::details::_Trace_agents_get_id(_PSource),
            ::Concurrency::details::_Trace_agents_get_id(this));

        _M_connectedSources.remove(_PSource);
    }

    
    virtual void unlink_sources()
    {
        for (source_iterator _Iter = _M_connectedSources.begin(); *_Iter != NULL; ++_Iter)
        {
             ISource<_Source_type> * _PSource = *_Iter;
             _PSource->unlink_target(this);
        }
    }

    
    virtual void process_message(message<_Source_type> *)
    {
    }

    //
    // Utility routines
    //

    
    void register_filter(filter_method const& _Filter)
    {
        if (_Filter != NULL)
        {
            _M_pFilter = new filter_method(_Filter);
        }
    }

    
    void decline_incoming_messages()
    {
        _M_fDeclineMessages = true;
    }

    
    void initialize_target(_Inout_opt_ Scheduler * _PScheduler = NULL, _Inout_opt_ ScheduleGroup * _PScheduleGroup = NULL)
    {
        // Register a callback with the processor
        _M_messageProcessor.initialize(_PScheduler, _PScheduleGroup,
            // Processing and Propagating function used by ordered_message_processors
            [this](message<_Source_type> * _PMessage)
            {
                // Handle message by calling process_message to maintain CRT100 compatibility
                this->process_message(_PMessage);
            });

        // Register this target block as the owner of the connected sources
        _M_connectedSources.register_target_block(this);
    }

    
    void enable_batched_processing()
    {
        _M_messageProcessor.initialize_batched_processing(
            // Processing function used by CRT110
            [this](message<_Source_type> * _PMessage)
            {
                // Handle message through new process_input_message to use CRT110 batch processing
                this->process_input_messages(_PMessage);
            }, nullptr);
    }

    
    void async_send(_Inout_opt_ message<_Source_type> * _PMessage)
    {
        _M_messageProcessor.async_send(_PMessage);
    }

    
    void sync_send(_Inout_opt_ message<_Source_type> * _PMessage)
    {
        _M_messageProcessor.sync_send(_PMessage);
    }

    
    void wait_for_async_sends()
    {
        // Decline new messages to ensure that messages are not dropped during the wait
        decline_incoming_messages();

        _M_messageProcessor.wait();
    }

    
    void remove_sources()
    {
        wait_for_async_sends();

        unlink_sources();
    }

    
    virtual void process_input_messages(_Inout_ message<_Source_type> * _PMessage)
    {
        throw invalid_operation("To use batched processing, you must override process_input_messages in the message block.");
    }

    
    _SourceLinkManager _M_connectedSources;

    
    filter_method * _M_pFilter;

    
    bool _M_fDeclineMessages;

    
    _MessageProcessorType _M_messageProcessor;
};

//**************************************************************************
// Source Block:
//**************************************************************************


template<class _TargetLinkRegistry,
    class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>
class source_block : public ISource<typename _TargetLinkRegistry::type::type>
{
public:

    
    typedef typename _TargetLinkRegistry::type::type _Target_type;

    
    typedef typename _TargetLinkRegistry::iterator target_iterator;

    
    source_block() :
      _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));
    }

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

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

    
    virtual void link_target(_Inout_ ITarget<_Target_type> * _PTarget)
    {
        _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);
    }

    
    virtual void unlink_target(_Inout_ ITarget<_Target_type> * _PTarget)
    {
        _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);
        }
    }

    
    virtual void unlink_targets()
    {
        _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);
    }

    
    virtual message<_Target_type> * accept(runtime_object_identity _MsgId, _Inout_ ITarget<_Target_type> * _PTarget)
    {
        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);
    }

    
    virtual bool reserve(runtime_object_identity _MsgId, _Inout_ ITarget<_Target_type> * _PTarget)
    {
        _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;
    }

    
    virtual message<_Target_type> * consume(runtime_object_identity _MsgId, _Inout_ ITarget<_Target_type> * _PTarget)
    {
        _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;
    }

    
    virtual void release(runtime_object_identity _MsgId, _Inout_ ITarget<_Target_type> * _PTarget)
    {
        _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();
    }

    
    virtual void acquire_ref(_Inout_ ITarget<_Target_type> *)
    {
        _InterlockedIncrement(&_M_referenceCount);
    }

    
    virtual void release_ref(_Inout_ ITarget<_Target_type> * _PTarget)
    {
        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
    }

protected:

    //
    // Protected methods that a derived class can override to customize
    // the functionality
    //

    
    virtual void link_target_notification(_Inout_ ITarget<_Target_type> *)
    {
        // By default, we restart propagation if there is no pending resrvation
        if (_M_pReservedFor == NULL)
        {
            propagate_to_any_targets(NULL);
        }
    }

    
    virtual void unlink_target_notification(_Inout_ ITarget<_Target_type> * _PTarget)
    {
        // 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);
        }
    }

    
    virtual message<_Target_type> * accept_message(runtime_object_identity _MsgId) = 0;

    
    virtual bool reserve_message(runtime_object_identity _MsgId) = 0;

    
    virtual message<_Target_type> * consume_message(runtime_object_identity _MsgId) = 0;

    
    virtual void release_message(runtime_object_identity _MsgId) = 0;

    
    virtual void resume_propagation() = 0;

    
    virtual void process_input_messages(_Inout_ message<_Target_type> * _PMessage)
    {
        // source_blocks do not need to process anything
    }

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

    
    virtual void propagate_to_any_targets(_Inout_opt_ message<_Target_type> * _PMessage)
    {
        throw invalid_operation("To use ordered message processing, you must override propagate_to_any_targets in the message block.");
    }

    //
    // Utility routines
    //
    
    void initialize_source(_Inout_opt_ Scheduler * _PScheduler = NULL, _Inout_opt_ ScheduleGroup * _PScheduleGroup = NULL)
    {
        // Register a callback
        _M_messageProcessor.initialize(_PScheduler, _PScheduleGroup,
            [this](message<_Target_type> * _PMessage)
            {
                this->_Handle_message(_PMessage);
            });
    }

    
    void enable_batched_processing()
    {
        // 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();
            });
    }

    
    virtual void sync_send(_Inout_opt_ message<_Target_type> * _Msg)
    {
        // Caller shall not be holding any locks when calling this routine
        _M_messageProcessor.sync_send(_Msg);
    }

    
    virtual void async_send(_Inout_opt_ message<_Target_type> * _Msg)
    {
        _M_messageProcessor.async_send(_Msg);
    }

    
    void wait_for_outstanding_async_sends()
    {
        _M_messageProcessor.wait();
    }

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

        unlink_targets();

        _Wait_on_ref();
    }

    //
    // Protected members
    //

    
    ITarget<_Target_type> * _M_pReservedFor;

    
    runtime_object_identity _M_reservedId;

    
    _TargetLinkRegistry _M_connectedTargets;

    
    _MessageProcessorType _M_messageProcessor;

private:



    // Message handler callback for the propagator. Invokes propagate_to_any_targets
    // which derived classes should implement.
    
    void _Handle_message(message<_Target_type> * _PMessage)
    {
        // Hold a lock to synchronize with unlink targets
        _R_lock _Lock(_M_internalLock);
        propagate_to_any_targets(_PMessage);
    }

    // Message handler callback for the processor. Invokes process_input_messages
    // which derived classes should implement.
    
    void _Process_message(message<_Target_type> * _PMessage)
    {
        // Don't need a lock to process the message
        process_input_messages(_PMessage);
    }

    // Message handler callback for the propagator. Invokes propagate_output_messages
    // which derived classes should implement.
    
    void _Propagate_message()
    {
        // Hold a lock to synchronize with unlink targets
        _R_lock _Lock(_M_internalLock);
        propagate_output_messages();
    }

    // Wait for the reference on this block to drop to zero
    
    void _Wait_on_ref(long _RefCount = 0)
    {
        ::Concurrency::details::_SpinWaitBackoffNone spinWait;
        while(_M_referenceCount != _RefCount)
        {
            spinWait._SpinOnce();
        }
    }

    // Private Data members

    
    ::Concurrency::details::_ReentrantPPLLock _M_internalLock;

    volatile long _M_referenceCount;

};

//**************************************************************************
// Propagator (source and target) Block:
//**************************************************************************

template<class _TargetLinkRegistry, class _SourceLinkRegistry,
    class _MessageProcessorType = ordered_message_processor<typename _TargetLinkRegistry::type::type>>
class propagator_block : public source_block<_TargetLinkRegistry, _MessageProcessorType>, public ITarget<typename _SourceLinkRegistry::type::source_type>
{
public:

    
    typedef typename _SourceLinkRegistry::type::source_type _Source_type;

    
    typedef source_link_manager<_SourceLinkRegistry> _SourceLinkManager;

    
    typedef typename _SourceLinkManager::iterator source_iterator;

    
    propagator_block() : _M_pFilter(NULL), _M_fDeclineMessages(false)
    {
    }

    
    virtual ~propagator_block()
    {
        remove_network_links();

        delete _M_pFilter;
    }

    
    virtual message_status propagate(_Inout_opt_ message<_Source_type> * _PMessage, _Inout_opt_ ISource<_Source_type> * _PSource)
    {
        // It is important that calls to propagate do *not* take the same lock on the
        // internal structure that is used by <c>consume</c> and the LWT.  Doing so could
        // result in a deadlock.

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

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

        if (_M_fDeclineMessages)
        {
            return declined;
        }

        if (_M_pFilter != NULL && !(*_M_pFilter)(_PMessage->payload))
        {
            return declined;
        }

        return propagate_message(_PMessage, _PSource);
    }

    
    virtual message_status send(_Inout_ message<_Source_type> * _PMessage, _Inout_ ISource<_Source_type> * _PSource)
    {
        if (_PMessage == NULL)
        {
            throw std::invalid_argument("_PMessage");
        }

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

        if (_M_fDeclineMessages)
        {
            return declined;
        }

        if (_M_pFilter != NULL && !(*_M_pFilter)(_PMessage->payload))
        {
            return declined;
        }

        return send_message(_PMessage, _PSource);
    }

protected:

    
    virtual message_status propagate_message(_Inout_ message<_Source_type> * _PMessage, _Inout_ ISource<_Source_type> * _PSource) = 0;

    
    virtual message_status send_message(_Inout_ message<_Source_type> *, _Inout_ ISource<_Source_type> *)
    {
        // By default we do not allow send()
        return declined;
    }

    
    virtual void link_source(_Inout_ ISource<_Source_type> * _PSource)
    {
        _M_connectedSources.add(_PSource);
        _Trace_agents(AGENTS_EVENT_LINK,
            ::Concurrency::details::_Trace_agents_get_id(_PSource),
            ::Concurrency::details::_Trace_agents_get_id(this));
    }

    
    virtual void unlink_source(_Inout_ ISource<_Source_type> * _PSource)
    {
        _Trace_agents(AGENTS_EVENT_UNLINK,
            ::Concurrency::details::_Trace_agents_get_id(_PSource),
            ::Concurrency::details::_Trace_agents_get_id(this));

        _M_connectedSources.remove(_PSource);
    }

    
    virtual void unlink_sources()
    {
        for (source_iterator _Iter = _M_connectedSources.begin(); *_Iter != NULL; ++_Iter)
        {
             ISource<_Source_type> * _PSource = *_Iter;
             _PSource->unlink_target(this);
        }
    }

    //
    // Utility routines
    //

    
    virtual void process_input_messages(_Inout_ message<_Target_type> * _PMessage)
    {
        throw invalid_operation("To use batched processing, you must override process_input_messages in the message block.");
    }

    
    void initialize_source_and_target(_Inout_opt_ Scheduler * _PScheduler = NULL, _Inout_opt_ ScheduleGroup * _PScheduleGroup = NULL)
    {
        initialize_source(_PScheduler, _PScheduleGroup);

        // Register this propagator block as the owner of the connected sources
        _M_connectedSources.register_target_block(this);
    }

    
    void register_filter(filter_method const& _Filter)
    {
        if (_Filter != NULL)
        {
            _M_pFilter = new filter_method(_Filter);
        }
    }

    
    void decline_incoming_messages()
    {
        _M_fDeclineMessages = true;
    }

    
    void remove_network_links()
    {
        // Decline messages while the links are being removed
        decline_incoming_messages();

        // Remove all the target links. This waits for
        // all outstanding async propagation operations.
        remove_targets();

        // unlink all sources. The above steps guarantee that
        // they can be removed safely.
        unlink_sources();
    }

    
    _SourceLinkManager _M_connectedSources;

    
    filter_method * _M_pFilter;

    
    volatile bool _M_fDeclineMessages;
};

//**************************************************************************
// Unbounded Buffers:
//**************************************************************************


template<class _Type>
class unbounded_buffer : public propagator_block<multi_link_registry<ITarget<_Type>>, multi_link_registry<ISource<_Type>>>
{
public:
    
    unbounded_buffer() :
      _M_fForceRepropagation(false)
    {
        initialize_source_and_target();
        enable_batched_processing();
    }

    
    unbounded_buffer(filter_method const& _Filter) :
      _M_fForceRepropagation(false)
    {
        initialize_source_and_target();
        enable_batched_processing();
        register_filter(_Filter);
    }

#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP
    
    unbounded_buffer(Scheduler& _PScheduler) :
      _M_fForceRepropagation(false)
    {
        initialize_source_and_target(&_PScheduler);
        enable_batched_processing();
    }

    
    unbounded_buffer(Scheduler& _PScheduler, filter_method const& _Filter) :
      _M_fForceRepropagation(false)
    {
        initialize_source_and_target(&_PScheduler);
        enable_batched_processing();
        register_filter(_Filter);
    }

    
    unbounded_buffer(ScheduleGroup& _PScheduleGroup) :
      _M_fForceRepropagation(false)
    {
        initialize_source_and_target(NULL, &_PScheduleGroup);
        enable_batched_processing();
    }

    
    unbounded_buffer(ScheduleGroup& _PScheduleGroup, filter_method const& _Filter) :
      _M_fForceRepropagation(false)
    {
        initialize_source_and_target(NULL, &_PScheduleGroup);
        enable_batched_processing();
        register_filter(_Filter);
    }
#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */

    
    ~unbounded_buffer()
    {
        // Remove all links
        remove_network_links();

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

    
    bool enqueue(_Type const& _Item)
    {
        return Concurrency::send<_Type>(this, _Item);
    }

    
    _Type dequeue()
    {
        return receive<_Type>(this);
    }


protected:

    //
    // propagator_block protected function implementations
    //

    
    virtual message_status propagate_message(_Inout_ message<_Type> * _PMessage, _Inout_ ISource<_Type> * _PSource)
    {
        // It is important that calls to propagate do *not* take the same lock on the
        // internal structure that is used by <c>consume</c> and the LWT.  Doing so could
        // result in a deadlock.

        message_status _Result = accepted;

        // Accept the message being propagated
        _PMessage = _PSource->accept(_PMessage->msg_id(), this);

        if (_PMessage != NULL)
        {
            async_send(_PMessage);
        }
        else
        {
            _Result = missed;
        }

        return _Result;
    }

    
    virtual message_status send_message(_Inout_ message<_Type> * _PMessage, _Inout_ ISource<_Type> * _PSource)
    {
        _PMessage = _PSource->accept(_PMessage->msg_id(), this);

        if (_PMessage != NULL)
        {
            sync_send(_PMessage);
        }
        else
        {
            return missed;
        }

        return accepted;
    }

    
    virtual bool supports_anonymous_source()
    {
        return true;
    }

    
    virtual message<_Type> * accept_message(runtime_object_identity _MsgId)
    {
        //
        // Peek at the head message in the message buffer.  If the IDs match
        // dequeue and transfer ownership
        //
        message<_Type> * _Msg = NULL;

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

        return _Msg;
    }

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

    
    virtual message<_Type> * consume_message(runtime_object_identity _MsgId)
    {
        // By default, accept the message
        return accept_message(_MsgId);
    }

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

    
    virtual void resume_propagation()
    {
        // If there are any messages in the buffer, propagate them out
        if (_M_messageBuffer._Count() > 0)
        {
            // Set the flag to force a repropagation. This flag is cleared when a propagation happens
            // The only functions that call this are release, consume, and link_target, all of which
            // hold the internal lock, so the flag is guaranteed to be read by propagation, which also
            // holds the same lock.
            _M_fForceRepropagation = true;

            // async send a NULL value to initiate the repropagation
            async_send(NULL);
        }
    }

    
    virtual void link_target_notification(_Inout_ ITarget<_Type> * _PTarget)
    {
        // If the message queue is blocked due to reservation
        // there is no need to do any message propagation
        if (_M_pReservedFor != NULL)
        {
            return;
        }

        message<_Type> * _Msg = _M_messageBuffer._Peek();

        if (_Msg != NULL)
        {
            // Propagate the head message to the new target
            message_status _Status = _PTarget->propagate(_Msg, this);

            if (_Status == accepted)
            {
                // The target accepted the message, restart propagation.
                _Propagate_priority_order(_M_messageBuffer);
            }

            // If the status is anything other than accepted, then leave
            // the message queue blocked.
        }
    }

    virtual void process_input_messages(_Inout_ message<_Type> * _PMessage)
    {
        if (_PMessage != NULL)
        {
            _M_processedMessages._Enqueue(_PMessage);
        }
    }

    
    virtual void propagate_output_messages()
    {
        // Move the messages from the processedMessages queue to the internal storage
        // to make them ready for propagating out

        // If there are messages in the message queue, the queue is blocked and a
        // propagation should not happen unless it has been forced using resume_propagation
        bool _FIsBlocked = (_M_messageBuffer._Count() > 0);

        for(;;)
        {
            message<_Type> * _PInputMessage = _M_processedMessages._Dequeue();
            if(_PInputMessage == NULL)
            {
                break;
            }
            _M_messageBuffer._Enqueue(_PInputMessage);
        }

        if (_M_fForceRepropagation == false && _FIsBlocked == true)
        {
            return;
        }

        // Reset the repropagation flag because a propagation has started.
        _M_fForceRepropagation = false;

        // Attempt to propagate messages to all the targets
        _Propagate_priority_order(_M_messageBuffer);
    }

private:

    
    void _Propagate_priority_order(::Concurrency::details::_Queue<message<_Target_type>> & _MessageBuffer)
    {
        message<_Target_type> * _Msg = _MessageBuffer._Peek();

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

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

            // Always start from the first target that linked
            for (target_iterator _Iter = _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 (_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();
        }
    }

    
    void _Delete_stored_messages()
    {
        // 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<_Type> * _Msg = _M_messageBuffer._Dequeue();
            if (_Msg == NULL)
            {
                break;
            }
            delete _Msg;
        }
    }

    
    ::Concurrency::details::_Queue<message<_Type>> _M_processedMessages;

    
    ::Concurrency::details::_Queue<message<_Type>> _M_messageBuffer;

    
    bool _M_fForceRepropagation;

private:
    //
    // Hide assignment operator and copy constructor
    //
    unbounded_buffer const &operator =(unbounded_buffer const&);  // no assignment operator
    unbounded_buffer(unbounded_buffer const &);                   // no copy constructor
};

//**************************************************************************
// Overwrite Buffers:
//**************************************************************************


template<class _Type>
class overwrite_buffer : public propagator_block<multi_link_registry<ITarget<_Type>>, multi_link_registry<ISource<_Type>>>
{
public:
    
    overwrite_buffer() :
        _M_fIsInitialized(false), _M_pMessage(NULL), _M_pReservedMessage(NULL)
    {
        initialize_source_and_target();
    }

    
    overwrite_buffer(filter_method const& _Filter) :
        _M_fIsInitialized(false), _M_pMessage(NULL), _M_pReservedMessage(NULL)
    {
        initialize_source_and_target();
        register_filter(_Filter);
    }

#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP
    
    overwrite_buffer(Scheduler& _PScheduler) :
        _M_fIsInitialized(false), _M_pMessage(NULL), _M_pReservedMessage(NULL)
    {
        initialize_source_and_target(&_PScheduler);
    }

    
    overwrite_buffer(Scheduler& _PScheduler,
        filter_method const& _Filter) :
        _M_fIsInitialized(false), _M_pMessage(NULL), _M_pReservedMessage(NULL)
    {
        initialize_source_and_target(&_PScheduler);
        register_filter(_Filter);
    }

    
    overwrite_buffer(ScheduleGroup& _PScheduleGroup) :
        _M_fIsInitialized(false), _M_pMessage(NULL), _M_pReservedMessage(NULL)
    {
        initialize_source_and_target(NULL, &_PScheduleGroup);
    }

    
    overwrite_buffer(ScheduleGroup& _PScheduleGroup,
        filter_method const& _Filter) :
        _M_fIsInitialized(false), _M_pMessage(NULL), _M_pReservedMessage(NULL)
    {
        initialize_source_and_target(NULL, &_PScheduleGroup);
        register_filter(_Filter);
    }
#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */

    
    ~overwrite_buffer()
    {
        // Remove all links that are targets of this overwrite_buffer
        remove_network_links();

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

    
    bool has_value() const
    {
        return _M_fIsInitialized != 0;
    }

    
    _Type value()
    {
        return receive<_Type>(this);
    }

protected:

    //
    // propagator_block protected function implementation
    //

    
    virtual message_status propagate_message(_Inout_ message<_Type> * _PMessage, _Inout_ ISource<_Type> * _PSource)
    {
        // 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 _Result = accepted;

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

        //
        // If message was accepted, set the member variables for
        // this block and start the asynchronous propagation task
        //
        if (_PMessage != NULL)
        {
            // Add a reference for the async_send holding the message
            _PMessage->add_ref();

            async_send(_PMessage);
        }
        else
        {
            _Result = missed;
        }

        return _Result;
    }

    
    virtual message_status send_message(_Inout_ message<_Type> * _PMessage, _Inout_ ISource<_Type> * _PSource)
    {
        _PMessage = _PSource->accept(_PMessage->msg_id(), this);

        //
        // If message was accepted, set the member variables for
        // this block and start the asynchronous propagation task
        //
        if (_PMessage != NULL)
        {
            // Add a reference for the sync_send holding the message
            _PMessage->add_ref();

            sync_send(_PMessage);
        }
        else
        {
            return missed;
        }

        return accepted;
    }

    
    virtual bool supports_anonymous_source()
    {
        return true;
    }

    
    virtual message<_Type> * accept_message(runtime_object_identity _MsgId)
    {
        //
        // If the internal message has not yet been initialized yet, return NULL
        //
        if (_M_pMessage == NULL)
        {
            return NULL;
        }

        //
        // Instead of returning the internal message, we return a copy of the
        // message stored.
        //
        // Because we are returning a copy, the accept routine for an overwritebuffer
        // does not need to grab the internalLock
        //
        message<_Type> * _Msg = NULL;
        if (_M_pMessage->msg_id() == _MsgId)
        {
            _Msg = new message<_Type>(_M_pMessage->payload);
        }

        return _Msg;
    }

    
    virtual bool reserve_message(runtime_object_identity _MsgId)
    {
        // Ensure that this message currently exists in the overwrite buffer
        if (_M_pMessage == NULL || _M_pMessage->msg_id() != _MsgId)
        {
            return false;
        }

        // Can only reserve one message, any other blocks trying to reserve
        // will return false
        _CONCRT_ASSERT(_M_pReservedMessage == NULL);

        // Save this message away
        _M_pReservedMessage = _M_pMessage;

        // Add a reference for this message to prevent deletion
        _M_pReservedMessage->add_ref();

        return true;
    }

    
    virtual message<_Type> * consume_message(runtime_object_identity _MsgId)
    {
        // Leave and return NULL if this msgId doesn't match the reserved message
        // Otherwise this is a pull of a later overwritten message, and messages
        // could them appear out of order.
        if (_M_pReservedMessage != NULL && _M_pReservedMessage->msg_id() != _MsgId)
        {
            return NULL;
        }
        // This redundant assert is specifically to make the /analyze switch happy, which cannot recognize the same assertion above in if stmnt.
        _CONCRT_ASSERT( _M_pReservedMessage != NULL );

        _Type _Payload = _M_pReservedMessage->payload;

        // Take the reserved message
        message<_Type> * _Result =  new message<_Type>(_Payload);

        if (_M_pReservedMessage->remove_ref() == 0)
        {
            delete _M_pReservedMessage;
        }
        _M_pReservedMessage = NULL;

        return _Result;
    }

    
    virtual void release_message(runtime_object_identity _MsgId)
    {
        _CONCRT_ASSERT(_M_fIsInitialized);
        _CONCRT_ASSERT(_M_pReservedMessage != NULL);

        if (_MsgId != _M_pReservedMessage->msg_id())
        {
            throw message_not_found();
        }

        if (_M_pReservedMessage->remove_ref() == 0)
        {
            delete _M_pReservedMessage;
        }
        _M_pReservedMessage = NULL;
    }

    
    virtual void resume_propagation()
    {
        // On reservation we do not stop propagation. So there
        // is nothing to be done to resume propagation.
    }

    
    virtual void link_target_notification(_Inout_ ITarget<_Type> * _PTarget)
    {
        // If there is a message available already, propagate it
        if (_M_pMessage != NULL)
        {
            _PTarget->propagate(_M_pMessage, this);
        }
    }

    
    virtual void propagate_to_any_targets(_Inout_ message<_Type> * _PMessage)
    {
        // Move the message from the queuedMessages Buffer to the internal storage

        // Add a reference for the overwrite_buffer holding the message
        _PMessage->add_ref();

        if (_M_pMessage != NULL)
        {
            if (_M_pMessage->remove_ref() == 0)
            {
                delete _M_pMessage;
            }
        }

        _M_pMessage = _PMessage;

        // Now that message has been received, set this block as initialized
        _M_fIsInitialized = true;

        for (target_iterator _Iter = _M_connectedTargets.begin(); *_Iter != NULL; ++_Iter)
        {
            // Overwrite buffers can propagate its message out
            // to any number of Targets

            ITarget<_Type> * _PTarget = *_Iter;
            _PTarget->propagate(_PMessage, this);
        }

        if (_PMessage->remove_ref() == 0)
        {
            delete _PMessage;
        }
    }

private:

    
    void _Delete_stored_messages()
    {
        // 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

        // The messages for an overwrite buffer are deleted when overwritten
        // through reference counting.  This final check is put in place in
        // case any message still exists in the buffer when the overwrite_buffer
        // is deleted.  The reference count of this message has not yet reached
        // zero because it hasn't been overwritten yet.  It is safe because of
        // we have finished all propagation.
        if (_M_pMessage != NULL)
        {
            // A block can only have a reserved message after receiving a message
            // at some point, so it must be within the above if-clause.
            // Now delete the reserved message if it is non-NULL and different from
            // the saved internal message
            if (_M_pReservedMessage != NULL && _M_pReservedMessage != _M_pMessage)
            {
                delete _M_pReservedMessage;
            }
            delete _M_pMessage;
        }
    }

    //
    //  Private Data Members
    //

    // The message being stored
    message<_Type> * _M_pMessage;

    // The message being reserved
    message<_Type> * _M_pReservedMessage;

    // The marker for whether the overwrite buffer has already been initialized
    volatile bool _M_fIsInitialized;

private:
    //
    // Hide assignment operator and copy constructor
    //
    overwrite_buffer const &operator =(overwrite_buffer const&);  // no assignment operator
    overwrite_buffer(overwrite_buffer const &);                   // no copy constructor
};

//**************************************************************************
// Call:
//**************************************************************************


template<class _Type, class _FunctorType = std::tr1::function<void(_Type const&)>>
class call : public target_block<multi_link_registry<ISource<_Type>>>
{
    
    typedef _FunctorType _Call_method;

public:
    
    call(_Call_method const& _Func) :
        _M_pFunc(_Func)
    {
        initialize_target();
        enable_batched_processing();
    }

    
    call(_Call_method const& _Func,
        filter_method const& _Filter) :
        _M_pFunc(_Func)
    {
        initialize_target();
        enable_batched_processing();
        register_filter(_Filter);
    }

#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP
    
    call(Scheduler& _PScheduler,
        _Call_method const& _Func) :
        _M_pFunc(_Func)
    {
        initialize_target(&_PScheduler);
        enable_batched_processing();
    }

    
    call(Scheduler& _PScheduler,
        _Call_method const& _Func,
        filter_method const& _Filter) :
        _M_pFunc(_Func)
    {
        initialize_target(&_PScheduler);
        enable_batched_processing();
        register_filter(_Filter);
    }

    
    call(ScheduleGroup& _PScheduleGroup,
        _Call_method const& _Func) :
        _M_pFunc(_Func)
    {
        initialize_target(NULL, &_PScheduleGroup);
        enable_batched_processing();
    }

    
    call(ScheduleGroup& _PScheduleGroup,
        _Call_method const& _Func,
        filter_method const& _Filter) :
        _M_pFunc(_Func)
    {
        initialize_target(NULL, &_PScheduleGroup);
        enable_batched_processing();
        register_filter(_Filter);
    }
#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */

    
    ~call()
    {
        remove_sources();
    }

protected:

    //
    // target_block protected function implementations
    //

    
    virtual message_status propagate_message(_Inout_ message<_Type> * _PMessage, _Inout_ ISource<_Type> * _PSource)
    {
        // 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 _Result = accepted;

        //
        // Accept the message being propagated
        // Note: depending on the source block propagating the message
        // this may not necessarily be the same message (pMessage) first
        // passed into the function.
        //
        _PMessage = _PSource->accept(_PMessage->msg_id(), this);

        if (_PMessage != NULL)
        {
            async_send(_PMessage);
        }
        else
        {
            _Result = missed;
        }

        return _Result;
    }

    
    virtual message_status send_message(_Inout_ message<_Type> * _PMessage, _Inout_ ISource<_Type> * _PSource)
    {
        message_status _Result = accepted;

        //
        // Accept the message being propagated
        // Note: depending on the source block propagating the message
        // this may not necessarily be the same message (pMessage) first
        // passed into the function.
        //
        _PMessage = _PSource->accept(_PMessage->msg_id(), this);

        if (_PMessage != NULL)
        {
            sync_send(_PMessage);
        }
        else
        {
            _Result = missed;
        }

        return _Result;
    }

    
    virtual bool supports_anonymous_source()
    {
        return true;
    }

    
    virtual void process_message(_Inout_ message<_Type> * _PMessage)
    {
        // No longer necessary with CRT110 change
    }

    
    virtual void process_input_messages(_Inout_ message<_Type> * _PMessage)
    {
        // Invoke the function provided by the user
        _CONCRT_ASSERT(_PMessage != NULL);
        _M_pFunc(_PMessage->payload);
        delete _PMessage;
    }

private:

    //
    //  Private Data Members
    //

    // The call method called by this block
    _Call_method _M_pFunc;

private:
    //
    // Hide assignment operator and copy constructor
    //
    call const &operator =(call const&);  // no assignment operator
    call(call const &);                   // no copy constructor
};

//**************************************************************************
// Transformer:
//**************************************************************************


template<class _Input, class _Output>
class transformer : public propagator_block<single_link_registry<ITarget<_Output>>, multi_link_registry<ISource<_Input>>>
{
    typedef std::tr1::function<_Output(_Input const&)> _Transform_method;

public:
    
    transformer(_Transform_method const& _Func,
        _Inout_opt_ ITarget<_Output> * _PTarget = NULL) :
        _M_pFunc(_Func)
    {
        initialize_source_and_target();

        if (_PTarget != NULL)
        {
            link_target(_PTarget);
        }
    }

    
    transformer(_Transform_method const& _Func,
        _Inout_opt_ ITarget<_Output> * _PTarget,
        filter_method const& _Filter) :
        _M_pFunc(_Func)
    {
        initialize_source_and_target();
        register_filter(_Filter);

        if (_PTarget != NULL)
        {
            link_target(_PTarget);
        }
    }

#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP
    
    transformer(Scheduler& _PScheduler,
        _Transform_method const& _Func,
        _Inout_opt_ ITarget<_Output> * _PTarget = NULL) :
        _M_pFunc(_Func)
    {
        initialize_source_and_target(&_PScheduler);

        if (_PTarget != NULL)
        {
            link_target(_PTarget);
        }
    }

    
    transformer(Scheduler& _PScheduler,
        _Transform_method const& _Func,
        _Inout_opt_ ITarget<_Output> * _PTarget,
        filter_method const& _Filter) :
        _M_pFunc(_Func)
    {
        initialize_source_and_target(&_PScheduler);
        register_filter(_Filter);

        if (_PTarget != NULL)
        {
            link_target(_PTarget);
        }
    }

    
    transformer(ScheduleGroup& _PScheduleGroup,
        _Transform_method const& _Func,
        _Inout_opt_ ITarget<_Output> * _PTarget = NULL) :
        _M_pFunc(_Func)
    {
        initialize_source_and_target(NULL, &_PScheduleGroup);

        if (_PTarget != NULL)
        {
            link_target(_PTarget);
        }
    }

    
    transformer(ScheduleGroup& _PScheduleGroup,
        _Transform_method const& _Func,
        _Inout_opt_ ITarget<_Output> * _PTarget,
        filter_method const& _Filter) :
        _M_pFunc(_Func)
    {
        initialize_source_and_target(NULL, &_PScheduleGroup);
        register_filter(_Filter);

        if (_PTarget != NULL)
        {
            link_target(_PTarget);
        }
    }
#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */

    
    ~transformer()
    {
        // Remove all links
        remove_network_links();

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

protected:

    // Propagator_block protected function implementations

    
    virtual message_status propagate_message(_Inout_ message<_Input> * _PMessage, _Inout_ ISource<_Input> * _PSource)
    {
        // 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 _Result = accepted;

        //
        // Accept the message being propagated
        // Note: depending on the source block propagating the message
        // this may not necessarily be the same message (pMessage) first
        // passed into the function.
        //
        _PMessage = _PSource->accept(_PMessage->msg_id(), this);

        if (_PMessage != NULL)
        {
            // Enqueue the input message
            _M_inputMessages.push(_PMessage);
            async_send(NULL);
        }
        else
        {
            _Result = missed;
        }

        return _Result;
    }

    
    virtual message_status send_message(_Inout_ message<_Input> * _PMessage, _Inout_ ISource<_Input> * _PSource)
    {
        _PMessage = _PSource->accept(_PMessage->msg_id(), this);

        if (_PMessage != NULL)
        {
            // Enqueue the input message
            _M_inputMessages.push(_PMessage);
            sync_send(NULL);
        }
        else
        {
            return missed;
        }

        return accepted;
    }

    
    virtual bool supports_anonymous_source()
    {
        return true;
    }

    
    virtual message<_Output> * accept_message(runtime_object_identity _MsgId)
    {
        //
        // Peek at the head message in the message buffer.  If the IDs match
        // dequeue and transfer ownership
        //
        message<_Output> * _Msg = NULL;

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

        return _Msg;
    }

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

    
    virtual message<_Output> * consume_message(runtime_object_identity _MsgId)
    {
        // By default, accept the message
        return accept_message(_MsgId);
    }

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

    
    virtual void resume_propagation()
    {
        // If there are any messages in the buffer, propagate them out
        if (_M_messageBuffer._Count() > 0)
        {
            // async send a NULL value to initiate the repropagation
            async_send(NULL);
        }
    }

    
    virtual void link_target_notification(_Inout_ ITarget<_Output> *)
    {
        // If the message queue is blocked due to reservation
        // there is no need to do any message propagation
        if (_M_pReservedFor != NULL)
        {
            return;
        }

        _Propagate_priority_order(_M_messageBuffer);
    }

    
    virtual void propagate_to_any_targets(_Inout_opt_ message<_Output> *)
    {
        message<_Output> * _Msg = NULL;

        // Process input message.
        message<_Input> * _PInputMessage = NULL;
        _M_inputMessages.try_pop(_PInputMessage);

        if (_PInputMessage != NULL)
        {
            // Invoke the TransformMethod on the data
            // Let exceptions flow
            _Output _Out = _M_pFunc(_PInputMessage->payload);

            // Reuse the input message ID
            _Msg = new message<_Output>(_Out, _PInputMessage->msg_id());
            _M_messageBuffer._Enqueue(_Msg);

            // Message cleanup
            delete _PInputMessage;

            if (!_M_messageBuffer._Is_head(_Msg->msg_id()))
            {
                return;
            }
        }

        _Propagate_priority_order(_M_messageBuffer);
    }

private:

    
    void _Propagate_priority_order(::Concurrency::details::_Queue<message<_Target_type>> & _MessageBuffer)
    {
        message<_Target_type> * _Msg = _MessageBuffer._Peek();

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

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

            // Always start from the first target that linked
            for (target_iterator _Iter = _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 (_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();
        }
    }

    
    void _Delete_stored_messages()
    {
        // Delete input messages
        // Because the transformer uses its own input queue, it's possible there are messages
        // in this queue and no LWT will be executed to handle them.
        message<_Input> * _PInputQueueMessage = NULL;

        while (_M_inputMessages.try_pop(_PInputQueueMessage))
        {
            // Message cleanup
            delete _PInputQueueMessage;
        }

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

    //
    //  Private Data Members
    //

    // The transformer method called by this block
    _Transform_method _M_pFunc;

    // The queue of input messages for this Transformer block
    concurrent_queue<message<_Input> *> _M_inputMessages;

    
    ::Concurrency::details::_Queue<message<_Output>> _M_messageBuffer;

private:
    //
    // Hide assignment operator and copy constructor
    //
    transformer const &operator =(transformer const &);  // no assignment operator
    transformer(transformer const &);                    // no copy constructor
};

//**************************************************************************
// Timer:
//**************************************************************************

template<class _Type>
class timer : public Concurrency::details::_Timer, public source_block<single_link_registry<ITarget<_Type>>>
{
private:

    
    enum State
    {
        
        Initialized,
        
        Started,
        
        Paused,
        
        Stopped
    };

public:

    
    timer(unsigned int _Ms, _Type const& _Value, ITarget<_Type> *_PTarget = NULL, bool _Repeating = false) :
        _Timer(_Ms, _Repeating)
    {
        _Initialize(_Value, _PTarget, _Repeating);
    }

#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP
    
    timer(Scheduler& _Scheduler, unsigned int _Ms, _Type const& _Value, _Inout_opt_ ITarget<_Type> *_PTarget = NULL, bool _Repeating = false) :
        _Timer(_Ms, _Repeating)
    {
        _Initialize(_Value, _PTarget, _Repeating, &_Scheduler);
    }

    
    timer(ScheduleGroup& _ScheduleGroup, unsigned int _Ms, _Type const& _Value, _Inout_opt_ ITarget<_Type> *_PTarget = NULL, bool _Repeating = false) :
        _Timer(_Ms, _Repeating)
    {
        _Initialize(_Value, _PTarget, _Repeating, NULL, &_ScheduleGroup);
    }
#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */

    
    ~timer()
    {
        //
        // Make sure there are no more outstanding timer fires.  Note that this does not mean that the LWT that was queued is finished, it only
        // means that no more timers will fire after the return from _Stop.  We still *MUST* wait on any outstanding LWTs.
        //
        if (_M_state == Started)
            _Stop();

        // Remove all the targets. This will wait for any outstanding LWTs
        remove_targets();

        //
        // No more asynchronous operations can happen as of this point.
        //

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

        if (_M_fReferencedScheduler)
        {
            ::Concurrency::details::_Scheduler(_M_pScheduler)._Release();
        }
    }

    
    void start()
    {
        if (_M_state == Initialized || _M_state == Paused)
        {
            _M_state = Started;
            _Start();
        }
    }

    
    void stop()
    {
        if (_M_state == Started)
            _Stop();

        _M_state = Stopped;
    }

    
    void pause()
    {
        //
        // Non repeating timers cannot pause.  They go to a final stopped state on pause.
        //
        if (!_M_fRepeating)
        {
            stop();
        }
        else
        {
            // Pause only a started timer.

            if (_M_state == Started)
            {
                _Stop();
                _M_state = Paused;
            }
        }
    }

protected:

    
    virtual message<_Type> * accept_message(runtime_object_identity _MsgId)
    {
        if (_M_pMessage == NULL || _MsgId != _M_pMessage->msg_id())
        {
            return NULL;
        }

        message<_Type> *_PMessage = _M_pMessage;
        _M_pMessage = NULL;

        return _PMessage;
    }

    
    virtual bool reserve_message(runtime_object_identity _MsgId)
    {
        //
        // Semantically, every timer tick is the same value -- it doesn't matter the message ID.  Because we can only
        // have one target as well, we do not need to track anything here.
        //
        if (_M_pMessage == NULL || _M_pMessage->msg_id() != _MsgId)
        {
            return false;
        }

        return true;
    }

    
    virtual message<_Type> * consume_message(runtime_object_identity _MsgId)
    {
        return accept_message(_MsgId);
    }

    
    virtual void release_message(runtime_object_identity _MsgId)
    {
        if (_M_pMessage == NULL || _M_pMessage->msg_id() != _MsgId)
        {
            throw message_not_found();
        }

        delete _M_pMessage;
        _M_pMessage = NULL;
    }

    
    virtual void resume_propagation()
    {
        // Because reservation doesn't prevent propagation there is
        // no need to resume on consume/release.
    }

    
    virtual void link_target_notification(_Inout_ ITarget<_Type> * _PTarget)
    {
        // If there is a timer message sitting around, it must be propagated to the target now.

        if (_M_pMessage != NULL)
        {
            _PTarget->propagate(_M_pMessage, this);
        }
    }

    
    virtual void propagate_to_any_targets(_Inout_opt_ message<_Type> *)
    {
        if (_M_pMessage == NULL)
        {
            _M_pMessage = _NewMessage();
            for (target_iterator _Iter = _M_connectedTargets.begin(); *_Iter != NULL; ++_Iter)
            {
                ITarget<_Type> * _PTarget = *_Iter;
                _PTarget->propagate(_M_pMessage, this);
            }
        }
    }

private:

    // The timer message we contain
    message<_Type> *_M_pMessage;

    // Current state of the timer.
    State _M_state;

    // The value to send on elapse of the timer.
    _Type _M_value;

    // An indication of whether the timer is repeating.
    bool _M_fRepeating;

    // A flag for whether we need to release a reference on the scheduler.
    bool _M_fReferencedScheduler;

    // Scheduler used for the timer
    Scheduler * _M_pScheduler;

    
    message<_Type>* _NewMessage() const
    {
        return new message<_Type>(_M_value);
    }

    
    virtual void _Fire()
    {
        async_send(NULL);
    }

    
    void _Initialize(const _Type& _Value, _Inout_ ITarget<_Type> *_PTarget, bool _Repeating, _Inout_opt_ Scheduler * _PScheduler = NULL, _Inout_opt_ ScheduleGroup * _PScheduleGroup = NULL)
    {
        _M_pMessage = NULL;
        _M_value = _Value;
        _M_fRepeating = _Repeating;
        _M_state = Initialized;
        _M_fReferencedScheduler = false;

        //
        // If we are going to utilize the current scheduler for timer firing, we need to capture it now.  Otherwise,
        // the timer threads fired from Windows (what _Fire executes within) will wind up with a default scheduler
        // attached -- probably not the semantic we want.
        //
        if (_PScheduleGroup == NULL && _PScheduler == NULL)
        {
            ::Concurrency::details::_Scheduler _sched = ::Concurrency::details::_CurrentScheduler::_Get();
            _PScheduler = _sched._GetScheduler();
            _sched._Reference();
            _M_fReferencedScheduler = true;
        }

        _M_pScheduler = _PScheduler;
        initialize_source(_PScheduler, _PScheduleGroup);

        if (_PTarget != NULL)
        {
            link_target(_PTarget);
        }
    }

    
    void _Delete_stored_messages()
    {
        // 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 the message remaining in the output queue
        if (_M_pMessage != NULL)
        {
            delete _M_pMessage;
        }
    }

private:
    //
    // Hide assignment operator and copy constructor
    //
    timer const &operator =(timer const &);  // no assignment operator
    timer(timer const &);                    // no copy constructor
};

//**************************************************************************
// Single assignment:
//**************************************************************************


template<class _Type>
class single_assignment : public propagator_block<multi_link_registry<ITarget<_Type>>, multi_link_registry<ISource<_Type>>>
{
public:

    
    single_assignment() :
        _M_fIsInitialized(false), _M_pMessage(NULL)
    {
        initialize_source_and_target();
    }

    
    single_assignment(filter_method const& _Filter) :
        _M_fIsInitialized(false), _M_pMessage(NULL)
    {
        initialize_source_and_target();
        register_filter(_Filter);
    }

#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP
    
    single_assignment(Scheduler& _PScheduler) :
        _M_fIsInitialized(false), _M_pMessage(NULL)
    {
        initialize_source_and_target(&_PScheduler);
    }

    
    single_assignment(Scheduler& _PScheduler, filter_method const& _Filter) :
        _M_fIsInitialized(false), _M_pMessage(NULL)
    {
        initialize_source_and_target(&_PScheduler);
        register_filter(_Filter);
    }

    
    single_assignment(ScheduleGroup& _PScheduleGroup) :
        _M_fIsInitialized(false), _M_pMessage(NULL)
    {
        initialize_source_and_target(NULL, &_PScheduleGroup);
    }

    
    single_assignment(ScheduleGroup& _PScheduleGroup, filter_method const& _Filter) :
        _M_fIsInitialized(false), _M_pMessage(NULL)
    {
        initialize_source_and_target(NULL, &_PScheduleGroup);
        register_filter(_Filter);
    }
#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */

    
    ~single_assignment()
    {
        // Remove all links
        remove_network_links();

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

    
    bool has_value() const
    {
        return (_M_pMessage != NULL);
    }


    
    _Type const & value()
    {
        if (_M_pMessage == NULL)
        {
            receive<_Type>(this);
        }
        _CONCRT_ASSERT(_M_pMessage != NULL);

        return _M_pMessage->payload;
    }


protected:

    
    virtual message_status propagate_message(_Inout_ message<_Type> * _PMessage, _Inout_ ISource<_Type> * _PSource)
    {
        // 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 _Result = accepted;

        // single_assignment messaging block can be initialized only once
        if (_M_fIsInitialized)
        {
            return declined;
        }

        {
            _NR_lock _Lock(_M_propagationLock);

            if (_M_fIsInitialized)
            {
                _Result = declined;
            }
            else
            {
                _PMessage = _PSource->accept(_PMessage->msg_id(), this);

                // Set initialized flag only if we have a message
                if (_PMessage != NULL)
                {
                    _M_fIsInitialized = true;
                }
                else
                {
                    _Result = missed;
                }
            }
        }

        //
        // If message was accepted, set the member variables for
        // this block and start the asynchronous propagation task
        //
        if (_Result == accepted)
        {
            async_send(_PMessage);
        }

        return _Result;
    }

    
    virtual message_status send_message(_Inout_ message<_Type> * _PMessage, _Inout_ ISource<_Type> * _PSource)
    {
        message_status _Result = accepted;

        // single_assignment messaging block can be initialized only once
        if (_M_fIsInitialized)
        {
            return declined;
        }

        {
            _NR_lock _Lock(_M_propagationLock);

            if (_M_fIsInitialized)
            {
                _Result = declined;
            }
            else
            {
                _PMessage = _PSource->accept(_PMessage->msg_id(), this);

                // Set initialized flag only if we have a message
                if (_PMessage != NULL)
                {
                    _M_fIsInitialized = true;
                }
                else
                {
                    _Result = missed;
                }
            }
        }

        //
        // If message was accepted, set the member variables for
        // this block and start the asynchronous propagation task
        //
        if (_Result == accepted)
        {
            sync_send(_PMessage);
        }

        return _Result;
    }

    
    virtual message<_Type> * accept_message(runtime_object_identity _MsgId)
    {
        // This check is to prevent spoofing and verify that the propagated message is
        // the one that is accepted at the end.
        if (_M_pMessage == NULL || _MsgId != _M_pMessage->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 single_assignment
        // does not need to grab the internal lock.
        //
        return (new message<_Type>(_M_pMessage->payload));
    }

    
    virtual bool reserve_message(runtime_object_identity _MsgId)
    {
        if (_M_pMessage == NULL)
        {
            return false;
        }

        if (_M_pMessage->msg_id() != _MsgId)
        {
            throw message_not_found();
        }

        return true;
    }

    
    virtual message<_Type> * consume_message(runtime_object_identity _MsgId)
    {
        _CONCRT_ASSERT(_M_fIsInitialized);

        return accept_message(_MsgId);
    }

    
    virtual void release_message(runtime_object_identity _MsgId)
    {
        _CONCRT_ASSERT(_M_fIsInitialized);

        if (_M_pMessage == NULL || _M_pMessage->msg_id() != _MsgId)
        {
            throw message_not_found();
        }
    }

    
    virtual void resume_propagation()
    {
        // Because reservation doesn't stop propagation, we don't
        // need to do anything on resume after consume/release.
    }

    
    virtual void link_target_notification(_Inout_ ITarget<_Type> * _PTarget)
    {
        // If there is a message available already, propagate it.

        if (_M_pMessage != NULL)
        {
            _PTarget->propagate(_M_pMessage, this);
        }
    }
    
    virtual void propagate_to_any_targets(_Inout_opt_ message<_Type> * _PMessage)
    {
        // Initialized flag should have been set by the propagate function using interlocked operation.
        _CONCRT_ASSERT(_M_fIsInitialized);

        // Move the message to the internal storage

        _CONCRT_ASSERT(_M_pMessage == NULL);
        _M_pMessage = _PMessage;

        for (target_iterator _Iter = _M_connectedTargets.begin(); *_Iter != NULL; ++_Iter)
        {
            // Single assignment can propagate its message out
            // to any number of Targets

            ITarget<_Type> * _PTarget = *_Iter;
            _PTarget->propagate(_PMessage, this);
        }
    }

private:

    
    void _Delete_stored_messages()
    {
        // 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

        // The messages for a single_assignment are deleted at the end when
        // single_assignment is deleted.
        delete _M_pMessage;
    }

    //
    //  Private Data Members
    //

    // The message being stored
    message<_Type> * _M_pMessage;

    // The lock used to protect propagation
    ::Concurrency::details::_NonReentrantPPLLock _M_propagationLock;

    // The marker for whether the single_assignment has already been initialized
    volatile bool _M_fIsInitialized;

private:
    //
    // Hide assignment operator and copy constructor
    //
    single_assignment const & operator=(single_assignment const &);  // no assignment operator
    single_assignment(single_assignment const &);                    // no copy constructor
};

//**************************************************************************
// Join (single-type)
//**************************************************************************


enum join_type {
    
    greedy = 0,
    
    non_greedy = 1
};


template<class _Type, join_type _Jtype = non_greedy>
class join : public propagator_block<single_link_registry<ITarget<std::vector<_Type>>>, multi_link_registry<ISource<_Type>>>
{
public:
    typedef typename std::vector<_Type> _OutputType;

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

    
    join(size_t _NumInputs, filter_method const& _Filter)
        : _M_messageArray(_NumInputs),
          _M_savedMessageIdArray(_NumInputs)
    {
        _Initialize(_NumInputs);
        register_filter(_Filter);
    }

#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP
    
    join(Scheduler& _PScheduler, size_t _NumInputs)
        : _M_messageArray(_NumInputs),
          _M_savedMessageIdArray(_NumInputs)
    {
        _Initialize(_NumInputs, &_PScheduler);
    }

    
    join(Scheduler& _PScheduler, size_t _NumInputs, filter_method const& _Filter)
        : _M_messageArray(_NumInputs),
          _M_savedMessageIdArray(_NumInputs)
    {
        _Initialize(_NumInputs, &_PScheduler);
        register_filter(_Filter);
    }

    
    join(ScheduleGroup& _PScheduleGroup, size_t _NumInputs)
        : _M_messageArray(_NumInputs),
          _M_savedMessageIdArray(_NumInputs)
    {
        _Initialize(_NumInputs, NULL, &_PScheduleGroup);
    }

    
    join(ScheduleGroup& _PScheduleGroup, size_t _NumInputs, filter_method const& _Filter)
        : _M_messageArray(_NumInputs),
          _M_savedMessageIdArray(_NumInputs)
    {
        _Initialize(_NumInputs, NULL, &_PScheduleGroup);
        register_filter(_Filter);
    }
#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */

    
    ~join()
    {
        // Remove all links that are targets of this join
        remove_network_links();

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

        delete [] _M_savedIdBuffer;
    }

protected:
    //
    // propagator_block protected function implementations
    //

    
    message_status propagate_message(_Inout_ message<_Type> * _PMessage, _Inout_ ISource<_Type> * _PSource)
    {
        // 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 = _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
                        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)
                {
                    async_send(NULL);
                }
            }

            // Always return postponed.  This message will be consumed
            // in the LWT
            _Ret_val = postponed;
        }

        return _Ret_val;
    }

    
    virtual message<_OutputType> * accept_message(runtime_object_identity _MsgId)
    {
        //
        // 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;
    }

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

    
    virtual message<_OutputType> * consume_message(runtime_object_identity _MsgId)
    {
        // By default, accept the message
        return accept_message(_MsgId);
    }

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

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

    
    virtual void link_target_notification(_Inout_ ITarget<std::vector<_Type>> *)
    {
        // If the message queue is blocked due to reservation
        // there is no need to do any message propagation
        if (_M_pReservedFor != NULL)
        {
            return;
        }

        _Propagate_priority_order(_M_messageBuffer);
    }

    
    void propagate_to_any_targets(_Inout_opt_ message<_OutputType> *)
    {
        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 = _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)
            {
                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);
    }

private:

    //
    //  Private Methods
    //

    
    void _Propagate_priority_order(::Concurrency::details::_Queue<message<_Target_type>> & _MessageBuffer)
    {
        message<_Target_type> * _Msg = _MessageBuffer._Peek();

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

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

            // Always start from the first target that linked
            for (target_iterator _Iter = _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 (_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();
        }
    }

    
    message<std::vector<_Type>> * __cdecl _Create_new_message()
    {
        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 = _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)
                            {
                                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));
    }

    
    void _Initialize(size_t _NumInputs, Scheduler * _PScheduler = NULL, ScheduleGroup * _PScheduleGroup = NULL)
    {
        initialize_source_and_target(_PScheduler, _PScheduleGroup);

        _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;
        }
    }

    
    void _Delete_stored_messages()
    {
        // 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;
        }
    }

    // The current number of messages remaining
    volatile size_t _M_messagesRemaining;

    // An array containing the accepted messages of this join.
    // Wrapped in a struct to enable debugger visualization.
    struct _MessageArray
    {
        size_t _M_count;
        message<_Type>** _M_messages;

        _MessageArray(size_t _NumInputs)
            : _M_count(_NumInputs),
              _M_messages(new message<_Type>*[_NumInputs])
        {
            memset(_M_messages,  0, sizeof(message<_Type> *) * _NumInputs);
        }

        ~_MessageArray()
        {
            for (size_t i = 0; i < _M_count; i++)
                delete _M_messages[i];
            delete [] _M_messages;
        }
    };
    _MessageArray _M_messageArray;

    // An array containing the msg IDs of messages propagated to the array
    // For greedy joins, this contains a log of other messages passed to this
    // join after the first has been accepted
    // For non-greedy joins, this contains the message ID of any message
    // passed to it.
    // Wrapped in a struct to enable debugger visualization.
    struct _SavedMessageIdArray
    {
        size_t _M_count;
        runtime_object_identity * _M_savedIds;

        _SavedMessageIdArray(size_t _NumInputs)
            : _M_count(_NumInputs),
              _M_savedIds(new runtime_object_identity[_NumInputs])
        {
            memset(_M_savedIds, -1, sizeof(runtime_object_identity) * _NumInputs);
        }

        ~_SavedMessageIdArray()
        {
            delete [] _M_savedIds;
        }
    };
    _SavedMessageIdArray _M_savedMessageIdArray;

    // Buffer for snapping saved IDs in non-greedy joins
    runtime_object_identity * _M_savedIdBuffer;

    // A lock for modifying the buffer or the connected blocks
    ::Concurrency::details::_NonReentrantPPLLock _M_propagationLock;

    // Queue to hold output messages
    ::Concurrency::details::_Queue<message<std::vector<_Type>>> _M_messageBuffer;
};


//**************************************************************************
// Multi-Type Choice and Join helper node:
//**************************************************************************


template<class _Type>
class _Order_node_base: public propagator_block<single_link_registry<ITarget<size_t>>, multi_link_registry<ISource<_Type>>>
{
public:

    
    _Order_node_base() :
        _M_index(0),
        _M_pReceiveMessage(NULL),
        _M_pSendMessage(NULL)
    {
    }

    
    ~_Order_node_base()
    {
        // The messages for an _Order_node_base are deleted at the end when
        // _Order_node_base is deleted.
        delete _M_pReceiveMessage;
        delete _M_pSendMessage;
    }

    
    bool has_value() const
    {
        return (_M_pReceiveMessage != NULL);
    }

    
    _Type const & value()
    {
        _CONCRT_ASSERT(_M_pReceiveMessage != NULL);

        return _M_pReceiveMessage->payload;
    }

    
    virtual void _Reset() = 0;

    
    virtual bool reserve_message(runtime_object_identity)
    {
        // reserve should never be called for this block.
        _CONCRT_ASSERT(false);

        return false;
    }

    
    virtual message<size_t> * consume_message(runtime_object_identity)
    {
        // consume should never be called for this block.
        _CONCRT_ASSERT(false);

        return NULL;
    }

    
    virtual void release_message(runtime_object_identity)
    {
        // release should never be called for this block.
        _CONCRT_ASSERT(false);
    }

protected:


    
    virtual void resume_propagation()
    {
        // Because there is only a single target, nothing needs
        // to be done on resume
    }

    
    virtual void link_target_notification(_Inout_ ITarget<size_t> *)
    {
        if (_M_pSendMessage != NULL)
        {
            propagate_to_any_targets(NULL);
        }
    }

    
    void _Create_send_message()
    {
        _M_pSendMessage = new message<size_t>(_M_index);
    }

    
    void _Initialize_order_node(ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget, Scheduler * _PScheduler = NULL, ScheduleGroup * _PScheduleGroup = NULL)
    {
        if (_Index < 0)
        {
            throw std::invalid_argument("_Index");
        }

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

        _M_index = _Index;

        initialize_source_and_target(_PScheduler, _PScheduleGroup);

        // Allow only a single source and ensure that they
        // cannot be unlinked and relinked.
        _M_connectedSources.set_bound(1);

        if (_PTarget != NULL)
        {
            link_target(_PTarget);
        }

        _PSource->link_target(this);
    }

    //
    //  Private Data Members
    //

    // The message to be received from the source
    message<_Type> * _M_pReceiveMessage;

    // The message to be sent to all targets
    message<size_t> * _M_pSendMessage;

    // The index of the _Order_node_base in the user's construct
    size_t _M_index;

private:
    //
    // Hide assignment operator and copy constructor
    //
    _Order_node_base const & operator=(_Order_node_base const &);  // no assignment operator
    _Order_node_base(_Order_node_base const &);                    // no copy constructor
};



template<class _Type>
class _Reserving_node: public _Order_node_base<_Type>
{
public:
    
    _Reserving_node(ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget = NULL) :
        _M_fIsInitialized(false),
        _M_savedId(-1),
        _M_pReservedSource(NULL)
    {
        _Initialize_order_node(_PSource, _Index, _PTarget);
    }

    
    _Reserving_node(ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget, filter_method const& _Filter) :
        _M_fIsInitialized(false),
        _M_savedId(-1),
        _M_pReservedSource(NULL)
    {
        register_filter(_Filter);
        _Initialize_order_node(_PSource, _Index, _PTarget);
    }

    
    _Reserving_node(Scheduler& _PScheduler, ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget = NULL) :
        _M_fIsInitialized(false),
        _M_savedId(-1),
        _M_pReservedSource(NULL)
    {
        _Initialize_order_node(_PSource, _Index, _PTarget, &_PScheduler);
    }

    
    _Reserving_node(Scheduler& _PScheduler, ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget, filter_method const& _Filter) :
        _M_fIsInitialized(false),
        _M_savedId(-1),
        _M_pReservedSource(NULL)
    {
        register_filter(_Filter);
        _Initialize_order_node(_PSource, _Index, _PTarget, &_PScheduler);
    }

    
    _Reserving_node(ScheduleGroup& _PScheduleGroup, ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget = NULL) :
        _M_fIsInitialized(false),
        _M_savedId(-1),
        _M_pReservedSource(NULL)
    {
        _Initialize_order_node(_PSource, _Index, _PTarget, NULL, &_PScheduleGroup);
    }

    
    _Reserving_node(ScheduleGroup& _PScheduleGroup, ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget, filter_method const& _Filter) :
        _M_fIsInitialized(false),
        _M_savedId(-1),
        _M_pReservedSource(NULL)
    {
        register_filter(_Filter);
        _Initialize_order_node(_PSource, _Index, _PTarget, NULL, &_PScheduleGroup);
    }

    
    ~_Reserving_node()
    {
        if (_M_pReservedSource != NULL)
        {
            _M_pReservedSource = NULL;
            _M_connectedSources.release();
        }

        // Remove all links
        remove_network_links();
    }


    
    virtual void _Reset()
    {
    }

protected:

    //
    // propagator_block protected function implementation
    //

    
    virtual message_status propagate_message(message<_Type> * _PMessage, ISource<_Type> * _PSource)
    {
        message_status _Result = postponed;

        // _Order_node messaging block can be initialized only once, just like single_assignment.
        if (_M_fIsInitialized)
        {
            return declined;
        }

        // Reserve a message on the source until this _Order_node gets the feedback from
        // the single_assignment on whether it has been selected.
        _M_fIsInitialized = _PSource->reserve(_PMessage->msg_id(), this);

        //
        // If message was successfully reserved, set the member variables for
        // this messaging block and start the asynchronous propagation task.
        //
        if (_M_fIsInitialized)
        {
            _M_savedId = _PMessage->msg_id();
            async_send(NULL);
        }
        else
        {
            _Result = missed;
        }

        return _Result;
    }

    
    virtual message<size_t> * accept_message(runtime_object_identity _MsgId)
    {
        // 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;
        }

        // If the source has disconnected then we can't allow for accept to succeed.
        source_iterator _Iter = _M_connectedSources.begin();
        ISource<_Type>* _PSource = *_Iter;

        if (_PSource == NULL)
        {
            // source was disconnected. Fail accept.
            return NULL;
        }

        _M_pReceiveMessage = _PSource->consume(_M_savedId, this);

        _CONCRT_ASSERT(_M_pReceiveMessage != 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 _Order_node
        // does not need to grab the internal lock.
        //
        return (new message<size_t>(_M_pSendMessage->payload));
    }

    
    virtual void propagate_to_any_targets(_Inout_opt_ message<size_t> *)
    {
        if (_M_pSendMessage == NULL)
        {
            _Create_send_message();
        }

        for (target_iterator _Iter = _M_connectedTargets.begin(); *_Iter != NULL; ++_Iter)
        {
            ITarget<size_t> * _PTarget = *_Iter;
            _Propagate_to_target(_PTarget);
        }
    }

private:

    
    message_status _Propagate_to_target(ITarget<size_t> * _PTarget)
    {
        message_status _Status = _PTarget->propagate(_M_pSendMessage, this);

        // If the message got rejected we have to release the hold on the source message.
        if (_Status != accepted)
        {
            if (_M_savedId != -1)
            {
                // Release the reservation
                source_iterator _Iter = _M_connectedSources.begin();
                ISource<_Type> * _PSource = *_Iter;

                if (_PSource != NULL)
                {
                    _PSource->release(_M_savedId, this);
                }

                // If the source was disconnected, then it would
                // automatically release any reservation. So we
                // should reset our savedId regardless.
                _M_savedId = -1;
            }

        }

        return _Status;
    }

    //
    //  Private Data Members
    //

    // The source where we have reserved a message
    ISource<_Type> * _M_pReservedSource;

    // For greedy order-nodes, the message ID of subsequent messages sent to this node
    // For non-greedy order nodes, the message ID of the message to reserve/consume
    runtime_object_identity _M_savedId;

    // The marker that indicates that _Reserving_node has reserved a message
    volatile bool _M_fIsInitialized;

private:
    //
    // Hide assignment operator and copy constructor
    //
    _Reserving_node const & operator=(_Reserving_node const &);  // no assignment operator
    _Reserving_node(_Reserving_node const &);                    // no copy constructor
};



template<class _Type>
class _Greedy_node: public _Order_node_base<_Type>
{
public:
    
    _Greedy_node(ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget = NULL) :
        _M_savedId(-1),
        _M_pGreedyMessage(NULL)
    {
        _Initialize_order_node(_PSource, _Index, _PTarget);
    }

    
    _Greedy_node(ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget, filter_method const& _Filter) :
        _M_savedId(-1),
        _M_pGreedyMessage(NULL)
    {
        register_filter(_Filter);
        _Initialize_order_node(_PSource, _Index, _PTarget);
    }

    
    _Greedy_node(Scheduler& _PScheduler, ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget = NULL) :
        _M_savedId(-1),
        _M_pGreedyMessage(NULL)
    {
        _Initialize_order_node(_PSource, _Index, _PTarget, &_PScheduler);
    }

    
    _Greedy_node(Scheduler& _PScheduler, ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget, filter_method const& _Filter) :
        _M_savedId(-1),
        _M_pGreedyMessage(NULL)
    {
        register_filter(_Filter);
        _Initialize_order_node(_PSource, _Index, _PTarget, &_PScheduler);
    }

    
    _Greedy_node(ScheduleGroup& _PScheduleGroup, ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget = NULL) :
        _M_savedId(-1),
        _M_pGreedyMessage(NULL)
    {
        _Initialize_order_node(_PSource, _Index, _PTarget, NULL, &_PScheduleGroup);
    }

    
    _Greedy_node(ScheduleGroup& _PScheduleGroup, ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget, filter_method const& _Filter) :
        _M_savedId(-1),
        _M_pGreedyMessage(NULL)
    {
        register_filter(_Filter);
        _Initialize_order_node(_PSource, _Index, _PTarget, NULL, &_PScheduleGroup);
    }

    
    ~_Greedy_node()
    {
        // Remove all links
        remove_network_links();

        if (_M_pGreedyMessage != _M_pReceiveMessage)
        {
            delete _M_pGreedyMessage;
        }
    }

    
    void _Reset()
    {
        _R_lock _Lock(_M_resetLock);

        delete _M_pReceiveMessage;
        _M_pReceiveMessage = NULL;

        delete _M_pSendMessage;
        _M_pSendMessage = NULL;

        //
        // For greedy type joins, look to see if any other messages have been
        // passed to this _Greedy_node while the join was waiting for other
        // messages to arrive.  This function is already called with _M_resetLock
        // held through propagate_to_any_targets().
        //
        for(;;)
        {
            // Set the current saved ID as -1.  Check to see if something was ready for consumption
            // (if _Saved_id != -1) and consume it if possible.
            runtime_object_identity _Saved_id;

            {
                _NR_lock lockHolder(_M_propagationLock);

                _Saved_id = _M_savedId;

                if (_Saved_id == -1)
                {
                    _M_pGreedyMessage = NULL;
                    break;
                }
                else
                {
                    _M_savedId = -1;
                }
            }

            if (_Saved_id != -1)
            {
                source_iterator _Iter = _M_connectedSources.begin();

                ISource<_Type> * _PSource = *_Iter;
                if ((_PSource != NULL) && _PSource->reserve(_Saved_id, this))
                {
                    _M_pGreedyMessage = _PSource->consume(_Saved_id, this);
                    async_send(NULL);
                    break;
                }
            }
        }
    }

protected:

    //
    // propagator_block protected function implementation
    //

    
    virtual message_status propagate_message(message<_Type> * _PMessage, ISource<_Type> * _PSource)
    {
        message_status _Result = postponed;

        bool _FDone = false;

        {
            _NR_lock lockHolder(_M_propagationLock);
            if (_M_pGreedyMessage != NULL)
            {
                _M_savedId = _PMessage->msg_id();
                _Result = postponed;
                _FDone = true;
            }
        }

        if (!_FDone)
        {
            _M_pGreedyMessage = _PSource->accept(_PMessage->msg_id(), this);

            if (_M_pGreedyMessage != NULL)
            {
                _Result = accepted;
                async_send(NULL);
            }
            else
            {
                _Result = missed;
            }
        }

        return _Result;
    }

    
    virtual message<size_t> * accept_message(runtime_object_identity _MsgId)
    {
        // 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 _Greedy_node
        // does not need to grab the internal lock.
        //
        return (new message<size_t>(_M_pSendMessage->payload));
    }


    
    virtual void propagate_to_any_targets(_Inout_opt_ message<size_t> *)
    {
        _R_lock _Lock(_M_resetLock);

        if (_M_pSendMessage == NULL)
        {
            // Save the incoming message so that it can be consumed in the accept function
            _M_pReceiveMessage = _M_pGreedyMessage;
            _Create_send_message();
        }

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

private:

    //
    //  Private Data Members
    //

    // The message to be saved by a greedy order node
    message<_Type> * _M_pGreedyMessage;

    // The lock used to protect propagation
    ::Concurrency::details::_NonReentrantPPLLock _M_propagationLock;

    // The lock used to protect modification during a reset
    ::Concurrency::details::_ReentrantPPLLock _M_resetLock;

    // For greedy order-nodes, the message ID of subsequent messages sent to this node
    // For non-greedy order nodes, the message ID of the message to reserve/consume
    runtime_object_identity _M_savedId;

private:
    //
    // Hide assignment operator and copy constructor
    //
    _Greedy_node const & operator=(_Greedy_node const &);  // no assignment operator
    _Greedy_node(_Greedy_node const &);                    // no copy constructor
};



template<class _Type>
class _Non_greedy_node: public _Order_node_base<_Type>
{
public:
    
    _Non_greedy_node(ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget = NULL) :
        _M_savedId(-1),
        _M_reservedId(-1),
        _M_pReservedSource(NULL)
    {
        _Initialize_order_node(_PSource, _Index, _PTarget);
    }

    
    _Non_greedy_node(ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget, filter_method const& _Filter) :
        _M_savedId(-1),
        _M_reservedId(-1),
        _M_pReservedSource(NULL)
    {
        register_filter(_Filter);
        _Initialize_order_node(_PSource, _Index, _PTarget);
    }

    
    _Non_greedy_node(Scheduler& _PScheduler, ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget = NULL) :
        _M_savedId(-1),
        _M_reservedId(-1),
        _M_pReservedSource(NULL)
    {
        _Initialize_order_node(_PSource, _Index, _PTarget, &_PScheduler);
    }

    
    _Non_greedy_node(Scheduler& _PScheduler, ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget, filter_method const& _Filter) :
        _M_savedId(-1),
        _M_reservedId(-1),
        _M_pReservedSource(NULL)
    {
        register_filter(_Filter);
        _Initialize_order_node(_PSource, _Index, _PTarget, &_PScheduler);
    }

    
    _Non_greedy_node(ScheduleGroup& _PScheduleGroup, ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget = NULL) :
        _M_savedId(-1),
        _M_reservedId(-1),
        _M_pReservedSource(NULL)
    {
        _Initialize_order_node(_PSource, _Index, _PTarget, NULL, &_PScheduleGroup);
    }

    
    _Non_greedy_node(ScheduleGroup& _PScheduleGroup, ISource<_Type> * _PSource, size_t _Index, ITarget<size_t> * _PTarget, filter_method const& _Filter) :
        _M_savedId(-1),
        _M_reservedId(-1),
        _M_pReservedSource(NULL)
    {
        register_filter(_Filter);
        _Initialize_order_node(_PSource, _Index, _PTarget, NULL, &_PScheduleGroup);
    }

    
    ~_Non_greedy_node()
    {
        if (_M_pReservedSource != NULL)
        {
            _M_pReservedSource = NULL;
            _M_connectedSources.release();
        }

        // Remove all links
        remove_network_links();
    }

    
    void _Reset()
    {
        _R_lock _Lock(_M_resetLock);

        delete _M_pReceiveMessage;
        _M_pReceiveMessage = NULL;

        delete _M_pSendMessage;
        _M_pSendMessage = NULL;
    }

    
    bool _Reserve_received_message()
    {
        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;
    }

    
    void _Consume_received_message()
    {
        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();
        }
    }

    
    bool _Release_received_message()
    {
        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;
    }

protected:

    //
    // propagator_block protected function implementation
    //

    
    virtual message_status propagate_message(message<_Type> * _PMessage, ISource<_Type> *)
    {
        // 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;
    }

    
    virtual message<size_t> * accept_message(runtime_object_identity _MsgId)
    {
        // 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));
    }

    
    virtual void propagate_to_any_targets(_Inout_opt_ message<size_t> *)
    {
        _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);
        }
    }

private:

    //
    //  Private Data Members
    //

    // The source where we have reserved a message
    ISource<_Type> * _M_pReservedSource;

    // The lock used to protect modification during a reset
    ::Concurrency::details::_ReentrantPPLLock _M_resetLock;

    // For non-greedy order nodes, the message ID of the message to reserve/consume
    runtime_object_identity _M_savedId;

    // For non-greedy order nodes, the reserved ID of the message that was reserved
    runtime_object_identity _M_reservedId;

    // The marker that indicates that _Non_greedy_node has reserved a message
    volatile bool _M_fIsInitialized;

private:
    //
    // Hide assignment operator and copy constructor
    //
    _Non_greedy_node const & operator=(_Non_greedy_node const &);  // no assignment operator
    _Non_greedy_node(_Non_greedy_node const &);                    // no copy constructor
};

//**************************************************************************
// Choice:
//**************************************************************************


template<class _Type>
class choice: public ISource<size_t>
{
public:

    
    explicit choice(_Type _Tuple) : _M_sourceTuple(_Tuple), _M_pScheduler(NULL), _M_pScheduleGroup(NULL)
    {
        _M_pSingleAssignment = new single_assignment<size_t>();
        _Initialize_choices<0>();
    }

#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP
    
    choice(Scheduler& _PScheduler, _Type _Tuple) : _M_sourceTuple(_Tuple), _M_pScheduler(&_PScheduler), _M_pScheduleGroup(NULL)
    {
        _M_pSingleAssignment = new single_assignment<size_t>(_PScheduler);
        _Initialize_choices<0>();
    }

    
    choice(ScheduleGroup& _PScheduleGroup, _Type _Tuple) : _M_sourceTuple(_Tuple), _M_pScheduler(NULL), _M_pScheduleGroup(&_PScheduleGroup)
    {
        _M_pSingleAssignment = new single_assignment<size_t>(_PScheduleGroup);
        _Initialize_choices<0>();
    }
#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */

    
    choice(choice && _Choice)
    {
        // Copy scheduler group or scheduler to the new object.
        _M_pScheduleGroup = _Choice._M_pScheduleGroup;
        _M_pScheduler = _Choice._M_pScheduler;

        // Single assignment is heap allocated, so simply copy the pointer. If it already has
        // a value, it will be preserved.
        _M_pSingleAssignment = _Choice._M_pSingleAssignment;
        _Choice._M_pSingleAssignment = NULL;

        // Invoke copy assignment for tuple to copy pointers to message blocks.
        _M_sourceTuple = _Choice._M_sourceTuple;

        // Copy the pointers to order nodes to a new object and zero out in the old object.
        memcpy(_M_pSourceChoices, _Choice._M_pSourceChoices, sizeof(_M_pSourceChoices));
        memset(_Choice._M_pSourceChoices, 0, sizeof(_M_pSourceChoices));
    }

    
    ~choice()
    {
        delete _M_pSingleAssignment;
        _Delete_choices<0>();
    }

    
    typedef typename _Type type;

    
    bool has_value() const
    {
        return _M_pSingleAssignment->has_value();
    }

    
    size_t index()
    {
        return _M_pSingleAssignment->value();
    }

    
    template <typename _Payload_type>
    _Payload_type const & value()
    {
        return reinterpret_cast<_Reserving_node<_Payload_type> *>(_M_pSourceChoices[_M_pSingleAssignment->value()])->value();
    }

    //
    // ISource public function implementations
    //

    
    virtual void link_target(_Inout_ ITarget<size_t> * _PTarget)
    {
        _M_pSingleAssignment->link_target(_PTarget);
    }

    
    virtual void unlink_target(_Inout_ ITarget<size_t> * _PTarget)
    {
        _M_pSingleAssignment->unlink_target(_PTarget);
    }

    
    virtual void unlink_targets()
    {
        _M_pSingleAssignment->unlink_targets();
    }

    
    virtual message<size_t> * accept(runtime_object_identity _MsgId, _Inout_ ITarget<size_t> * _PTarget)
    {
        return _M_pSingleAssignment->accept(_MsgId, _PTarget);
    }

    
    virtual bool reserve(runtime_object_identity _MsgId, _Inout_ ITarget<size_t> * _PTarget)
    {
        return _M_pSingleAssignment->reserve(_MsgId, _PTarget);
    }

    
    virtual message<size_t> * consume(runtime_object_identity _MsgId, _Inout_ ITarget<size_t> * _PTarget)
    {
        return _M_pSingleAssignment->consume(_MsgId, _PTarget);
    }

    
    virtual void release(runtime_object_identity _MsgId, _Inout_ ITarget<size_t> * _PTarget)
    {
        _M_pSingleAssignment->release(_MsgId, _PTarget);
    }

    
    virtual void acquire_ref(_Inout_ ITarget<size_t> * _PTarget)
    {
        _M_pSingleAssignment->acquire_ref(_PTarget);
    }

    
    virtual void release_ref(_Inout_ ITarget<size_t> * _PTarget)
    {
        _M_pSingleAssignment->release_ref(_PTarget);
    }

private:

    
    template<int _Index>
    void _Initialize_choices()
    {
        std::tr1::tuple_element<_Index, _Type>::type _Item = std::tr1::get<_Index>(_M_sourceTuple);
        _Reserving_node<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> * _Order_node_element = NULL;

        if (_M_pScheduleGroup != NULL)
        {
            _Order_node_element = new _Reserving_node<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> (*_M_pScheduleGroup, _Item, _Index);
        }
        else if (_M_pScheduler != NULL)
        {
            _Order_node_element = new _Reserving_node<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> (*_M_pScheduler, _Item, _Index);
        }
        else
        {
            _Order_node_element = new _Reserving_node<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> (_Item, _Index);
        }

        _M_pSourceChoices[_Index] = _Order_node_element;
        _Order_node_element->link_target(_M_pSingleAssignment);
        _Initialize_choices<_Index + 1>();
    }

    
    template<> void _Initialize_choices<std::tr1::tuple_size<_Type>::value>()
    {
    }

    
    template<int _Index>
    void _Delete_choices()
    {
        delete reinterpret_cast<_Reserving_node<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> *>(_M_pSourceChoices[_Index]);
        _M_pSourceChoices[_Index] = NULL;
        _Delete_choices<_Index + 1>();
    }

    
    template<> void _Delete_choices<std::tr1::tuple_size<_Type>::value>()
    {
    }

    // Array of pointers to _Reserving_node elements representing each source
    void * _M_pSourceChoices[std::tr1::tuple_size<_Type>::value];

    // Single assignment which chooses between source messaging blocks
    single_assignment<size_t> * _M_pSingleAssignment;

    // Tuple of messaging blocks that are sources to this choice
    _Type _M_sourceTuple;

    // The scheduler to propagate messages on
    Scheduler * _M_pScheduler;

    // The schedule group to propagate messages on
    ScheduleGroup * _M_pScheduleGroup;

private:
    //
    // Hide assignment operator
    //
    choice const &operator =(choice const &);                      // no assignment operator
    choice(choice const &);                                        // no copy constructor
};

// Templated factory functions that create a choice, three flavors

#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP

template<typename _Type1, typename _Type2, typename... _Types>
choice<std::tuple<_Type1, _Type2, _Types...>>
make_choice(Scheduler& _PScheduler, _Type1 _Item1, _Type2 _Item2, _Types... _Items)
{
    return choice<std::tuple<_Type1, _Type2, _Types...>>(_PScheduler, std::make_tuple(_Item1, _Item2, _Items...));
}


template<typename _Type1, typename _Type2, typename... _Types>
choice<std::tuple<_Type1, _Type2, _Types...>>
make_choice(ScheduleGroup& _PScheduleGroup, _Type1 _Item1, _Type2 _Item2, _Types... _Items)
{
    return choice<std::tuple<_Type1, _Type2, _Types...>>(_PScheduleGroup, std::make_tuple(_Item1, _Item2, _Items...));
}

#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */


template<typename _Type1, typename _Type2, typename... _Types>
choice<std::tuple<_Type1, _Type2, _Types...>>
make_choice(_Type1 _Item1, _Type2 _Item2, _Types... _Items)
{
    return choice<std::tuple<_Type1, _Type2, _Types...>>(std::make_tuple(_Item1, _Item2, _Items...));
}

//**************************************************************************
// Join:
//**************************************************************************

// Template specialization used to unwrap the types from within a tuple.


template <typename _Tuple> struct _Unwrap;


template <typename... _Types>
struct _Unwrap<std::tuple<_Types...>>
{
    typedef std::tuple<typename std::remove_pointer<_Types>::type::source_type...> type;
};


template<typename _Type, typename _Destination_type, join_type _Jtype>
class _Join_node: public propagator_block<single_link_registry<ITarget<_Destination_type>>, multi_link_registry<ISource<size_t>>>
{
public:

    
    _Join_node() : _M_counter(std::tr1::tuple_size<_Destination_type>::value)
    {
        initialize_source_and_target();
    }

    
    _Join_node(Scheduler& _PScheduler) : _M_counter(std::tr1::tuple_size<_Destination_type>::value)
    {
        initialize_source_and_target(&_PScheduler);
    }

    
    _Join_node(ScheduleGroup& _PScheduleGroup) : _M_counter(std::tr1::tuple_size<_Destination_type>::value)
    {
        initialize_source_and_target(NULL, &_PScheduleGroup);
    }

    
    ~_Join_node()
    {
        // Remove all links
        remove_network_links();

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

protected:

    
    virtual message_status propagate_message(message<size_t> * _PMessage, ISource<size_t> * _PSource)
    {
        // This join block is connected to the _Order_node sources, which know not to send
        // any more messages until join propagates them further. That is why join can
        // always accept the incoming messages.

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

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

        long _Ret_val = _InterlockedDecrement(&_M_counter);

        _CONCRT_ASSERT(_Ret_val >= 0);

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

        return accepted;
    }

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

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

        return _Msg;
    }

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

    
    virtual message<_Destination_type> * consume_message(runtime_object_identity _MsgId)
    {
        // By default, accept the message
        return accept_message(_MsgId);
    }

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

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

    
    virtual void link_target_notification(_Inout_ ITarget<_Destination_type> *)
    {
        // There is only a single target.
        _Propagate_priority_order(_M_messageBuffer);
    }

    
    virtual void propagate_to_any_targets(_Inout_opt_ message<_Destination_type> *)
    {
        message<_Destination_type> * _Msg = NULL;

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

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

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

            _Msg = _Create_send_message();
        }

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

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

        _Propagate_priority_order(_M_messageBuffer);
    }

private:

    
    template<int _Index>
    bool _Try_consume_source_messages(_Destination_type & _Destination_tuple, ISource<size_t> ** _Sources)
    {
        _Non_greedy_node<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> * _Node =
            static_cast<_Non_greedy_node<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> *>(_Sources[_Index]);

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

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

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

            return _Ret_val;
        }

        return false;
    }

    
    template<> bool _Try_consume_source_messages<std::tr1::tuple_size<_Type>::value>(_Destination_type &, ISource<size_t> **)
    {
        return true;
    }

    
    bool _Non_greedy_acquire_messages()
    {
        _Destination_type _Destination_tuple;

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

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

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

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

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

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

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

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

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

        return _IsAcquireSuccessful;
    }

    
    void _Propagate_priority_order(::Concurrency::details::_Queue<message<_Target_type>> & _MessageBuffer)
    {
        message<_Target_type> * _Msg = _MessageBuffer._Peek();

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

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

            // Always start from the first target that linked
            for (target_iterator _Iter = _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 (_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();
        }
    }

    
    message<_Destination_type> * _Create_send_message()
    {
        _Destination_type _Destination_tuple;

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

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

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

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

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

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

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

        _Populate_destination_tuple<0>(_Destination_tuple, _Sources);

        return new message<_Destination_type>(_Destination_tuple);
    }

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

    
    template<int _Index>
    void _Populate_destination_tuple(_Destination_type & _Destination_tuple, ISource<size_t> ** _Sources)
    {
        _Order_node_base<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> * _Node =
            static_cast<_Order_node_base<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> *>(_Sources[_Index]);

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

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

    
    template<> void _Populate_destination_tuple<std::tr1::tuple_size<_Type>::value>(_Destination_type &, ISource<size_t> **)
    {
    }

    // A tuple containing a collection of source messaging blocks
    _Type _M_sourceTuple;

    // Counts messages received by sources of this node and is used to trigger propagation to targets
    // This value starts at the total number of inputs and counts down to zero.  When it reaches zero,
    // a join of the inputs is started.
    volatile long _M_counter;

    // Buffer to hold outgoing messages
    ::Concurrency::details::_Queue<message<_Destination_type>> _M_messageBuffer;

private:
    //
    // Hide assignment operator and copy constructor
    //
    _Join_node(const _Join_node & _Join);                                  // no copy constructor
    _Join_node const &operator =(_Join_node const &);                      // no assignment operator
};


template<typename _Type, join_type _Jtype = non_greedy>
class multitype_join: public ISource<typename _Unwrap<_Type>::type>
{
public:

    typedef typename _Unwrap<_Type>::type _Destination_type;

    
    explicit multitype_join(_Type _Tuple) : _M_sourceTuple(_Tuple), _M_pScheduler(NULL), _M_pScheduleGroup(NULL)
    {
        _M_pJoinNode = new _Join_node<_Type, _Destination_type, _Jtype>();
        _Initialize_joins<0>();
    }

#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP
    
    multitype_join(Scheduler& _PScheduler, _Type _Tuple) : _M_sourceTuple(_Tuple), _M_pScheduler(&_PScheduler), _M_pScheduleGroup(NULL)
    {
        _M_pJoinNode = new _Join_node<_Type, _Destination_type, _Jtype>(_PScheduler);
        _Initialize_joins<0>();
    }

    
    multitype_join(ScheduleGroup& _PScheduleGroup, _Type _Tuple) : _M_sourceTuple(_Tuple), _M_pScheduler(NULL), _M_pScheduleGroup(&_PScheduleGroup)
    {
        _M_pJoinNode = new _Join_node<_Type, _Destination_type, _Jtype>(_PScheduleGroup);
        _Initialize_joins<0>();
    }
#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */

    
    multitype_join(multitype_join && _Join)
    {
        // Copy scheduler group or scheduler to the new object.
        _M_pScheduleGroup = _Join._M_pScheduleGroup;
        _M_pScheduler = _Join._M_pScheduler;

        // Single assignment is heap allocated, so simply copy the pointer. If it already has
        // a value, it will be preserved.
        _M_pJoinNode = _Join._M_pJoinNode;
        _Join._M_pJoinNode = NULL;

        // Invoke copy assignment for tuple to copy pointers to message blocks.
        _M_sourceTuple = _Join._M_sourceTuple;

        // Copy the pointers to order nodes to a new object and zero out in the old object.
        memcpy(_M_pSourceJoins, _Join._M_pSourceJoins, sizeof(_M_pSourceJoins));
        memset(_Join._M_pSourceJoins, 0, sizeof(_M_pSourceJoins));
    }

    
    ~multitype_join()
    {
        delete _M_pJoinNode;
        _Delete_joins<0>();
    }

    
    typedef typename _Type type;

    //
    // ISource public function implementations
    //

    
    virtual void link_target(_Inout_ ITarget<_Destination_type> * _PTarget)
    {
        _M_pJoinNode->link_target(_PTarget);
    }

    
    virtual void unlink_target(_Inout_ ITarget<_Destination_type> * _PTarget)
    {
        _M_pJoinNode->unlink_target(_PTarget);
    }

    
    virtual void unlink_targets()
    {
        _M_pJoinNode->unlink_targets();
    }

    
    virtual message<_Destination_type> * accept(runtime_object_identity _MsgId, _Inout_ ITarget<_Destination_type> * _PTarget)
    {
        return _M_pJoinNode->accept(_MsgId, _PTarget);
    }

    
    virtual bool reserve(runtime_object_identity _MsgId, _Inout_ ITarget<_Destination_type> * _PTarget)
    {
        return _M_pJoinNode->reserve(_MsgId, _PTarget);
    }

    
    virtual message<_Destination_type> * consume(runtime_object_identity _MsgId, _Inout_ ITarget<_Destination_type> * _PTarget)
    {
        return _M_pJoinNode->consume(_MsgId, _PTarget);
    }

    
    virtual void release(runtime_object_identity _MsgId, _Inout_ ITarget<_Destination_type> * _PTarget)
    {
        _M_pJoinNode->release(_MsgId, _PTarget);
    }

    
    virtual void acquire_ref(_Inout_ ITarget<_Destination_type> * _PTarget)
    {
        _M_pJoinNode->acquire_ref(_PTarget);
    }

    
    virtual void release_ref(_Inout_ ITarget<_Destination_type> * _PTarget)
    {
        _M_pJoinNode->release_ref(_PTarget);
    }

private:

    
    template<int _Index>
    void _Initialize_joins()
    {
        std::tr1::tuple_element<_Index, _Type>::type _Item = std::tr1::get<_Index>(_M_sourceTuple);
        _Order_node_base<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> * _Order_node_element = NULL;

        bool fIsNonGreedy = (_Jtype == non_greedy);

        if (fIsNonGreedy)
        {
            if (_M_pScheduleGroup != NULL)
            {
                _Order_node_element = new _Non_greedy_node<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> (*_M_pScheduleGroup, _Item, _Index);
            }
            else if (_M_pScheduler != NULL)
            {
                _Order_node_element = new _Non_greedy_node<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> (*_M_pScheduler, _Item, _Index);
            }
            else
            {
                _Order_node_element = new _Non_greedy_node<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> (_Item, _Index);
            }
        }
        else
        {
            if (_M_pScheduleGroup != NULL)
            {
                _Order_node_element = new _Greedy_node<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> (*_M_pScheduleGroup, _Item, _Index);
            }
            else if (_M_pScheduler != NULL)
            {
                _Order_node_element = new _Greedy_node<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> (*_M_pScheduler, _Item, _Index);
            }
            else
            {
                _Order_node_element = new _Greedy_node<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> (_Item, _Index);
            }
        }
        _M_pSourceJoins[_Index] = _Order_node_element;
        _Order_node_element->link_target(_M_pJoinNode);
        _Initialize_joins<_Index + 1>();
    }

    
    template<> void _Initialize_joins<std::tr1::tuple_size<_Type>::value>()
    {
    }

    
    template<int _Index>
    void _Delete_joins()
    {
        delete reinterpret_cast<_Order_node_base<std::tr1::remove_pointer<std::tr1::tuple_element<_Index, _Type>::type>::type::source_type> *>(_M_pSourceJoins[_Index]);
        _M_pSourceJoins[_Index] = NULL;
        _Delete_joins<_Index + 1>();
    }

    
    template<> void _Delete_joins<std::tr1::tuple_size<_Type>::value>()
    {
    }

    // Array of pointers to _Order_node elements representing each source
    void * _M_pSourceJoins[std::tr1::tuple_size<_Type>::value];

    // Join node that collects source messaging block messages
    _Join_node<_Type, _Destination_type, _Jtype> * _M_pJoinNode;

    // Tuple of messaging blocks that are sources to this multitype_join
    _Type _M_sourceTuple;

    // The scheduler to propagate messages on
    Scheduler * _M_pScheduler;

    // The schedule group to propagate messages on
    ScheduleGroup * _M_pScheduleGroup;

private:
    //
    // Hide assignment operator
    //
    multitype_join const &operator =(multitype_join const &);                      // no assignment operator
    multitype_join(multitype_join const &);                                        // no copy constructor
};

// Templated factory functions that create a join, three flavors

#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP

template<typename _Type1, typename _Type2, typename... _Types>
multitype_join<std::tuple<_Type1, _Type2, _Types...>>
make_join(Scheduler& _PScheduler, _Type1 _Item1, _Type2 _Item2, _Types... _Items)
{
    return multitype_join<std::tuple<_Type1, _Type2, _Types...>>(_PScheduler, std::make_tuple(_Item1, _Item2, _Items...));
}


template<typename _Type1, typename _Type2, typename... _Types>
multitype_join<std::tuple<_Type1, _Type2, _Types...>>
make_join(ScheduleGroup& _PScheduleGroup, _Type1 _Item1, _Type2 _Item2, _Types... _Items)
{
    return multitype_join<std::tuple<_Type1, _Type2, _Types...>>(_PScheduleGroup, std::make_tuple(_Item1, _Item2, _Items...));
}

#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */


template<typename _Type1, typename _Type2, typename... _Types>
multitype_join<std::tuple<_Type1, _Type2, _Types...>>
make_join(_Type1 _Item1, _Type2 _Item2, _Types... _Items)
{
    return multitype_join<std::tuple<_Type1, _Type2, _Types...>>(std::make_tuple(_Item1, _Item2, _Items...));
}

// Templated factory functions that create a *greedy* join, three flavors
#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP

template<typename _Type1, typename _Type2, typename... _Types>
multitype_join<std::tuple<_Type1, _Type2, _Types...>, greedy>
make_greedy_join(Scheduler& _PScheduler, _Type1 _Item1, _Type2 _Item2, _Types... _Items)
{
    return multitype_join<std::tuple<_Type1, _Type2, _Types...>, greedy>(_PScheduler, std::make_tuple(_Item1, _Item2, _Items...));
}


template<typename _Type1, typename _Type2, typename... _Types>
multitype_join<std::tuple<_Type1, _Type2, _Types...>, greedy>
make_greedy_join(ScheduleGroup& _PScheduleGroup, _Type1 _Item1, _Type2 _Item2, _Types... _Items)
{
    return multitype_join<std::tuple<_Type1, _Type2, _Types...>, greedy>(_PScheduleGroup, std::make_tuple(_Item1, _Item2, _Items...));
}

#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */


template<typename _Type1, typename _Type2, typename... _Types>
multitype_join<std::tuple<_Type1, _Type2, _Types...>, greedy>
make_greedy_join(_Type1 _Item1, _Type2 _Item2, _Types... _Items)
{
    return multitype_join<std::tuple<_Type1, _Type2, _Types...>, greedy>(std::make_tuple(_Item1, _Item2, _Items...));
}

//**************************************************************************
// Agents:
//**************************************************************************


enum agent_status {
    
    agent_created,
    
    agent_runnable,
    
    agent_started,
    
    agent_done,
    
    agent_canceled
};


class agent
{
public:
    
    _CRTIMP2 agent();

#ifdef _CRT_USE_WINAPI_FAMILY_DESKTOP_APP
    
    _CRTIMP2 agent(Scheduler& _PScheduler);

    
    _CRTIMP2 agent(ScheduleGroup& _PGroup);
#endif  /* _CRT_USE_WINAPI_FAMILY_DESKTOP_APP */

    
    _CRTIMP2 virtual ~agent();

    
    _CRTIMP2 ISource<agent_status> * status_port();

    
    _CRTIMP2 agent_status status();

    
    _CRTIMP2 bool start();

    
    _CRTIMP2 bool cancel();

    
    _CRTIMP2 static agent_status __cdecl wait(_Inout_ agent * _PAgent, unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE);

    
    _CRTIMP2 static void __cdecl wait_for_all(size_t _Count, _In_reads_(_Count) agent ** _PAgents,
        _Out_writes_opt_(_Count) agent_status * _PStatus = NULL, unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE);

    
    _CRTIMP2 static void __cdecl wait_for_one(size_t _Count, _In_reads_(_Count) agent ** _PAgents, agent_status& _Status,
                                      size_t& _Index, unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE);

protected:
    
    virtual void run() = 0;

    
    _CRTIMP2 bool done();

    
    overwrite_buffer<agent_status> _M_status;

private:

    // A flag to check of whether the agent can be started
    // This is initialized to TRUE and there is a race between Start() and Cancel() to set it
    // to FALSE.  Once Started or Canceled, further calls to Start() or Cancel() will return false.
    
    volatile long _M_fStartable;

    // A flag to check of whether the agent can be canceled
    // This is initailized to TRUE and there is a race between Cancel() and the LWT executing
    // a task that has been started to set it to FALSE.  If Cancel() wins, the task will not be
    // executed.  If the LWT wins, Cancel() will return false.
    
    volatile long _M_fCancelable;

    // A static wrapper function that calls the Run() method.  Used for scheduling of the task
    
    static void __cdecl _Agent_task_wrapper(void * data);

    Scheduler * _M_pScheduler;
    ScheduleGroup * _M_pScheduleGroup;

    //
    // Hide assignment operator and copy constructor
    //
    agent const &operator =(agent const&);  // no assignment operator
    agent(agent const &);                   // no copy constructor
};

//**************************************************************************
// Direct Messaging APIs:
//**************************************************************************


template <class _Type>
_Type _Receive_impl(ISource<_Type> * _Src, unsigned int _Timeout, typename ITarget<_Type>::filter_method const* _Filter_proc)
{
    // The Blocking Recipient messaging block class is internal to the receive function
    class _Blocking_recipient : public ITarget<_Type>
    {
    public:
        // Create an Blocking Recipient
        _Blocking_recipient(ISource<_Type> * _PSource,
            unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE) :
            _M_pFilter(NULL), _M_pConnectedTo(NULL), _M_pMessage(NULL), _M_fState(_NotInitialized), _M_timeout(_Timeout)
        {
            _Connect(_PSource);
        }

        // Create an Blocking Recipient
        _Blocking_recipient(ISource<_Type> * _PSource,
            filter_method const& _Filter,
            unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE) :
            _M_pFilter(NULL), _M_pConnectedTo(NULL), _M_pMessage(NULL), _M_fState(_NotInitialized), _M_timeout(_Timeout)
        {
            if (_Filter != NULL)
            {
                _M_pFilter = new filter_method(_Filter);
            }

            _Connect(_PSource);
        }

        // Cleans up any resources that may have been created by the BlockingRecipient.
        ~_Blocking_recipient()
        {
            _Disconnect();

            delete _M_pFilter;
            delete _M_pMessage;
        }

        // Gets the value of the message sent to this BlockingRecipient.  Blocks by
        // spinning until a message has arrived.
        _Type _Value()
        {
            _Wait_for_message();

            return _M_pMessage->payload;
        }

        // The main propagation function for ITarget blocks.  Called by a source
        // block, generally within an asynchronous task to send messages to its targets.
        virtual message_status propagate(message<_Type> * _PMessage, ISource<_Type> * _PSource)
        {
            // Throw exception if the message being propagated to this block is NULL
            if (_PMessage == NULL)
            {
                throw std::invalid_argument("_PMessage");
            }

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

            // Reject if the recipient has already received a message
            if (_M_fState == _Initialized)
            {
                return declined;
            }

            // Reject if the message does not meet the filter requirements
            if (_M_pFilter != NULL && !(*_M_pFilter)(_PMessage->payload))
            {
                return declined;
            }

            // Accept the message
            _CONCRT_ASSERT(_PSource != NULL);
            _M_pMessage = _PSource->accept(_PMessage->msg_id(), this);

            if (_M_pMessage != NULL)
            {
                // Set the initialized flag on this block
                if (_InterlockedExchange(&_M_fState, _Initialized) == _Blocked)
                {
                    _M_ev.set();
                }

                return accepted;
            }

            return missed;
        }

        // Synchronously sends a message to this block.  When this function completes the message will
        // already have propagated into the block.
        virtual message_status send(message<_Type> * _PMessage, ISource<_Type> * _PSource)
        {
            if (_PMessage == NULL)
            {
                throw std::invalid_argument("_PMessage");
            }

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

            // Only the connected source is allowed to send messages
            // to the blocking recepient. Decline messages without
            // a source.

            return declined;
        }

    private:

        // Link a source block
        virtual void link_source(ISource<_Type> * _PSrc)
        {
            _M_pConnectedTo = _PSrc;
            _PSrc->acquire_ref(this);
        }

        // Remove a source messaging block for this BlockingRecipient
        virtual void unlink_source(ISource<_Type> * _PSource)
        {
            if (_InterlockedCompareExchangePointer(reinterpret_cast<void *volatile *>(&_M_pConnectedTo), (void *)NULL, _PSource) == _PSource)
            {
                _PSource->release_ref(this);
            }
        }

        // Remove the source messaging block for this BlockingRecipient
        virtual void unlink_sources()
        {
            ISource<_Type> * _PSource = reinterpret_cast<ISource<_Type> *>(_InterlockedExchangePointer(reinterpret_cast<void *volatile *>(&_M_pConnectedTo), (void *)NULL));
            if (_PSource != NULL)
            {
                _PSource->unlink_target(this);
                _PSource->release_ref(this);
            }
        }


        // Connect the blocking recipient to the source
        void _Connect(ISource<_Type> * _PSource)
        {
            if (_PSource == NULL)
            {
                throw std::invalid_argument("_PSource");
            }

            _PSource->link_target(this);
        }

        // Cleanup the connection to the blocking recipient's source. There is no need
        // to do anything about the associated context.
        void _Disconnect()
        {
            unlink_sources();
        }

        // Internal function used to block while waiting for a message to arrive
        // at this BlockingRecipient
        void _Wait_for_message()
        {
            bool _Timeout = false;

            // If we haven't received a message yet, cooperatively block.
            if (_InterlockedCompareExchange(&_M_fState, _Blocked, _NotInitialized) == _NotInitialized)
            {
                if (_M_ev.wait(_M_timeout) == COOPERATIVE_WAIT_TIMEOUT)
                {
                    _Timeout = true;
                }
            }

            // Unlinking from our source guarantees that there are no threads in propagate
            _Disconnect();

            if (_M_fState != _Initialized)
            {
                // We had to have timed out if we came out of the wait
                // without being initialized.
                _CONCRT_ASSERT(_Timeout);

                throw operation_timed_out();
            }
        }

        // States for this block
        enum
        {
            _NotInitialized,
            _Blocked,
            _Initialized
        };

        volatile long _M_fState;

        // The source messaging block connected to this Recipient
        ISource<_Type> * _M_pConnectedTo;

        // The message that was received
        message<_Type> * volatile _M_pMessage;

        // The timeout.
        unsigned int _M_timeout;

        // The event we wait upon
        event _M_ev;

        // The filter that is called on this block before accepting a message
        filter_method * _M_pFilter;
    };

    if (_Filter_proc != NULL)
    {
        _Blocking_recipient _Recipient(_Src, *_Filter_proc, _Timeout);
        return _Recipient._Value();
    }
    else
    {
        _Blocking_recipient _Recipient(_Src, _Timeout);
        return _Recipient._Value();
    }
}


template <class _Type>
_Type receive(_Inout_ ISource<_Type> * _Src, unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE)
{
    return _Receive_impl(_Src, _Timeout, NULL);
}


template <class _Type>
_Type receive(_Inout_ ISource<_Type> * _Src, typename ITarget<_Type>::filter_method const& _Filter_proc, unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE)
{
    return _Receive_impl(_Src, _Timeout, &_Filter_proc);
}


template <class _Type>
_Type receive(ISource<_Type> &_Src, unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE)
{
    return _Receive_impl(&_Src, _Timeout, NULL);
}


template <class _Type>
_Type receive(ISource<_Type> &_Src, typename ITarget<_Type>::filter_method const& _Filter_proc, unsigned int _Timeout = COOPERATIVE_TIMEOUT_INFINITE)
{
    return _Receive_impl(&_Src, _Timeout, &_Filter_proc);
}


template <class _Type>
bool _Try_receive_impl(ISource<_Type> * _Src, _Type & _value, typename ITarget<_Type>::filter_method const * _Filter_proc)
{
    // The Immediate Recipient messaging block class is internal to the receive function
    class _Immediate_recipient : public ITarget<_Type>
    {
    public:
        // Create an Immediate Recipient
        _Immediate_recipient(ISource<_Type> * _PSource) :
            _M_pFilter(NULL), _M_pConnectedTo(NULL), _M_pMessage(NULL), _M_isInitialized(0)
        {
            _Connect(_PSource);
        }

        // Create an Immediate Recipient
        _Immediate_recipient(ISource<_Type> * _PSource,
            filter_method const& _Filter) :
            _M_pFilter(NULL), _M_pConnectedTo(NULL), _M_pMessage(NULL), _M_isInitialized(0)
        {
            if (_Filter != NULL)
            {
                _M_pFilter = new filter_method(_Filter);
            }

            _Connect(_PSource);
        }

        // Cleans up any resources that may have been created by the ImmediateRecipient.
        ~_Immediate_recipient()
        {
            _Disconnect();

            delete _M_pFilter;
            delete _M_pMessage;
        }

        // Gets the value of the message sent to this ImmediateRecipient.
        bool _Value(_Type & _value)
        {
            // Unlinking from our source guarantees that there are no threads in propagate
            _Disconnect();

            if (_M_pMessage != NULL)
            {
                _value = _M_pMessage->payload;
                return true;
            }

            return false;
        }

        // The main propagation function for ITarget blocks.  Called by a source
        // block, generally within an asynchronous task to send messages to its targets.
        virtual message_status propagate(message<_Type> * _PMessage, ISource<_Type> * _PSource)
        {
            message_status _Result = accepted;

            // Throw exception if the message being propagated to this block is NULL
            if (_PMessage == NULL)
            {
                throw std::invalid_argument("_PMessage");
            }

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

            // Reject if the recipient has already received a message
            if (_M_isInitialized == 1)
            {
                return declined;
            }

            // Reject if the message does not meet the filter requirements
            if (_M_pFilter != NULL && !(*_M_pFilter)(_PMessage->payload))
            {
                return declined;
            }

            // Accept the message
            _CONCRT_ASSERT(_PSource != NULL);
            _M_pMessage = _PSource->accept(_PMessage->msg_id(), this);

            // Set the initialized flag on this block

            if (_M_pMessage != NULL)
            {
                // Fence to ensure that the above update to _M_pMessage is visible
                _InterlockedExchange(&_M_isInitialized, 1);
                _Result = accepted;
            }
            else
            {
                _Result = missed;
            }

            return _Result;
        }


        // Synchronously sends a message to this block.  When this function completes the message will
        // already have propagated into the block.
        virtual message_status send(message<_Type> * _PMessage, ISource<_Type> * _PSource)
        {
            if (_PMessage == NULL)
            {
                throw std::invalid_argument("_PMessage");
            }

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

            // Only the connected source is allowed to send messages
            // to the blocking recepient. Decline messages without
            // a source.

            return declined;
        }

    private:

        // Add a source messaging block
        virtual void link_source(ISource<_Type> * _PSrc)
        {
            _M_pConnectedTo = _PSrc;
            _PSrc->acquire_ref(this);
        }

        // Remove a source messaging block for this BlockingRecipient
        virtual void unlink_source(ISource<_Type> * _PSource)
        {
            if (_InterlockedCompareExchangePointer(reinterpret_cast<void *volatile *>(&_M_pConnectedTo), (void *)NULL, _PSource) == _PSource)
            {
                _PSource->release_ref(this);
            }
        }

        // Remove the source messaging block for this BlockingRecipient
        virtual void unlink_sources()
        {
            ISource<_Type> * _PSource = reinterpret_cast<ISource<_Type> *>(_InterlockedExchangePointer(reinterpret_cast<void *volatile *>(&_M_pConnectedTo), (void *)NULL));
            if (_PSource != NULL)
            {
                _PSource->unlink_target(this);
                _PSource->release_ref(this);
            }
        }

        // Connect to a source block
        void _Connect(ISource<_Type> * _PSource)
        {
            if (_PSource == NULL)
            {
                throw std::invalid_argument("_PSource");
            }

            _CONCRT_ASSERT(_M_isInitialized == 0);

            _PSource->link_target(this);
        }

        //
        // Cleanup the connection to the trigger's source. There is no need
        // to do anything about the associated context.
        //
        void _Disconnect()
        {
            unlink_sources();
        }

        // The source messaging block connected to this Recipient
        ISource<_Type> * _M_pConnectedTo;

        // The message that was received
        message<_Type> * volatile _M_pMessage;

        // A flag for whether or not this block has been initialized with a value
        volatile long _M_isInitialized;

        // The filter that is called on this block before accepting a message
        filter_method * _M_pFilter;
    };

    if (_Filter_proc != NULL)
    {
        _Immediate_recipient _Recipient(_Src, *_Filter_proc);
        return _Recipient._Value(_value);
    }
    else
    {
        _Immediate_recipient _Recipient(_Src);
        return _Recipient._Value(_value);
    }
}


template <class _Type>
bool try_receive(_Inout_ ISource<_Type> * _Src, _Type & _value)
{
    return _Try_receive_impl(_Src, _value, NULL);
}


template <class _Type>
bool try_receive(_Inout_ ISource<_Type> * _Src, _Type & _value, typename ITarget<_Type>::filter_method const& _Filter_proc)
{
    return _Try_receive_impl(_Src, _value, &_Filter_proc);
}


template <class _Type>
bool try_receive(ISource<_Type> & _Src, _Type & _value)
{
    return _Try_receive_impl(&_Src, _value, NULL);
}


template <class _Type>
bool try_receive(ISource<_Type> & _Src, _Type & _value, typename ITarget<_Type>::filter_method const& _Filter_proc)
{
    return _Try_receive_impl(&_Src, _value, &_Filter_proc);
}

namespace details
{
    //**************************************************************************
    // Supporting blocks for send and asend
    //**************************************************************************

    // Originator block that pushes messages to a target
    template <class _Type>
    class _AnonymousOriginator : public ISource<_Type>
    {
    public:

        typedef single_link_registry<ITarget<_Type>> _Target_registry;

        // Create an Originator
        _AnonymousOriginator() : _M_pMessage(NULL),  _M_pTarget(NULL)
        {
        }

        // Cleans up any resources that may have been created by the Originator.
        virtual ~_AnonymousOriginator()
        {
            delete _M_pMessage;
        }

        // Removes a target messaging block for this Originator
        virtual void unlink_target(ITarget<_Type> * _PTarget)
        {
            throw invalid_operation("unlink_target is not supported on _AnonymousOriginator");
        }

        // Removes the target messaging block from this Originator
        virtual void unlink_targets()
        {
            throw invalid_operation("unlink_targets is not supported on _AnonymousOriginator");
        }

        // Accept on this Originator is called by a target to take ownership of a
        // propagated message
        virtual message<_Type> * accept(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget)
        {
            if (_PTarget != _M_pTarget)
            {
                return NULL;
            }

            if (_M_pMessage == NULL || _M_pMessage->msg_id() != _MsgId)
            {
                return NULL;
            }

            // The IDs match, actaully 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;

            return _Result;
        }

        // Reserve shall not be called by blocks that supports push
        virtual bool reserve(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget)
        {
            throw invalid_operation("reserve is not supported on _AnonymousOriginator");
        }

        // Consume shall not be called by blocks that supports push
        virtual message<_Type> * consume(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget)
        {
            throw invalid_operation("consume is not supported on _AnonymousOriginator");
        }

        // Release needs to be defined for ISource blocks, but Originator doesn't need to
        // do anything for reservation release because there can only be one target block to read
        // the data at a later time.
        virtual void release(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget)
        {
            throw invalid_operation("release is not supported on _AnonymousOriginator");
        }

        virtual void acquire_ref(_Inout_ ITarget<_Type> *)
        {
            throw invalid_operation("acquire_ref is not supported on _AnonymousOriginator");
        }

        virtual void release_ref(_Inout_ ITarget<_Type> *)
        {
            throw invalid_operation("release_ref is not supported on _AnonymousOriginator");
        }

    private:
        friend class _Originator;

        // Send the given value to the target
        bool _internal_send(ITarget<_Type> * _PTarget, _Type const & _Value)
        {
            _M_pTarget = _PTarget;

            _CONCRT_ASSERT(_M_pTarget != NULL);
            _CONCRT_ASSERT(_M_pTarget->supports_anonymous_source());

            // Create the message
            message_status _Status = declined;
            message<_Type> * _Msg = new message<_Type>(_Value);

            _CONCRT_ASSERT(_M_pMessage == NULL);
            _M_pMessage = _Msg;

            // Send the message
            _Status = _M_pTarget->send(_M_pMessage, this);

            // If the message is declined, the destructor will
            // delete the message

            // status should not be postponed.
            _CONCRT_ASSERT(_Status != postponed);
            return (_Status == accepted);
        }

        bool _internal_asend(ITarget<_Type> * _PTarget, _Type const & _Value)
        {
            _M_pTarget = _PTarget;

            _CONCRT_ASSERT(_M_pTarget != NULL);
            _CONCRT_ASSERT(_M_pTarget->supports_anonymous_source());

            // Create the message
            message_status _Status = declined;
            message<_Type> * _Msg = new message<_Type>(_Value);

            _CONCRT_ASSERT(_M_pMessage == NULL);
            _M_pMessage = _Msg;

            // Send the message
            _Status = _M_pTarget->propagate(_M_pMessage, this);

            // If the message is declined, the destructor will
            // delete the message

            // status should not be postponed.
            if (_Status == postponed)
            {
                throw invalid_operation("Messages offered by _AnonymousOriginator shall not be postponed");
            }

            return (_Status == accepted);
        }

        // Add a target messaging block for this Originator
        virtual void link_target(ITarget<_Type> * _PTarget)
        {
            throw invalid_operation("link_target is not supported on _AnonymousOriginator");
        }

        // The message that will be propagated by the Originator
        message<_Type> * _M_pMessage;

        // The single target for this block
        ITarget<_Type> * _M_pTarget;
    };

    // The Originator messaging block class is internal to the send function.
    template <class _Type>
    class _SyncOriginator : public ISource<_Type>
    {
    public:

        typedef single_link_registry<ITarget<_Type>> _Target_registry;

        // Create an Originator
        _SyncOriginator() :
          _M_pMessage(NULL),
          _M_fStatus(postponed),
          _M_referenceCount(0)
        {
        }

        // Cleans up any resources that may have been created by the Originator.
        virtual ~_SyncOriginator()
        {
            unlink_targets();

            _Wait_on_ref();

            delete _M_pMessage;
        }

        // Removes a target messaging block for this Originator
        virtual void unlink_target(ITarget<_Type> * _PTarget)
        {
            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))
                {
                    _Invoke_unlink_source(_PTarget);

                    // Indicate that the send is complete
                    _Done(declined);
                }
            }
        }

        // Removes the target messaging block from this Originator
        virtual void unlink_targets()
        {
            // Hold the lock to ensure that the target doesn't unlink while
            // propagation is in progress.
            _R_lock _Lock(_M_internalLock);

            for (_Target_registry::iterator _Iter = _M_connectedTargets.begin(); *_Iter != NULL; ++_Iter)
            {
                ITarget<_Type> * _PTarget = *_Iter;
                if (_M_connectedTargets.remove(_PTarget))
                {
                    _Invoke_unlink_source(_PTarget);
                }
            }

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

            // Indicate that the send is complete
            _Done(declined);
        }

        // Accept on this Originator is called by a target to take ownership of a
        // propagated message
        virtual message<_Type> * accept(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget)
        {
            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, actaully 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;
        }

        // Reserve needs to be defined for ISource blocks, but Originator doesn't need to
        // do anything for reservation because there can only be one target block to read
        // the data at a later time.
        virtual bool reserve(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget)
        {
            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;
        }

        // Consume is called by a target messaging block to take ownership of a
        // previously reserved message.
        virtual message<_Type> * consume(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget)
        {
            if (_PTarget == NULL)
            {
                throw std::invalid_argument("_PTarget");
            }

            if (!_M_connectedTargets.contains(_PTarget))
            {
                throw bad_target();
            }

            return accept(_MsgId, _PTarget);
        }

        // Release needs to be defined for ISource blocks, but Originator doesn't need to
        // do anything for reservation release because there can only be one target block to read
        // the data at a later time.
        virtual void release(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget)
        {
            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);
        }

        virtual void acquire_ref(_Inout_ ITarget<_Type> *)
        {
            _InterlockedIncrement(&_M_referenceCount);
        }

        virtual void release_ref(_Inout_ ITarget<_Type> *)
        {
            _InterlockedDecrement(&_M_referenceCount);
        }

    private:

        friend class _Originator;

        // Send the given value to the target
        bool _internal_send(ITarget<_Type> * _PTarget, _Type const & _Value)
        {
            // _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);
        }

        // Add a target messaging block for this Originator
        virtual void link_target(ITarget<_Type> * _PTarget)
        {
            if (_PTarget == NULL)
            {
                throw std::invalid_argument("_PTarget");
            }

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

            // There should be no pending messages to propagate at this time.
            _CONCRT_ASSERT(_M_pMessage == NULL);
        }

        // Wait for the status to reach one of the terminal
        // states (!= postponed)
        void _Wait_for_completion()
        {
            // Wait for the event to be signalled
            _M_ev.wait(COOPERATIVE_TIMEOUT_INFINITE);
            _CONCRT_ASSERT(_M_fStatus != postponed);

        }

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

        // Indicate that the send operation has completed
        void _Done(message_status _Status)
        {
            // postponed is not a done state
            _CONCRT_ASSERT(_Status != postponed);

            _M_fStatus = _Status;
            _M_ev.set();
        }

        // The message that will be propagated by the Originator
        message<_Type> * _M_pMessage;

        // Event to indicate completion
        event _M_ev;

        // Final status of the send
        volatile message_status _M_fStatus;

        // A lock for modifying the buffer or the connected blocks
        ::Concurrency::details::_ReentrantPPLLock _M_internalLock;

        // Connected targets
        _Target_registry _M_connectedTargets;

        volatile long _M_referenceCount;
    };

    // The Originator messaging block class is internal to the send function.
    template <class _Type>
    class _AsyncOriginator : public ISource<_Type>
    {
    public:

        typedef single_link_registry<ITarget<_Type>> _Target_registry;

        // Cleans up any resources that may have been created by the AsyncOriginator.
        virtual ~_AsyncOriginator()
        {
            unlink_targets();

            delete _M_pMessage;
        }

        // Removes a target messaging block for this AsyncOriginator
        virtual void unlink_target(ITarget<_Type> * _PTarget)
        {
            if (_PTarget == NULL)
            {
                throw std::invalid_argument("_PTarget");
            }

            bool _Unlinked = false;
            {
                // 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))
                {
                    _Invoke_unlink_source(_PTarget);
                    _Unlinked = true;
                }
            }

            // Release the lock before decrementing the refcount. Otherwise, the
            // lock release could corrupt memory.
            if (_Unlinked)
            {
                _Release_ref();
            }
        }

        // Removes the target messaging block from this AsyncOriginator
        virtual void unlink_targets()
        {
            bool _Unlinked = false;
            {
                // Hold the lock to ensure that the target doesn't unlink while
                // propagation is in progress.
                _R_lock _Lock(_M_internalLock);

                for (_Target_registry::iterator _Iter = _M_connectedTargets.begin();
                    *_Iter != NULL;
                    ++_Iter)
                {
                    ITarget<_Type> * _PTarget = *_Iter;
                    if (_M_connectedTargets.remove(_PTarget))
                    {
                        _Invoke_unlink_source(_PTarget);
                        _Unlinked = true;
                    }

                }

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

            // Release the lock before decrementing the refcount. Otherwise, the
            // lock release could corrupt memory.
            if (_Unlinked)
            {
                _Release_ref();
            }
        }

        // Accept on this AsyncOriginator is called by a target to take ownership of a
        // propagated message. This can only be called from propagate.
        virtual message<_Type> * accept(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget)
        {
            if (_PTarget == NULL)
            {
                return NULL;
            }

            if (!_M_connectedTargets.contains(_PTarget))
            {
                return NULL;
            }

            if (_M_pMessage == NULL || _M_pMessage->msg_id() != _MsgId)
            {
                return NULL;
            }

            //
            // If the IDs match, actaully transfer ownership of the message.
            //
            message<_Type> * _Result = _M_pMessage;
            _M_pMessage = NULL;

            return _Result;
        }

        // Reserve needs to be defined for ISource blocks, but AsyncOriginator doesn't need to
        // do anything for reservation because there can only be one target block to read
        // the data at a later time.
        virtual bool reserve(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget)
        {
            if (_PTarget == NULL)
            {
                throw std::invalid_argument("_PTarget");
            }

            if (!_M_connectedTargets.contains(_PTarget))
            {
                return false;
            }

            if (_M_pMessage == NULL || _M_pMessage->msg_id() != _MsgId)
            {
                return false;
            }

            return true;
        }

        // Consume is called by a target messaging block to take ownership of a
        // previously reserved message.
        virtual message<_Type> * consume(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget)
        {
            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)
            {
                return NULL;
            }

            // The ownership of this message has changed.  Set the internal pointer to NULL
            // so it won't be deleted in the destructor

            message<_Type> * _Result = _M_pMessage;
            _M_pMessage = NULL;

            // We are done. Unlink from the target. DO NOT TOUCH "this" pointer after unlink
            unlink_target(_PTarget);

            return _Result;
        }

        // Release needs to be defined for ISource blocks, but AsyncOriginator doesn't need to
        // do anything for reservation release because there can only be one target block to read
        // the data at a later time.
        virtual void release(runtime_object_identity _MsgId, ITarget<_Type> * _PTarget)
        {
            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();
            }

            // We can be connected to only 1 target. Unlink from the target.
            // DO NOT TOUCH "this" pointer after unlink
            unlink_target(_PTarget);
        }

        virtual void acquire_ref(_Inout_ ITarget<_Type> *)
        {
            _Acquire_ref();
        }

        virtual void release_ref(_Inout_ ITarget<_Type> *)
        {
            _Release_ref();
        }

    private:

        friend class _Originator;

        // Create an AsyncOriginator (constructor is private to ensure that
        // it is allocated on the heap).
        _AsyncOriginator() :
          _M_pMessage(NULL),
          _M_refcount(0)
        {
        }

        // Send the given value to the target
        bool _internal_send(ITarget<_Type> * _PTarget, _Type const & _Value)
        {
            // Keep a refcount so that this object doesn't get deleted if
            // the target decides to unlink before we release our lock
            _Acquire_ref();

            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;

                _Status = _PTarget->propagate(_M_pMessage, this);
            }

            // If the status is anything other than postponed, unlink away
            // from the target and delete the AsyncOriginator.
            if (_Status != postponed)
            {
                unlink_target(_PTarget);
            }

            // Release the reference acquired above
            _Release_ref();

            return (_Status == accepted);
        }

        // Add a target messaging block for this AsyncOriginator
        virtual void link_target(ITarget<_Type> * _PTarget)
        {
            if (_PTarget == NULL)
            {
                throw std::invalid_argument("_PTarget");
            }

            // Acquire a reference that will be released by unlink_target
            _Acquire_ref();
            _M_connectedTargets.add(_PTarget);
            _Invoke_link_source(_PTarget);

            // There should be no pending messages to propagate at this time.
            _CONCRT_ASSERT(_M_pMessage == NULL);

        }

        // Acquire a reference on the async originator object
        void _Acquire_ref()
        {
            _InterlockedIncrement(&_M_refcount);
        }

        // Release the reference on the async originator object. The object
        // will be deleted when the reference count goes to 0.
        void _Release_ref()
        {
            _CONCRT_ASSERT(_M_refcount > 0);
            if (_InterlockedDecrement(&_M_refcount) == 0)
            {
                delete this;
            }
        }

        // The message that will be propagated by the AsyncOriginator
        message<_Type> * _M_pMessage;

        // Reference count to manage object lifetime
        volatile long _M_refcount;

        // The internal lock for this block for its message
        ::Concurrency::details::_ReentrantPPLLock _M_internalLock;

        // connected targets
        _Target_registry _M_connectedTargets;
    };

    // static class that exposes methods to initiate messages into
    // a dataflow network
    class _Originator
    {
    public:

        // Synchronous initiation of messages
        template <class _Type>
        static bool _send(ITarget<_Type> * _Trg, const _Type& _Data)
        {
            if (_Trg != NULL && _Trg->supports_anonymous_source())
            {
                // _send will block until the message is accepted/rejected.
                // Note that this invokes the send method on the target which
                // would synchronously process the message.
                _AnonymousOriginator<_Type> _Send_block;
                return _Send_block._internal_send(_Trg, _Data);
            }
            else
            {
                // Create a blocking originator on the stack. _send will block until the
                // message is accepted/rejected.
                _SyncOriginator<_Type> _Orig;
                return _Orig._internal_send(_Trg, _Data);
            }
        }

        // Asynchronous initiation of messages
        template <class _Type>
        static bool _asend(ITarget<_Type> * _Trg, const _Type& _Data)
        {
            // If the block can participate in posting messages without requiring a call back, use that
            // method of initiating the message rather for efficiency purposes.
            if (_Trg != NULL && _Trg->supports_anonymous_source())
            {
                _AnonymousOriginator<_Type> _Asend_block;
                return _Asend_block._internal_asend(_Trg, _Data);
            }
            else
            {
                // Needs to be allocated on the heap
                _AsyncOriginator<_Type> * _AsyncOrig = new _AsyncOriginator<_Type>;
                return _AsyncOrig->_internal_send(_Trg, _Data);
            }
        }
    };

} // namespace details


template <class _Type>
bool send(_Inout_ ITarget<_Type> * _Trg, const _Type& _Data)
{
    return details::_Originator::_send(_Trg, _Data);
}



template <class _Type>
bool send(ITarget<_Type> &_Trg, const _Type &_Data)
{
    return send(&_Trg, _Data);
}


template <class _Type>
bool asend(_Inout_ ITarget<_Type> * _Trg, const _Type& _Data)
{
    return details::_Originator::_asend(_Trg, _Data);
}



template <class _Type>
bool asend(ITarget<_Type> &_Trg, const _Type &_Data)
{
    return asend(&_Trg, _Data);
}

template <class _Type>
void Trace_agents_register_name(_Inout_ _Type * _PObject, _In_z_ const wchar_t * _Name)
{
    _Trace_agents(AGENTS_EVENT_NAME, ::Concurrency::details::_Trace_agents_get_id(_PObject), _Name);
}

} // namespace Concurrency

namespace concurrency = Concurrency;

#pragma warning(pop)
#pragma pack(pop)