+semver:minor - Refactor Not in Parallel Exclusive Locks (#488)

* Refactor Not in Parallel Exclusive Locks

* Formatting Markdown

* Correctly release semaphore lock

* CancellationTokenSource

* Order keys

* Fix waiting for multiple not-in-parallel constraint keys

---------

Co-authored-by: Tom Longhurst <[email protected]>

Author: thomhurst

src/ModularPipelines.Build/ReleaseNotes.md

@@ -1 +1 @@
-*   Refactor how modules wait for their dependencies
+*   Refactor grabbing exclusive locks for Not-In-Parallel modules

src/ModularPipelines/Engine/IModuleExecutor.cs

@@ -5,6 +5,4 @@ namespace ModularPipelines.Engine;
 internal interface IModuleExecutor
 {
     Task<IEnumerable<ModuleBase>> ExecuteAsync(IReadOnlyList<ModuleBase> modules);
-
-    Task<ModuleBase> ExecuteAsync(ModuleBase module);
 }

src/ModularPipelines/Engine/ModuleExecutor.cs

@@ -20,170 +20,174 @@ internal class ModuleExecutor : IModuleExecutor
     private readonly IModuleDisposer _moduleDisposer;
     private readonly IEnumerable<ModuleBase> _allModules;
     private readonly IExceptionContainer _exceptionContainer;
+    private readonly ILogger<ModuleExecutor> _logger;
 
     private readonly ConcurrentDictionary<ModuleBase, Task<ModuleBase>> _moduleExecutionTasks = new();
+    private readonly object _moduleDictionaryLock = new();
 
+    private readonly ConcurrentDictionary<string, Semaphore> _notInParallelKeyedLocks = new();
+    private readonly object _notInParallelDictionaryLock = new();
+    
     public ModuleExecutor(IPipelineSetupExecutor pipelineSetupExecutor,
         IOptions<PipelineOptions> pipelineOptions,
         ISafeModuleEstimatedTimeProvider moduleEstimatedTimeProvider,
         IModuleDisposer moduleDisposer,
         IEnumerable<ModuleBase> allModules,
-        IExceptionContainer exceptionContainer)
+        IExceptionContainer exceptionContainer,
+        ILogger<ModuleExecutor> logger)
     {
         _pipelineSetupExecutor = pipelineSetupExecutor;
         _pipelineOptions = pipelineOptions;
         _moduleEstimatedTimeProvider = moduleEstimatedTimeProvider;
         _moduleDisposer = moduleDisposer;
         _allModules = allModules;
         _exceptionContainer = exceptionContainer;
+        _logger = logger;
     }
 
     public async Task<IEnumerable<ModuleBase>> ExecuteAsync(IReadOnlyList<ModuleBase> modules)
     {
-        var moduleResults = new List<ModuleBase>();
-
-        var nonParallelModules = modules
-            .Where(x => x.GetType().GetCustomAttribute<NotInParallelAttribute>() != null)
-            .OrderBy(x => x.DependentModules.Count)
-            .ToList();
+        try
+        {
+            var nonParallelModules = modules
+                .Where(x => x.GetType().GetCustomAttribute<NotInParallelAttribute>() != null)
+                .OrderBy(x => x.DependentModules.Count)
+                .ToList();
 
-        var unKeyedNonParallelModules = nonParallelModules
-            .Where(x => x.GetType().GetCustomAttribute<NotInParallelAttribute>()!.ConstraintKeys.Length == 0)
-            .ToList();
+            var unKeyedNonParallelModules = nonParallelModules
+                .Where(x => x.GetType().GetCustomAttribute<NotInParallelAttribute>()!.ConstraintKeys.Length == 0)
+                .ToList();
 
-        foreach (var nonParallelModule in unKeyedNonParallelModules)
-        {
-            moduleResults.Add(await ExecuteAsync(nonParallelModule));
-        }
+            foreach (var nonParallelModule in unKeyedNonParallelModules)
+            {
+                await StartModule(nonParallelModule);
+            }
 
-        var keyedNonParallelModules = nonParallelModules
-            .Where(x => x.GetType().GetCustomAttribute<NotInParallelAttribute>()!.ConstraintKeys.Length != 0)
-            .ToList();
+            var keyedNonParallelModules = nonParallelModules
+                .Where(x => x.GetType().GetCustomAttribute<NotInParallelAttribute>()!.ConstraintKeys.Length != 0)
+                .ToList();
+            
+            await ProcessKeyedNonParallelModules(keyedNonParallelModules.ToList());
 
-        moduleResults.AddRange(
-            await ProcessKeyedNonParallelModules(keyedNonParallelModules.ToList(), moduleResults)
-        );
+            var parallelModuleTasks = modules.Except(nonParallelModules)
+                .Select(StartModule)
+                .ToArray();
 
-        var parallelModuleTasks = modules.Except(nonParallelModules)
-            .Select(ExecuteAsync)
-            .ToArray();
+            if (_pipelineOptions.Value.ExecutionMode == ExecutionMode.StopOnFirstException)
+            {
+                await parallelModuleTasks.WhenAllFailFast();
+            }
+            else
+            {
+                await Task.WhenAll(parallelModuleTasks);
+            }
 
-        if (_pipelineOptions.Value.ExecutionMode == ExecutionMode.StopOnFirstException)
-        {
-            moduleResults.AddRange(await parallelModuleTasks.WhenAllFailFast());
+            return modules;
         }
-        else
+        catch
         {
-            moduleResults.AddRange(await Task.WhenAll(parallelModuleTasks));
-        }
+            foreach (var moduleBase in modules.Where(x => x.ModuleRunType == ModuleRunType.AlwaysRun))
+            {
+                try
+                {
+                    await StartModule(moduleBase);
+                }
+                catch
+                {
+                    // Ignored
+                }
+            }
 
-        return moduleResults;
+            throw;
+        }
     }
 
