Mirrors/PowerToys
1
0
Fork 0
mirror of https://github.com/microsoft/PowerToys.git synced 2026-08-29 10:09:43 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fix CmdPal startup and shutdown lifecycle races

Browse Source 
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Copilot-Session: 7a42451c-0ec4-4778-86fd-c77d7ac9d122
This commit is contained in:
Michael Jolley
2026-08-27 19:21:14 -05:00
parent d9e47478d1
commit 88238db78b
10 changed files with 729 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/modules/cmdpal/Microsoft.CmdPal.JsonRpc/Models/JSObservableProxyBase.cs
View File

@@ -118,12 +118,12 @@ internal abstract class JSObservableProxyBase : BaseObservable, IJSPropertyChang
_commandId = notificationId;
}
Volatile.Write(ref _data, new DataBox(data));
if (changed.Count == 0)
{
return;
}
Volatile.Write(ref _data, new DataBox(data));
OnPropertyChangesApplied(changed);
}

32
src/modules/cmdpal/Microsoft.CmdPal.UI.ViewModels/Services/ExtensionTaskCoordinator.cs
View File

@@ -16,13 +16,20 @@ internal static class ExtensionTaskCoordinator
IEnumerable<TInput> inputs,
Func<TInput, Task<TResult?>> operation,
Action<TInput, Exception> onError,
int maxConcurrency,
CancellationToken cancellationToken)
where TResult : class
{
ArgumentOutOfRangeException.ThrowIfLessThan(maxConcurrency, 1);
using var concurrencyGate = new SemaphoreSlim(maxConcurrency, maxConcurrency);
var tasks = inputs.Select(async input =>
{
var entered = false;
try
{
await concurrencyGate.WaitAsync(cancellationToken).ConfigureAwait(false);
entered = true;
cancellationToken.ThrowIfCancellationRequested();
return await operation(input).ConfigureAwait(false);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
@@ -34,12 +41,37 @@ internal static class ExtensionTaskCoordinator
onError(input, ex);
return null;
}
finally
{
if (entered)
{
concurrencyGate.Release();
}
}
});
var results = await Task.WhenAll(tasks).ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
return results.OfType<TResult>().ToArray();
}
internal static async Task<TResult> RunWithConcurrencyLimitAsync<TResult>(
SemaphoreSlim concurrencyGate,
Func<Task<TResult>> operation,
CancellationToken cancellationToken)
{
await concurrencyGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
cancellationToken.ThrowIfCancellationRequested();
return await operation().ConfigureAwait(false);
}
finally
{
concurrencyGate.Release();
}
}
internal static async Task RunBlockingConcurrentlyAsync<T>(
IReadOnlyList<T> inputs,
Action<T> operation,

24
src/modules/cmdpal/Microsoft.CmdPal.UI.ViewModels/Services/JsonRpcExtensionService.cs
View File

@@ -77,10 +77,12 @@ public sealed partial class JsonRpcExtensionService : IExtensionService, IDispos
private const int ManifestStabilityAttempts = 20;
private static readonly TimeSpan ManifestStabilityDelay = TimeSpan.FromMilliseconds(250);
private static readonly TimeSpan ExtensionTeardownTimeout = TimeSpan.FromSeconds(6);
private static readonly int MaxConcurrentExtensionStarts = Math.Max(1, Math.Min(Environment.ProcessorCount, 8));
private static readonly string ExtensionsPath = GetDefaultExtensionsPath();
private readonly TaskScheduler _taskScheduler;
private readonly SemaphoreSlim _extensionStartupGate = new(MaxConcurrentExtensionStarts, MaxConcurrentExtensionStarts);
private readonly Lock _extensionsLock = new();
private readonly List<JSExtensionWrapper> _extensions = [];
private readonly List<CommandProviderWrapper> _providerWrappers = [];
@@ -218,6 +220,7 @@ public sealed partial class JsonRpcExtensionService : IExtensionService, IDispos
accepted,
item => AddExtensionGatedAsync(item.Directory, item.Manifest, ct),
(item, ex) => Logger.LogError($"Failed to load JS extension from {item.Directory}", ex),
MaxConcurrentExtensionStarts,
ct)
.ConfigureAwait(false)).ToList();
@@ -887,6 +890,7 @@ public sealed partial class JsonRpcExtensionService : IExtensionService, IDispos
candidates,
item => AddExtensionGatedAsync(item.Directory, item.Manifest, ct),
(item, ex) => Logger.LogError($"Failed to load JS extension from {item.Directory}", ex),
MaxConcurrentExtensionStarts,
ct)
.ConfigureAwait(false);
@@ -957,6 +961,26 @@ public sealed partial class JsonRpcExtensionService : IExtensionService, IDispos
return null;
}
try
{
return await ExtensionTaskCoordinator.RunWithConcurrencyLimitAsync(
_extensionStartupGate,
() => StartInstanceWithinStartupSlotAsync(directory, manifest, ct),
ct).ConfigureAwait(false);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
return null;
}
}
private async Task<StartedInstance?> StartInstanceWithinStartupSlotAsync(string directory, JSExtensionManifest manifest, CancellationToken ct)
{
if (IsStopping(ct))
{
return null;
}
JSExtensionWrapper? extensionWrapper = null;
try
{

40
src/modules/cmdpal/Microsoft.CmdPal.UI/App.xaml.cs
View File

@@ -51,6 +51,9 @@ namespace Microsoft.CmdPal.UI;
public partial class App : Application, IDisposable
{
private readonly GlobalErrorHandler _globalErrorHandler = new();
private readonly ProcessShutdownCoordinator _processShutdownCoordinator = new(
TimeSpan.FromSeconds(12),
static ex => Logger.LogError("Failed while shutting down Command Palette", ex));
/// <summary>
/// Gets the current <see cref="App"/> instance in use.
@@ -93,15 +96,12 @@ public partial class App : Application, IDisposable
NativeEventWaiter.WaitForEventLoop(
"Local\\PowerToysCmdPal-ExitEvent-eb73f6be-3f22-4b36-aee3-62924ba40bfd", () =>
{
EtwTrace?.Dispose();
if (AppWindow is not null)
var window = AppWindow;
RequestProcessExit(() =>
{
AppWindow.Close();
}
else
{
Environment.Exit(0);
}
window?.Close();
EtwTrace?.Dispose();
});
});
// Connect the PT logging to the core project's logging.
@@ -113,6 +113,30 @@ public partial class App : Application, IDisposable
appInfoService.SetLogDirectory(() => Logger.CurrentVersionLogDirectoryPath);
}
internal void RequestProcessExit(System.Action? closeWindow = null)
{
_processShutdownCoordinator.RequestExit(
closeWindow,
GetExtensionShutdownOperations,
static () => Environment.Exit(0));
}
private IEnumerable<Func<Task>> GetExtensionShutdownOperations()
{
return Services.GetServices<IExtensionService>()
.Select<IExtensionService, Func<Task>>(extensionService => async () =>
{
try
{
await extensionService.SignalStopAsync().ConfigureAwait(false);
}
catch (Exception ex)
{
Logger.LogError($"Failed to stop extension service {extensionService.GetType().Name}", ex);
}
});
}
/// <summary>
/// Invoked when the application is launched.
/// </summary>

49
src/modules/cmdpal/Microsoft.CmdPal.UI/MainWindow.xaml.cs
View File

@@ -5,7 +5,6 @@
using System.Collections.Immutable;
using System.Diagnostics;
using System.Runtime.InteropServices;
using System.Threading;
using CmdPalKeyboardService;
using CommunityToolkit.Mvvm.Messaging;
using ManagedCommon;
@@ -62,8 +61,6 @@ public sealed partial class MainWindow : WindowEx,
IDisposable,
IHostWindow
{
private static readonly TimeSpan ExtensionServicesShutdownTimeout = TimeSpan.FromSeconds(12);
[System.Diagnostics.CodeAnalysis.SuppressMessage("StyleCop.CSharp.NamingRules", "SA1310:Field names should not contain underscore", Justification = "Stylistically, window messages are WM_")]
[System.Diagnostics.CodeAnalysis.SuppressMessage("StyleCop.CSharp.NamingRules", "SA1306:Field names should begin with lower-case letter", Justification = "Stylistically, window messages are WM_")]
private readonly uint WM_TASKBAR_RESTART;
@@ -86,7 +83,6 @@ public sealed partial class MainWindow : WindowEx,
private bool _allowBreakthroughShortcut;
private bool _suppressDpiChange;
private bool _themeServiceInitialized;
private int _shutdownStarted;
// The snapshot of settings last consumed by HotReloadSettings. Used to skip redundant
// hot-reloads when a SettingsChanged notification touches settings this window doesn't
@@ -1170,8 +1166,6 @@ public sealed partial class MainWindow : WindowEx,
}
}
var extensionServices = serviceProvider.GetServices<IExtensionService>().ToArray();
App.Current.Services.GetService<TrayIconService>()!.Destroy();
// WinUI bug is causing a crash on shutdown when FailFastOnErrors is set to true (#51773592).
@@ -1181,48 +1175,7 @@ public sealed partial class MainWindow : WindowEx,
DisposeAcrylic();
_keyboardListener.Stop();
if (Interlocked.Exchange(ref _shutdownStarted, 1) == 0)
{
var shutdownThread = new Thread(() => StopExtensionServicesAndExit(extensionServices))
{
IsBackground = false,
Name = "CmdPal extension shutdown",
};
shutdownThread.Start();
}
}
private static void StopExtensionServicesAndExit(IReadOnlyList<IExtensionService> extensionServices)
{
try
{
Task.WhenAll(extensionServices.Select(ObserveExtensionStopAsync))
.WaitAsync(ExtensionServicesShutdownTimeout)
.GetAwaiter()
.GetResult();
}
catch (TimeoutException ex)
{
Logger.LogError(
$"Timed out waiting for extension services to stop after {ExtensionServicesShutdownTimeout.TotalSeconds} seconds",
ex);
}
finally
{
Environment.Exit(0);
}
}
private static async Task ObserveExtensionStopAsync(IExtensionService extensionService)
{
try
{
await extensionService.SignalStopAsync().ConfigureAwait(false);
}
catch (Exception ex)
{
Logger.LogError($"Failed to stop extension service {extensionService.GetType().Name}", ex);
}
App.Current.RequestProcessExit();
}
private void DisposeAcrylic()

