From e3007d68bebb11f1fb4ce97b57f3e3de44951698 Mon Sep 17 00:00:00 2001 From: Abdulmalik Alquwayfili Date: Wed, 7 Oct 2026 14:16:03 +0300 Subject: [PATCH] Sch: Propagate a failure to dependents before deleting the failed task Both failure paths (`finish_task!` and `set_failed!`) called `delete_unused_task!` and only then `schedule_dependents!`. Each dependent builds its own `DTaskFailedException` from the upstream error with `load_result(state, origin)`. If nothing else referenced the upstream task, the delete had just cleared its result, so `load_result` hit AssertionError: Thunk[1] is not yet finished on the scheduler task, and the eager scheduler exited. Every task then in flight failed with `SchedulingException (Scheduler exited)`. This needs a failed task whose `DTask` is no longer referenced and that still has dependents, which is what the intermediate tasks of a chain or of a tree reduction are. Propagate first, then delete. --- src/sch/Sch.jl | 9 +++++---- src/sch/util.jl | 4 +++- test/thunk.jl | 12 ++++++++++++ 3 files changed, 20 insertions(+), 5 deletions(-) diff --git a/src/sch/Sch.jl b/src/sch/Sch.jl index 299462f9f..2700fbb6d 100644 --- a/src/sch/Sch.jl +++ b/src/sch/Sch.jl @@ -1162,8 +1162,10 @@ function finish_task!(ctx, state, node, thunk_failed, ready::Vector{Thunk}) # permanently unresolved. Handle the three steps directly instead. fill_registered_futures!(state, node, true) node.sch_accessible = false - delete_unused_task!(state, node) + # Dependents read this error with load_result, so propagate before + # the result can be deleted schedule_dependents!(state, node, true, ready) + delete_unused_task!(state, node) else # Success path: seal dependents (collecting newly-ready ones into # `ready` for the caller to schedule outside state.lock), fulfill @@ -1197,9 +1199,8 @@ function delete_unused_task!(state, thunk; recycle::Bool=false) end # `recycle=true` returns the Thunk to the global pool. Only callers for whom # the futures AND dependents Treiber lists are already sealed-and-drained may -# pass it (success-path finish_task! and unref_thunk!): the failure paths run -# schedule_dependents! AFTER deletion, and a recycled thunk's re-SEALED -# dependents head would silently skip failure propagation. +# pass it (success-path finish_task! and unref_thunk!): a recycled thunk's +# re-SEALED dependents head would silently skip failure propagation. function task_delete!(state, thunk; recycle::Bool=false) clear_result!(state, thunk) @atomic thunk.valid = false diff --git a/src/sch/util.jl b/src/sch/util.jl index d1e07439f..3334c2011 100644 --- a/src/sch/util.jl +++ b/src/sch/util.jl @@ -346,7 +346,6 @@ function set_failed!(state, origin::Thunk, thunk::Thunk=origin; ex=nothing) fill_registered_futures!(state, thunk, true) thunk.sch_accessible = false - delete_unused_task!(state, thunk) # Seal the dependents list and propagate failure transitively. # schedule_dependents! with failed=true immediately calls set_failed! for @@ -358,6 +357,9 @@ function set_failed!(state, origin::Thunk, thunk::Thunk=origin; ex=nothing) failed_ready = Thunk[] schedule_dependents!(state, thunk, true, failed_ready) @assert isempty(failed_ready) "set_failed! produced ready thunks on the failure path" + + # Only now: each dependent above read this thunk's error with load_result + delete_unused_task!(state, thunk) end "Internal utility, useful for debugging scheduler state." diff --git a/test/thunk.jl b/test/thunk.jl index a0489793c..f1490170b 100644 --- a/test/thunk.jl +++ b/test/thunk.jl @@ -271,6 +271,18 @@ end c = @spawn b+2 @test_throws_unwrap (Dagger.DTaskFailedException, ErrorException) fetch(c) end + @testset "dependent chain, only the last task referenced" begin + # The upstream tasks can be deleted as soon as they fail + function chain() + s = @spawn sleep(1) + a = @spawn (s->error("Test"))(s) + b = @spawn a+1 + return @spawn b+2 + end + c = chain() + GC.gc(); GC.gc() + @test_throws_unwrap (Dagger.DTaskFailedException, ErrorException) fetch(c) + end @testset "single input" begin a = @spawn 1+1 b = @spawn (a->error("Test"))(a)