Improve pipeline execution performance

src/Polly.Core/CircuitBreaker/Controller/CircuitStateController.cs

@@ -18,7 +18,7 @@ internal sealed class CircuitStateController<T> : IDisposable
     private readonly TimeSpan _breakDuration;
     private readonly Func<BreakDurationGeneratorArguments, ValueTask<TimeSpan>>? _breakDurationGenerator;
     private DateTimeOffset _blockedUntil;
-    private CircuitState _circuitState = CircuitState.Closed;
+    private volatile CircuitState _circuitState = CircuitState.Closed;
     private Outcome<T>? _lastOutcome;
     private Exception? _breakingException;
     private bool _disposed;
@@ -125,6 +125,15 @@ public Task CloseCircuitAsync(ResilienceContext context)
     {
         EnsureNotDisposed();
 
+        // Fast path for the common case. A closed circuit cannot transition to another state in this method,
+        // so there is no need to take the lock. The state is only ever changed while holding the lock and
+        // the field is volatile, so the read observes the latest state. A concurrent transition after the
+        // read is indistinguishable from one that happens right after the lock would have been released.
+        if (_circuitState == CircuitState.Closed)
+        {
+            return default;
+        }
+
         BrokenCircuitException? exception = null;
         bool isHalfOpen = false;
 

src/Polly.Core/Utils/Pipeline/CompositeComponent.cs

@@ -90,6 +90,16 @@ private ValueTask<Outcome<TResult>> ExecuteCoreWithoutTelemetry<TResult, TState>
         {
             return Outcome.FromExceptionAsValueTask<TResult>(new OperationCanceledException(context.CancellationToken).TrySetStackTrace());
         }
+        else if (FirstComponent is DelegatingComponent delegating)
+        {
+            // Avoid the generic virtual method dispatch for the common case of multiple components.
+            return delegating.Execute(callback, context, state);
+        }
+        else if (FirstComponent is BridgeComponent bridge)
+        {
+            // Single non-generic strategy, skip the bridge and its generic virtual method dispatch.
+            return bridge.Strategy.ExecuteCore(callback, context, state);
+        }
         else
         {
             return FirstComponent.ExecuteCore(callback, context, state);

src/Polly.Core/Utils/Pipeline/DelegatingComponent.cs

@@ -9,15 +9,39 @@ namespace Polly.Utils.Pipeline;
 internal sealed class DelegatingComponent : PipelineComponent
 {
     private readonly PipelineComponent _component;
+    private readonly ResilienceStrategy? _strategy;
 
-    public DelegatingComponent(PipelineComponent component) => _component = component;
+    public DelegatingComponent(PipelineComponent component)
+    {
+        _component = component;
+
+        // Non-generic strategies are wrapped in a trivial bridge component. Unwrap them here so that
+        // the execution can skip the bridge and its generic virtual method dispatch.
+        _strategy = (component as BridgeComponent)?.Strategy;
+    }
 
     public PipelineComponent? Next { get; set; }
 
     public override ValueTask DisposeAsync() => default;
 
     [ExcludeFromCodeCoverage]
     internal override ValueTask<Outcome<TResult>> ExecuteCore<TResult, TState>(
+        Func<ResilienceContext, TState, ValueTask<Outcome<TResult>>> callback,
+        ResilienceContext context,
+        TState state) => Execute(callback, context, state);
+
+    /// <summary>
+    /// Non-virtual equivalent of <see cref="ExecuteCore{TResult, TState}"/>.
+    /// </summary>
+    /// <remarks>
+    /// Generic virtual method calls are dispatched through a runtime helper and are considerably more
+    /// expensive than regular calls. Callers that statically know they are dealing with a
+    /// <see cref="DelegatingComponent"/> (the composite root and the previous link in the chain)
+    /// use this method to bypass that dispatch.
+    /// </remarks>
+    [ExcludeFromCodeCoverage]
+    [MethodImpl(MethodImplOptions.AggressiveInlining)]
+    internal ValueTask<Outcome<TResult>> Execute<TResult, TState>(
         Func<ResilienceContext, TState, ValueTask<Outcome<TResult>>> callback,
         ResilienceContext context,
         TState state) =>
@@ -39,6 +63,19 @@ private static ValueTask<Outcome<TResult>> ExecuteNext<TResult, TState>(
             return Outcome.FromExceptionAsValueTask<TResult>(new OperationCanceledException(context.CancellationToken).TrySetStackTrace());
         }
 
+        // All components except the last one are delegating components, so avoid the
+        // generic virtual method dispatch for them by calling the non-virtual method directly.
+        if (next is DelegatingComponent delegating)
+        {
+            return delegating.Execute(callback, context, state);
+        }
+
+        // The last component is commonly a bridge to a non-generic strategy, skip the bridge.
+        if (next is BridgeComponent bridge)
+        {
+            return bridge.Strategy.ExecuteCore(callback, context, state);
+        }
+
         return next.ExecuteCore(callback, context, state);
     }
 
@@ -47,10 +84,20 @@ internal ValueTask<Outcome<TResult>> ExecuteComponent<TResult, TState>(
         Func<ResilienceContext, TState, ValueTask<Outcome<TResult>>> callback,
         ResilienceContext context,
         TState state)
-        => _component.ExecuteCore(
+    {
+        if (_strategy is not null)
+        {
+            return _strategy.ExecuteCore(
                 static (context, state) => ExecuteNext(state.Next!, state.callback, context, state.state),
                 context,
                 (Next, callback, state));
+        }
+
+        return _component.ExecuteCore(
+            static (context, state) => ExecuteNext(state.Next!, state.callback, context, state.state),
+            context,
+            (Next, callback, state));
+    }
 
 #if NET6_0_OR_GREATER
     // Custom state object is used to cast the callback and state to prevent infinite
@@ -60,8 +107,22 @@ internal ValueTask<Outcome<TResult>> ExecuteComponent<TResult, TState>(
     internal ValueTask<Outcome<TResult>> ExecuteComponentAot<TResult, TState>(
         Func<ResilienceContext, TState, ValueTask<Outcome<TResult>>> callback,
         ResilienceContext context,
-        TState state) =>
-        _component.ExecuteCore(
+        TState state)
+    {
+        if (_strategy is not null)
+        {
+            return _strategy.ExecuteCore(
+                static (context, wrapper) =>
+                {
+                    var callback = (Func<ResilienceContext, TState, ValueTask<Outcome<TResult>>>)wrapper.Callback;
+                    var state = (TState)wrapper.State;
+                    return ExecuteNext(wrapper.Next, callback, context, state);
+                },
+                context,
+                new StateWrapper(Next!, callback, state!));
+        }
+
+        return _component.ExecuteCore(
             static (context, wrapper) =>
             {
                 var callback = (Func<ResilienceContext, TState, ValueTask<Outcome<TResult>>>)wrapper.Callback;
@@ -70,6 +131,7 @@ internal ValueTask<Outcome<TResult>> ExecuteComponentAot<TResult, TState>(
             },
             context,
             new StateWrapper(Next!, callback, state!));
+    }
 
     private readonly record struct StateWrapper(PipelineComponent Next, object Callback, object State);
 #endif

test/Polly.Core.Tests/CircuitBreaker/Controller/CircuitStateControllerTests.cs

@@ -68,6 +68,55 @@ public async Task IsolateAsync_Ok()
         _telemetryListener.GetArgs<OnCircuitOpenedArguments<int>>().ShouldNotBeEmpty();
     }
 
+    [Fact]
+    public async Task OnActionPreExecute_CircuitClosedConcurrently_Ok()
+    {
+        // arrange
+        var cancellationToken = TestCancellation.Token;
+        using var controller = CreateController();
+        await TransitionToState(controller, CircuitState.Isolated, ResilienceContextPool.Shared.Get(cancellationToken));
+
+        var context = ResilienceContextPool.Shared.Get(cancellationToken);
+        var syncRoot = controller.GetType().GetField("_lock", BindingFlags.Instance | BindingFlags.NonPublic)!.GetValue(controller)!;
+
+        ValueTask<Outcome<int>?> pending = default;
+        var thread = new Thread(() => pending = controller.OnActionPreExecuteAsync(context));
+
+        Task closeTask;
+
+        // act
+        Monitor.Enter(syncRoot);
+
+        try
+        {
+            // The circuit is not closed so the thread skips the fast path and blocks waiting for the lock
+            thread.Start();
+
+            var stopwatch = Stopwatch.StartNew();
+
+            while (!thread.ThreadState.HasFlag(System.Threading.ThreadState.WaitSleepJoin))
+            {
+                stopwatch.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(30));
+                Thread.Yield();
+            }
+
+            // The lock is re-entrant, so the circuit can be closed before the blocked thread acquires it
+            closeTask = controller.CloseCircuitAsync(ResilienceContextPool.Shared.Get(cancellationToken));
+        }
+        finally
+        {
+            Monitor.Exit(syncRoot);
+        }
+
+        await closeTask;
+        thread.Join(TimeSpan.FromSeconds(30)).ShouldBeTrue();
+
+        // assert
+        var outcome = await pending;
+        outcome.ShouldBeNull();
+        controller.CircuitState.ShouldBe(CircuitState.Closed);
+    }
+
     [Fact]
     public async Task BreakAsync_Ok()
     {

test/Polly.Core.Tests/Utils/Pipeline/DelegatingComponentTests.cs

@@ -40,8 +40,50 @@ async static (_, state) => await Outcome.FromResultAsValueTask(state + 1),
 
         actual.Result.ShouldBe(2);
     }
+
+    [Fact]
+    public static async Task ExecuteComponentAot_WithBridgeComponent_ExecutesStrategy()
+    {
+        var executed = false;
+        var strategy = new TestResilienceStrategy { Before = (_, _) => executed = true };
+        await using var component = new BridgeComponent(strategy);
+        var next = new CallbackComponent();
+        var context = ResilienceContextPool.Shared.Get(TestCancellation.Token);
+        var state = 1;
+
+        await using var delegating = new DelegatingComponent(component) { Next = next };
+
+        var actual = await delegating.ExecuteComponentAot(
+            async static (_, state) => await Outcome.FromResultAsValueTask(state + 1),
+            context,
+            state);
+
+        actual.Result.ShouldBe(2);
+        executed.ShouldBeTrue();
+    }
 #endif
 
+    [Fact]
+    public static async Task ExecuteComponent_WithBridgeComponent_ExecutesStrategy()
+    {
+        var executed = false;
+        var strategy = new TestResilienceStrategy { Before = (_, _) => executed = true };
+        await using var component = new BridgeComponent(strategy);
+        var next = new CallbackComponent();
+        var context = ResilienceContextPool.Shared.Get(TestCancellation.Token);
+        var state = 1;
+
+        await using var delegating = new DelegatingComponent(component) { Next = next };
+
+        var actual = await delegating.ExecuteComponent(
+            async static (_, state) => await Outcome.FromResultAsValueTask(state + 1),
+            context,
+            state);
+
+        actual.Result.ShouldBe(2);
+        executed.ShouldBeTrue();
+    }
+
     private sealed class CallbackComponent : PipelineComponent
     {
         public override ValueTask DisposeAsync() => default;