113
src/modules/cmdpal/Microsoft.CmdPal.UI/Services/ProcessShutdownCoordinator.cs Normal file
View File

@@ -0,0 +1,113 @@
// Copyright (c) Microsoft Corporation
// The Microsoft Corporation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
namespace Microsoft.CmdPal.UI.Services;
internal sealed class ProcessShutdownCoordinator
{
private readonly TimeSpan _timeout;
private readonly Action<Action> _startWorker;
private readonly Action<Exception> _onError;
private int _shutdownStarted;
private int _exitStarted;
internal ProcessShutdownCoordinator(TimeSpan timeout, Action<Exception> onError)
: this(timeout, StartForegroundWorker, onError)
{
}
internal ProcessShutdownCoordinator(
TimeSpan timeout,
Action<Action> startWorker,
Action<Exception> onError)
{
ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(timeout, TimeSpan.Zero);
_timeout = timeout;
_startWorker = startWorker;
_onError = onError;
}
internal bool RequestExit(
Action? closeWindow,
Func<IEnumerable<Func<Task>>> getShutdownOperations,
Action exitProcess)
{
if (Interlocked.Exchange(ref _shutdownStarted, 1) != 0)
{
return false;
}
try
{
closeWindow?.Invoke();
}
catch (Exception ex)
{
_onError(ex);
}
try
{
_startWorker(() => StopAndExit(getShutdownOperations, exitProcess));
}
catch (Exception ex)
{
_onError(ex);
ExitOnce(exitProcess);
}
return true;
}
private static void StartForegroundWorker(Action action)
{
var shutdownThread = new Thread(() => action())
{
IsBackground = false,
Name = "CmdPal extension shutdown",
};
shutdownThread.Start();
}
private void StopAndExit(
Func<IEnumerable<Func<Task>>> getShutdownOperations,
Action exitProcess)
{
try
{
RunShutdownOperationsAsync(getShutdownOperations)
.WaitAsync(_timeout)
.GetAwaiter()
.GetResult();
}
catch (Exception ex)
{
_onError(ex);
}
finally
{
ExitOnce(exitProcess);
}
}
private static async Task RunOperationAsync(Func<Task> operation)
{
await Task.Yield();
await operation().ConfigureAwait(false);
}
private static async Task RunShutdownOperationsAsync(Func<IEnumerable<Func<Task>>> getShutdownOperations)
{
await Task.Yield();
await Task.WhenAll(getShutdownOperations().Select(RunOperationAsync)).ConfigureAwait(false);
}
private void ExitOnce(Action exitProcess)
{
if (Interlocked.Exchange(ref _exitStarted, 1) == 0)
{
exitProcess();
}
}
}

