Support parallel execution of tasks

twenzel · twenzel · commit 875bedb2f510 · 2022-10-04T10:34:57.000+02:00
diff --git a/src/Cake.Core.Tests/Unit/CakeEngineTests.cs b/src/Cake.Core.Tests/Unit/CakeEngineTests.cs
@@ -1558,6 +1558,118 @@ public async Task Should_Return_Report_That_Marks_Failed_Tasks_As_Failed()
                 Assert.Equal(CakeTaskExecutionStatus.Delegated, report.First(e => e.TaskName == "A").ExecutionStatus);
                 Assert.Equal(CakeTaskExecutionStatus.Failed, report.First(e => e.TaskName == "B").ExecutionStatus);
             }
+
+            [Fact]
+            public async Task Should_Throw_Exception_For_Circular_Dependencies()
+            {
+                // Given
+                var fixture = new CakeEngineFixture();
+                var settings = new ExecutionSettings().SetTarget("B");
+                var engine = fixture.CreateEngine();
+                engine.RegisterTask("B").IsDependentOn("C");
+                engine.RegisterTask("C").IsDependentOn("D");
+                engine.RegisterTask("D").IsDependentOn("B");
+
+                // When
+                var result = await Record.ExceptionAsync(() => engine.RunTargetAsync(fixture.Context, fixture.ExecutionStrategy, settings));
+
+                // Then
+                Assert.IsType<CakeException>(result);
+                Assert.Equal("Graph contains circular references.", result?.Message);
+            }
+
+            [Fact]
+            public async Task Should_Throw_Exception_For_Circular_Dependencies_In_Parallel()
+            {
+                // Given
+                var fixture = new CakeEngineFixture();
+                var settings = new ExecutionSettings().SetTarget("B").RunInParallel();
+                var engine = fixture.CreateEngine();
+                engine.RegisterTask("B").IsDependentOn("C");
+                engine.RegisterTask("C").IsDependentOn("D");
+                engine.RegisterTask("D").IsDependentOn("B");
+
+                // When
+                var result = await Record.ExceptionAsync(() => engine.RunTargetAsync(fixture.Context, fixture.ExecutionStrategy, settings));
+
+                // Then
+                Assert.IsType<CakeException>(result);
+                Assert.Equal("Graph contains circular references.", result?.Message);
+            }
+
+            [Fact]
+            public async Task Should_Execute_Tasks_In_Order_In_Parallel()
+            {
+                // Given
+                var result = new List<string>();
+                var fixture = new CakeEngineFixture();
+                var settings = new ExecutionSettings().SetTarget("E").RunInParallel();
+                var engine = fixture.CreateEngine();
+                engine.RegisterTask("A").Does(() => result.Add("A"));
+                engine.RegisterTask("B").IsDependentOn("A").Does(() => result.Add("B"));
+                engine.RegisterTask("C").IsDependentOn("B").Does(() => result.Add("C"));
+                engine.RegisterTask("D").IsDependentOn("C").IsDependeeOf("E").Does(() => { result.Add("D"); });
+                engine.RegisterTask("E").Does(() => { result.Add("E"); });
+
+                // When
+                await engine.RunTargetAsync(fixture.Context, fixture.ExecutionStrategy, settings);
+
+                // Then
+                Assert.Equal(5, result.Count);
+                Assert.Equal("A", result[0]);
+                Assert.Equal("B", result[1]);
+                Assert.Equal("C", result[2]);
+                Assert.Equal("D", result[3]);
+                Assert.Equal("E", result[4]);
+            }
+
+            [Fact]
+            public async Task Should_Execute_Tasks_In_Parallel()
+            {
+                // Given
+                var result = new List<string>();
+                var fixture = new CakeEngineFixture();
+                var settings = new ExecutionSettings().SetTarget("E").RunInParallel();
+                var engine = fixture.CreateEngine();
+                engine.RegisterTask("A").Does(() => result.Add("A"));
+                engine.RegisterTask("B").IsDependentOn("A").Does(async () => { await Task.Delay(20); result.Add("B"); });
+                engine.RegisterTask("C").IsDependentOn("A").Does(async () => { await Task.Delay(5); result.Add("C"); });
+                engine.RegisterTask("D").IsDependentOn("A").Does(() => result.Add("D"));
+                engine.RegisterTask("E").IsDependentOn("B").IsDependentOn("C").IsDependentOn("D").Does(() => result.Add("E"));
+
+                // When
+                await engine.RunTargetAsync(fixture.Context, fixture.ExecutionStrategy, settings);
+
+                // Then
+                Assert.Equal(5, result.Count);
+                Assert.Equal("A", result[0]);
+                Assert.Equal("D", result[1]);
+                Assert.Equal("C", result[2]);
+                Assert.Equal("B", result[3]);
+                Assert.Equal("E", result[4]);
+            }
+
+            [Fact]
+            public async Task Should_Not_Catch_Exceptions_From_Task_If_ContinueOnError_Is_Not_Set_In_Parallel()
+            {
+                // Given
+                var fixture = new CakeEngineFixture();
+                var settings = new ExecutionSettings().SetTarget("E").RunInParallel();
+                var engine = fixture.CreateEngine();
+                engine.RegisterTask("A");
+                engine.RegisterTask("B").IsDependentOn("A");
+                engine.RegisterTask("C").IsDependentOn("A").Does(() => throw new InvalidOperationException("Whoopsie"));
+                engine.RegisterTask("D").IsDependentOn("A");
+                engine.RegisterTask("E").IsDependentOn("B").IsDependentOn("C").IsDependentOn("D");
+
+                // When
+                var result = await Record.ExceptionAsync(() =>
+                    engine.RunTargetAsync(fixture.Context, fixture.ExecutionStrategy, settings));
+
+                // Then
+                Assert.IsType<InvalidOperationException>(result);
+                Assert.Equal("Whoopsie", result?.Message);
+            }
         }
 
         public sealed class TheSetupEvent
