Code:
/ 4.0 / 4.0 / untmp / DEVDIV_TFS / Dev10 / Releases / RTMRel / ndp / fx / src / Sys / System / Collections / Concurrent / BlockingCollection.cs / 1305376 / BlockingCollection.cs
// ==++==
//
// Copyright (c) Microsoft Corporation. All rights reserved.
//
// ==--==
// =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
//
// BlockingCollection.cs
//
// [....]
//
// A class that implements the bounding and blocking functionality while abstracting away
// the underlying storage mechanism. This file also contains BlockingCollection's
// associated debugger view type.
//
// =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
#pragma warning disable 0420
using System;
using System.Collections.Generic;
using System.Collections;
using System.Diagnostics;
using System.Globalization;
using System.Security.Permissions;
using System.Runtime.InteropServices;
using System.Threading;
namespace System.Collections.Concurrent
{
///
/// Provides blocking and bounding capabilities for thread-safe collections that
/// implement
///
///
/// Specifies the type of elements in the collection.
[ComVisible(false)]
[HostProtection(SecurityAction.LinkDemand, Synchronization = true, ExternalThreading = true)]
[DebuggerTypeProxy(typeof(SystemThreadingCollections_BlockingCollectionDebugView<>))]
[DebuggerDisplay("Count = {Count}, Type = {m_collection}")]
public class BlockingCollection : IEnumerable, ICollection, IDisposable
{
private IProducerConsumerCollection m_collection;
private int m_boundedCapacity;
private const int NON_BOUNDED = -1;
private SemaphoreSlim m_freeNodes;
private SemaphoreSlim m_occupiedNodes;
private volatile bool m_isDisposed;
private CancellationTokenSource m_ConsumersCancellationTokenSource;
private CancellationTokenSource m_ProducersCancellationTokenSource;
private volatile int m_currentAdders;
private const int COMPLETE_ADDING_ON_MASK = unchecked((int)0x80000000);
#region Enums
/// An enumerated data type used internal to the class to specify to a generic method
/// the current mode of operation.
private enum OperationMode
{
Add, Take
};
#endregion
#region Properties
/// Gets the bounded capacity of this
/// The bounded capacity of this collection, or int.MaxValue if no bound was supplied.
/// The
public int BoundedCapacity
{
get
{
CheckDisposed();
return m_boundedCapacity;
}
}
/// Gets whether this
/// Whether this collection has been marked as complete for adding.
/// The
public bool IsAddingCompleted
{
get
{
CheckDisposed();
return (m_currentAdders == COMPLETE_ADDING_ON_MASK);
}
}
/// Gets whether this
/// Whether this collection has been marked as complete for adding and is empty.
/// The
public bool IsCompleted
{
get
{
CheckDisposed();
return (IsAddingCompleted && (m_occupiedNodes.CurrentCount == 0));
}
}
/// Gets the number of items contained in the
/// The number of items contained in the
/// The
public int Count
{
get
{
CheckDisposed();
return m_occupiedNodes.CurrentCount;
}
}
/// Gets a value indicating whether access to the
/// The
bool ICollection.IsSynchronized
{
get
{
CheckDisposed();
return false;
}
}
///
/// Gets an object that can be used to synchronize access to the
/// The SyncRoot property is not supported.
object ICollection.SyncRoot
{
get
{
throw new NotSupportedException(SR.GetString(SR.ConcurrentCollection_SyncRoot_NotSupported));
}
}
#endregion
/// Initializes a new instance of the
///
///
/// The default underlying collection is a ConcurrentQueue<T> .
///
public BlockingCollection()
: this(new ConcurrentQueue())
{
}
/// Initializes a new instance of the
/// The bounded size of the collection.
/// The
///
/// The default underlying collection is a ConcurrentQueue<T> .
///
public BlockingCollection(int boundedCapacity)
: this(new ConcurrentQueue(), boundedCapacity)
{
}
/// Initializes a new instance of the
/// The collection to use as the underlying data store.
/// The bounded size of the collection.
/// The
/// The
/// The supplied
public BlockingCollection(IProducerConsumerCollection collection, int boundedCapacity)
{
if (boundedCapacity < 1)
{
throw new ArgumentOutOfRangeException(
"boundedCapacity", boundedCapacity,
SR.GetString(SR.BlockingCollection_ctor_BoundedCapacityRange));
}
if (collection == null)
{
throw new ArgumentNullException("collection");
}
int count = collection.Count;
if (count > boundedCapacity)
{
throw new ArgumentException(SR.GetString(SR.BlockingCollection_ctor_CountMoreThanCapacity));
}
Initialize(collection, boundedCapacity, count);
}
/// Initializes a new instance of the
/// The collection to use as the underlying data store.
/// The
public BlockingCollection(IProducerConsumerCollection collection)
{
if (collection == null)
{
throw new ArgumentNullException("collection");
}
Initialize(collection, NON_BOUNDED, collection.Count);
}
/// Initializes the BlockingCollection instance.
/// The collection to use as the underlying data store.
/// The bounded size of the collection.
/// The number of items currently in the underlying collection.
private void Initialize(IProducerConsumerCollection collection, int boundedCapacity, int collectionCount)
{
Debug.Assert(boundedCapacity > 0 || boundedCapacity == NON_BOUNDED);
m_collection = collection;
m_boundedCapacity = boundedCapacity; ;
m_isDisposed = false;
m_ConsumersCancellationTokenSource = new CancellationTokenSource();
m_ProducersCancellationTokenSource = new CancellationTokenSource();
if (boundedCapacity == NON_BOUNDED)
{
m_freeNodes = null;
}
else
{
Debug.Assert(boundedCapacity > 0);
m_freeNodes = new SemaphoreSlim(boundedCapacity - collectionCount);
}
m_occupiedNodes = new SemaphoreSlim(collectionCount);
}
///
/// Adds the item to the
/// The item to be added to the collection. The value can be a null reference.
/// The
/// The
/// The underlying collection didn't accept the item.
///
/// If a bounded capacity was specified when this instance of
///
public void Add(T item)
{
#if DEBUG
bool tryAddReturnValue =
#endif
TryAddWithNoTimeValidation(item, Timeout.Infinite, new CancellationToken());
#if DEBUG
Debug.Assert(tryAddReturnValue, "TryAdd() was expected to return true.");
#endif
}
///
/// Adds the item to the
/// The item to be added to the collection. The value can be a null reference.
/// A cancellation token to observe.
/// If the
/// The
/// The
/// The underlying collection didn't accept the item.
///
/// If a bounded capacity was specified when this instance of
///
public void Add(T item, CancellationToken cancellationToken)
{
#if DEBUG
bool tryAddReturnValue =
#endif
TryAddWithNoTimeValidation(item, Timeout.Infinite, cancellationToken);
#if DEBUG
Debug.Assert(tryAddReturnValue, "TryAdd() was expected to return true.");
#endif
}
///
/// Attempts to add the specified item to the
/// The item to be added to the collection.
/// true if the
/// The
/// The
/// The underlying collection didn't accept the item.
public bool TryAdd(T item)
{
return TryAddWithNoTimeValidation(item, 0, new CancellationToken());
}
///
/// Attempts to add the specified item to the
/// The item to be added to the collection.
/// A true if the
/// The
/// The
/// The underlying collection didn't accept the item.
public bool TryAdd(T item, TimeSpan timeout)
{
ValidateTimeout(timeout);
return TryAddWithNoTimeValidation(item, (int)timeout.TotalMilliseconds, new CancellationToken());
}
///
/// Attempts to add the specified item to the
/// The item to be added to the collection.
/// The number of milliseconds to wait, or true if the
/// The
/// The
/// The underlying collection didn't accept the item.
public bool TryAdd(T item, int millisecondsTimeout)
{
ValidateMillisecondsTimeout(millisecondsTimeout);
return TryAddWithNoTimeValidation(item, millisecondsTimeout, new CancellationToken());
}
///
/// Attempts to add the specified item to the
/// The item to be added to the collection.
/// The number of milliseconds to wait, or true if the
/// If the
/// The
/// The
/// The underlying collection didn't accept the item.
public bool TryAdd(T item, int millisecondsTimeout, CancellationToken cancellationToken)
{
ValidateMillisecondsTimeout(millisecondsTimeout);
return TryAddWithNoTimeValidation(item, millisecondsTimeout, cancellationToken);
}
/// Adds an item into the underlying data store using its IProducerConsumerCollection<T>.Add
/// method. If a bounded capacity was specified and the collection was full,
/// this method will wait for, at most, the timeout period trying to add the item.
/// If the timeout period was exhaused before successfully adding the item this method will
/// return false.
/// The item to be added to the collection.
/// The number of milliseconds to wait for the collection to accept the item,
/// or Timeout.Infinite to wait indefinitely.
/// A cancellation token to observe.
/// False if the collection remained full till the timeout period was exhausted.True otherwise.
/// If the
/// the collection has already been marked
/// as complete with regards to additions.
/// If the collection has been disposed.
/// The underlying collection didn't accept the item.
private bool TryAddWithNoTimeValidation(T item, int millisecondsTimeout, CancellationToken cancellationToken)
{
CheckDisposed();
if (cancellationToken.IsCancellationRequested)
throw new OperationCanceledException(SR.GetString(SR.Common_OperationCanceled), cancellationToken);
if (IsAddingCompleted)
{
throw new InvalidOperationException(SR.GetString(SR.BlockingCollection_Completed));
}
bool waitForSemaphoreWasSuccessful = true;
if (m_freeNodes != null)
{
//If the m_freeNodes semaphore threw OperationCanceledException then this means that CompleteAdding()
//was called concurrently with Adding which is not supported by BlockingCollection.
CancellationTokenSource linkedTokenSource = null;
try
{
waitForSemaphoreWasSuccessful = m_freeNodes.Wait(0);
if (waitForSemaphoreWasSuccessful == false && millisecondsTimeout != 0)
{
linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(
cancellationToken, m_ProducersCancellationTokenSource.Token);
waitForSemaphoreWasSuccessful = m_freeNodes.Wait(millisecondsTimeout, linkedTokenSource.Token);
}
}
catch (OperationCanceledException)
{
//if cancellation was via external token, throw an OCE
if (cancellationToken.IsCancellationRequested)
throw new OperationCanceledException(SR.GetString(SR.Common_OperationCanceled), cancellationToken);
//if cancellation was via internal token, this indicates invalid use, hence InvalidOpEx.
//Contract.Assert(m_ProducersCancellationTokenSource.Token.IsCancellationRequested);
throw new InvalidOperationException
(SR.GetString(SR.BlockingCollection_Add_ConcurrentCompleteAdd));
}
finally
{
if (linkedTokenSource != null)
{
linkedTokenSource.Dispose();
}
}
}
if (waitForSemaphoreWasSuccessful)
{
// Update the adders count if the complete adding was not requested, otherwise
// spins until all adders finish then throw IOE
// The idea behind to spin untill all adders finish, is to avoid to return to the caller with IOE while there are still some adders have
// not been finished yet
SpinWait spinner = new SpinWait();
while (true)
{
int observedAdders = m_currentAdders;
if ((observedAdders & COMPLETE_ADDING_ON_MASK) != 0)
{
spinner.Reset();
// CompleteAdding is requested, spin then throw
while (m_currentAdders != COMPLETE_ADDING_ON_MASK) spinner.SpinOnce();
throw new InvalidOperationException(SR.GetString(SR.BlockingCollection_Completed));
}
if (Interlocked.CompareExchange(ref m_currentAdders, observedAdders + 1, observedAdders) == observedAdders)
{
Debug.Assert((observedAdders + 1) <= (~COMPLETE_ADDING_ON_MASK), "The number of concurrent adders thread excceeded the maximum limit.");
break;
}
spinner.SpinOnce();
}
// This outer try/finally to workaround of repeating the decrement adders code 3 times, because we should decrement the adders if:
// 1- m_collection.TryAdd threw an exception
// 2- m_collection.TryAdd succeeded
// 3- m_collection.TryAdd returned false
// so we put the decrement code in the finally block
try
{
//TryAdd is guaranteed to find a place to add the element. Its return value depends
//on the semantics of the underlying store. Some underlying stores will not add an already
//existing item and thus TryAdd returns false indicating that the size of the underlying
//store did not increase.
bool addingSucceeded = false;
try
{
//The token may have been canceled before the collection had space available, so we need a check after the wait has completed.
//This fixes bug #702328, case 2 of 2.
cancellationToken.ThrowIfCancellationRequested();
addingSucceeded = m_collection.TryAdd(item);
}
catch
{
//TryAdd did not result in increasing the size of the underlying store and hence we need
//to increment back the count of the m_freeNodes semaphore.
if (m_freeNodes != null)
{
m_freeNodes.Release();
}
throw;
}
if (addingSucceeded)
{
//After adding an element to the underlying storage, signal to the consumers
//waiting on m_occupiedNodes that there is a new item added ready to be consumed.
m_occupiedNodes.Release();
}
else
{
throw new InvalidOperationException(SR.GetString(SR.BlockingCollection_Add_Failed));
}
}
finally
{
// decrement the adders count
Debug.Assert((m_currentAdders & ~COMPLETE_ADDING_ON_MASK) > 0);
Interlocked.Decrement(ref m_currentAdders);
}
}
return waitForSemaphoreWasSuccessful;
}
/// Takes an item from the
/// The item removed from the collection.
/// The
/// The
/// The underlying collection was modified
/// outside of this
/// A call to
public T Take()
{
T item;
if (!TryTake(out item, Timeout.Infinite, CancellationToken.None))
{
throw new InvalidOperationException(SR.GetString(SR.BlockingCollection_CantTakeWhenDone));
}
return item;
}
/// Takes an item from the
/// The item removed from the collection.
/// If the
/// The
/// The underlying collection was modified
/// outside of this
/// A call to
public T Take(CancellationToken cancellationToken)
{
T item;
if (!TryTake(out item, Timeout.Infinite, cancellationToken))
{
throw new InvalidOperationException(SR.GetString(SR.BlockingCollection_CantTakeWhenDone));
}
return item;
}
///
/// Attempts to remove an item from the
/// The item removed from the collection.
/// true if an item could be removed; otherwise, false.
/// The
/// The underlying collection was modified
/// outside of this
public bool TryTake(out T item)
{
return TryTake(out item, 0, CancellationToken.None);
}
///
/// Attempts to remove an item from the
/// The item removed from the collection.
/// A true if an item could be removed from the collection within
/// the alloted time; otherwise, false.
/// The
/// The underlying collection was modified
/// outside of this
public bool TryTake(out T item, TimeSpan timeout)
{
ValidateTimeout(timeout);
return TryTakeWithNoTimeValidation(out item, (int)timeout.TotalMilliseconds, CancellationToken.None, null);
}
///
/// Attempts to remove an item from the
/// The item removed from the collection.
/// The number of milliseconds to wait, or true if an item could be removed from the collection within
/// the alloted time; otherwise, false.
/// The
/// The underlying collection was modified
/// outside of this
public bool TryTake(out T item, int millisecondsTimeout)
{
ValidateMillisecondsTimeout(millisecondsTimeout);
return TryTakeWithNoTimeValidation(out item, millisecondsTimeout, CancellationToken.None, null);
}
///
/// Attempts to remove an item from the
/// The item removed from the collection.
/// The number of milliseconds to wait, or true if an item could be removed from the collection within
/// the alloted time; otherwise, false.
/// If the
/// The
/// The underlying collection was modified
/// outside of this
public bool TryTake(out T item, int millisecondsTimeout, CancellationToken cancellationToken)
{
ValidateMillisecondsTimeout(millisecondsTimeout);
return TryTakeWithNoTimeValidation(out item, millisecondsTimeout, cancellationToken, null);
}
/// Takes an item from the underlying data store using its IProducerConsumerCollection<T>.Take
/// method. If the collection was empty, this method will wait for, at most, the timeout period (if AddingIsCompleted is false)
/// trying to remove an item. If the timeout period was exhaused before successfully removing an item
/// this method will return false.
/// A
/// The item removed from the collection.
/// The number of milliseconds to wait for the collection to have an item available
/// for removal, or Timeout.Infinite to wait indefinitely.
/// A cancellation token to observe.
/// A combined cancellation token if created, it is only created by GetConsumingEnumerable to avoid creating the linked token
/// multiple times.
/// False if the collection remained empty till the timeout period was exhausted. True otherwise.
/// If the
/// If the collection has been disposed.
private bool TryTakeWithNoTimeValidation(out T item, int millisecondsTimeout, CancellationToken cancellationToken, CancellationTokenSource combinedTokenSource)
{
CheckDisposed();
item = default(T);
if (cancellationToken.IsCancellationRequested)
throw new OperationCanceledException(SR.GetString(SR.Common_OperationCanceled), cancellationToken);
//If the collection is completed then there is no need to wait.
if (IsCompleted)
{
return false;
}
bool waitForSemaphoreWasSuccessful = false;
// set the combined token source to the combinedToken paramater if it is not null (came from GetConsumingEnumerable)
CancellationTokenSource linkedTokenSource = combinedTokenSource;
try
{
waitForSemaphoreWasSuccessful = m_occupiedNodes.Wait(0);
if (waitForSemaphoreWasSuccessful == false && millisecondsTimeout != 0)
{
// create the linked token if it is not created yet
if (combinedTokenSource == null)
linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken,
m_ConsumersCancellationTokenSource.Token);
waitForSemaphoreWasSuccessful = m_occupiedNodes.Wait(millisecondsTimeout, linkedTokenSource.Token);
}
}
//The collection became completed while waiting on the semaphore.
catch (OperationCanceledException)
{
if (cancellationToken.IsCancellationRequested)
throw new OperationCanceledException(SR.GetString(SR.Common_OperationCanceled), cancellationToken);
return false;
}
finally
{
// only dispose the combined token source if we created it here, otherwise the caller (GetConsumingEnumerable) is responsible for disposing it
if (linkedTokenSource != null && combinedTokenSource == null)
{
linkedTokenSource.Dispose();
}
}
if (waitForSemaphoreWasSuccessful)
{
bool removeSucceeded = false;
bool removeFaulted = true;
try
{
//The token may have been canceled before an item arrived, so we need a check after the wait has completed.
//This fixes bug #702328, case 1 of 2.
cancellationToken.ThrowIfCancellationRequested();
//If an item was successfully removed from the underlying collection.
removeSucceeded = m_collection.TryTake(out item);
removeFaulted = false;
if (!removeSucceeded)
{
// Check if the collection is empty which means that the collection was modified outside BlockingCollection
throw new InvalidOperationException
(SR.GetString(SR.BlockingCollection_Take_CollectionModified));
}
}
finally
{
// removeFaulted implies !removeSucceeded, but the reverse is not true.
if (removeSucceeded)
{
if (m_freeNodes != null)
{
Debug.Assert(m_boundedCapacity != NON_BOUNDED);
m_freeNodes.Release();
}
}
else if (removeFaulted)
{
m_occupiedNodes.Release();
}
//Last remover will detect that it has actually removed the last item from the
//collection and that CompleteAdding() was called previously. Thus, it will cancel the semaphores
//so that any thread waiting on them wakes up. Note several threads may call CancelWaitingConsumers
//but this is not a problem.
if (IsCompleted)
{
CancelWaitingConsumers();
}
}
}
return waitForSemaphoreWasSuccessful;
}
///
/// Adds the specified item to any one of the specified
///
/// The array of collections.
/// The item to be added to one of the collections.
/// The index of the collection in the
/// The
/// The
/// At least one of the
/// At least one underlying collection didn't accept the item.
/// The count of
///
/// If a bounded capacity was specified when all of the
///
public static int AddToAny(BlockingCollection[] collections, T item)
{
#if DEBUG
int tryAddAnyReturnValue =
#else
return
#endif
TryAddToAny(collections, item, Timeout.Infinite, new CancellationToken());
#if DEBUG
Debug.Assert((tryAddAnyReturnValue >= 0 && tryAddAnyReturnValue < collections.Length)
, "TryAddToAny() was expected to return an index within the bounds of the collections array.");
return tryAddAnyReturnValue;
#endif
}
///
/// Adds the specified item to any one of the specified
///
/// The array of collections.
/// The item to be added to one of the collections.
/// A cancellation token to observe.
/// The index of the collection in the
/// If the
/// The
/// The
/// At least one of the
/// At least one underlying collection didn't accept the item.
/// The count of
///
/// If a bounded capacity was specified when all of the
///
public static int AddToAny(BlockingCollection[] collections, T item, CancellationToken cancellationToken)
{
#if DEBUG
int tryAddAnyReturnValue =
#else
return
#endif
TryAddToAny(collections, item, Timeout.Infinite, cancellationToken);
#if DEBUG
Debug.Assert((tryAddAnyReturnValue >= 0 && tryAddAnyReturnValue < collections.Length)
, "TryAddToAny() was expected to return an index within the bounds of the collections array.");
return tryAddAnyReturnValue;
#endif
}
///
/// Attempts to add the specified item to any one of the specified
///
/// The array of collections.
/// The item to be added to one of the collections.
/// The index of the collection in the
/// The
/// The
/// At least one of the
/// At least one underlying collection didn't accept the item.
/// The count of
public static int TryAddToAny(BlockingCollection[] collections, T item)
{
return TryAddToAny(collections, item, 0, new CancellationToken());
}
///
/// Attempts to add the specified item to any one of the specified
///
/// The array of collections.
/// The item to be added to one of the collections.
/// A The index of the collection in the
/// The
/// The
/// At least one of the
/// At least one underlying collection didn't accept the item.
/// The count of
public static int TryAddToAny(BlockingCollection[] collections, T item, TimeSpan timeout)
{
ValidateTimeout(timeout);
return TryAddTakeAny(collections, ref item, (int)timeout.TotalMilliseconds, OperationMode.Add, new CancellationToken());
}
///
/// Attempts to add the specified item to any one of the specified
///
/// The array of collections.
/// The item to be added to one of the collections.
/// The number of milliseconds to wait, or The index of the collection in the
/// The
/// The
/// At least one of the
/// At least one underlying collection didn't accept the item.
/// The count of
public static int TryAddToAny(BlockingCollection[] collections, T item, int millisecondsTimeout)
{
ValidateMillisecondsTimeout(millisecondsTimeout);
return TryAddTakeAny(collections, ref item, millisecondsTimeout, OperationMode.Add, new CancellationToken());
}
///
/// Attempts to add the specified item to any one of the specified
///
/// The array of collections.
/// The item to be added to one of the collections.
/// The number of milliseconds to wait, or The index of the collection in the
/// A cancellation token to observe.
/// If the
/// The
/// The
/// At least one of the
/// At least one underlying collection didn't accept the item.
/// The count of
public static int TryAddToAny(BlockingCollection[] collections, T item, int millisecondsTimeout, CancellationToken cancellationToken)
{
ValidateMillisecondsTimeout(millisecondsTimeout);
return TryAddTakeAny(collections, ref item, millisecondsTimeout, OperationMode.Add, cancellationToken);
}
/// Adds/Takes an item to/from anyone of the specified collections.
/// A
/// The collections into which the item can be added.
/// The item to be added or the item removed and returned to the caller.
/// The number of milliseconds to wait for a collection to accept the
/// operation, or -1 to wait indefinitely.
/// Indicates whether this method is called to Add or Take.
/// A cancellation token to observe.
/// The index into collections for the collection which accepted the
/// adding/removal of the item; -1 if the item could not be added/removed.
/// If the
/// If the collections argument is null.
/// If the collections argument is a 0-length array or contains a
/// null element. Also, if atleast one of the collections has been marked complete for adds.
/// If atleast one of the collections has been disposed.
private static int TryAddTakeAny(BlockingCollection[] collections, ref T item, int millisecondsTimeout, OperationMode operationMode, CancellationToken externalCancellationToken)
{
BlockingCollection[] collectionsCopy = ValidateCollectionsArray(collections, operationMode);
#if DEBUG
collections = null; // Ensure we always go through collectionsCopy from now on.
#endif
const int OPERATION_FAILED = -1;
// Copy the wait time to another local variable to update it
int timeout = millisecondsTimeout;
long startTimeTicks = 0;
if (millisecondsTimeout != Timeout.Infinite)
{
startTimeTicks = DateTime.UtcNow.Ticks;
}
// Fast path for adding if there is at least one unbounded collection
if (operationMode == OperationMode.Add)
{
for (int i = 0; i < collectionsCopy.Length; i++)
{
if (collectionsCopy[i].m_freeNodes == null)
{
#if DEBUG
bool result =
#endif
collectionsCopy[i].TryAdd(item);
#if DEBUG
Debug.Assert(result);
#endif
return i;
}
}
}
// Get wait handles and the tokens for all collections
CancellationToken[] collatedCancellationTokens;
List handles = GetHandles(collectionsCopy, operationMode, externalCancellationToken, false, out collatedCancellationTokens);
//Loop until the timeout period is exhausted or the operation was successful.
while (millisecondsTimeout == Timeout.Infinite || timeout >= 0)
{
#region Update Timeout
#endregion
//Wait for any collection to become available.
int index = -1;
CancellationTokenSource linkedTokenSource = null;
try
{
index = WaitHandle_WaitAny(handles, 0, externalCancellationToken, externalCancellationToken);
if (index == WaitHandle.WaitTimeout)
{
linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(collatedCancellationTokens);
index = WaitHandle_WaitAny(handles, timeout, linkedTokenSource.Token,
externalCancellationToken);
}
}
catch (OperationCanceledException ex)
{
// throw OCE if it is external cancellation
if (externalCancellationToken.IsCancellationRequested)
{
Debug.Assert(ex.CancellationToken == externalCancellationToken);
throw;
}
// If the Operation is TryTake, the exception will not be propagated, it will try another collection, and return false if timeout expired
// or all collection completed addition
if (operationMode == OperationMode.Take && millisecondsTimeout != Timeout.Infinite)
{
// Get the new handles and exclude the collections that completed addition
handles = GetHandles(collectionsCopy, operationMode, externalCancellationToken, true, out collatedCancellationTokens);
timeout = UpdateTimeOut(startTimeTicks, millisecondsTimeout);
if (handles.Count == 0 || timeout == 0)
{
return OPERATION_FAILED;
}
continue; //retry for the updated handles
}
// Otherwise if it is Take, Add or Try Add ArgumentException will be thrown
throw new ArgumentException(
SR.GetString(SR.BlockingCollection_CantTakeAnyWhenDone), "collections");
}
finally
{
if (linkedTokenSource != null)
{
linkedTokenSource.Dispose();
}
}
Debug.Assert((index == WaitHandle.WaitTimeout) || (index >= 0 && index < handles.Count));
//If the timeout period was not exhausted and the appropriate operation succeeded.
if (index != WaitHandle.WaitTimeout)
{
if (operationMode == OperationMode.Add)
{
if (collectionsCopy[index].TryAdd(item))
{
return index;
}
}
else if (operationMode == OperationMode.Take)
{
if (collectionsCopy[index].TryTake(out item))
{
return index;
}
}
}
else
{
return OPERATION_FAILED;
}
// Update the timeout
if (millisecondsTimeout > 0)
{
timeout = UpdateTimeOut(startTimeTicks, millisecondsTimeout);
if (timeout <= 0)
{
return OPERATION_FAILED;
}
}
}
return OPERATION_FAILED;
}
///
/// Local static method, used by TryAddTakeAny to get the wait handles for the collection, with exclude option to exclude the Compeleted collections
///
/// The blocking collections
/// Add or Take operation
/// The original CancellationToken
/// True to exclude the compeleted collections
/// Complete list of cancellationTokens to observe
/// The collections wait handles
private static List GetHandles(BlockingCollection[] collections, OperationMode operationMode, CancellationToken externalCancellationToken, bool excludeCompleted, out CancellationToken[] cancellationTokens)
{
Debug.Assert(collections != null);
List handlesList = new List(collections.Length);
List tokensList = new List(collections.Length + 1); // + 1 for the external token
tokensList.Add(externalCancellationToken);
//Read the appropriate WaitHandle based on the operation mode.
if (operationMode == OperationMode.Add)
{
for (int i = 0; i < collections.Length; i++)
{
if (collections[i].m_freeNodes != null)
{
handlesList.Add(collections[i].m_freeNodes.AvailableWaitHandle);
tokensList.Add(collections[i].m_ProducersCancellationTokenSource.Token);
}
}
}
else
{
for (int i = 0; i < collections.Length; i++)
{
if (excludeCompleted && collections[i].IsCompleted)
{
continue;
}
handlesList.Add(collections[i].m_occupiedNodes.AvailableWaitHandle);
tokensList.Add(collections[i].m_ConsumersCancellationTokenSource.Token);
}
}
cancellationTokens = tokensList.ToArray();
return handlesList;
}
///
/// Helper to perform WaitHandle.WaitAny(.., CancellationToken)
/// this should eventually appear on the WaitHandle class.
///
///
///
///
///
/// handles, int millisecondsTimeout, CancellationToken combinedToken, CancellationToken externalToken)
{
WaitHandle[] handlesPlusCancelToken = new WaitHandle[handles.Count + 1]; // +1 for the combinedToken handle
for (int i = 0; i < handles.Count; i++)
handlesPlusCancelToken[i] = handles[i];
handlesPlusCancelToken[handles.Count] = combinedToken.WaitHandle;
int retVal = WaitHandle.WaitAny(handlesPlusCancelToken, millisecondsTimeout, false);
if (combinedToken.IsCancellationRequested)
{
//Add the original token if it is external cancellation
if (externalToken.IsCancellationRequested)
throw new OperationCanceledException(SR.GetString(SR.Common_OperationCanceled), externalToken);
throw new OperationCanceledException(SR.GetString(SR.Common_OperationCanceled));
}
return retVal;
}
///
/// Helper function to measure and update the wait time
///
/// The first time (in Ticks) observed when the wait started
/// The orginal wait timeoutout in milliseconds
/// The new wait time in milliseconds, -1 if the time expired
private static int UpdateTimeOut(long startTimeTicks, int originalWaitMillisecondsTimeout)
{
if (originalWaitMillisecondsTimeout == 0)
{
return 0;
}
// The function must be called in case the time out is not infinite
Debug.Assert(originalWaitMillisecondsTimeout != Timeout.Infinite);
long elapsedMilliseconds = (DateTime.UtcNow.Ticks - startTimeTicks) / TimeSpan.TicksPerMillisecond;
// Check the elapsed milliseconds is greater than max int because this property is long
if (elapsedMilliseconds > int.MaxValue)
{
return 0;
}
// Subtract the elapsed time from the current wait time
int currentWaitTimeout = originalWaitMillisecondsTimeout - (int)elapsedMilliseconds; ;
if (currentWaitTimeout <= 0)
{
return 0;
}
return currentWaitTimeout;
}
///
/// Takes an item from any one of the specified
///
/// The array of collections.
/// The item removed from one of the collections.
/// The index of the collection in the
/// The
/// The
/// At least one of the
/// At least one of the underlying collections was modified
/// outside of its
/// The count of
/// A call to TakeFromAny may block until an item is available to be removed.
public static int TakeFromAny(BlockingCollection[] collections, out T item)
{
#if DEBUG
int tryTakeAnyReturnValue =
#else
return
#endif
TryTakeFromAny(collections, out item, Timeout.Infinite);
#if DEBUG
Debug.Assert((tryTakeAnyReturnValue >= 0 && tryTakeAnyReturnValue < collections.Length)
, "TryTakeFromAny() was expected to return an index within the bounds of the collections array.");
return tryTakeAnyReturnValue;
#endif
}
///
/// Takes an item from any one of the specified
///
/// The array of collections.
/// The item removed from one of the collections.
/// A cancellation token to observe.
/// The index of the collection in the
/// The
/// If the
/// The
/// At least one of the
/// At least one of the underlying collections was modified
/// outside of its
/// The count of
/// A call to TakeFromAny may block until an item is available to be removed.
public static int TakeFromAny(BlockingCollection[] collections, out T item, CancellationToken cancellationToken)
{
#if DEBUG
int tryTakeAnyReturnValue =
#else
return
#endif
TryTakeFromAny(collections, out item, Timeout.Infinite, cancellationToken);
#if DEBUG
Debug.Assert((tryTakeAnyReturnValue >= 0 && tryTakeAnyReturnValue < collections.Length)
, "TryTakeFromAny() was expected to return an index within the bounds of the collections array.");
return tryTakeAnyReturnValue;
#endif
}
///
/// Attempts to remove an item from any one of the specified
///
/// The array of collections.
/// The item removed from one of the collections.
/// The index of the collection in the
/// The
/// The
/// At least one of the
/// At least one of the underlying collections was modified
/// outside of its
/// The count of
/// A call to TryTakeFromAny may block until an item is available to be removed.
public static int TryTakeFromAny(BlockingCollection[] collections, out T item)
{
return TryTakeFromAny(collections, out item, 0);
}
///
/// Attempts to remove an item from any one of the specified
///
/// The array of collections.
/// The item removed from one of the collections.
/// A The index of the collection in the
/// The
/// The
/// At least one of the
/// At least one of the underlying collections was modified
/// outside of its
/// The count of
/// A call to TryTakeFromAny may block until an item is available to be removed.
public static int TryTakeFromAny(BlockingCollection[] collections, out T item, TimeSpan timeout)
{
ValidateTimeout(timeout);
T tempItem = default(T);
int returnValue = TryAddTakeAny(collections, ref tempItem, (int)timeout.TotalMilliseconds, OperationMode.Take, new CancellationToken());
item = tempItem;
return returnValue;
}
///
/// Attempts to remove an item from any one of the specified
///
/// The array of collections.
/// The item removed from one of the collections.
/// The number of milliseconds to wait, or The index of the collection in the
/// The
/// The
/// At least one of the
/// At least one of the underlying collections was modified
/// outside of its
/// The count of
/// A call to TryTakeFromAny may block until an item is available to be removed.
public static int TryTakeFromAny(BlockingCollection[] collections, out T item, int millisecondsTimeout)
{
ValidateMillisecondsTimeout(millisecondsTimeout);
T tempItem = default(T);
int returnValue = TryAddTakeAny(collections, ref tempItem, millisecondsTimeout, OperationMode.Take, new CancellationToken());
item = tempItem;
return returnValue;
}
///
/// Attempts to remove an item from any one of the specified
///
/// The array of collections.
/// The item removed from one of the collections.
/// The number of milliseconds to wait, or The index of the collection in the
/// If the
/// The
/// The
/// At least one of the
/// At least one of the underlying collections was modified
/// outside of its
/// The count of
/// A call to TryTakeFromAny may block until an item is available to be removed.
public static int TryTakeFromAny(BlockingCollection[] collections, out T item, int millisecondsTimeout, CancellationToken cancellationToken)
{
ValidateMillisecondsTimeout(millisecondsTimeout);
T tempItem = default(T);
int returnValue = TryAddTakeAny(collections, ref tempItem, millisecondsTimeout, OperationMode.Take, cancellationToken);
item = tempItem;
return returnValue;
}
///
/// Marks the
///
/// After a collection has been marked as complete for adding, adding to the collection is not permitted
/// and attempts to remove from the collection will not wait when the collection is empty.
///
/// The
public void CompleteAdding()
{
CheckDisposed();
if (IsAddingCompleted)
return;
SpinWait spinner = new SpinWait();
while (true)
{
int observedAdders = m_currentAdders;
if ((observedAdders & COMPLETE_ADDING_ON_MASK) != 0)
{
spinner.Reset();
// If there is another COmpleteAdding in progress waiting the current adders, then spin until it finishes
while (m_currentAdders != COMPLETE_ADDING_ON_MASK) spinner.SpinOnce();
return;
}
if (Interlocked.CompareExchange(ref m_currentAdders, observedAdders | COMPLETE_ADDING_ON_MASK, observedAdders) == observedAdders)
{
spinner.Reset();
while (m_currentAdders != COMPLETE_ADDING_ON_MASK) spinner.SpinOnce();
if (Count == 0)
{
CancelWaitingConsumers();
}
// We should always wake waiting producers, and have them throw exceptions as
// Add&CompleteAdding should not be used concurrently.
CancelWaitingProducers();
return;
}
spinner.SpinOnce();
}
}
/// Cancels the semaphores.
private void CancelWaitingConsumers()
{
m_ConsumersCancellationTokenSource.Cancel();
}
private void CancelWaitingProducers()
{
m_ProducersCancellationTokenSource.Cancel();
}
///
/// Releases resources used by the
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
///
/// Releases resources used by the
/// Whether being disposed explicitly (true) or due to a finalizer (false).
protected virtual void Dispose(bool disposing)
{
if (!m_isDisposed)
{
if (m_freeNodes != null)
{
m_freeNodes.Dispose();
}
m_occupiedNodes.Dispose();
m_isDisposed = true;
}
}
/// Copies the items from the
/// An array containing copies of the elements of the collection.
/// The
///
/// The copied elements are not removed from the collection.
///
public T[] ToArray()
{
CheckDisposed();
return m_collection.ToArray();
}
/// Copies all of the items in the
/// The one-dimensional array that is the destination of the elements copied from
/// the The
/// The
/// The
/// The
public void CopyTo(T[] array, int index)
{
((ICollection)this).CopyTo(array, index);
}
/// Copies all of the items in the
/// The one-dimensional array that is the destination of the elements copied from
/// the The
/// The
/// The
/// The
void ICollection.CopyTo(Array array, int index)
{
CheckDisposed();
//We don't call m_collection.CopyTo() directly because we rely on Array.Copy method to customize
//all array exceptions.
T[] collectionSnapShot = m_collection.ToArray();
try
{
Array.Copy(collectionSnapShot, 0, array, index, collectionSnapShot.Length);
}
catch (ArgumentNullException)
{
throw new ArgumentNullException("array");
}
catch (ArgumentOutOfRangeException)
{
throw new ArgumentOutOfRangeException("index", index, SR.GetString(SR.BlockingCollection_CopyTo_NonNegative));
}
catch (ArgumentException)
{
throw new ArgumentException(SR.GetString(SR.BlockingCollection_CopyTo_TooManyElems), "index");
}
catch (RankException)
{
throw new ArgumentException(SR.GetString(SR.BlockingCollection_CopyTo_MultiDim), "array");
}
catch (InvalidCastException)
{
throw new ArgumentException(SR.GetString(SR.BlockingCollection_CopyTo_IncorrectType), "array");
}
catch (ArrayTypeMismatchException)
{
throw new ArgumentException(SR.GetString(SR.BlockingCollection_CopyTo_IncorrectType), "array");
}
}
/// Provides a consuming
/// An
/// The
public IEnumerable GetConsumingEnumerable()
{
return GetConsumingEnumerable(CancellationToken.None);
}
/// Provides a consuming
/// A cancellation token to observe.
/// An
/// The
/// If the
public IEnumerable GetConsumingEnumerable(CancellationToken cancellationToken)
{
CancellationTokenSource linkedTokenSource = null;
try
{
linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, m_ConsumersCancellationTokenSource.Token);
while (!IsCompleted)
{
T item;
if (TryTakeWithNoTimeValidation(out item, Timeout.Infinite, cancellationToken, linkedTokenSource))
{
yield return item;
}
}
}
finally
{
if (linkedTokenSource != null)
{
linkedTokenSource.Dispose();
}
}
}
/// Provides an
/// An
/// The
IEnumerator IEnumerable.GetEnumerator()
{
CheckDisposed();
return m_collection.GetEnumerator();
}
/// Provides an
/// An
/// The
IEnumerator IEnumerable.GetEnumerator()
{
return ((IEnumerable)this).GetEnumerator();
}
/// Centralizes the logic for validating the BlockingCollections array passed to TryAddToAny()
/// and TryTakeFromAny().
/// The collections to/from which an item should be added/removed.
/// Indicates whether this method is called to Add or Take.
/// A copy of the collections array that acts as a defense to prevent an “outsider” from changing
/// elements of the array after we have done the validation on them.
/// If the collections argument is null.
/// If the collections argument is a 0-length array or contains a
/// null element. Also, if atleast one of the collections has been marked complete for adds.
/// If atleast one of the collections has been disposed.
private static BlockingCollection[] ValidateCollectionsArray(BlockingCollection[] collections, OperationMode operationMode)
{
if (collections == null)
{
throw new ArgumentNullException("collections");
}
else if (collections.Length < 1)
{
throw new ArgumentException(
SR.GetString(SR.BlockingCollection_ValidateCollectionsArray_ZeroSize), "collections");
}
else if ((Thread.CurrentThread.GetApartmentState() == ApartmentState.MTA && collections.Length > 63)
|| (Thread.CurrentThread.GetApartmentState() == ApartmentState.STA && collections.Length > 62))
//The number of WaitHandles must be <= 64 for MTA, and <=63 for STA, and we reserve one for CancellationToken
{
throw new ArgumentOutOfRangeException(
"collections", SR.GetString(SR.BlockingCollection_ValidateCollectionsArray_LargeSize));
}
BlockingCollection[] collectionsCopy = new BlockingCollection[collections.Length];
for (int i = 0; i < collectionsCopy.Length; ++i)
{
collectionsCopy[i] = collections[i];
if (collectionsCopy[i] == null)
{
throw new ArgumentException(
SR.GetString(SR.BlockingCollection_ValidateCollectionsArray_NullElems), "collections");
}
if (collectionsCopy[i].m_isDisposed)
throw new ObjectDisposedException(
"collections", SR.GetString(SR.BlockingCollection_ValidateCollectionsArray_DispElems));
if (operationMode == OperationMode.Add && collectionsCopy[i].IsAddingCompleted)
{
throw new ArgumentException(
SR.GetString(SR.BlockingCollection_CantTakeAnyWhenDone), "collections");
}
}
return collectionsCopy;
}
// ---------
// Private Helpers.
/// Centeralizes the logic of validating the timeout input argument.
/// The TimeSpan to wait for to successfully complete an operation on the collection.
/// If the number of millseconds represented by the timeout
/// TimeSpan is less than 0 or is larger than Int32.MaxValue and not Timeout.Infinite
private static void ValidateTimeout(TimeSpan timeout)
{
long totalMilliseconds = (long)timeout.TotalMilliseconds;
if ((totalMilliseconds < 0 || totalMilliseconds > Int32.MaxValue) && (totalMilliseconds != Timeout.Infinite))
{
throw new ArgumentOutOfRangeException("timeout", timeout,
String.Format(CultureInfo.InvariantCulture, SR.GetString(SR.BlockingCollection_TimeoutInvalid), Int32.MaxValue));
}
}
/// Centralizes the logic of validating the millisecondsTimeout input argument.
/// The number of milliseconds to wait for to successfully complete an
/// operation on the collection.
/// If the number of millseconds is less than 0 and not
/// equal to Timeout.Infinite.
private static void ValidateMillisecondsTimeout(int millisecondsTimeout)
{
if ((millisecondsTimeout < 0) && (millisecondsTimeout != Timeout.Infinite))
{
throw new ArgumentOutOfRangeException("millisecondsTimeout", millisecondsTimeout,
String.Format(CultureInfo.InvariantCulture, SR.GetString(SR.BlockingCollection_TimeoutInvalid), Int32.MaxValue));
}
}
/// Throws a System.ObjectDisposedException if the collection was disposed
/// If the collection has been disposed.
private void CheckDisposed()
{
if (m_isDisposed)
{
throw new ObjectDisposedException("BlockingCollection", SR.GetString(SR.BlockingCollection_Disposed));
}
}
}
/// A debugger view of the blocking collection that makes it simple to browse the
/// collection's contents at a point in time.
/// The type of element that the BlockingCollection will hold.
internal sealed class SystemThreadingCollections_BlockingCollectionDebugView
{
private BlockingCollection m_blockingCollection; // The collection being viewed.
/// Constructs a new debugger view object for the provided blocking collection object.
/// A blocking collection to browse in the debugger.
public SystemThreadingCollections_BlockingCollectionDebugView(BlockingCollection collection)
{
if (collection == null)
{
throw new ArgumentNullException("collection");
}
m_blockingCollection = collection;
}
/// Returns a snapshot of the underlying collection's elements.
[DebuggerBrowsable(DebuggerBrowsableState.RootHidden)]
public T[] Items
{
get
{
return m_blockingCollection.ToArray();
}
}
}
}
#pragma warning restore 0420
// File provided for Reference Use Only by Microsoft Corporation (c) 2007.
Link Menu
This book is available now!
Buy at Amazon US or
Buy at Amazon UK
- Identity.cs
- ConfigXmlText.cs
- SqlUtils.cs
- PKCS1MaskGenerationMethod.cs
- DBBindings.cs
- MaskedTextBox.cs
- BackoffTimeoutHelper.cs
- Vector3DAnimationBase.cs
- DataGridViewRowCancelEventArgs.cs
- CryptoKeySecurity.cs
- XmlFormatExtensionAttribute.cs
- PageEventArgs.cs
- MemberInfoSerializationHolder.cs
- QueryResponse.cs
- DataGridRow.cs
- ActivityContext.cs
- XmlStreamStore.cs
- Empty.cs
- PointConverter.cs
- Latin1Encoding.cs
- DataGridPageChangedEventArgs.cs
- TimerTable.cs
- ObjectDataSourceDesigner.cs
- ColorConvertedBitmap.cs
- StreamMarshaler.cs
- ResourceBinder.cs
- PrintPageEvent.cs
- CompositeDataBoundControl.cs
- PersistenceTypeAttribute.cs
- URLIdentityPermission.cs
- PasswordPropertyTextAttribute.cs
- SafePipeHandle.cs
- CriticalExceptions.cs
- PagedControl.cs
- MemoryMappedViewAccessor.cs
- HealthMonitoringSection.cs
- ConfigurationUtility.cs
- Error.cs
- XdrBuilder.cs
- CodeConditionStatement.cs
- CodeTypeParameter.cs
- HttpHandlerActionCollection.cs
- DataViewSettingCollection.cs
- HtmlMeta.cs
- MarshalByValueComponent.cs
- ExeContext.cs
- RegexReplacement.cs
- DataObjectAttribute.cs
- EntityObject.cs
- WindowsTab.cs
- TailCallAnalyzer.cs
- TTSEngineTypes.cs
- XhtmlBasicCommandAdapter.cs
- RectKeyFrameCollection.cs
- GetParentChain.cs
- FrameworkContextData.cs
- SizeAnimation.cs
- KeyGestureConverter.cs
- XPathDocument.cs
- ImpersonateTokenRef.cs
- XmlSchemaAttributeGroupRef.cs
- NamespaceDisplay.xaml.cs
- ThreadPool.cs
- NamespaceQuery.cs
- CookielessHelper.cs
- IndentedWriter.cs
- AudioLevelUpdatedEventArgs.cs
- TimeSpanSecondsConverter.cs
- WebBrowserContainer.cs
- SequenceFullException.cs
- Triplet.cs
- BinaryCommonClasses.cs
- SQLDecimal.cs
- ComponentChangingEvent.cs
- RelatedEnd.cs
- NetworkInformationException.cs
- SemanticKeyElement.cs
- SystemDiagnosticsSection.cs
- SingleStorage.cs
- DataGridComboBoxColumn.cs
- TabPanel.cs
- RunClient.cs
- LocalizationComments.cs
- XmlnsCompatibleWithAttribute.cs
- PersistChildrenAttribute.cs
- BamlLocalizerErrorNotifyEventArgs.cs
- CompleteWizardStep.cs
- FastEncoder.cs
- PagesChangedEventArgs.cs
- ApplicationCommands.cs
- ToolTipService.cs
- HtmlControlPersistable.cs
- OleStrCAMarshaler.cs
- Effect.cs
- TextAnchor.cs
- PathFigureCollectionConverter.cs
- Transactions.cs
- MDIControlStrip.cs
- Function.cs
- GraphicsContext.cs