107
src/modules/cmdpal/Tests/Microsoft.CmdPal.JsonRpc.UnitTests/JSAdapterLifecycleTests.cs
View File

@@ -18,6 +18,8 @@ namespace Microsoft.CmdPal.JsonRpc.UnitTests;
public partial class JSAdapterTests
{
private static readonly string[] SubtitleProperty = ["Subtitle"];
[TestMethod]
public void PropertyChangeRegistry_PrunesDeadTargetsAndDeduplicatesLiveTargets()
{
@@ -487,6 +489,91 @@ public partial class JSAdapterTests
Assert.AreNotSame(originalMoreCommands, item.MoreCommands);
}
[TestMethod]
public void ReplaceData_IdAndUnknownFieldChangesUpdateBackingDataWithoutNotifications()
{
using var fake = new JSFakeExtension();
using var proxy = new RecordingObservableProxy(
fake.Connection,
ParseElement(new JsonObject
{
["id"] = "before",
["title"] = "Title",
["subtitle"] = "Subtitle",
["unknown"] = "before",
}));
var changedProperties = new List<string>();
proxy.PropChanged += (_, args) => changedProperties.Add(args.PropertyName);
proxy.Update(
ParseElement(new JsonObject
{
["id"] = "after",
["title"] = "Title",
["subtitle"] = "Subtitle",
["unknown"] = "after",
}));
Assert.AreEqual("after", proxy.CurrentData.GetProperty("id").GetString());
Assert.AreEqual("after", proxy.CurrentData.GetProperty("unknown").GetString());
Assert.IsEmpty(changedProperties);
}
[TestMethod]
public void ReplaceData_UnchangedVisiblePropertiesStillStoresLatestPayload()
{
using var fake = new JSFakeExtension();
using var proxy = new RecordingObservableProxy(
fake.Connection,
ParseElement(new JsonObject
{
["title"] = "Title",
["subtitle"] = "Subtitle",
["metadata"] = new JsonObject { ["version"] = 1 },
}));
var changedProperties = new List<string>();
proxy.PropChanged += (_, args) => changedProperties.Add(args.PropertyName);
proxy.Update(
ParseElement(new JsonObject
{
["title"] = "Title",
["subtitle"] = "Subtitle",
["metadata"] = new JsonObject { ["version"] = 2 },
}));
Assert.AreEqual(2, proxy.CurrentData.GetProperty("metadata").GetProperty("version").GetInt32());
Assert.IsEmpty(changedProperties);
}
[TestMethod]
public void ReplaceData_SubtitleChangeUpdatesBackingDataAndRaisesOnlySubtitle()
{
using var fake = new JSFakeExtension();
using var proxy = new RecordingObservableProxy(
fake.Connection,
ParseElement(new JsonObject
{
["title"] = "Title",
["subtitle"] = "Before",
["unknown"] = "before",
}));
var changedProperties = new List<string>();
proxy.PropChanged += (_, args) => changedProperties.Add(args.PropertyName);
proxy.Update(
ParseElement(new JsonObject
{
["title"] = "Title",
["subtitle"] = "After",
["unknown"] = "after",
}));
Assert.AreEqual("After", proxy.CurrentData.GetProperty("subtitle").GetString());
Assert.AreEqual("after", proxy.CurrentData.GetProperty("unknown").GetString());
CollectionAssert.AreEqual(SubtitleProperty, changedProperties);
}
[TestMethod]
public void ProviderSettings_ConcurrentReadsReturnOneProxy()
{
@@ -782,4 +869,24 @@ public partial class JSAdapterTests
ApplyCount++;
}
}
private sealed class RecordingObservableProxy : JSObservableProxyBase
{
internal RecordingObservableProxy(JsonRpcConnection connection, JsonElement data)
: base("recording", connection, data)
{
}
internal JsonElement CurrentData => Data;
internal void Update(JsonElement data)
{
ReplaceData(data, ["title", "subtitle"]);
}
protected override bool SupportsProperty(string propertyName)
{
return propertyName is "title" or "subtitle";
}
}
}

