From 850353c629209c4ad8486b76c10ddc3c4262bdea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mos=C3=A8=20Giordano?= Date: Sat, 8 Aug 2026 21:21:49 +0100 Subject: [PATCH 1/2] Stop worker processes even when finalization rethrows a task failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Finalization waited on the printer task and the worker tasks — rethrowing any non-interrupt failure (e.g. a throwing `test_worker` hook) — before stopping the workers left in the pool, so a propagated failure leaked live worker processes for the rest of the parent session. A failure during a serial phase additionally stranded the shared serial worker in its Ref, which even the pool cleanup would have missed, since returning it to the pool is skipped when the phase's `@sync` throws. Run the worker cleanup in a `finally` block and also stop a stranded serial worker there. Co-Authored-By: Claude Fable 5 --- src/ParallelTestRunner.jl | 46 ++++++++++++++++++++++++--------------- test/runtests.jl | 43 ++++++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 18 deletions(-) diff --git a/src/ParallelTestRunner.jl b/src/ParallelTestRunner.jl index 3bd5491..4913c87 100644 --- a/src/ParallelTestRunner.jl +++ b/src/ParallelTestRunner.jl @@ -1458,29 +1458,39 @@ function _runtests(mod::Module, args::ParsedArgs; # finalization # - # wait for the printer to finish so that all results have been printed - close(printer_channel) - wait(printer_task) + try + # wait for the printer to finish so that all results have been printed + close(printer_channel) + wait(printer_task) + + # wait for worker tasks to catch unhandled exceptions + for task in worker_tasks + try + wait(task) + catch err + # unwrap TaskFailedException + while isa(err, TaskFailedException) + err = current_exceptions(err.task)[1].exception + end - # wait for worker tasks to catch unhandled exceptions - for task in worker_tasks - try - wait(task) - catch err - # unwrap TaskFailedException - while isa(err, TaskFailedException) - err = current_exceptions(err.task)[1].exception + isa(err, InterruptException) || rethrow() end + end + finally + # clean up remaining workers even when a worker or printer task failed and + # its exception is propagated, so worker processes don't outlive the run - isa(err, InterruptException) || rethrow() + # a failure during a serial phase can leave the shared worker in its Ref + # instead of the pool + if serial_worker[] !== nothing && Malt.isrunning(serial_worker[]) + Malt.stop(serial_worker[]) end - end - # clean up remaining workers in the pool - close(worker_pool) - for p in worker_pool - if p !== nothing && Malt.isrunning(p) - Malt.stop(p) + close(worker_pool) + for p in worker_pool + if p !== nothing && Malt.isrunning(p) + Malt.stop(p) + end end end diff --git a/test/runtests.jl b/test/runtests.jl index c79d79c..b0a4048 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -1207,6 +1207,49 @@ end @test ParallelTestRunner.ID_COUNTER[] == old_id_counter + jobs + 1 end + @testset "workers stopped when a task fails" begin + # a `test_worker` hook throwing during the serial phase strands the shared + # worker outside the pool while the task failure is rethrown from + # finalization; the worker process must still be stopped + before = _count_child_pids() + if before < 0 + # counting child PIDs not supported on this platform + @test_skip false + else + testsuite = Dict( + "s1" => :( @test true ), + "s2" => :( @test true ), + ) + exception = ErrorException("test_worker exploded") + test_worker(name) = name == "s2" ? throw(exception) : nothing + io = IOBuffer() + try + ParallelTestRunner._runtests( + ParallelTestRunner, parse_args(["--jobs=1"]); + testsuite, + tests=["s1", "s2"], + serial=["s1", "s2"], + test_worker, + stdout=io, + stderr=io, + ) + # the error must propagate out of `_runtests` + @test false + catch e + @test typeof(e) === TaskFailedException + @test first(Base.current_exceptions(e.task)).exception == exception + end + # allow a moment for worker processes to exit + after = -1 + for _ in 1:50 + sleep(0.1) + after = _count_child_pids() + after >= 0 && after <= before && break + end + @test after == before + end + end + @testset "quickfail in serial phase before parallel" begin # The failing serial test runs first: the remaining serial tests and the whole # parallel batch should never be started. From 7a7453e437b937fd7e13a675a157556406c28729 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mos=C3=A8=20Giordano?= Date: Sat, 8 Aug 2026 22:04:55 +0100 Subject: [PATCH 2/2] Replace useless `IOBuffer()` with `devnull` --- test/runtests.jl | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/test/runtests.jl b/test/runtests.jl index cd3c846..385e992 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -1272,7 +1272,6 @@ end ) exception = ErrorException("test_worker exploded") test_worker(name) = name == "s2" ? throw(exception) : nothing - io = IOBuffer() try ParallelTestRunner._runtests( ParallelTestRunner, parse_args(["--jobs=1"]); @@ -1280,8 +1279,8 @@ end tests=["s1", "s2"], serial=["s1", "s2"], test_worker, - stdout=io, - stderr=io, + stdout=devnull, + stderr=devnull, ) # the error must propagate out of `_runtests` @test false