diff --git a/src/Cake.Core/CakeEngine.cs b/src/Cake.Core/CakeEngine.cs
@@ -8,6 +8,7 @@
 using System.Globalization;
 using System.Linq;
 using System.Reflection;
+using System.Threading;
 using System.Threading.Tasks;
 using Cake.Core.Diagnostics;
 using Cake.Core.Graph;
@@ -211,19 +212,48 @@ public async Task<CakeReport> RunTargetAsync(ICakeContext context, IExecutionStr
                 {
                     // Execute only the target task.
                     var task = _tasks.FirstOrDefault(x => x.Name.Equals(settings.Target, StringComparison.OrdinalIgnoreCase));
-                    await RunTask(context, strategy, task, target, stopWatch, report);
+                    await RunTask(context, strategy, task, target, stopWatch, report, null);
+                }
+                else if (settings.Parallel)
+                {
+                    await graph.TraverseAsync(target, async (taskName, cancellationTokenSource) =>
+                    {
+                        if (cancellationTokenSource.IsCancellationRequested)
+                        {
+                            return;
+                        }
+
+                        var task = _tasks.FirstOrDefault(_ => _.Name.Equals(taskName, StringComparison.OrdinalIgnoreCase));
+                        Debug.Assert(task != null, "Node should not be null");
+
+                        var isTarget = task.Name.Equals(target, StringComparison.OrdinalIgnoreCase);
+
+                        await RunTask(context, strategy, task, target, stopWatch, report, cancellationTokenSource);
+                    });
                 }
                 else
                 {
                     // Execute all scheduled tasks.
                     foreach (var task in orderedTasks)
                     {
-                        await RunTask(context, strategy, task, target, stopWatch, report);
+                        await RunTask(context, strategy, task, target, stopWatch, report, null);
                     }
                 }
 
                 return report;
             }
+            catch (TaskCanceledException)
+            {
+                exceptionWasThrown = true;
+                throw;
+            }
+            catch (AggregateException ex)
+            {
+                exceptionWasThrown = true;
+                thrownException = ex.InnerException;
+
+                throw ex.GetBaseException();
+            }
             catch (Exception ex)
             {
                 exceptionWasThrown = true;
@@ -236,7 +266,7 @@ public async Task<CakeReport> RunTargetAsync(ICakeContext context, IExecutionStr
             }
         }
 
