More task scheduling improvements.

Author: electricessence

Open.ChannelExtensions.Tests/Open.ChannelExtensions.Tests.csproj

@@ -5,6 +5,7 @@
 		<TargetFramework>net9.0</TargetFramework>
 		<LangVersion>latest</LangVersion>
 		<IsPackable>false</IsPackable>
+		<RunSettingsFilePath>D:\Users\essence\Development\- GitHub\- Open-NET-Libraries\Open.ChannelExtensions\Open.ChannelExtensions.Tests\bin\Debug\net9.0\fine-code-coverage\coverage-tool-output\Open.ChannelExtensions.Tests-fcc-mscodecoverage-generated.runsettings</RunSettingsFilePath>
 	</PropertyGroup>
 
 	<ItemGroup>

Open.ChannelExtensions.Tests/TaskSchedulerTests.cs

@@ -0,0 +1,49 @@
+namespace Open.ChannelExtensions.Tests;
+
+public static class TaskSchedulerTests
+{
+	[Fact]
+	public static async Task ReadAllConcurrentlyAsync_UsesCurrentScheduler()
+	{
+		// Arrange
+		var channel = Channel.CreateUnbounded<int>();
+		var testScheduler = new TestTaskScheduler();
+		const int maxConcurrency = 3;
+		const int itemsToProcess = 10000;
+
+		// Populate the channel with test data
+		for (int i = 0; i < itemsToProcess; i++)
+		{
+			await channel.Writer.WriteAsync(i);
+		}
+		channel.Writer.Complete();
+
+		// Act
+		var itemsProcessed = await channel.Reader
+			.ReadAllConcurrentlyAsync(
+				maxConcurrency,
+				testScheduler,
+				_ =>
+				{
+					Assert.True(testScheduler == TaskScheduler.Current, "The custom scheduler was not used.");
+					return ValueTask.CompletedTask;
+				});
+
+		// Assert
+		Assert.Equal(itemsToProcess, itemsProcessed);
+	}
+
+	private class TestTaskScheduler : TaskScheduler
+	{
+		protected override IEnumerable<Task> GetScheduledTasks() => null;
+
+		protected override void QueueTask(Task task)
+			=> Task.Factory.StartNew(() => TryExecuteTask(task),
+				CancellationToken.None,
+				TaskCreationOptions.None,
+				TaskScheduler.Default);
+
+		protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
+			=> TryExecuteTask(task);
+	}
+}

Open.ChannelExtensions/Extensions.Pipe.cs

