Optimizations and ensured live changing the timeout during operation.

Author: electricessence

Open.ChannelExtensions.Tests/AssumptionTests.cs

@@ -0,0 +1,43 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Open.ChannelExtensions.Tests;
+
+public static class AssumptionTests
+{
+	[Fact]
+	public static async Task WaitCancellation()
+	{
+		var channel = Channel.CreateUnbounded<int>();
+		using (var tokenSource = new CancellationTokenSource())
+		{
+			CancellationToken token = tokenSource.Token;
+
+			var t = channel.Reader.WaitToReadAsync(token).ConfigureAwait(false);
+			tokenSource.Cancel();
+
+			// NOTE: a cancelled WaitToReadAsync will throw. 
+			await Assert.ThrowsAsync<OperationCanceledException>(async () => await t);
+		}
+
+		using (var tokenSource = new CancellationTokenSource())
+		{
+			CancellationToken token = tokenSource.Token;
+
+			var t1 = channel.Reader.WaitToReadAsync(token);
+			var t2 = channel.Reader.WaitToReadAsync();
+			tokenSource.Cancel();
+
+			// NOTE: a cancelled WhenAny will not throw!
+			var result = await Task.WhenAny(t1.AsTask(), t2.AsTask()).ConfigureAwait(false);
+			Assert.True(result.IsCanceled);
+		}
+
+	}
+}

Open.ChannelExtensions.Tests/BatchTests.cs

@@ -340,7 +340,7 @@ public static async Task BatchReadBehavior()
 
 		async ValueTask Dequeue()
 		{
-			Assert.True(c.Writer.TryWrite(e));
+			if (!c.Writer.TryWrite(e)) return;
 			while (queue.TryDequeue(out e) && c.Writer.TryWrite(e))
 				await Task.Yield();
 		}

Open.ChannelExtensions/BatchingChannelReader.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Diagnostics.Contracts;
+using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
@@ -68,9 +69,27 @@ public BatchingChannelReader<T> WithTimeout(long millisecondsTimeout)
 		LazyInitializer.EnsureInitialized(ref _timer,
 			() => new Timer(obj => ForceBatch()));
 
+		if (_batch is null) return this;
+		
+		// Might be in the middle of a batch so we need to update the timeout.
+		lock(Buffer)
+		{
+			if (_batch is not null)	RefreshTimeout();
+		}
+
 		return this;
 	}
 
+	/// <summary>
+	/// If one exists, updates the timer's timeout value.
+	/// </summary>
+	[MethodImpl(MethodImplOptions.AggressiveInlining)]
+	protected void RefreshTimeout()
+	{
+		var ok = _timer?.Change(_timeout, 0);
+		Debug.Assert(ok ?? true);
+	}
+
 	/// <param name="timeout">
 	/// The timeout value where after a batch is forced.<br/>
 	/// A value of zero or less cancels/clears any timeout.<br/>
@@ -139,13 +158,9 @@ protected override bool TryPipeItems(bool flush)
 			Emit(ref c);
 
 		finalizeTimer:
-			
+
 			// Are we adding to the existing batch (active timeout) or did we create a new one?
-			if (newBatch && _batch is not null)
-			{
-				var ok = _timer?.Change(_timeout, 0);
-				Debug.Assert(ok ?? true);
-			}
+			if (newBatch && _batch is not null)	RefreshTimeout();
 
 			return batched;
 
@@ -167,16 +182,16 @@ protected override async ValueTask<bool> WaitToReadAsyncCore(
 		CancellationToken cancellationToken)
 	{
 		ChannelReader<T>? source = Source;
-		if (source is null) return await bufferWait.ConfigureAwait(false);
 
-		Task<bool>? b = bufferWait.AsTask();
+		if (source is null || bufferWait.IsCompleted)
+			return await bufferWait.ConfigureAwait(false);
+
+		var b = bufferWait.AsTask();
 		using var tokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
 		CancellationToken token = tokenSource.Token;
 
 	start:
 
-		if (b.IsCompleted) return await b.ConfigureAwait(false);
-
 		ValueTask<bool> s = source.WaitToReadAsync(token);
 		if (s.IsCompleted && !b.IsCompleted) TryPipeItems(false);
 
@@ -186,14 +201,19 @@ protected override async ValueTask<bool> WaitToReadAsyncCore(
 			return await b.ConfigureAwait(false);
 		}
 
+		// WhenAny will not throw when a task is cancelled.
 		await Task.WhenAny(s.AsTask(), b).ConfigureAwait(false);
-		if (b.IsCompleted)
+		if (b.IsCompleted) // Assuming it was bufferWait that completed.
 		{
 			tokenSource.Cancel();
 			return await b.ConfigureAwait(false);
 		}
 
 		TryPipeItems(false);
+
+		if (b.IsCompleted || token.IsCancellationRequested)
+			return await b.ConfigureAwait(false);
+
 		goto start;
 	}
 }

Open.ChannelExtensions/BufferingChannelReader.cs

@@ -48,6 +48,9 @@ protected BufferingChannelReader(ChannelReader<TIn> source, bool singleReader, b
 				// Need to be sure writing is done before we continue...
 				lock (Buffer)
 				{
+					/* When the source is complete,
+					 * we dump all remaining into the buffer 
+					 * in order to propagate the completion and any exception. */
 					TryPipeItems(true);
 					Buffer.Writer.Complete(t.Exception);
 				}
@@ -114,15 +117,14 @@ public override ValueTask<bool> WaitToReadAsync(CancellationToken cancellationTo
 	protected virtual async ValueTask<bool> WaitToReadAsyncCore(ValueTask<bool> bufferWait, CancellationToken cancellationToken)
 	{
 		ChannelReader<TIn>? source = Source;
-		if (source is null) return await bufferWait.ConfigureAwait(false);
+		if (source is null || bufferWait.IsCompleted) 
+			return await bufferWait.ConfigureAwait(false);
 
 		using var tokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
 		CancellationToken token = tokenSource.Token;
 
 	start:
 
-		if (bufferWait.IsCompleted) return await bufferWait.ConfigureAwait(false);
-
 		ValueTask<bool> s = source.WaitToReadAsync(token);
 		if (s.IsCompleted && !bufferWait.IsCompleted) TryPipeItems(false);
 
@@ -131,13 +133,11 @@ protected virtual async ValueTask<bool> WaitToReadAsyncCore(ValueTask<bool> buff
 			tokenSource.Cancel();
 			return await bufferWait.ConfigureAwait(false);
 		}
+
 		await s.ConfigureAwait(false);
-		if (bufferWait.IsCompleted)
-		{
-			tokenSource.Cancel();
-			return await bufferWait.ConfigureAwait(false);
-		}
+		if (bufferWait.IsCompleted)	return await bufferWait.ConfigureAwait(false);
 		TryPipeItems(false);
+		if (bufferWait.IsCompleted) return await bufferWait.ConfigureAwait(false);
 
 		goto start;
 	}

Open.ChannelExtensions/Documentation.xml

@@ -20,7 +20,7 @@
 		<RepositoryUrl>https://github.com/Open-NET-Libraries/Open.ChannelExtensions</RepositoryUrl>
 		<RepositoryType>git</RepositoryType>
 		<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
-		<Version>5.1.1</Version>
+		<Version>5.1.2</Version>
 		<PackageReleaseNotes></PackageReleaseNotes>
 		<PackageLicenseExpression>MIT</PackageLicenseExpression>
 		<PublishRepositoryUrl>true</PublishRepositoryUrl>