-        private async Task RunTask(ICakeContext context, IExecutionStrategy strategy, CakeTask task, string target, Stopwatch stopWatch, CakeReport report)
+        private async Task RunTask(ICakeContext context, IExecutionStrategy strategy, CakeTask task, string target, Stopwatch stopWatch, CakeReport report, CancellationTokenSource cancellationTokenSource)
         {
             // Is this the current target?
             var isTarget = task.Name.Equals(target, StringComparison.OrdinalIgnoreCase);
@@ -255,7 +285,7 @@ private async Task RunTask(ICakeContext context, IExecutionStrategy strategy, Ca
 
             if (!skipped)
             {
-                await ExecuteTaskAsync(context, strategy, stopWatch, task, report).ConfigureAwait(false);
+                await ExecuteTaskAsync(context, strategy, stopWatch, task, report, cancellationTokenSource).ConfigureAwait(false);
             }
         }
 
@@ -309,7 +339,7 @@ private static bool ShouldTaskExecute(ICakeContext context, CakeTask task, CakeT
         }
 
         private async Task ExecuteTaskAsync(ICakeContext context, IExecutionStrategy strategy, Stopwatch stopWatch,
-            CakeTask task, CakeReport report)
+            CakeTask task, CakeReport report, CancellationTokenSource cancellationTokenSource)
         {
             stopWatch.Restart();
 
@@ -321,6 +351,11 @@ private async Task ExecuteTaskAsync(ICakeContext context, IExecutionStrategy str
                 // Execute the task.
                 await strategy.ExecuteAsync(task, context).ConfigureAwait(false);
             }
+            catch (TaskCanceledException exception)
+            {
+                taskException = exception;
+                throw;
+            }
             catch (Exception exception)
             {
                 _log.Error("An error occurred when executing task '{0}'.", task.Name);
@@ -340,6 +375,8 @@ private async Task ExecuteTaskAsync(ICakeContext context, IExecutionStrategy str
                 }
                 else
                 {
+                    cancellationTokenSource?.Cancel();
+
                     // No error handler defined for this task.
                     // Rethrow the exception and let it propagate.
                     throw;
diff --git a/src/Cake.Core/ExecutionSettings.cs b/src/Cake.Core/ExecutionSettings.cs
@@ -19,6 +19,11 @@ public sealed class ExecutionSettings
         /// </summary>
         public bool Exclusive { get; private set; }
 
+        /// <summary>
+        /// Gets a value indicating whether the dependend task of the target should be run in parallel (if possible).
+        /// </summary>
+        public bool Parallel { get; private set; }
+
         /// <summary>
         /// Sets the target to be executed.
         /// </summary>
@@ -39,5 +44,15 @@ public ExecutionSettings UseExclusiveTarget()
             Exclusive = true;
             return this;
         }
+
+        /// <summary>
+        /// Whether or not to run the dependend task in parallel.
+        /// </summary>
+        /// <returns>The same <see cref="ExecutionSettings"/> instance so that multiple calls can be chained.</returns>
+        public ExecutionSettings RunInParallel()
+        {
+            Parallel = true;
+            return this;
+        }
     }
 }
diff --git a/src/Cake.Core/Graph/CakeGraph.cs b/src/Cake.Core/Graph/CakeGraph.cs
@@ -5,6 +5,8 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace Cake.Core.Graph
 {
@@ -112,6 +114,29 @@ public IEnumerable<string> Traverse(string target)
             return result;
         }
 
+        /// <summary>
+        /// Traverses the graph asynchrounus leading to the specified target.
+        /// </summary>
+        /// <param name="target">The target to traverse to.</param>
+        /// <param name="executeTask">Action which will be called on each task.</param>
+        /// <returns>A task to wait for.</returns>
+        public async Task TraverseAsync(string target, Func<string, CancellationTokenSource, Task> executeTask)
+        {
+            if (!Exist(target))
+            {
+                return;
+            }
+
+            if (HasCircularReferences(target))
+            {
+                throw new CakeException("Graph contains circular references.");
+            }
+
+            var cancellationTokenSource = new CancellationTokenSource();
+            var visitedNodes = new Dictionary<string, Task>();
+            await TraverseAsync(target, executeTask, cancellationTokenSource, visitedNodes);
+        }
+
         private void Traverse(string node, ICollection<string> result, ISet<string> visited = null)
         {
             visited = visited ?? new HashSet<string>(StringComparer.OrdinalIgnoreCase);
@@ -130,5 +155,59 @@ private void Traverse(string node, ICollection<string> result, ISet<string> visi
                 throw new CakeException("Graph contains circular references.");
             }
         }
