<commit_before>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using BenchmarkDotNet;
using HellBrick.Collections;

namespace HellBrick.AsyncCollections.Benchmark
{
	internal class AsyncQueueBenchmark
	{
		private const int _consumerThreadCount = 3;
		private const int _producerThreadCount = 3;
		private const int _itemsAddedPerThread = 10000;
		private const int _itemsAddedTotal = _producerThreadCount * _itemsAddedPerThread;

		private readonly BenchmarkCompetition _competition;

		private IAsyncCollection<int> _currentQueue;
		private CancellationTokenSource _cancelSource;
		private int _itemsTaken;

		public AsyncQueueBenchmark()
		{
			_competition = new BenchmarkCompetition();
			AddTask( "HellBrick.AsyncCollections.AsyncQueue", () => new AsyncQueue<int>() );
			AddTask( "Nito.AsyncEx.AsyncCollection", () => new NitoAsyncCollectionAdapter<int>() );
			AddTask( "System.Concurrent.BlockingCollection", () => new BlockingCollectionAdapter<int>() );
			AddTask( "System.Threading.Tasks.Dataflow.BufferBlock", () => new TplDataflowAdapter<int>() );
		}

		private void AddTask( string name, Func<IAsyncCollection<int>> factoryMethod )
		{
			_competition.AddTask(
				name,
				initialize : () => Initialize( factoryMethod ),
				clean : CleanUp,
				action : () => DdosCurrentQueue() );
		}

		private void Initialize( Func<IAsyncCollection<int>> factoryMethod )
		{
			_currentQueue = factoryMethod();
			_itemsTaken = 0;
			_cancelSource = new CancellationTokenSource();
		}

		private void CleanUp()
		{
			_currentQueue = null;
			_cancelSource = null;
		}

		private void DdosCurrentQueue()
		{
			Task[] consumerTasks = Enumerable.Range( 0, _consumerThreadCount )
				.Select( _ => Task.Run( () => RunConsumerAsync() ) )
				.ToArray();

			Task[] 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 ).ConfigureAwait( false );
					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 );
			}
		}<commit_msg>Simplified consumer loop (we're breaking out when TaskCanceledException is thrown anyway)<commit_after>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using BenchmarkDotNet;
using HellBrick.Collections;

namespace HellBrick.AsyncCollections.Benchmark
{
	internal class AsyncQueueBenchmark
	{
		private const int _consumerThreadCount = 3;
		private const int _producerThreadCount = 3;
		private const int _itemsAddedPerThread = 10000;
		private const int _itemsAddedTotal = _producerThreadCount * _itemsAddedPerThread;

		private readonly BenchmarkCompetition _competition;

		private IAsyncCollection<int> _currentQueue;
		private CancellationTokenSource _cancelSource;
		private int _itemsTaken;

		public AsyncQueueBenchmark()
		{
			_competition = new BenchmarkCompetition();
			AddTask( "HellBrick.AsyncCollections.AsyncQueue", () => new AsyncQueue<int>() );
			AddTask( "Nito.AsyncEx.AsyncCollection", () => new NitoAsyncCollectionAdapter<int>() );
			AddTask( "System.Concurrent.BlockingCollection", () => new BlockingCollectionAdapter<int>() );
			AddTask( "System.Threading.Tasks.Dataflow.BufferBlock", () => new TplDataflowAdapter<int>() );
		}

		private void AddTask( string name, Func<IAsyncCollection<int>> factoryMethod )
		{
			_competition.AddTask(
				name,
				initialize : () => Initialize( factoryMethod ),
				clean : CleanUp,
				action : () => DdosCurrentQueue() );
		}

		private void Initialize( Func<IAsyncCollection<int>> factoryMethod )
		{
			_currentQueue = factoryMethod();
			_itemsTaken = 0;
			_cancelSource = new CancellationTokenSource();
		}

		private void CleanUp()
		{
			_currentQueue = null;
			_cancelSource = null;
		}

		private void DdosCurrentQueue()
		{
			Task[] consumerTasks = Enumerable.Range( 0, _consumerThreadCount )
				.Select( _ => Task.Run( () => RunConsumerAsync() ) )
				.ToArray();

			Task[] 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 ( true )
				{
					int item = await _currentQueue.TakeAsync( cancelToken ).ConfigureAwait( false );
					int itemsTakenLocal = Interlocked.Increment( ref _itemsTaken );

					if ( itemsTakenLocal >= _itemsAddedTotal )
						_cancelSource.Cancel();
				}
			}
			catch ( OperationCanceledException )
			{
			}
		}

		private void RunProducer()
		{
			for ( int i = 0; i < _itemsAddedPerThread; i++ )
			{
				int item = 42;
				_currentQueue.Add( item );
			}
		}
{'fraction_non_alphanumeric': 0.10759148006553795, 'fraction_numerical': 0.004733296923357, 'mean_word_length': 3.99909008189263, 'pattern_counts': {'":': 0, '<': 24, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '128202', 'n_tokens_mistral': 1990, 'n_tokens_neox': 1648, 'n_words': 371}