From d61115300e51bc24fb9297e694c98234a91661ed Mon Sep 17 00:00:00 2001 From: Volodymyr Dombrovskyi <5788605+dombrovsky@users.noreply.github.com> Date: Sun, 16 Aug 2026 17:07:47 -0600 Subject: [PATCH 1/2] Add extensible task scheduler middleware pipeline --- .../LoggingTaskSchedulerExtensionsFixture.cs | 2 +- .../LoggingTaskSchedulerExtensions.cs | 181 +++----- ...ExceptionTaskSchedulerExtensionsFixture.cs | 54 +++ ...erceptionTaskSchedulerExtensionsFixture.cs | 6 +- ...askSchedulerMiddlewareExtensionsFixture.cs | 421 ++++++++++++++++++ .../TimeoutTaskSchedulerExtensionsFixture.cs | 4 +- ...OwnershipTaskSchedulerExtensionsFixture.cs | 16 + TaskFlow.sln | 5 +- .../AnnotatingTaskSchedulerExtensions.cs | 13 +- .../CancelPreviousTaskSchedulerExtensions.cs | 17 +- ...ancellationScopeTaskSchedulerExtensions.cs | 19 +- .../ExceptionTaskSchedulerExtensions.cs | 40 +- .../InterceptionTaskSchedulerExtensions.cs | 111 ++--- .../TaskSchedulerInterceptionContext.cs | 12 +- .../ThrottlingTaskSchedulerExtensions.cs | 14 +- .../TimeoutTaskSchedulerExtensions.cs | 25 +- TaskFlow/Internal/AnnotationScope.cs | 38 ++ TaskFlow/Internal/MiddlewareRegistration.cs | 15 + TaskFlow/Internal/PipelineOperation.cs | 78 ++++ TaskFlow/Internal/TaskSchedulerPipeline.cs | 171 +++++++ .../ITaskSchedulerCompletionMiddleware.cs | 32 ++ .../ITaskSchedulerEnqueueMiddleware.cs | 26 ++ .../ITaskSchedulerExecutionMiddleware.cs | 25 ++ .../Middleware/ITaskSchedulerMiddleware.cs | 15 + .../TaskSchedulerCompletionDelegate.cs | 20 + .../Middleware/TaskSchedulerEnqueueContext.cs | 89 ++++ .../TaskSchedulerEnqueueDelegate.cs | 18 + .../TaskSchedulerExecutionDelegate.cs | 17 + .../TaskSchedulerMiddlewareExtensions.cs | 133 ++++++ .../TaskSchedulerOperationContext.cs | 83 ++++ .../TaskSchedulerOperationOutcome.cs | 72 +++ docs/customization.md | 216 ++++++++- docs/extensions/cancellation.md | 2 + docs/extensions/observability.md | 16 +- docs/extensions/reliability.md | 2 + docs/semantics-and-pitfalls.md | 12 +- 36 files changed, 1737 insertions(+), 283 deletions(-) create mode 100644 TaskFlow.Tests/Extensions/TaskSchedulerMiddlewareExtensionsFixture.cs create mode 100644 TaskFlow/Internal/AnnotationScope.cs create mode 100644 TaskFlow/Internal/MiddlewareRegistration.cs create mode 100644 TaskFlow/Internal/PipelineOperation.cs create mode 100644 TaskFlow/Internal/TaskSchedulerPipeline.cs create mode 100644 TaskFlow/Middleware/ITaskSchedulerCompletionMiddleware.cs create mode 100644 TaskFlow/Middleware/ITaskSchedulerEnqueueMiddleware.cs create mode 100644 TaskFlow/Middleware/ITaskSchedulerExecutionMiddleware.cs create mode 100644 TaskFlow/Middleware/ITaskSchedulerMiddleware.cs create mode 100644 TaskFlow/Middleware/TaskSchedulerCompletionDelegate.cs create mode 100644 TaskFlow/Middleware/TaskSchedulerEnqueueContext.cs create mode 100644 TaskFlow/Middleware/TaskSchedulerEnqueueDelegate.cs create mode 100644 TaskFlow/Middleware/TaskSchedulerExecutionDelegate.cs create mode 100644 TaskFlow/Middleware/TaskSchedulerMiddlewareExtensions.cs create mode 100644 TaskFlow/Middleware/TaskSchedulerOperationContext.cs create mode 100644 TaskFlow/Middleware/TaskSchedulerOperationOutcome.cs diff --git a/TaskFlow.Extensions.Microsoft.Logging.Tests/LoggingTaskSchedulerExtensionsFixture.cs b/TaskFlow.Extensions.Microsoft.Logging.Tests/LoggingTaskSchedulerExtensionsFixture.cs index 92509fc..db8805d 100644 --- a/TaskFlow.Extensions.Microsoft.Logging.Tests/LoggingTaskSchedulerExtensionsFixture.cs +++ b/TaskFlow.Extensions.Microsoft.Logging.Tests/LoggingTaskSchedulerExtensionsFixture.cs @@ -29,7 +29,7 @@ public async Task WithLogging_LogsFullLifecycleAtTraceByDefault() { _taskFlow = new TaskFlow(); var logger = new RecordingLogger(LogLevel.Trace); - Assert.That(await _taskFlow.WithLogging(logger).WithOperationName("answer").Enqueue(() => 42), Is.EqualTo(42)); + Assert.That(await _taskFlow.WithOperationName("answer").WithLogging(logger).Enqueue(() => 42), Is.EqualTo(42)); Assert.That(logger.Entries.Select(x => x.EventId.Id), Is.EqualTo(new[] { EnqueuedEventId, StartedEventId, SucceededEventId, FinishedEventId })); Assert.That(logger.Entries, Has.All.Property(nameof(LogEntry.Level)).EqualTo(LogLevel.Trace)); Assert.That(logger.Entries.Select(x => x.Message), Has.All.Contains("operation 1")); diff --git a/TaskFlow.Extensions.Microsoft.Logging/LoggingTaskSchedulerExtensions.cs b/TaskFlow.Extensions.Microsoft.Logging/LoggingTaskSchedulerExtensions.cs index 15a129e..fa2140f 100644 --- a/TaskFlow.Extensions.Microsoft.Logging/LoggingTaskSchedulerExtensions.cs +++ b/TaskFlow.Extensions.Microsoft.Logging/LoggingTaskSchedulerExtensions.cs @@ -14,80 +14,76 @@ public static class LoggingTaskSchedulerExtensions private static readonly EventId FailedEvent = new EventId(0x5446_0005, "TaskFlowOperationFailed"); private static readonly EventId FinishedEvent = new EventId(0x5446_0006, "TaskFlowOperationFinished"); - /// Wraps a scheduler with configurable structured lifecycle logging. + /// Registers structured enqueue and execution lifecycle logging for every scheduled operation. /// The scheduler whose operations will be logged. /// The logger that receives lifecycle events. /// /// An optional callback that configures event levels. When omitted, every lifecycle event uses /// . /// - /// An that logs the lifecycle of every enqueued operation. + /// A new immutable scheduler snapshot that logs operation lifecycles. /// - /// Thrown when or is null. + /// or is null. /// /// /// - /// The decorator emits enqueue, start, cancellation-request, success or failure, and finish events. Events - /// contain structured operation ID, optional operation name, result type, and elapsed-duration fields where - /// applicable. Failure events include the operation exception. + /// The compound middleware emits enqueue, start, cancellation-request, success or failure, and finish events. + /// Every event carries an increasing operation ID and the visible when + /// this logging registration is created. Result type and elapsed duration are included where applicable. /// /// - /// is checked before logging each event and before optional timing - /// work is started. Disabled events do not call . + /// Call before + /// WithLogging. Metadata is forward-scoped, so an operation name registered later does not retroactively + /// change this logging registration. /// /// - /// Cancellation requests are observed synchronously from the cancellation token. This event does not mean - /// that cancellation was accepted or that the operation ultimately completed as canceled. - /// - /// - /// Place WithOperationName outside this decorator, for example - /// scheduler.WithLogging(logger).WithOperationName("Import"), so logging can observe the annotation. + /// is checked before each event and before optional timing work begins. + /// Disabled events do not invoke . + /// Cancellation-request logging observes the submitting caller's token and does not imply that the operation + /// ultimately completes as canceled. Logging never suppresses or replaces the operation outcome. /// + /// The returned pipeline snapshot is non-owning and does not dispose the underlying scheduler or logger. /// /// /// /// var scheduler = taskFlow - /// .WithLogging(logger, options => + /// .WithOperationName("imports.run") + /// .WithLogging(logger, options => /// { + /// options.StartedLogLevel = LogLevel.Information; /// options.FailedLogLevel = LogLevel.Error; /// options.FinishedLogLevel = LogLevel.Debug; - /// }) - /// .WithOperationName("Import"); + /// }); /// - /// await scheduler.Enqueue(() => ImportAsync()); + /// await scheduler.Enqueue(token => ImportAsync(token)); /// /// public static ITaskScheduler WithLogging(this ITaskScheduler taskScheduler, ILogger logger, Action? configure = null) { Argument.NotNull(taskScheduler); Argument.NotNull(logger); - var options = new TaskFlowLoggingOptions(); configure?.Invoke(options); - return new LoggingTaskSchedulerWrapper(taskScheduler, logger, options); + return taskScheduler.UseMiddleware(new LoggingMiddleware(logger, options)); } - private sealed class LoggingTaskSchedulerWrapper : ITaskScheduler + private sealed class LoggingMiddleware : ITaskSchedulerEnqueueMiddleware, ITaskSchedulerExecutionMiddleware { private readonly ILogger _logger; private readonly TaskFlowLoggingOptions _options; - private readonly ITaskScheduler _interceptedScheduler; private long _lastOperationId; - public LoggingTaskSchedulerWrapper(ITaskScheduler taskScheduler, ILogger logger, TaskFlowLoggingOptions options) + public LoggingMiddleware(ILogger logger, TaskFlowLoggingOptions options) { _logger = logger; _options = options; - _interceptedScheduler = taskScheduler.Intercept(new LoggingInterceptor(logger, options)); } - public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + public async Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) { - var operation = new LoggingOperationState( + var operation = context.GetOrCreateLocalState(() => new LoggingOperationState( Interlocked.Increment(ref _lastOperationId), - TaskSchedulerInterceptionContext.GetAnnotation(state)?.OperationName, - taskFunc, - state); + context.GetAnnotation()?.OperationName)); if (_logger.IsEnabled(_options.EnqueuedLogLevel)) { @@ -95,7 +91,7 @@ public async Task Enqueue(Func> t "TaskFlow operation {OperationId} ({OperationName}) enqueued", operation.OperationId, operation.OperationName); } - using var registration = cancellationToken.Register(() => + using var registration = context.CallerCancellationToken.Register(() => { if (_logger.IsEnabled(_options.CancellationRequestedLogLevel)) { @@ -104,118 +100,81 @@ public async Task Enqueue(Func> t } }); - return await _interceptedScheduler.Enqueue(Execute, operation, cancellationToken).ConfigureAwait(false); - - static ValueTask Execute(object? operationState, CancellationToken token) - { - var loggingState = (LoggingOperationState)operationState!; - return loggingState.TaskFunc(loggingState.State, token); - } + return await continuation(context).ConfigureAwait(false); } - } - - private struct LoggingInterceptor : ITaskSchedulerInterceptor - { - private readonly ILogger _logger; - private readonly TaskFlowLoggingOptions _options; - private long _startTimestamp; - public LoggingInterceptor(ILogger logger, TaskFlowLoggingOptions options) + public async ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation) { - _logger = logger; - _options = options; - _startTimestamp = 0; - } + var operation = context.GetOrCreateLocalState(() => new LoggingOperationState( + Interlocked.Increment(ref _lastOperationId), + context.GetAnnotation()?.OperationName)); - public void OnBefore(TaskSchedulerInterceptionContext context) - { if (_logger.IsEnabled(_options.SucceededLogLevel) || _logger.IsEnabled(_options.FailedLogLevel) || _logger.IsEnabled(_options.FinishedLogLevel)) { - _startTimestamp = Stopwatch.GetTimestamp(); + operation.StartTimestamp = Stopwatch.GetTimestamp(); } if (_logger.IsEnabled(_options.StartedLogLevel)) { - var operation = GetLoggingState(context); _logger.Log(_options.StartedLogLevel, StartedEvent, "TaskFlow operation {OperationId} ({OperationName}) started", operation.OperationId, operation.OperationName); } - } - - public void OnSuccess(TaskSchedulerInterceptionContext context, TResult result) - { - if (_logger.IsEnabled(_options.SucceededLogLevel)) + try { - var operation = GetLoggingState(context); - _logger.Log(_options.SucceededLogLevel, SucceededEvent, - "TaskFlow operation {OperationId} ({OperationName}) succeeded with result type {ResultType} in {ElapsedMilliseconds} ms", - operation.OperationId, operation.OperationName, typeof(TResult).FullName, GetElapsedMilliseconds()); - } + TResult result; + try + { + result = await continuation(context).ConfigureAwait(true); + } + catch (Exception exception) + { + if (_logger.IsEnabled(_options.FailedLogLevel)) + { + _logger.Log(_options.FailedLogLevel, FailedEvent, exception, + "TaskFlow operation {OperationId} ({OperationName}) failed in {ElapsedMilliseconds} ms", + operation.OperationId, operation.OperationName, operation.GetElapsedMilliseconds()); + } + + throw; + } - } + if (_logger.IsEnabled(_options.SucceededLogLevel)) + { + _logger.Log(_options.SucceededLogLevel, SucceededEvent, + "TaskFlow operation {OperationId} ({OperationName}) succeeded with result type {ResultType} in {ElapsedMilliseconds} ms", + operation.OperationId, operation.OperationName, typeof(TResult).FullName, operation.GetElapsedMilliseconds()); + } - public void OnError(TaskSchedulerInterceptionContext context, Exception exception) - { - if (_logger.IsEnabled(_options.FailedLogLevel)) - { - var operation = GetLoggingState(context); - _logger.Log(_options.FailedLogLevel, FailedEvent, exception, - "TaskFlow operation {OperationId} ({OperationName}) failed in {ElapsedMilliseconds} ms", - operation.OperationId, operation.OperationName, GetElapsedMilliseconds()); + return result; } - - } - - public void OnFinally(TaskSchedulerInterceptionContext context) - { - LogFinished(context); - } - - private void LogFinished(TaskSchedulerInterceptionContext context) - { - if (_logger.IsEnabled(_options.FinishedLogLevel)) + finally { - var operation = GetLoggingState(context); - _logger.Log(_options.FinishedLogLevel, FinishedEvent, - "TaskFlow operation {OperationId} ({OperationName}) finished in {ElapsedMilliseconds} ms", - operation.OperationId, operation.OperationName, GetElapsedMilliseconds()); + if (_logger.IsEnabled(_options.FinishedLogLevel)) + { + _logger.Log(_options.FinishedLogLevel, FinishedEvent, + "TaskFlow operation {OperationId} ({OperationName}) finished in {ElapsedMilliseconds} ms", + operation.OperationId, operation.OperationName, operation.GetElapsedMilliseconds()); + } } } - - private double GetElapsedMilliseconds() - { - return _startTimestamp != 0 - ? (Stopwatch.GetTimestamp() - _startTimestamp) * 1000d / Stopwatch.Frequency - : 0d; - } - - private static ILoggingOperationState GetLoggingState(TaskSchedulerInterceptionContext context) - { - return (ILoggingOperationState)context.State!; - } - } - - private interface ILoggingOperationState - { - long OperationId { get; } - string? OperationName { get; } } - private sealed class LoggingOperationState : ILoggingOperationState + private sealed class LoggingOperationState { - public LoggingOperationState(long operationId, string? operationName, Func> taskFunc, object? state) + public LoggingOperationState(long operationId, string? operationName) { OperationId = operationId; OperationName = operationName; - TaskFunc = taskFunc; - State = state; } public long OperationId { get; } public string? OperationName { get; } - public Func> TaskFunc { get; } - public object? State { get; } + public long StartTimestamp { get; set; } + + public double GetElapsedMilliseconds() => StartTimestamp == 0 + ? 0d + : (Stopwatch.GetTimestamp() - StartTimestamp) * 1000d / Stopwatch.Frequency; } } } diff --git a/TaskFlow.Tests/Extensions/ExceptionTaskSchedulerExtensionsFixture.cs b/TaskFlow.Tests/Extensions/ExceptionTaskSchedulerExtensionsFixture.cs index 02d8422..211e820 100644 --- a/TaskFlow.Tests/Extensions/ExceptionTaskSchedulerExtensionsFixture.cs +++ b/TaskFlow.Tests/Extensions/ExceptionTaskSchedulerExtensionsFixture.cs @@ -105,5 +105,59 @@ public void Enqueue_MultipleHandlers_ShouldExecuteAllMatchingHandlers(ITaskFlow Assert.That(exceptions, Is.EqualTo(Generic)); exceptions.Clear(); } + + [Test] + public void OnError_DedicatedThread_AsynchronousFailureCallbackRetainsFlowContext() + { + var flow = new DedicatedThreadTaskFlow(); + _taskFlow = flow; + var operationThread = 0; + var callbackThread = 0; + SynchronizationContext? operationContext = null; + SynchronizationContext? callbackContext = null; + + var task = flow.OnError(_ => + { + callbackThread = Environment.CurrentManagedThreadId; + callbackContext = SynchronizationContext.Current; + }) + .Enqueue(FailAsync); + + Assert.That(async () => await task, Throws.InvalidOperationException); + Assert.That(callbackThread, Is.EqualTo(operationThread)); + Assert.That(callbackContext, Is.SameAs(operationContext).And.Not.Null); + + async ValueTask FailAsync(CancellationToken _) + { + await Task.Yield(); + operationThread = Environment.CurrentManagedThreadId; + operationContext = SynchronizationContext.Current; + throw new InvalidOperationException("expected"); + } + } + + [TestCaseSource(typeof(TaskFlows), nameof(TaskFlows.CreateTaskFlows))] + public void OnError_CallbackFailureBecomesOutcomeForLaterHandlers(ITaskFlow taskFlow) + { + _taskFlow = taskFlow; + Exception? observed = null; + var scheduler = taskFlow + .OnError(_ => throw new NotSupportedException("replacement")) + .Intercept(new NoOpInterceptor()) + .OnError(exception => observed = exception); + + var task = scheduler.Enqueue(() => throw new InvalidOperationException("original")); + + Assert.That(async () => await task, Throws.TypeOf().With.Message.EqualTo("replacement")); + Assert.That(observed, Is.TypeOf().With.Message.EqualTo("replacement")); + } + + private readonly struct NoOpInterceptor : ITaskSchedulerInterceptor + { + public void OnBefore(TaskSchedulerInterceptionContext context) { } + public void OnSuccess(TaskSchedulerInterceptionContext context, TResult result) { } + public void OnError(TaskSchedulerInterceptionContext context, Exception exception) { } + public void OnFinally(TaskSchedulerInterceptionContext context) { } + } } } diff --git a/TaskFlow.Tests/Extensions/InterceptionTaskSchedulerExtensionsFixture.cs b/TaskFlow.Tests/Extensions/InterceptionTaskSchedulerExtensionsFixture.cs index dccdba2..369c773 100644 --- a/TaskFlow.Tests/Extensions/InterceptionTaskSchedulerExtensionsFixture.cs +++ b/TaskFlow.Tests/Extensions/InterceptionTaskSchedulerExtensionsFixture.cs @@ -46,7 +46,7 @@ public void Intercept_ObservesErrorsAndMetadata(ITaskFlow taskFlow) var interceptor = new RecordingInterceptor(events, false); var state = new object(); Func> operation = (_, _) => throw new InvalidOperationException("boom"); - var task = taskFlow.Intercept(interceptor).WithOperationName("failure") + var task = taskFlow.WithOperationName("failure").Intercept(interceptor) .Enqueue(operation, state, CancellationToken.None); Assert.That(async () => await task, Throws.InvalidOperationException.With.Message.EqualTo("boom")); Assert.That(events, Is.EqualTo(FailedLifecycle)); @@ -174,8 +174,8 @@ public async Task Intercept_SynchronousHooksRunInOrderAndObserveContext(ITaskFlo var state = new object(); using var cts = new CancellationTokenSource(); - var result = await taskFlow.Intercept(new SynchronousLifecycleInterceptor(recorder)) - .WithOperationName("sync") + var result = await taskFlow.WithOperationName("sync") + .Intercept(new SynchronousLifecycleInterceptor(recorder)) .Enqueue((operationState, token) => { recorder.Events.Add("operation"); diff --git a/TaskFlow.Tests/Extensions/TaskSchedulerMiddlewareExtensionsFixture.cs b/TaskFlow.Tests/Extensions/TaskSchedulerMiddlewareExtensionsFixture.cs new file mode 100644 index 0000000..a8b2166 --- /dev/null +++ b/TaskFlow.Tests/Extensions/TaskSchedulerMiddlewareExtensionsFixture.cs @@ -0,0 +1,421 @@ +namespace TaskFlow.Tests.Extensions +{ + using NUnit.Framework; + using System.Collections.Concurrent; + using System.Threading.Tasks.Flow; + + [TestFixture] + internal sealed class TaskSchedulerMiddlewareExtensionsFixture + { + private static readonly string[] ExpectedPhaseEvents = + { + "enqueue:second", "enqueue:first", "terminal", "execute:first", "execute:second", + "operation", "complete:first", "complete:second", "terminal-return" + }; + + [Test] + public async Task Pipeline_UsesPhaseOrderingAndOneTerminalDelegate() + { + var events = new List(); + var terminal = new RecordingTerminal(events); + var scheduler = terminal + .UseMiddleware(new RecordingMiddleware("first", events)) + .UseMiddleware(new RecordingMiddleware("second", events)); + + Assert.That(await scheduler.Enqueue(() => { events.Add("operation"); return 42; }), Is.EqualTo(42)); + Assert.That(terminal.EnqueueCount, Is.EqualTo(1)); + Assert.That(events, Is.EqualTo(ExpectedPhaseEvents)); + } + + [Test] + public async Task Metadata_IsForwardScopedShadowedAndBranchIndependent() + { + var observations = new ConcurrentBag(); + var root = new InlineTerminal().WithAnnotation(new TestAnnotation("root")); + var parent = root.UseMiddleware(new AnnotationMiddleware("parent", observations)); + var left = parent.WithAnnotation(new TestAnnotation("left")) + .UseMiddleware(new AnnotationMiddleware("left", observations)); + var right = parent.WithAnnotation(new TestAnnotation("right")) + .UseMiddleware(new AnnotationMiddleware("right", observations)); + + await Task.WhenAll(left.Enqueue(() => 1), right.Enqueue(() => 2)); + + Assert.That(observations.Count(x => x == "parent:root"), Is.EqualTo(2)); + Assert.That(observations, Does.Contain("left:left")); + Assert.That(observations, Does.Contain("right:right")); + } + + [Test] + public async Task EnqueueMiddleware_MayShortCircuitWithoutReachingTerminal() + { + var terminal = new RecordingTerminal(new List()); + var scheduler = terminal.UseMiddleware(new ConstantMiddleware(17)); + + Assert.That(await scheduler.Enqueue(() => 42), Is.EqualTo(17)); + Assert.That(terminal.EnqueueCount, Is.Zero); + } + + [Test] + public async Task ExecutionMiddleware_MayInvokeContinuationMoreThanOnceInOneQueueTurn() + { + var terminal = new RecordingTerminal(new List()); + var calls = 0; + var scheduler = terminal.UseMiddleware(new TwiceMiddleware()); + + Assert.That(await scheduler.Enqueue(() => ++calls), Is.EqualTo(2)); + Assert.That(calls, Is.EqualTo(2)); + Assert.That(terminal.EnqueueCount, Is.EqualTo(1)); + } + + [Test] + public async Task CompoundMiddleware_SharesOnlyItsRegistrationLocalState() + { + var middleware = new CompoundStateMiddleware(); + var scheduler = new InlineTerminal().UseMiddleware(middleware); + + await Task.WhenAll(scheduler.Enqueue(() => 1), scheduler.Enqueue(() => 2)); + + Assert.That(middleware.ObservedIds, Has.Count.EqualTo(2)); + Assert.That(middleware.ObservedIds.Distinct().Count(), Is.EqualTo(2)); + } + + [Test] + public async Task Terminal_ReceivesOriginalStateAndProducerToken() + { + var terminal = new RecordingTerminal(new List()); + var state = new object(); + using var source = new CancellationTokenSource(); + var scheduler = terminal.UseMiddleware(new AnnotationMiddleware("unused", new ConcurrentBag())); + + await scheduler.Enqueue((s, _) => new ValueTask(s), state, source.Token); + + Assert.That(terminal.State, Is.SameAs(state)); + Assert.That(terminal.CancellationToken, Is.EqualTo(source.Token)); + } + + [Test] + public async Task EnqueueWithContext_ReceivesFinalStateTokensAndAnnotations() + { + var state = new object(); + using var source = new CancellationTokenSource(); + var scheduler = new InlineTerminal().WithAnnotation(new TestAnnotation("final")); + + var result = await scheduler.EnqueueWithContext( + context => new ValueTask<(object?, CancellationToken, CancellationToken, string?)>(( + context.State, + context.CallerCancellationToken, + context.CancellationToken, + context.GetAnnotation()?.Value)), + state, + source.Token); + + Assert.That(result.Item1, Is.SameAs(state)); + Assert.That(result.Item2, Is.EqualTo(source.Token)); + Assert.That(result.Item3, Is.EqualTo(source.Token)); + Assert.That(result.Item4, Is.EqualTo("final")); + } + + [Test] + public async Task EnqueueMiddleware_MayReplaceProducerTokenWithoutReplacingCallerToken() + { + using var callerSource = new CancellationTokenSource(); + using var producerSource = new CancellationTokenSource(); + var terminal = new RecordingTerminal(new List()); + var observer = new TokenObservingMiddleware(); + var scheduler = terminal + .UseMiddleware(observer) + .UseMiddleware(new ProducerTokenMiddleware(producerSource.Token)); + + await scheduler.Enqueue( + (_, _) => new ValueTask(42), + null, + callerSource.Token); + + Assert.That(terminal.CancellationToken, Is.EqualTo(producerSource.Token)); + Assert.That(observer.CallerToken, Is.EqualTo(callerSource.Token)); + Assert.That(observer.ProducerToken, Is.EqualTo(producerSource.Token)); + } + + [Test] + public void TerminalEnqueueFailure_IsProcessedByCompletionExactlyOnce() + { + var failure = new InvalidOperationException("rejected"); + var completion = new ObservingCompletionMiddleware(); + var scheduler = new ThrowingTerminal(failure).UseMiddleware(completion); + + var thrown = Assert.ThrowsAsync(async () => await scheduler.Enqueue(() => 42)); + + Assert.That(thrown, Is.SameAs(failure)); + Assert.That(completion.CallCount, Is.EqualTo(1)); + Assert.That(completion.ObservedException, Is.SameAs(failure)); + } + + [Test] + public void ExecutionFailure_PreservesExceptionAndRunsCompletionExactlyOnce() + { + var failure = CreateFailureWithCapturedStack(); + var completion = new ObservingCompletionMiddleware(); + var scheduler = new InlineTerminal().UseMiddleware(completion); + + var thrown = Assert.ThrowsAsync(async () => await scheduler.Enqueue( + (_, _) => ValueTask.FromException(failure), + null, + CancellationToken.None)); + + Assert.That(thrown, Is.SameAs(failure)); + Assert.That(thrown!.StackTrace, Does.Contain(nameof(CreateFailureWithCapturedStack))); + Assert.That(completion.CallCount, Is.EqualTo(1)); + Assert.That(completion.ObservedException, Is.SameAs(failure)); + } + + [Test] + public void CompletionReplacement_IsVisibleToLaterMiddlewareAndCaller() + { + var original = new InvalidOperationException("original"); + var replacement = new NotSupportedException("replacement"); + var observer = new ObservingCompletionMiddleware(); + var scheduler = new InlineTerminal() + .UseMiddleware(new ReplacingCompletionMiddleware(replacement)) + .UseMiddleware(observer); + + var thrown = Assert.ThrowsAsync(async () => await scheduler.Enqueue( + (_, _) => ValueTask.FromException(original), + null, + CancellationToken.None)); + + Assert.That(thrown, Is.SameAs(replacement)); + Assert.That(observer.ObservedException, Is.SameAs(replacement)); + } + + [Test] + public void CompletionFailureBeforeContinuation_IsVisibleToLaterMiddleware() + { + var replacement = new NotSupportedException("callback failed"); + var observer = new ObservingCompletionMiddleware(); + var scheduler = new InlineTerminal() + .UseMiddleware(new ThrowingCompletionMiddleware(replacement, throwAfterContinuation: false)) + .UseMiddleware(observer); + + var thrown = Assert.ThrowsAsync(async () => await scheduler.Enqueue(() => 42)); + + Assert.That(thrown, Is.SameAs(replacement)); + Assert.That(observer.CallCount, Is.EqualTo(1)); + Assert.That(observer.ObservedException, Is.SameAs(replacement)); + } + + [Test] + public void CompletionFailureAfterContinuation_ReplacesFinalOutcomeWithoutRepeatingLaterMiddleware() + { + var replacement = new NotSupportedException("callback failed after next"); + var observer = new ObservingCompletionMiddleware(); + var scheduler = new InlineTerminal() + .UseMiddleware(new ThrowingCompletionMiddleware(replacement, throwAfterContinuation: true)) + .UseMiddleware(observer); + + var thrown = Assert.ThrowsAsync(async () => await scheduler.Enqueue(() => 42)); + + Assert.That(thrown, Is.SameAs(replacement)); + Assert.That(observer.CallCount, Is.EqualTo(1)); + Assert.That(observer.ObservedException, Is.Null); + } + + [Test] + public void UseMiddleware_RejectsMarkerWithoutPhase() + { + Assert.That( + () => new InlineTerminal().UseMiddleware(new MarkerOnlyMiddleware()), + Throws.ArgumentException.With.Property("ParamName").EqualTo("middleware")); + } + + private static InvalidOperationException CreateFailureWithCapturedStack() + { + try + { + throw new InvalidOperationException("failure"); + } + catch (InvalidOperationException exception) + { + return exception; + } + } + + private sealed class TestAnnotation : IOperationAnnotation + { + public TestAnnotation(string value) => Value = value; + public string Value { get; } + } + + private sealed class RecordingMiddleware : ITaskSchedulerEnqueueMiddleware, ITaskSchedulerExecutionMiddleware, ITaskSchedulerCompletionMiddleware + { + private readonly string _name; + private readonly IList _events; + public RecordingMiddleware(string name, IList events) { _name = name; _events = events; } + + public async Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) + { + _events.Add("enqueue:" + _name); + return await continuation(context); + } + + public async ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation) + { + _events.Add("execute:" + _name); + return await continuation(context); + } + + public async ValueTask> InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerOperationOutcome outcome, TaskSchedulerCompletionDelegate continuation) + { + _events.Add("complete:" + _name); + return await continuation(context, outcome); + } + } + + private sealed class AnnotationMiddleware : ITaskSchedulerExecutionMiddleware + { + private readonly string _name; + private readonly ConcurrentBag _observations; + public AnnotationMiddleware(string name, ConcurrentBag observations) { _name = name; _observations = observations; } + public ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation) + { + _observations.Add(_name + ":" + context.GetAnnotation()?.Value); + return continuation(context); + } + } + + private sealed class ConstantMiddleware : ITaskSchedulerEnqueueMiddleware + { + private readonly int _value; + public ConstantMiddleware(int value) => _value = value; + public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) + => Task.FromResult((TResult)(object)_value); + } + + private sealed class ProducerTokenMiddleware : ITaskSchedulerEnqueueMiddleware + { + private readonly CancellationToken _producerToken; + public ProducerTokenMiddleware(CancellationToken producerToken) => _producerToken = producerToken; + public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) + => continuation(context.WithCancellationToken(_producerToken)); + } + + private sealed class TokenObservingMiddleware : ITaskSchedulerExecutionMiddleware + { + public CancellationToken CallerToken { get; private set; } + public CancellationToken ProducerToken { get; private set; } + public ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation) + { + CallerToken = context.CallerCancellationToken; + ProducerToken = context.CancellationToken; + return continuation(context); + } + } + + private sealed class ObservingCompletionMiddleware : ITaskSchedulerCompletionMiddleware + { + public int CallCount { get; private set; } + public Exception? ObservedException { get; private set; } + public ValueTask> InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerOperationOutcome outcome, TaskSchedulerCompletionDelegate continuation) + { + CallCount++; + ObservedException = outcome.Exception; + return continuation(context, outcome); + } + } + + private sealed class ReplacingCompletionMiddleware : ITaskSchedulerCompletionMiddleware + { + private readonly Exception _replacement; + public ReplacingCompletionMiddleware(Exception replacement) => _replacement = replacement; + public ValueTask> InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerOperationOutcome outcome, TaskSchedulerCompletionDelegate continuation) + => continuation(context, TaskSchedulerOperationOutcome.FromException(_replacement)); + } + + private sealed class ThrowingCompletionMiddleware : ITaskSchedulerCompletionMiddleware + { + private readonly Exception _exception; + private readonly bool _throwAfterContinuation; + public ThrowingCompletionMiddleware(Exception exception, bool throwAfterContinuation) + { + _exception = exception; + _throwAfterContinuation = throwAfterContinuation; + } + + public async ValueTask> InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerOperationOutcome outcome, TaskSchedulerCompletionDelegate continuation) + { + if (!_throwAfterContinuation) throw _exception; + _ = await continuation(context, outcome); + throw _exception; + } + } + + private sealed class MarkerOnlyMiddleware : ITaskSchedulerMiddleware + { + } + + private sealed class TwiceMiddleware : ITaskSchedulerExecutionMiddleware + { + public async ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation) + { + _ = await continuation(context); + return await continuation(context); + } + } + + private sealed class CompoundStateMiddleware : ITaskSchedulerEnqueueMiddleware, ITaskSchedulerExecutionMiddleware + { + private int _nextId; + public ConcurrentBag ObservedIds { get; } = new ConcurrentBag(); + public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) + { + context.GetOrCreateLocalState(() => new LocalState(Interlocked.Increment(ref _nextId))); + return continuation(context); + } + + public ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation) + { + ObservedIds.Add(context.GetLocalState()!.Id); + return continuation(context); + } + } + + private sealed class LocalState + { + public LocalState(int id) => Id = id; + public int Id { get; } + } + + private sealed class RecordingTerminal : ITaskScheduler + { + private readonly IList _events; + public RecordingTerminal(IList events) => _events = events; + public int EnqueueCount { get; private set; } + public object? State { get; private set; } + public CancellationToken CancellationToken { get; private set; } + + public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + { + EnqueueCount++; + State = state; + CancellationToken = cancellationToken; + _events.Add("terminal"); + var result = await taskFunc(state, cancellationToken); + _events.Add("terminal-return"); + return result; + } + } + + private sealed class InlineTerminal : ITaskScheduler + { + public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + => await taskFunc(state, cancellationToken); + } + + private sealed class ThrowingTerminal : ITaskScheduler + { + private readonly Exception _exception; + public ThrowingTerminal(Exception exception) => _exception = exception; + public Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + => Task.FromException(_exception); + } + } +} diff --git a/TaskFlow.Tests/Extensions/TimeoutTaskSchedulerExtensionsFixture.cs b/TaskFlow.Tests/Extensions/TimeoutTaskSchedulerExtensionsFixture.cs index 7f90066..120582c 100644 --- a/TaskFlow.Tests/Extensions/TimeoutTaskSchedulerExtensionsFixture.cs +++ b/TaskFlow.Tests/Extensions/TimeoutTaskSchedulerExtensionsFixture.cs @@ -75,8 +75,8 @@ public void Timeout_WhenOperationNameSpecified_ShouldThrowTimeoutExceptionWithOp _taskFlow = taskFlow; var task = taskFlow - .WithTimeout(TimeSpan.FromMilliseconds(100)) .WithOperationName("inner") + .WithTimeout(TimeSpan.FromMilliseconds(100)) .CreateCancelPrevious() .WithOperationName("outer") .Enqueue( @@ -88,7 +88,7 @@ public void Timeout_WhenOperationNameSpecified_ShouldThrowTimeoutExceptionWithOp 42, CancellationToken.None); - Assert.That(async () => await task.ConfigureAwait(false), Throws.InstanceOf().With.Message.Contain("outer")); + Assert.That(async () => await task.ConfigureAwait(false), Throws.InstanceOf().With.Message.Contain("inner")); } [TestCaseSource(typeof(TaskFlows), nameof(TaskFlows.CreateTaskFlows))] diff --git a/TaskFlow.Tests/Extensions/WrapperOwnershipTaskSchedulerExtensionsFixture.cs b/TaskFlow.Tests/Extensions/WrapperOwnershipTaskSchedulerExtensionsFixture.cs index 3bfe54c..ab0e070 100644 --- a/TaskFlow.Tests/Extensions/WrapperOwnershipTaskSchedulerExtensionsFixture.cs +++ b/TaskFlow.Tests/Extensions/WrapperOwnershipTaskSchedulerExtensionsFixture.cs @@ -24,6 +24,8 @@ public void ExtensionWrappers_ShouldNotOwnTaskFlow_DisposableInterfacesAreNotExp var cancelPrevious = taskFlow.CreateCancelPrevious(); var cancellationScope = taskFlow.CreateCancellationScope(CancellationToken.None); var intercepted = taskFlow.Intercept(new NoOpAsyncInterceptorFactory()); + var annotated = taskFlow.WithOperationName("non-owning"); + var middleware = taskFlow.UseMiddleware(new NoOpMiddleware()); Assert.That(timeout, Is.Not.InstanceOf()); Assert.That(timeout, Is.Not.InstanceOf()); @@ -39,6 +41,12 @@ public void ExtensionWrappers_ShouldNotOwnTaskFlow_DisposableInterfacesAreNotExp Assert.That(intercepted, Is.Not.InstanceOf()); Assert.That(intercepted, Is.Not.InstanceOf()); + + Assert.That(annotated, Is.Not.InstanceOf()); + Assert.That(annotated, Is.Not.InstanceOf()); + + Assert.That(middleware, Is.Not.InstanceOf()); + Assert.That(middleware, Is.Not.InstanceOf()); } private sealed class NoOpAsyncInterceptorFactory : IAsyncTaskSchedulerInterceptor @@ -71,5 +79,13 @@ public ValueTask OnSuccessAsync(TaskSchedulerInterceptionContext contex return default; } } + + private sealed class NoOpMiddleware : ITaskSchedulerExecutionMiddleware + { + public ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation) + { + return continuation(context); + } + } } } diff --git a/TaskFlow.sln b/TaskFlow.sln index 4586601..d4e8130 100644 --- a/TaskFlow.sln +++ b/TaskFlow.sln @@ -1,7 +1,7 @@  Microsoft Visual Studio Solution File, Format Version 12.00 -# Visual Studio Version 17 -VisualStudioVersion = 17.3.32901.215 +# Visual Studio Version 18 +VisualStudioVersion = 18.9.12105.275 stable MinimumVisualStudioVersion = 10.0.40219.1 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "TaskFlow", "TaskFlow\TaskFlow.csproj", "{7B846A17-1146-4BF1-9E01-81AAFB097A02}" EndProject @@ -68,6 +68,7 @@ Global GlobalSection(SharedMSBuildProjectFiles) = preSolution TaskFlow.Annotations\TaskFlow.Annotations.projitems*{38b4a885-e50e-4339-91f6-b07976dce941}*SharedItemsImports = 5 TaskFlow.Annotations\TaskFlow.Annotations.projitems*{7b846a17-1146-4bf1-9e01-81aafb097a02}*SharedItemsImports = 5 + TaskFlow.Annotations\TaskFlow.Annotations.projitems*{a8d73e43-4792-4a8d-a57a-2df8e91ccf88}*SharedItemsImports = 5 TaskFlow.Annotations\TaskFlow.Annotations.projitems*{b9a7fa49-0771-408d-82cf-42ea7ebe0f31}*SharedItemsImports = 13 TaskFlow.Annotations\TaskFlow.Annotations.projitems*{dd1d5309-b058-4563-8bb3-96b93ab7731f}*SharedItemsImports = 5 EndGlobalSection diff --git a/TaskFlow/Extensions/AnnotatingTaskSchedulerExtensions.cs b/TaskFlow/Extensions/AnnotatingTaskSchedulerExtensions.cs index f920562..b250628 100644 --- a/TaskFlow/Extensions/AnnotatingTaskSchedulerExtensions.cs +++ b/TaskFlow/Extensions/AnnotatingTaskSchedulerExtensions.cs @@ -111,7 +111,7 @@ public static ITaskScheduler WithOperationName(this ITaskScheduler taskScheduler { Argument.NotEmpty(operationName); - return taskScheduler.WithExtendedState(new OperationNameAnnotation( operationName)); + return taskScheduler.WithAnnotation(new OperationNameAnnotation(operationName)); } /// @@ -180,11 +180,10 @@ public static Task AnnotatedEnqueue(this ITaskScheduler taskS Argument.NotNull(taskScheduler); Argument.NotNull(taskFunc); - var annotation = (state as ExtendedState) - .Unwrap() - .FirstOrDefault(); - - return taskScheduler.Enqueue((s, c) => taskFunc(s, annotation, c), state, cancellationToken); + return taskScheduler.EnqueueWithContext( + context => taskFunc(context.State, context.GetAnnotation(), context.CancellationToken), + state, + cancellationToken); } } -} \ No newline at end of file +} diff --git a/TaskFlow/Extensions/CancelPreviousTaskSchedulerExtensions.cs b/TaskFlow/Extensions/CancelPreviousTaskSchedulerExtensions.cs index 68a326f..7a2880f 100644 --- a/TaskFlow/Extensions/CancelPreviousTaskSchedulerExtensions.cs +++ b/TaskFlow/Extensions/CancelPreviousTaskSchedulerExtensions.cs @@ -183,30 +183,27 @@ public static class CancelPreviousTaskSchedulerExtensions /// public static ITaskScheduler CreateCancelPrevious(this ITaskScheduler taskScheduler) { - return new CancelPreviousTaskSchedulerWrapper(taskScheduler); + Argument.NotNull(taskScheduler); + return taskScheduler.UseMiddleware(new CancelPreviousMiddleware()); } - private sealed class CancelPreviousTaskSchedulerWrapper : ITaskScheduler + private sealed class CancelPreviousMiddleware : ITaskSchedulerEnqueueMiddleware { - private readonly ITaskScheduler _baseTaskScheduler; private readonly CancelAllTokensAllocator _cancelAllTokensAllocator; - public CancelPreviousTaskSchedulerWrapper(ITaskScheduler baseTaskScheduler) + public CancelPreviousMiddleware() { - Argument.NotNull(baseTaskScheduler); - - _baseTaskScheduler = baseTaskScheduler; _cancelAllTokensAllocator = new CancelAllTokensAllocator(); } - public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + public async Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) { _cancelAllTokensAllocator.Cancel(); using (_cancelAllTokensAllocator.AllocateCancellationToken(out var allocatedToken)) { - using var linkedToken = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, allocatedToken); - return await _baseTaskScheduler.Enqueue(taskFunc, state, linkedToken.Token).ConfigureAwait(false); + using var linkedToken = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken, allocatedToken); + return await continuation(context.WithCancellationToken(linkedToken.Token)).ConfigureAwait(false); } } } diff --git a/TaskFlow/Extensions/CancellationScopeTaskSchedulerExtensions.cs b/TaskFlow/Extensions/CancellationScopeTaskSchedulerExtensions.cs index 1ce38f8..ff09c7c 100644 --- a/TaskFlow/Extensions/CancellationScopeTaskSchedulerExtensions.cs +++ b/TaskFlow/Extensions/CancellationScopeTaskSchedulerExtensions.cs @@ -137,27 +137,24 @@ public static class CancellationScopeTaskSchedulerExtensions /// public static ITaskScheduler CreateCancellationScope(this ITaskScheduler taskScheduler, CancellationToken scopeCancellationToken) { - return new CancellationScopeTaskSchedulerWrapper(taskScheduler, scopeCancellationToken); + Argument.NotNull(taskScheduler); + return taskScheduler.UseMiddleware(new CancellationScopeMiddleware(scopeCancellationToken)); } - private sealed class CancellationScopeTaskSchedulerWrapper : ITaskScheduler + private sealed class CancellationScopeMiddleware : ITaskSchedulerEnqueueMiddleware { - private readonly ITaskScheduler _baseTaskScheduler; private readonly CancellationToken _scopedCancellationToken; - public CancellationScopeTaskSchedulerWrapper(ITaskScheduler baseTaskScheduler, CancellationToken scopedCancellationToken) + public CancellationScopeMiddleware(CancellationToken scopedCancellationToken) { - Argument.NotNull(baseTaskScheduler); - - _baseTaskScheduler = baseTaskScheduler; _scopedCancellationToken = scopedCancellationToken; } - public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + public async Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) { - using var linkedToken = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _scopedCancellationToken); - return await _baseTaskScheduler.Enqueue(taskFunc, state, linkedToken.Token).ConfigureAwait(false); + using var linkedToken = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken, _scopedCancellationToken); + return await continuation(context.WithCancellationToken(linkedToken.Token)).ConfigureAwait(false); } } } -} \ No newline at end of file +} diff --git a/TaskFlow/Extensions/ExceptionTaskSchedulerExtensions.cs b/TaskFlow/Extensions/ExceptionTaskSchedulerExtensions.cs index 504caca..62e8a42 100644 --- a/TaskFlow/Extensions/ExceptionTaskSchedulerExtensions.cs +++ b/TaskFlow/Extensions/ExceptionTaskSchedulerExtensions.cs @@ -120,7 +120,7 @@ public static class ExceptionTaskSchedulerExtensions public static ITaskScheduler OnError(this ITaskScheduler taskScheduler, Action errorAction, Func? errorFilter = null) where TException : Exception { - return new AnnotatedExceptionTaskSchedulerWrapper(taskScheduler, errorFilter ?? DefaultErrorFilter, (scheduler, exception, _) => errorAction(scheduler, exception)); + return taskScheduler.UseMiddleware(new AnnotatedExceptionMiddleware(errorFilter ?? DefaultErrorFilter, (scheduler, exception, _) => errorAction(scheduler, exception))); } /// @@ -141,7 +141,7 @@ public static ITaskScheduler OnError(this ITaskScheduler taskSchedul { Argument.NotNull(errorAction); - return new AnnotatedExceptionTaskSchedulerWrapper(taskScheduler, errorFilter ?? DefaultErrorFilter, (_, exception, _) => errorAction(exception)); + return taskScheduler.UseMiddleware(new AnnotatedExceptionMiddleware(errorFilter ?? DefaultErrorFilter, (_, exception, _) => errorAction(exception))); } /// @@ -192,7 +192,7 @@ public static ITaskScheduler OnError(this ITaskScheduler taskScheduler, Action(this ITaskScheduler taskScheduler, Action errorAction, Func? errorFilter = null) where TException : Exception { - return new AnnotatedExceptionTaskSchedulerWrapper(taskScheduler, errorFilter ?? DefaultErrorFilter, errorAction); + return taskScheduler.UseMiddleware(new AnnotatedExceptionMiddleware(errorFilter ?? DefaultErrorFilter, errorAction)); } /// @@ -214,7 +214,7 @@ public static ITaskScheduler OnError(this ITaskSchedule where TException : Exception where TAnnotation : IOperationAnnotation { - return new AnnotatedExceptionTaskSchedulerWrapper(taskScheduler, errorFilter ?? DefaultErrorFilter, errorAction); + return taskScheduler.UseMiddleware(new AnnotatedExceptionMiddleware(errorFilter ?? DefaultErrorFilter, errorAction)); } private static bool DefaultErrorFilter(TException _) @@ -223,47 +223,45 @@ private static bool DefaultErrorFilter(TException _) return true; } - private sealed class AnnotatedExceptionTaskSchedulerWrapper : ITaskScheduler + private sealed class AnnotatedExceptionMiddleware : ITaskSchedulerCompletionMiddleware where TException : Exception where TAnnotation : IOperationAnnotation { - private readonly ITaskScheduler _baseTaskScheduler; private readonly Func _errorFilter; private readonly Action _errorAction; - public AnnotatedExceptionTaskSchedulerWrapper( - ITaskScheduler baseTaskScheduler, + public AnnotatedExceptionMiddleware( Func errorFilter, Action errorAction) { - Argument.NotNull(baseTaskScheduler); Argument.NotNull(errorFilter); Argument.NotNull(errorAction); - _baseTaskScheduler = baseTaskScheduler; _errorFilter = errorFilter; _errorAction = errorAction; } - public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + public async ValueTask> InvokeAsync( + TaskSchedulerOperationContext context, + TaskSchedulerOperationOutcome outcome, + TaskSchedulerCompletionDelegate continuation) { TAnnotation? annotation = default; if (!typeof(TAnnotation).IsInterface) { - annotation = (state as ExtendedState) - .Unwrap() - .FirstOrDefault(); + var value = context.GetAnnotation(typeof(TAnnotation)); + if (value != null) + { + annotation = (TAnnotation)value; + } } - try + if (outcome.Exception is TException exception && _errorFilter(exception)) { - return await _baseTaskScheduler.Enqueue(taskFunc, state, cancellationToken).ConfigureAwait(false); - } - catch (TException exception) when (_errorFilter(exception)) - { - _errorAction(this, exception, annotation); - throw; + _errorAction(context.Scheduler, exception, annotation); } + + return await continuation(context, outcome).ConfigureAwait(true); } } } diff --git a/TaskFlow/Extensions/InterceptionTaskSchedulerExtensions.cs b/TaskFlow/Extensions/InterceptionTaskSchedulerExtensions.cs index c6f4387..c374405 100644 --- a/TaskFlow/Extensions/InterceptionTaskSchedulerExtensions.cs +++ b/TaskFlow/Extensions/InterceptionTaskSchedulerExtensions.cs @@ -1,94 +1,63 @@ namespace System.Threading.Tasks.Flow { + using System.Threading.Tasks; using System.Threading.Tasks.Flow.Annotations; /// Provides synchronous and asynchronous operation interception for . public static class InterceptionTaskSchedulerExtensions { - /// Wraps a scheduler with a synchronous value-type interceptor. - /// The value type that implements the lifecycle callbacks. - /// The scheduler whose operations will be intercepted. - /// The interceptor template to copy for each operation. - /// A scheduler that invokes around every operation. - /// Thrown when is null. + /// Registers a synchronous value-type interceptor in the scheduled execution phase. + /// The value type copied to create isolated state for each operation. + /// The scheduler or pipeline snapshot to extend. + /// The interceptor template copied for each operation. + /// A new immutable, non-owning scheduler snapshot. + /// is null. /// - /// - /// A separate copy of is made when each operation starts. Mutable fields on the - /// copy may be used as allocation-free per-operation state and are preserved across all lifecycle callbacks. - /// - /// - /// Interceptor callbacks run on the selected TaskFlow scheduler context. A callback exception follows the - /// replacement semantics documented by . - /// + /// Callbacks run on the selected terminal scheduler context in before, success/error, and finally order. + /// Mutable fields belong to the per-operation struct copy. Callback exceptions follow the replacement + /// semantics documented by . /// - /// - /// - /// var intercepted = scheduler.Intercept(new TimingInterceptor()); - /// var result = await intercepted.Enqueue(() => ComputeAsync()); - /// - /// public static ITaskScheduler Intercept(this ITaskScheduler taskScheduler, TInterceptor interceptor) where TInterceptor : struct, ITaskSchedulerInterceptor { Argument.NotNull(taskScheduler); - return new InterceptionTaskSchedulerWrapper(taskScheduler, interceptor); + return taskScheduler.UseMiddleware(new SynchronousInterceptionMiddleware(interceptor)); } - /// Wraps a scheduler with a factory for asynchronous per-operation interceptors. - /// The scheduler whose operations will be intercepted. - /// The factory that creates an interceptor for each operation. - /// A scheduler that asynchronously intercepts every operation. - /// - /// Thrown when or is null. - /// + /// Registers a factory for asynchronous per-operation interceptors in the scheduled execution phase. + /// The scheduler or pipeline snapshot to extend. + /// The factory that creates one asynchronous interceptor per operation. + /// A new immutable, non-owning scheduler snapshot. + /// An argument is null. /// - /// - /// The factory is invoked once per operation on the selected scheduler. Every lifecycle - /// is awaited before execution advances, and continuations retain the TaskFlow synchronization context. - /// - /// - /// The returned scheduler preserves the enqueue operation's cancellation and exception behavior when all - /// interceptor callbacks complete successfully. - /// + /// The factory and every lifecycle callback run inside the terminal scheduler delegate. Returned + /// instances are awaited with scheduler-context capture so asynchronous callbacks, + /// the user operation, and later completion middleware retain the selected synchronization context. /// public static ITaskScheduler Intercept(this ITaskScheduler taskScheduler, IAsyncTaskSchedulerInterceptor interceptor) { Argument.NotNull(taskScheduler); Argument.NotNull(interceptor); - return new AsyncInterceptionTaskSchedulerWrapper(taskScheduler, interceptor); + return taskScheduler.UseMiddleware(new AsyncInterceptionMiddleware(interceptor)); } - private sealed class InterceptionTaskSchedulerWrapper : ITaskScheduler + private sealed class SynchronousInterceptionMiddleware : ITaskSchedulerExecutionMiddleware where TInterceptor : struct, ITaskSchedulerInterceptor { - private readonly ITaskScheduler _baseTaskScheduler; private readonly TInterceptor _interceptor; + public SynchronousInterceptionMiddleware(TInterceptor interceptor) => _interceptor = interceptor; - public InterceptionTaskSchedulerWrapper(ITaskScheduler baseTaskScheduler, TInterceptor interceptor) + public async ValueTask InvokeAsync(TaskSchedulerOperationContext operationContext, TaskSchedulerExecutionDelegate continuation) { - _baseTaskScheduler = baseTaskScheduler; - _interceptor = interceptor; - } - - public Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) - { - var context = new TaskSchedulerInterceptionContext(state, cancellationToken); - return _baseTaskScheduler.Enqueue((_, token) => Execute(taskFunc, state, context, token), state, cancellationToken); - } - - private async ValueTask Execute(Func> taskFunc, object? state, TaskSchedulerInterceptionContext context, CancellationToken token) - { - // A value-type assignment creates the independent interceptor instance used by this operation. var interceptor = _interceptor; + var context = new TaskSchedulerInterceptionContext(operationContext); try { interceptor.OnBefore(context); - - T result; + TResult result; try { - // Keep the TaskFlow synchronization context so subsequent interceptor callbacks run on the selected scheduler. - result = await taskFunc(state, token).ConfigureAwait(true); + result = await continuation(operationContext).ConfigureAwait(true); } catch (Exception exception) { @@ -106,34 +75,22 @@ private async ValueTask Execute(Func Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) - { - var context = new TaskSchedulerInterceptionContext(state, cancellationToken); - return _baseTaskScheduler.Enqueue((_, token) => Execute(taskFunc, state, context, token), state, cancellationToken); - } + private readonly IAsyncTaskSchedulerInterceptor _factory; + public AsyncInterceptionMiddleware(IAsyncTaskSchedulerInterceptor factory) => _factory = factory; - private async ValueTask Execute(Func> taskFunc, object? state, TaskSchedulerInterceptionContext context, CancellationToken token) + public async ValueTask InvokeAsync(TaskSchedulerOperationContext operationContext, TaskSchedulerExecutionDelegate continuation) { - var interceptor = _interceptor.CreateInterceptor(context); + var context = new TaskSchedulerInterceptionContext(operationContext); + var interceptor = _factory.CreateInterceptor(context); try { - // ConfigureAwait(false) could move execution to the thread pool and make the operation or later callbacks leave the selected scheduler. await interceptor.OnBeforeAsync(context).ConfigureAwait(true); - - T result; + TResult result; try { - result = await taskFunc(state, token).ConfigureAwait(true); + result = await continuation(operationContext).ConfigureAwait(true); } catch (Exception exception) { diff --git a/TaskFlow/Extensions/TaskSchedulerInterceptionContext.cs b/TaskFlow/Extensions/TaskSchedulerInterceptionContext.cs index 4864fc0..cc22585 100644 --- a/TaskFlow/Extensions/TaskSchedulerInterceptionContext.cs +++ b/TaskFlow/Extensions/TaskSchedulerInterceptionContext.cs @@ -12,14 +12,24 @@ namespace System.Threading.Tasks.Flow public readonly struct TaskSchedulerInterceptionContext { private readonly ExtendedState? _extendedState; + private readonly TaskSchedulerOperationContext? _operationContext; internal TaskSchedulerInterceptionContext(object? state, CancellationToken cancellationToken) { _extendedState = state as ExtendedState; + _operationContext = null; State = UnwrapState(state); CancellationToken = cancellationToken; } + internal TaskSchedulerInterceptionContext(TaskSchedulerOperationContext context) + { + _extendedState = null; + _operationContext = context; + State = context.State; + CancellationToken = context.CancellationToken; + } + /// Gets the original operation state after TaskFlow extended-state wrappers have been removed. /// The caller-provided state, or null when no state was supplied. public object? State { get; } @@ -35,7 +45,7 @@ internal TaskSchedulerInterceptionContext(object? state, CancellationToken cance public TAnnotation? GetAnnotation() where TAnnotation : class, IOperationAnnotation { - return GetAnnotation(_extendedState); + return _operationContext?.GetAnnotation() ?? GetAnnotation(_extendedState); } /// Gets the first annotation of the requested type from raw scheduler state. diff --git a/TaskFlow/Extensions/ThrottlingTaskSchedulerExtensions.cs b/TaskFlow/Extensions/ThrottlingTaskSchedulerExtensions.cs index e5b385f..2493b65 100644 --- a/TaskFlow/Extensions/ThrottlingTaskSchedulerExtensions.cs +++ b/TaskFlow/Extensions/ThrottlingTaskSchedulerExtensions.cs @@ -126,12 +126,12 @@ public static class ThrottlingTaskSchedulerExtensions /// public static ITaskScheduler WithThrottle(this ITaskScheduler taskScheduler, TimeSpan interval, TimeProvider? timeProvider = null) { - return new ThrottleTaskSchedulerWrapper(taskScheduler, timeProvider ?? TimeProvider.System, interval); + Argument.NotNull(taskScheduler); + return taskScheduler.UseMiddleware(new ThrottleMiddleware(timeProvider ?? TimeProvider.System, interval)); } - private sealed class ThrottleTaskSchedulerWrapper : ITaskScheduler + private sealed class ThrottleMiddleware : ITaskSchedulerEnqueueMiddleware { - private readonly ITaskScheduler _baseTaskScheduler; private readonly TimeProvider _timeProvider; private readonly TimeSpan _interval; private readonly object _lastAdmissionLock; @@ -139,19 +139,17 @@ private sealed class ThrottleTaskSchedulerWrapper : ITaskScheduler private long _lastAdmissionTimestamp; private bool _hasAdmission; - public ThrottleTaskSchedulerWrapper(ITaskScheduler baseTaskScheduler, TimeProvider timeProvider, TimeSpan interval) + public ThrottleMiddleware(TimeProvider timeProvider, TimeSpan interval) { - Argument.NotNull(baseTaskScheduler); Argument.NotNull(timeProvider); Argument.Assert(interval, ts => ts > TimeSpan.Zero, "Interval should be greater than zero"); - _baseTaskScheduler = baseTaskScheduler; _interval = interval; _timeProvider = timeProvider; _lastAdmissionLock = new object(); } - public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) { lock (_lastAdmissionLock) { @@ -166,7 +164,7 @@ public async Task Enqueue(Func> t _hasAdmission = true; } - return await _baseTaskScheduler.Enqueue(taskFunc, state, cancellationToken).ConfigureAwait(false); + return continuation(context); } } } diff --git a/TaskFlow/Extensions/TimeoutTaskSchedulerExtensions.cs b/TaskFlow/Extensions/TimeoutTaskSchedulerExtensions.cs index 823da41..729c0a8 100644 --- a/TaskFlow/Extensions/TimeoutTaskSchedulerExtensions.cs +++ b/TaskFlow/Extensions/TimeoutTaskSchedulerExtensions.cs @@ -212,43 +212,38 @@ public static class TimeoutTaskSchedulerExtensions /// public static ITaskScheduler WithTimeout(this ITaskScheduler taskScheduler, TimeSpan timeout) { - return new TimeoutTaskSchedulerWrapper(taskScheduler, timeout); + Argument.NotNull(taskScheduler); + return taskScheduler.UseMiddleware(new TimeoutMiddleware(timeout)); } - private sealed class TimeoutTaskSchedulerWrapper : ITaskScheduler + private sealed class TimeoutMiddleware : ITaskSchedulerEnqueueMiddleware { - private readonly ITaskScheduler _baseTaskScheduler; private readonly TimeSpan _timeout; - public TimeoutTaskSchedulerWrapper(ITaskScheduler baseTaskScheduler, TimeSpan timeout) + public TimeoutMiddleware(TimeSpan timeout) { - Argument.NotNull(baseTaskScheduler); Argument.Assert(timeout, t => t > TimeSpan.Zero && t.TotalMilliseconds <= int.MaxValue || t == Timeout.InfiniteTimeSpan, "Wrong timeout value"); - - _baseTaskScheduler = baseTaskScheduler; _timeout = timeout; } - public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + public async Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) { - return await Internal.TaskExtensions.WhenAnyCancelRest(new[] { TimeoutAsync, EnqueueInternalAsync }, cancellationToken).ConfigureAwait(false); + return await Internal.TaskExtensions.WhenAnyCancelRest(new[] { TimeoutAsync, EnqueueInternalAsync }, context.CancellationToken).ConfigureAwait(false); - async Task TimeoutAsync(CancellationToken token) + async Task TimeoutAsync(CancellationToken token) { await Task.Delay(_timeout, token).ConfigureAwait(false); throw new TimeoutException(FormatExceptionMessage()); } - async Task EnqueueInternalAsync(CancellationToken token) + Task EnqueueInternalAsync(CancellationToken token) { - return await _baseTaskScheduler.Enqueue(taskFunc, state, token).ConfigureAwait(false); + return continuation(context.WithCancellationToken(token)); } string FormatExceptionMessage() { - var operationName = (state as ExtendedState) - .Unwrap() - .FirstOrDefault()?.OperationName; + var operationName = context.GetAnnotation()?.OperationName; return operationName == null ? string.Format(CultureInfo.InvariantCulture, "Operation has timed out in {0}", _timeout) diff --git a/TaskFlow/Internal/AnnotationScope.cs b/TaskFlow/Internal/AnnotationScope.cs new file mode 100644 index 0000000..83f37f1 --- /dev/null +++ b/TaskFlow/Internal/AnnotationScope.cs @@ -0,0 +1,38 @@ +namespace System.Threading.Tasks.Flow.Internal +{ + using System; + + internal sealed class AnnotationScope + { + private readonly AnnotationScope? _parent; + private readonly Type _type; + private readonly IOperationAnnotation _annotation; + + public AnnotationScope(AnnotationScope? parent, Type type, IOperationAnnotation annotation) + { + _parent = parent; + _type = type; + _annotation = annotation; + } + + public T? Get() where T : class, IOperationAnnotation + { + for (var current = this; current != null; current = current._parent) + { + if (current._type == typeof(T)) return (T)current._annotation; + } + + return null; + } + + public IOperationAnnotation? Get(Type type) + { + for (var current = this; current != null; current = current._parent) + { + if (current._type == type) return current._annotation; + } + + return null; + } + } +} diff --git a/TaskFlow/Internal/MiddlewareRegistration.cs b/TaskFlow/Internal/MiddlewareRegistration.cs new file mode 100644 index 0000000..cdbafd5 --- /dev/null +++ b/TaskFlow/Internal/MiddlewareRegistration.cs @@ -0,0 +1,15 @@ +namespace System.Threading.Tasks.Flow.Internal +{ + internal sealed class MiddlewareRegistration + { + public MiddlewareRegistration(object middleware, AnnotationScope? annotations) + { + Middleware = middleware; + Annotations = annotations; + } + + public object Middleware { get; } + public AnnotationScope? Annotations { get; } + public ITaskScheduler Scheduler { get; set; } = null!; + } +} diff --git a/TaskFlow/Internal/PipelineOperation.cs b/TaskFlow/Internal/PipelineOperation.cs new file mode 100644 index 0000000..e95fc8b --- /dev/null +++ b/TaskFlow/Internal/PipelineOperation.cs @@ -0,0 +1,78 @@ +namespace System.Threading.Tasks.Flow.Internal +{ + using System; + using System.Threading; + using System.Threading.Tasks; + + internal abstract class PipelineOperation + { + private readonly object?[] _localStates; + private readonly object _localStatesLock = new object(); + private int _completionClaimed; + + protected PipelineOperation(object? state, int registrationCount, CancellationToken callerCancellationToken) + { + State = state; + CallerCancellationToken = callerCancellationToken; + ProducerCancellationToken = callerCancellationToken; + _localStates = new object?[registrationCount]; + } + + public object? State { get; } + public CancellationToken CallerCancellationToken { get; } + public CancellationToken ProducerCancellationToken { get; set; } + + public T? GetLocalState(int index) where T : class + { + var value = Volatile.Read(ref _localStates[index]); + if (value == null) + { + return null; + } + + return value as T ?? throw new InvalidOperationException("The middleware registration-local state has a different type."); + } + + public T GetOrCreateLocalState(int index, Func factory) where T : class + { + var existing = GetLocalState(index); + if (existing != null) + { + return existing; + } + + lock (_localStatesLock) + { + existing = GetLocalState(index); + if (existing != null) + { + return existing; + } + + var created = factory() ?? throw new InvalidOperationException("A registration-local state factory cannot return null."); + _localStates[index] = created; + return created; + } + } + + public bool TryClaimCompletion() => Interlocked.CompareExchange(ref _completionClaimed, 1, 0) == 0; + } + + internal sealed class PipelineOperation : PipelineOperation + { + public PipelineOperation( + Func> taskFunc, + object? state, + int registrationCount, + AnnotationScope? finalAnnotations, + CancellationToken callerCancellationToken) + : base(state, registrationCount, callerCancellationToken) + { + TaskFunc = taskFunc; + FinalAnnotations = finalAnnotations; + } + + public Func> TaskFunc { get; } + public AnnotationScope? FinalAnnotations { get; } + } +} diff --git a/TaskFlow/Internal/TaskSchedulerPipeline.cs b/TaskFlow/Internal/TaskSchedulerPipeline.cs new file mode 100644 index 0000000..51191ad --- /dev/null +++ b/TaskFlow/Internal/TaskSchedulerPipeline.cs @@ -0,0 +1,171 @@ +namespace System.Threading.Tasks.Flow.Internal +{ + using System; + using System.Diagnostics.CodeAnalysis; + using System.Threading; + using System.Threading.Tasks; + + internal sealed class PipelineTaskScheduler : ITaskScheduler + { + private readonly ITaskScheduler _terminal; + private readonly MiddlewareRegistration[] _registrations; + private readonly AnnotationScope? _annotations; + + public PipelineTaskScheduler(ITaskScheduler terminal) + : this(terminal, Array.Empty(), null) + { + } + + private PipelineTaskScheduler(ITaskScheduler terminal, MiddlewareRegistration[] registrations, AnnotationScope? annotations) + { + _terminal = terminal; + _registrations = registrations; + _annotations = annotations; + } + + public PipelineTaskScheduler Append(object middleware) + { + var registrations = new MiddlewareRegistration[_registrations.Length + 1]; + Array.Copy(_registrations, registrations, _registrations.Length); + var registration = new MiddlewareRegistration(middleware, _annotations); + registrations[registrations.Length - 1] = registration; + var snapshot = new PipelineTaskScheduler(_terminal, registrations, _annotations); + registration.Scheduler = snapshot; + return snapshot; + } + + public PipelineTaskScheduler WithAnnotation(Type type, IOperationAnnotation annotation) + => new PipelineTaskScheduler(_terminal, _registrations, new AnnotationScope(_annotations, type, annotation)); + + public Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + { + Annotations.Argument.NotNull(taskFunc); + var operation = new PipelineOperation(taskFunc, state, _registrations.Length, _annotations, cancellationToken); + return EnqueueCore(operation); + } + + private async Task EnqueueCore(PipelineOperation operation) + { + try + { + return await InvokeEnqueue(operation, _registrations.Length - 1, operation.CallerCancellationToken).ConfigureAwait(false); + } + catch (Exception exception) + { + if (!operation.TryClaimCompletion()) + { + throw; + } + + var outcome = await InvokeCompletion(operation, 0, TaskSchedulerOperationOutcome.FromException(exception), operation.CallerCancellationToken).ConfigureAwait(false); + return outcome.GetResultOrThrow(); + } + } + + private Task InvokeEnqueue(PipelineOperation operation, int index, CancellationToken cancellationToken) + { + while (index >= 0 && !(_registrations[index].Middleware is ITaskSchedulerEnqueueMiddleware)) + { + index--; + } + + if (index < 0) + { + return ScheduleTerminal(operation, cancellationToken); + } + + var registrationIndex = index; + var registration = _registrations[index]; + var middleware = (ITaskSchedulerEnqueueMiddleware)registration.Middleware; + var context = new TaskSchedulerEnqueueContext(operation, registration.Annotations, registrationIndex, cancellationToken); + return middleware.InvokeAsync(context, nextContext => InvokeEnqueue(operation, registrationIndex - 1, nextContext.CancellationToken)); + } + + private Task ScheduleTerminal(PipelineOperation operation, CancellationToken cancellationToken) + { + operation.ProducerCancellationToken = cancellationToken; + return _terminal.Enqueue((_, token) => ExecuteEnvelope(operation, token), operation.State, cancellationToken); + } + + [SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "The pipeline converts every operation failure into a completion outcome.")] + private async ValueTask ExecuteEnvelope(PipelineOperation operation, CancellationToken cancellationToken) + { + TaskSchedulerOperationOutcome outcome; + try + { + var result = await InvokeExecution(operation, 0, cancellationToken).ConfigureAwait(true); + outcome = TaskSchedulerOperationOutcome.FromResult(result); + } + catch (Exception exception) + { + outcome = TaskSchedulerOperationOutcome.FromException(exception); + } + + if (operation.TryClaimCompletion()) + { + outcome = await InvokeCompletion(operation, 0, outcome, cancellationToken).ConfigureAwait(true); + } + + return outcome.GetResultOrThrow(); + } + + private ValueTask InvokeExecution(PipelineOperation operation, int index, CancellationToken cancellationToken) + { + while (index < _registrations.Length && !(_registrations[index].Middleware is ITaskSchedulerExecutionMiddleware)) + { + index++; + } + + if (index >= _registrations.Length) + { + var finalContext = new TaskSchedulerOperationContext(operation, operation.FinalAnnotations, -1, cancellationToken); + return operation.TaskFunc(operation.State, finalContext.CancellationToken); + } + + var registrationIndex = index; + var registration = _registrations[index]; + var middleware = (ITaskSchedulerExecutionMiddleware)registration.Middleware; + var context = new TaskSchedulerOperationContext(operation, registration.Annotations, registrationIndex, operation.ProducerCancellationToken, registration.Scheduler); + return middleware.InvokeAsync(context, nextContext => InvokeExecution(operation, registrationIndex + 1, nextContext.CancellationToken)); + } + + [SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "Completion middleware failures replace the current outcome by contract.")] + private async ValueTask> InvokeCompletion( + PipelineOperation operation, + int index, + TaskSchedulerOperationOutcome outcome, + CancellationToken cancellationToken) + { + while (index < _registrations.Length && !(_registrations[index].Middleware is ITaskSchedulerCompletionMiddleware)) + { + index++; + } + + if (index >= _registrations.Length) + { + return outcome; + } + + var registrationIndex = index; + var registration = _registrations[index]; + var middleware = (ITaskSchedulerCompletionMiddleware)registration.Middleware; + var context = new TaskSchedulerOperationContext(operation, registration.Annotations, registrationIndex, operation.ProducerCancellationToken, registration.Scheduler); + var nextCalled = false; + try + { + return await middleware.InvokeAsync(context, outcome, async (nextContext, nextOutcome) => + { + nextCalled = true; + return await InvokeCompletion(operation, registrationIndex + 1, nextOutcome, nextContext.CancellationToken).ConfigureAwait(true); + }).ConfigureAwait(true); + } + catch (Exception exception) + { + var replacement = TaskSchedulerOperationOutcome.FromException(exception); + return nextCalled + ? replacement + : await InvokeCompletion(operation, registrationIndex + 1, replacement, cancellationToken).ConfigureAwait(true); + } + } + } +} diff --git a/TaskFlow/Middleware/ITaskSchedulerCompletionMiddleware.cs b/TaskFlow/Middleware/ITaskSchedulerCompletionMiddleware.cs new file mode 100644 index 0000000..7e9fa4f --- /dev/null +++ b/TaskFlow/Middleware/ITaskSchedulerCompletionMiddleware.cs @@ -0,0 +1,32 @@ +namespace System.Threading.Tasks.Flow +{ + using System.Threading.Tasks; + + /// Defines middleware that observes, handles, or replaces an operation's success or failure outcome. + /// + /// Completion middleware is invoked in registration order after execution, or after an enqueue or terminal submission failure. + /// Execution failures run completion on the terminal context. Failures that occur before the terminal invokes the operation cannot guarantee + /// terminal thread or synchronization-context affinity. + /// + public interface ITaskSchedulerCompletionMiddleware : ITaskSchedulerMiddleware + { + /// Processes the current outcome and continues completion processing. + /// The operation result type. + /// Immutable operation facts, effective cancellation, captured annotations, and this registration's local state. + /// The success or failure produced by execution or preceding completion middleware. + /// The remaining completion pipeline. + /// A value task containing the outcome that should ultimately complete the scheduled task. + /// + /// Pass unchanged to preserve the identity and captured dispatch information of an existing exception. Pass an + /// outcome created by or + /// to intentionally replace it. + /// Middleware should normally invoke exactly once. If this method throws before invoking it, the thrown + /// exception replaces the current outcome and later completion middleware still runs. If it throws after invoking it, later middleware is not + /// repeated and the thrown exception becomes the final failure. + /// + ValueTask> InvokeAsync( + TaskSchedulerOperationContext context, + TaskSchedulerOperationOutcome outcome, + TaskSchedulerCompletionDelegate continuation); + } +} diff --git a/TaskFlow/Middleware/ITaskSchedulerEnqueueMiddleware.cs b/TaskFlow/Middleware/ITaskSchedulerEnqueueMiddleware.cs new file mode 100644 index 0000000..20da15d --- /dev/null +++ b/TaskFlow/Middleware/ITaskSchedulerEnqueueMiddleware.cs @@ -0,0 +1,26 @@ +namespace System.Threading.Tasks.Flow +{ + using System.Threading.Tasks; + + /// Defines admission middleware that runs before an operation reaches the terminal scheduler. + /// + /// Enqueue middleware is invoked in reverse registration order, so the most recently registered enqueue component runs first. + /// Use this phase for admission, rejection, coalescing, throttling, or producer cancellation-token selection. Invoke the continuation at most + /// once. Returning a task without invoking it short-circuits the remaining enqueue pipeline and the terminal scheduler. + /// + public interface ITaskSchedulerEnqueueMiddleware : ITaskSchedulerMiddleware + { + /// Processes an operation before it is submitted to the terminal scheduler. + /// The operation result type. + /// Immutable operation facts, the effective producer token, captured annotations, and this registration's local state. + /// The remaining enqueue pipeline. Pass the original context or a copy returned by + /// . + /// A task representing admission and, when the continuation is invoked, terminal scheduling and operation completion. + /// + /// 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. + /// + Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation); + } +} diff --git a/TaskFlow/Middleware/ITaskSchedulerExecutionMiddleware.cs b/TaskFlow/Middleware/ITaskSchedulerExecutionMiddleware.cs new file mode 100644 index 0000000..a8a9dae --- /dev/null +++ b/TaskFlow/Middleware/ITaskSchedulerExecutionMiddleware.cs @@ -0,0 +1,25 @@ +namespace System.Threading.Tasks.Flow +{ + using System.Threading.Tasks; + + /// Defines middleware that surrounds the user operation inside the delegate invoked by the terminal scheduler. + /// + /// Execution middleware is invoked in registration order and runs on the execution context established by the terminal scheduler. + /// Use this phase for timing, tracing, resource scopes, retries, or other behavior that must execute with the scheduled operation. + /// + public interface ITaskSchedulerExecutionMiddleware : ITaskSchedulerMiddleware + { + /// Invokes, replaces, or repeatedly orchestrates the remaining execution pipeline. + /// The operation result type. + /// Immutable operation facts, effective cancellation, captured annotations, and this registration's local state. + /// The next execution middleware or the user operation. + /// A value task containing the result selected by this middleware. + /// + /// Ordinary middleware should invoke once. Orchestration middleware may invoke it multiple times—for + /// example, to retry—without enqueueing another terminal delegate. Returning without invoking it replaces the user operation. + /// A thrown exception or a faulted value task becomes a failed and is passed to + /// completion middleware. Use ConfigureAwait(true) when an asynchronous continuation must retain terminal affinity. + /// + ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation); + } +} diff --git a/TaskFlow/Middleware/ITaskSchedulerMiddleware.cs b/TaskFlow/Middleware/ITaskSchedulerMiddleware.cs new file mode 100644 index 0000000..2345ca7 --- /dev/null +++ b/TaskFlow/Middleware/ITaskSchedulerMiddleware.cs @@ -0,0 +1,15 @@ +namespace System.Threading.Tasks.Flow +{ + /// Identifies a component that participates in one or more task-scheduler middleware phases. + /// + /// This marker is the common registration type accepted by . + /// Implement at least one of , , or + /// . A component may implement multiple phase interfaces; one registration then captures one + /// annotation scope and provides one operation-local state slot shared by all implemented phases. + /// Middleware instances are reused across scheduled operations. Instance-level mutable state must therefore be safe for concurrent use. + /// Store state belonging to one operation in the context's local-state slot. + /// + public interface ITaskSchedulerMiddleware + { + } +} diff --git a/TaskFlow/Middleware/TaskSchedulerCompletionDelegate.cs b/TaskFlow/Middleware/TaskSchedulerCompletionDelegate.cs new file mode 100644 index 0000000..b696237 --- /dev/null +++ b/TaskFlow/Middleware/TaskSchedulerCompletionDelegate.cs @@ -0,0 +1,20 @@ +namespace System.Threading.Tasks.Flow +{ + using System.Diagnostics.CodeAnalysis; + using System.Threading.Tasks; + + /// Represents the remaining completion pipeline supplied to completion middleware. + /// The type of result carried by a successful outcome. + /// The current operation context to pass to the next completion phase. + /// The success or failure to expose to later completion middleware. + /// A value task containing the final outcome produced by the remaining completion middleware. + /// + /// Invoke this delegate once to preserve the normal completion chain. Passing the received outcome unchanged preserves captured exception identity + /// and stack information. Completion processing is claimed once per scheduled operation; calling this delegate does not enqueue or execute the user + /// operation again. + /// + [SuppressMessage("Naming", "CA1711:Identifiers should not have incorrect suffix", Justification = "The public API deliberately identifies continuation delegate types.")] + public delegate ValueTask> TaskSchedulerCompletionDelegate( + TaskSchedulerOperationContext context, + TaskSchedulerOperationOutcome outcome); +} diff --git a/TaskFlow/Middleware/TaskSchedulerEnqueueContext.cs b/TaskFlow/Middleware/TaskSchedulerEnqueueContext.cs new file mode 100644 index 0000000..4c5f267 --- /dev/null +++ b/TaskFlow/Middleware/TaskSchedulerEnqueueContext.cs @@ -0,0 +1,89 @@ +namespace System.Threading.Tasks.Flow +{ + using System; + using System.Threading; + + /// Provides immutable enqueue facts, cancellation, metadata, and local state to enqueue middleware. + /// The scheduled operation result type. + /// + /// Each context is associated with one scheduled operation and one middleware registration. Context instances are immutable; changing the + /// producer cancellation token creates a copy for the same operation. + /// The context distinguishes the caller's original token from the effective producer token flowing toward the terminal scheduler. It also + /// exposes only annotations that existed when the current middleware was registered. + /// + public sealed class TaskSchedulerEnqueueContext + { + private readonly Internal.PipelineOperation _operation; + private readonly Internal.AnnotationScope? _annotations; + private readonly int _registrationIndex; + + internal TaskSchedulerEnqueueContext( + Internal.PipelineOperation operation, + Internal.AnnotationScope? annotations, + int registrationIndex, + CancellationToken cancellationToken) + { + _operation = operation; + _annotations = annotations; + _registrationIndex = registrationIndex; + CancellationToken = cancellationToken; + } + + /// Gets the unmodified state supplied to . + /// The original state reference, which may be null. + /// Middleware transport does not wrap or replace this value, and the same reference is forwarded to the terminal scheduler. + public object? State => _operation.State; + + /// Gets the cancellation token supplied by the submitting caller. + /// The original caller token, even when enqueue middleware changes . + /// Use this token for caller-wait behavior that must remain independent of shared or transformed producer work. + public CancellationToken CallerCancellationToken => _operation.CallerCancellationToken; + + /// Gets the effective producer cancellation token for the next enqueue phase. + /// The token currently flowing toward the terminal scheduler. + /// The initial value is the caller token. Earlier enqueue middleware may replace it by passing a context returned by + /// to its continuation. + public CancellationToken CancellationToken { get; } + + /// Gets the annotation visible when this middleware registration was created. + /// The annotation type. + /// The nearest visible annotation registered under exactly , or null when no such annotation + /// was visible at registration time. + /// Annotations are forward-scoped. Values added after this middleware registration are not visible. A later value registered under the + /// same type shadows an earlier value only for later registrations and the final operation. + public TAnnotation? GetAnnotation() where TAnnotation : class, IOperationAnnotation + => _annotations?.Get(); + + /// Gets this middleware registration's local state for the current operation. + /// The state type. + /// The existing state when this registration initialized a slot with ; otherwise, null. + /// The slot was initialized with a different state type. + /// The slot is isolated from other operations and other registrations, including another registration of the same middleware object. + public TState? GetLocalState() where TState : class + => _operation.GetLocalState(_registrationIndex); + + /// Gets or atomically creates this middleware registration's local state for the current operation. + /// The reference-type state container. + /// Creates the state when it has not been initialized. + /// The existing state, or the non-null state created by . + /// is null. + /// The factory returns null, or the slot was initialized with a different state type. + /// The factory is used to initialize a per-operation, per-registration slot. Compound middleware registered through + /// can retrieve the same value from its execution and completion phases. + public TState GetOrCreateLocalState(Func stateFactory) where TState : class + { + Annotations.Argument.NotNull(stateFactory); + return _operation.GetOrCreateLocalState(_registrationIndex, stateFactory); + } + + /// Creates an enqueue context for the same operation with a different effective producer cancellation token. + /// The token to pass to the next enqueue phase and terminal scheduler. + /// A new context retaining the original state, caller token, captured annotations, operation identity, and registration-local state. + /// Pass the returned context to the enqueue continuation. This method does not cancel, link, or dispose either token and does not mutate + /// the current context. + public TaskSchedulerEnqueueContext WithCancellationToken(CancellationToken cancellationToken) + => new TaskSchedulerEnqueueContext(_operation, _annotations, _registrationIndex, cancellationToken); + + internal Internal.PipelineOperation Operation => _operation; + } +} diff --git a/TaskFlow/Middleware/TaskSchedulerEnqueueDelegate.cs b/TaskFlow/Middleware/TaskSchedulerEnqueueDelegate.cs new file mode 100644 index 0000000..45a8f24 --- /dev/null +++ b/TaskFlow/Middleware/TaskSchedulerEnqueueDelegate.cs @@ -0,0 +1,18 @@ +namespace System.Threading.Tasks.Flow +{ + using System.Diagnostics.CodeAnalysis; + using System.Threading.Tasks; + + /// Represents the remaining enqueue pipeline supplied to enqueue middleware. + /// The type of result produced by the scheduled operation. + /// The immutable enqueue context to pass onward. Its producer cancellation token is forwarded to later enqueue middleware + /// and eventually to the terminal scheduler. + /// A task representing the remaining admission steps, terminal scheduling, and operation completion. + /// + /// Invoke this delegate at most once for a given middleware invocation. Enqueue middleware may omit the call to short-circuit scheduling. The + /// delegate accepts only contexts belonging to the current operation; use + /// to change the effective producer token without losing operation identity or metadata. + /// + [SuppressMessage("Naming", "CA1711:Identifiers should not have incorrect suffix", Justification = "The public API deliberately identifies continuation delegate types.")] + public delegate Task TaskSchedulerEnqueueDelegate(TaskSchedulerEnqueueContext context); +} diff --git a/TaskFlow/Middleware/TaskSchedulerExecutionDelegate.cs b/TaskFlow/Middleware/TaskSchedulerExecutionDelegate.cs new file mode 100644 index 0000000..47a6f35 --- /dev/null +++ b/TaskFlow/Middleware/TaskSchedulerExecutionDelegate.cs @@ -0,0 +1,17 @@ +namespace System.Threading.Tasks.Flow +{ + using System.Diagnostics.CodeAnalysis; + using System.Threading.Tasks; + + /// Represents the remaining execution pipeline supplied to execution middleware. + /// The type of result produced by the scheduled operation. + /// The current operation context to pass to the next execution phase. + /// A value task containing the result produced by the remaining execution middleware or user operation. + /// + /// The delegate runs inside the terminal scheduler's submitted envelope. It may be invoked more than once by orchestration middleware without + /// submitting additional terminal delegates. Each invocation may execute the user operation again. Exceptions are captured as operation outcomes + /// and processed by completion middleware. + /// + [SuppressMessage("Naming", "CA1711:Identifiers should not have incorrect suffix", Justification = "The public API deliberately identifies continuation delegate types.")] + public delegate ValueTask TaskSchedulerExecutionDelegate(TaskSchedulerOperationContext context); +} diff --git a/TaskFlow/Middleware/TaskSchedulerMiddlewareExtensions.cs b/TaskFlow/Middleware/TaskSchedulerMiddlewareExtensions.cs new file mode 100644 index 0000000..7189aa3 --- /dev/null +++ b/TaskFlow/Middleware/TaskSchedulerMiddlewareExtensions.cs @@ -0,0 +1,133 @@ +namespace System.Threading.Tasks.Flow +{ + using System; + using System.Threading; + using System.Threading.Tasks; + using System.Threading.Tasks.Flow.Annotations; + using System.Threading.Tasks.Flow.Internal; + + /// Provides immutable middleware and annotation composition for . + /// + /// Each method returns a new non-owning scheduler snapshot. The source scheduler remains unchanged and may be reused to create independent + /// branches. + /// Pipeline snapshots do not dispose or otherwise assume ownership of terminal schedulers, middleware instances, or their collaborators. + /// + public static class TaskSchedulerMiddlewareExtensions + { + /// Creates a scheduler snapshot containing every phase implemented by one middleware object. + /// The scheduler or pipeline snapshot to extend. + /// The middleware object implementing one or more phase interfaces. + /// A new immutable, non-owning scheduler snapshot. The supplied scheduler is not modified. + /// or is null. + /// implements no phase interface. + /// + /// A single-phase object registers that phase. When one object implements multiple phase interfaces, all phases are registered atomically + /// at one pipeline position, capture the same forward annotation scope, and share one local-state slot per operation. + /// Enqueue middleware runs newest registration first. Execution and completion middleware run in registration order. Registration does not + /// invoke or take ownership of the middleware. + /// If is not already a TaskFlow pipeline snapshot, it is treated as an opaque terminal boundary; wrappers + /// are not inspected or flattened. + /// + /// + /// + /// ITaskScheduler scheduler = terminal + /// .UseMiddleware(new AdmissionMiddleware()) + /// .UseMiddleware(new TelemetryMiddleware()); + /// + /// + public static ITaskScheduler UseMiddleware(this ITaskScheduler taskScheduler, ITaskSchedulerMiddleware middleware) + { + Argument.NotNull(taskScheduler); + Argument.NotNull(middleware); + if (!(middleware is ITaskSchedulerEnqueueMiddleware) && + !(middleware is ITaskSchedulerExecutionMiddleware) && + !(middleware is ITaskSchedulerCompletionMiddleware)) + { + throw new ArgumentException("Middleware must implement at least one task scheduler middleware phase.", nameof(middleware)); + } + + return GetPipeline(taskScheduler).Append(middleware); + } + + /// Creates a scheduler snapshot with an additional forward-scoped annotation. + /// The annotation type used as the lookup key. + /// The scheduler or pipeline snapshot to extend. + /// The annotation visible to later registrations and the final operation. + /// A new immutable scheduler snapshot with the extended annotation scope. The supplied scheduler is not modified. + /// or is null. + /// + /// The annotation is visible to middleware registered after this call and to the final annotation-aware operation. Middleware already + /// registered on retains its earlier scope. + /// Lookup uses as the exact key. A later annotation registered with the same key shadows this value for + /// later registrations without mutating earlier snapshots or sibling branches. + /// The returned pipeline does not dispose or otherwise own . + /// + /// + /// + /// ITaskScheduler scheduler = terminal + /// .WithAnnotation<TenantAnnotation>(new TenantAnnotation("north")) + /// .UseMiddleware(new TenantTelemetryMiddleware()); + /// + /// + public static ITaskScheduler WithAnnotation(this ITaskScheduler taskScheduler, TAnnotation annotation) + where TAnnotation : class, IOperationAnnotation + { + Argument.NotNull(taskScheduler); + Argument.NotNull(annotation); + return GetPipeline(taskScheduler).WithAnnotation(typeof(TAnnotation), annotation); + } + + /// Enqueues an operation that receives the final pipeline operation context. + /// The operation result type. + /// The scheduler used to enqueue the operation. + /// The context-aware operation delegate. + /// Optional caller state exposed through the operation context. + /// The submitting caller's cancellation token. + /// A task representing the scheduled operation. + /// or is null. + /// The terminal scheduler has been disposed. + /// On a non-pipeline scheduler the context contains the supplied state and token but no annotations. + internal static Task EnqueueWithContext( + this ITaskScheduler taskScheduler, + TaskSchedulerExecutionDelegate taskFunc, + object? state, + CancellationToken cancellationToken) + { + Argument.NotNull(taskScheduler); + Argument.NotNull(taskFunc); + + if (taskScheduler is PipelineTaskScheduler pipeline) + { + var scheduler = pipeline.UseMiddleware(new ContextOperationMiddleware(taskFunc)); + return scheduler.Enqueue( + (Func>)((_, __) => throw new InvalidOperationException("The context operation middleware did not execute.")), + state, + cancellationToken); + } + + return taskScheduler.Enqueue(async (s, token) => + { + var operation = new PipelineOperation((_, __) => throw new InvalidOperationException(), s, 0, null, cancellationToken); + var context = new TaskSchedulerOperationContext(operation, null, -1, token); + return await taskFunc(context).ConfigureAwait(true); + }, state, cancellationToken); + } + + private static PipelineTaskScheduler GetPipeline(ITaskScheduler taskScheduler) + => taskScheduler as PipelineTaskScheduler ?? new PipelineTaskScheduler(taskScheduler); + + private sealed class ContextOperationMiddleware : ITaskSchedulerExecutionMiddleware + { + private readonly TaskSchedulerExecutionDelegate _taskFunc; + public ContextOperationMiddleware(TaskSchedulerExecutionDelegate taskFunc) => _taskFunc = taskFunc; + + public async ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate next) + { + if (typeof(T) != typeof(TResult)) return await next(context).ConfigureAwait(true); + var result = await _taskFunc(context).ConfigureAwait(true); + return (T)(object?)result!; + } + } + + } +} diff --git a/TaskFlow/Middleware/TaskSchedulerOperationContext.cs b/TaskFlow/Middleware/TaskSchedulerOperationContext.cs new file mode 100644 index 0000000..acc8de9 --- /dev/null +++ b/TaskFlow/Middleware/TaskSchedulerOperationContext.cs @@ -0,0 +1,83 @@ +namespace System.Threading.Tasks.Flow +{ + using System; + using System.Threading; + + /// Provides immutable operation facts and registration-scoped metadata to execution and completion middleware. + /// + /// The context belongs to one scheduled operation. For middleware, it is also associated with one registration and therefore one captured + /// annotation scope and one local-state slot. + /// The final context used by annotation-aware operation invocation is not associated with a middleware registration and cannot create local + /// state. + /// + public sealed class TaskSchedulerOperationContext + { + private readonly Internal.PipelineOperation _operation; + private readonly Internal.AnnotationScope? _annotations; + private readonly int _registrationIndex; + private readonly ITaskScheduler? _scheduler; + + internal TaskSchedulerOperationContext( + Internal.PipelineOperation operation, + Internal.AnnotationScope? annotations, + int registrationIndex, + CancellationToken cancellationToken, + ITaskScheduler? scheduler = null) + { + _operation = operation; + _annotations = annotations; + _registrationIndex = registrationIndex; + _scheduler = scheduler; + CancellationToken = cancellationToken; + } + + /// Gets the unmodified state supplied to . + /// The original state reference, which may be null. + public object? State => _operation.State; + + /// Gets the effective producer cancellation token captured before terminal scheduling. + /// The token selected by the enqueue pipeline and supplied to the terminal scheduler. + /// This value can differ from when enqueue middleware separates caller waiting from producer work. + public CancellationToken CancellationToken { get; } + + /// Gets the submitting caller's original cancellation token. + /// The token supplied when the operation was enqueued, independently of producer-token transformations. + public CancellationToken CallerCancellationToken => _operation.CallerCancellationToken; + + /// Gets the annotation visible when this middleware registration was created. + /// The annotation type. + /// The nearest visible annotation registered under exactly , or null. + /// A middleware context sees the annotation scope captured when that middleware was registered. The final operation context sees the + /// pipeline's final scope. Lookup is by the exact registered type rather than assignability. + public TAnnotation? GetAnnotation() where TAnnotation : class, IOperationAnnotation + => _annotations?.Get(); + + /// Gets this middleware registration's local state for the current operation. + /// The state type. + /// The existing state when initialized with ; otherwise, null. + /// The slot was initialized with a different state type. + /// Returns null for a final operation context because it has no middleware registration slot. + public TState? GetLocalState() where TState : class + => _registrationIndex < 0 ? null : _operation.GetLocalState(_registrationIndex); + + /// Gets or atomically creates this middleware registration's local state for the current operation. + /// The reference-type state container. + /// Creates the state when this registration has not initialized it. + /// The existing state, or the non-null state created by . + /// is null. + /// The context has no middleware registration, the factory returns null, or the slot was + /// initialized with a different state type. + /// All phases belonging to one compound registration share this slot. + /// Slots remain isolated across operations and registrations. + public TState GetOrCreateLocalState(Func stateFactory) where TState : class + { + Annotations.Argument.NotNull(stateFactory); + if (_registrationIndex < 0) throw new InvalidOperationException("The final operation context has no middleware registration-local state."); + return _operation.GetOrCreateLocalState(_registrationIndex, stateFactory); + } + + internal Internal.PipelineOperation Operation => _operation; + internal ITaskScheduler Scheduler => _scheduler ?? throw new InvalidOperationException("The context is not associated with a middleware registration."); + internal IOperationAnnotation? GetAnnotation(Type type) => _annotations?.Get(type); + } +} diff --git a/TaskFlow/Middleware/TaskSchedulerOperationOutcome.cs b/TaskFlow/Middleware/TaskSchedulerOperationOutcome.cs new file mode 100644 index 0000000..09a6797 --- /dev/null +++ b/TaskFlow/Middleware/TaskSchedulerOperationOutcome.cs @@ -0,0 +1,72 @@ +namespace System.Threading.Tasks.Flow +{ + using System; + using System.Diagnostics.CodeAnalysis; + using System.Runtime.ExceptionServices; + + /// Represents the current successful result or captured failure of a scheduled operation. + /// The successful result type. + /// + /// Completion middleware receives and returns this value to observe or transform completion without throwing merely to represent failure. + /// A failure captures when it is created. Passing that outcome onward unchanged preserves exception identity + /// and its captured stack when the pipeline ultimately rethrows it. + /// The default value represents a successful outcome whose is the default value of . Prefer + /// the named factories when intentionally creating an outcome. + /// + [SuppressMessage("Design", "CA1000:Do not declare static members on generic types", Justification = "Named factories avoid ambiguous constructors when TResult is Exception.")] + [SuppressMessage("Performance", "CA1815:Override equals and operator equals on value types", Justification = "Operation outcomes have no value equality semantics.")] + public readonly struct TaskSchedulerOperationOutcome + { + private readonly TResult _result; + private readonly ExceptionDispatchInfo? _exception; + + private TaskSchedulerOperationOutcome(TResult result) + { + _result = result; + _exception = null; + } + + private TaskSchedulerOperationOutcome(Exception exception) + { + _result = default!; + Annotations.Argument.NotNull(exception); + _exception = ExceptionDispatchInfo.Capture(exception); + } + + /// Gets a value indicating whether the outcome contains a successful result. + /// true when no exception is captured; otherwise, false. + public bool IsSuccess => _exception == null; + + /// Gets the successful operation result. + /// The result supplied to . + /// The outcome represents a failure. + public TResult Result => IsSuccess + ? _result + : throw new InvalidOperationException("A failed operation outcome does not have a result."); + + /// Gets the original exception represented by a failed outcome. + /// The captured exception instance, or null when is true. + /// Reading this property does not throw the exception or alter its captured dispatch information. + public Exception? Exception => _exception?.SourceException; + + /// Creates an outcome representing successful operation completion. + /// The operation result. + /// A successful outcome containing . The result may be null when permitted by + /// . + public static TaskSchedulerOperationOutcome FromResult(TResult result) => new TaskSchedulerOperationOutcome(result); + + /// Creates a failed outcome and captures the exception's dispatch information. + /// The operation or middleware exception. + /// A failed outcome containing the same exception instance. + /// is null. + /// Create a new failure only when intentionally introducing or replacing a failure. Forward an existing outcome unchanged to preserve + /// the dispatch information it already contains. + public static TaskSchedulerOperationOutcome FromException(Exception exception) => new TaskSchedulerOperationOutcome(exception); + + internal TResult GetResultOrThrow() + { + _exception?.Throw(); + return _result; + } + } +} diff --git a/docs/customization.md b/docs/customization.md index aba22dd..4751296 100644 --- a/docs/customization.md +++ b/docs/customization.md @@ -62,11 +62,221 @@ Derive from `TaskFlowBase` only when the implementation needs an owned lifecycle Use `ThisLock` for state coordinated with the base disposal state. Do not claim built-in FIFO or canceled-delegate behavior unless the custom implementation actually preserves it and tests it. -## Interceptors and decorators +## Middleware and terminal schedulers -Prefer an `ITaskScheduler` decorator when adding a cross-cutting policy without owning the lane. Forward the original state and cancellation token unless the policy intentionally transforms them. Preserve the returned task's result and failure unless replacement behavior is part of the documented contract. +Implement `ITaskScheduler` when you need a new terminal scheduling strategy. Implement middleware when the terminal scheduling strategy is already correct and you only need to add admission, execution, or outcome policy. Middleware does not require changes to `ITaskScheduler.Enqueue`, so it can be placed around built-in flows and third-party schedulers. -Use `ITaskSchedulerInterceptor` or `IAsyncTaskSchedulerInterceptor` for operation lifecycle callbacks. See [Observability extensions](extensions/observability.md) for callback ordering and exception replacement rules. +The pipeline has three phases: + +| Phase | Runs | Typical uses | Continuation behavior | +| --- | --- | --- | --- | +| Enqueue | Before the terminal accepts the operation | admission, coalescing, throttling, cancellation-token transformation | Call zero or one time | +| Execution | Inside the delegate invoked by the terminal | instrumentation, retries, operation wrapping | Call one or more times | +| Completion | After a result or exception is available | error handling, result transformation, final telemetry | Normally call exactly once | + +Implement one or more of these interfaces: + +- `ITaskSchedulerEnqueueMiddleware` +- `ITaskSchedulerExecutionMiddleware` +- `ITaskSchedulerCompletionMiddleware` + +Every middleware also implements the marker interface `ITaskSchedulerMiddleware`. Register it with `UseMiddleware`. A single-phase object registers that phase; a compound object atomically registers every phase it implements. + +### Register middleware + +Registration returns a new immutable scheduler snapshot. It does not modify the source scheduler: + +```csharp +ITaskScheduler terminal = new TaskFlow(); + +ITaskScheduler observed = terminal + .UseMiddleware(new TraceExecutionMiddleware()) + .UseMiddleware(new TraceCompletionMiddleware()); + +// `terminal` has no middleware. `observed` has both registrations. +``` + +Snapshots may be reused concurrently and branched safely: + +```csharp +ITaskScheduler common = terminal.UseMiddleware(new MetricsMiddleware()); +ITaskScheduler interactive = common.UseMiddleware(new InteractiveAdmissionMiddleware()); +ITaskScheduler background = common.UseMiddleware(new BackgroundAdmissionMiddleware()); +``` + +The snapshot is non-owning. Disposing a terminal that implements `IDisposable` remains the caller's responsibility. A pipeline never disposes the terminal, middleware objects, or collaborators held by middleware. + +### Phase ordering + +For registrations `A`, then `B`, the phases run in this order: + +```text +enqueue: B -> A -> terminal +execution: terminal -> A -> B -> operation +completion: terminal outcome -> A -> B -> caller +``` + +Enqueue uses reverse registration order because the newest admission policy is the outermost policy. Execution and completion use registration order. A compound middleware occupies one registration position in every phase it implements. + +An existing scheduler wrapper that does not expose the middleware pipeline is an opaque boundary. Adding middleware outside that wrapper creates a new pipeline; TaskFlow does not inspect, flatten, or reorder the wrapper and any pipeline hidden inside it. + +### Enqueue middleware + +Enqueue middleware receives immutable operation facts and decides whether or how to continue toward the terminal: + +```csharp +public sealed class RejectWhenStopping : ITaskSchedulerEnqueueMiddleware +{ + private readonly Func _isStopping; + + public RejectWhenStopping(Func isStopping) => _isStopping = isStopping; + + public Task InvokeAsync( + TaskSchedulerEnqueueContext context, + TaskSchedulerEnqueueDelegate continuation) + { + if (_isStopping()) + return Task.FromException(new InvalidOperationException("The service is stopping.")); + + return continuation(context); + } +} +``` + +Returning without invoking `continuation` prevents terminal scheduling. This supports rejection and shared-work policies. Invoke the enqueue continuation at most once: the terminal contract represents one submitted operation. + +`context.CallerCancellationToken` is always the token supplied by the caller. `context.CancellationToken` is the effective producer token currently flowing toward the terminal. To transform only the producer token, pass a copied context onward: + +```csharp +return continuation(context.WithCancellationToken(producerToken)); +``` + +`WithCancellationToken` retains the original state, caller token, annotations, operation identity, and registration-local state. Middleware that separates a caller's wait from shared producer work should observe `CallerCancellationToken` for the wait and pass the shared producer token through `WithCancellationToken`. + +### Execution middleware + +Execution middleware runs on the context chosen by the terminal scheduler. It can surround the operation in the same way application middleware surrounds a request: + +```csharp +public sealed class TraceExecutionMiddleware : ITaskSchedulerExecutionMiddleware +{ + public async ValueTask InvokeAsync( + TaskSchedulerOperationContext context, + TaskSchedulerExecutionDelegate continuation) + { + Console.WriteLine("Starting"); + try + { + return await continuation(context).ConfigureAwait(true); + } + finally + { + Console.WriteLine("Finished execution"); + } + } +} +``` + +Ordinary middleware should invoke `continuation` once. Orchestration middleware may invoke it repeatedly—for example, to implement retry—without enqueueing another terminal delegate. Such middleware owns the semantics of repeated operation execution, including which failures are retryable and how cancellation is handled. + +Use `ConfigureAwait(true)` when asynchronous middleware must preserve the synchronization context or scheduler affinity established by the terminal. + +### Completion middleware and outcomes + +Completion middleware receives a `TaskSchedulerOperationOutcome`. Inspect `IsSuccess` before reading `Result`; `Exception` is non-null for a failed outcome. + +```csharp +public sealed class TraceCompletionMiddleware : ITaskSchedulerCompletionMiddleware +{ + public ValueTask> InvokeAsync( + TaskSchedulerOperationContext context, + TaskSchedulerOperationOutcome outcome, + TaskSchedulerCompletionDelegate continuation) + { + if (!outcome.IsSuccess) + Console.Error.WriteLine(outcome.Exception); + + return continuation(context, outcome); + } +} +``` + +Pass the same outcome onward to preserve an exception's identity and captured stack. Use `TaskSchedulerOperationOutcome.FromResult` or `FromException` only when intentionally replacing the current result or failure. If completion middleware throws, that exception replaces the current outcome and later completion middleware sees the replacement. + +Completion is claimed atomically and runs at most once for an operation, including races between rejection, timeout, and the scheduled delegate. Failures produced during terminal execution are completed on the terminal context. Failures raised before the terminal accepts or invokes the operation cannot claim terminal thread or synchronization-context affinity. + +### Share per-operation state between phases + +One middleware object may implement multiple phase interfaces. Register it once with `UseMiddleware` to give all of its phases one operation-local state slot: + +```csharp +public sealed class TimingMiddleware : + ITaskSchedulerExecutionMiddleware, + ITaskSchedulerCompletionMiddleware +{ + private sealed class TimingState + { + public Stopwatch Stopwatch { get; } = new Stopwatch(); + } + + public async ValueTask InvokeAsync( + TaskSchedulerOperationContext context, + TaskSchedulerExecutionDelegate continuation) + { + context.GetOrCreateLocalState(() => new TimingState()).Stopwatch.Start(); + return await continuation(context).ConfigureAwait(true); + } + + public ValueTask> InvokeAsync( + TaskSchedulerOperationContext context, + TaskSchedulerOperationOutcome outcome, + TaskSchedulerCompletionDelegate continuation) + { + TimingState? state = context.GetLocalState(); + state?.Stopwatch.Stop(); + Console.WriteLine(state?.Stopwatch.Elapsed); + return continuation(context, outcome); + } +} +``` + +The slot is isolated per scheduled operation and per registration. `GetOrCreateLocalState` creates its value atomically. A different state type cannot later occupy the same registration slot. Registering the same middleware object twice creates two independent slots; registering its phases separately also creates separate slots. + +Middleware objects themselves are shared by all operations and must therefore keep any mutable registration-wide state thread-safe. Put operation-specific mutable state in the context slot rather than in middleware instance fields. + +### Forward-scoped annotations + +Annotations are immutable metadata scopes keyed by their registered type: + +```csharp +public sealed class TenantAnnotation : IOperationAnnotation +{ + public TenantAnnotation(string tenantId) => TenantId = tenantId; + public string TenantId { get; } +} + +ITaskScheduler tenantPipeline = terminal + .WithAnnotation(new TenantAnnotation("north")) + .UseMiddleware(new TenantTelemetryMiddleware()); +``` + +A middleware registration captures only annotations added before that registration. Later annotations are visible only to later registrations and to the final context-aware operation. Adding another annotation with the same registered type shadows the earlier value without changing existing registrations or sibling pipeline branches. + +Read captured metadata with `context.GetAnnotation()`. The lookup uses the exact type supplied to `WithAnnotation`; registering an implementation as an interface and querying by its concrete type are different keys. + +Use `AnnotatedEnqueue` when the final operation needs an annotation from the final scope: + +```csharp +string tenant = await tenantPipeline.AnnotatedEnqueue( + (state, annotation, token) => new ValueTask( + annotation?.TenantId ?? "unknown"), + state: null, + cancellationToken: CancellationToken.None); +``` + +### Choosing middleware or interceptors + +Use middleware for reusable policy that needs to affect admission, wrap execution, transform outcomes, carry metadata, or coordinate state across phases. Use `ITaskSchedulerInterceptor` or `IAsyncTaskSchedulerInterceptor` for the established operation lifecycle callback model. The built-in interception, error, timeout, throttling, cancellation, and logging extensions are composed through the middleware pipeline but retain their existing public APIs. See [Observability extensions](extensions/observability.md) for callback ordering and exception replacement rules. ## Testing custom implementations diff --git a/docs/extensions/cancellation.md b/docs/extensions/cancellation.md index 24afa23..c2fbcfd 100644 --- a/docs/extensions/cancellation.md +++ b/docs/extensions/cancellation.md @@ -6,6 +6,8 @@ permalink: /extensions/cancellation/ # Cancellation extensions +Cancellation scope and cancel-previous are enqueue middleware. They link or replace the producer token before the terminal scheduler accepts the operation while retaining the submitting caller's token separately. Pipeline wrappers remain non-owning and dispose only the linked token sources created for an invocation. + Cancellation policies compose additional cancellation sources with the caller token and the underlying flow's disposal token. They request cancellation; they cannot force a delegate to stop. ## Component cancellation scopes diff --git a/docs/extensions/observability.md b/docs/extensions/observability.md index 967d5fa..d13b56c 100644 --- a/docs/extensions/observability.md +++ b/docs/extensions/observability.md @@ -8,19 +8,19 @@ permalink: /extensions/observability/ ## Operation names -`WithOperationName` attaches an `OperationNameAnnotation` to submissions. Place it outside consumers such as logging or timeout so they can read the annotation. +`WithOperationName` adds forward-scoped metadata. Place it before consumers such as logging or timeout so those registrations capture the name. ```csharp ITaskScheduler named = flow - .WithLogging(logger) - .WithOperationName("orders.persist"); + .WithOperationName("orders.persist") + .WithLogging(logger); await named.Enqueue(token => PersistAsync(token)); static Task PersistAsync(CancellationToken token) => Task.CompletedTask; ``` -Decorator order is observable: `flow.WithOperationName(...).WithLogging(logger)` places logging outside the annotation and therefore does not provide that name to the logging wrapper. +Later annotations never change middleware that is already registered. Replacing an annotation type affects subsequent registrations and the final operation only, so immutable branches can carry independent metadata. ## Microsoft logging @@ -32,6 +32,7 @@ dotnet add package TaskFlow.Microsoft.Extensions.Logging ```csharp ITaskScheduler logged = flow + .WithOperationName("imports.run") .WithLogging(logger, options => { options.EnqueuedLogLevel = LogLevel.Debug; @@ -39,8 +40,7 @@ ITaskScheduler logged = flow options.SucceededLogLevel = LogLevel.Information; options.FailedLogLevel = LogLevel.Error; options.FinishedLogLevel = LogLevel.Debug; - }) - .WithOperationName("imports.run"); + }); await logged.Enqueue(token => ImportAsync(token)); @@ -67,3 +67,7 @@ Callbacks run inside the selected scheduler context. A callback failure faults t ## Ownership Annotations, interception, and logging are scheduler decorators. They neither own nor dispose the underlying flow. Keep the original `ITaskFlow` and dispose it at the component boundary. + +## Middleware ordering + +TaskFlow registrations have distinct phases. Enqueue middleware handles admission, timeout, and cancellation before the terminal scheduler. Execution middleware surrounds the operation on the selected scheduler context. Completion middleware processes the resulting success or failure in registration order. An execution failure therefore reaches interception error/finally callbacks before `OnError` completion callbacks. diff --git a/docs/extensions/reliability.md b/docs/extensions/reliability.md index edf75ea..e88af06 100644 --- a/docs/extensions/reliability.md +++ b/docs/extensions/reliability.md @@ -6,6 +6,8 @@ permalink: /extensions/reliability/ # Reliability extensions +Timeout and throttle run in the enqueue phase, so timeout still includes queue waiting and throttle still rejects before a terminal queue turn is consumed. Failures produced before scheduled execution pass through completion middleware once, but no terminal thread or synchronization-context affinity can be promised for those admission failures. + ## Timeout `WithTimeout` applies one budget to queue waiting and delegate execution. It throws `TimeoutException` when the timer wins and requests cancellation of the underlying operation. diff --git a/docs/semantics-and-pitfalls.md b/docs/semantics-and-pitfalls.md index 4b7adcb..2a998d6 100644 --- a/docs/semantics-and-pitfalls.md +++ b/docs/semantics-and-pitfalls.md @@ -58,18 +58,20 @@ Disposal waits for lane completion and suppresses operation failures internally. Disposal does not propagate an ignored operation's exception. Await or return the task when its outcome belongs to a caller. For intentionally discarded work, handle failures inside the delegate or use a decorator such as `OnError` to report them. `OnError` observes and rethrows, so its diagnostic side effect still runs even when the returned task is deliberately ignored. -## Decorator order changes what a policy sees +## Middleware order and forward metadata -Decorators wrap from left to right. In this chain, `WithOperationName` is outermost, so its annotation reaches the logging decorator: +Metadata configures registrations to its right. In this chain, both logging and timeout capture the operation name: ```csharp ITaskScheduler operations = flow + .WithOperationName("orders.persist") .WithLogging(logger) - .WithTimeout(TimeSpan.FromSeconds(10)) - .WithOperationName("orders.persist"); + .WithTimeout(TimeSpan.FromSeconds(10)); ``` -Moving `WithOperationName` inside `WithLogging` prevents that logging wrapper from seeing the annotation. Cancellation and error wrappers can likewise observe different failures depending on their order. +Moving `WithOperationName` after `WithLogging` does not retroactively rename that logging registration. Enqueue middleware is entered newest-first, execution middleware is entered in registration order, and completion middleware processes outcomes in registration order. + +Opaque third-party scheduler wrappers remain supported, but each wrapper is a pipeline boundary. Use the public middleware interfaces when cross-decorator phase ordering is required. ## Value-returning async lambdas can be ambiguous From 46dfb18d786a7986618992a8fe15151dec4b168c Mon Sep 17 00:00:00 2001 From: Volodymyr Dombrovskyi <5788605+dombrovsky@users.noreply.github.com> Date: Sun, 16 Aug 2026 19:13:37 -0600 Subject: [PATCH 2/2] Validate future middleware policy seams --- .../FuturePolicyMiddlewareSeamsFixture.cs | 524 ++++++++++++++++++ docs/customization.md | 14 + 2 files changed, 538 insertions(+) create mode 100644 TaskFlow.Tests/Extensions/FuturePolicyMiddlewareSeamsFixture.cs diff --git a/TaskFlow.Tests/Extensions/FuturePolicyMiddlewareSeamsFixture.cs b/TaskFlow.Tests/Extensions/FuturePolicyMiddlewareSeamsFixture.cs new file mode 100644 index 0000000..8d7209a --- /dev/null +++ b/TaskFlow.Tests/Extensions/FuturePolicyMiddlewareSeamsFixture.cs @@ -0,0 +1,524 @@ +namespace TaskFlow.Tests.Extensions +{ + using NUnit.Framework; + using System.Collections.Concurrent; + using System.Diagnostics.CodeAnalysis; + using System.Threading.Tasks.Flow; + + [TestFixture] + internal sealed class FuturePolicyMiddlewareSeamsFixture + { + [SuppressMessage("Maintainability", "CA1515:Consider making public types internal", Justification = "NUnit requires the parameter type of this public parameterized test to be publicly accessible.")] + public enum OwnershipCancellation + { + LocalScope, + SharedLane, + } + + [Test] + public async Task Retry_UsesOneQueueTurnAndInvokesDownstreamOncePerAttempt() + { + var terminal = new CountingInlineTerminal(); + var downstream = new ExecutionCounterMiddleware(); + var completion = new CompletionCounterMiddleware(); + var attempts = 0; + var scheduler = terminal + .UseMiddleware(new RetryMiddleware(3)) + .UseMiddleware(downstream) + .UseMiddleware(completion); + + var result = await scheduler.Enqueue(() => + { + attempts++; + if (attempts < 3) throw new InvalidOperationException("retryable"); + return 42; + }); + + Assert.That(result, Is.EqualTo(42)); + Assert.That(terminal.EnqueueCount, Is.EqualTo(1)); + Assert.That(attempts, Is.EqualTo(3)); + Assert.That(downstream.CallCount, Is.EqualTo(3)); + Assert.That(completion.CallCount, Is.EqualTo(1)); + } + + [Test] + public async Task SingleFlight_SharesOneProducerAndAllowsIndependentWaitCancellation() + { + var terminal = new CountingInlineTerminal(); + var middleware = new SingleFlightIntMiddleware(); + var execution = new ExecutionCounterMiddleware(); + var completion = new CompletionCounterMiddleware(); + var scheduler = terminal + .UseMiddleware(middleware) + .UseMiddleware(execution) + .UseMiddleware(completion); + var producerGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var firstWait = new CancellationTokenSource(); + + var first = scheduler.Enqueue((_, _) => new ValueTask(producerGate.Task), "key", firstWait.Token); + await middleware.ProducerStarted; + var follower = scheduler.Enqueue((_, _) => throw new AssertionException("A follower must not create another producer."), "key", CancellationToken.None); + + await firstWait.CancelAsync(); + Assert.That(async () => await first, Throws.InstanceOf()); + Assert.That(follower.IsCompleted, Is.False); + + producerGate.SetResult(42); + Assert.That(await follower, Is.EqualTo(42)); + Assert.That(terminal.EnqueueCount, Is.EqualTo(1)); + Assert.That(execution.CallCount, Is.EqualTo(1)); + Assert.That(completion.CallCount, Is.EqualTo(1)); + Assert.That(middleware.ActiveCount, Is.Zero); + } + + [Test] + public async Task DuplicateSuppression_RejectsConcurrentDuplicateAndReleasesKeyAfterCompletion() + { + var terminal = new CountingInlineTerminal(); + var middleware = new DuplicateSuppressionMiddleware(); + var execution = new ExecutionCounterMiddleware(); + var scheduler = terminal + .UseMiddleware(middleware) + .UseMiddleware(execution); + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + var accepted = scheduler.Enqueue((_, _) => new ValueTask(gate.Task), "key", CancellationToken.None); + await middleware.Accepted; + var duplicate = scheduler.Enqueue((_, _) => new ValueTask(99), "key", CancellationToken.None); + + Assert.That(async () => await duplicate, Throws.TypeOf()); + Assert.That(terminal.EnqueueCount, Is.EqualTo(1)); + Assert.That(execution.CallCount, Is.EqualTo(1)); + + gate.SetResult(42); + Assert.That(await accepted, Is.EqualTo(42)); + Assert.That(middleware.ActiveCount, Is.Zero); + + Assert.That(await scheduler.Enqueue((_, _) => new ValueTask(7), "key", CancellationToken.None), Is.EqualTo(7)); + Assert.That(terminal.EnqueueCount, Is.EqualTo(2)); + Assert.That(execution.CallCount, Is.EqualTo(2)); + } + + [Test] + public async Task KeyedCoordinatorAdapter_EvictsInactiveKeysAndDoesNotTransferOwnership() + { + var coordinator = new TestKeyedCoordinator(); + var terminal = new CountingInlineTerminal(); + var scheduler = terminal.UseMiddleware(new KeyedCoordinatorMiddleware(coordinator)); + var firstGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var entered = 0; + + var first = scheduler.Enqueue(async (_, _) => + { + Interlocked.Increment(ref entered); + await firstGate.Task; + return 1; + }, "same", CancellationToken.None); + await coordinator.FirstEntered; + var second = scheduler.Enqueue((_, _) => + { + Interlocked.Increment(ref entered); + return new ValueTask(2); + }, "same", CancellationToken.None); + var differentKey = scheduler.Enqueue((_, _) => new ValueTask(3), "other", CancellationToken.None); + + await Task.Yield(); + Assert.That(Volatile.Read(ref entered), Is.EqualTo(1)); + Assert.That(await differentKey, Is.EqualTo(3), "A different key should not wait for the occupied lane."); + firstGate.SetResult(null); + var results = await Task.WhenAll(first, second); + Assert.That(results[0], Is.EqualTo(1)); + Assert.That(results[1], Is.EqualTo(2)); + Assert.That(coordinator.MaximumSameKeyConcurrency, Is.EqualTo(1)); + Assert.That(coordinator.ActiveKeyCount, Is.Zero); + Assert.That(coordinator.IsDisposed, Is.False); + Assert.That(terminal.EnqueueCount, Is.EqualTo(3)); + + coordinator.Dispose(); + Assert.That(coordinator.IsDisposed, Is.True); + } + + [Test] + public async Task LocalScope_TracksSubmissionBeforeMiddlewareAdmissionOnSharedLane() + { + var lane = new CountingInlineTerminal(); + var admission = new BlockingAdmissionMiddleware(); + var decoratedLane = lane.UseMiddleware(admission); + var firstScope = new LocalOperationScope(decoratedLane); + var secondScope = new LocalOperationScope(decoratedLane); + + var operation = firstScope.Enqueue((_, _) => new ValueTask(42), null, CancellationToken.None); + await admission.Entered; + + Assert.That(firstScope.ActiveCount, Is.EqualTo(1)); + Assert.That(secondScope.ActiveCount, Is.Zero); + Assert.That(lane.EnqueueCount, Is.Zero); + + admission.Release(); + Assert.That(await operation, Is.EqualTo(42)); + Assert.That(firstScope.ActiveCount, Is.Zero); + Assert.That(lane.EnqueueCount, Is.EqualTo(1)); + } + + [TestCase(OwnershipCancellation.LocalScope)] + [TestCase(OwnershipCancellation.SharedLane)] + public async Task SharedProducer_DetachesCallerButPreservesOwnershipCancellation(OwnershipCancellation cancellation) + { + using var localOwner = new CancellationTokenSource(); + using var sharedLane = new CancellationTokenSource(); + using var caller = new CancellationTokenSource(); + var terminal = new CountingInlineTerminal(); + var ownership = new OwnershipSingleFlightMiddleware(localOwner.Token, sharedLane.Token); + var scheduler = terminal + .UseMiddleware(ownership) + .UseMiddleware(new DropProducerTokenMiddleware()); + var producerEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + var first = scheduler.Enqueue(async (_, producerToken) => + { + producerEntered.TrySetResult(null); + await Task.Delay(Timeout.InfiniteTimeSpan, producerToken); + return 42; + }, "key", caller.Token); + await producerEntered.Task; + var follower = scheduler.Enqueue((_, _) => throw new AssertionException("A follower must not invoke the producer."), "key", CancellationToken.None); + + await caller.CancelAsync(); + Assert.That(async () => await first, Throws.InstanceOf()); + Assert.That(follower.IsCompleted, Is.False, "Canceling one caller must not cancel shared producer work."); + + if (cancellation == OwnershipCancellation.LocalScope) + { + await localOwner.CancelAsync(); + } + else + { + await sharedLane.CancelAsync(); + } + + Assert.That(async () => await follower, Throws.InstanceOf()); + Assert.That(terminal.EnqueueCount, Is.EqualTo(1)); + Assert.That(() => ownership.ActiveCount, Is.Zero.After(100, 10)); + } + + private sealed class RetryMiddleware : ITaskSchedulerExecutionMiddleware + { + private readonly int _maximumAttempts; + public RetryMiddleware(int maximumAttempts) => _maximumAttempts = maximumAttempts; + + public async ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation) + { + for (var attempt = 1; ; attempt++) + { + try + { + return await continuation(context); + } + catch (InvalidOperationException) when (attempt < _maximumAttempts) + { + } + } + } + } + + private sealed class SingleFlightIntMiddleware : ITaskSchedulerEnqueueMiddleware + { + private readonly ConcurrentDictionary _producers = new ConcurrentDictionary(); + private readonly TaskCompletionSource _producerStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + public Task ProducerStarted => _producerStarted.Task; + public int ActiveCount => _producers.Count; + + public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) + { + if (typeof(TResult) != typeof(int)) throw new NotSupportedException("This test prototype supports Int32 results only."); + var key = (string)context.State!; + var candidate = new Producer(); + var producer = _producers.GetOrAdd(key, candidate); + if (ReferenceEquals(candidate, producer)) + { + _ = Produce(key, producer, context, continuation); + } + + return Wait(producer.Task, context.CallerCancellationToken); + } + + [SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "The single-flight prototype must transfer every producer failure to all follower tasks.")] + private async Task Produce(string key, Producer producer, TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) + { + _producerStarted.TrySetResult(null); + try + { + var result = await continuation(context.WithCancellationToken(CancellationToken.None)); + producer.TrySetResult((int)(object)result!); + } + catch (Exception exception) + { + producer.TrySetException(exception); + } + finally + { + _producers.TryRemove(new KeyValuePair(key, producer)); + } + } + + private static async Task Wait(Task producer, CancellationToken cancellationToken) + => (TResult)(object)await producer.WaitAsync(cancellationToken); + + private sealed class Producer + { + private readonly TaskCompletionSource _completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + public Task Task => _completion.Task; + public void TrySetResult(int result) => _completion.TrySetResult(result); + public void TrySetException(Exception exception) => _completion.TrySetException(exception); + } + } + + private sealed class DuplicateSuppressionMiddleware : ITaskSchedulerEnqueueMiddleware + { + private readonly ConcurrentDictionary _active = new ConcurrentDictionary(); + private readonly TaskCompletionSource _accepted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + public Task Accepted => _accepted.Task; + public int ActiveCount => _active.Count; + + public async Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) + { + var key = (string)context.State!; + if (!_active.TryAdd(key, null)) throw new InvalidOperationException("An equivalent operation is already active."); + _accepted.TrySetResult(null); + try + { + return await continuation(context); + } + finally + { + _active.TryRemove(key, out _); + } + } + } + + private sealed class KeyedCoordinatorMiddleware : ITaskSchedulerEnqueueMiddleware + { + private readonly TestKeyedCoordinator _coordinator; + public KeyedCoordinatorMiddleware(TestKeyedCoordinator coordinator) => _coordinator = coordinator; + public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) + => _coordinator.Run((string)context.State!, () => continuation(context)); + } + + private sealed class TestKeyedCoordinator : IDisposable + { + private readonly object _sync = new object(); + private readonly Dictionary _lanes = new Dictionary(); + private readonly TaskCompletionSource _firstEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + private int _maximumSameKeyConcurrency; + private bool _isDisposed; + + public Task FirstEntered => _firstEntered.Task; + public int ActiveKeyCount + { + get + { + lock (_sync) return _lanes.Count; + } + } + public int MaximumSameKeyConcurrency => _maximumSameKeyConcurrency; + public bool IsDisposed => _isDisposed; + + public async Task Run(string key, Func> operation) + { + Lane lane; + lock (_sync) + { + ObjectDisposedException.ThrowIf(_isDisposed, this); + if (!_lanes.TryGetValue(key, out lane!)) + { + lane = new Lane(); + _lanes.Add(key, lane); + } + + lane.Users++; + } + + await lane.Gate.WaitAsync(); + var concurrency = Interlocked.Increment(ref lane.Executing); + UpdateMaximum(concurrency); + _firstEntered.TrySetResult(null); + try + { + return await operation(); + } + finally + { + Interlocked.Decrement(ref lane.Executing); + lane.Gate.Release(); + lock (_sync) + { + lane.Users--; + if (lane.Users == 0 && _lanes.TryGetValue(key, out var current) && ReferenceEquals(current, lane)) + { + _lanes.Remove(key); + } + } + } + } + + public void Dispose() + { + lock (_sync) _isDisposed = true; + } + + private void UpdateMaximum(int value) + { + int current; + do + { + current = _maximumSameKeyConcurrency; + if (current >= value) return; + } + while (Interlocked.CompareExchange(ref _maximumSameKeyConcurrency, value, current) != current); + } + + private sealed class Lane + { + public SemaphoreSlim Gate { get; } = new SemaphoreSlim(1, 1); + public int Users; + public int Executing; + } + } + + private sealed class LocalOperationScope : ITaskScheduler + { + private readonly ITaskScheduler _sharedLane; + private int _activeCount; + public LocalOperationScope(ITaskScheduler sharedLane) => _sharedLane = sharedLane; + public int ActiveCount => Volatile.Read(ref _activeCount); + + public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + { + Interlocked.Increment(ref _activeCount); + try + { + return await _sharedLane.Enqueue(taskFunc, state, cancellationToken); + } + finally + { + Interlocked.Decrement(ref _activeCount); + } + } + } + + private sealed class DropProducerTokenMiddleware : ITaskSchedulerEnqueueMiddleware + { + public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) + => continuation(context.WithCancellationToken(CancellationToken.None)); + } + + private sealed class OwnershipSingleFlightMiddleware : ITaskSchedulerEnqueueMiddleware + { + private readonly CancellationToken _localOwnerToken; + private readonly CancellationToken _sharedLaneToken; + private readonly ConcurrentDictionary _producers = new ConcurrentDictionary(); + + public OwnershipSingleFlightMiddleware(CancellationToken localOwnerToken, CancellationToken sharedLaneToken) + { + _localOwnerToken = localOwnerToken; + _sharedLaneToken = sharedLaneToken; + } + + public int ActiveCount => _producers.Count; + + public Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) + { + if (typeof(TResult) != typeof(int)) throw new NotSupportedException("This test prototype supports Int32 results only."); + var key = (string)context.State!; + var candidate = new OwnedProducer(); + var producer = _producers.GetOrAdd(key, candidate); + if (ReferenceEquals(candidate, producer)) + { + _ = Produce(key, producer, context, continuation); + } + + return Wait(producer.Task, context.CallerCancellationToken, _localOwnerToken, _sharedLaneToken); + } + + [SuppressMessage("Design", "CA1031:Do not catch general exception types", Justification = "The ownership prototype must transfer every producer failure to all follower tasks.")] + private async Task Produce(string key, OwnedProducer producer, TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) + { + using var ownership = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken, _localOwnerToken, _sharedLaneToken); + try + { + var result = await continuation(context.WithCancellationToken(ownership.Token)); + producer.TrySetResult((int)(object)result!); + } + catch (Exception exception) + { + producer.TrySetException(exception); + } + finally + { + _producers.TryRemove(new KeyValuePair(key, producer)); + } + } + + private static async Task Wait(Task producer, CancellationToken callerToken, CancellationToken localOwnerToken, CancellationToken sharedLaneToken) + { + using var wait = CancellationTokenSource.CreateLinkedTokenSource(callerToken, localOwnerToken, sharedLaneToken); + return (TResult)(object)await producer.WaitAsync(wait.Token); + } + + private sealed class OwnedProducer + { + private readonly TaskCompletionSource _completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + public Task Task => _completion.Task; + public void TrySetResult(int result) => _completion.TrySetResult(result); + public void TrySetException(Exception exception) => _completion.TrySetException(exception); + } + } + + private sealed class BlockingAdmissionMiddleware : ITaskSchedulerEnqueueMiddleware + { + private readonly TaskCompletionSource _entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + public Task Entered => _entered.Task; + public void Release() => _release.TrySetResult(null); + + public async Task InvokeAsync(TaskSchedulerEnqueueContext context, TaskSchedulerEnqueueDelegate continuation) + { + _entered.TrySetResult(null); + await _release.Task; + return await continuation(context); + } + } + + private sealed class ExecutionCounterMiddleware : ITaskSchedulerExecutionMiddleware + { + public int CallCount { get; private set; } + public ValueTask InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerExecutionDelegate continuation) + { + CallCount++; + return continuation(context); + } + } + + private sealed class CompletionCounterMiddleware : ITaskSchedulerCompletionMiddleware + { + public int CallCount { get; private set; } + public ValueTask> InvokeAsync(TaskSchedulerOperationContext context, TaskSchedulerOperationOutcome outcome, TaskSchedulerCompletionDelegate continuation) + { + CallCount++; + return continuation(context, outcome); + } + } + + private sealed class CountingInlineTerminal : ITaskScheduler + { + public int EnqueueCount { get; private set; } + public async Task Enqueue(Func> taskFunc, object? state, CancellationToken cancellationToken) + { + EnqueueCount++; + return await taskFunc(state, cancellationToken); + } + } + + } +} diff --git a/docs/customization.md b/docs/customization.md index 4751296..e333ae8 100644 --- a/docs/customization.md +++ b/docs/customization.md @@ -145,6 +145,8 @@ public sealed class RejectWhenStopping : ITaskSchedulerEnqueueMiddleware Returning without invoking `continuation` prevents terminal scheduling. This supports rejection and shared-work policies. Invoke the enqueue continuation at most once: the terminal contract represents one submitted operation. +Middleware placement also defines an observation boundary. Enqueue middleware outside a short-circuiting registration observes every submission. Execution and completion middleware inside that registration observes only work that reaches the terminal. For a future shared-work policy, followers may therefore be visible to outer admission metrics while inner execution metrics run once for the shared producer. + `context.CallerCancellationToken` is always the token supplied by the caller. `context.CancellationToken` is the effective producer token currently flowing toward the terminal. To transform only the producer token, pass a copied context onward: ```csharp @@ -244,6 +246,18 @@ The slot is isolated per scheduled operation and per registration. `GetOrCreateL Middleware objects themselves are shared by all operations and must therefore keep any mutable registration-wide state thread-safe. Put operation-specific mutable state in the context slot rather than in middleware instance fields. +### Middleware author invariants + +Custom middleware should preserve these composition rules: + +- Treat scheduler configuration and contexts as immutable; return a new scheduler snapshot or pass a copied context instead of mutating shared data. +- Invoke an enqueue continuation zero or one time. A zero-call path must return an explicit result, cancellation, rejection, or shared task. +- Document execution middleware that invokes its continuation more than once. Every invocation may run all downstream execution middleware and the user operation again within the same terminal queue turn. +- Pass one final outcome through completion middleware. Forward an existing outcome unchanged unless intentionally replacing its result or exception. +- Keep registration-wide shared state concurrency-safe and operation-specific state in the registration-local context slot. +- Do not dispose the terminal scheduler, coordinators, lanes, or other collaborators unless a separate API explicitly transfers their ownership. +- Preserve the distinction between caller-wait cancellation and producer cancellation when sharing work between callers. + ### Forward-scoped annotations Annotations are immutable metadata scopes keyed by their registered type: