AsyncCollections: the story of one bike

False Trace: Nito.AsyncEx

Asyncquueue

	public interface IAsyncCollection: IEnumerable
	{
		int Count { get; }
		void Add( T item );
		Task TakeAsync();
	}

• Awaiter obviously needs to be stored somewhere. The queue begs, more specifically, ConcurrentQueue.
• If a call to TakeAsync () occurs, we have a new awaiter, drop it into the awaiter queue.
• If a call to Add () occurs, we have a new element with which you can instantly take one of the awaiter and complete it.

• Items need to be stored somewhere. Where? In ConcurrentQueue, where else.
• If a call to Add () occurs, a new element appears, drop it into the element queue.
• If a call to TakeAsync () occurs, a new awaiter appears, which can be terminated instantly by taking the top item from the queue.

• If Add () is called, a new element appears, try to take awaiter from the queue for it, it is empty there, try to add it to the element queue ...
• At this very moment, a call to TakeAsync () occurs, a new awaiter appears, we try to take an element from the queue for it, it is empty there for now, try to add it to the awaiter queue ...
• Oops We broke everything: awaiter and the element are sitting in different lines and waiting for each other. What to do?

• understand what condition we are in (1 or 2)
• at the same time report that we started our operation, whether it was adding awaiter or adding an element
• depending on the state, either add an awaiter / element to the queue or turn around until the / awaiter element is added to the opposite queue, which we will immediately pick up

	public class AsyncQueue: IAsyncCollection
	{
		private ConcurrentQueue _itemQueue = new ConcurrentQueue();
		private ConcurrentQueue> _awaiterQueue = new ConcurrentQueue>();
		//	_queueBalance < 0 means there are free awaiters and not enough items.
		//	_queueBalance > 0 means the opposite is true.
		private long _queueBalance = 0;
		public void Add( T item )
		{
			long balanceAfterCurrentItem = Interlocked.Increment( ref _queueBalance );
			if ( balanceAfterCurrentItem > 0 )
			{
				//	Items are dominating, so we can safely add a new item to the queue.
				_itemQueue.Enqueue( item );
			}
			else
			{
				//	There's at least one awaiter available or being added as we're speaking, so we're giving the item to it.
				TaskCompletionSource awaiter;
				SpinWait spin = new SpinWait();
				while ( !_awaiterQueue.TryDequeue( out awaiter ) )
					spin.SpinOnce();
				awaiter.SetResult( item );
			}
		}
		public Task TakeAsync()
		{
			long balanceAfterCurrentAwaiter = Interlocked.Decrement( ref _queueBalance );
			if ( balanceAfterCurrentAwaiter < 0 )
			{
				//	Awaiters are dominating, so we can safely add a new awaiter to the queue.
				var taskSource = new TaskCompletionSource();
				_awaiterQueue.Enqueue( taskSource );
				return taskSource.Task;
			}
			else
			{
				//	There's at least one item available or being added, so we're returning it directly.
				T item;
				SpinWait spin = new SpinWait();
				while ( !_itemQueue.TryTake( out item ) )
					spin.SpinOnce();
				return Task.FromResult( item );
			}
		}
	}

Useful (and not so) goodies

		public Task TakeAsync( CancellationToken cancellationToken )
		{
			//	...
			if ( balanceAfterCurrentAwaiter < 0 )
			{
				var taskSource = new TaskCompletionSource();
				_awaiterQueue.Enqueue( taskSource );
				cancellationToken.Register(
					state =>
					{
						TaskCompletionSource awaiter = state as TaskCompletionSource;
						awaiter.TrySetCanceled();
					},
					taskSource,
					useSynchronizationContext : false );
				return taskSource.Task;
			}
			else
			{
				//	...
			}
		}

		private bool TryAdd( TItem item )
		{
			long balanceAfterCurrentItem = Interlocked.Increment( ref _queueBalance );
			if ( balanceAfterCurrentItem > 0 )
			{
				_itemQueue.Enqueue( item );
				return true;
			}
			else
			{
				TaskCompletionSource awaiter;
				SpinWait spin = new SpinWait();
				while ( !_awaiterQueue.TryDequeue( out awaiter ) )
					spin.SpinOnce();
				//	Returns false if the cancellation occurred earlier.
				return awaiter.TrySetResult( item );
			}
		}
		public void Add( TItem item )
		{
			while ( !TryAdd( item ) ) ;
		}

	public interface IAsyncCollection: IEnumerable
	{
		int Count { get; }
		void Add( T item );
		Task TakeAsync( CancellationToken cancellationToken );
	}
	public static class AsyncCollectionExtensions
	{
		public static Task TakeAsync( this IAsyncCollection collection )
		{
			return collection.TakeAsync( CancellationToken.None );
		}
	}

	public class AsyncCollection: IAsyncCollection
		where TItemQueue: IProducerConsumerCollection, new()
	{
		private TItemQueue _itemQueue = new TItemQueue();
		private ConcurrentQueue> _awaiterQueue = new ConcurrentQueue>();
		//	...
	}
	public class AsyncQueue: AsyncCollection>
	{
	}
	public class AsyncStack: AsyncCollection>
	{
	}