-    public async Task<ModuleBase> ExecuteAsync(ModuleBase module)
+    private Semaphore GetLockForKey(string key)
     {
-        try
+        lock (_notInParallelDictionaryLock)
         {
-            return await StartModule(module);
-        }
-        catch (TaskCanceledException)
-        {
-            // If the pipeline failed, sometimes a TaskCanceledException can throw before the original exception
-            // So delay a bit to let the original exception throw first
-            await Task.Delay(TimeSpan.FromMilliseconds(500));
-            throw;
+            return _notInParallelKeyedLocks.GetOrAdd(key, _ => new Semaphore(1, 1));
         }
     }
 
-    private async Task<ModuleBase[]> ProcessKeyedNonParallelModules(List<ModuleBase> keyedNonParallelModules,
-        List<ModuleBase> moduleResults)
+    private async Task ProcessKeyedNonParallelModules(List<ModuleBase> keyedNonParallelModules)
     {
-        var currentlyExecutingByKeysLock = new object();
-        var currentlyExecutingByKeys = new List<(string[] Keys, Task)>();
-
-        var executing = new List<Task<ModuleBase>>();
-
-        while (keyedNonParallelModules.Count > 0)
-        {
-            // Reversing allows us to remove from the collection
-            for (var i = keyedNonParallelModules.Count - 1; i >= 0; i--)
+        await keyedNonParallelModules
+            .OrderBy(x => x.GetType().GetCustomAttribute<NotInParallelAttribute>()!.ConstraintKeys.Length)
+            .ForEachAsync(async module =>
             {
-                var module = keyedNonParallelModules[i];
-
-                var notInParallelKeys =
-                    module.GetType().GetCustomAttribute<NotInParallelAttribute>()!.ConstraintKeys;
-
-                lock (currentlyExecutingByKeysLock)
+                var keys = module.GetType()
+                    .GetCustomAttribute<NotInParallelAttribute>()!
+                    .ConstraintKeys
+                    .OrderBy(x => x)
+                    .ToArray();
+                
+                _logger.LogDebug("Grabbing not in parallel locks for keys {Keys}", string.Join(", ", keys));
+                
+                var locks = keys.Select(GetLockForKey).ToArray();
+                
+                while (!WaitHandle.WaitAll(locks, TimeSpan.FromMilliseconds(100), false))
                 {
-                    if (currentlyExecutingByKeys.Any(x => x.Keys.Intersect(notInParallelKeys).Any()))
-                    {
-                        // There are currently executing tasks with that same 
-                        continue;
-                    }
+                    await Task.Delay(TimeSpan.FromMilliseconds(500));
                 }
 
-                // Remove from collection as we're now processing it
-                keyedNonParallelModules.RemoveAt(i);
-
-                var executionTask = ExecuteAsync(module);
-
-                var tuple = (notInParallelKeys, executionTask);
-
-                lock (currentlyExecutingByKeysLock)
+                try
                 {
-                    currentlyExecutingByKeys.Add(tuple);
+                    await StartModule(module);
                 }
-
-                _ = executionTask.ContinueWith(_ =>
+                finally
                 {
-                    lock (currentlyExecutingByKeysLock)
+                    foreach (var semaphore in locks)
                     {
-                        return currentlyExecutingByKeys.Remove(tuple);
+                        semaphore.Release();
                     }
-                });
-
-                executing.Add(executionTask);
-            }
-        }
-
-        return await Task.WhenAll(executing);
+                }
+            })
+            .ProcessInParallel();
     }
 
