From ca1580a8351ded1ae4dadcb7cf732f3b11b55642 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 15:05:20 -0700 Subject: [PATCH 1/3] Reduce per-call allocations in ChannelRecreatingCallInvoker outcome tracking ObserveOutcome previously passed a (this, methodFullName) ValueTuple as the object? state argument to Task.ContinueWith, which boxes the tuple on every unary RPC. It also relied on a non-capturing lambda for the continuation. Cache a single per-instance Action delegate (onUnaryCallCompleted) created once in the constructor, and pass the method-name string directly as ContinueWith state (already a reference type, so no boxing). ObserveOutcome now takes a non-generic Task parameter so the call resolves to Task's inherited non-generic ContinueWith(Action, object?, ...) overload for both generic and non-generic Task-returning calls. Success/failure classification, method-name diagnostics, TaskContinuationOptions.ExecuteSynchronously + TaskScheduler.Default scheduling, thread safety, channel lifecycle/disposal behavior, and the public API are all unchanged -- ObserveOutcome/OnUnaryCallCompleted are private implementation details. A standalone allocation comparison (200,000 iterations, Release, warmed up) showed a ~14.8% reduction in per-call allocation (216 -> 184 bytes/call) from removing the boxed tuple; the remainder is inherent to Task.ContinueWith's internal continuation/Task bookkeeping and out of scope per the issue. Added focused regression tests to GrpcDurableTaskClientChannelRecreationTests.cs covering: failure-count reset after an interleaved success, DeadlineExceeded suppression on long-poll methods, DeadlineExceeded counting/triggering recreate on regular methods, and cached-delegate identity across multiple calls. Fixes #777 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 3c971e29-a4bb-494b-b272-dce0c7aede23 --- .../Grpc/ChannelRecreatingCallInvoker.cs | 42 ++- ...DurableTaskClientChannelRecreationTests.cs | 268 +++++++++++++++++- 2 files changed, 286 insertions(+), 24 deletions(-) diff --git a/src/Client/Grpc/ChannelRecreatingCallInvoker.cs b/src/Client/Grpc/ChannelRecreatingCallInvoker.cs index f68ccfe9..a4857c83 100644 --- a/src/Client/Grpc/ChannelRecreatingCallInvoker.cs +++ b/src/Client/Grpc/ChannelRecreatingCallInvoker.cs @@ -38,6 +38,12 @@ sealed class ChannelRecreatingCallInvoker : CallInvoker, IAsyncDisposable readonly bool ownsChannel; readonly ILogger logger; + // Cached once per invoker instance (instead of once per RPC) so ObserveOutcome's ContinueWith call + // does not allocate a new delegate for every unary call. The method-name diagnostic is threaded + // through as the ContinueWith `state` argument (a string, already a reference type) instead of a + // boxed (self, methodName) tuple, eliminating the per-call heap allocation entirely. + readonly Action onUnaryCallCompleted; + // Cancelled in DisposeAsync so an in-flight RecreateAsync stops promptly and does not leak the // freshly created channel back into our state after we've disposed. readonly CancellationTokenSource disposalCts = new(); @@ -68,6 +74,7 @@ public ChannelRecreatingCallInvoker( this.minRecreateInterval = minRecreateInterval; this.ownsChannel = ownsChannel; this.logger = logger; + this.onUnaryCallCompleted = this.OnUnaryCallCompleted; this.state = new TransportState(initialChannel, initialChannel.CreateCallInvoker()); // Backdate the initial timestamp so the first recreate is never blocked by the cooldown. @@ -195,28 +202,33 @@ static TimeSpan ElapsedSince(long previousTimestamp, long nowTimestamp) return TimeSpan.FromSeconds((double)elapsedTicks / Stopwatch.Frequency); } - void ObserveOutcome(Task responseAsync, string methodFullName) + void ObserveOutcome(Task responseAsync, string methodFullName) { - // Use ContinueWith with TaskScheduler.Default so we don't capture sync context. + // Use ContinueWith with TaskScheduler.Default so we don't capture sync context. Both the + // continuation delegate (this.onUnaryCallCompleted, cached once per invoker instance) and the + // state (methodFullName, already a reference-typed string) are allocation-free per call: no + // boxed tuple and no per-call delegate closure. responseAsync.ContinueWith( - (t, state) => - { - var (self, name) = ((ChannelRecreatingCallInvoker, string))state!; - if (t.Status == TaskStatus.RanToCompletion) - { - self.RecordSuccess(); - } - else if (t.Exception?.InnerException is RpcException rpcEx) - { - self.RecordFailure(rpcEx.StatusCode, name); - } - }, - (this, methodFullName), + this.onUnaryCallCompleted, + methodFullName, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); } + void OnUnaryCallCompleted(Task task, object? state) + { + string methodFullName = (string)state!; + if (task.Status == TaskStatus.RanToCompletion) + { + this.RecordSuccess(); + } + else if (task.Exception?.InnerException is RpcException rpcEx) + { + this.RecordFailure(rpcEx.StatusCode, methodFullName); + } + } + void RecordSuccess() { Volatile.Write(ref this.consecutiveFailures, 0); diff --git a/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs b/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs index 29d3e910..9424a2fc 100644 --- a/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs +++ b/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs @@ -22,6 +22,12 @@ public class GrpcDurableTaskClientChannelRecreationTests "TestMethod", StringMarshaller, StringMarshaller); + static readonly Method WaitForInstanceCompletionMethod = new( + MethodType.Unary, + "TaskHubSidecarService", + "WaitForInstanceCompletion", + StringMarshaller, + StringMarshaller); static readonly MethodInfo GetCallInvokerMethod = typeof(GrpcDurableTaskClient) .GetMethod("GetCallInvoker", BindingFlags.Static | BindingFlags.NonPublic)!; static readonly MethodInfo ToStopwatchTicksMethod = typeof(ChannelRecreatingCallInvoker) @@ -93,14 +99,207 @@ public async Task GetCallInvoker_WithProvidedChannel_RecreatesTransportAfterUnar } [Fact] - public async Task GetCallInvoker_WithAddressAndRecreator_UsesWrapperThatOwnsCreatedChannel() + public async Task AsyncUnaryCall_SuccessAfterFailure_ResetsConsecutiveFailureCountAndSuppressesRecreate() + { + // Arrange: threshold of 2 consecutive failures, but a success is interleaved so the counter + // should reset and the recreator should never be invoked. + int callIndex = 0; + CallbackHttpMessageHandler handler = new((request, cancellationToken) => + { + int index = Interlocked.Increment(ref callIndex); + return Task.FromResult(index == 2 + ? CreateSuccessResponse("pong") + : CreateFailureResponse(StatusCode.Unavailable, "transient failure")); + }); + + GrpcChannel channel = CreateChannel("http://success-reset.client.test", handler); + GrpcDurableTaskClientOptions options = new() { Channel = channel }; + options.Internal.ChannelRecreateFailureThreshold = 2; + options.Internal.MinRecreateInterval = TimeSpan.Zero; + + int recreatorCalls = 0; + options.SetChannelRecreator((existingChannel, ct) => + { + Interlocked.Increment(ref recreatorCalls); + return Task.FromResult(existingChannel); + }); + + try + { + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + try + { + // Act: failure, success, failure -- the success in the middle must reset the streak so + // the trailing failure alone never reaches the threshold of 2. + await AssertRpcFailureAsync(callInvoker); + await AssertRpcSuccessAsync(callInvoker); + await AssertRpcFailureAsync(callInvoker); + + await Task.Delay(TimeSpan.FromMilliseconds(200)); + + // Assert + GetConsecutiveFailures(callInvoker).Should().Be(1); + recreatorCalls.Should().Be(0); + } + finally + { + await disposable.DisposeAsync(); + } + } + finally + { + await DisposeChannelAsync(channel); + } + } + + [Fact] + public async Task AsyncUnaryCall_DeadlineExceededOnAllowedLongPollMethod_DoesNotCountTowardThreshold() + { + // Arrange: WaitForInstanceCompletion is long-poll and DeadlineExceeded is expected behavior + // there, so repeated timeouts on it must never count toward the recreate threshold. + CallbackHttpMessageHandler handler = new((request, cancellationToken) => + Task.FromResult(CreateFailureResponse(StatusCode.DeadlineExceeded, "long-poll wait elapsed"))); + + GrpcChannel channel = CreateChannel("http://deadline-allowed.client.test", handler); + GrpcDurableTaskClientOptions options = new() { Channel = channel }; + options.Internal.ChannelRecreateFailureThreshold = 1; + options.Internal.MinRecreateInterval = TimeSpan.Zero; + + int recreatorCalls = 0; + options.SetChannelRecreator((existingChannel, ct) => + { + Interlocked.Increment(ref recreatorCalls); + return Task.FromResult(existingChannel); + }); + + try + { + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + try + { + // Act + await AssertRpcFailureAsync( + callInvoker, WaitForInstanceCompletionMethod, StatusCode.DeadlineExceeded); + await AssertRpcFailureAsync( + callInvoker, WaitForInstanceCompletionMethod, StatusCode.DeadlineExceeded); + + await Task.Delay(TimeSpan.FromMilliseconds(200)); + + // Assert + GetConsecutiveFailures(callInvoker).Should().Be(0); + recreatorCalls.Should().Be(0); + } + finally + { + await disposable.DisposeAsync(); + } + } + finally + { + await DisposeChannelAsync(channel); + } + } + + [Fact] + public async Task AsyncUnaryCall_DeadlineExceededOnRegularMethod_CountsTowardThresholdAndTriggersRecreate() + { + // Arrange: same DeadlineExceeded status as above, but on a method that is NOT in the long-poll + // allow-list, so it must count toward the threshold and trigger a recreate. + CallbackHttpMessageHandler handler = new((request, cancellationToken) => + Task.FromResult(CreateFailureResponse(StatusCode.DeadlineExceeded, "unexpected timeout"))); + + GrpcChannel channel = CreateChannel("http://deadline-counts.client.test", handler); + GrpcDurableTaskClientOptions options = new() { Channel = channel }; + options.Internal.ChannelRecreateFailureThreshold = 1; + options.Internal.MinRecreateInterval = TimeSpan.Zero; + + TaskCompletionSource recreateRequested = new(TaskCreationOptions.RunContinuationsAsynchronously); + options.SetChannelRecreator((existingChannel, ct) => { - // Arrange - GrpcDurableTaskClientOptions options = new() + recreateRequested.TrySetResult(existingChannel); + return Task.FromResult(existingChannel); + }); + + try + { + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + try { - Address = "http://owned.client.test", - }; - options.SetChannelRecreator((existingChannel, ct) => Task.FromResult(existingChannel)); + // Act + await AssertRpcFailureAsync(callInvoker, TestMethod, StatusCode.DeadlineExceeded); + GrpcChannel recreatorChannel = await recreateRequested.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + // Assert + recreatorChannel.Should().BeSameAs(channel); + } + finally + { + await disposable.DisposeAsync(); + } + } + finally + { + await DisposeChannelAsync(channel); + } + } + + [Fact] + public async Task AsyncUnaryCall_MultipleCalls_ReuseSameCachedOutcomeDelegateInstance() + { + // Arrange/Act: the outcome-observation delegate must be created exactly once per invoker + // instance (in the constructor) and reused for every call, rather than allocated per RPC. + GrpcChannel channel = GrpcChannel.ForAddress("http://cached-delegate.client.test"); + GrpcDurableTaskClientOptions options = new() { Channel = channel }; + options.SetChannelRecreator((existingChannel, ct) => Task.FromResult(existingChannel)); + + try + { + (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); + try + { + object? delegateBeforeCalls = GetOnUnaryCallCompletedDelegate(callInvoker); + delegateBeforeCalls.Should().NotBeNull(); + + for (int i = 0; i < 5; i++) + { + try + { + using AsyncUnaryCall call = callInvoker.AsyncUnaryCall( + TestMethod, host: null, new CallOptions(), request: "ping"); + await call.ResponseAsync; + } + catch + { + // The call itself fails fast against a fake address; only the delegate identity + // across calls is under test here. + } + } + + // Assert: still the exact same delegate instance (a readonly field set once in the + // constructor), proving no per-call delegate is allocated. + object? delegateAfterCalls = GetOnUnaryCallCompletedDelegate(callInvoker); + delegateAfterCalls.Should().BeSameAs(delegateBeforeCalls); + } + finally + { + await disposable.DisposeAsync(); + } + } + finally + { + await DisposeChannelAsync(channel); + } + } + + [Fact] + public async Task GetCallInvoker_WithAddressAndRecreator_UsesWrapperThatOwnsCreatedChannel() + { + // Arrange + GrpcDurableTaskClientOptions options = new() + { + Address = "http://owned.client.test", + }; + options.SetChannelRecreator((existingChannel, ct) => Task.FromResult(existingChannel)); // Act (AsyncDisposable disposable, CallInvoker callInvoker) = InvokeGetCallInvoker(options); @@ -218,17 +417,34 @@ static void SetDisposed(CallInvoker callInvoker, int value) .SetValue(callInvoker, value); } + static int GetConsecutiveFailures(CallInvoker callInvoker) + { + return (int)typeof(ChannelRecreatingCallInvoker) + .GetField("consecutiveFailures", BindingFlags.Instance | BindingFlags.NonPublic)! + .GetValue(callInvoker)!; + } + + static object GetOnUnaryCallCompletedDelegate(CallInvoker callInvoker) + { + return typeof(ChannelRecreatingCallInvoker) + .GetField("onUnaryCallCompleted", BindingFlags.Instance | BindingFlags.NonPublic)! + .GetValue(callInvoker)!; + } + static long InvokeToStopwatchTicks(TimeSpan interval) { return (long)ToStopwatchTicksMethod.Invoke(null, new object?[] { interval })!; } - static async Task AssertRpcFailureAsync(CallInvoker callInvoker) + static async Task AssertRpcFailureAsync(CallInvoker callInvoker) => + await AssertRpcFailureAsync(callInvoker, TestMethod, StatusCode.Unavailable); + + static async Task AssertRpcFailureAsync(CallInvoker callInvoker, Method method, StatusCode expectedStatus) { Func act = async () => { using AsyncUnaryCall call = callInvoker.AsyncUnaryCall( - TestMethod, + method, host: null, new CallOptions(deadline: DateTime.UtcNow.AddSeconds(1)), request: "ping"); @@ -237,7 +453,19 @@ static async Task AssertRpcFailureAsync(CallInvoker callInvoker) }; RpcException rpcException = (await act.Should().ThrowAsync()).Which; - rpcException.StatusCode.Should().Be(StatusCode.Unavailable); + rpcException.StatusCode.Should().Be(expectedStatus); + } + + static async Task AssertRpcSuccessAsync(CallInvoker callInvoker) + { + using AsyncUnaryCall call = callInvoker.AsyncUnaryCall( + TestMethod, + host: null, + new CallOptions(deadline: DateTime.UtcNow.AddSeconds(5)), + request: "ping"); + + string response = await call.ResponseAsync; + response.Should().Be("pong"); } static GrpcChannel CreateChannel(string address, HttpMessageHandler handler) @@ -268,6 +496,28 @@ static HttpResponseMessage CreateFailureResponse(StatusCode statusCode, string d return response; } + static HttpResponseMessage CreateSuccessResponse(string payload) + { + byte[] messageBytes = StringMarshaller.Serializer(payload); + byte[] frame = new byte[5 + messageBytes.Length]; + uint length = (uint)messageBytes.Length; + frame[1] = (byte)(length >> 24); + frame[2] = (byte)(length >> 16); + frame[3] = (byte)(length >> 8); + frame[4] = (byte)length; + Buffer.BlockCopy(messageBytes, 0, frame, 5, messageBytes.Length); + + HttpResponseMessage response = new(System.Net.HttpStatusCode.OK) + { + Version = new Version(2, 0), + Content = new ByteArrayContent(frame), + }; + + response.Content.Headers.ContentType = new System.Net.Http.Headers.MediaTypeHeaderValue("application/grpc"); + response.TrailingHeaders.Add("grpc-status", "0"); + return response; + } + sealed class CallbackHttpMessageHandler : HttpMessageHandler { readonly Func> callback; From 7f73ac8cf11b0ceebc2db59df9e72563c5a618d0 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 16:13:53 -0700 Subject: [PATCH 2/3] Make cached-delegate test deterministic (address PR review) AsyncUnaryCall_MultipleCalls_ReuseSameCachedOutcomeDelegateInstance used GrpcChannel.ForAddress against a real (unresolvable) hostname with no deadline and a broad catch, which could hang or behave unpredictably in CI depending on DNS/proxy/network availability. Switch to the same deterministic CallbackHttpMessageHandler pattern used by the other tests in this file (always returns Unavailable), add an explicit short deadline, and narrow the catch to the expected RpcException. The delegate-identity assertion under test is unchanged. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 3c971e29-a4bb-494b-b272-dce0c7aede23 --- ...cDurableTaskClientChannelRecreationTests.cs | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs b/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs index 9424a2fc..6945ead5 100644 --- a/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs +++ b/test/Client/Grpc.Tests/GrpcDurableTaskClientChannelRecreationTests.cs @@ -248,7 +248,12 @@ public async Task AsyncUnaryCall_MultipleCalls_ReuseSameCachedOutcomeDelegateIns { // Arrange/Act: the outcome-observation delegate must be created exactly once per invoker // instance (in the constructor) and reused for every call, rather than allocated per RPC. - GrpcChannel channel = GrpcChannel.ForAddress("http://cached-delegate.client.test"); + // Use the deterministic fake-handler pattern (as in the other tests in this file) instead of + // a real address, so the test never depends on DNS/proxy/network and cannot hang in CI. + CallbackHttpMessageHandler handler = new((request, cancellationToken) => + Task.FromResult(CreateFailureResponse(StatusCode.Unavailable, "deterministic failure"))); + + GrpcChannel channel = CreateChannel("http://cached-delegate.client.test", handler); GrpcDurableTaskClientOptions options = new() { Channel = channel }; options.SetChannelRecreator((existingChannel, ct) => Task.FromResult(existingChannel)); @@ -265,13 +270,16 @@ public async Task AsyncUnaryCall_MultipleCalls_ReuseSameCachedOutcomeDelegateIns try { using AsyncUnaryCall call = callInvoker.AsyncUnaryCall( - TestMethod, host: null, new CallOptions(), request: "ping"); + TestMethod, + host: null, + new CallOptions(deadline: DateTime.UtcNow.AddSeconds(1)), + request: "ping"); await call.ResponseAsync; } - catch + catch (RpcException) { - // The call itself fails fast against a fake address; only the delegate identity - // across calls is under test here. + // Each call deterministically fails via the fake handler above; only the + // delegate identity across calls is under test here. } } From 23d11e3fad00d89d37c05e61fc28f4f11c758ca5 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Mon, 27 Jul 2026 11:35:33 -0700 Subject: [PATCH 3/3] Address PR hygiene comments: comment accuracy and parameter shadowing - Reworded the cached-delegate comment to accurately scope the allocation claim to the per-call delegate and boxed tuple that were eliminated, rather than implying all of Task.ContinueWith's internal continuation/Task bookkeeping is allocation-free. - Renamed OnUnaryCallCompleted's `state` parameter to `continuationState` to avoid shadowing the instance `state` (TransportState) field. No behavior change; targeted channel-recreation tests still pass. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 3c971e29-a4bb-494b-b272-dce0c7aede23 --- src/Client/Grpc/ChannelRecreatingCallInvoker.cs | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/src/Client/Grpc/ChannelRecreatingCallInvoker.cs b/src/Client/Grpc/ChannelRecreatingCallInvoker.cs index a4857c83..f0928876 100644 --- a/src/Client/Grpc/ChannelRecreatingCallInvoker.cs +++ b/src/Client/Grpc/ChannelRecreatingCallInvoker.cs @@ -41,7 +41,8 @@ sealed class ChannelRecreatingCallInvoker : CallInvoker, IAsyncDisposable // Cached once per invoker instance (instead of once per RPC) so ObserveOutcome's ContinueWith call // does not allocate a new delegate for every unary call. The method-name diagnostic is threaded // through as the ContinueWith `state` argument (a string, already a reference type) instead of a - // boxed (self, methodName) tuple, eliminating the per-call heap allocation entirely. + // boxed (self, methodName) tuple, eliminating that per-call delegate and boxed-tuple allocation + // (Task.ContinueWith's own internal continuation/Task bookkeeping is unaffected). readonly Action onUnaryCallCompleted; // Cancelled in DisposeAsync so an in-flight RecreateAsync stops promptly and does not leak the @@ -206,8 +207,8 @@ void ObserveOutcome(Task responseAsync, string methodFullName) { // Use ContinueWith with TaskScheduler.Default so we don't capture sync context. Both the // continuation delegate (this.onUnaryCallCompleted, cached once per invoker instance) and the - // state (methodFullName, already a reference-typed string) are allocation-free per call: no - // boxed tuple and no per-call delegate closure. + // continuation state (methodFullName, already a reference-typed string) are allocation-free + // per call: no boxed tuple and no per-call delegate closure. responseAsync.ContinueWith( this.onUnaryCallCompleted, methodFullName, @@ -216,9 +217,9 @@ void ObserveOutcome(Task responseAsync, string methodFullName) TaskScheduler.Default); } - void OnUnaryCallCompleted(Task task, object? state) + void OnUnaryCallCompleted(Task task, object? continuationState) { - string methodFullName = (string)state!; + string methodFullName = (string)continuationState!; if (task.Status == TaskStatus.RanToCompletion) { this.RecordSuccess();