diff --git a/TaskFlow.Tests/Extensions/CompletionContinuationLifetimeFixture.cs b/TaskFlow.Tests/Extensions/CompletionContinuationLifetimeFixture.cs index 4c0943e..744b0a1 100644 --- a/TaskFlow.Tests/Extensions/CompletionContinuationLifetimeFixture.cs +++ b/TaskFlow.Tests/Extensions/CompletionContinuationLifetimeFixture.cs @@ -90,13 +90,13 @@ public async Task IndependentChildren_DrainBeforeOwnedCleanup(bool explicitLifet } [Test] - public async Task SubmissionFailure_IndependentCompletionStillDrains() + public async Task ReturnedSubmissionFailure_IndependentCompletionStillDrains() { var events = new ConcurrentQueue(); await using var lifetime = new TaskSchedulerMiddlewareLifetime(); var outer = new IndependentCompletion(events); var inner = new GatedCompletion(events); - var scheduler = new FailingTerminal().UseMiddleware(new Factory(outer), lifetime) + var scheduler = new FaultedTerminal().UseMiddleware(new Factory(outer), lifetime) .UseMiddleware(new Factory(inner)); var operation = scheduler.Enqueue(() => 42); try @@ -250,10 +250,10 @@ public async Task DisposalRacingChildCompletion_CleansUpOnceAfterChild() } } - private sealed class FailingTerminal : ITaskScheduler + private sealed class FaultedTerminal : ITaskScheduler { public Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) - => throw new FormatException("submission"); + => Task.FromException(new FormatException("submission")); } private sealed class Factory(ITaskSchedulerMiddleware middleware) : ITaskSchedulerMiddlewareFactory diff --git a/TaskFlow.Tests/Extensions/LifetimeTrackingRegressionFixture.cs b/TaskFlow.Tests/Extensions/LifetimeTrackingRegressionFixture.cs index 2fe26e3..9fd9b10 100644 --- a/TaskFlow.Tests/Extensions/LifetimeTrackingRegressionFixture.cs +++ b/TaskFlow.Tests/Extensions/LifetimeTrackingRegressionFixture.cs @@ -93,8 +93,8 @@ public async Task AnnotationOnlySnapshot_DrainsAcceptedWorkAndRejectsClosedEntry var shutdown = flow.DisposeAsync().AsTask(); Assert.That(shutdown.IsCompleted, Is.False); Task rejected = null!; - Assert.That(() => { rejected = scheduler.Enqueue(() => 0); }, Throws.Nothing); - await Assert.ThatAsync(async () => await rejected, Throws.TypeOf()); + Assert.That(() => { rejected = scheduler.Enqueue(() => 0); }, Throws.TypeOf()); + Assert.That(rejected, Is.Null, "Closed entry must reject the enqueue call before returning a task."); release.SetResult(true); Assert.That(await pending, Is.EqualTo(42)); await shutdown; @@ -223,7 +223,8 @@ public async Task InlineTerminals_PreserveContinuationResultsAndCancellation(boo { await Assert.ThatAsync(async () => await task, Throws.Exception.SameAs(failure)); Assert.That(task.IsCanceled, Is.EqualTo(failure is OperationCanceledException)); - Assert.That(observer.Status, Is.EqualTo(failure is OperationCanceledException ? TaskStatus.Canceled : TaskStatus.Faulted)); + // A synchronous next rejection returns no continuation task for this observer to inspect. + Assert.That(observer.Status, Is.EqualTo(!faultedTask ? TaskStatus.Created : failure is OperationCanceledException ? TaskStatus.Canceled : TaskStatus.Faulted)); } } } diff --git a/TaskFlow.Tests/TaskFlowSchedulerAdapterFixture.cs b/TaskFlow.Tests/TaskFlowSchedulerAdapterFixture.cs new file mode 100644 index 0000000..f719593 --- /dev/null +++ b/TaskFlow.Tests/TaskFlowSchedulerAdapterFixture.cs @@ -0,0 +1,37 @@ +namespace TaskFlow.Tests; + +using NUnit.Framework; +using System.Threading.Tasks.Flow; + +[TestFixture] +internal sealed class TaskFlowSchedulerAdapterFixture +{ + [TestCase(false)] + [TestCase(true)] + public async Task Enqueue_QueueTaskFailure_ShouldPreserveStartNewException(bool disposed) + { + Exception failure = disposed ? new ObjectDisposedException("scheduler") : new InvalidOperationException("rejected"); + var scheduler = new TaskFlowSchedulerAdapter(new RejectingScheduler(failure)); + Task? pending = null; + + Assert.That(() => { pending = scheduler.Enqueue((_, _) => new ValueTask(42), null, CancellationToken.None); }, Throws.Nothing); + await Assert.ThatAsync(async () => await pending!, + Throws.TypeOf().With.Property(nameof(Exception.InnerException)).SameAs(failure)); + } + + [Test] + public async Task Enqueue_NullDelegate_ShouldRetainTaskBasedValidation() + { + var scheduler = new TaskFlowSchedulerAdapter(TaskScheduler.Default); + Task? pending = null; + Assert.That(() => { pending = scheduler.Enqueue(null!, null, CancellationToken.None); }, Throws.Nothing); + await Assert.ThatAsync(async () => await pending!, Throws.TypeOf()); + } + + private sealed class RejectingScheduler(Exception failure) : TaskScheduler + { + protected override void QueueTask(Task task) => throw failure; + protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) => false; + protected override IEnumerable GetScheduledTasks() => []; + } +} \ No newline at end of file diff --git a/TaskFlow.Tests/TaskSchedulerDisposalConformance.cs b/TaskFlow.Tests/TaskSchedulerDisposalConformance.cs new file mode 100644 index 0000000..08bcd3d --- /dev/null +++ b/TaskFlow.Tests/TaskSchedulerDisposalConformance.cs @@ -0,0 +1,153 @@ +namespace TaskFlow.Tests; + +using NUnit.Framework; +using System.Threading.Tasks.Flow; + +internal static class TaskSchedulerDisposalConformance +{ + public static readonly string[] Overloads = + [ + "DirectGeneric", "Action", "ActionWithToken", "TokenAction", "TokenActionWithToken", + "Task", "TokenTask", "TokenTaskWithToken", "StateTask", + "ValueTask", "TokenValueTask", "TokenValueTaskWithToken", "StateValueTask", + "TaskResult", "TokenTaskResult", "TokenTaskResultWithToken", "StateTaskResult", + "ValueTaskResult", "TokenValueTaskResult", "TokenValueTaskResultWithToken", "StateValueTaskResult", + "Func", "TokenFunc", + ]; + + public static void AssertRejected(ITaskScheduler scheduler, string overload, bool keyed) + { + var invoked = false; + Task? returnedTask = null; + Assert.That(() => { returnedTask = Enqueue(scheduler, overload, () => invoked = true, keyed); }, Throws.TypeOf()); + Assert.That(returnedTask, Is.Null, "A disposed scheduler must reject the call before returning a task."); + Assert.That(invoked, Is.False, "A rejected delegate must never execute."); + } + + public static Task Enqueue(ITaskScheduler scheduler, string overload, Action action, bool keyed) + { + Task TaskOperation() + { action(); return Task.CompletedTask; } + ValueTask ValueTaskOperation() + { action(); return ValueTask.CompletedTask; } + if (keyed) + { + return overload switch + { + "DirectGeneric" => scheduler.Enqueue("key", (key, state, token) => { action(); return new ValueTask(42); }, 42, CancellationToken.None), + "Action" => scheduler.Enqueue("key", action), + "ActionWithToken" => scheduler.Enqueue("key", action, CancellationToken.None), + "TokenAction" => scheduler.Enqueue("key", (string key) => { action(); }), + "TokenActionWithToken" => scheduler.Enqueue("key", (string key) => { action(); }, CancellationToken.None), + "Task" => scheduler.Enqueue("key", TaskOperation), + "TokenTask" => scheduler.Enqueue("key", (CancellationToken token) => TaskOperation()), + "TokenTaskWithToken" => scheduler.Enqueue("key", (CancellationToken token) => TaskOperation(), CancellationToken.None), + "StateTask" => scheduler.Enqueue("key", (key, state, token) => { action(); return Task.FromResult(state); }, 42, CancellationToken.None), + "ValueTask" => scheduler.Enqueue("key", ValueTaskOperation), + "TokenValueTask" => scheduler.Enqueue("key", (CancellationToken token) => ValueTaskOperation()), + "TokenValueTaskWithToken" => scheduler.Enqueue("key", (CancellationToken token) => ValueTaskOperation(), CancellationToken.None), + "StateValueTask" => scheduler.Enqueue("key", (key, state, token) => { action(); return new ValueTask(state); }, 42, CancellationToken.None), + "TaskResult" => scheduler.Enqueue("key", () => { action(); return Task.FromResult(42); }), + "TokenTaskResult" => scheduler.Enqueue("key", (CancellationToken token) => { action(); return Task.FromResult(42); }), + "TokenTaskResultWithToken" => scheduler.Enqueue("key", (CancellationToken token) => { action(); return Task.FromResult(42); }, CancellationToken.None), + "StateTaskResult" => scheduler.Enqueue("key", (key, state, token) => { action(); return Task.FromResult(state); }, 42, CancellationToken.None), + "ValueTaskResult" => scheduler.Enqueue("key", () => { action(); return new ValueTask(42); }), + "TokenValueTaskResult" => scheduler.Enqueue("key", (CancellationToken token) => { action(); return new ValueTask(42); }), + "TokenValueTaskResultWithToken" => scheduler.Enqueue("key", (CancellationToken token) => { action(); return new ValueTask(42); }, CancellationToken.None), + "StateValueTaskResult" => scheduler.Enqueue("key", (key, state, token) => { action(); return new ValueTask(state); }, 42, CancellationToken.None), + "Func" => scheduler.Enqueue("key", () => { action(); return 42; }), + "TokenFunc" => scheduler.Enqueue("key", (string key) => { action(); return 42; }, CancellationToken.None), + _ => throw new ArgumentOutOfRangeException(nameof(overload)), + }; + } + + return overload switch + { + "DirectGeneric" => scheduler.Enqueue((state, token) => { action(); return new ValueTask(42); }, null, CancellationToken.None), + "Action" => scheduler.Enqueue(action), + "ActionWithToken" => scheduler.Enqueue(action, CancellationToken.None), + "TokenAction" => scheduler.Enqueue(token => { action(); }), + "TokenActionWithToken" => scheduler.Enqueue(token => { action(); }, CancellationToken.None), + "Task" => scheduler.Enqueue(TaskOperation), + "TokenTask" => scheduler.Enqueue(token => TaskOperation()), + "TokenTaskWithToken" => scheduler.Enqueue(token => TaskOperation(), CancellationToken.None), + "StateTask" => scheduler.Enqueue((int state, CancellationToken token) => TaskOperation(), 42, CancellationToken.None), + "ValueTask" => scheduler.Enqueue(ValueTaskOperation, null), + "TokenValueTask" => scheduler.Enqueue(token => ValueTaskOperation(), null), + "TokenValueTaskWithToken" => scheduler.Enqueue(token => ValueTaskOperation(), CancellationToken.None, null), + "StateValueTask" => scheduler.Enqueue((int state, CancellationToken token) => ValueTaskOperation(), 42, CancellationToken.None), + "TaskResult" => scheduler.Enqueue(() => { action(); return Task.FromResult(42); }), + "TokenTaskResult" => scheduler.Enqueue(token => { action(); return Task.FromResult(42); }), + "TokenTaskResultWithToken" => scheduler.Enqueue(token => { action(); return Task.FromResult(42); }, CancellationToken.None), + "StateTaskResult" => scheduler.Enqueue((int state, CancellationToken token) => { action(); return Task.FromResult(state); }, 42, CancellationToken.None), + "ValueTaskResult" => scheduler.Enqueue(() => { action(); return new ValueTask(42); }, null), + "TokenValueTaskResult" => scheduler.Enqueue(token => { action(); return new ValueTask(42); }, null), + "TokenValueTaskResultWithToken" => scheduler.Enqueue(token => { action(); return new ValueTask(42); }, CancellationToken.None), + "StateValueTaskResult" => scheduler.Enqueue((int state, CancellationToken token) => { action(); return new ValueTask(state); }, 42, CancellationToken.None), + "Func" => scheduler.Enqueue(() => { action(); return 42; }), + "TokenFunc" => scheduler.Enqueue(token => { action(); return 42; }, CancellationToken.None), + _ => throw new ArgumentOutOfRangeException(nameof(overload)), + }; + } + + public static Task EnqueueNullDelegate(ITaskScheduler scheduler, string overload, bool keyed) + { + if (keyed) + { + return overload switch + { + "DirectGeneric" => scheduler.Enqueue("key", (Func>)null!, 42, CancellationToken.None), + "Action" => scheduler.Enqueue("key", (Action)null!), + "ActionWithToken" => scheduler.Enqueue("key", (Action)null!, CancellationToken.None), + "TokenAction" => scheduler.Enqueue("key", (Action)null!), + "TokenActionWithToken" => scheduler.Enqueue("key", (Action)null!, CancellationToken.None), + "Task" => scheduler.Enqueue("key", (Func)null!), + "TokenTask" => scheduler.Enqueue("key", (Func)null!), + "TokenTaskWithToken" => scheduler.Enqueue("key", (Func)null!, CancellationToken.None), + "StateTask" => scheduler.Enqueue("key", (Func>)null!, 42, CancellationToken.None), + "ValueTask" => scheduler.Enqueue("key", (Func)null!), + "TokenValueTask" => scheduler.Enqueue("key", (Func)null!), + "TokenValueTaskWithToken" => scheduler.Enqueue("key", (Func)null!, CancellationToken.None), + "StateValueTask" => scheduler.Enqueue("key", (Func>)null!, 42, CancellationToken.None), + "TaskResult" => scheduler.Enqueue("key", (Func>)null!), + "TokenTaskResult" => scheduler.Enqueue("key", (Func>)null!), + "TokenTaskResultWithToken" => scheduler.Enqueue("key", (Func>)null!, CancellationToken.None), + "StateTaskResult" => scheduler.Enqueue("key", (Func>)null!, 42, CancellationToken.None), + "ValueTaskResult" => scheduler.Enqueue("key", (Func>)null!), + "TokenValueTaskResult" => scheduler.Enqueue("key", (Func>)null!), + "TokenValueTaskResultWithToken" => scheduler.Enqueue("key", (Func>)null!, CancellationToken.None), + "StateValueTaskResult" => scheduler.Enqueue("key", (Func>)null!, 42, CancellationToken.None), + "Func" => scheduler.Enqueue("key", (Func)null!), + "TokenFunc" => scheduler.Enqueue("key", (Func)null!, CancellationToken.None), + _ => throw new ArgumentOutOfRangeException(nameof(overload)), + }; + } + return overload switch + { + "DirectGeneric" => scheduler.Enqueue(null!, null, CancellationToken.None), + "Action" => scheduler.Enqueue((Action)null!), + "ActionWithToken" => scheduler.Enqueue((Action)null!, CancellationToken.None), + "TokenAction" => scheduler.Enqueue((Action)null!), + "TokenActionWithToken" => scheduler.Enqueue((Action)null!, CancellationToken.None), + "Task" => scheduler.Enqueue((Func)null!), + "TokenTask" => scheduler.Enqueue((Func)null!), + "TokenTaskWithToken" => scheduler.Enqueue((Func)null!, CancellationToken.None), + "StateTask" => scheduler.Enqueue((Func)null!, 42, CancellationToken.None), + "ValueTask" => scheduler.Enqueue((Func)null!, _: null), + "TokenValueTask" => scheduler.Enqueue((Func)null!, _: null), + "TokenValueTaskWithToken" => scheduler.Enqueue((Func)null!, CancellationToken.None, null), + "StateValueTask" => scheduler.Enqueue((Func)null!, 42, CancellationToken.None), + "TaskResult" => scheduler.Enqueue((Func>)null!), + "TokenTaskResult" => scheduler.Enqueue((Func>)null!), + "TokenTaskResultWithToken" => scheduler.Enqueue((Func>)null!, CancellationToken.None), + "StateTaskResult" => scheduler.Enqueue((Func>)null!, 42, CancellationToken.None), + "ValueTaskResult" => scheduler.Enqueue((Func>)null!, _: null), + "TokenValueTaskResult" => scheduler.Enqueue((Func>)null!, _: null), + "TokenValueTaskResultWithToken" => scheduler.Enqueue((Func>)null!, CancellationToken.None), + "StateValueTaskResult" => scheduler.Enqueue((Func>)null!, 42, CancellationToken.None), + "Func" => scheduler.Enqueue((Func)null!), + "TokenFunc" => scheduler.Enqueue((Func)null!, CancellationToken.None), + _ => throw new ArgumentOutOfRangeException(nameof(overload)), + }; + } +} \ No newline at end of file diff --git a/TaskFlow.Tests/TaskSchedulerDisposalConformanceFixture.cs b/TaskFlow.Tests/TaskSchedulerDisposalConformanceFixture.cs new file mode 100644 index 0000000..5a44ff8 --- /dev/null +++ b/TaskFlow.Tests/TaskSchedulerDisposalConformanceFixture.cs @@ -0,0 +1,211 @@ +namespace TaskFlow.Tests; + +using NUnit.Framework; +using System.Threading.Tasks.Flow; + +[TestFixtureSource(nameof(Scenarios))] +[CancelAfter(10000)] +internal sealed class TaskSchedulerDisposalConformanceFixture +{ + private readonly string _implementation; + private readonly string _configuration; + + private ITaskFlow? _flow; + private DisposableTaskScheduler? _adapterOwner; + private TaskSchedulerMiddlewareLifetime? _lifetime; + private ITaskScheduler _scheduler = null!; + private PhaseObserver _observer = null!; + + public TaskSchedulerDisposalConformanceFixture(string implementation, string configuration) + { + _implementation = implementation; + _configuration = configuration; + } + + public static IEnumerable Scenarios() + { + foreach (var implementation in new[] { "TaskFlow", "DedicatedThread", "CurrentThread" }) + foreach (var configuration in new[] { "Plain", "Annotated", "Pipeline", "ExtendedState", "Bounded" }) + yield return new TestFixtureData(implementation, configuration); + foreach (var configuration in new[] { "Concurrent", "Keyed", "KeyedBounded" }) + yield return new TestFixtureData("TaskFlow", configuration); + yield return new TestFixtureData("Adapter", "ExplicitLifetime"); + } + + [SetUp] + public async Task SetUp() + { + _observer = new PhaseObserver(); + _flow = null; + _adapterOwner = null; + _lifetime = null; + if (_implementation == "Adapter") + { + _adapterOwner = new DisposableTaskScheduler(); + _scheduler = new TaskFlowSchedulerAdapter(_adapterOwner); + } + else + { + _flow = _implementation switch + { + "TaskFlow" => new TaskFlow(), + "DedicatedThread" => new DedicatedThreadTaskFlow(), + "CurrentThread" => StartCurrentThread(), + _ => throw new ArgumentOutOfRangeException(nameof(_implementation)), + }; + _scheduler = _flow; + } + + _scheduler = _configuration switch + { + "Plain" => _scheduler, + "Annotated" => _scheduler.WithOperationName("disposal-conformance"), + "Pipeline" => _scheduler.UseMiddleware(_observer), + "ExtendedState" => _scheduler.WithExtendedState(42), + "Bounded" => _scheduler.WithBoundedAdmission(new BoundedAdmissionOptions { Capacity = 1 }).UseMiddleware(_observer), + "Concurrent" => _scheduler.WithConcurrencyLimit(1).UseMiddleware(_observer), + "Keyed" => _scheduler.WithKeyedScheduling().UseMiddleware(_observer), + "KeyedBounded" => _scheduler.WithKeyedScheduling().WithPerKeyBoundedAdmission(new BoundedAdmissionOptions { Capacity = 1 }).UseMiddleware(_observer), + "ExplicitLifetime" => WithExplicitLifetime(_scheduler), + _ => throw new ArgumentOutOfRangeException(nameof(_configuration)), + }; + // Exercise the same API before disposal so a permanently rejecting scheduler cannot pass. + await TaskSchedulerDisposalConformance.Enqueue(_scheduler, "DirectGeneric", () => { }, IsKeyed); + _observer.Invocations = 0; + } + + private bool IsKeyed => _configuration is "Keyed" or "KeyedBounded"; + + [TearDown] + public async Task TearDown() => await DisposeOwnerAsync(); + + [TestCaseSource(typeof(TaskSchedulerDisposalConformance), nameof(TaskSchedulerDisposalConformance.Overloads))] + public void Enqueue_AfterDispose_ShouldRejectSynchronously(string overload) + { + DisposeOwner(); + TaskSchedulerDisposalConformance.AssertRejected(_scheduler, overload, IsKeyed); + Assert.That(_observer.Invocations, Is.Zero, "Disposed middleware must not execute any phase."); + } + + [TestCaseSource(typeof(TaskSchedulerDisposalConformance), nameof(TaskSchedulerDisposalConformance.Overloads))] + public async Task Enqueue_AfterDisposeAsync_ShouldRejectSynchronously(string overload) + { + await DisposeOwnerAsync(); + TaskSchedulerDisposalConformance.AssertRejected(_scheduler, overload, IsKeyed); + Assert.That(_observer.Invocations, Is.Zero, "Disposed middleware must not execute any phase."); + } + + [TestCaseSource(typeof(TaskSchedulerDisposalConformance), nameof(TaskSchedulerDisposalConformance.Overloads))] + public async Task Enqueue_AfterShutdownStarts_ShouldRejectSynchronously(string overload) + { + var started = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + async Task Operation() + { + started.SetResult(); + await release.Task.ConfigureAwait(false); + } + var held = IsKeyed ? _scheduler.Enqueue("key", Operation) : _scheduler.Enqueue(Operation); + Task? shutdown = null; + try + { + await started.Task.WaitAsync(TimeSpan.FromSeconds(5)); + var invocations = _observer.Invocations; + shutdown = DisposeOwnerAsync(); + TaskSchedulerDisposalConformance.AssertRejected(_scheduler, overload, IsKeyed); + Assert.That(_observer.Invocations, Is.EqualTo(invocations), "Entry rejection must not invoke middleware while accepted work drains."); + } + finally + { + release.TrySetResult(); + if (shutdown != null) + await shutdown; + await held; + } + } + + [TestCaseSource(typeof(TaskSchedulerDisposalConformance), nameof(TaskSchedulerDisposalConformance.Overloads))] + public void Enqueue_NullDelegate_ShouldThrowSynchronously(string overload) + { + Task? task = null; + Assert.That(() => { task = TaskSchedulerDisposalConformance.EnqueueNullDelegate(_scheduler, overload, IsKeyed); }, Throws.TypeOf()); + Assert.That(task, Is.Null); + Assert.That(_observer.Invocations, Is.Zero, "Invalid arguments must be rejected before middleware executes."); + } + + [TestCase("DirectGeneric", "disposed")] + [TestCase("DirectGeneric", "argument")] + [TestCase("Action", "disposed")] + [TestCase("Action", "argument")] + public async Task Enqueue_AcceptedOperationThrows_ShouldFaultTask(string overload, string exceptionType) + { + Exception failure = exceptionType == "disposed" ? new ObjectDisposedException("operation-resource") : new ArgumentException("Operation failed"); + Task? task = null; + Assert.That(() => { task = TaskSchedulerDisposalConformance.Enqueue(_scheduler, overload, () => throw failure, IsKeyed); }, Throws.Nothing); + await Assert.ThatAsync(async () => await task!, Throws.Exception.SameAs(failure)); + Assert.That(task!.IsFaulted, Is.True); + } + + private ITaskScheduler WithExplicitLifetime(ITaskScheduler scheduler) + { + _lifetime = new TaskSchedulerMiddlewareLifetime(); + return scheduler.WithBoundedAdmission(new BoundedAdmissionController(new BoundedAdmissionOptions { Capacity = 1 }), _lifetime).UseMiddleware(_observer); + } + + private void DisposeOwner() + { + _lifetime?.DisposeAsync().AsTask().GetAwaiter().GetResult(); + _flow?.Dispose(); + _adapterOwner?.Dispose(); + } + + private async Task DisposeOwnerAsync() + { + if (_lifetime != null) + await _lifetime.DisposeAsync(); + if (_flow != null) + await _flow.DisposeAsync(); + _adapterOwner?.Dispose(); + } + + private static CurrentThreadTaskFlow StartCurrentThread() + { + var flow = new CurrentThreadTaskFlow(); + new Thread(flow.Run) { IsBackground = true }.Start(); + return flow; + } + + private sealed class PhaseObserver : ITaskSchedulerEnqueueMiddleware, ITaskSchedulerExecutionMiddleware, ITaskSchedulerCompletionMiddleware + { + public int Invocations { get; set; } + public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate next) + { + Invocations++; + return next(context); + } + public ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate next) + { + Invocations++; + return next(context); + } + public ValueTask> InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerOperationOutcome outcome, TaskSchedulerCompletionDelegate next) + { + Invocations++; + return next(context, outcome); + } + } + + private sealed class DisposableTaskScheduler : TaskScheduler, IDisposable + { + private int _disposed; + + public void Dispose() => Volatile.Write(ref _disposed, 1); + protected override void QueueTask(Task task) + { + ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposed) != 0, this); + ThreadPool.QueueUserWorkItem(_ => TryExecuteTask(task)); + } + protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) => false; + protected override IEnumerable GetScheduledTasks() => []; + } +} \ No newline at end of file diff --git a/TaskFlow.Tests/TaskSchedulerEnqueueExtensionsFixture.cs b/TaskFlow.Tests/TaskSchedulerEnqueueExtensionsFixture.cs new file mode 100644 index 0000000..ca37a6c --- /dev/null +++ b/TaskFlow.Tests/TaskSchedulerEnqueueExtensionsFixture.cs @@ -0,0 +1,78 @@ +namespace TaskFlow.Tests; + +using NUnit.Framework; +using System.Threading.Tasks.Flow; + +[TestFixture] +internal sealed class TaskSchedulerEnqueueExtensionsFixture +{ + private static IEnumerable ExtensionOverloads => TaskSchedulerDisposalConformance.Overloads.Where(overload => overload != "DirectGeneric"); + + [TestCase("Action")] + [TestCase("ActionWithToken")] + [TestCase("TokenAction")] + [TestCase("TokenActionWithToken")] + [TestCase("Task")] + [TestCase("TokenTask")] + [TestCase("TokenTaskWithToken")] + [TestCase("StateTask")] + [TestCase("ValueTask")] + [TestCase("TokenValueTask")] + [TestCase("TokenValueTaskWithToken")] + [TestCase("StateValueTask")] + public void Enqueue_ShouldReturnSchedulerTaskWithoutAdditionalContinuations(string overload) + { + var scheduler = new PendingScheduler(); + var task = Enqueue(scheduler, overload); + + // An extra async wrapper could still be pending after the scheduler has drained its queue. + Assert.That(task, Is.SameAs(scheduler.ScheduledTask)); + Assert.That(task.IsCompleted, Is.False); + scheduler.Complete!(); + Assert.That(task.IsCompletedSuccessfully, Is.True); + } + + [TestCaseSource(nameof(ExtensionOverloads))] + public void Enqueue_InvalidScheduler_ShouldThrowSynchronously(string overload) + { + Task? task = null; + Assert.That(() => { task = Enqueue(null!, overload); }, Throws.TypeOf()); + Assert.That(task, Is.Null); + } + + [TestCaseSource(typeof(TaskSchedulerDisposalConformance), nameof(TaskSchedulerDisposalConformance.Overloads))] + public void Enqueue_SchedulingFailure_ShouldThrowSynchronously(string overload) + { + var failure = new ArgumentException("Scheduling rejected"); + var scheduler = new ThrowingScheduler(failure); + Task? task = null; + Assert.That(() => { task = Enqueue(scheduler, overload); }, Throws.Exception.SameAs(failure)); + Assert.That(task, Is.Null); + } + + private static Task Enqueue(ITaskScheduler scheduler, string overload) + => TaskSchedulerDisposalConformance.Enqueue(scheduler, overload, () => { }, false); + + private sealed class ThrowingScheduler(Exception failure) : ITaskScheduler + { + public Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + { + throw failure; + } + } + + private sealed class PendingScheduler : ITaskScheduler + { + public Task? ScheduledTask { get; private set; } + + public Action? Complete { get; private set; } + + public Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + { + var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + ScheduledTask = completion.Task; + Complete = () => completion.SetResult(default!); + return completion.Task; + } + } +} \ No newline at end of file diff --git a/TaskFlow.Tests/TaskSchedulerForwardingConformanceFixture.cs b/TaskFlow.Tests/TaskSchedulerForwardingConformanceFixture.cs new file mode 100644 index 0000000..093f834 --- /dev/null +++ b/TaskFlow.Tests/TaskSchedulerForwardingConformanceFixture.cs @@ -0,0 +1,385 @@ +namespace TaskFlow.Tests; + +using Microsoft.Extensions.Time.Testing; +using NUnit.Framework; +using System.Threading.Tasks.Flow; + +[TestFixture] +[CancelAfter(10000)] +internal sealed class TaskSchedulerForwardingConformanceFixture +{ + [TestCase("Annotated", "disposed", "DirectGeneric")] + [TestCase("Forwarding", "disposed", "DirectGeneric")] + [TestCase("Execution", "disposed", "Task")] + [TestCase("Completion", "disposed", "Action")] + [TestCase("Nested", "disposed", "ValueTask")] + [TestCase("Owned", "disposed", "DirectGeneric")] + [TestCase("Throttle", "disposed", "DirectGeneric")] + [TestCase("Forwarding", "argument", "Action")] + [TestCase("Forwarding", "failure", "Task")] + [TestCase("Forwarding", "cancellation", "ValueTask")] + public async Task Enqueue_ImmediateTerminalRejection_ShouldThrowSynchronously( + string configuration, string failureType, string overload) + { + var failure = CreateFailure(failureType); + var terminal = new Terminal(failure, throwSynchronously: true); + await using var lifetime = new TaskSchedulerMiddlewareLifetime(); + using var observer = new Observer(); + var scheduler = Configure(terminal, configuration, lifetime, observer); + Task? returned = null; + + Assert.That(() => { returned = TaskSchedulerDisposalConformance.Enqueue(scheduler, overload, () => { }, false); }, Throws.Exception.SameAs(failure)); + Assert.That(returned, Is.Null); + Assert.That(terminal.Calls, Is.EqualTo(1)); + Assert.That(observer.Completions, Is.Zero, "Immediate scheduling rejection must reach the caller before a task is returned."); + + await lifetime.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5)); + if (configuration == "Owned") + Assert.That(observer.Disposed, Is.True, "Rejected forwarding must release the pipeline's lifetime entry."); + } + + [TestCase("Forwarding", "disposed", "DirectGeneric")] + [TestCase("Completion", "disposed", "Action")] + [TestCase("Owned", "argument", "DirectGeneric")] + [TestCase("CancellationScope", "failure", "Task")] + [TestCase("CancelPrevious", "cancellation", "ValueTask")] + [TestCase("Timeout", "disposed", "DirectGeneric")] + public async Task Enqueue_ReturnedFailureTask_ShouldStayTaskBased( + string configuration, string failureType, string overload) + { + var failure = CreateFailure(failureType); + var terminal = new Terminal(failure, throwSynchronously: false); + await using var lifetime = new TaskSchedulerMiddlewareLifetime(); + using var observer = new Observer(); + var scheduler = Configure(terminal, configuration, lifetime, observer); + Task? returned = null; + + Assert.That(() => { returned = TaskSchedulerDisposalConformance.Enqueue(scheduler, overload, () => { }, false); }, Throws.Nothing); + await Assert.ThatAsync(async () => await returned!, Throws.Exception.SameAs(failure)); + Assert.That(terminal.Calls, Is.EqualTo(1)); + Assert.That(observer.Completions, Is.EqualTo(configuration is "Completion" or "Owned" ? 1 : 0)); + } + + [TestCase("Timeout")] + [TestCase("CancellationScope")] + [TestCase("CancelPrevious")] + public async Task Enqueue_ExistingAsyncWrapper_ShouldReportImmediateTerminalRejectionThroughTask(string configuration) + { + var failure = new ObjectDisposedException("inner"); + var terminal = new Terminal(failure, throwSynchronously: true); + using var observer = new Observer(); + await using var lifetime = new TaskSchedulerMiddlewareLifetime(); + var scheduler = Configure(terminal.UseMiddleware((ITaskSchedulerCompletionMiddleware)observer), configuration, lifetime, observer); + Task? returned = null; + + Assert.That(() => { returned = scheduler.Enqueue(() => 42); }, Throws.Nothing); + await Assert.ThatAsync(async () => await returned!, Throws.Exception.SameAs(failure)); + Assert.That(terminal.Calls, Is.EqualTo(1)); + Assert.That(observer.Completions, Is.EqualTo(1)); + } + + [TestCase("CancellationScope", true)] + [TestCase("CancellationScope", false)] + [TestCase("CancelPrevious", true)] + [TestCase("CancelPrevious", false)] + public async Task Enqueue_CancellationResources_ShouldLiveUntilCompletionOrRejection(string configuration, bool reject) + { + var failure = new ObjectDisposedException("inner"); + var terminal = new TokenCapturingTerminal(failure, reject); + using var caller = new CancellationTokenSource(); + var scheduler = configuration == "CancellationScope" + ? terminal.CreateCancellationScope(CancellationToken.None) + : terminal.CreateCancelPrevious(); + try + { + if (reject) + { + Task? pending = null; + Assert.That(() => { pending = scheduler.Enqueue((_, _) => new ValueTask(42), null, caller.Token); }, Throws.Nothing); + await Assert.ThatAsync(async () => await pending!, Throws.Exception.SameAs(failure)); + } + else + { + var pending = scheduler.Enqueue((_, _) => new ValueTask(42), null, caller.Token); + Assert.That(() => terminal.Token.WaitHandle, Throws.Nothing, "Accepted work must retain its linked source."); + await caller.CancelAsync(); + Assert.That(terminal.Token.IsCancellationRequested, Is.True); + terminal.Release.SetResult(); + await pending; + } + Assert.That(() => terminal.Token.WaitHandle, Throws.TypeOf(), "The linked source must be released."); + } + finally { terminal.Release.TrySetResult(); } + } + + [Test] + public async Task Enqueue_ShortCircuit_ShouldNotProbeOrCallRejectedTerminal() + { + var terminal = new Terminal(new ObjectDisposedException("inner"), throwSynchronously: true); + var scheduler = terminal.UseMiddleware(new ConstantAdmission()); + Assert.That(await scheduler.Enqueue(() => 42), Is.EqualTo(17)); + Assert.That(terminal.Calls, Is.Zero); + } + + [Test] + public async Task Enqueue_DeferredForwarding_ShouldReportLaterTerminalRejectionThroughTask() + { + var failure = new ObjectDisposedException("inner"); + var terminal = new Terminal(failure, throwSynchronously: true); + var admission = new DeferredAdmission(); + using var observer = new Observer(); + var scheduler = terminal.UseMiddleware((ITaskSchedulerCompletionMiddleware)observer).UseMiddleware(admission); + Task? returned = null; + Assert.That(() => { returned = scheduler.Enqueue((_, _) => new ValueTask(42), null, CancellationToken.None); }, Throws.Nothing); + Assert.That(returned!.IsCompleted, Is.False); + Assert.That(terminal.Calls, Is.Zero); + + admission.Release.SetResult(); + await Assert.ThatAsync(async () => await returned, Throws.Exception.SameAs(failure)); + Assert.That(terminal.Calls, Is.EqualTo(1)); + Assert.That(observer.Completions, Is.EqualTo(1)); + } + + [Test] + public async Task Enqueue_ClosedOwnLifetime_ShouldRejectWithoutCallingAvailableTerminal() + { + var terminal = new Terminal(new InvalidOperationException("terminal must not be reached"), throwSynchronously: true); + await using var lifetime = new TaskSchedulerMiddlewareLifetime(); + using var observer = new Observer(); + var scheduler = terminal.UseMiddleware(new Factory(observer), lifetime); + await lifetime.DisposeAsync(); + Assert.That(() => scheduler.Enqueue(() => 42), Throws.TypeOf()); + Assert.That(terminal.Calls, Is.Zero); + Assert.That(observer.Forwards, Is.Zero); + Assert.That(observer.Disposed, Is.True); + } + + [Test] + public async Task Enqueue_SynchronousFailureAfterForwarding_ShouldRetainAcceptedChildUntilQuiescent() + { + await using var lifetime = new TaskSchedulerMiddlewareLifetime(); + using var observer = new Observer(); + var failure = new InvalidOperationException("after forwarding"); + var forwarding = new ThrowAfterForward(failure); + var scheduler = new TaskFlowSchedulerAdapter(TaskScheduler.Default) + .UseMiddleware(new Factory(observer), lifetime).UseMiddleware(forwarding); + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + async ValueTask Work(object? state, CancellationToken token) + { + entered.SetResult(); + await release.Task.ConfigureAwait(false); + return 42; + } + try + { + Assert.That(() => scheduler.Enqueue(Work, null, CancellationToken.None), Throws.Exception.SameAs(failure)); + await entered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + var shutdown = lifetime.DisposeAsync().AsTask(); + Assert.That(shutdown.IsCompleted, Is.False); + Assert.That(observer.Disposed, Is.False); + release.SetResult(); + await forwarding.Child!.WaitAsync(TimeSpan.FromSeconds(5)); + await shutdown.WaitAsync(TimeSpan.FromSeconds(5)); + Assert.That(observer.Completions, Is.EqualTo(1)); + Assert.That(observer.Disposed, Is.True); + } + finally + { + release.TrySetResult(); + if (forwarding.Child != null) + await forwarding.Child.WaitAsync(TimeSpan.FromSeconds(5)); + } + } + + [Test] + public async Task Enqueue_Timeout_ShouldIncludeTimeSpentForwarding() + { + var clock = new FakeTimeProvider(); + var terminal = new TimeAdvancingTerminal(clock); + var scheduler = terminal.WithTimeout(TimeSpan.FromSeconds(30), clock); + try + { + var task = scheduler.Enqueue(() => 42); + await Assert.ThatAsync(async () => await task.WaitAsync(TimeSpan.FromSeconds(5)), + Throws.TypeOf().With.Message.Contains("Operation has timed out in")); + } + finally { terminal.Release.TrySetResult(); } + } + + [Test] + public async Task Enqueue_Timeout_ShouldRetainAcceptedExecutionUntilLifetimeDrains() + { + await using var flow = new TaskFlow(); + using var observer = new Observer(); + var clock = new FakeTimeProvider(); + var scheduler = flow.UseMiddleware(new Factory(observer)).WithTimeout(TimeSpan.FromSeconds(30), clock); + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + try + { + var task = scheduler.Enqueue(async (_, _) => + { + entered.SetResult(); + await release.Task.ConfigureAwait(false); + return 42; + }, null, CancellationToken.None); + await entered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + clock.Advance(TimeSpan.FromSeconds(31)); + await Assert.ThatAsync(async () => await task, Throws.TypeOf()); + Assert.That(observer.Completions, Is.EqualTo(1)); + var shutdown = flow.DisposeAsync().AsTask(); + Assert.That(shutdown.IsCompleted, Is.False); + Assert.That(observer.Disposed, Is.False); + release.SetResult(); + await shutdown.WaitAsync(TimeSpan.FromSeconds(5)); + Assert.That(observer.Completions, Is.EqualTo(1)); + Assert.That(observer.Disposed, Is.True); + } + finally + { + release.TrySetResult(); + await flow.DisposeAsync(); + } + } + + private static ITaskScheduler Configure(ITaskScheduler terminal, string configuration, + TaskSchedulerMiddlewareLifetime lifetime, Observer observer) + => configuration switch + { + "Annotated" => terminal.WithOperationName("forwarding"), + "Forwarding" => terminal.UseMiddleware(new Forwarding()).UseMiddleware(new Forwarding()), + "Execution" => terminal.UseMiddleware(new Execution()), + "Completion" => terminal.UseMiddleware((ITaskSchedulerCompletionMiddleware)observer), + "Nested" => new Opaque(terminal.UseMiddleware(new Forwarding())).UseMiddleware(new Forwarding()), + "Owned" => terminal.UseMiddleware(new Factory(observer), lifetime), + "Throttle" => terminal.WithThrottle(TimeSpan.FromHours(1)), + "CancellationScope" => terminal.CreateCancellationScope(CancellationToken.None), + "CancelPrevious" => terminal.CreateCancelPrevious(), + "Timeout" => terminal.WithTimeout(TimeSpan.FromHours(1)), + _ => throw new ArgumentOutOfRangeException(nameof(configuration)), + }; + + private static Exception CreateFailure(string failureType) + => failureType switch + { + "disposed" => new ObjectDisposedException("inner"), + "argument" => new ArgumentException("rejected"), + "failure" => new InvalidOperationException("rejected"), + "cancellation" => new OperationCanceledException(new CancellationToken(canceled: true)), + _ => throw new ArgumentOutOfRangeException(nameof(failureType)), + }; + + private sealed class Terminal(Exception failure, bool throwSynchronously) : ITaskScheduler + { + public int Calls { get; private set; } + public Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + { + Calls++; + if (throwSynchronously) + throw failure; + return Task.FromException(failure); + } + } + + private sealed class TokenCapturingTerminal(Exception failure, bool reject) : ITaskScheduler + { + public CancellationToken Token { get; private set; } + public TaskCompletionSource Release { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + { + Token = cancellationToken; + if (reject) + throw failure; + return AwaitRelease(); + } + private async Task AwaitRelease() + { + await Release.Task.ConfigureAwait(false); + return default!; + } + } + + private sealed class TimeAdvancingTerminal(FakeTimeProvider clock) : ITaskScheduler + { + public TaskCompletionSource Release { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + { + clock.Advance(TimeSpan.FromSeconds(31)); + return AwaitRelease(); + } + private async Task AwaitRelease() + { + await Release.Task.ConfigureAwait(false); + return default!; + } + } + + private sealed class Opaque(ITaskScheduler inner) : ITaskScheduler + { + public Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + => inner.Enqueue(taskFunc, state, cancellationToken); + } + + private sealed class Forwarding : ITaskSchedulerEnqueueMiddleware + { + public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate next) => next(context); + } + + private sealed class Execution : ITaskSchedulerExecutionMiddleware + { + public ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate next) => next(context); + } + + private sealed class ConstantAdmission : ITaskSchedulerEnqueueMiddleware + { + public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate next) + => Task.FromResult((T)(object)17); + } + + private sealed class ThrowAfterForward(Exception failure) : ITaskSchedulerEnqueueMiddleware + { + public Task? Child { get; private set; } + public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate next) + { + Child = next(context); + throw failure; + } + } + + private sealed class DeferredAdmission : ITaskSchedulerEnqueueMiddleware + { + public TaskCompletionSource Release { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public async Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate next) + { + await Release.Task.ConfigureAwait(false); + return await next(context).ConfigureAwait(false); + } + } + + private sealed class Observer : ITaskSchedulerEnqueueMiddleware, ITaskSchedulerCompletionMiddleware, IDisposable + { + public int Forwards { get; private set; } + public int Completions { get; private set; } + public bool Disposed { get; private set; } + public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate next) + { + ObjectDisposedException.ThrowIf(Disposed, this); + Forwards++; + return next(context); + } + public ValueTask> InvokeAsync(TaskSchedulerOperationContext context, + TaskSchedulerOperationOutcome outcome, TaskSchedulerCompletionDelegate next) + { + ObjectDisposedException.ThrowIf(Disposed, this); + Completions++; + return next(context, outcome); + } + public void Dispose() => Disposed = true; + } + + private sealed class Factory(Observer observer) : ITaskSchedulerMiddlewareFactory + { + public ITaskSchedulerMiddleware Create(TaskSchedulerMiddlewareLifetime lifetime) => observer; + } +} \ No newline at end of file diff --git a/TaskFlow.Tests/ThreadTaskFlowDisposalFixture.cs b/TaskFlow.Tests/ThreadTaskFlowDisposalFixture.cs new file mode 100644 index 0000000..f61ce46 --- /dev/null +++ b/TaskFlow.Tests/ThreadTaskFlowDisposalFixture.cs @@ -0,0 +1,104 @@ +namespace TaskFlow.Tests; + +using NUnit.Framework; +using System.Threading.Tasks.Flow; + +[TestFixture] +[CancelAfter(10000)] +internal sealed class ThreadTaskFlowDisposalFixture +{ + private static readonly int[] ExpectedExecutionOrder = [1, 4, 2, 3]; + + [TestCase(false, false)] + [TestCase(false, true)] + [TestCase(true, false)] + [TestCase(true, true)] + public async Task DisposeAsync_ShouldSettleCallerTasksBeforeReturning(bool currentThread, bool asynchronous) + { + await using var sut = CreateFlow(currentThread); + var started = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var lastStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var lastRelease = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var cancellation = new CancellationTokenSource(); + var failure = new InvalidOperationException("Queued failure"); + var executed = new List(); + var observedCancellation = CancellationToken.None; + var first = sut.Enqueue(async _ => + { + started.SetResult(); + await release.Task; + }); + + Task Result(object? state, CancellationToken token) => RunAsync(1, token); + Task Fault(CancellationToken token) => RunAsync(2, token); + async ValueTask Canceled(CancellationToken token) + { + observedCancellation = token; + await RunAsync(3, token); + } + async Task RunAsync(int value, CancellationToken token) + { + if (asynchronous) + await Task.Yield(); + executed.Add(value); + if (value == 2) + throw failure; + if (value == 3) + token.ThrowIfCancellationRequested(); + return value; + } + + try + { + await started.Task.WaitAsync(TimeSpan.FromSeconds(5)); + var result = sut.Enqueue((state, token) => new ValueTask(Result(state, token)), null, CancellationToken.None); + var action = sut.Enqueue(() => executed.Add(4)); + var fault = sut.Enqueue(Fault); + var canceled = sut.Enqueue(Canceled, cancellation.Token, null); + var last = sut.Enqueue(async _ => + { + lastStarted.SetResult(); + await lastRelease.Task; + }); + await cancellation.CancelAsync(); + + var disposal = sut.DisposeAsync().AsTask(); + Assert.That(disposal.IsCompleted, Is.False); + release.SetResult(); + await lastStarted.Task.WaitAsync(TimeSpan.FromSeconds(5)); + Assert.That(disposal.IsCompleted, Is.False, "Disposal must wait for the last asynchronous operation."); + lastRelease.SetResult(); + await disposal.WaitAsync(TimeSpan.FromSeconds(5)); + + using (Assert.EnterMultipleScope()) + { + Assert.That(first.IsCompletedSuccessfully, Is.True); + Assert.That(result.IsCompletedSuccessfully, Is.True); + Assert.That(action.IsCompletedSuccessfully, Is.True); + Assert.That(fault.IsFaulted, Is.True); + Assert.That(canceled.IsCanceled, Is.True); + Assert.That(last.IsCompletedSuccessfully, Is.True); + Assert.That(executed, Is.EqualTo(ExpectedExecutionOrder)); + } + Assert.That(await result, Is.EqualTo(1)); + await Assert.ThatAsync(async () => await fault, Throws.Exception.SameAs(failure)); + await Assert.ThatAsync(async () => await canceled, + Throws.InstanceOf().With.Property("CancellationToken").EqualTo(observedCancellation)); + } + finally + { + release.TrySetResult(); + lastRelease.TrySetResult(); + } + } + + private static ThreadTaskFlow CreateFlow(bool currentThread) + { + if (!currentThread) + return new DedicatedThreadTaskFlow(); + var flow = new CurrentThreadTaskFlow(); + new Thread(flow.Run) { IsBackground = true }.Start(); + return flow; + } +} \ No newline at end of file diff --git a/TaskFlow/Extensions/ExtendedStateTaskSchedulerExtensions.cs b/TaskFlow/Extensions/ExtendedStateTaskSchedulerExtensions.cs index 5f360a2..daed38a 100644 --- a/TaskFlow/Extensions/ExtendedStateTaskSchedulerExtensions.cs +++ b/TaskFlow/Extensions/ExtendedStateTaskSchedulerExtensions.cs @@ -22,11 +22,13 @@ public ExtendedStateTaskSchedulerWrapper(ITaskScheduler baseTaskScheduler, TStat _extendedState = extendedState; } - public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + public Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) { + Argument.NotNull(taskFunc); + var extendedState = new ExtendedState(state, _extendedState); - return await _baseTaskScheduler.Enqueue(Action, extendedState, cancellationToken).ConfigureAwait(false); + return _baseTaskScheduler.Enqueue(Action, extendedState, cancellationToken); async ValueTask Action(ExtendedState s, CancellationToken token) { diff --git a/TaskFlow/Extensions/ThrottlingTaskSchedulerExtensions.cs b/TaskFlow/Extensions/ThrottlingTaskSchedulerExtensions.cs index 610bd98..2d44ae5 100644 --- a/TaskFlow/Extensions/ThrottlingTaskSchedulerExtensions.cs +++ b/TaskFlow/Extensions/ThrottlingTaskSchedulerExtensions.cs @@ -161,7 +161,7 @@ public Task InvokeAsync(TaskSchedulerEnqueueContext c var elapsed = _timeProvider.GetElapsedTime(_lastAdmissionTimestamp, currentTimestamp); if (_hasAdmission && elapsed < _interval) { - throw new OperationThrottledException($"Operation did not execute because the throttle interval has not elapsed. Interval: {_interval}. Elapsed: {elapsed}."); + return Task.FromException(new OperationThrottledException($"Operation did not execute because the throttle interval has not elapsed. Interval: {_interval}. Elapsed: {elapsed}.")); } _lastAdmissionTimestamp = currentTimestamp; diff --git a/TaskFlow/Internal/TaskSchedulerPipeline.cs b/TaskFlow/Internal/TaskSchedulerPipeline.cs index c77db1a..eb6dc09 100644 --- a/TaskFlow/Internal/TaskSchedulerPipeline.cs +++ b/TaskFlow/Internal/TaskSchedulerPipeline.cs @@ -99,19 +99,10 @@ internal Task Submit(Func(taskFunc, state, _registrations.Length, _annotations, cancellationToken), request); } - [SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "Pipeline submission reports terminal failures through its returned task.")] - private Task EnqueueTerminal(Func> taskFunc, - object? state, CancellationToken cancellationToken) - { - try - { return _terminal.Enqueue(taskFunc, state, cancellationToken); } - catch (Exception exception) { return EnqueueFailure(exception); } - } - internal Task SubmitKeyed(TKey key, Func> taskFunc, object? state, AdmissionRequest? request, CancellationToken cancellationToken) where TKey : notnull { @@ -124,24 +115,48 @@ private Task SubmitOperation(PipelineOperation operat { operation.AdmissionRequest = request; if (request != null) + { operation.OnQuiesced(request.OnQuiesced); - return EnqueueCore(operation); + return EnqueueCore(operation); + } + // Plain enqueue invokes forwarding before handing ownership to the task observer. + Task pending; + try + { + if (Lifetime != null) + { + Lifetime.Enter(); + operation.Lifetime = Lifetime; + } + pending = InvokeEnqueueCore(operation, _registrations.Length - 1, operation.CallerCancellationToken); + Annotations.Argument.NotNull(pending); + } + catch + { + operation.ReleaseActivity(); + throw; + } + return EnqueueCore(operation, pending); } - private async Task EnqueueCore(PipelineOperation operation) + private async Task EnqueueCore(PipelineOperation operation, Task? pending = null) { try { TResult result; try { - if (Lifetime != null) + if (pending == null) { - Lifetime.Enter(); - operation.Lifetime = Lifetime; + if (Lifetime != null) + { + Lifetime.Enter(); + operation.Lifetime = Lifetime; + } + // Explicit submissions report admission and scheduling failures through their tasks. + pending = InvokeEnqueueCore(operation, _registrations.Length - 1, operation.CallerCancellationToken); } - // The root activity already covers this invocation, including completion middleware. - result = await InvokeEnqueueCore(operation, _registrations.Length - 1, operation.CallerCancellationToken).ConfigureAwait(false); + result = await pending.ConfigureAwait(false); } catch (Exception exception) { @@ -161,16 +176,28 @@ private async Task EnqueueCore(PipelineOperation oper finally { operation.ReleaseActivity(); } } - private async Task InvokeEnqueue(PipelineOperation operation, int index, CancellationToken cancellationToken) + private Task InvokeEnqueue(PipelineOperation operation, int index, CancellationToken cancellationToken) { operation.AddActivity(); + Task pending; try - { return await InvokeEnqueueCore(operation, index, cancellationToken).ConfigureAwait(false); } - finally { operation.ReleaseActivity(); } - } + { + pending = InvokeEnqueueCore(operation, index, cancellationToken); + } + catch + { + operation.ReleaseActivity(); + throw; + } + return AwaitActivity(operation, pending); - private static async Task EnqueueFailure(Exception exception) - => await Task.FromException(exception).ConfigureAwait(false); + static async Task AwaitActivity(PipelineOperation operation, Task pending) + { + try + { return await pending.ConfigureAwait(false); } + finally { operation.ReleaseActivity(); } + } + } private Task InvokeEnqueueCore(PipelineOperation operation, int index, CancellationToken cancellationToken) { diff --git a/TaskFlow/Middleware/ITaskSchedulerEnqueueMiddleware.cs b/TaskFlow/Middleware/ITaskSchedulerEnqueueMiddleware.cs index cad97be..2d38795 100644 --- a/TaskFlow/Middleware/ITaskSchedulerEnqueueMiddleware.cs +++ b/TaskFlow/Middleware/ITaskSchedulerEnqueueMiddleware.cs @@ -19,7 +19,9 @@ public interface ITaskSchedulerEnqueueMiddleware : ITaskSchedulerMiddleware /// /// Return a completed, failed, or canceled task without calling to satisfy or reject the request without /// submitting a terminal delegate. Do not call the continuation more than once. - /// Exceptions returned or thrown by this method enter the completion pipeline even though terminal execution has not started. + /// Failures reported through the returned task enter the completion pipeline even though terminal execution has not started. + /// Exceptions thrown by this method before it returns a task propagate from enqueue without being converted into completion outcomes. + /// Middleware and wrappers that return their own tasks retain their exception timing and resource ownership. /// Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation); } \ No newline at end of file diff --git a/TaskFlow/TaskSchedulerEnqueueExtensions.cs b/TaskFlow/TaskSchedulerEnqueueExtensions.cs index 33df8b2..6f78931 100644 --- a/TaskFlow/TaskSchedulerEnqueueExtensions.cs +++ b/TaskFlow/TaskSchedulerEnqueueExtensions.cs @@ -16,6 +16,10 @@ namespace System.Threading.Tasks.Flow; /// synchronous functions, and actions) and provide convenient overloads for common scenarios. /// /// +/// Argument validation and scheduler rejection exceptions, including , +/// are propagated synchronously. Failures from accepted operations remain observable through their returned tasks. +/// +/// /// All extension methods perform argument validation and ensure proper exception propagation. /// Methods that don't specify a will use . /// @@ -73,7 +77,7 @@ public static class TaskSchedulerEnqueueExtensions /// The function to execute that accepts a cancellation token and returns a . /// A cancellation token that can be used to cancel the operation. /// A representing the result of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method adapts a function that takes only a cancellation token to the scheduler's signature @@ -84,6 +88,7 @@ public static class TaskSchedulerEnqueueExtensions public static Task Enqueue(this ITaskScheduler taskScheduler, Func> taskFunc, CancellationToken cancellationToken) { Argument.NotNull(taskScheduler); + Argument.NotNull(taskFunc); return taskScheduler.Enqueue(TaskFunc, null, cancellationToken); @@ -101,7 +106,7 @@ ValueTask TaskFunc(object? state, CancellationToken token) /// The function to execute that accepts a cancellation token and returns a . /// A cancellation token that can be used to cancel the operation. /// A representing the result of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method wraps the returned by the function in a @@ -112,6 +117,7 @@ ValueTask TaskFunc(object? state, CancellationToken token) public static Task Enqueue(this ITaskScheduler taskScheduler, Func> taskFunc, CancellationToken cancellationToken) { Argument.NotNull(taskScheduler); + Argument.NotNull(taskFunc); return taskScheduler.Enqueue(TaskFunc, null, cancellationToken); @@ -131,7 +137,7 @@ ValueTask TaskFunc(object? state, CancellationToken token) /// The state object to pass to the function. /// A cancellation token that can be used to cancel the operation. /// A representing the result of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method adapts a strongly-typed state parameter function to the scheduler's object-based state signature @@ -140,6 +146,7 @@ ValueTask TaskFunc(object? state, CancellationToken token) public static Task Enqueue(this ITaskScheduler taskScheduler, Func> taskFunc, TState state, CancellationToken cancellationToken) { Argument.NotNull(taskScheduler); + Argument.NotNull(taskFunc); return taskScheduler.Enqueue(TaskFunc, state, cancellationToken); @@ -158,7 +165,7 @@ ValueTask TaskFunc(object? s, CancellationToken token) /// The state object to pass to the function. /// A cancellation token that can be used to cancel the operation. /// A representing the completion of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method wraps the non-generic in a with a @@ -167,6 +174,7 @@ ValueTask TaskFunc(object? s, CancellationToken token) public static Task Enqueue(this ITaskScheduler taskScheduler, Func taskFunc, TState state, CancellationToken cancellationToken) { Argument.NotNull(taskScheduler); + Argument.NotNull(taskFunc); return taskScheduler.Enqueue(TaskFunc, state, cancellationToken); @@ -187,7 +195,7 @@ async ValueTask TaskFunc(object? s, CancellationToken token) /// The state object to pass to the function. /// A cancellation token that can be used to cancel the operation. /// A representing the result of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method wraps the returned by the function in a @@ -196,6 +204,7 @@ async ValueTask TaskFunc(object? s, CancellationToken token) public static Task Enqueue(this ITaskScheduler taskScheduler, Func> taskFunc, TState state, CancellationToken cancellationToken) { Argument.NotNull(taskScheduler); + Argument.NotNull(taskFunc); return taskScheduler.Enqueue(TaskFunc, state, cancellationToken); @@ -214,7 +223,7 @@ ValueTask TaskFunc(object? s, CancellationToken token) /// The state object to pass to the function. /// A cancellation token that can be used to cancel the operation. /// A representing the completion of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method wraps the non-generic in a to match the scheduler's @@ -223,6 +232,7 @@ ValueTask TaskFunc(object? s, CancellationToken token) public static Task Enqueue(this ITaskScheduler taskScheduler, Func taskFunc, TState state, CancellationToken cancellationToken) { Argument.NotNull(taskScheduler); + Argument.NotNull(taskFunc); return taskScheduler.Enqueue(TaskFunc, (object?)state, cancellationToken); @@ -239,19 +249,20 @@ ValueTask TaskFunc(object? s, CancellationToken token) /// The function to execute that accepts a cancellation token and returns a . /// A cancellation token that can be used to cancel the operation. /// A representing the completion of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method wraps the non-generic in a with a /// result to match the scheduler's signature requirement. /// - public static async Task Enqueue(this ITaskScheduler taskScheduler, Func taskFunc, CancellationToken cancellationToken) + public static Task Enqueue(this ITaskScheduler taskScheduler, Func taskFunc, CancellationToken cancellationToken) { Argument.NotNull(taskScheduler); + Argument.NotNull(taskFunc); - await taskScheduler.Enqueue(TaskFunc, cancellationToken).ConfigureAwait(false); + return taskScheduler.Enqueue(TaskFunc, null, cancellationToken); - async ValueTask TaskFunc(CancellationToken token) + async ValueTask TaskFunc(object? state, CancellationToken token) { await taskFunc(token).ConfigureAwait(false); return default; @@ -264,16 +275,14 @@ async ValueTask TaskFunc(CancellationToken token) /// The task scheduler to enqueue the operation on. /// The function to execute that accepts a cancellation token and returns a . /// A representing the completion of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method uses for the cancellation token. /// - public static async Task Enqueue(this ITaskScheduler taskScheduler, Func taskFunc) + public static Task Enqueue(this ITaskScheduler taskScheduler, Func taskFunc) { - Argument.NotNull(taskScheduler); - - await taskScheduler.Enqueue(taskFunc, CancellationToken.None).ConfigureAwait(false); + return taskScheduler.Enqueue(taskFunc, CancellationToken.None); } /// @@ -283,7 +292,7 @@ public static async Task Enqueue(this ITaskScheduler taskScheduler, FuncThe task scheduler to enqueue the operation on. /// The function to execute that accepts a cancellation token and returns a . /// A representing the result of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method uses for the cancellation token and wraps the @@ -294,6 +303,7 @@ public static async Task Enqueue(this ITaskScheduler taskScheduler, Func Enqueue(this ITaskScheduler taskScheduler, Func> taskFunc) { Argument.NotNull(taskScheduler); + Argument.NotNull(taskFunc); return taskScheduler.Enqueue(token => new ValueTask(taskFunc(token)), CancellationToken.None); } @@ -305,7 +315,7 @@ public static Task Enqueue(this ITaskScheduler taskScheduler, FuncThe task scheduler to enqueue the operation on. /// The function to execute that returns a . /// A representing the result of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method uses for the cancellation token and ignores the @@ -314,6 +324,7 @@ public static Task Enqueue(this ITaskScheduler taskScheduler, Func Enqueue(this ITaskScheduler taskScheduler, Func> taskFunc) { Argument.NotNull(taskScheduler); + Argument.NotNull(taskFunc); return taskScheduler.Enqueue(_ => new ValueTask(taskFunc()), CancellationToken.None); } @@ -324,7 +335,7 @@ public static Task Enqueue(this ITaskScheduler taskScheduler, Func /// The task scheduler to enqueue the operation on. /// The function to execute that returns a . /// A representing the completion of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method uses for the cancellation token and ignores the @@ -333,6 +344,7 @@ public static Task Enqueue(this ITaskScheduler taskScheduler, Func public static Task Enqueue(this ITaskScheduler taskScheduler, Func taskFunc) { Argument.NotNull(taskScheduler); + Argument.NotNull(taskFunc); return taskScheduler.Enqueue(_ => taskFunc(), CancellationToken.None); } @@ -345,7 +357,7 @@ public static Task Enqueue(this ITaskScheduler taskScheduler, Func taskFun /// A cancellation token that can be used to cancel the operation. /// Dummy parameter to disambiguate from other overloads. Should always be null. /// A representing the completion of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method wraps the non-generic in a with a @@ -353,13 +365,14 @@ public static Task Enqueue(this ITaskScheduler taskScheduler, Func taskFun /// to avoid method signature conflicts with other overloads. /// [SuppressMessage("ApiDesign", "RS0027:Public API with optional parameter(s) should have the most parameters amongst its public overloads", Justification = "The released optional dummy parameter disambiguates Task and ValueTask lambdas and cannot be removed without a source break.")] - public static async Task Enqueue(this ITaskScheduler taskScheduler, Func valueTaskFunc, CancellationToken cancellationToken, DummyParameter? _ = null) + public static Task Enqueue(this ITaskScheduler taskScheduler, Func valueTaskFunc, CancellationToken cancellationToken, DummyParameter? _ = null) { Argument.NotNull(taskScheduler); + Argument.NotNull(valueTaskFunc); - await taskScheduler.Enqueue(TaskFunc, cancellationToken).ConfigureAwait(false); + return taskScheduler.Enqueue(TaskFunc, null, cancellationToken); - async ValueTask TaskFunc(CancellationToken token) + async ValueTask TaskFunc(object? state, CancellationToken token) { await valueTaskFunc(token).ConfigureAwait(false); return default; @@ -373,7 +386,7 @@ async ValueTask TaskFunc(CancellationToken token) /// The function to execute that accepts a cancellation token and returns a . /// Dummy parameter to disambiguate from other overloads. Should always be null. /// A representing the completion of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method uses for the cancellation token. The dummy parameter @@ -382,11 +395,9 @@ async ValueTask TaskFunc(CancellationToken token) /// overload; use a named function or an explicit delegate cast. /// [SuppressMessage("ApiDesign", "RS0027:Public API with optional parameter(s) should have the most parameters amongst its public overloads", Justification = "The released optional dummy parameter disambiguates Task and ValueTask lambdas and cannot be removed without a source break.")] - public static async Task Enqueue(this ITaskScheduler taskScheduler, Func valueTaskFunc, DummyParameter? _ = null) + public static Task Enqueue(this ITaskScheduler taskScheduler, Func valueTaskFunc, DummyParameter? _ = null) { - Argument.NotNull(taskScheduler); - - await taskScheduler.Enqueue(valueTaskFunc, CancellationToken.None).ConfigureAwait(false); + return taskScheduler.Enqueue(valueTaskFunc, CancellationToken.None, _); } /// @@ -397,7 +408,7 @@ public static async Task Enqueue(this ITaskScheduler taskScheduler, FuncThe function to execute that accepts a cancellation token and returns a . /// Dummy parameter to disambiguate from other overloads. Should always be null. /// A representing the result of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method uses for the cancellation token. The dummy parameter @@ -407,6 +418,7 @@ public static async Task Enqueue(this ITaskScheduler taskScheduler, Func Enqueue(this ITaskScheduler taskScheduler, Func> valueTaskFunc, DummyParameter? _ = null) { Argument.NotNull(taskScheduler); + Argument.NotNull(valueTaskFunc); return taskScheduler.Enqueue(valueTaskFunc, CancellationToken.None); } @@ -419,7 +431,7 @@ public static Task Enqueue(this ITaskScheduler taskScheduler, FuncThe function to execute that returns a . /// Dummy parameter to disambiguate from other overloads. Should always be null. /// A representing the result of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method uses for the cancellation token and ignores the @@ -430,6 +442,7 @@ public static Task Enqueue(this ITaskScheduler taskScheduler, Func Enqueue(this ITaskScheduler taskScheduler, Func> valueTaskFunc, DummyParameter? _ = null) { Argument.NotNull(taskScheduler); + Argument.NotNull(valueTaskFunc); return taskScheduler.Enqueue(_ => valueTaskFunc(), CancellationToken.None); } @@ -441,7 +454,7 @@ public static Task Enqueue(this ITaskScheduler taskScheduler, FuncThe function to execute that returns a . /// Dummy parameter to disambiguate from other overloads. Should always be null. /// A representing the completion of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method uses for the cancellation token and ignores the @@ -452,6 +465,7 @@ public static Task Enqueue(this ITaskScheduler taskScheduler, Func valueTaskFunc, DummyParameter? _ = null) { Argument.NotNull(taskScheduler); + Argument.NotNull(valueTaskFunc); return taskScheduler.Enqueue(_ => valueTaskFunc(), CancellationToken.None, _); } @@ -463,24 +477,25 @@ public static Task Enqueue(this ITaskScheduler taskScheduler, Func va /// The action to execute that accepts a cancellation token. /// A cancellation token that can be used to cancel the operation. /// A representing the completion of the enqueued action. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method wraps the synchronous action in a to match the scheduler's /// asynchronous interface. /// - public static async Task Enqueue(this ITaskScheduler taskScheduler, Action action, CancellationToken cancellationToken) + public static Task Enqueue(this ITaskScheduler taskScheduler, Action action, CancellationToken cancellationToken) { Argument.NotNull(taskScheduler); - - await taskScheduler.Enqueue( - token => - { - action(token); - return new ValueTask(Task.CompletedTask); - }, - cancellationToken) - .ConfigureAwait(false); + Argument.NotNull(action); + + return taskScheduler.Enqueue( + (_, token) => + { + action(token); + return new ValueTask(default(Unit)); + }, + null, + cancellationToken); } /// @@ -489,24 +504,15 @@ await taskScheduler.Enqueue( /// The task scheduler to enqueue the operation on. /// The action to execute that accepts a cancellation token. /// A representing the completion of the enqueued action. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method uses for the cancellation token and wraps the /// synchronous action in a . /// - public static async Task Enqueue(this ITaskScheduler taskScheduler, Action action) + public static Task Enqueue(this ITaskScheduler taskScheduler, Action action) { - Argument.NotNull(taskScheduler); - - await taskScheduler.Enqueue( - token => - { - action(token); - return new ValueTask(Task.CompletedTask); - }, - CancellationToken.None) - .ConfigureAwait(false); + return taskScheduler.Enqueue(action, CancellationToken.None); } /// @@ -515,24 +521,15 @@ await taskScheduler.Enqueue( /// The task scheduler to enqueue the operation on. /// The action to execute. /// A representing the completion of the enqueued action. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method uses for the cancellation token, ignores the /// cancellation token parameter when calling the action, and wraps the synchronous action in a . /// - public static async Task Enqueue(this ITaskScheduler taskScheduler, Action action) + public static Task Enqueue(this ITaskScheduler taskScheduler, Action action) { - Argument.NotNull(taskScheduler); - - await taskScheduler.Enqueue( - _ => - { - action(); - return new ValueTask(Task.CompletedTask); - }, - CancellationToken.None) - .ConfigureAwait(false); + return taskScheduler.Enqueue(action, CancellationToken.None); } /// @@ -542,24 +539,25 @@ await taskScheduler.Enqueue( /// The action to execute. /// A cancellation token that can be used to cancel the operation. /// A representing the completion of the enqueued action. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method ignores the cancellation token parameter when calling the action and wraps the /// synchronous action in a . /// - public static async Task Enqueue(this ITaskScheduler taskScheduler, Action action, CancellationToken cancellationToken) + public static Task Enqueue(this ITaskScheduler taskScheduler, Action action, CancellationToken cancellationToken) { Argument.NotNull(taskScheduler); - - await taskScheduler.Enqueue( - _ => - { - action(); - return new ValueTask(Task.CompletedTask); - }, - cancellationToken) - .ConfigureAwait(false); + Argument.NotNull(action); + + return taskScheduler.Enqueue( + (_, token) => + { + action(); + return new ValueTask(default(Unit)); + }, + null, + cancellationToken); } /// @@ -569,7 +567,7 @@ await taskScheduler.Enqueue( /// The task scheduler to enqueue the operation on. /// The synchronous function to execute. /// A representing the result of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method uses for the cancellation token, ignores the @@ -578,6 +576,7 @@ await taskScheduler.Enqueue( public static Task Enqueue(this ITaskScheduler taskScheduler, Func func) { Argument.NotNull(taskScheduler); + Argument.NotNull(func); return taskScheduler.Enqueue(_ => new ValueTask(func()), CancellationToken.None); } @@ -590,7 +589,7 @@ public static Task Enqueue(this ITaskScheduler taskScheduler, Func func /// The synchronous function to execute that accepts a cancellation token. /// A cancellation token that can be used to cancel the operation. /// A representing the result of the enqueued function. - /// Thrown when is null. + /// Thrown synchronously when or is null. /// Thrown if the scheduler has been disposed. /// /// This method wraps the synchronous result in a to match the scheduler's @@ -599,6 +598,7 @@ public static Task Enqueue(this ITaskScheduler taskScheduler, Func func public static Task Enqueue(this ITaskScheduler taskScheduler, Func func, CancellationToken cancellationToken) { Argument.NotNull(taskScheduler); + Argument.NotNull(func); return taskScheduler.Enqueue(token => new ValueTask(func(token)), cancellationToken); } diff --git a/TaskFlow/ThreadTaskFlow.cs b/TaskFlow/ThreadTaskFlow.cs index f1c367f..c906ced 100644 --- a/TaskFlow/ThreadTaskFlow.cs +++ b/TaskFlow/ThreadTaskFlow.cs @@ -86,24 +86,27 @@ protected ThreadTaskFlow(TaskFlowOptions options) /// continuations within the task function will also be marshaled back to the same thread. /// /// + /// Calls made after disposal are rejected synchronously with . + /// Failures from accepted operations are reported through their returned tasks. + /// + /// /// Cancellation does not remove the queued item. The delegate is invoked with a linked token that may already /// be canceled so queue progression remains deterministic. /// /// - public override async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + public override Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) { Argument.NotNull(taskFunc); - ExecutionItem executionItem; lock (ThisLock) { CheckDisposed(); - executionItem = new ExecutionItem(TaskFunc, state, _synchronizationContext, cancellationToken); + var executionItem = new ExecutionItem(TaskFunc, state, _synchronizationContext, cancellationToken); _blockingCollection.Add(executionItem, CancellationToken.None); + return executionItem.Task; } - return await executionItem.GetTypedTask().ConfigureAwait(false); - async Task TaskFunc(object? stateObject, CancellationToken token) + async Task TaskFunc(object? stateObject, CancellationToken token) { var result = await taskFunc(stateObject, token).ConfigureAwait(false); return result; @@ -206,13 +209,21 @@ protected void ThreadStart(object? _) /// This class encapsulates the task function, state, and completion source for an enqueued task. /// It handles the execution of the task and propagation of results and exceptions. /// - private sealed class ExecutionItem + private abstract class ExecutionItem + { + public abstract void Execute(CancellationToken cancellationToken); + + public abstract void Cancel(CancellationToken token); + } + + // Complete the caller-facing typed task before Execute returns and the queue can shut down. + private sealed class ExecutionItem : ExecutionItem { - private readonly Func> _taskFunc; + private readonly Func> _taskFunc; private readonly object? _state; private readonly SynchronizationContext? _synchronizationContext; private readonly CancellationToken _cancellationToken; - private readonly TaskCompletionSource _taskCompletionSource; + private readonly TaskCompletionSource _taskCompletionSource; /// /// Initializes a new instance of the class. @@ -222,7 +233,7 @@ private sealed class ExecutionItem /// The synchronization context to use for continuations. /// A cancellation token that can be used to cancel the operation. public ExecutionItem( - Func> taskFunc, + Func> taskFunc, object? state, SynchronizationContext? synchronizationContext, CancellationToken cancellationToken) @@ -231,19 +242,14 @@ public ExecutionItem( _state = state; _synchronizationContext = synchronizationContext; _cancellationToken = cancellationToken; - _taskCompletionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _taskCompletionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); } /// /// Gets a task that represents the completion of this execution item with the expected result type. /// - /// The expected result type. /// A task that completes with the result of the execution item. - public async Task GetTypedTask() - { - var result = await _taskCompletionSource.Task.ConfigureAwait(false); - return (T)result!; - } + public Task Task => _taskCompletionSource.Task; /// /// Executes the task function and handles the result. @@ -251,7 +257,7 @@ public async Task GetTypedTask() /// A cancellation token that can be used to cancel the operation. [SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "Exception propagated to TaskCompletionSource")] [SuppressMessage("Reliability", "CA2000:Dispose objects before losing scope", Justification = "linkedCts disposed after awaiter is done")] - public void Execute(CancellationToken cancellationToken) + public override void Execute(CancellationToken cancellationToken) { var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cancellationToken); var blockingCollection = new BlockingCollection<(SendOrPostCallback Callback, object? State)>(); @@ -285,7 +291,7 @@ public void Execute(CancellationToken cancellationToken) return; - void HandleAwaiterCompletion(TaskAwaiter awaiter) + void HandleAwaiterCompletion(TaskAwaiter awaiter) { try { @@ -310,7 +316,7 @@ void HandleAwaiterCompletion(TaskAwaiter awaiter) /// Cancels the execution of this item. /// /// The cancellation token that caused the cancellation. - public void Cancel(CancellationToken token) + public override void Cancel(CancellationToken token) { _taskCompletionSource.TrySetCanceled(token); } diff --git a/docs/design/middleware-lifetime.md b/docs/design/middleware-lifetime.md index 2ef8c6c..975f449 100644 --- a/docs/design/middleware-lifetime.md +++ b/docs/design/middleware-lifetime.md @@ -28,10 +28,15 @@ its explicit middleware lifetime has finished draining and cleaning up. ## Operation activities -PipelineOperation starts with one activity owned by EnqueueCore. After successful -lifetime entry it stores the owner directly, without a bound exit delegate. -Failure to enter must not invoke completion middleware whose resources may already -have been cleaned up, or decrement a lifetime it never entered. +PipelineOperation starts with one activity. Plain Enqueue enters its lifetime +synchronously to protect its own middleware resources before forwarding. It does +not probe the terminal scheduler's disposal state. Entry and closure use the same +lock: accepted work is counted before shutdown can drain it. After successful entry, the operation stores its +owner directly and EnqueueCore owns the initial activity once forwarding returns +a task. If entry or synchronous forwarding fails first, submission releases the +activity and propagates the exception without invoking completion middleware. +A failed entry never exits a lifetime it did not enter. Explicit admission Submit retains its +separate asynchronous decision/completion contract. Every invocation of an enqueue or execution continuation acquires an activity before calling user code. Independently pending work retains its activity even @@ -52,9 +57,28 @@ can race to report it; the winning outcome drives completion middleware. Caller admission outcomes are recorded after completion middleware has processed failures or recovery, rather than from the raw terminal outcome. -Each enqueue continuation uses async try/finally to retain its activity until its -own returned task settles, even if middleware independently completes the -caller-facing task earlier. This also preserves normal async exception semantics. +Each enqueue continuation forwards synchronously before observing the returned +task. Immediate scheduling rejection releases its activity and propagates directly +through forwarding pipelines, preserving the original exception. Once a task is +returned, observation retains the activity until that task settles, even when +middleware completes the caller-facing task earlier. Returned task failures enter +completion middleware; they are never promoted into synchronous exceptions. + +Admission middleware can short-circuit or defer forwarding according to its own +state. It does not inspect terminal disposal to manufacture an early rejection. +Transparent wrappers preserve exceptions from synchronous forwarding. Middleware +that owns asynchronous admission or scheduling, including timeout and queuing, +may report forwarding failures through its returned task even if it invokes the +continuation immediately. Such task failures still enter completion middleware. +Explicit Submit continues to report admission and scheduling outcomes through +its decision/completion tasks. + +Cancellation scope and cancel-previous retain their existing async wrappers and +linked-token ownership. An immediate forwarding failure is represented by their +returned task, and their using scopes release resources when that task settles. +Timeout likewise retains its competing-task semantics. The pipeline does not +impose a new exception-timing contract on these wrappers or reinterpret the +exceptions produced by Task.Factory.StartNew in TaskFlowSchedulerAdapter. ## Annotation-only forwarding @@ -65,10 +89,10 @@ terminals retain pipeline activity tracking; their completion hooks may not drai all submitted work. Context-aware submissions still construct an operation so the final annotations remain available to the delegate. -The successful forwarding path returns the terminal task directly. A cold catch -preserves pipeline submission's asynchronous failure behavior when a terminal -throws before returning a task. Its async failure helper also preserves the -canceled-task semantics of OperationCanceledException, including uncanceled tokens. +Forwarding returns the terminal task directly. Enqueue-time validation and +scheduler rejection propagate synchronously, including ObjectDisposedException +when the terminal has closed entry. Failures and cancellation from accepted +operations remain observable through their returned tasks. ExecutionContext flow, caller cancellation and terminal scheduler selection are left to the existing scheduling paths. No lifetime scope needs to flow with them. \ No newline at end of file