From fa48d63135693641f5d56ad49a47199e23299c96 Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 25 Aug 2026 14:42:26 -0400 Subject: [PATCH 1/2] Release abandoned orchestrator tasks when an executor is retired TaskOrchestrationContext creates a TaskCompletionSource per activity, sub-orchestration and timer, and the orchestrator awaits it. When an episode ends while the orchestrator is parked on one of those awaits, the TCS is deliberately abandoned in a pending state -- that is inherent to the replay model. When a debugger is attached, TaskAwaiter.OnCompletedInternal calls OutputWaitEtwEvents, which calls Task.AddToActiveTasks because Task.s_asyncDebuggingEnabled is true. That inserts the task into the process-wide, strongly-referenced static dictionary Task.s_currentActiveTasks. The matching Task.RemoveFromActiveTasks only runs inside the awaiter continuation wrapper, i.e. only when the awaited task completes. Abandoned tasks are therefore rooted forever, and each rooted Task pins its continuation -> async state machine -> TaskOrchestrationContext -> history/inputs/outputs. Only the AddToActiveTasks branch roots the task, which is why the leak is debugger-specific. Fix: cancel the open tasks once the executor is guaranteed never to run again, so the abandoned awaiter continuations run and unregister themselves. - TaskOrchestrationContext.ReleaseOpenTasks() cancels every open task. It is idempotent, sets isReleased before cancelling, and snapshots openTasks before iterating because resumed user code can mutate the dictionary. A ThrowIfReleased guard on ScheduleTaskInternal, CreateSubOrchestrationInstanceCore and CreateTimer stops orchestrator code that swallows the cancellation from scheduling new work. - TaskOrchestrationExecutor implements IDisposable. Dispose() restores the orchestrator ambient environment, releases the open tasks, and observes a faulted result so nothing surfaces as an UnobservedTaskException. - TaskOrchestrationDispatcher owns executor lifetime via workItem.Cursor, so it retires the executor from a new finally on OnProcessWorkItemSessionAsync and at the continue-as-new site. Extended sessions are unaffected: release only happens when the executor is retired, never between episodes. ExtendedSession_OpenTasksSurviveBetweenEpisodes covers this. Ref https://github.com/Azure/azure-functions-durable-extension/issues/340 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../TaskOrchestrationExecutorTests.cs | 344 ++++++++++++++++++ .../TaskOrchestrationContext.cs | 77 ++++ .../TaskOrchestrationDispatcher.cs | 22 +- .../TaskOrchestrationExecutor.cs | 48 ++- 4 files changed, 489 insertions(+), 2 deletions(-) create mode 100644 Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs diff --git a/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs b/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs new file mode 100644 index 000000000..b0e503053 --- /dev/null +++ b/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs @@ -0,0 +1,344 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- + +namespace DurableTask.Core.Tests +{ + using System; + using System.Collections.Generic; + using System.Linq; + using System.Reflection; + using System.Threading.Tasks; + using DurableTask.Core.Command; + using DurableTask.Core.History; + using DurableTask.Core.Serializing; + using Microsoft.VisualStudio.TestTools.UnitTesting; + + /// + /// Tests for lifetime management. + /// + /// + /// An orchestrator parks on a for every activity, + /// sub-orchestration, and timer it is waiting on, and those tasks are abandoned in a pending state when + /// the episode ends. When a debugger is attached, the CLR keeps every awaited task in the process-wide + /// Task.s_currentActiveTasks dictionary until the task completes, so abandoned awaits permanently + /// root the orchestration object graph. Disposing the executor cancels the open tasks, which lets those + /// awaiter continuations run and unregister themselves. + /// Regression coverage for https://github.com/Azure/azure-functions-durable-extension/issues/340. + /// + [TestClass] + public class TaskOrchestrationExecutorTests + { + const string ActivityName = "SayHello"; + + [TestMethod] + public void Dispose_ResumesAbandonedOrchestratorContinuations() + { + var orchestration = new FanOutOrchestration(fanOut: 3); + using (var executor = CreateExecutor(orchestration)) + { + executor.Execute(); + + Assert.AreEqual(3, orchestration.StartedTaskCount, "The orchestrator should have scheduled 3 activities."); + Assert.AreEqual(0, orchestration.ReleasedTaskCount, "Abandoned awaits should still be pending at the end of the episode."); + } + + Assert.AreEqual( + 3, + orchestration.ReleasedTaskCount, + "Disposing the executor should resume every abandoned await so its continuation can unregister itself."); + } + + [TestMethod] + public void Dispose_WithAsyncDebuggingEnabled_DoesNotLeakActiveTasks() + { + // This is the actual bug: with a debugger attached, every abandoned await stays in + // Task.s_currentActiveTasks forever, which roots the entire orchestration object graph. + // + // Task.s_currentActiveTasks is process-wide, so the test host's own async plumbing shows up in + // it as well. Rather than trying to subtract that noise, this measures how the growth scales + // with the number of abandoned awaits: a leak is proportional to the fan-out, while unrelated + // noise is not. + const int Episodes = 20; + const int SmallFanOut = 1; + const int LargeFanOut = 25; + + using (AsyncDebuggingScope.Enable()) + { + // Warm up so that one-time allocations aren't counted against the measurement. + RunEpisodes(Episodes, SmallFanOut, dispose: true); + + int leakySensitivity = MeasureFanOutSensitivity(Episodes, SmallFanOut, LargeFanOut, dispose: false); + Assert.IsTrue( + leakySensitivity > Episodes * (LargeFanOut - SmallFanOut), + "Undisposed executors are expected to leak one entry per abandoned await; if they no longer " + + $"do, this test can no longer detect the regression. Measured {leakySensitivity}."); + + int fixedSensitivity = MeasureFanOutSensitivity(Episodes, SmallFanOut, LargeFanOut, dispose: true); + Assert.IsTrue( + fixedSensitivity * 20 < leakySensitivity, + "Disposing the executor must stop Task.s_currentActiveTasks from growing with the number of " + + $"abandoned awaits, but growth was still {fixedSensitivity} against {leakySensitivity} when " + + "the executors were left undisposed."); + } + } + + /// + /// Returns how much more the active task table grows for abandoned + /// awaits per episode than it does for . Constant per-episode overhead + /// and unrelated test host activity cancel out, leaving only growth caused by abandoned awaits. + /// + static int MeasureFanOutSensitivity(int episodes, int smallFanOut, int largeFanOut, bool dispose) + { + int small = RunEpisodes(episodes, smallFanOut, dispose); + int large = RunEpisodes(episodes, largeFanOut, dispose); + return large - small; + } + + static int RunEpisodes(int episodes, int fanOut, bool dispose) + { + int before = AsyncDebuggingScope.ActiveTaskCount; + for (int i = 0; i < episodes; i++) + { + TaskOrchestrationExecutor executor = CreateExecutor(new FanOutOrchestration(fanOut)); + executor.Execute(); + if (dispose) + { + executor.Dispose(); + } + } + + return AsyncDebuggingScope.ActiveTaskCount - before; + } + + [TestMethod] + public void Dispose_DoesNotChangeTheDecisionsAlreadyProduced() + { + var orchestration = new FanOutOrchestration(fanOut: 3); + using (var executor = CreateExecutor(orchestration)) + { + OrchestratorExecutionResult result = executor.Execute(); + List before = result.Actions.ToList(); + + executor.Dispose(); + + CollectionAssert.AreEqual( + before, + result.Actions.ToList(), + "Releasing abandoned tasks must not add or remove orchestrator actions."); + } + } + + [TestMethod] + public void Dispose_OrchestratorThatSwallowsCancellation_CannotScheduleMoreWork() + { + var orchestration = new SwallowsCancellationOrchestration(); + using (var executor = CreateExecutor(orchestration)) + { + executor.Execute(); + } + + Assert.IsInstanceOfType( + orchestration.RescheduleFailure, + typeof(OperationCanceledException), + "A released context must refuse to open new tasks, otherwise resumed orchestrator code can leak again."); + } + + [TestMethod] + public void Dispose_IsIdempotent() + { + var orchestration = new FanOutOrchestration(fanOut: 2); + var executor = CreateExecutor(orchestration); + executor.Execute(); + + executor.Dispose(); + executor.Dispose(); + + Assert.AreEqual(2, orchestration.ReleasedTaskCount, "Repeated disposal should not resume continuations more than once."); + } + + [TestMethod] + public void ExtendedSession_OpenTasksSurviveBetweenEpisodes() + { + // Extended sessions reuse the executor across episodes, so open tasks must stay pending + // until their results arrive. Only the end of the session may release them. + var orchestration = new FanOutOrchestration(fanOut: 1); + OrchestrationRuntimeState runtimeState = CreateRuntimeState(); + + using (var executor = new TaskOrchestrationExecutor(runtimeState, orchestration, BehaviorOnContinueAsNew.Carryover)) + { + OrchestratorExecutionResult firstEpisode = executor.Execute(); + Assert.AreEqual(1, firstEpisode.Actions.Count(), "The first episode should schedule the activity."); + Assert.IsFalse(executor.IsCompleted); + + runtimeState.NewEvents.Clear(); + runtimeState.AddEvent(new OrchestratorStartedEvent(-1)); + runtimeState.AddEvent(new TaskCompletedEvent(-1, taskScheduledId: 0, result: JsonDataConverter.Default.Serialize("Hello"))); + + OrchestratorExecutionResult secondEpisode = executor.ExecuteNewEvents(); + + Assert.IsTrue(executor.IsCompleted, "The activity result should have been delivered to the still-open task."); + Assert.AreEqual(1, orchestration.ReleasedTaskCount, "The await should have been resumed by its result, not by cancellation."); + Assert.IsTrue( + secondEpisode.Actions.OfType().Any(), + "The orchestration should have completed on the second episode."); + } + } + + static TaskOrchestrationExecutor CreateExecutor(TaskOrchestration orchestration) => + new TaskOrchestrationExecutor(CreateRuntimeState(), orchestration, BehaviorOnContinueAsNew.Carryover); + + static OrchestrationRuntimeState CreateRuntimeState() + { + var runtimeState = new OrchestrationRuntimeState(); + runtimeState.AddEvent(new OrchestratorStartedEvent(-1)); + runtimeState.AddEvent(new ExecutionStartedEvent(-1, null) + { + OrchestrationInstance = new OrchestrationInstance + { + InstanceId = Guid.NewGuid().ToString("N"), + ExecutionId = Guid.NewGuid().ToString("N"), + }, + Name = "TestOrchestration", + Version = string.Empty, + }); + + return runtimeState; + } + + /// + /// Fans out to several activities and then parks, which is the state an orchestrator is in at the + /// end of a typical episode. Each await records whether it was ever resumed. + /// + class FanOutOrchestration : TaskOrchestration + { + readonly int fanOut; + + public FanOutOrchestration(int fanOut) + { + this.fanOut = fanOut; + } + + public int StartedTaskCount { get; private set; } + + public int ReleasedTaskCount { get; private set; } + + public override async Task RunTask(OrchestrationContext context, string input) + { + var tasks = new List>(this.fanOut); + for (int i = 0; i < this.fanOut; i++) + { + this.StartedTaskCount++; + tasks.Add(this.AwaitActivityAsync(context)); + } + + await Task.WhenAll(tasks); + return string.Empty; + } + + async Task AwaitActivityAsync(OrchestrationContext context) + { + try + { + return await context.ScheduleTask(ActivityName, string.Empty); + } + finally + { + this.ReleasedTaskCount++; + } + } + } + + /// + /// Mimics orchestrator code with a catch-all handler: it swallows the cancellation raised while the + /// executor is being released and then tries to schedule more work. + /// + class SwallowsCancellationOrchestration : TaskOrchestration + { + public Exception RescheduleFailure { get; private set; } + + public override async Task RunTask(OrchestrationContext context, string input) + { + try + { + return await context.ScheduleTask(ActivityName, string.Empty); + } + catch (Exception) + { + try + { + return await context.ScheduleTask(ActivityName, string.Empty); + } + catch (Exception e) + { + this.RescheduleFailure = e; + throw; + } + } + } + } + + /// + /// Turns on the CLR's async debugging bookkeeping for the duration of a test, which is what a + /// attached debugger does, and exposes the size of the tracking dictionary. + /// + sealed class AsyncDebuggingScope : IDisposable + { + const BindingFlags StaticFlags = BindingFlags.Static | BindingFlags.NonPublic | BindingFlags.Public; + + static readonly FieldInfo EnabledField = typeof(Task).GetField("s_asyncDebuggingEnabled", StaticFlags); + static readonly FieldInfo ActiveTasksField = typeof(Task).GetField("s_currentActiveTasks", StaticFlags); + + readonly bool previousValue; + + AsyncDebuggingScope(bool previousValue) + { + this.previousValue = previousValue; + } + + public static AsyncDebuggingScope Enable() + { + if (EnabledField == null || ActiveTasksField == null) + { + Assert.Inconclusive("This runtime does not expose the async debugging state that this test relies on."); + } + + var scope = new AsyncDebuggingScope((bool)EnabledField.GetValue(null)); + EnabledField.SetValue(null, true); + return scope; + } + + /// + /// The size of the process-wide Task.s_currentActiveTasks table. + /// + public static int ActiveTaskCount + { + get + { + object activeTasks = ActiveTasksField.GetValue(null); + if (activeTasks == null) + { + return 0; + } + + PropertyInfo count = activeTasks.GetType().GetProperty("Count"); + lock (activeTasks) + { + return (int)count.GetValue(activeTasks); + } + } + } + + public void Dispose() => EnabledField.SetValue(null, this.previousValue); + } + } +} diff --git a/src/DurableTask.Core/TaskOrchestrationContext.cs b/src/DurableTask.Core/TaskOrchestrationContext.cs index e4846124c..cbb9bffd2 100644 --- a/src/DurableTask.Core/TaskOrchestrationContext.cs +++ b/src/DurableTask.Core/TaskOrchestrationContext.cs @@ -35,6 +35,7 @@ internal class TaskOrchestrationContext : OrchestrationContext private OrchestrationCompleteOrchestratorAction continueAsNew; private static readonly ContinueAsNewOptions DefaultContinueAsNewOptions = new ContinueAsNewOptions(); private bool executionCompletedOrTerminated; + private bool isReleased; private int idCounter; private readonly Queue eventsWhileSuspended; private readonly IDictionary suspendedActionsMap; @@ -75,6 +76,76 @@ public TaskOrchestrationContext( public bool HasOpenTasks => this.openTasks.Count > 0; + /// + /// Cancels every open task so that the orchestrator's abandoned await continuations are released. + /// + /// + /// + /// Orchestrator code parks on a for every activity, + /// sub-orchestration, and timer it is waiting on. When an episode ends, those tasks are simply + /// abandoned in a pending state, because their results are not yet known. That is harmless to the + /// garbage collector on its own, but when a debugger is attached the CLR records every awaited task + /// in the process-wide Task.s_currentActiveTasks dictionary and only removes the entry when + /// the awaited task completes. Abandoned tasks therefore stay rooted forever, and with them the + /// entire orchestration object graph (context, history, inputs, and outputs). + /// + /// + /// Cancelling the open tasks lets each awaiter continuation run and unregister itself, which is what + /// allows the graph to be collected. This is only safe once the executor is guaranteed never to be + /// used again, since a cancelled task can no longer receive a result on a subsequent episode. + /// + /// + internal void ReleaseOpenTasks() + { + if (this.isReleased) + { + return; + } + + // Set this before cancelling anything: cancellation resumes orchestrator code, and that code must + // not be able to open new tasks that would leak in exactly the same way. + this.isReleased = true; + + if (this.openTasks.Count == 0) + { + return; + } + + // Resumed orchestrator code can mutate openTasks (for example via a timer cancellation callback), + // so cancel from a snapshot rather than while enumerating the live dictionary. + List abandonedTasks = this.openTasks.Values.ToList(); + this.openTasks.Clear(); + + foreach (OpenTaskInfo info in abandonedTasks) + { + try + { + info.Result.TrySetCanceled(); + } + catch (Exception e) when (!Utils.IsFatal(e)) + { + // Orchestrator code observed the cancellation and threw. The episode is already over and + // its decisions have already been captured, so there is nothing to report. Swallow the + // exception and keep going so the remaining tasks still get released. + TraceHelper.TraceSession( + TraceEventType.Warning, + "TaskOrchestrationContext-ReleaseOpenTasks", + OrchestrationInstance?.InstanceId, + "Exception while releasing abandoned orchestrator tasks: {0}", + e); + } + } + } + + private void ThrowIfReleased() + { + if (this.isReleased) + { + throw new OperationCanceledException( + "This orchestration episode has ended and the orchestration context can no longer schedule work."); + } + } + internal void ClearPendingActions() { this.orchestratorActionsMap.Clear(); @@ -126,6 +197,8 @@ public async Task ScheduleTaskToWorker(string name, string ver public async Task ScheduleTaskInternal(string name, string version, string taskList, Type resultType, ScheduleTaskOptions options, params object[] parameters) { + ThrowIfReleased(); + int id = this.idCounter++; string serializedInput = this.MessageDataConverter.SerializeInternal(parameters); var scheduleTaskTaskAction = new ScheduleTaskOrchestratorAction @@ -189,6 +262,8 @@ async Task CreateSubOrchestrationInstanceCore( object input, IDictionary tags) { + ThrowIfReleased(); + int id = this.idCounter++; string serializedInput = this.MessageDataConverter.SerializeInternal(input); @@ -295,6 +370,8 @@ public override async Task CreateTimer(DateTime fireAt, T state, Cancellat paramName: nameof(state)); } + ThrowIfReleased(); + int id = this.idCounter++; var createTimerOrchestratorAction = new CreateTimerOrchestratorAction { diff --git a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs index 649e7b47a..95474a122 100644 --- a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs +++ b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs @@ -308,6 +308,24 @@ async Task OnProcessWorkItemSessionAsync(TaskOrchestrationWorkItem workItem) TraceHelper.TraceInstance(TraceEventType.Warning, "TaskOrchestrationDispatcher-ExecutionAborted", instance, "{0}", e.Message); await this.orchestrationService.AbandonTaskOrchestrationWorkItemAsync(workItem); } + finally + { + // The session is over and the executor will never run again, so release the orchestrator + // continuations it abandoned while waiting on activities, sub-orchestrations, and timers. + // Leaving them pending leaks the whole orchestration object graph when a debugger is + // attached. See https://github.com/Azure/azure-functions-durable-extension/issues/340. + ReleaseCursor(ref workItem.Cursor); + } + } + + /// + /// Retires the executor held by and clears the reference. + /// + static void ReleaseCursor(ref OrchestrationExecutionCursor? cursor) + { + OrchestrationExecutionCursor? retiredCursor = cursor; + cursor = null; + retiredCursor?.OrchestrationExecutor?.Dispose(); } /// @@ -669,7 +687,9 @@ protected async Task OnProcessWorkItemAsync(TaskOrchestrationWorkItem work runtimeState.AddEvent(new OrchestratorCompletedEvent(-1)); workItem.OrchestrationRuntimeState = runtimeState; - workItem.Cursor = null; + // The continued-as-new execution gets a brand new executor, so retire this + // one instead of just dropping the reference. + ReleaseCursor(ref workItem.Cursor); traceActivity = RestartTraceActivityForContinueAsNewIfNeeded( traceActivity, diff --git a/src/DurableTask.Core/TaskOrchestrationExecutor.cs b/src/DurableTask.Core/TaskOrchestrationExecutor.cs index 540851e50..0f4f6545a 100644 --- a/src/DurableTask.Core/TaskOrchestrationExecutor.cs +++ b/src/DurableTask.Core/TaskOrchestrationExecutor.cs @@ -28,7 +28,7 @@ namespace DurableTask.Core /// /// Utility for executing task orchestrators. /// - public class TaskOrchestrationExecutor + public class TaskOrchestrationExecutor : IDisposable { readonly TaskOrchestrationContext context; readonly TaskScheduler decisionScheduler; @@ -135,6 +135,52 @@ public OrchestratorExecutionResult ExecuteNewEvents() newEvents: this.orchestrationRuntimeState.NewEvents); } + /// + /// Releases the orchestrator continuations that this executor abandoned while waiting on + /// activities, sub-orchestrations, or timers. + /// + /// + /// + /// Call this once the executor is guaranteed never to run again, i.e. when the orchestration + /// session ends. It must not be called between episodes of an extended session, because an open + /// task still needs to be able to receive its result on a later episode. + /// + /// + /// Skipping this call is not a correctness problem, but it leaks memory whenever a debugger is + /// attached: the CLR keeps every awaited task in a process-wide dictionary until that task + /// completes, so abandoned orchestrator awaits permanently root the orchestration object graph. + /// See https://github.com/Azure/azure-functions-durable-extension/issues/340. + /// + /// + public void Dispose() + { + SynchronizationContext prevCtx = SynchronizationContext.Current; + bool prevIsOrchestratorThread = OrchestrationContext.IsOrchestratorThread; + + try + { + // Cancelling the open tasks resumes orchestrator code, so give it the same ambient + // environment it sees during a normal episode. + SynchronizationContext.SetSynchronizationContext( + new TaskOrchestrationSynchronizationContext(this.decisionScheduler)); + OrchestrationContext.IsOrchestratorThread = true; + + this.context.ReleaseOpenTasks(); + } + finally + { + SynchronizationContext.SetSynchronizationContext(prevCtx); + OrchestrationContext.IsOrchestratorThread = prevIsOrchestratorThread; + + // Unwinding may have faulted the orchestrator's top-level task. Nothing observes it at this + // point, so observe it here to keep it from surfacing as an unobserved task exception. + if (this.result?.IsFaulted == true) + { + _ = this.result.Exception; + } + } + } + OrchestratorExecutionResult ExecuteCore(IEnumerable pastEvents, IEnumerable newEvents) { SynchronizationContext prevCtx = SynchronizationContext.Current; From da21249cf73ea21765560b712970cc2433594187 Mon Sep 17 00:00:00 2001 From: wangbill Date: Thu, 27 Aug 2026 17:44:27 -0400 Subject: [PATCH 2/2] Make the executor release hook internal instead of IDisposable TaskOrchestrationExecutor is a public type, so implementing IDisposable added public API surface that no caller outside DurableTask.Core can use correctly: executor lifetime is owned by TaskOrchestrationDispatcher, which is the only component that knows when an executor will never run again. It also risked tripping CA2000 in downstream repos that construct an executor directly. Replace the public Dispose() with an internal Release() and have ReleaseCursor call that instead. The cleanup behavior is unchanged: install the orchestrator ambient context, call context.ReleaseOpenTasks(), restore ambient state in a finally, and observe the top-level task exception so it cannot surface as an UnobservedTaskException. Cursor clearing and all lifecycle placement in the dispatcher are untouched. Tests call Release() explicitly rather than relying on a using block, and are renamed Dispose_* -> Release_*. All six behavioral guarantees are preserved, including ExtendedSession_OpenTasksSurviveBetweenEpisodes. The public surface of TaskOrchestrationExecutor is now exactly what it was before this PR: three constructors, Execute, ExecuteNewEvents, and IsCompleted. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 019e7654-1037-4e02-b2fd-c03e09d6b7e2 --- .../TaskOrchestrationExecutorTests.cs | 114 +++++++++--------- .../TaskOrchestrationDispatcher.cs | 2 +- .../TaskOrchestrationExecutor.cs | 10 +- 3 files changed, 65 insertions(+), 61 deletions(-) diff --git a/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs b/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs index b0e503053..6d7403b10 100644 --- a/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs +++ b/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs @@ -31,7 +31,7 @@ namespace DurableTask.Core.Tests /// sub-orchestration, and timer it is waiting on, and those tasks are abandoned in a pending state when /// the episode ends. When a debugger is attached, the CLR keeps every awaited task in the process-wide /// Task.s_currentActiveTasks dictionary until the task completes, so abandoned awaits permanently - /// root the orchestration object graph. Disposing the executor cancels the open tasks, which lets those + /// root the orchestration object graph. Releasing the executor cancels the open tasks, which lets those /// awaiter continuations run and unregister themselves. /// Regression coverage for https://github.com/Azure/azure-functions-durable-extension/issues/340. /// @@ -41,25 +41,25 @@ public class TaskOrchestrationExecutorTests const string ActivityName = "SayHello"; [TestMethod] - public void Dispose_ResumesAbandonedOrchestratorContinuations() + public void Release_ResumesAbandonedOrchestratorContinuations() { var orchestration = new FanOutOrchestration(fanOut: 3); - using (var executor = CreateExecutor(orchestration)) - { - executor.Execute(); + TaskOrchestrationExecutor executor = CreateExecutor(orchestration); + executor.Execute(); - Assert.AreEqual(3, orchestration.StartedTaskCount, "The orchestrator should have scheduled 3 activities."); - Assert.AreEqual(0, orchestration.ReleasedTaskCount, "Abandoned awaits should still be pending at the end of the episode."); - } + Assert.AreEqual(3, orchestration.StartedTaskCount, "The orchestrator should have scheduled 3 activities."); + Assert.AreEqual(0, orchestration.ReleasedTaskCount, "Abandoned awaits should still be pending at the end of the episode."); + + executor.Release(); Assert.AreEqual( 3, orchestration.ReleasedTaskCount, - "Disposing the executor should resume every abandoned await so its continuation can unregister itself."); + "Releasing the executor should resume every abandoned await so its continuation can unregister itself."); } [TestMethod] - public void Dispose_WithAsyncDebuggingEnabled_DoesNotLeakActiveTasks() + public void Release_WithAsyncDebuggingEnabled_DoesNotLeakActiveTasks() { // This is the actual bug: with a debugger attached, every abandoned await stays in // Task.s_currentActiveTasks forever, which roots the entire orchestration object graph. @@ -75,20 +75,20 @@ public void Dispose_WithAsyncDebuggingEnabled_DoesNotLeakActiveTasks() using (AsyncDebuggingScope.Enable()) { // Warm up so that one-time allocations aren't counted against the measurement. - RunEpisodes(Episodes, SmallFanOut, dispose: true); + RunEpisodes(Episodes, SmallFanOut, release: true); - int leakySensitivity = MeasureFanOutSensitivity(Episodes, SmallFanOut, LargeFanOut, dispose: false); + int leakySensitivity = MeasureFanOutSensitivity(Episodes, SmallFanOut, LargeFanOut, release: false); Assert.IsTrue( leakySensitivity > Episodes * (LargeFanOut - SmallFanOut), - "Undisposed executors are expected to leak one entry per abandoned await; if they no longer " + + "Unreleased executors are expected to leak one entry per abandoned await; if they no longer " + $"do, this test can no longer detect the regression. Measured {leakySensitivity}."); - int fixedSensitivity = MeasureFanOutSensitivity(Episodes, SmallFanOut, LargeFanOut, dispose: true); + int fixedSensitivity = MeasureFanOutSensitivity(Episodes, SmallFanOut, LargeFanOut, release: true); Assert.IsTrue( fixedSensitivity * 20 < leakySensitivity, - "Disposing the executor must stop Task.s_currentActiveTasks from growing with the number of " + + "Releasing the executor must stop Task.s_currentActiveTasks from growing with the number of " + $"abandoned awaits, but growth was still {fixedSensitivity} against {leakySensitivity} when " + - "the executors were left undisposed."); + "the executors were left unreleased."); } } @@ -97,23 +97,23 @@ public void Dispose_WithAsyncDebuggingEnabled_DoesNotLeakActiveTasks() /// awaits per episode than it does for . Constant per-episode overhead /// and unrelated test host activity cancel out, leaving only growth caused by abandoned awaits. /// - static int MeasureFanOutSensitivity(int episodes, int smallFanOut, int largeFanOut, bool dispose) + static int MeasureFanOutSensitivity(int episodes, int smallFanOut, int largeFanOut, bool release) { - int small = RunEpisodes(episodes, smallFanOut, dispose); - int large = RunEpisodes(episodes, largeFanOut, dispose); + int small = RunEpisodes(episodes, smallFanOut, release); + int large = RunEpisodes(episodes, largeFanOut, release); return large - small; } - static int RunEpisodes(int episodes, int fanOut, bool dispose) + static int RunEpisodes(int episodes, int fanOut, bool release) { int before = AsyncDebuggingScope.ActiveTaskCount; for (int i = 0; i < episodes; i++) { TaskOrchestrationExecutor executor = CreateExecutor(new FanOutOrchestration(fanOut)); executor.Execute(); - if (dispose) + if (release) { - executor.Dispose(); + executor.Release(); } } @@ -121,31 +121,30 @@ static int RunEpisodes(int episodes, int fanOut, bool dispose) } [TestMethod] - public void Dispose_DoesNotChangeTheDecisionsAlreadyProduced() + public void Release_DoesNotChangeTheDecisionsAlreadyProduced() { var orchestration = new FanOutOrchestration(fanOut: 3); - using (var executor = CreateExecutor(orchestration)) - { - OrchestratorExecutionResult result = executor.Execute(); - List before = result.Actions.ToList(); + TaskOrchestrationExecutor executor = CreateExecutor(orchestration); - executor.Dispose(); + OrchestratorExecutionResult result = executor.Execute(); + List before = result.Actions.ToList(); - CollectionAssert.AreEqual( - before, - result.Actions.ToList(), - "Releasing abandoned tasks must not add or remove orchestrator actions."); - } + executor.Release(); + + CollectionAssert.AreEqual( + before, + result.Actions.ToList(), + "Releasing abandoned tasks must not add or remove orchestrator actions."); } [TestMethod] - public void Dispose_OrchestratorThatSwallowsCancellation_CannotScheduleMoreWork() + public void Release_OrchestratorThatSwallowsCancellation_CannotScheduleMoreWork() { var orchestration = new SwallowsCancellationOrchestration(); - using (var executor = CreateExecutor(orchestration)) - { - executor.Execute(); - } + TaskOrchestrationExecutor executor = CreateExecutor(orchestration); + executor.Execute(); + + executor.Release(); Assert.IsInstanceOfType( orchestration.RescheduleFailure, @@ -154,16 +153,16 @@ public void Dispose_OrchestratorThatSwallowsCancellation_CannotScheduleMoreWork( } [TestMethod] - public void Dispose_IsIdempotent() + public void Release_IsIdempotent() { var orchestration = new FanOutOrchestration(fanOut: 2); - var executor = CreateExecutor(orchestration); + TaskOrchestrationExecutor executor = CreateExecutor(orchestration); executor.Execute(); - executor.Dispose(); - executor.Dispose(); + executor.Release(); + executor.Release(); - Assert.AreEqual(2, orchestration.ReleasedTaskCount, "Repeated disposal should not resume continuations more than once."); + Assert.AreEqual(2, orchestration.ReleasedTaskCount, "Repeated release should not resume continuations more than once."); } [TestMethod] @@ -174,24 +173,23 @@ public void ExtendedSession_OpenTasksSurviveBetweenEpisodes() var orchestration = new FanOutOrchestration(fanOut: 1); OrchestrationRuntimeState runtimeState = CreateRuntimeState(); - using (var executor = new TaskOrchestrationExecutor(runtimeState, orchestration, BehaviorOnContinueAsNew.Carryover)) - { - OrchestratorExecutionResult firstEpisode = executor.Execute(); - Assert.AreEqual(1, firstEpisode.Actions.Count(), "The first episode should schedule the activity."); - Assert.IsFalse(executor.IsCompleted); + var executor = new TaskOrchestrationExecutor(runtimeState, orchestration, BehaviorOnContinueAsNew.Carryover); - runtimeState.NewEvents.Clear(); - runtimeState.AddEvent(new OrchestratorStartedEvent(-1)); - runtimeState.AddEvent(new TaskCompletedEvent(-1, taskScheduledId: 0, result: JsonDataConverter.Default.Serialize("Hello"))); + OrchestratorExecutionResult firstEpisode = executor.Execute(); + Assert.AreEqual(1, firstEpisode.Actions.Count(), "The first episode should schedule the activity."); + Assert.IsFalse(executor.IsCompleted); - OrchestratorExecutionResult secondEpisode = executor.ExecuteNewEvents(); + runtimeState.NewEvents.Clear(); + runtimeState.AddEvent(new OrchestratorStartedEvent(-1)); + runtimeState.AddEvent(new TaskCompletedEvent(-1, taskScheduledId: 0, result: JsonDataConverter.Default.Serialize("Hello"))); - Assert.IsTrue(executor.IsCompleted, "The activity result should have been delivered to the still-open task."); - Assert.AreEqual(1, orchestration.ReleasedTaskCount, "The await should have been resumed by its result, not by cancellation."); - Assert.IsTrue( - secondEpisode.Actions.OfType().Any(), - "The orchestration should have completed on the second episode."); - } + OrchestratorExecutionResult secondEpisode = executor.ExecuteNewEvents(); + + Assert.IsTrue(executor.IsCompleted, "The activity result should have been delivered to the still-open task."); + Assert.AreEqual(1, orchestration.ReleasedTaskCount, "The await should have been resumed by its result, not by cancellation."); + Assert.IsTrue( + secondEpisode.Actions.OfType().Any(), + "The orchestration should have completed on the second episode."); } static TaskOrchestrationExecutor CreateExecutor(TaskOrchestration orchestration) => diff --git a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs index 95474a122..a15a4ce96 100644 --- a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs +++ b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs @@ -325,7 +325,7 @@ static void ReleaseCursor(ref OrchestrationExecutionCursor? cursor) { OrchestrationExecutionCursor? retiredCursor = cursor; cursor = null; - retiredCursor?.OrchestrationExecutor?.Dispose(); + retiredCursor?.OrchestrationExecutor?.Release(); } /// diff --git a/src/DurableTask.Core/TaskOrchestrationExecutor.cs b/src/DurableTask.Core/TaskOrchestrationExecutor.cs index 0f4f6545a..97b6a9d94 100644 --- a/src/DurableTask.Core/TaskOrchestrationExecutor.cs +++ b/src/DurableTask.Core/TaskOrchestrationExecutor.cs @@ -28,7 +28,7 @@ namespace DurableTask.Core /// /// Utility for executing task orchestrators. /// - public class TaskOrchestrationExecutor : IDisposable + public class TaskOrchestrationExecutor { readonly TaskOrchestrationContext context; readonly TaskScheduler decisionScheduler; @@ -151,8 +151,14 @@ public OrchestratorExecutionResult ExecuteNewEvents() /// completes, so abandoned orchestrator awaits permanently root the orchestration object graph. /// See https://github.com/Azure/azure-functions-durable-extension/issues/340. /// + /// + /// This is intentionally internal rather than an implementation. + /// Executor lifetime is owned by , which is the only + /// component that knows when an executor will never run again, so exposing this externally would + /// add public surface that no caller outside this assembly can use correctly. + /// /// - public void Dispose() + internal void Release() { SynchronizationContext prevCtx = SynchronizationContext.Current; bool prevIsOrchestratorThread = OrchestrationContext.IsOrchestratorThread;