diff --git a/src/ParallelTestRunner.jl b/src/ParallelTestRunner.jl index 45f8f3b..4eee57d 100644 --- a/src/ParallelTestRunner.jl +++ b/src/ParallelTestRunner.jl @@ -1480,29 +1480,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 1037322..385e992 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -1257,6 +1257,48 @@ 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 + try + ParallelTestRunner._runtests( + ParallelTestRunner, parse_args(["--jobs=1"]); + testsuite, + tests=["s1", "s2"], + serial=["s1", "s2"], + test_worker, + stdout=devnull, + stderr=devnull, + ) + # 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.