diff --git a/Directory.Packages.props b/Directory.Packages.props index 47d6595c7cd..3bcd8b9892e 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -12,7 +12,6 @@ - diff --git a/src/TUnit.Engine/Building/TestBuilderPipeline.cs b/src/TUnit.Engine/Building/TestBuilderPipeline.cs index 12d27b4ae65..ad005d8872b 100644 --- a/src/TUnit.Engine/Building/TestBuilderPipeline.cs +++ b/src/TUnit.Engine/Building/TestBuilderPipeline.cs @@ -1,6 +1,4 @@ -using System.Collections.Concurrent; using System.Diagnostics.CodeAnalysis; -using EnumerableAsyncProcessor.Extensions; using Microsoft.Testing.Platform.Requests; using TUnit.Core; using TUnit.Core.Helpers; @@ -13,19 +11,6 @@ namespace TUnit.Engine.Building; -/// -/// Threshold below which sequential processing is used instead of parallel. -/// For small test sets, the overhead of task scheduling exceeds parallelization benefits. -/// -internal static class ParallelThresholds -{ - /// - /// Minimum number of items before parallel processing is used. - /// Below this threshold, sequential processing avoids task scheduling overhead. - /// - public const int MinItemsForParallel = 8; -} - internal sealed class TestBuilderPipeline { private readonly ITestDataCollector _dataCollector; @@ -67,17 +52,6 @@ private TestBuilderContext CreateTestBuilderContext(TestMetadata metadata) return testBuilderContext; } - [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Reflection mode is not used in AOT/trimmed scenarios")] - [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Reflection mode is not used in AOT scenarios")] - public async Task> BuildTestsAsync(string testSessionId) - { - var collectedMetadata = await _dataCollector.CollectTestsAsync(testSessionId).ConfigureAwait(false); - - // For this method (non-streaming), we're not in execution mode so no filter optimization - var buildingContext = new TestBuildingContext(IsForExecution: false, Filter: null); - return await BuildTestsFromMetadataAsync(collectedMetadata, buildingContext).ConfigureAwait(false); - } - /// /// Collects all test metadata without building tests. /// This is a lightweight operation used for dependency analysis. @@ -97,22 +71,18 @@ public async Task> CollectTestMetadataAsync(string tes } /// - /// Streaming version that yields tests as they're built without buffering + /// Collects metadata for the session and builds all tests, optionally pre-filtering metadata. /// /// The test session identifier /// Context for test building /// Optional predicate to filter which metadata should be built (null means build all) /// Cancellation token - [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Reflection mode is not used in AOT/trimmed scenarios")] - [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Reflection mode is not used in AOT scenarios")] - public async Task> BuildTestsStreamingAsync( + public async Task> BuildTestsAsync( string testSessionId, TestBuildingContext buildingContext, Func? metadataFilter = null, CancellationToken cancellationToken = default) { - // Get metadata streaming if supported - // Fall back to non-streaming collection var collectedMetadata = await _dataCollector.CollectTestsAsync(testSessionId).ConfigureAwait(false); // Apply metadata filter if provided (for dependency-aware filtering optimization) @@ -121,9 +91,7 @@ public async Task> BuildTestsStreamingAsync( collectedMetadata = collectedMetadata.Where(metadataFilter); } - return await collectedMetadata - .SelectManyAsync(metadata => BuildTestsFromSingleMetadataAsync(metadata, buildingContext, cancellationToken), cancellationToken: cancellationToken) - .ProcessInParallel(cancellationToken: cancellationToken); + return await BuildTestsFromMetadataAsync(collectedMetadata, buildingContext, cancellationToken).ConfigureAwait(false); } /// @@ -137,62 +105,55 @@ public async Task> BuildTestsFromMetadataAsy TestBuildingContext buildingContext, CancellationToken cancellationToken = default) { - // Materialize to check count - for small sets, sequential processing is faster - var metadataList = testMetadata as IList ?? testMetadata.ToList(); + var metadataList = testMetadata as IReadOnlyList ?? testMetadata.ToList(); + + var testGroups = await ParallelMap.SelectParallelAsync( + metadataList, + metadata => BuildTestsForMetadataAsync(metadata, buildingContext, cancellationToken), + Environment.ProcessorCount, + cancellationToken).ConfigureAwait(false); - IEnumerable> testGroups; + var totalCount = 0; + foreach (var group in testGroups) + { + totalCount += group.Count; + } - if (metadataList.Count < ParallelThresholds.MinItemsForParallel) + var allTests = new List(totalCount); + foreach (var group in testGroups) { - // Sequential processing for small sets - avoids task scheduling overhead - var results = new List>(metadataList.Count); - foreach (var metadata in metadataList) + allTests.AddRange(group); + } + + return allTests; + } + + /// + /// Builds all tests for a single metadata item. Exceptions surface as a single failed + /// placeholder test rather than propagating, so one bad data source cannot abort discovery. + /// + [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Reflection mode is not used in AOT/trimmed scenarios")] + private async Task> BuildTestsForMetadataAsync( + TestMetadata metadata, + TestBuildingContext buildingContext, + CancellationToken cancellationToken) + { + try + { + // Dynamic test metadata bypasses normal test building + if (metadata is IDynamicTestMetadata) { - cancellationToken.ThrowIfCancellationRequested(); - try - { - if (metadata is IDynamicTestMetadata) - { - results.Add(await GenerateDynamicTests(metadata).ConfigureAwait(false)); - } - else - { - results.Add(await _testBuilder.BuildTestsFromMetadataAsync(metadata, buildingContext, cancellationToken).ConfigureAwait(false)); - } - } - catch (Exception ex) - { - var failedTest = CreateFailedTestForDataGenerationError(metadata, ex); - results.Add([failedTest]); - } + return await GenerateDynamicTests(metadata).ConfigureAwait(false); } - testGroups = results; + + var tests = await _testBuilder.BuildTestsFromMetadataAsync(metadata, buildingContext, cancellationToken).ConfigureAwait(false); + + return tests as IReadOnlyList ?? tests.ToList(); } - else + catch (Exception ex) { - // Parallel processing for larger sets - testGroups = await metadataList.SelectAsync(async metadata => - { - try - { - // Check if this is a dynamic test metadata that should bypass normal test building - if (metadata is IDynamicTestMetadata) - { - return await GenerateDynamicTests(metadata).ConfigureAwait(false); - } - - return await _testBuilder.BuildTestsFromMetadataAsync(metadata, buildingContext, cancellationToken).ConfigureAwait(false); - } - catch (Exception ex) - { - var failedTest = CreateFailedTestForDataGenerationError(metadata, ex); - return (IEnumerable)[failedTest]; - } - }, cancellationToken: cancellationToken) - .ProcessInParallel(Environment.ProcessorCount); + return [CreateFailedTestForDataGenerationError(metadata, ex)]; } - - return testGroups.SelectMany(x => x); } private async Task GenerateDynamicTests(TestMetadata metadata) @@ -208,8 +169,9 @@ private async Task GenerateDynamicTests(TestMetadata m var attributesByType = attributes.ToAttributeDictionary(); Func> instanceFactory = () => Task.FromResult(metadata.InstanceFactory(Type.EmptyTypes, [])); - return await Enumerable.Range(0, repeatCount + 1) - .SelectAsync(async repeatIndex => + return await ParallelMap.ForParallelAsync( + repeatCount + 1, + async repeatIndex => { // Create a simple TestData for ID generation // Use DynamicTestIndex from the metadata to ensure unique test IDs for multiple dynamic tests @@ -285,160 +247,8 @@ private async Task GenerateDynamicTests(TestMetadata m }; return metadata.CreateExecutableTestFactory(executableTestContext, metadata); - }) - .ProcessInParallel(Environment.ProcessorCount); - } - - /// - /// Build tests from a single metadata item, yielding them as they're created - /// -#if NET8_0_OR_GREATER - [RequiresUnreferencedCode("Test building in reflection mode uses generic type resolution which requires unreferenced code")] -#endif - private async IAsyncEnumerable BuildTestsFromSingleMetadataAsync(TestMetadata metadata, TestBuildingContext buildingContext, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default) - { - TestMetadata resolvedMetadata; - Exception? resolutionError = null; - - try - { - resolvedMetadata = metadata; - } - catch (Exception ex) - { - resolutionError = ex; - resolvedMetadata = metadata; // Use original for error reporting - } - - if (resolutionError != null) - { - yield return CreateFailedTestForGenericResolutionError(metadata, resolutionError); - yield break; - } - - List? testsToYield = null; - Exception? buildError = null; - - try - { - // Check if this is a dynamic test metadata that should bypass normal test building - if (resolvedMetadata is IDynamicTestMetadata) - { - testsToYield = - [ - ]; - - // Use pre-extracted repeat count from metadata (avoids instantiating attributes) - var repeatCount = resolvedMetadata.RepeatCount ?? 0; - - // Get attributes for test details - var attributes = resolvedMetadata.GetOrCreateAttributes(); - - // Hoist loop-invariant state so repeat iterations share one attribute dictionary and factory delegate - var attributesByType = attributes.ToAttributeDictionary(); - Func> instanceFactory = () => Task.FromResult(resolvedMetadata.InstanceFactory(Type.EmptyTypes, [])); - - // Dynamic tests need to honor attributes like RepeatCount, RetryCount, etc. - // We'll create multiple test instances based on RepeatCount - // Use DynamicTestIndex from the metadata to ensure unique test IDs for multiple dynamic tests - var dynamicTestIndex = ((IDynamicTestMetadata)resolvedMetadata).DynamicTestIndex; - for (var repeatIndex = 0; repeatIndex < repeatCount + 1; repeatIndex++) - { - // Create a simple TestData for ID generation - var testData = new TestBuilder.TestData - { - TestClassInstanceFactory = instanceFactory, - ClassDataSourceAttributeIndex = 0, - ClassDataLoopIndex = 0, - ClassData = [], - MethodDataSourceAttributeIndex = 0, - MethodDataLoopIndex = dynamicTestIndex, - MethodData = [], - RepeatIndex = repeatIndex, - InheritanceDepth = resolvedMetadata.InheritanceDepth, - ResolvedClassGenericArguments = Type.EmptyTypes, - ResolvedMethodGenericArguments = Type.EmptyTypes - }; - - var testId = TestIdentifierService.GenerateTestId(resolvedMetadata, testData); - var dynamicMetadata = (IDynamicTestMetadata)resolvedMetadata; - var baseDisplayName = dynamicMetadata.DisplayName ?? resolvedMetadata.TestName; - var displayName = repeatCount > 0 - ? $"{baseDisplayName} (Repeat {repeatIndex + 1}/{repeatCount + 1})" - : baseDisplayName; - - // Create TestDetails for dynamic tests - var testDetails = new TestDetails(attributes) - { - TestId = testId, - TestName = resolvedMetadata.TestName, - ClassType = resolvedMetadata.TestClassType, - MethodName = resolvedMetadata.TestMethodName, - ClassInstance = PlaceholderInstance.Instance, - TestMethodArguments = [], - TestClassArguments = [], - TestFilePath = resolvedMetadata.FilePath ?? "Unknown", - TestLineNumber = resolvedMetadata.LineNumber, - TestStartColumnNumber = resolvedMetadata.StartColumnNumber, - TestEndLineNumber = resolvedMetadata.EndLineNumber, - TestEndColumnNumber = resolvedMetadata.EndColumnNumber, - ReturnType = typeof(Task), - MethodMetadata = resolvedMetadata.MethodMetadata, - AttributesByType = attributesByType - }; - - var context = _contextProvider.CreateTestContext( - resolvedMetadata.TestClassType, - CreateTestBuilderContext(resolvedMetadata), - testDetails, - CancellationToken.None); - - // Set custom display name for dynamic tests if specified - if (dynamicMetadata.DisplayName != null) - { - context.Metadata.DisplayName = dynamicMetadata.DisplayName; - } - - // Invoke discovery event receivers to properly handle all attribute behaviors - await InvokeDiscoveryEventReceiversAsync(context).ConfigureAwait(false); - - var executableTestContext = new ExecutableTestCreationContext - { - TestId = testId, - DisplayName = displayName, - Arguments = [], - ClassArguments = [], - Context = context, - TestClassInstanceFactory = testData.TestClassInstanceFactory - }; - - var executableTest = resolvedMetadata.CreateExecutableTestFactory(executableTestContext, resolvedMetadata); - testsToYield.Add(executableTest); - } - } - else - { - // Normal test metadata goes through the standard test builder - var testsFromMetadata = await _testBuilder.BuildTestsFromMetadataAsync(resolvedMetadata, buildingContext, cancellationToken).ConfigureAwait(false); - testsToYield = new List(testsFromMetadata); - } - } - catch (Exception ex) - { - buildError = ex; - } - - if (buildError != null) - { - yield return CreateFailedTestForDataGenerationError(resolvedMetadata, buildError); - } - else if (testsToYield != null) - { - foreach (var test in testsToYield) - { - yield return test; - } - } + }, + Environment.ProcessorCount).ConfigureAwait(false); } private AbstractExecutableTest CreateFailedTestForDataGenerationError(TestMetadata metadata, Exception exception) @@ -494,59 +304,6 @@ private AbstractExecutableTest CreateFailedTestForDataGenerationError(TestMetada }; } - private AbstractExecutableTest CreateFailedTestForGenericResolutionError(TestMetadata metadata, Exception exception) - { - var testId = TestIdentifierService.GenerateFailedTestId(metadata); - var displayName = $"{metadata.TestName} [GENERIC RESOLUTION ERROR]"; - - var testDetails = new TestDetails([]) - { - TestId = testId, - TestName = metadata.TestName, - ClassType = metadata.TestClassType, - MethodName = metadata.TestMethodName, - ClassInstance = null!, - TestMethodArguments = [], - TestClassArguments = [], - TestFilePath = metadata.FilePath ?? "Unknown", - TestLineNumber = metadata.LineNumber, - TestStartColumnNumber = metadata.StartColumnNumber, - TestEndLineNumber = metadata.EndLineNumber, - TestEndColumnNumber = metadata.EndColumnNumber, - ReturnType = typeof(Task), - MethodMetadata = metadata.MethodMetadata, - AttributesByType = AttributeDictionaryHelper.Empty - }; - - var context = _contextProvider.CreateTestContext( - metadata.TestClassType, - CreateTestBuilderContext(metadata), - testDetails, - CancellationToken.None); - - var now = DateTimeOffset.UtcNow; - - return new FailedExecutableTest(exception) - { - TestId = testId, - Metadata = metadata, - Arguments = [], - ClassArguments = [], - Context = context, - State = TestState.Failed, - Result = new TestResult - { - State = TestState.Failed, - Start = now, - End = now, - Duration = TimeSpan.Zero, - Exception = exception, - ComputerName = EnvironmentHelper.MachineName, - TestContext = context - } - }; - } - private async Task InvokeDiscoveryEventReceiversAsync(TestContext context) { var discoveredContext = new DiscoveredTestContext( diff --git a/src/TUnit.Engine/TUnit.Engine.csproj b/src/TUnit.Engine/TUnit.Engine.csproj index c8aee10ebd2..ad9bd89d734 100644 --- a/src/TUnit.Engine/TUnit.Engine.csproj +++ b/src/TUnit.Engine/TUnit.Engine.csproj @@ -13,7 +13,6 @@ - diff --git a/src/TUnit.Engine/TestDiscoveryService.cs b/src/TUnit.Engine/TestDiscoveryService.cs index 10f128ff331..8123099b69e 100644 --- a/src/TUnit.Engine/TestDiscoveryService.cs +++ b/src/TUnit.Engine/TestDiscoveryService.cs @@ -233,7 +233,7 @@ private async IAsyncEnumerable DiscoverTestsStreamAsync( using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); cts.CancelAfter(EngineDefaults.DiscoveryTimeout); - var tests = await _testBuilderPipeline.BuildTestsStreamingAsync(testSessionId, buildingContext, metadataFilter: null, cts.Token).ConfigureAwait(false); + var tests = await _testBuilderPipeline.BuildTestsAsync(testSessionId, buildingContext, metadataFilter: null, cts.Token).ConfigureAwait(false); foreach (var test in tests) { @@ -382,34 +382,12 @@ public async IAsyncEnumerable DiscoverTestsFullyStreamin - private async Task InvokePostResolutionEventsInParallelAsync(List allTests) + private Task InvokePostResolutionEventsInParallelAsync(List allTests) { - if (allTests.Count < Building.ParallelThresholds.MinItemsForParallel) - { - foreach (var test in allTests) - { - await _testBuilderPipeline.InvokePostResolutionEventsAsync(test).ConfigureAwait(false); - } - return; - } - -#if NET8_0_OR_GREATER - await Parallel.ForEachAsync( + return Utilities.ParallelMap.ForEachParallelAsync( allTests, - new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount }, - async (test, _) => - { - await _testBuilderPipeline.InvokePostResolutionEventsAsync(test).ConfigureAwait(false); - } - ).ConfigureAwait(false); -#else - var tasks = new Task[allTests.Count]; - for (var i = 0; i < allTests.Count; i++) - { - tasks[i] = _testBuilderPipeline.InvokePostResolutionEventsAsync(allTests[i]).AsTask(); - } - await Task.WhenAll(tasks).ConfigureAwait(false); -#endif + test => _testBuilderPipeline.InvokePostResolutionEventsAsync(test), + Environment.ProcessorCount); } public IEnumerable GetCachedTestContexts() diff --git a/src/TUnit.Engine/Utilities/ParallelMap.cs b/src/TUnit.Engine/Utilities/ParallelMap.cs new file mode 100644 index 00000000000..e53192c093e --- /dev/null +++ b/src/TUnit.Engine/Utilities/ParallelMap.cs @@ -0,0 +1,164 @@ +namespace TUnit.Engine.Utilities; + +/// +/// Allocation-light bounded-parallelism async map. +/// Results preserve source order. A fixed set of workers pulls items via an interlocked +/// cursor, so no partitioner, channel, or per-item task is allocated. +/// Sets smaller than run sequentially, where task +/// scheduling overhead would exceed the parallelization benefit. +/// +internal static class ParallelMap +{ + /// + /// Minimum number of items before parallel processing is used. + /// + public const int SequentialThreshold = 8; + + /// + /// Void counterpart of — same worker + /// model without the results array. Takes a ValueTask action so synchronously-completing + /// callees don't allocate a Task per item. + /// + public static async Task ForEachParallelAsync( + IReadOnlyList source, + Func action, + int maxDegreeOfParallelism, + CancellationToken cancellationToken = default) + { + var count = source.Count; + + if (count == 0) + { + return; + } + + var workerCount = Math.Min(maxDegreeOfParallelism, count); + + if (workerCount <= 1 || count < SequentialThreshold) + { + for (var i = 0; i < count; i++) + { + cancellationToken.ThrowIfCancellationRequested(); + await action(source[i]).ConfigureAwait(false); + } + + return; + } + + var cursor = -1; + + Func worker = WorkerAsync; + var workers = new Task[workerCount]; + for (var w = 0; w < workerCount; w++) + { + workers[w] = Task.Run(worker, CancellationToken.None); + } + + await Task.WhenAll(workers).ConfigureAwait(false); + + return; + + async Task WorkerAsync() + { + while (true) + { + var index = Interlocked.Increment(ref cursor); + + if (index >= count) + { + return; + } + + cancellationToken.ThrowIfCancellationRequested(); + + try + { + await action(source[index]).ConfigureAwait(false); + } + catch + { + // Park the cursor so sibling workers stop pulling new items; + // Task.WhenAll surfaces this fault once in-flight items finish. + Volatile.Write(ref cursor, count); + throw; + } + } + } + } + + public static Task SelectParallelAsync( + IReadOnlyList source, + Func> selector, + int maxDegreeOfParallelism, + CancellationToken cancellationToken = default) + => ForParallelAsync(source.Count, index => selector(source[index]), maxDegreeOfParallelism, cancellationToken); + + public static async Task ForParallelAsync( + int count, + Func> selector, + int maxDegreeOfParallelism, + CancellationToken cancellationToken = default) + { + if (count == 0) + { + return []; + } + + var results = new TResult[count]; + var workerCount = Math.Min(maxDegreeOfParallelism, count); + + if (workerCount <= 1 || count < SequentialThreshold) + { + for (var i = 0; i < count; i++) + { + cancellationToken.ThrowIfCancellationRequested(); + results[i] = await selector(i).ConfigureAwait(false); + } + + return results; + } + + var cursor = -1; + + // Task.Run rather than invoking WorkerAsync directly: a selector that completes + // synchronously would otherwise run the entire loop inline on this thread and + // serialize all the other workers. + Func worker = WorkerAsync; + var workers = new Task[workerCount]; + for (var w = 0; w < workerCount; w++) + { + workers[w] = Task.Run(worker, CancellationToken.None); + } + + await Task.WhenAll(workers).ConfigureAwait(false); + + return results; + + async Task WorkerAsync() + { + while (true) + { + var index = Interlocked.Increment(ref cursor); + + if (index >= count) + { + return; + } + + cancellationToken.ThrowIfCancellationRequested(); + + try + { + results[index] = await selector(index).ConfigureAwait(false); + } + catch + { + // Park the cursor so sibling workers stop pulling new items; + // Task.WhenAll surfaces this fault once in-flight items finish. + Volatile.Write(ref cursor, count); + throw; + } + } + } + } +} diff --git a/tests/TUnit.UnitTests/ParallelMapTests.cs b/tests/TUnit.UnitTests/ParallelMapTests.cs new file mode 100644 index 00000000000..f0a00bf4100 --- /dev/null +++ b/tests/TUnit.UnitTests/ParallelMapTests.cs @@ -0,0 +1,195 @@ +using TUnit.Engine.Utilities; + +namespace TUnit.UnitTests; + +/// +/// Tests for the internal bounded-parallelism map that backs test discovery. +/// +public class ParallelMapTests +{ + [Test] + public async Task EmptySource_ReturnsEmptyArray() + { + var results = await ParallelMap.SelectParallelAsync( + [], x => Task.FromResult(x), Environment.ProcessorCount); + + await Assert.That(results).IsEmpty(); + } + + [Test] + [Arguments(1)] + [Arguments(ParallelMap.SequentialThreshold - 1)] + [Arguments(ParallelMap.SequentialThreshold)] + [Arguments(ParallelMap.SequentialThreshold + 1)] + [Arguments(100)] + public async Task ResultsPreserveSourceOrder(int count) + { + var source = Enumerable.Range(0, count).ToArray(); + + var results = await ParallelMap.SelectParallelAsync( + source, + async x => + { + // Stagger completion so later items often finish before earlier ones + await Task.Delay(x % 3); + return x * 2; + }, + maxDegreeOfParallelism: 4); + + await Assert.That(results.Length).IsEqualTo(count); + + for (var i = 0; i < count; i++) + { + await Assert.That(results[i]).IsEqualTo(i * 2); + } + } + + [Test] + public async Task EveryIndexInvokedExactlyOnce() + { + const int count = 500; + var invocations = new int[count]; + + await ParallelMap.ForParallelAsync( + count, + async i => + { + Interlocked.Increment(ref invocations[i]); + await Task.Yield(); + return i; + }, + Environment.ProcessorCount); + + for (var i = 0; i < count; i++) + { + await Assert.That(invocations[i]).IsEqualTo(1); + } + } + + [Test] + public async Task MaxDegreeOfParallelism_IsNotExceeded() + { + const int count = 64; + const int dop = 4; + var current = 0; + var maxObserved = 0; + + await ParallelMap.ForParallelAsync( + count, + async _ => + { + var now = Interlocked.Increment(ref current); + InterlockedMax(ref maxObserved, now); + await Task.Delay(5); + Interlocked.Decrement(ref current); + return 0; + }, + dop); + + await Assert.That(maxObserved).IsLessThanOrEqualTo(dop); + await Assert.That(maxObserved).IsGreaterThan(1); + } + + [Test] + public async Task SelectorException_PropagatesToCaller() + { + var exception = await Assert.ThrowsAsync(() => + ParallelMap.ForParallelAsync( + 100, + i => i == 5 + ? throw new InvalidOperationException("boom") + : Task.FromResult(i), + Environment.ProcessorCount)); + + await Assert.That(exception.Message).IsEqualTo("boom"); + } + + [Test] + public async Task SelectorException_StopsSiblingsPullingNewItems() + { + const int count = 10_000; + var invoked = 0; + + await Assert.ThrowsAsync(() => + ParallelMap.ForParallelAsync( + count, + async i => + { + Interlocked.Increment(ref invoked); + + if (i == 0) + { + throw new InvalidOperationException("fail fast"); + } + + await Task.Yield(); + return i; + }, + Environment.ProcessorCount)); + + // The parked cursor must prevent the full list from being processed. + // In-flight items may still complete, so allow generous slack. + await Assert.That(invoked).IsLessThan(count); + } + + [Test] + public async Task PreCancelledToken_Throws() + { + using var cts = new CancellationTokenSource(); + cts.Cancel(); + + await Assert.ThrowsAsync(() => + ParallelMap.ForParallelAsync( + 100, + i => Task.FromResult(i), + Environment.ProcessorCount, + cts.Token)); + } + + [Test] + public async Task ForEachParallelAsync_VisitsEveryItemExactlyOnce() + { + const int count = 500; + var source = Enumerable.Range(0, count).ToArray(); + var visits = new int[count]; + + await ParallelMap.ForEachParallelAsync( + source, + async i => + { + Interlocked.Increment(ref visits[i]); + await Task.Yield(); + }, + Environment.ProcessorCount); + + for (var i = 0; i < count; i++) + { + await Assert.That(visits[i]).IsEqualTo(1); + } + } + + [Test] + public async Task ForEachParallelAsync_ExceptionPropagates() + { + var source = Enumerable.Range(0, 100).ToArray(); + + await Assert.ThrowsAsync(() => + ParallelMap.ForEachParallelAsync( + source, + i => i == 50 ? throw new InvalidOperationException("boom") : default, + Environment.ProcessorCount)); + } + + private static void InterlockedMax(ref int location, int value) + { + int snapshot; + do + { + snapshot = Volatile.Read(ref location); + if (value <= snapshot) + { + return; + } + } while (Interlocked.CompareExchange(ref location, value, snapshot) != snapshot); + } +} diff --git a/tools/TUnit.Pipeline/Modules/UploadToNuGetModule.cs b/tools/TUnit.Pipeline/Modules/UploadToNuGetModule.cs index f64065c96f2..215b3024476 100644 --- a/tools/TUnit.Pipeline/Modules/UploadToNuGetModule.cs +++ b/tools/TUnit.Pipeline/Modules/UploadToNuGetModule.cs @@ -1,5 +1,4 @@ -using EnumerableAsyncProcessor.Extensions; -using Microsoft.Extensions.Options; +using Microsoft.Extensions.Options; using ModularPipelines.Attributes; using ModularPipelines.Configuration; using ModularPipelines.Context; @@ -66,22 +65,27 @@ protected override ModuleConfiguration Configure() => ModuleConfiguration.Create $"Refusing to publish to NuGet: found a package stamped with the dummy local version '{DummyLocalVersion}'."); } - return await nupkgs.SelectAsync(file => - context.DotNet().Nuget.Push(new DotNetNugetPushOptions - { - Path = file.Path, - Source = "https://api.nuget.org/v3/index.json", - ApiKey = options.Value.ApiKey, - }, new CommandExecutionOptions + var results = new CommandResult[nupkgs.Length]; + + for (var i = 0; i < nupkgs.Length; i++) + { + results[i] = await context.DotNet().Nuget.Push(new DotNetNugetPushOptions + { + Path = nupkgs[i].Path, + Source = "https://api.nuget.org/v3/index.json", + ApiKey = options.Value.ApiKey, + }, new CommandExecutionOptions + { + LogSettings = new CommandLoggingOptions { - LogSettings = new CommandLoggingOptions - { - ShowCommandArguments = true, - ShowStandardError = true, - ShowExecutionTime = true, - ShowExitCode = true - } - }, cancellationToken), cancellationToken: cancellationToken) - .ProcessOneAtATime(); + ShowCommandArguments = true, + ShowStandardError = true, + ShowExecutionTime = true, + ShowExitCode = true + } + }, cancellationToken); + } + + return results; } }