Add batching with IMemoryOwner<T> and related tests (#49)

* Add batching with IMemoryOwner<T> and related tests

Introduced a new feature for batching results as `IMemoryOwner<T>` to improve memory management. Updated `.editorconfig` for better code style. Modified `SpecialTests.cs` to correct the expected count. Added documentation and comments in `Extensions.Batch.cs`. Updated `BatchingChannelReader.cs` with new methods and classes for batching. Added comprehensive unit tests in `BatchTests.Memory.cs` and `BatchTests.Queue.cs` to ensure the new batching methods work correctly under various conditions.

* Add benchmarks for batching methods and update batch creation

Added a new project `Open.ChannelExtensions.Benchmarks` to the
solution, including a `BatchDrain` class to measure performance
of different batching methods. Updated `Batch` extension methods
to include an optional `batchFactory` parameter. Modified
`BatchingChannelReader` to use `batchFactory` for batch creation.
Added benchmark tests in `BatchDrain.cs` for list-based,
queue-based, and memory-pooled batching methods. Updated solution
file to include the new benchmark project. Added `Program.cs` to
run benchmarks using `BenchmarkDotNet`.

* Refactor BatchDrain and update Join methods

Refactored `BatchDrain` class in `BatchDrain.cs` by removing `_noOpTarget`, `NoOp`, and `GlobalCleanup`. Updated benchmark methods to use `Join` with a `true` parameter and static lambda for `ReadAll`. Modified `BatchListDrainPooled` and `BatchQueueDrainPooled` to use `Join` with recycler action.

In `Extensions.Join.cs`, updated `JoiningChannelReader` to accept an optional `recycler` action. Added `QueueJoiningChannelReader` and `MemoryJoiningChannelReader` for `Queue<T>` and `IMemoryOwner<T>`. Updated `Join` methods to support new types and optional `recycler` action. Added `TryDequeue` for `Queue<T>` to support .NET Standard 2.0.

Author: electricessence

.editorconfig

@@ -93,7 +93,7 @@ csharp_style_pattern_matching_over_as_with_null_check = true:warning
 csharp_style_pattern_matching_over_is_with_cast_check = true:warning
 csharp_style_prefer_not_pattern = true:warning
 csharp_style_prefer_pattern_matching = true:suggestion
-csharp_style_prefer_switch_expression = true
+csharp_style_prefer_switch_expression = true:suggestion
 
 # Null-checking preferences
 csharp_style_conditional_delegate_call = true:warning
@@ -143,7 +143,7 @@ csharp_new_line_between_query_expression_clauses = true
 csharp_indent_block_contents = true
 csharp_indent_braces = false
 csharp_indent_case_contents = true
-csharp_indent_case_contents_when_block = true
+csharp_indent_case_contents_when_block = false
 csharp_indent_labels = one_less_than_current
 csharp_indent_switch_labels = true
 

Open.ChannelExtensions.Benchmarks/BatchDrain.cs

@@ -0,0 +1,82 @@
+using BenchmarkDotNet.Attributes;
+using System.Threading.Channels;
+
+namespace Open.ChannelExtensions.Benchmarks;
+
+[MemoryDiagnoser]
+public class BatchDrain
+{
+	[Params(10, 100, 500, 1000, 3000, 12000)]
+	public int BatchSize { get; set; }
+
+	public int TotalItems = 10000000;
+
+	private Channel<int>? _channel;
+
+	private readonly Queue<List<int>> _listPool = new();
+	private readonly Queue<Queue<int>> _queuePool = new();
+
+	[IterationSetup]
+	public void IterationSetup()
+	{
+		_channel = Channel.CreateUnbounded<int>(new UnboundedChannelOptions
+		{
+			SingleReader = true,
+			SingleWriter = true
+		});
+
+		for (int i = 0; i < TotalItems; i++)
+		{
+			_channel.Writer.TryWrite(i);
+		}
+
+		_channel.Writer.Complete();
+
+		_listPool.Clear();
+		_queuePool.Clear();
+
+		_listPool.Enqueue(new List<int>(BatchSize));
+		_listPool.Enqueue(new List<int>(BatchSize));
+		_queuePool.Enqueue(new Queue<int>(BatchSize));
+		_queuePool.Enqueue(new Queue<int>(BatchSize));
+	}
+
+	[Benchmark(Baseline = true)]
+	public async Task BatchListDrain()
+		=> await _channel!.Reader
+			.Batch(BatchSize)
+			.Join(true)
+			.ReadAll(static _ => { });
+
+	[Benchmark]
+	public async Task BatchListDrainPooled()
+		=> await _channel!.Reader
+			.Batch(BatchSize, batchFactory: _ => _listPool.Dequeue())
+			.Join(true, e =>
+			{
+				e.Clear(); // Simulate resetting the size.
+				_listPool.Enqueue(e);
+			})
+			.ReadAll(static _ => { });
+
+	[Benchmark]
+	public async Task BatchQueueDrain()
+		=> await _channel!.Reader
+			.BatchToQueues(BatchSize)
+			.Join(true)
+			.ReadAll(static _ => { });
+
+	[Benchmark]
+	public async Task BatchQueueDrainPooled()
+		=> await _channel!.Reader
+			.BatchToQueues(BatchSize, batchFactory: _ => _queuePool.Dequeue())
+			.Join(true, _queuePool.Enqueue)
+			.ReadAll(static _ => { });
+
+	[Benchmark]
+	public async Task BatchMemoryOwnerDrain()
+		=> await _channel!.Reader
+			.BatchAsMemory(BatchSize)
+			.Join(true)
+			.ReadAll(static _ => { });
+}

Open.ChannelExtensions.Benchmarks/Open.ChannelExtensions.Benchmarks.csproj

@@ -0,0 +1,18 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net8.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="BenchmarkDotNet" Version="0.14.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Open.ChannelExtensions\Open.ChannelExtensions.csproj" />
+  </ItemGroup>
+
+</Project>

Open.ChannelExtensions.Benchmarks/Program.cs

@@ -0,0 +1,3 @@
+using Open.ChannelExtensions.Benchmarks;
+using BenchmarkDotNet.Running;
+BenchmarkRunner.Run<BatchDrain>();