Benchmarks!

	class AsyncQueueBenchmark
	{
		private const int _consumerThreadCount = 3;
		private const int _producerThreadCount = 3;
		private const int _itemsAddedPerThread = 10000;
		private const int _itemsAddedTotal = _producerThreadCount * _itemsAddedPerThread;
		private IAsyncCollection _currentQueue;
		private CancellationTokenSource _cancelSource;
		private int _itemsTaken;
		// Выполнение этого метода будет измеряться
		private void DdosCurrentQueue()
		{
			_consumerTasks = Enumerable.Range( 0, _consumerThreadCount )
				.Select( _ => Task.Run( () => RunConsumerAsync() ) )
				.ToArray();
			_producerTasks = Enumerable.Range( 0, _producerThreadCount )
				.Select( _ => Task.Run( () => RunProducer() ) )
				.ToArray();
			Task.WaitAll( _producerTasks );
			Task.WaitAll( _consumerTasks );
		}
		private async Task RunConsumerAsync()
		{
			try
			{
				CancellationToken cancelToken = _cancelSource.Token;
				while ( _itemsTaken < _itemsAddedTotal && !cancelToken.IsCancellationRequested )
				{
					int item = await _currentQueue.TakeAsync( cancelToken );
					int itemsTakenLocal = Interlocked.Increment( ref _itemsTaken );
					if ( itemsTakenLocal >= _itemsAddedTotal )
					{
						_cancelSource.Cancel();
						break;
					}
				}
			}
			catch ( OperationCanceledException )
			{
			}
		}
		private void RunProducer()
		{
			for ( int i = 0; i < _itemsAddedPerThread; i++ )
			{
				int item = 42;
				_currentQueue.Add( item );
			}
		}

	class NitoAsyncCollectionAdapter: IAsyncCollection
	{
		private Nito.AsyncEx.AsyncCollection _collection;
		public NitoAsyncCollectionAdapter()
		{
			_collection = new Nito.AsyncEx.AsyncCollection();
		}
		#region IAsyncCollection Members
		public void Add( T item )
		{
			_collection.Add( item );
		}
		public Task TakeAsync( System.Threading.CancellationToken cancellationToken )
		{
			return _collection.TakeAsync( cancellationToken );
		}
		#endregion
	}

	class BlockingCollectionAdapter: IAsyncCollection
	{
		private BlockingCollection _collection;
		public BlockingCollectionAdapter()
		{
			_collection = new BlockingCollection();
		}
		#region IAsyncCollection Members
		public void Add( T item )
		{
			_collection.Add( item );
		}
		public Task TakeAsync( System.Threading.CancellationToken cancellationToken )
		{
			T item = _collection.Take( cancellationToken );
			return Task.FromResult( item );
		}
		#endregion
	}

HellBrick.AsyncCollections.AsyncQueue: 1ms | Stats: MedianTicks = 3368, MedianMs = 1, Error = 06.34%
Nito.AsyncEx.AsyncCollection: 12ms | Stats: MedianTicks = 40503, MedianMs = 12, Error = 31.36%
System.Concurrent.BlockingCollection: 2ms | Stats: MedianTicks = 7222, MedianMs = 2, Error = 38.82%

AsyncBatchQueue

	public class AsyncBatchQueue
	{
		private int _batchSize;
		private Batch _currentBatch;
		private AsyncQueue> _batchQueue = new AsyncQueue>();
		public AsyncBatchQueue( int batchSize )
		{
			_batchSize = batchSize;
			_currentBatch = new Batch( this );
		}
		public void Add( T item )
		{
			SpinWait spin = new SpinWait();
			while ( !_currentBatch.TryAdd( item ) )
				spin.SpinOnce();
		}
		public Task> TakeAsync( CancellationToken cancellationToken )
		{
			return _batchQueue.TakeAsync( cancellationToken );
		}
		private class Batch: IReadOnlyList
		{
			private AsyncBatchQueue _queue;
			//	?
			public Batch( AsyncBatchQueue queue )
			{
				_queue = queue;
			}
			public bool TryAdd( T item )
			{
				//	?
			}
		}
	}

• We store the elements in a regular array, since we know the size in advance
• Problems with concurrency are solved using the Interlocked.Increment field, where the index of the last inserted item is stored
• If the thread has captured the last slot of the array, then it [the thread, not the slot] is responsible for executing the flush of the current batch
• If the stream has captured a slot that goes beyond the boundaries of the array, then we are out of luck: this batch is already full and the stream needs to spin in anticipation of a new

		private class Batch: IReadOnlyList
		{
			private AsyncBatchQueue _queue;
			private T[] _items;
			private int _lastReservationIndex = -1;
			private int _count = -1;
			public Batch( AsyncBatchQueue queue )
			{
				_queue = queue;
				_items = new T[ _queue._batchSize ];
			}
			public bool TryAdd( T item )
			{
				int index = Interlocked.Increment( ref _lastReservationIndex );
				//	The following is true if someone has beaten us to the last slot and we have to wait until the next batch comes along.
				if ( index >= _queue._batchSize )
					return false;
				//	The following is true if we've taken the last slot, which means we're obligated to flush the current batch and create a new one.
				if ( index == _queue._batchSize - 1 )
					FlushInternal( _queue._batchSize );
				_items[ index ] = item;
				return true;
			}
			private void FlushInternal( int count )
			{
				_count = count;
				_queue._currentBatch = new Batch( _queue );
				_queue._batchQueue.Add( this );
			}
		}

		private class Batch: IReadOnlyList
		{
			//	...
			private bool[] _finalizationFlags;
			public Batch( AsyncBatchQueue queue )
			{
				//	...
				_finalizationFlags = new bool[ _queue._batchSize ];
			}
			public bool TryAdd( T item )
			{
				//	...
				_items[ index ] = item;
				_finalizationFlags[ index ] = true;
				return true;
			}
			public T this[ int index ]
			{
				get
				{
					if ( index >= _count )
						throw new IndexOutOfRangeException();
					return GetItemWithoutValidation( index );
				}
			}
			private T GetItemWithoutValidation( int index )
			{
				SpinWait spin = new SpinWait();
				while ( !_finalizationFlags[ index ] )
					spin.SpinOnce();
				return _items[ index ];
			}
			//	... остальные методы реализуется через GetItemWithoutValidation
		}

	public class AsyncBatchQueue
	{
		//	...
		private volatile Batch _currentBatch;
		//	...
	}

	private void FlushInternal( int count )
	{
		_count = count;
		_queue._currentBatch = new Batch( _queue );
		//	The full fence ensures that the current batch will never be added to the queue before _count is set.
		Thread.MemoryBarrier();
		_queue._batchQueue.Add( this );
	}

	public bool TryAdd( T item )
	{
		//	...
		//	The full fence prevents setting finalization flag before the actual item value is written.
		_items[ index ] = item;
		Thread.MemoryBarrier();
		_finalizationFlags[ index ] = true;
		return true;
	}

	private T GetItemWithoutValidation( int index )
	{
		SpinWait spin = new SpinWait();
		while ( !_finalizationFlags[ index ] )
			spin.SpinOnce();
		//	The full fence prevents reading item value before finalization flag is set.
		Thread.MemoryBarrier();
		return _items[ index ];
	}

	private T GetItemWithoutValidation( int index )
	{
		SpinWait spin = new SpinWait();
		while ( !_finalizationFlags[ index ] )
		{
			spin.SpinOnce();
			//	The full fence prevents caching any part of _finalizationFlags[ index ] expression.
			Thread.MemoryBarrier();
		}
		//	...
	}

	internal volatile VolatileBool[] m_state;
	struct VolatileBool
	{
		public VolatileBool(bool value)
		{
			m_value = value;
		}
		public volatile bool m_value;
	}

Now get the timer back.

1. _lastReservationIndex <0. There is nothing to flush.
2. _lastReservationIndex> = _queue._batchSize. FlushInternal () will execute the thread that grabs the last slot, nothing needs to be done.
3. _lastReservationIndex is valid and we managed to atomically set it to _queue._batchSize. We know the actual number of elements in the array, you can do FlushInternal ().
4. Between reading the past value of _lastReservationIndex and writing there the new value, another thread crawled in and grabbed the last element. In fact, the situation repeats option number 2: we do nothing.
5. The same as in No. 4, but the batch is not filled. Spin, try again.

	public class AsyncBatchQueue: IEnumerable>
	{
		//	...
		public void Flush()
		{
			SpinWait spin = new SpinWait();
			while ( !_currentBatch.TryFlush() )
				spin.SpinOnce();
		}
		//	...
		private class Batch: IReadOnlyList
		{
			//	...			
			public bool TryFlush()
			{
				int expectedPreviousReservation = Volatile.Read( ref _lastReservationIndex );
				//	We don't flush if the batch doesn't have any items or if another thread is about to flush it
				//	However, we report success to avoid unnecessary spinning.
				if ( expectedPreviousReservation < 0 || expectedPreviousReservation >= _queue._batchSize )
					return true;
				int previousReservation = Interlocked.CompareExchange( ref _lastReservationIndex, _queue._batchSize, expectedPreviousReservation );
				//	Flush reservation has succeeded.
				if ( expectedPreviousReservation == previousReservation )
				{
					FlushInternal( previousReservation + 1 );
					return true;
				}
				//	The following is true if someone has completed the batch by the time we tried to flush it.
				//	Therefore the batch will be flushed anyway even if we don't do anything.
				//	The opposite means someone has slipped in an update and we have to spin.
				return previousReservation >= _queue._batchSize;
			}
			//	...
		}
	}

What's next?

try
{
}
finally
{
	//	Insert Thread.Abort()-safe code here
}

PS