+
+        private async Task TraverseAsync(string node, Func<string, CancellationTokenSource, Task> executeTask,
+            CancellationTokenSource cancellationTokenSource, IDictionary<string, Task> visitedNodes)
+        {
+            if (visitedNodes.ContainsKey(node))
+            {
+                await visitedNodes[node];
+                return;
+            }
+
+            var token = cancellationTokenSource.Token;
+            var dependentTasks = _edges
+                .Where(x => x.End.Equals(node, StringComparison.OrdinalIgnoreCase))
+                .Select(x =>
+                {
+                    var task = TraverseAsync(x.Start, executeTask, cancellationTokenSource, visitedNodes);
+                    visitedNodes[x.Start] = task;
+
+                    if (task.IsFaulted)
+                    {
+                        throw task.Exception;
+                    }
+
+                    return task;
+                })
+                .ToArray();
+
+            if (dependentTasks.Any())
+            {
+                TaskCompletionSource<object> tcs = new TaskCompletionSource<object>();
+                token.Register(() => tcs.TrySetCanceled(), false);
+                await Task.WhenAny(Task.WhenAll(dependentTasks), tcs.Task);
+            }
+
+            await executeTask(node, cancellationTokenSource);
+        }
+
+        private bool HasCircularReferences(string node, Stack<string> visited = null)
+        {
+            visited = visited ?? new Stack<string>();
+
+            if (visited.Contains(node))
+            {
+                return true;
+            }
+
+            visited.Push(node);
+            var hasCircularReference = _edges
+                .Where(x => x.End.Equals(node, StringComparison.OrdinalIgnoreCase))
+                .Any(x => HasCircularReferences(x.Start, visited));
+            visited.Pop();
+
+            return hasCircularReference;
+        }
     }
 }
diff --git a/src/Cake.Frosting/Internal/Commands/DefaultCommand.cs b/src/Cake.Frosting/Internal/Commands/DefaultCommand.cs
@@ -68,6 +68,11 @@ public override int Execute(CommandContext context, DefaultCommandSettings setti
                     runner.Settings.UseExclusiveTarget();
                 }
 
+                if (settings.Parallel)
+                {
+                    runner.Settings.RunInParallel();
+                }
+
                 runner.Run(settings.Target);
             }
             catch (Exception ex)
diff --git a/src/Cake.Frosting/Internal/Commands/DefaultCommandSettings.cs b/src/Cake.Frosting/Internal/Commands/DefaultCommandSettings.cs
@@ -51,5 +51,9 @@ internal sealed class DefaultCommandSettings : CommandSettings
         [CommandOption("--info")]
         [Description("Displays additional information about Cake.")]
         public bool Info { get; set; }
+
+        [CommandOption("--parallel|-p")]
+        [Description("Enables the support for parallel tasks.")]
+        public bool Parallel { get; set; }
     }
 }
diff --git a/src/Cake.Tests/Unit/ProgramTests.cs b/src/Cake.Tests/Unit/ProgramTests.cs
@@ -1,4 +1,3 @@
-using System.Linq;
 using System.Threading.Tasks;
 using Autofac;
 using Cake.Cli;
@@ -32,6 +31,7 @@ public async Task Should_Use_Default_Parameters_By_Default()
                     settings.BuildHostKind == BuildHostKind.Build &&
                     settings.Debug == false &&
                     settings.Exclusive == false &&
+                    settings.Parallel == false &&
                     settings.Script.FullPath == "build.cake" &&
                     settings.Verbosity == Verbosity.Normal &&
                     settings.NoBootstrapping == false));
