Add throw keyword argument in waitall

JuliaLang · vtjnash · Mar 11, 2024 · Jan 22, 2024 · Feb 4, 2024 · Feb 8, 2024
commit f1f400e705b598af8ca9a2df0509342f9e858a12
diff --git a/base/task.jl b/base/task.jl
@@ -367,9 +367,9 @@ end
 
 # Wait multiple tasks
 waitany(tasks) = _wait_multiple(tasks)
-waitall(tasks; failfast=false) = _wait_multiple(tasks, true, failfast)
+waitall(tasks; failfast=false, throw=false) = _wait_multiple(tasks, true, failfast, throw)
 
-function _wait_multiple(waiting_tasks, all=false, failfast=false)
+function _wait_multiple(waiting_tasks, all=false, failfast=false, throwexc=false)
     tasks = Task[]
 
     for t in waiting_tasks
@@ -378,12 +378,19 @@ function _wait_multiple(waiting_tasks, all=false, failfast=false)
     end
 
     if all && !failfast
+        exception = false
         # Force everything to finish synchronously for the case of waitall
         # with failfast=false
         for t in tasks
             _wait(t)
+            exception |= istaskfailed(t)
+        end
+        if exception && throwexc
+            exceptions = [t.exception for t in tasks if istaskfailed(t)]
+            throw(CompositeException(exceptions))
+        else
+            return tasks, Task[]
         end
-        return tasks, Task[]
     end
 
     exception = false
@@ -402,7 +409,12 @@ function _wait_multiple(waiting_tasks, all=false, failfast=false)
     if nremaining == 0
         return tasks, Task[]
     elseif any(done_mask) && (!all || (failfast && exception))
-        return tasks[done_mask], tasks[.~done_mask]
+        if throwexc && failfast && exception
+            exceptions = [t.exception for t in tasks[done_mask] if istaskfailed(t)]
+            throw(CompositeException(exceptions))
+        else
+            return tasks[done_mask], tasks[.~done_mask]
+        end
     end
 
     chan = Channel{Int}(Inf)
@@ -451,7 +463,13 @@ function _wait_multiple(waiting_tasks, all=false, failfast=false)
             donenotify = tasks[i].donenotify::ThreadSynchronizer
             @lock donenotify Base.list_deletefirst!(donenotify.waitq, waiter)
         end
-        return tasks[done_mask], tasks[remaining_mask]
+        done_tasks = tasks[done_mask]
+        if throwexc && exception
+            exceptions = [t.exception for t in done_tasks if istaskfailed(t)]
+            throw(CompositeException(exceptions))
+        else
+            return done_tasks, tasks[remaining_mask]
+        end
     end
 end
 

diff --git a/test/threads_exec.jl b/test/threads_exec.jl
@@ -1210,6 +1210,12 @@ end
         return tasks, event
     end
 
+    function teardown(tasks, event)
+        notify(event)
+        yield()
+        wait(tasks[3])
+    end
+
     for tasks_type in (Vector{Task}, Set{Task}, Tuple{Task})
         @testset "waitany" begin
             tasks, event = create_tasks()
@@ -1220,9 +1226,7 @@ end
             @test tasks[2] ∈ done
             @test length(pending) == 1
             @test tasks[3] ∈ pending
-            notify(event)
-            yield()
-            wait(tasks[3])
+            teardown(tasks, event)
         end
 
         @testset "waitall" begin
@@ -1257,9 +1261,36 @@ end
                 @test length(pending) == 1
                 @test tasks[3] ∈ pending
 
+                teardown(tasks, event)
+            end
+
+            @testset "throw=true" begin
+                tasks, event = create_tasks()
+                push!(tasks, Threads.@spawn error("Error"))
+
+                sleep(0.1)
                 notify(event)
-                yield()
-                wait(tasks[3])
+
+                @test_throws CompositeException begin
+                    waitall(convert_tasks(tasks_type, tasks); throw=true)
+                end
+
+                @test all(istaskdone.(tasks))
+
+                teardown(tasks, event)
+            end
+
+            @testset "failfast=true and throw=true" begin
+                tasks, event = create_tasks()
+                push!(tasks, Threads.@spawn error("Error"))
+
+                @test_throws CompositeException begin
+                    waitall(convert_tasks(tasks_type, tasks); failfast=true, throw=true)
+                end
+
+                @test !istaskdone(tasks[3])
+
+                teardown(tasks, event)
             end
         end
     end