-    private async Task<ModuleBase> StartModule(ModuleBase module)
+    private Task<ModuleBase> StartModule(ModuleBase module)
     {
-        return await _moduleExecutionTasks.GetOrAdd(module, async @base =>
+        lock (_moduleDictionaryLock)
         {
-            var dependencies = module.GetModuleDependencies();
-
-            foreach (var dependency in dependencies)
+            return _moduleExecutionTasks.GetOrAdd(module, async @base =>
             {
-                await StartDependency(module, dependency.DependencyType, dependency.IgnoreIfNotRegistered);
-            }
+                _logger.LogDebug("Starting Module {Module}", module.GetType().Name);
 
-            try
-            {
-                await _pipelineSetupExecutor.OnBeforeModuleStartAsync(module);
+                var dependencies = module.GetModuleDependencies();
 
-                await module.StartInternal();
+                foreach (var dependency in dependencies)
+                {
+                    await StartDependency(module, dependency.DependencyType, dependency.IgnoreIfNotRegistered);
+                }
 
-                await _moduleEstimatedTimeProvider.SaveModuleTimeAsync(module.GetType(), module.Duration);
+                try
+                {
+                    await _pipelineSetupExecutor.OnBeforeModuleStartAsync(module);
 
-                await _pipelineSetupExecutor.OnAfterModuleEndAsync(module);
+                    await module.StartInternal();
 
-                return module;
-            }
-            finally
-            {
-                if (!_pipelineOptions.Value.ShowProgressInConsole)
+                    await _moduleEstimatedTimeProvider.SaveModuleTimeAsync(module.GetType(), module.Duration);
+
+                    await _pipelineSetupExecutor.OnAfterModuleEndAsync(module);
+
+                    return module;
+                }
+                finally
                 {
-                    await _moduleDisposer.DisposeAsync(module);
+                    if (!_pipelineOptions.Value.ShowProgressInConsole)
+                    {
+                        await _moduleDisposer.DisposeAsync(module);
+                    }
                 }
-            }
-        });
+            });
+        }
     }
 
     private async Task StartDependency(ModuleBase requestingModule, Type dependencyType, bool ignoreIfNotRegistered)
     {
+        _logger.LogDebug("Starting Dependency {Dependency} for Module {Module}", dependencyType.Name, requestingModule.GetType().Name);
+        
         var module = _allModules.FirstOrDefault(x => x.GetType() == dependencyType);
 
         if (module is null && ignoreIfNotRegistered)