diff --git a/src/Cake/Commands/DefaultCommand.cs b/src/Cake/Commands/DefaultCommand.cs
@@ -75,6 +75,7 @@ public override int Execute(CommandContext context, DefaultCommandSettings setti
                     Script = settings.Script,
                     Verbosity = settings.Verbosity,
                     Exclusive = settings.Exclusive,
+                    Parallel = settings.Parallel,
                     Debug = settings.Debug,
                     NoBootstrapping = settings.SkipBootstrap,
                 });
diff --git a/src/Cake/Commands/DefaultCommandSettings.cs b/src/Cake/Commands/DefaultCommandSettings.cs
diff --git a/src/Cake/Features/Building/BuildFeature.cs b/src/Cake/Features/Building/BuildFeature.cs
diff --git a/src/Cake/Features/Building/BuildFeatureSettings.cs b/src/Cake/Features/Building/BuildFeatureSettings.cs

Original file line number	Diff line number	Diff line change
`@@ -8,6 +8,7 @@`
`8`	`8`	`using System.Globalization;`
`9`	`9`	`using System.Linq;`
`10`	`10`	`using System.Reflection;`
	`11`	`+using System.Threading;`
`11`	`12`	`using System.Threading.Tasks;`
`12`	`13`	`using Cake.Core.Diagnostics;`
`13`	`14`	`using Cake.Core.Graph;`
`@@ -211,19 +212,48 @@ public async Task<CakeReport> RunTargetAsync(ICakeContext context, IExecutionStr`
`211`	`212`	`{`
`212`	`213`	`// Execute only the target task.`
`213`	`214`	`var task = _tasks.FirstOrDefault(x => x.Name.Equals(settings.Target, StringComparison.OrdinalIgnoreCase));`
`214`		`- await RunTask(context, strategy, task, target, stopWatch, report);`
	`215`	`+ await RunTask(context, strategy, task, target, stopWatch, report, null);`
	`216`	`+ }`
	`217`	`+ else if (settings.Parallel)`
	`218`	`+ {`
	`219`	`+ await graph.TraverseAsync(target, async (taskName, cancellationTokenSource) =>`
	`220`	`+ {`
	`221`	`+ if (cancellationTokenSource.IsCancellationRequested)`
	`222`	`+ {`
	`223`	`+ return;`
	`224`	`+ }`
	`225`	`+`
	`226`	`+ var task = _tasks.FirstOrDefault(_ => _.Name.Equals(taskName, StringComparison.OrdinalIgnoreCase));`
	`227`	`+ Debug.Assert(task != null, "Node should not be null");`
	`228`	`+`
	`229`	`+ var isTarget = task.Name.Equals(target, StringComparison.OrdinalIgnoreCase);`
	`230`	`+`
	`231`	`+ await RunTask(context, strategy, task, target, stopWatch, report, cancellationTokenSource);`
	`232`	`+ });`
`215`	`233`	`}`
`216`	`234`	`else`
`217`	`235`	`{`
`218`	`236`	`// Execute all scheduled tasks.`
`219`	`237`	`foreach (var task in orderedTasks)`
`220`	`238`	`{`
`221`		`- await RunTask(context, strategy, task, target, stopWatch, report);`
	`239`	`+ await RunTask(context, strategy, task, target, stopWatch, report, null);`
`222`	`240`	`}`
`223`	`241`	`}`
`224`	`242`
`225`	`243`	`return report;`
`226`	`244`	`}`
	`245`	`+ catch (TaskCanceledException)`
	`246`	`+ {`
	`247`	`+ exceptionWasThrown = true;`
	`248`	`+ throw;`
	`249`	`+ }`
	`250`	`+ catch (AggregateException ex)`
	`251`	`+ {`
	`252`	`+ exceptionWasThrown = true;`
	`253`	`+ thrownException = ex.InnerException;`
	`254`	`+`
	`255`	`+ throw ex.GetBaseException();`
	`256`	`+ }`
