diff --git a/TUnit.Engine.Tests/GlobalHooks.cs b/TUnit.Engine.Tests/GlobalHooks.cs index 939b43dea5c..4f2f262d99e 100644 --- a/TUnit.Engine.Tests/GlobalHooks.cs +++ b/TUnit.Engine.Tests/GlobalHooks.cs @@ -8,7 +8,7 @@ public class GlobalHooks public static async Task BuildTestProject() { var result = await CliWrap.Cli.Wrap("dotnet") - .WithArguments(["build", "-c", GetConfiguration(), "--no-restore"]) + .WithArguments(["build", "-c", GetConfiguration()]) .WithWorkingDirectory(FileSystemHelpers.FindFile(x => x.Name == "TUnit.TestProject.csproj")!.DirectoryName!) .WithValidation(CliWrap.CommandResultValidation.None) .ExecuteBufferedAsync(); diff --git a/TUnit.Engine/Framework/TUnitServiceProvider.cs b/TUnit.Engine/Framework/TUnitServiceProvider.cs index 81c83f70a9c..53a5228f0a9 100644 --- a/TUnit.Engine/Framework/TUnitServiceProvider.cs +++ b/TUnit.Engine/Framework/TUnitServiceProvider.cs @@ -16,6 +16,7 @@ using TUnit.Engine.Helpers; using TUnit.Engine.Interfaces; using TUnit.Engine.Logging; +using TUnit.Engine.Scheduling; using TUnit.Engine.Services; namespace TUnit.Engine.Framework; @@ -44,6 +45,7 @@ public ITestExecutionFilter? Filter public ITestFinder TestFinder { get; } public TUnitInitializer Initializer { get; } public CancellationTokenSource FailFastCancellationSource { get; } + public ParallelLimitLockProvider ParallelLimitLockProvider { get; } public TUnitServiceProvider(IExtension extension, ExecuteRequestContext context, @@ -59,7 +61,7 @@ public TUnitServiceProvider(IExtension extension, var outputDevice = frameworkServiceProvider.GetOutputDevice(); CommandLineOptions = frameworkServiceProvider.GetCommandLineOptions(); var configuration = frameworkServiceProvider.GetConfiguration(); - + TestContext.Configuration = new ConfigurationAdapter(configuration); VerbosityService = Register(new VerbosityService(CommandLineOptions)); @@ -86,6 +88,8 @@ public TUnitServiceProvider(IExtension extension, HookCollectionService = Register(new HookCollectionService()); + ParallelLimitLockProvider = Register(new ParallelLimitLockProvider()); + ContextProvider = Register(new ContextProvider(this, TestSessionId, Filter?.ToString())); HookOrchestrator = Register(new HookOrchestrator(HookCollectionService, Logger, ContextProvider, this)); @@ -129,7 +133,7 @@ public TUnitServiceProvider(IExtension extension, // Create single test executor with ExecutionContext support var singleTestExecutor = Register( - new SingleTestExecutor(Logger, EventReceiverOrchestrator, HookCollectionService, context.Request.Session.SessionUid)); + new SingleTestExecutor(Logger, EventReceiverOrchestrator, HookCollectionService, CancellationToken, context.Request.Session.SessionUid)); // Create the HookOrchestratingTestExecutorAdapter // Note: We'll need to update this to handle dynamic dependencies properly @@ -146,14 +150,24 @@ public TUnitServiceProvider(IExtension extension, isFailFastEnabled, FailFastCancellationSource, Logger, - HookOrchestrator)); + HookOrchestrator, + ParallelLimitLockProvider)); + + // Create scheduler configuration from command line options + var schedulerConfig = GetSchedulerConfiguration(); + var testGroupingService = Register(new TestGroupingService()); + var testScheduler = Register(new Scheduling.TestScheduler( + Logger, + testGroupingService, + MessageBus, + schedulerConfig)); TestExecutor = Register(new TestExecutor( singleTestExecutor, CommandLineOptions, Logger, loggerFactory, - testScheduler: null, + testScheduler, serviceProvider: this, hookOrchestratingTestExecutorAdapter, ContextProvider, @@ -230,6 +244,40 @@ private static bool GetUseSourceGeneration(ICommandLineOptions commandLineOption return SourceRegistrar.IsEnabled; } + private SchedulerConfiguration GetSchedulerConfiguration() + { + var config = new SchedulerConfiguration(); + + // Handle --maximum-parallel-tests + if (CommandLineOptions.TryGetOptionArgumentList( + MaximumParallelTestsCommandProvider.MaximumParallelTests, + out var args) && args.Length > 0) + { + if (int.TryParse(args[0], out var maxParallelTests) && maxParallelTests > 0) + { + config.MaxParallelism = maxParallelTests; + config.AdaptiveMaxParallelism = maxParallelTests; + } + } + + // Handle --parallelism-strategy + if (CommandLineOptions.TryGetOptionArgumentList( + ParallelismStrategyCommandProvider.ParallelismStrategy, + out var strategyArgs) && strategyArgs.Length > 0) + { + var strategy = strategyArgs[0].ToLowerInvariant(); + config.Strategy = strategy == "fixed" ? ParallelismStrategy.Fixed : ParallelismStrategy.Adaptive; + } + + // Handle --adaptive-metrics + if (CommandLineOptions.IsOptionSet(AdaptiveMetricsCommandProvider.AdaptiveMetrics)) + { + config.EnableAdaptiveMetrics = true; + } + + return config; + } + public async ValueTask DisposeAsync() { foreach (var service in _services.Values) diff --git a/TUnit.Engine/Scheduling/AdaptiveParallelismController.cs b/TUnit.Engine/Scheduling/AdaptiveParallelismController.cs deleted file mode 100644 index e49e0a50bde..00000000000 --- a/TUnit.Engine/Scheduling/AdaptiveParallelismController.cs +++ /dev/null @@ -1,243 +0,0 @@ -using TUnit.Core.Logging; -using TUnit.Engine.Logging; -using TUnit.Engine.Services; - -namespace TUnit.Engine.Scheduling; - -/// -/// Controller that runs in the background and adjusts parallelism based on system metrics -/// -internal sealed class AdaptiveParallelismController : IDisposable -{ - private readonly AdaptiveSemaphore _semaphore; - private readonly SystemMetricsCollector _metricsCollector; - private readonly ParallelismAdjustmentStrategy _adjustmentStrategy; - private readonly TUnitFrameworkLogger _logger; - private readonly bool _enableMetricsLogging; - private readonly CancellationTokenSource _cancellationSource; - private readonly Task _adjustmentTask; - private readonly Task? _metricsLoggingTask; - private int _currentParallelism; - private bool _disposed; - - public AdaptiveParallelismController( - AdaptiveSemaphore semaphore, - TUnitFrameworkLogger logger, - int minParallelism, - int maxParallelism, - int initialParallelism, - bool enableMetricsLogging) - { - _semaphore = semaphore ?? throw new ArgumentNullException(nameof(semaphore)); - _logger = logger ?? throw new ArgumentNullException(nameof(logger)); - _enableMetricsLogging = enableMetricsLogging; - _currentParallelism = initialParallelism; - - _metricsCollector = new SystemMetricsCollector(); - _adjustmentStrategy = new ParallelismAdjustmentStrategy(minParallelism, maxParallelism); - _cancellationSource = new CancellationTokenSource(); - - // Start background tasks - _adjustmentTask = RunAdjustmentLoopAsync(_cancellationSource.Token); - - if (_enableMetricsLogging) - { - _metricsLoggingTask = RunMetricsLoggingLoopAsync(_cancellationSource.Token); - } - } - - /// - /// Gets the current parallelism level - /// - public int CurrentParallelism => _currentParallelism; - - /// - /// Records a test completion for metrics - /// - public void RecordTestCompletion(TimeSpan executionTime) - { - _adjustmentStrategy.RecordTestCompletion(executionTime); - } - - private async Task RunAdjustmentLoopAsync(CancellationToken cancellationToken) - { -#if NET6_0_OR_GREATER - // Use PeriodicTimer for cleaner async timing (500ms intervals) - using var timer = new PeriodicTimer(TimeSpan.FromMilliseconds(500)); - - while (!cancellationToken.IsCancellationRequested) - { - try - { - await timer.WaitForNextTickAsync(cancellationToken); - await AdjustParallelismAsync(); - } - catch (OperationCanceledException) - { - // Expected when cancellation is requested - break; - } - catch (Exception ex) - { - // Log error but don't crash the adjustment loop - await _logger.LogErrorAsync($"Error in adaptive parallelism adjustment: {ex.Message}"); - } - } -#else - // Fallback for netstandard2.0 - while (!cancellationToken.IsCancellationRequested) - { - try - { - await Task.Delay(500, cancellationToken); - await AdjustParallelismAsync(); - } - catch (OperationCanceledException) - { - // Expected when cancellation is requested - break; - } - catch (Exception ex) - { - // Log error but don't crash the adjustment loop - await _logger.LogErrorAsync($"Error in adaptive parallelism adjustment: {ex.Message}"); - } - } -#endif - } - - private async Task RunMetricsLoggingLoopAsync(CancellationToken cancellationToken) - { - // Initial delay to let tests start - await Task.Delay(1000, cancellationToken); - -#if NET6_0_OR_GREATER - // Use PeriodicTimer for metrics logging (3 second intervals) - using var timer = new PeriodicTimer(TimeSpan.FromSeconds(3)); - - while (!cancellationToken.IsCancellationRequested) - { - try - { - await timer.WaitForNextTickAsync(cancellationToken); - - // Get current metrics - var metrics = _metricsCollector.GetMetrics(); - await LogMetrics(metrics); - } - catch (OperationCanceledException) - { - // Expected when cancellation is requested - break; - } - catch (Exception ex) - { - // Log error but continue - await _logger.LogErrorAsync($"Error logging adaptive metrics: {ex.Message}"); - } - } -#else - // Fallback for netstandard2.0 - while (!cancellationToken.IsCancellationRequested) - { - try - { - await Task.Delay(3000, cancellationToken); - - // Get current metrics - var metrics = _metricsCollector.GetMetrics(); - await LogMetrics(metrics); - } - catch (OperationCanceledException) - { - // Expected when cancellation is requested - break; - } - catch (Exception ex) - { - // Log error but continue - await _logger.LogErrorAsync($"Error logging adaptive metrics: {ex.Message}"); - } - } -#endif - } - - private async Task AdjustParallelismAsync() - { - // Collect metrics - var metrics = _metricsCollector.GetMetrics(); - - // Calculate adjustment - var recommendation = _adjustmentStrategy.CalculateAdjustment(metrics, _currentParallelism); - - // Apply adjustment if needed - if (recommendation.NewParallelism != _currentParallelism) - { - _semaphore.AdjustMaxCount(recommendation.NewParallelism); - var oldParallelism = _currentParallelism; - _currentParallelism = recommendation.NewParallelism; - - if (_enableMetricsLogging) - { - await LogAdjustment(oldParallelism, recommendation, metrics); - } - } - } - - private async Task LogAdjustment(int oldParallelism, AdjustmentRecommendation recommendation, SystemMetrics metrics) - { - var direction = recommendation.Direction == AdjustmentDirection.Increase ? "↑" : "↓"; - await _logger.LogDebugAsync( - $"[Adaptive] Parallelism adjusted: {oldParallelism} {direction} {recommendation.NewParallelism} | " + - $"Reason: {recommendation.Reason} | " + - $"CPU: {metrics.SystemCpuUsagePercent:F1}% | " + - $"Threads: {metrics.AvailableWorkerThreads}/{metrics.MaxWorkerThreads} | " + - $"Memory: {metrics.TotalMemoryBytes / 1_000_000}MB"); - } - - private async Task LogMetrics(SystemMetrics metrics) - { - var semaphoreAvailable = _semaphore.CurrentCount; - var activeTests = _currentParallelism - semaphoreAvailable; - - await _logger.LogDebugAsync( - $"[Adaptive] Metrics | " + - $"Parallelism: {_currentParallelism} (Active: {activeTests}, Available: {semaphoreAvailable}) | " + - $"CPU: {metrics.SystemCpuUsagePercent:F1}% | " + - $"Threads: {metrics.AvailableWorkerThreads}/{metrics.MaxWorkerThreads} | " + - $"Pending: {metrics.PendingWorkItems} | " + - $"Memory: {metrics.TotalMemoryBytes / 1_000_000}MB"); - } - - public void Dispose() - { - if (_disposed) - return; - - _disposed = true; - - // Cancel background tasks - _cancellationSource.Cancel(); - - // Wait for tasks to complete (with timeout) - try - { - var tasksToWait = new[] { _adjustmentTask, _metricsLoggingTask } - .Where(t => t != null) - .Cast() - .ToArray(); - - if (tasksToWait.Length > 0) - { - Task.WaitAll(tasksToWait, TimeSpan.FromSeconds(5)); - } - } - catch (AggregateException) - { - // Tasks may have been cancelled, which is expected - } - - _cancellationSource.Dispose(); - _metricsCollector?.Dispose(); - } -} \ No newline at end of file diff --git a/TUnit.Engine/Scheduling/AdaptiveSemaphore.cs b/TUnit.Engine/Scheduling/AdaptiveSemaphore.cs deleted file mode 100644 index eb9d53c19dd..00000000000 --- a/TUnit.Engine/Scheduling/AdaptiveSemaphore.cs +++ /dev/null @@ -1,184 +0,0 @@ -using System.Collections.Concurrent; - -namespace TUnit.Engine.Scheduling; - -/// -/// A semaphore that supports dynamic adjustment of maximum count -/// -internal sealed class AdaptiveSemaphore : IDisposable -{ - private readonly object _lock = new(); - private readonly ConcurrentQueue> _waiters = new(); - private int _currentCount; - private int _maxCount; - private bool _disposed; - - public AdaptiveSemaphore(int initialCount, int maxCount) - { - if (initialCount < 0) throw new ArgumentOutOfRangeException(nameof(initialCount)); - if (maxCount < 1) throw new ArgumentOutOfRangeException(nameof(maxCount)); - if (initialCount > maxCount) throw new ArgumentException("Initial count cannot exceed max count"); - - _currentCount = initialCount; - _maxCount = maxCount; - } - - /// - /// Gets the current available count - /// - public int CurrentCount - { - get - { - lock (_lock) - { - return _currentCount; - } - } - } - - /// - /// Gets the current maximum count - /// - public int MaxCount - { - get - { - lock (_lock) - { - return _maxCount; - } - } - } - - /// - /// Waits to enter the semaphore - /// - public async Task WaitAsync(CancellationToken cancellationToken = default) - { - TaskCompletionSource? waiter = null; - - lock (_lock) - { - if (_disposed) - throw new ObjectDisposedException(nameof(AdaptiveSemaphore)); - - if (_currentCount > 0) - { - _currentCount--; - return; - } - - waiter = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - _waiters.Enqueue(waiter); - } - - using (cancellationToken.Register(() => waiter.TrySetCanceled())) - { - await waiter.Task.ConfigureAwait(false); - } - } - - /// - /// Releases one count back to the semaphore - /// - public void Release() - { - TaskCompletionSource? waiterToRelease = null; - - lock (_lock) - { - if (_disposed) - throw new ObjectDisposedException(nameof(AdaptiveSemaphore)); - - if (_waiters.TryDequeue(out waiterToRelease)) - { - // Release directly to a waiter without incrementing count - } - else - { - // Don't throw if we're at max - this can happen when max count is reduced - // while tests are still running. Just silently ignore the release. - if (_currentCount < _maxCount) - { - _currentCount++; - } - } - } - - waiterToRelease?.TrySetResult(true); - } - - /// - /// Adjusts the maximum count of the semaphore - /// - public void AdjustMaxCount(int newMaxCount) - { - if (newMaxCount < 1) - throw new ArgumentOutOfRangeException(nameof(newMaxCount)); - - var waitersToRelease = new List>(); - - lock (_lock) - { - if (_disposed) - throw new ObjectDisposedException(nameof(AdaptiveSemaphore)); - - var oldMaxCount = _maxCount; - _maxCount = newMaxCount; - - // If we're increasing the max count, we might be able to release some waiters - if (newMaxCount > oldMaxCount) - { - var additionalCapacity = newMaxCount - oldMaxCount; - _currentCount += additionalCapacity; - - // Release waiters if we have capacity - while (_currentCount > 0 && _waiters.TryDequeue(out var waiter)) - { - waitersToRelease.Add(waiter); - _currentCount--; - } - } - else if (newMaxCount < oldMaxCount) - { - // If decreasing, cap the current count at the new max - // This prevents issues but allows running tests to complete - if (_currentCount > newMaxCount) - { - _currentCount = newMaxCount; - } - } - } - - // Release waiters outside the lock to avoid potential deadlocks - foreach (var waiter in waitersToRelease) - { - waiter.TrySetResult(true); - } - } - - public void Dispose() - { - List> waitersToCancel; - - lock (_lock) - { - if (_disposed) - return; - - _disposed = true; - waitersToCancel = new List>(); - - while (_waiters.TryDequeue(out var waiter)) - { - waitersToCancel.Add(waiter); - } - } - - foreach (var waiter in waitersToCancel) - { - waiter.TrySetCanceled(); - } - } -} \ No newline at end of file diff --git a/TUnit.Engine/Scheduling/ParallelismAdjustmentStrategy.cs b/TUnit.Engine/Scheduling/ParallelismAdjustmentStrategy.cs deleted file mode 100644 index d57164df055..00000000000 --- a/TUnit.Engine/Scheduling/ParallelismAdjustmentStrategy.cs +++ /dev/null @@ -1,226 +0,0 @@ -using System.Collections.Concurrent; -using TUnit.Engine.Services; - -namespace TUnit.Engine.Scheduling; - -/// -/// Strategy for adjusting parallelism based on system metrics -/// -internal sealed class ParallelismAdjustmentStrategy -{ - private readonly int _minParallelism; - private readonly int _maxParallelism; - private readonly ConcurrentQueue _completedTests = new(); - private DateTime _lastMeasurementTime = DateTime.UtcNow; - private int _lastCompletedCount; - private const double CpuLowThreshold = 70.0; - private const double CpuHighThreshold = 90.0; - private const double MinIncreaseFactor = 0.25; // 25% increase (more aggressive) - private const double MinDecreaseFactor = 0.15; // 15% decrease (more conservative) - - public ParallelismAdjustmentStrategy(int minParallelism, int maxParallelism) - { - _minParallelism = Math.Max(1, minParallelism); - _maxParallelism = Math.Max(_minParallelism, maxParallelism); - } - - /// - /// Records a test completion for rate calculation - /// - public void RecordTestCompletion(TimeSpan executionTime) - { - _completedTests.Enqueue(new TestCompletionInfo - { - CompletionTime = DateTime.UtcNow, - ExecutionTime = executionTime - }); - - // Clean up old entries (older than 10 seconds) - var cutoff = DateTime.UtcNow.AddSeconds(-10); - while (_completedTests.TryPeek(out var oldest) && oldest.CompletionTime < cutoff) - { - _completedTests.TryDequeue(out _); - } - } - - /// - /// Calculates the recommended parallelism adjustment - /// - public AdjustmentRecommendation CalculateAdjustment(SystemMetrics metrics, int currentParallelism) - { - var decision = MakeDecision(metrics, currentParallelism); - - // No sliding window - make immediate adjustments - if (decision.Direction == AdjustmentDirection.None) - { - return new AdjustmentRecommendation - { - NewParallelism = currentParallelism, - Direction = AdjustmentDirection.None, - Reason = decision.Reason - }; - } - - // Calculate new parallelism based on direction and current metrics - int newParallelism; - if (decision.Direction == AdjustmentDirection.Increase) - { - // Calculate optimal increase based on current CPU usage - // If we're at 2% CPU with 176 tests, and we want to reach ~70% CPU, - // we can estimate: newParallelism = currentParallelism * (targetCPU / currentCPU) - var currentCpu = metrics.SystemCpuUsagePercent; - if (currentCpu > 0 && currentCpu < 10.0) // Very low CPU usage - { - // Aggressive scaling: try to reach 60% CPU utilization - var scaleFactor = Math.Min(60.0 / currentCpu, 3.0); // Cap at 3x to avoid overshooting - var targetParallelism = (int)(currentParallelism * scaleFactor); - newParallelism = Math.Min(_maxParallelism, targetParallelism); - } - else - { - // Normal increase by 25% - var adjustmentSize = Math.Max(1, (int)(currentParallelism * MinIncreaseFactor)); - newParallelism = Math.Min(_maxParallelism, currentParallelism + adjustmentSize); - } - } - else - { - // Decrease more conservatively - var adjustmentSize = Math.Max(1, (int)(currentParallelism * MinDecreaseFactor)); - newParallelism = Math.Max(_minParallelism, currentParallelism - adjustmentSize); - } - - return new AdjustmentRecommendation - { - NewParallelism = newParallelism, - Direction = decision.Direction, - Reason = decision.Reason - }; - } - - private AdjustmentDecision MakeDecision(SystemMetrics metrics, int currentParallelism) - { - // Check for thread pool starvation - var threadUtilization = CalculateThreadUtilization(metrics); - if (threadUtilization > 0.9 || metrics.PendingWorkItems > 100) - { - return new AdjustmentDecision - { - Direction = AdjustmentDirection.Decrease, - Reason = $"Thread pool starvation detected (utilization: {threadUtilization:P1}, pending: {metrics.PendingWorkItems})" - }; - } - - // Check CPU usage - if (metrics.SystemCpuUsagePercent > CpuHighThreshold) - { - return new AdjustmentDecision - { - Direction = AdjustmentDirection.Decrease, - Reason = $"High CPU usage ({metrics.SystemCpuUsagePercent:F1}%)" - }; - } - - // Check memory pressure - if (metrics.TotalMemoryBytes > 1_000_000_000) // Over 1GB - { - return new AdjustmentDecision - { - Direction = AdjustmentDirection.Decrease, - Reason = $"High memory usage ({metrics.TotalMemoryBytes / 1_000_000}MB)" - }; - } - - // Calculate test completion rate - var completionRate = CalculateCompletionRate(); - - // Check if we can increase parallelism - if (metrics.SystemCpuUsagePercent < CpuLowThreshold && - threadUtilization < 0.7 && - currentParallelism < _maxParallelism) - { - // Check if completion rate is stable or improving - if (completionRate >= 0) // Not declining - { - return new AdjustmentDecision - { - Direction = AdjustmentDirection.Increase, - Reason = $"Resources available (CPU: {metrics.SystemCpuUsagePercent:F1}%, threads: {threadUtilization:P1})" - }; - } - } - - // If completion rate is declining significantly, decrease - if (completionRate < -0.2) // More than 20% decline - { - return new AdjustmentDecision - { - Direction = AdjustmentDirection.Decrease, - Reason = $"Test completion rate declining ({completionRate:P1})" - }; - } - - return new AdjustmentDecision - { - Direction = AdjustmentDirection.None, - Reason = "System metrics stable" - }; - } - - private double CalculateThreadUtilization(SystemMetrics metrics) - { - if (metrics.MaxWorkerThreads == 0) return 0; - return 1.0 - (double)metrics.AvailableWorkerThreads / metrics.MaxWorkerThreads; - } - - private double CalculateCompletionRate() - { - var now = DateTime.UtcNow; - var timeDelta = (now - _lastMeasurementTime).TotalSeconds; - if (timeDelta < 1) return 0; // Not enough time passed - - var currentCount = _completedTests.Count; - var completedInPeriod = currentCount - _lastCompletedCount; - - var currentRate = completedInPeriod / timeDelta; - var previousRate = _lastCompletedCount / 10.0; // Over 10 second window - - _lastCompletedCount = currentCount; - _lastMeasurementTime = now; - - if (previousRate == 0) return 0; - return (currentRate - previousRate) / previousRate; // Percentage change - } - - private sealed class AdjustmentDecision - { - public AdjustmentDirection Direction { get; init; } - public string Reason { get; init; } = ""; - } - - private sealed class TestCompletionInfo - { - public DateTime CompletionTime { get; init; } - public TimeSpan ExecutionTime { get; init; } - } -} - -/// -/// Adjustment direction -/// -internal enum AdjustmentDirection -{ - None, - Increase, - Decrease -} - -/// -/// Adjustment recommendation -/// -internal sealed class AdjustmentRecommendation -{ - public int NewParallelism { get; init; } - public AdjustmentDirection Direction { get; init; } - public string Reason { get; init; } = ""; -} \ No newline at end of file diff --git a/TUnit.Engine/Scheduling/TestExecutor.cs b/TUnit.Engine/Scheduling/TestExecutor.cs index 27a3e3d0628..78e8df1381a 100644 --- a/TUnit.Engine/Scheduling/TestExecutor.cs +++ b/TUnit.Engine/Scheduling/TestExecutor.cs @@ -20,6 +20,7 @@ internal sealed class TestExecutor : ITestExecutor, IDataProducer private readonly CancellationTokenSource _failFastCancellationSource; private readonly TUnitFrameworkLogger _logger; private readonly HookOrchestrator _hookOrchestrator; + private readonly ParallelLimitLockProvider _parallelLimitLockProvider; // IDataProducer implementation public string Uid => "TUnit.TestExecutor"; @@ -37,7 +38,8 @@ public TestExecutor( bool isFailFastEnabled, CancellationTokenSource failFastCancellationSource, TUnitFrameworkLogger logger, - HookOrchestrator hookOrchestrator) + HookOrchestrator hookOrchestrator, + ParallelLimitLockProvider parallelLimitLockProvider) { _innerExecutor = innerExecutor; _messageBus = messageBus; @@ -47,6 +49,7 @@ public TestExecutor( _failFastCancellationSource = failFastCancellationSource; _logger = logger; _hookOrchestrator = hookOrchestrator; + _parallelLimitLockProvider = parallelLimitLockProvider; } public async Task ExecuteTestAsync(AbstractExecutableTest test, CancellationToken cancellationToken) @@ -72,23 +75,33 @@ public async Task ExecuteTestAsync(AbstractExecutableTest test, CancellationToke return; } - // Simple state management - scheduler ensures we only get here for executable tests - test.State = TestState.Running; - test.StartTime = DateTimeOffset.UtcNow; - - // Report test started - await _tunitMessageBus.InProgress(test.Context); + // Acquire semaphore for parallel limit if configured + SemaphoreSlim? parallelLimitSemaphore = null; + if (test.Context.ParallelLimiter != null) + { + parallelLimitSemaphore = _parallelLimitLockProvider.GetLock(test.Context.ParallelLimiter); + await parallelLimitSemaphore.WaitAsync(cancellationToken); + } try { - if (test.Context.TestDetails.ClassInstance is PlaceholderInstance) + // Simple state management - scheduler ensures we only get here for executable tests + test.State = TestState.Running; + test.StartTime = DateTimeOffset.UtcNow; + + // Report test started + await _tunitMessageBus.InProgress(test.Context); + + try { - var instance = await test.CreateInstanceAsync(); - test.Context.TestDetails.ClassInstance = instance; - } + if (test.Context.TestDetails.ClassInstance is PlaceholderInstance) + { + var instance = await test.CreateInstanceAsync(); + test.Context.TestDetails.ClassInstance = instance; + } - // Execute class/assembly hooks on first test - var executionContext = await _hookOrchestrator.OnTestStartingAsync(test, cancellationToken); + // Execute class/assembly hooks on first test + var executionContext = await _hookOrchestrator.OnTestStartingAsync(test, cancellationToken); #if NET // Restore the accumulated context from all hooks to flow AsyncLocal values to the test @@ -98,52 +111,58 @@ public async Task ExecuteTestAsync(AbstractExecutableTest test, CancellationToke } #endif - // Execute the test and get the result message - var updateMessage = await _innerExecutor.ExecuteTestAsync(test, cancellationToken); + // Execute the test and get the result message + var updateMessage = await _innerExecutor.ExecuteTestAsync(test, cancellationToken); - // Route the result to the appropriate ITUnitMessageBus method - await RouteTestResult(test, updateMessage); + // Route the result to the appropriate ITUnitMessageBus method + await RouteTestResult(test, updateMessage); - // Check if we should trigger fail-fast - if (_isFailFastEnabled && test.Result?.State == TestState.Failed) - { - await _logger.LogErrorAsync($"Test {test.TestId} failed. Triggering fail-fast cancellation."); - _failFastCancellationSource.Cancel(); + // Check if we should trigger fail-fast + if (_isFailFastEnabled && test.Result?.State == TestState.Failed) + { + await _logger.LogErrorAsync($"Test {test.TestId} failed. Triggering fail-fast cancellation."); + _failFastCancellationSource.Cancel(); + } } - } - catch (Exception ex) - { - // Set test state - test.State = TestState.Failed; - test.Result = new TestResult + catch (Exception ex) { - State = TestState.Failed, - Start = test.StartTime, - End = DateTimeOffset.Now, - Duration = DateTimeOffset.Now - test.StartTime.GetValueOrDefault(), - Exception = ex, - ComputerName = Environment.MachineName - }; - - // Report the failure - await _tunitMessageBus.Failed(test.Context, ex, test.StartTime.GetValueOrDefault()); - - // Log the exception - await _logger.LogErrorAsync($"Unhandled exception in test {test.TestId}: {ex}"); - - // If fail-fast is enabled, cancel all remaining tests - if (_isFailFastEnabled) + // Set test state + test.State = TestState.Failed; + test.Result = new TestResult + { + State = TestState.Failed, + Start = test.StartTime, + End = DateTimeOffset.Now, + Duration = DateTimeOffset.Now - test.StartTime.GetValueOrDefault(), + Exception = ex, + ComputerName = Environment.MachineName + }; + + // Report the failure + await _tunitMessageBus.Failed(test.Context, ex, test.StartTime.GetValueOrDefault()); + + // Log the exception + await _logger.LogErrorAsync($"Unhandled exception in test {test.TestId}: {ex}"); + + // If fail-fast is enabled, cancel all remaining tests + if (_isFailFastEnabled) + { + await _logger.LogErrorAsync("Unhandled exception occurred. Triggering fail-fast cancellation."); + _failFastCancellationSource.Cancel(); + } + + // Re-throw to maintain existing behavior + throw; + } + finally { - await _logger.LogErrorAsync("Unhandled exception occurred. Triggering fail-fast cancellation."); - _failFastCancellationSource.Cancel(); + test.EndTime = DateTimeOffset.UtcNow; } - - // Re-throw to maintain existing behavior - throw; } finally { - test.EndTime = DateTimeOffset.UtcNow; + // Release semaphore if we acquired one + parallelLimitSemaphore?.Release(); } } diff --git a/TUnit.Engine/Scheduling/TestScheduler.cs b/TUnit.Engine/Scheduling/TestScheduler.cs index e3a4da79465..58e2df5144f 100644 --- a/TUnit.Engine/Scheduling/TestScheduler.cs +++ b/TUnit.Engine/Scheduling/TestScheduler.cs @@ -1,4 +1,5 @@ using System.Collections.Concurrent; +using EnumerableAsyncProcessor.Extensions; using TUnit.Core; using TUnit.Core.Logging; using TUnit.Engine.Logging; @@ -10,14 +11,12 @@ namespace TUnit.Engine.Scheduling; /// /// A clean, simplified test scheduler that uses an execution plan /// -internal sealed class TestScheduler : ITestScheduler, IDisposable +internal sealed class TestScheduler : ITestScheduler { private readonly TUnitFrameworkLogger _logger; private readonly ITestGroupingService _groupingService; private readonly ITUnitMessageBus _messageBus; private readonly SchedulerConfiguration _configuration; - private AdaptiveParallelismController? _adaptiveController; - private IDisposable? _semaphore; public TestScheduler( TUnitFrameworkLogger logger, @@ -42,7 +41,6 @@ public async Task ScheduleAndExecuteAsync( // Create execution plan upfront var plan = ExecutionPlan.Create(tests); - if (plan.ExecutableTests.Count == 0) { await _logger.LogDebugAsync("No executable tests found"); @@ -65,29 +63,13 @@ private async Task ExecuteGroupedTestsAsync( var runningTasks = new ConcurrentDictionary(); var completedTests = new ConcurrentDictionary(); - // Create appropriate semaphore based on strategy - object semaphore; - if (_configuration.Strategy == ParallelismStrategy.Adaptive) - { - var initialParallelism = Math.Min(Environment.ProcessorCount * 4, _configuration.AdaptiveMaxParallelism); - var adaptiveSemaphore = new AdaptiveSemaphore(initialParallelism, _configuration.AdaptiveMaxParallelism); - _adaptiveController = new AdaptiveParallelismController( - adaptiveSemaphore, - _logger, - _configuration.AdaptiveMinParallelism, - _configuration.AdaptiveMaxParallelism, - initialParallelism, - _configuration.EnableAdaptiveMetrics); - semaphore = adaptiveSemaphore; - _semaphore = adaptiveSemaphore; - } - else + // Determine parallelism level + int? maxParallelism = null; + if (_configuration.Strategy != ParallelismStrategy.Adaptive) { - var maxParallelism = _configuration.MaxParallelism > 0 ? _configuration.MaxParallelism : Environment.ProcessorCount * 4; - var fixedSemaphore = new SemaphoreSlim(maxParallelism, maxParallelism); - semaphore = fixedSemaphore; - _semaphore = fixedSemaphore; + maxParallelism = _configuration.MaxParallelism > 0 ? _configuration.MaxParallelism : Environment.ProcessorCount * 4; } + // For adaptive, we pass null to let EnumerableAsyncProcessor manage concurrency // Process all test groups var allTestTasks = new List(); @@ -125,7 +107,7 @@ private async Task ExecuteGroupedTestsAsync( executor, runningTasks, completedTests, - semaphore, + maxParallelism, cancellationToken); allTestTasks.Add(groupTask); } @@ -135,7 +117,7 @@ private async Task ExecuteGroupedTestsAsync( executor, runningTasks, completedTests, - semaphore, + maxParallelism, cancellationToken); allTestTasks.Add(parallelTask); @@ -157,24 +139,33 @@ private async Task ExecuteNotInParallelTestsAsync( testsWithPriority.Add((test, priority)); } - // Sort by NotInParallel Order first, then by execution order from the plan for dependency resolution - testsWithPriority.Sort((a, b) => + // Group tests by class + var testsByClass = testsWithPriority + .GroupBy(t => t.Test.Context.TestDetails.ClassType) + .ToList(); + + // Sort classes by their minimum test Order + testsByClass.Sort((a, b) => { - // Primary sort: NotInParallel Order (from TestPriority) - var priorityComparison = a.Priority.CompareTo(b.Priority); - if (priorityComparison != 0) - return priorityComparison; - - // Secondary sort: ExecutionPlan order for dependency resolution when NotInParallel orders are equal - var aOrder = plan.ExecutionOrder.TryGetValue(a.Test, out var ao) ? ao : int.MaxValue; - var bOrder = plan.ExecutionOrder.TryGetValue(b.Test, out var bo) ? bo : int.MaxValue; - return aOrder.CompareTo(bOrder); + var aMinOrder = a.Min(t => t.Priority.Order); + var bMinOrder = b.Min(t => t.Priority.Order); + return aMinOrder.CompareTo(bMinOrder); }); - // Execute sequentially - foreach (var (test, _) in testsWithPriority) + // Execute class by class + foreach (var classGroup in testsByClass) { - await ExecuteTestWhenReadyAsync(test, executor, runningTasks, completedTests, cancellationToken); + // Sort tests within the class by Order, then by execution plan order + var classTests = classGroup.OrderBy(t => t.Priority.Order) + .ThenBy(t => plan.ExecutionOrder.TryGetValue(t.Test, out var order) ? order : int.MaxValue) + .Select(t => t.Test) + .ToList(); + + // Execute all tests from this class sequentially + foreach (var test in classTests) + { + await ExecuteTestWhenReadyAsync(test, executor, runningTasks, completedTests, cancellationToken); + } } } @@ -192,24 +183,33 @@ private async Task ExecuteKeyedNotInParallelTestsAsync( testsWithPriority.Add((test, priority)); } - // Sort by NotInParallel Order first, then by execution order from the plan for dependency resolution - testsWithPriority.Sort((a, b) => + // Group tests by class + var testsByClass = testsWithPriority + .GroupBy(t => t.Test.Context.TestDetails.ClassType) + .ToList(); + + // Sort classes by their minimum test Order + testsByClass.Sort((a, b) => { - // Primary sort: NotInParallel Order (from TestPriority) - var priorityComparison = a.Priority.CompareTo(b.Priority); - if (priorityComparison != 0) - return priorityComparison; - - // Secondary sort: ExecutionPlan order for dependency resolution when NotInParallel orders are equal - var aOrder = plan.ExecutionOrder.TryGetValue(a.Test, out var ao) ? ao : int.MaxValue; - var bOrder = plan.ExecutionOrder.TryGetValue(b.Test, out var bo) ? bo : int.MaxValue; - return aOrder.CompareTo(bOrder); + var aMinOrder = a.Min(t => t.Priority.Order); + var bMinOrder = b.Min(t => t.Priority.Order); + return aMinOrder.CompareTo(bMinOrder); }); - // Execute sequentially within this key - foreach (var (test, _) in testsWithPriority) + // Execute class by class within this key + foreach (var classGroup in testsByClass) { - await ExecuteTestWhenReadyAsync(test, executor, runningTasks, completedTests, cancellationToken); + // Sort tests within the class by Order, then by execution plan order + var classTests = classGroup.OrderBy(t => t.Priority.Order) + .ThenBy(t => plan.ExecutionOrder.TryGetValue(t.Test, out var order) ? order : int.MaxValue) + .Select(t => t.Test) + .ToList(); + + // Execute all tests from this class sequentially + foreach (var test in classTests) + { + await ExecuteTestWhenReadyAsync(test, executor, runningTasks, completedTests, cancellationToken); + } } } @@ -217,42 +217,26 @@ private async Task ExecuteParallelGroupAsync(SortedDictionary runningTasks, ConcurrentDictionary completedTests, - object semaphore, + int? maxParallelism, CancellationToken cancellationToken) { // Execute order groups sequentially foreach (var orderGroup in orderGroups.OrderBy(og => og.Key)) { - var tasks = new List(); - - foreach (var test in orderGroup.Value) + // Use EnumerableAsyncProcessor to execute tests in parallel + if (maxParallelism.HasValue) { - if (semaphore is AdaptiveSemaphore adaptive) - await adaptive.WaitAsync(cancellationToken); - else - await ((SemaphoreSlim)semaphore).WaitAsync(cancellationToken); - - // Create a task that releases semaphore on completion without Task.Run overhead - async Task ExecuteWithSemaphoreRelease() - { - try - { - await ExecuteTestWhenReadyAsync(test, executor, runningTasks, completedTests, cancellationToken); - } - finally - { - if (semaphore is AdaptiveSemaphore adaptive) - adaptive.Release(); - else - ((SemaphoreSlim)semaphore).Release(); - } - } - - tasks.Add(ExecuteWithSemaphoreRelease()); + await orderGroup.Value + .ForEachAsync(async test => await ExecuteTestWhenReadyAsync(test, executor, runningTasks, completedTests, cancellationToken)) + .ProcessInParallel(maxParallelism.Value); + } + else + { + // Adaptive parallelism - no limit specified + await orderGroup.Value + .ForEachAsync(async test => await ExecuteTestWhenReadyAsync(test, executor, runningTasks, completedTests, cancellationToken)) + .ProcessInParallel(); } - - // Wait for all tests in this order group to complete - await Task.WhenAll(tasks); } } @@ -260,38 +244,23 @@ private async Task ExecuteParallelTestsAsync(IEnumerable ITestExecutor executor, ConcurrentDictionary runningTasks, ConcurrentDictionary completedTests, - object semaphore, + int? maxParallelism, CancellationToken cancellationToken) { - var tasks = new List(); - - foreach (var test in tests) + // Use EnumerableAsyncProcessor to execute tests in parallel + if (maxParallelism.HasValue) { - if (semaphore is AdaptiveSemaphore adaptive) - await adaptive.WaitAsync(cancellationToken); - else - await ((SemaphoreSlim)semaphore).WaitAsync(cancellationToken); - - // Create a task that releases semaphore on completion without Task.Run overhead - async Task ExecuteWithSemaphoreRelease() - { - try - { - await ExecuteTestWhenReadyAsync(test, executor, runningTasks, completedTests, cancellationToken); - } - finally - { - if (semaphore is AdaptiveSemaphore adaptive) - adaptive.Release(); - else - ((SemaphoreSlim)semaphore).Release(); - } - } - - tasks.Add(ExecuteWithSemaphoreRelease()); + await tests + .ForEachAsync(async test => await ExecuteTestWhenReadyAsync(test, executor, runningTasks, completedTests, cancellationToken)) + .ProcessInParallel(maxParallelism.Value); + } + else + { + // Adaptive parallelism - no limit specified + await tests + .ForEachAsync(async test => await ExecuteTestWhenReadyAsync(test, executor, runningTasks, completedTests, cancellationToken)) + .ProcessInParallel(); } - - await Task.WhenAll(tasks); } private async Task ExecuteTestWhenReadyAsync(AbstractExecutableTest test, @@ -327,7 +296,6 @@ await _messageBus.Failed(test.Context, private async Task ExecuteTestDirectlyAsync(AbstractExecutableTest test, ITestExecutor executor, ConcurrentDictionary completedTests, CancellationToken cancellationToken) { - var startTime = DateTime.UtcNow; try { await executor.ExecuteTestAsync(test, cancellationToken); @@ -335,19 +303,6 @@ private async Task ExecuteTestDirectlyAsync(AbstractExecutableTest test, ITestEx finally { completedTests[test] = true; - - // Record completion for adaptive metrics - if (_adaptiveController != null) - { - var executionTime = DateTime.UtcNow - startTime; - _adaptiveController.RecordTestCompletion(executionTime); - } } } - - public void Dispose() - { - _adaptiveController?.Dispose(); - _semaphore?.Dispose(); - } } diff --git a/TUnit.Engine/Services/ITestResultFactory.cs b/TUnit.Engine/Services/ITestResultFactory.cs index 375d8cddf50..4e654e89de5 100644 --- a/TUnit.Engine/Services/ITestResultFactory.cs +++ b/TUnit.Engine/Services/ITestResultFactory.cs @@ -11,4 +11,5 @@ internal interface ITestResultFactory TestResult CreateFailedResult(DateTimeOffset startTime, Exception exception); TestResult CreateSkippedResult(DateTimeOffset startTime, string reason); TestResult CreateTimeoutResult(DateTimeOffset startTime, int timeoutMs); + TestResult? CreateCancelledResult(DateTimeOffset testStartTime); } diff --git a/TUnit.Engine/Services/SingleTestExecutor.cs b/TUnit.Engine/Services/SingleTestExecutor.cs index 3eb41e2ed75..339a03e6d4b 100644 --- a/TUnit.Engine/Services/SingleTestExecutor.cs +++ b/TUnit.Engine/Services/SingleTestExecutor.cs @@ -18,13 +18,19 @@ internal class SingleTestExecutor : ISingleTestExecutor private readonly ITestResultFactory _resultFactory; private readonly EventReceiverOrchestrator _eventReceiverOrchestrator; private readonly IHookCollectionService _hookCollectionService; + private readonly EngineCancellationToken _engineCancellationToken; private SessionUid _sessionUid; - public SingleTestExecutor(TUnitFrameworkLogger logger, EventReceiverOrchestrator eventReceiverOrchestrator, IHookCollectionService hookCollectionService, SessionUid sessionUid) + public SingleTestExecutor(TUnitFrameworkLogger logger, + EventReceiverOrchestrator eventReceiverOrchestrator, + IHookCollectionService hookCollectionService, + EngineCancellationToken engineCancellationToken, + SessionUid sessionUid) { _logger = logger; _eventReceiverOrchestrator = eventReceiverOrchestrator; _hookCollectionService = hookCollectionService; + _engineCancellationToken = engineCancellationToken; _sessionUid = sessionUid; _resultFactory = new TestResultFactory(); } @@ -74,95 +80,102 @@ private async Task ExecuteTestInternalAsync( test.StartTime = DateTimeOffset.Now; test.State = TestState.Running; - if (!string.IsNullOrEmpty(test.Context.SkipReason)) - { - return await HandleSkippedTestInternalAsync(test, cancellationToken); - } + if (!string.IsNullOrEmpty(test.Context.SkipReason)) + { + return await HandleSkippedTestInternalAsync(test, cancellationToken); + } - if (test.Context.TestDetails.ClassInstance is SkippedTestInstance) - { - return await HandleSkippedTestInternalAsync(test, cancellationToken); - } + if (test.Context.TestDetails.ClassInstance is SkippedTestInstance) + { + return await HandleSkippedTestInternalAsync(test, cancellationToken); + } - if (test.Context.TestDetails.ClassInstance is PlaceholderInstance) - { - var createdInstance = await test.CreateInstanceAsync(); - if (createdInstance == null) + if (test.Context.TestDetails.ClassInstance is PlaceholderInstance) + { + var createdInstance = await test.CreateInstanceAsync(); + if (createdInstance == null) + { + throw new InvalidOperationException($"CreateInstanceAsync returned null for test {test.Context.GetDisplayName()}. This is likely a framework bug."); + } + test.Context.TestDetails.ClassInstance = createdInstance; + } + + var instance = test.Context.TestDetails.ClassInstance; + + if (instance == null) { - throw new InvalidOperationException($"CreateInstanceAsync returned null for test {test.Context.GetDisplayName()}. This is likely a framework bug."); + throw new InvalidOperationException( + $"Test instance is null for test {test.Context.GetDisplayName()} after instance creation. ClassInstance type: {test.Context.TestDetails.ClassInstance?.GetType()?.Name ?? "null"}"); } - test.Context.TestDetails.ClassInstance = createdInstance; - } - var instance = test.Context.TestDetails.ClassInstance; - - if (instance == null) - { - throw new InvalidOperationException($"Test instance is null for test {test.Context.GetDisplayName()} after instance creation. ClassInstance type: {test.Context.TestDetails.ClassInstance?.GetType()?.Name ?? "null"}"); - } - - if (instance is PlaceholderInstance) - { - throw new InvalidOperationException($"Test instance is still PlaceholderInstance for test {test.Context.GetDisplayName()}. This should have been replaced."); - } + if (instance is PlaceholderInstance) + { + throw new InvalidOperationException($"Test instance is still PlaceholderInstance for test {test.Context.GetDisplayName()}. This should have been replaced."); + } - await PropertyInjectionService.InjectPropertiesIntoArgumentsAsync(test.ClassArguments, test.Context.ObjectBag, test.Context.TestDetails.MethodMetadata, test.Context.Events); - await PropertyInjectionService.InjectPropertiesIntoArgumentsAsync(test.Arguments, test.Context.ObjectBag, test.Context.TestDetails.MethodMetadata, test.Context.Events); + await PropertyInjectionService.InjectPropertiesIntoArgumentsAsync(test.ClassArguments, test.Context.ObjectBag, test.Context.TestDetails.MethodMetadata, + test.Context.Events); + await PropertyInjectionService.InjectPropertiesIntoArgumentsAsync(test.Arguments, test.Context.ObjectBag, test.Context.TestDetails.MethodMetadata, + test.Context.Events); - await PropertyInjectionService.InjectPropertiesAsync( - test.Context, - instance, - test.Metadata.PropertyDataSources, - test.Metadata.PropertyInjections, - test.Metadata.MethodMetadata, - test.Context.TestDetails.TestId); + await PropertyInjectionService.InjectPropertiesAsync( + test.Context, + instance, + test.Metadata.PropertyDataSources, + test.Metadata.PropertyInjections, + test.Metadata.MethodMetadata, + test.Context.TestDetails.TestId); - await _eventReceiverOrchestrator.InitializeAllEligibleObjectsAsync(test.Context, cancellationToken); + await _eventReceiverOrchestrator.InitializeAllEligibleObjectsAsync(test.Context, cancellationToken); - CheckDependenciesAndThrowIfShouldSkip(test); + CheckDependenciesAndThrowIfShouldSkip(test); - var classContext = test.Context.ClassContext; - var assemblyContext = classContext.AssemblyContext; - var sessionContext = assemblyContext.TestSessionContext; + var classContext = test.Context.ClassContext; + var assemblyContext = classContext.AssemblyContext; + var sessionContext = assemblyContext.TestSessionContext; - await _eventReceiverOrchestrator.InvokeFirstTestInSessionEventReceiversAsync(test.Context, sessionContext, cancellationToken); + await _eventReceiverOrchestrator.InvokeFirstTestInSessionEventReceiversAsync(test.Context, sessionContext, cancellationToken); - await _eventReceiverOrchestrator.InvokeFirstTestInAssemblyEventReceiversAsync(test.Context, assemblyContext, cancellationToken); + await _eventReceiverOrchestrator.InvokeFirstTestInAssemblyEventReceiversAsync(test.Context, assemblyContext, cancellationToken); - await _eventReceiverOrchestrator.InvokeFirstTestInClassEventReceiversAsync(test.Context, classContext, cancellationToken); - await _eventReceiverOrchestrator.InvokeTestStartEventReceiversAsync(test.Context, cancellationToken); + await _eventReceiverOrchestrator.InvokeFirstTestInClassEventReceiversAsync(test.Context, classContext, cancellationToken); + await _eventReceiverOrchestrator.InvokeTestStartEventReceiversAsync(test.Context, cancellationToken); - try - { - if (!string.IsNullOrEmpty(test.Context.SkipReason)) + try + { + if (!string.IsNullOrEmpty(test.Context.SkipReason)) + { + return await HandleSkippedTestInternalAsync(test, cancellationToken); + } + + if (test.Context is { RetryFunc: not null, TestDetails.RetryLimit: > 0 }) + { + await ExecuteTestWithRetries(() => ExecuteTestWithHooksAsync(test, instance, cancellationToken), test.Context, cancellationToken); + } + else + { + await ExecuteTestWithHooksAsync(test, instance, cancellationToken); + } + } + catch (TestDependencyException e) { + test.Context.SkipReason = e.Message; return await HandleSkippedTestInternalAsync(test, cancellationToken); } - - if(test.Context is { RetryFunc: not null, TestDetails.RetryLimit: > 0 }) + catch (Exception exception) when (_engineCancellationToken.Token.IsCancellationRequested && exception is OperationCanceledException or TaskCanceledException) { - await ExecuteTestWithRetries(() => ExecuteTestWithHooksAsync(test, instance, cancellationToken), test.Context, cancellationToken); + HandleCancellation(test); } - else + catch (Exception ex) { - await ExecuteTestWithHooksAsync(test, instance, cancellationToken); + HandleTestFailure(test, ex); } - } - catch (TestDependencyException e) - { - test.Context.SkipReason = e.Message; - return await HandleSkippedTestInternalAsync(test, cancellationToken); - } - catch (Exception ex) - { - HandleTestFailure(test, ex); - } - finally - { - test.EndTime = DateTimeOffset.Now; + finally + { + test.EndTime = DateTimeOffset.Now; - await _eventReceiverOrchestrator.InvokeTestEndEventReceiversAsync(test.Context!, cancellationToken); - } + await _eventReceiverOrchestrator.InvokeTestEndEventReceiversAsync(test.Context!, cancellationToken); + } if (test.Result == null) { @@ -371,6 +384,11 @@ private void HandleTestFailure(AbstractExecutableTest test, Exception ex) } } + private void HandleCancellation(AbstractExecutableTest test) + { + test.State = TestState.Cancelled; + test.Result = _resultFactory.CreateCancelledResult(test.StartTime!.Value); + } private TestNodeUpdateMessage CreateUpdateMessage(AbstractExecutableTest test) { diff --git a/TUnit.Engine/Services/SystemMetricsCollector.cs b/TUnit.Engine/Services/SystemMetricsCollector.cs deleted file mode 100644 index 214355b2edf..00000000000 --- a/TUnit.Engine/Services/SystemMetricsCollector.cs +++ /dev/null @@ -1,143 +0,0 @@ -using System.Diagnostics; - -namespace TUnit.Engine.Services; - -/// -/// Collects system metrics for adaptive parallelism -/// -internal sealed class SystemMetricsCollector : IDisposable -{ - private readonly Process _currentProcess; - private readonly Timer _gcMemoryTimer; - private long _lastGcMemory; - private DateTime _lastCpuTime; - private TimeSpan _lastTotalProcessorTime; - private double _lastCpuUsage; - - public SystemMetricsCollector() - { - _currentProcess = Process.GetCurrentProcess(); - _lastCpuTime = DateTime.UtcNow; - _lastTotalProcessorTime = _currentProcess.TotalProcessorTime; - - // Update GC memory periodically to avoid blocking - _gcMemoryTimer = new Timer(_ => _lastGcMemory = GC.GetTotalMemory(false), null, TimeSpan.Zero, TimeSpan.FromSeconds(1)); - } - - /// - /// Gets current system metrics snapshot - /// - public SystemMetrics GetMetrics() - { - var now = DateTime.UtcNow; - var currentTotalProcessorTime = _currentProcess.TotalProcessorTime; - var timeDelta = (now - _lastCpuTime).TotalMilliseconds; - - double processCpuUsage = 0; - if (timeDelta > 0) - { - var cpuTimeDelta = (currentTotalProcessorTime - _lastTotalProcessorTime).TotalMilliseconds; - processCpuUsage = (cpuTimeDelta / timeDelta) / Environment.ProcessorCount * 100; - _lastCpuUsage = processCpuUsage; - } - else - { - processCpuUsage = _lastCpuUsage; - } - - _lastCpuTime = now; - _lastTotalProcessorTime = currentTotalProcessorTime; - - // Thread pool statistics - ThreadPool.GetAvailableThreads(out var workerThreads, out var ioThreads); - ThreadPool.GetMaxThreads(out var maxWorkerThreads, out var maxIoThreads); - long pendingWorkItems = 0; -#if NET5_0_OR_GREATER - pendingWorkItems = ThreadPool.PendingWorkItemCount; -#endif - - // Memory metrics - var totalMemory = GC.GetTotalMemory(false); - var gen0Collections = GC.CollectionCount(0); - var gen1Collections = GC.CollectionCount(1); - var gen2Collections = GC.CollectionCount(2); - - return new SystemMetrics - { - ProcessCpuUsagePercent = processCpuUsage, - SystemCpuUsagePercent = processCpuUsage, // Use process CPU as approximation - AvailableWorkerThreads = workerThreads, - AvailableIoThreads = ioThreads, - MaxWorkerThreads = maxWorkerThreads, - MaxIoThreads = maxIoThreads, - PendingWorkItems = pendingWorkItems, - TotalMemoryBytes = totalMemory, - Gen0Collections = gen0Collections, - Gen1Collections = gen1Collections, - Gen2Collections = gen2Collections, - Timestamp = now - }; - } - - /// - /// Detects if the thread pool is experiencing starvation - /// - public bool IsThreadPoolStarved() - { - ThreadPool.GetAvailableThreads(out var workerThreads, out var ioThreads); - ThreadPool.GetMaxThreads(out var maxWorkerThreads, out var maxIoThreads); - - // Consider starved if less than 10% of threads available - var workerUtilization = 1.0 - (double)workerThreads / maxWorkerThreads; - var ioUtilization = 1.0 - (double)ioThreads / maxIoThreads; - - bool hasPendingWork = false; -#if NET5_0_OR_GREATER - hasPendingWork = ThreadPool.PendingWorkItemCount > 100; -#endif - return workerUtilization > 0.9 || ioUtilization > 0.9 || hasPendingWork; - } - - /// - /// Detects memory pressure - /// - public bool IsMemoryPressureHigh() - { - // Simple heuristic: check if we're using more than 90% of max generation size - // or if Gen2 collections are happening frequently - var totalMemory = GC.GetTotalMemory(false); - var gen2Size = GC.GetGeneration(new object()) == 2 ? totalMemory : 0; - - // Check if memory is growing rapidly - var memoryGrowth = totalMemory - _lastGcMemory; - _lastGcMemory = totalMemory; - - return totalMemory > 1_000_000_000 || // Over 1GB - memoryGrowth > 100_000_000; // Growing by more than 100MB/sec - } - - public void Dispose() - { - _gcMemoryTimer?.Dispose(); - _currentProcess?.Dispose(); - } -} - -/// -/// System metrics snapshot -/// -internal sealed class SystemMetrics -{ - public double ProcessCpuUsagePercent { get; init; } - public double SystemCpuUsagePercent { get; init; } - public int AvailableWorkerThreads { get; init; } - public int AvailableIoThreads { get; init; } - public int MaxWorkerThreads { get; init; } - public int MaxIoThreads { get; init; } - public long PendingWorkItems { get; init; } - public long TotalMemoryBytes { get; init; } - public int Gen0Collections { get; init; } - public int Gen1Collections { get; init; } - public int Gen2Collections { get; init; } - public DateTime Timestamp { get; init; } -} \ No newline at end of file diff --git a/TUnit.Engine/Services/TestResultFactory.cs b/TUnit.Engine/Services/TestResultFactory.cs index 3a024f50858..c6cfc7f375a 100644 --- a/TUnit.Engine/Services/TestResultFactory.cs +++ b/TUnit.Engine/Services/TestResultFactory.cs @@ -61,4 +61,19 @@ public TestResult CreateTimeoutResult(DateTimeOffset startTime, int timeoutMs) OverrideReason = $"Test exceeded timeout of {timeoutMs}ms" }; } + + public TestResult? CreateCancelledResult(DateTimeOffset startTime) + { + var endTime = DateTimeOffset.Now; + + return new TestResult + { + State = TestState.Cancelled, + Start = startTime, + End = endTime, + Duration = endTime - startTime, + Exception = null, + ComputerName = Environment.MachineName + }; + } } diff --git a/TUnit.TestProject/NotInParallelClassGroupingTests.cs b/TUnit.TestProject/NotInParallelClassGroupingTests.cs new file mode 100644 index 00000000000..869c6bc7499 --- /dev/null +++ b/TUnit.TestProject/NotInParallelClassGroupingTests.cs @@ -0,0 +1,139 @@ +using System.Collections.Concurrent; +using TUnit.TestProject.Attributes; + +namespace TUnit.TestProject; + +// This test verifies that NotInParallel tests are grouped by class +// and executed sequentially within each class before moving to the next class + +// Test classes for NotInParallel grouping +[NotInParallel] +[EngineTest(ExpectedResult.Pass)] +public class NotInParallelClassGroupingTests_ClassA +{ + internal static readonly ConcurrentQueue ExecutionOrder = new(); + + [Test, NotInParallel(Order = 1)] + public async Task Test1() + { + ExecutionOrder.Enqueue($"ClassA.Test1"); + await Task.Delay(10); + } + + [Test, NotInParallel(Order = 2)] + public async Task Test2() + { + ExecutionOrder.Enqueue($"ClassA.Test2"); + await Task.Delay(10); + } + + [Test, NotInParallel(Order = 3)] + public async Task Test3() + { + ExecutionOrder.Enqueue($"ClassA.Test3"); + await Task.Delay(10); + } +} + +[NotInParallel] +[EngineTest(ExpectedResult.Pass)] +public class NotInParallelClassGroupingTests_ClassB +{ + [Test, NotInParallel(Order = 1)] + public async Task Test1() + { + NotInParallelClassGroupingTests_ClassA.ExecutionOrder.Enqueue($"ClassB.Test1"); + await Task.Delay(10); + } + + [Test, NotInParallel(Order = 2)] + public async Task Test2() + { + NotInParallelClassGroupingTests_ClassA.ExecutionOrder.Enqueue($"ClassB.Test2"); + await Task.Delay(10); + } +} + +[NotInParallel] +[EngineTest(ExpectedResult.Pass)] +public class NotInParallelClassGroupingTests_ClassC +{ + [Test, NotInParallel(Order = 1)] + public async Task Test1() + { + NotInParallelClassGroupingTests_ClassA.ExecutionOrder.Enqueue($"ClassC.Test1"); + await Task.Delay(10); + } + + [Test, NotInParallel(Order = 2)] + public async Task Test2() + { + NotInParallelClassGroupingTests_ClassA.ExecutionOrder.Enqueue($"ClassC.Test2"); + await Task.Delay(10); + } + + [Test, NotInParallel(Order = 3)] + public async Task Test3() + { + NotInParallelClassGroupingTests_ClassA.ExecutionOrder.Enqueue($"ClassC.Test3"); + await Task.Delay(10); + } +} + +// Verification test that runs last +[EngineTest(ExpectedResult.Pass)] +public class NotInParallelClassGroupingTests_Verify +{ + [Test, NotInParallel(Order = int.MaxValue)] + public async Task VerifyClassGrouping() + { + // Allow time for all tests to complete + await Task.Delay(200); + + var order = NotInParallelClassGroupingTests_ClassA.ExecutionOrder.ToList(); + + // We should have 8 test executions (3 from ClassA, 2 from ClassB, 3 from ClassC) + await Assert.That(order).HasCount(8); + + // Verify that all tests from one class complete before another class starts + var classSequence = new List(); + string? lastClass = null; + + foreach (var execution in order) + { + var className = execution.Split('.')[0]; + if (className != lastClass) + { + classSequence.Add(className); + lastClass = className; + } + } + + // Each class should appear exactly once in the sequence + // (meaning no interleaving of classes) + await Assert.That(classSequence.Distinct().Count()).IsEqualTo(3); + await Assert.That(classSequence).HasCount(3); + + // Verify test order within each class + var classATests = order.Where(o => o.StartsWith("ClassA.")).ToList(); + var classBTests = order.Where(o => o.StartsWith("ClassB.")).ToList(); + var classCTests = order.Where(o => o.StartsWith("ClassC.")).ToList(); + + // Check ClassA test order + await Assert.That(classATests).HasCount(3); + await Assert.That(classATests[0]).IsEqualTo("ClassA.Test1"); + await Assert.That(classATests[1]).IsEqualTo("ClassA.Test2"); + await Assert.That(classATests[2]).IsEqualTo("ClassA.Test3"); + + // Check ClassB test order + await Assert.That(classBTests).HasCount(2); + await Assert.That(classBTests[0]).IsEqualTo("ClassB.Test1"); + await Assert.That(classBTests[1]).IsEqualTo("ClassB.Test2"); + + // Check ClassC test order + await Assert.That(classCTests).HasCount(3); + await Assert.That(classCTests[0]).IsEqualTo("ClassC.Test1"); + await Assert.That(classCTests[1]).IsEqualTo("ClassC.Test2"); + await Assert.That(classCTests[2]).IsEqualTo("ClassC.Test3"); + } +} \ No newline at end of file