1
src/modules/cmdpal/Tests/Microsoft.CmdPal.UI.UnitTests/Microsoft.CmdPal.UI.UnitTests.csproj
View File

@@ -29,5 +29,6 @@
<Compile Include="..\..\Microsoft.CmdPal.UI\Helpers\Icons\IconSourceProvider.cs" Link="Helpers\Icons\IconSourceProvider.cs" />
<Compile Include="..\..\Microsoft.CmdPal.UI\Helpers\Icons\IIconLoaderService.cs" Link="Helpers\Icons\IIconLoaderService.cs" />
<Compile Include="..\..\Microsoft.CmdPal.UI\Helpers\Icons\IIconSourceProvider.cs" Link="Helpers\Icons\IIconSourceProvider.cs" />
<Compile Include="..\..\Microsoft.CmdPal.UI\Services\ProcessShutdownCoordinator.cs" Link="Services\ProcessShutdownCoordinator.cs" />
</ItemGroup>
</Project>

218
src/modules/cmdpal/Tests/Microsoft.CmdPal.UI.UnitTests/ProcessShutdownCoordinatorTests.cs Normal file
View File

@@ -0,0 +1,218 @@
// Copyright (c) Microsoft Corporation
// The Microsoft Corporation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.
using System.Diagnostics;
using Microsoft.CmdPal.UI.Services;
using Microsoft.VisualStudio.TestTools.UnitTesting;
namespace Microsoft.CmdPal.UI.UnitTests;
[TestClass]
public sealed class ProcessShutdownCoordinatorTests
{
[TestMethod]
public void RequestExit_WithLiveWindow_ClosesWindowAndExitsOnce()
{
var coordinator = CreateCoordinator();
var closeCount = 0;
var stopCount = 0;
var exitCount = 0;
var started = coordinator.RequestExit(
() =>
{
closeCount++;
coordinator.RequestExit(null, () => [], () => exitCount++);
},
() => GetShutdownOperations(() => stopCount++),
() => exitCount++);
Assert.IsTrue(started);
Assert.AreEqual(1, closeCount);
Assert.AreEqual(1, stopCount);
Assert.AreEqual(1, exitCount);
}
[TestMethod]
public void RequestExit_WithoutWindow_StillStopsAndExits()
{
var coordinator = CreateCoordinator();
var stopCount = 0;
var exitCount = 0;
coordinator.RequestExit(
null,
() => GetShutdownOperations(() => stopCount++),
() => exitCount++);
Assert.AreEqual(1, stopCount);
Assert.AreEqual(1, exitCount);
}
[TestMethod]
public void RequestExit_WhenWindowCloseFails_StillStopsAndExits()
{
var errors = new List<Exception>();
var coordinator = CreateCoordinator(errors.Add);
var stopCount = 0;
var exitCount = 0;
coordinator.RequestExit(
() => throw new InvalidOperationException("stale window"),
() => GetShutdownOperations(() => stopCount++),
() => exitCount++);
Assert.AreEqual(1, stopCount);
Assert.AreEqual(1, exitCount);
Assert.HasCount(1, errors);
Assert.IsInstanceOfType<InvalidOperationException>(errors[0]);
}
[TestMethod]
public void RequestExit_WhenCloseDoesNotRaiseEvent_StillExitsOnce()
{
var coordinator = CreateCoordinator();
var closeCount = 0;
var exitCount = 0;
Assert.IsTrue(coordinator.RequestExit(
() => closeCount++,
() => Array.Empty<Func<Task>>(),
() => exitCount++));
Assert.IsFalse(coordinator.RequestExit(
null,
() => Array.Empty<Func<Task>>(),
() => exitCount++));
Assert.AreEqual(1, closeCount);
Assert.AreEqual(1, exitCount);
}
[TestMethod]
public void RequestExit_WhenShutdownTimesOut_StillExits()
{
var errors = new List<Exception>();
var coordinator = new ProcessShutdownCoordinator(
TimeSpan.FromMilliseconds(20),
action => action(),
errors.Add);
var exitCount = 0;
coordinator.RequestExit(
null,
() => new List<Func<Task>> { () => new TaskCompletionSource().Task },
() => exitCount++);
Assert.AreEqual(1, exitCount);
Assert.HasCount(1, errors);
Assert.IsInstanceOfType<TimeoutException>(errors[0]);
}
[TestMethod]
public void RequestExit_SynchronousStopWorkCannotBypassAggregateTimeout()
{
using var release = new ManualResetEventSlim();
var errors = new List<Exception>();
var coordinator = new ProcessShutdownCoordinator(
TimeSpan.FromMilliseconds(50),
action => action(),
errors.Add);
var exitCount = 0;
var stopwatch = Stopwatch.StartNew();
try
{
coordinator.RequestExit(
null,
() => GetShutdownOperations(() => release.Wait(TimeSpan.FromSeconds(5))),
() => exitCount++);
}
finally
{
release.Set();
}
Assert.IsTrue(stopwatch.Elapsed < TimeSpan.FromSeconds(1));
Assert.AreEqual(1, exitCount);
Assert.HasCount(1, errors);
Assert.IsInstanceOfType<TimeoutException>(errors[0]);
}
[TestMethod]
public void RequestExit_SynchronousOperationDiscoveryCannotBypassAggregateTimeout()
{
using var release = new ManualResetEventSlim();
var errors = new List<Exception>();
var coordinator = new ProcessShutdownCoordinator(
TimeSpan.FromMilliseconds(50),
action => action(),
errors.Add);
var exitCount = 0;
var stopwatch = Stopwatch.StartNew();
try
{
coordinator.RequestExit(
null,
() =>
{
release.Wait(TimeSpan.FromSeconds(5));
return Array.Empty<Func<Task>>();
},
() => exitCount++);
}
finally
{
release.Set();
}
Assert.IsTrue(stopwatch.Elapsed < TimeSpan.FromSeconds(1));
Assert.AreEqual(1, exitCount);
Assert.HasCount(1, errors);
Assert.IsInstanceOfType<TimeoutException>(errors[0]);
}
[TestMethod]
public void RequestExit_WhenWorkerCannotStart_StillExitsExactlyOnce()
{
var errors = new List<Exception>();
var coordinator = new ProcessShutdownCoordinator(
TimeSpan.FromSeconds(1),
action =>
{
action();
throw new InvalidOperationException("worker failed after running");
},
errors.Add);
var exitCount = 0;
coordinator.RequestExit(
null,
() => Array.Empty<Func<Task>>(),
() => exitCount++);
Assert.AreEqual(1, exitCount);
Assert.HasCount(1, errors);
Assert.IsInstanceOfType<InvalidOperationException>(errors[0]);
}
private static ProcessShutdownCoordinator CreateCoordinator(Action<Exception>? onError = null)
{
return new ProcessShutdownCoordinator(
TimeSpan.FromSeconds(1),
action => action(),
onError ?? (_ => Assert.Fail("Shutdown should not fail")));
}
private static Task StopAsync(Action stop)
{
stop();
return Task.CompletedTask;
}
private static IReadOnlyList<Func<Task>> GetShutdownOperations(Action stop)
{
return new List<Func<Task>> { () => StopAsync(stop) };
}
}

