fix waitall deadlock if any errors occur (#60030)

When errors occur, `waitall` may skip allocating Channel producers,
leading to deadlock in the subsequent loop in the event that the user
asked it to failfast (ironically). This is seen often in the failing of
the threads_exec test ever since the test was added for this call.
Simplify this to just use separate loops for the wait and the return
computation.

(cherry picked from commit e2f3178)

Author: vtjnash

base/task.jl

@@ -383,8 +383,11 @@ completed tasks, and the other consists of uncompleted tasks.
     each runs serially, since this needs to scan the list of `tasks` each time and
     synchronize with each one every time this is called. Or consider using
     [`waitall(tasks; failfast=true)`](@ref waitall) instead.
+
+!!! compat "Julia 1.12"
+    This function requires at least Julia 1.12.
 """
-waitany(tasks; throw=true) = _wait_multiple(tasks, throw)
+waitany(tasks; throw=true) = _wait_multiple(collect_tasks(tasks), throw)
 
 """
     waitall(tasks; failfast=true, throw=true) -> (done_tasks, remaining_tasks)
@@ -400,17 +403,22 @@ given tasks is finished by exception. If `throw` is `true`, throw
 
 The return value consists of two task vectors. The first one consists of
 completed tasks, and the other consists of uncompleted tasks.
+
+!!! compat "Julia 1.12"
+    This function requires at least Julia 1.12.
 """
-waitall(tasks; failfast=true, throw=true) = _wait_multiple(tasks, throw, true, failfast)
+waitall(tasks; failfast=true, throw=true) = _wait_multiple(collect_tasks(tasks), throw, true, failfast)
 
-function _wait_multiple(waiting_tasks, throwexc=false, all=false, failfast=false)
+function collect_tasks(waiting_tasks)
     tasks = Task[]
-
     for t in waiting_tasks
         t isa Task || error("Expected an iterator of `Task` object")
         push!(tasks, t)
     end
+    return tasks
+end
 
+function _wait_multiple(tasks::Vector{Task}, throwexc::Bool=false, all::Bool=false, failfast::Bool=false)
     if (all && !failfast) || length(tasks) <= 1
         exception = false
         # Force everything to finish synchronously for the case of waitall
@@ -474,22 +482,36 @@ function _wait_multiple(waiting_tasks, throwexc=false, all=false, failfast=false
     end
 
     while nremaining > 0
+        exception && failfast && break
         i = take!(chan)
         t = tasks[i]
         waiter_tasks[i] = sentinel
         done_mask[i] = true
         exception |= istaskfailed(t)
         nremaining -= 1
-
-        # stop early if requested, unless there is something immediately
-        # ready to consume from the channel (using a race-y check)
-        if (!all || (failfast && exception)) && !isready(chan)
-            break
-        end
+        # stop early if requested
+        all || break
     end
 
     close(chan)
 
+    # now just read which tasks finished directly: the channel is not needed anymore for that
+    # repeat until we get (acquire) the list of all dependent-exited tasks
+    changed = true
+    while changed
+        changed = false
+        for (i, done) in enumerate(done_mask)
+            done && continue
+            t = tasks[i]
+            if istaskdone(t)
+                done_mask[i] = true
+                exception |= istaskfailed(t)
+                nremaining -= 1
+                changed = true
+            end
+        end
+    end
+
     if nremaining == 0
         if throwexc && exception
             exceptions = [TaskFailedException(t) for t in tasks if istaskfailed(t)]
@@ -500,6 +522,7 @@ function _wait_multiple(waiting_tasks, throwexc=false, all=false, failfast=false
         remaining_mask = .~done_mask
         for i in findall(remaining_mask)
             waiter = waiter_tasks[i]
+            waiter === sentinel && continue
             donenotify = tasks[i].donenotify::ThreadSynchronizer
             @lock donenotify list_deletefirst!(donenotify.waitq, waiter)
         end

test/threads_exec.jl

@@ -1372,9 +1372,7 @@ end
                 tasks = [Threads.@spawn(div(1, i)) for i = 0:1]
                 wait(tasks[1]; throw=false)
                 wait(tasks[2]; throw=false)
-                @test_throws CompositeException begin
-                    waitall(Threads.@spawn(div(1, i)) for i = 0:1)
-                end
+                @test_throws CompositeException waitall(tasks)
             end
         end
     end