@@ -54,21 +54,49 @@ public static async ValueTask<long> PipeTo<T>(this ChannelReader<T> source,
 	/// <typeparam name="T">The type contained by the source channel and written to the target..</typeparam>
 	/// <param name="source">The source channel.</param>
 	/// <param name="target">The target channel.</param>
+	/// <param name="scheduler">The task scheduler to use for the operation.</param>
+	/// <param name="taskCreationOptions">The options to use for starting the task.</param>
 	/// <param name="cancellationToken">An optional cancellation token.</param>
 	/// <returns>The channel reader of the target.</returns>
 	public static ChannelReader<T> PipeTo<T>(this ChannelReader<T> source,
 		Channel<T> target,
+		TaskScheduler scheduler,
+		TaskCreationOptions taskCreationOptions = TaskCreationOptions.None,
 		CancellationToken cancellationToken = default)
 	{
 		if (source is null) throw new ArgumentNullException(nameof(source));
 		if (target is null) throw new ArgumentNullException(nameof(target));
 		Contract.EndContractBlock();
 
-		Task.Run(() => PipeTo(source, target.Writer, true, cancellationToken));
+		_ = Task.Factory.StartNew(
+			() => PipeTo(source, target.Writer, true, cancellationToken),
+			cancellationToken,
+			taskCreationOptions,
+			scheduler);
 
 		return target.Reader;
 	}
 
+	/// <inheritdoc cref="PipeTo{T}(ChannelReader{T}, Channel{T}, TaskScheduler, TaskCreationOptions, CancellationToken)"/>
+	public static ChannelReader<T> PipeTo<T>(this ChannelReader<T> source,
+		Channel<T> target,
+		TaskScheduler scheduler,
+		CancellationToken cancellationToken)
+		=> PipeTo<T>(source, target, scheduler, TaskCreationOptions.None, cancellationToken);
+
+	/// <inheritdoc cref="PipeTo{T}(ChannelReader{T}, Channel{T}, TaskScheduler, TaskCreationOptions, CancellationToken)"/>
+	public static ChannelReader<T> PipeTo<T>(this ChannelReader<T> source,
+		Channel<T> target,
+		CancellationToken cancellationToken)
+		=> PipeTo<T>(source, target, TaskScheduler.Current, TaskCreationOptions.None, cancellationToken);
+
+	/// <inheritdoc cref="PipeTo{T}(ChannelReader{T}, Channel{T}, TaskScheduler, TaskCreationOptions, CancellationToken)"/>
+	public static ChannelReader<T> PipeTo<T>(this ChannelReader<T> source,
+		Channel<T> target,
+		TaskCreationOptions taskCreationOptions = TaskCreationOptions.None,
+		CancellationToken cancellationToken = default)
+		=> PipeTo<T>(source, target, TaskScheduler.Current, taskCreationOptions, cancellationToken);
+
 	/// <summary>
 	/// Reads all entries concurrently and applies the values to the provided transform function before buffering the results into another channel for consumption.
 	/// </summary>

Open.ChannelExtensions/Extensions.ReadConcurrently.cs

@@ -8,9 +8,34 @@ public static partial class Extensions
 	/// <typeparam name="T">The item type.</typeparam>
 	/// <param name="reader">The channel reader to read from.</param>
 	/// <param name="maxConcurrency">The maximum number of concurrent operations.  Greater than 1 may likely cause results to be out of order.</param>
+	/// <param name="taskCreationOptions">The task creation options to use.</param>
+	/// <param name="scheduler">The task scheduler to use.</param>
 	/// <param name="receiver">The async receiver function.</param>
-	/// <param name="cancellationToken">An optional cancellation token.</param>
+	/// <param name="cancellationToken">The cancellation token.</param>
 	/// <returns>A task that completes when no more reading is to be done.</returns>
+	public static Task<long> ReadAllConcurrentlyAsync<T>(this ChannelReader<T> reader,
+		int maxConcurrency,
+		TaskScheduler scheduler,
+		TaskCreationOptions taskCreationOptions,
+		Func<T, ValueTask> receiver,
+		CancellationToken cancellationToken = default)
+		=> Task.Factory
+			.StartNew(
+				() => ReadAllConcurrentlyAsync(reader, maxConcurrency, receiver, cancellationToken),
+				cancellationToken,
+				taskCreationOptions,
+				scheduler)
+			.Unwrap();
+
+	/// <inheritdoc cref="ReadAllConcurrentlyAsync{T}(ChannelReader{T}, int, TaskScheduler, TaskCreationOptions, Func{T, ValueTask}, CancellationToken)"/>
+	public static Task<long> ReadAllConcurrentlyAsync<T>(this ChannelReader<T> reader,
+		int maxConcurrency,
+		TaskScheduler scheduler,
+		Func<T, ValueTask> receiver,
+		CancellationToken cancellationToken = default)
+		=> ReadAllConcurrentlyAsync(reader, maxConcurrency, scheduler, TaskCreationOptions.None, receiver, cancellationToken);
+
+	/// <inheritdoc cref="ReadAllConcurrentlyAsync{T}(ChannelReader{T}, int, TaskScheduler, TaskCreationOptions, Func{T, ValueTask}, CancellationToken)"/>
 	public static Task<long> ReadAllConcurrentlyAsync<T>(this ChannelReader<T> reader,
 		int maxConcurrency,
 		Func<T, ValueTask> receiver,
@@ -37,7 +62,7 @@ static async Task<long> ReadAllConcurrentlyAsyncCore(
 			var readers = new Task<long>[maxConcurrency];
 			var scheduler = TaskScheduler.Current;
 			for (int r = 0; r < maxConcurrency; ++r)
-				readers[r] = Read();
+				readers[r] = Task.Factory.StartNew(Read, cancellationToken, TaskCreationOptions.None, scheduler).Unwrap();
 
 			// This produces the most accurate/reliable exception and cancellation results.
 			return await Task
@@ -73,22 +98,33 @@ static async Task<long> SumAsync(Task<long[]> counts)
 		}
 	}
 
-	/// <summary>
-	/// Reads items from the channel and passes them to the receiver.
-	/// </summary>
-	/// <typeparam name="T">The item type.</typeparam>
-	/// <param name="reader">The channel reader to read from.</param>
-	/// <param name="maxConcurrency">The maximum number of concurrent operations.  Greater than 1 may likely cause results to be out of order.</param>
-	/// <param name="cancellationToken">The cancellation token.</param>
-	/// <param name="receiver">The async receiver function.</param>
-	/// <returns>A task that completes when no more reading is to be done.</returns>
+	/// <inheritdoc cref="ReadAllConcurrentlyAsync{T}(ChannelReader{T}, int, TaskScheduler, TaskCreationOptions, Func{T, ValueTask}, CancellationToken)"/>
 	[SuppressMessage("Design", "CA1068:CancellationToken parameters must come last", Justification = "Provided for aesthetic convenience.")]
 	public static Task<long> ReadAllConcurrentlyAsync<T>(this ChannelReader<T> reader,
 		int maxConcurrency,
 		CancellationToken cancellationToken,
 		Func<T, ValueTask> receiver)
 		=> ReadAllConcurrentlyAsync(reader, maxConcurrency, receiver, cancellationToken);
 
+	/// <inheritdoc cref="ReadAllConcurrentlyAsync{T}(ChannelReader{T}, int, TaskScheduler, TaskCreationOptions, Func{T, ValueTask}, CancellationToken)"/>
+	[SuppressMessage("Design", "CA1068:CancellationToken parameters must come last", Justification = "Provided for aesthetic convenience.")]
+	public static Task<long> ReadAllConcurrentlyAsync<T>(this ChannelReader<T> reader,
+		int maxConcurrency,
+		TaskScheduler scheduler,
+		TaskCreationOptions taskCreationOptions,
+		CancellationToken cancellationToken,
+		Func<T, ValueTask> receiver)
+		=> ReadAllConcurrentlyAsync(reader, maxConcurrency, scheduler, taskCreationOptions, receiver, cancellationToken);
+
+	/// <inheritdoc cref="ReadAllConcurrentlyAsync{T}(ChannelReader{T}, int, TaskScheduler, TaskCreationOptions, Func{T, ValueTask}, CancellationToken)"/>
+	[SuppressMessage("Design", "CA1068:CancellationToken parameters must come last", Justification = "Provided for aesthetic convenience.")]
+	public static Task<long> ReadAllConcurrentlyAsync<T>(this ChannelReader<T> reader,
+		int maxConcurrency,
+		TaskScheduler scheduler,
+		CancellationToken cancellationToken,
+		Func<T, ValueTask> receiver)
+		=> ReadAllConcurrentlyAsync(reader, maxConcurrency, scheduler, TaskCreationOptions.None, receiver, cancellationToken);
+
 	/// <summary>
 	/// Reads items from the channel and passes them to the receiver.
 	/// </summary>

Open.ChannelExtensions/Open.ChannelExtensions.csproj

@@ -22,7 +22,7 @@
 		<RepositoryType>git</RepositoryType>
 		<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
 		<GenerateDocumentationFile>true</GenerateDocumentationFile>
-		<Version>9.0.0</Version>
+		<Version>9.1.0</Version>
 		<PackageReleaseNotes>Added .Merge, .PipeAsync, and .PropagateCompletion extensions.</PackageReleaseNotes>
 		<PackageLicenseExpression>MIT</PackageLicenseExpression>
 		<PublishRepositoryUrl>true</PublishRepositoryUrl>