200
src/modules/cmdpal/Tests/Microsoft.CmdPal.UI.ViewModels.UnitTests/ExtensionTaskCoordinatorTests.cs
View File

@@ -18,6 +18,9 @@ namespace Microsoft.CmdPal.UI.ViewModels.UnitTests;
public sealed class ExtensionTaskCoordinatorTests
{
private static readonly int[] AllInputs = [1, 2, 3];
private static readonly int[] FiveInputs = [1, 2, 3, 4, 5];
private static readonly int[] FirstInput = [1];
private static readonly int[] RetryInput = [2];
[TestMethod]
public async Task RunConcurrentlyAsync_PreservesOrderAndIsolatesFailures()
@@ -47,6 +50,7 @@ public sealed class ExtensionTaskCoordinatorTests
AllInputs,
LoadAsync,
(_, exception) => errors.Enqueue(exception),
3,
CancellationToken.None);
await allStarted.Task.WaitAsync(TimeSpan.FromSeconds(1));
@@ -60,6 +64,202 @@ public sealed class ExtensionTaskCoordinatorTests
Assert.IsInstanceOfType<InvalidOperationException>(errors.Single());
}
[TestMethod]
public async Task RunConcurrentlyAsync_RespectsLimitAndStartsQueuedWork()
{
var firstWaveStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var activeLock = new object();
var started = 0;
var active = 0;
var maxActive = 0;
async Task<string?> LoadAsync(int value)
{
var currentActive = Interlocked.Increment(ref active);
lock (activeLock)
{
maxActive = Math.Max(maxActive, currentActive);
}
if (Interlocked.Increment(ref started) == 2)
{
firstWaveStarted.SetResult();
}
await release.Task;
Interlocked.Decrement(ref active);
return value.ToString(CultureInfo.InvariantCulture);
}
var loadTask = ExtensionTaskCoordinator.RunConcurrentlyAsync(
FiveInputs,
LoadAsync,
(_, _) => Assert.Fail(),
2,
CancellationToken.None);
await firstWaveStarted.Task.WaitAsync(TimeSpan.FromSeconds(1));
Assert.AreEqual(2, Volatile.Read(ref started));
Assert.AreEqual(2, Volatile.Read(ref maxActive));
release.SetResult();
var results = await loadTask.WaitAsync(TimeSpan.FromSeconds(1));
Assert.HasCount(5, results);
Assert.AreEqual(5, started);
Assert.AreEqual(2, maxActive);
}
[TestMethod]
public async Task RunConcurrentlyAsync_CancellationDoesNotStartQueuedWorkAndAllowsRetry()
{
using var cancellationTokenSource = new CancellationTokenSource();
var firstStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var releaseFirst = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var started = new ConcurrentQueue<int>();
async Task<string?> LoadAsync(int value)
{
started.Enqueue(value);
firstStarted.TrySetResult();
await releaseFirst.Task;
return value.ToString(CultureInfo.InvariantCulture);
}
var loadTask = ExtensionTaskCoordinator.RunConcurrentlyAsync(
AllInputs,
LoadAsync,
(_, _) => Assert.Fail(),
1,
cancellationTokenSource.Token);
await firstStarted.Task.WaitAsync(TimeSpan.FromSeconds(1));
cancellationTokenSource.Cancel();
releaseFirst.SetResult();
await Assert.ThrowsExactlyAsync<OperationCanceledException>(
() => loadTask);
CollectionAssert.AreEqual(FirstInput, started.ToArray());
var retry = await ExtensionTaskCoordinator.RunConcurrentlyAsync(
RetryInput,
value => Task.FromResult<string?>(value.ToString(CultureInfo.InvariantCulture)),
(_, _) => Assert.Fail(),
1,
CancellationToken.None);
Assert.AreEqual("2", retry.Single());
}
[TestMethod]
public async Task RunConcurrentlyAsync_FailureReleasesPermit()
{
var errors = new ConcurrentQueue<Exception>();
var results = await ExtensionTaskCoordinator.RunConcurrentlyAsync(
AllInputs,
value => value == 1
? Task.FromException<string?>(new InvalidOperationException("failed"))
: Task.FromResult<string?>(value.ToString(CultureInfo.InvariantCulture)),
(_, exception) => errors.Enqueue(exception),
1,
CancellationToken.None);
Assert.HasCount(2, results);
Assert.AreEqual("2", results[0]);
Assert.AreEqual("3", results[1]);
Assert.HasCount(1, errors);
}
[TestMethod]
public async Task RunWithConcurrencyLimitAsync_SharesLimitAcrossIndependentCallers()
{
using var concurrencyGate = new SemaphoreSlim(2, 2);
var firstWaveStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var started = 0;
var active = 0;
var maxActive = 0;
var activeLock = new object();
async Task<int> StartAsync(int value)
{
var currentActive = Interlocked.Increment(ref active);
lock (activeLock)
{
maxActive = Math.Max(maxActive, currentActive);
}
if (Interlocked.Increment(ref started) == 2)
{
firstWaveStarted.SetResult();
}
await release.Task;
Interlocked.Decrement(ref active);
return value;
}
var tasks = FiveInputs.Select(value =>
ExtensionTaskCoordinator.RunWithConcurrencyLimitAsync(
concurrencyGate,
() => StartAsync(value),
CancellationToken.None)).ToArray();
await firstWaveStarted.Task.WaitAsync(TimeSpan.FromSeconds(1));
Assert.AreEqual(2, started);
release.SetResult();
await Task.WhenAll(tasks).WaitAsync(TimeSpan.FromSeconds(1));
Assert.AreEqual(5, started);
Assert.AreEqual(2, maxActive);
}
[TestMethod]
public async Task RunWithConcurrencyLimitAsync_CanceledWaiterDoesNotRunAndPermitCanBeReused()
{
using var concurrencyGate = new SemaphoreSlim(1, 1);
using var cancellationTokenSource = new CancellationTokenSource();
await concurrencyGate.WaitAsync();
var operationStarted = false;
var canceled = ExtensionTaskCoordinator.RunWithConcurrencyLimitAsync(
concurrencyGate,
() =>
{
operationStarted = true;
return Task.FromResult(1);
},
cancellationTokenSource.Token);
cancellationTokenSource.Cancel();
await Assert.ThrowsExactlyAsync<OperationCanceledException>(() => canceled);
Assert.IsFalse(operationStarted);
concurrencyGate.Release();
var result = await ExtensionTaskCoordinator.RunWithConcurrencyLimitAsync(
concurrencyGate,
() => Task.FromResult(2),
CancellationToken.None);
Assert.AreEqual(2, result);
}
[TestMethod]
public async Task RunWithConcurrencyLimitAsync_FailureReleasesPermit()
{
using var concurrencyGate = new SemaphoreSlim(1, 1);
await Assert.ThrowsExactlyAsync<InvalidOperationException>(
() => ExtensionTaskCoordinator.RunWithConcurrencyLimitAsync<int>(
concurrencyGate,
() => throw new InvalidOperationException("failed"),
CancellationToken.None));
var result = await ExtensionTaskCoordinator.RunWithConcurrencyLimitAsync(
concurrencyGate,
() => Task.FromResult(2),
CancellationToken.None);
Assert.AreEqual(2, result);
}
[TestMethod]
public async Task RunBlockingConcurrentlyAsync_UsesOneAggregateTimeout()
{