`227`	`257`	`catch (Exception ex)`
`228`	`258`	`{`
`229`	`259`	`exceptionWasThrown = true;`
`@@ -236,7 +266,7 @@ public async Task<CakeReport> RunTargetAsync(ICakeContext context, IExecutionStr`
`236`	`266`	`}`
`237`	`267`	`}`
`238`	`268`
`239`		`- private async Task RunTask(ICakeContext context, IExecutionStrategy strategy, CakeTask task, string target, Stopwatch stopWatch, CakeReport report)`
	`269`	`+ private async Task RunTask(ICakeContext context, IExecutionStrategy strategy, CakeTask task, string target, Stopwatch stopWatch, CakeReport report, CancellationTokenSource cancellationTokenSource)`
`240`	`270`	`{`
`241`	`271`	`// Is this the current target?`
`242`	`272`	`var isTarget = task.Name.Equals(target, StringComparison.OrdinalIgnoreCase);`
`@@ -255,7 +285,7 @@ private async Task RunTask(ICakeContext context, IExecutionStrategy strategy, Ca`
`255`	`285`
`256`	`286`	`if (!skipped)`
`257`	`287`	`{`
`258`		`- await ExecuteTaskAsync(context, strategy, stopWatch, task, report).ConfigureAwait(false);`
	`288`	`+ await ExecuteTaskAsync(context, strategy, stopWatch, task, report, cancellationTokenSource).ConfigureAwait(false);`
`259`	`289`	`}`
`260`	`290`	`}`
`261`	`291`
`@@ -309,7 +339,7 @@ private static bool ShouldTaskExecute(ICakeContext context, CakeTask task, CakeT`
`309`	`339`	`}`
`310`	`340`
`311`	`341`	`private async Task ExecuteTaskAsync(ICakeContext context, IExecutionStrategy strategy, Stopwatch stopWatch,`
`312`		`- CakeTask task, CakeReport report)`
	`342`	`+ CakeTask task, CakeReport report, CancellationTokenSource cancellationTokenSource)`
`313`	`343`	`{`
`314`	`344`	`stopWatch.Restart();`
`315`	`345`
`@@ -321,6 +351,11 @@ private async Task ExecuteTaskAsync(ICakeContext context, IExecutionStrategy str`
`321`	`351`	`// Execute the task.`
`322`	`352`	`await strategy.ExecuteAsync(task, context).ConfigureAwait(false);`
`323`	`353`	`}`
	`354`	`+ catch (TaskCanceledException exception)`
	`355`	`+ {`
	`356`	`+ taskException = exception;`
	`357`	`+ throw;`
	`358`	`+ }`
`324`	`359`	`catch (Exception exception)`
`325`	`360`	`{`
`326`	`361`	`_log.Error("An error occurred when executing task '{0}'.", task.Name);`
`@@ -340,6 +375,8 @@ private async Task ExecuteTaskAsync(ICakeContext context, IExecutionStrategy str`
`340`	`375`	`}`
`341`	`376`	`else`
`342`	`377`	`{`
	`378`	`+ cancellationTokenSource?.Cancel();`
	`379`	`+`
`343`	`380`	`// No error handler defined for this task.`
`344`	`381`	`// Rethrow the exception and let it propagate.`
`345`	`382`	`throw;`
Original file line number	Diff line number	Diff line change
`@@ -68,6 +68,11 @@ public override int Execute(CommandContext context, DefaultCommandSettings setti`
`68`	`68`	`runner.Settings.UseExclusiveTarget();`
`69`	`69`	`}`
`70`	`70`
	`71`	`+ if (settings.Parallel)`
	`72`	`+ {`
	`73`	`+ runner.Settings.RunInParallel();`
	`74`	`+ }`
	`75`	`+`
`71`	`76`	`runner.Run(settings.Target);`
`72`	`77`	`}`
`73`	`78`	`catch (Exception ex)`
Original file line number	Diff line number	Diff line change
`@@ -51,5 +51,9 @@ internal sealed class DefaultCommandSettings : CommandSettings`
`51`	`51`	`[CommandOption("--info")]`
`52`	`52`	`[Description("Displays additional information about Cake.")]`
`53`	`53`	`public bool Info { get; set; }`
	`54`	`+`
	`55`	`+ [CommandOption("--parallel\|-p")]`
	`56`	`+ [Description("Enables the support for parallel tasks.")]`
	`57`	`+ public bool Parallel { get; set; }`
`54`	`58`	`}`
`55`	`59`	`}`