Warn when the pre-test init time becomes much slower than a worker restart

docs/src/advanced.md

@@ -225,11 +225,25 @@ and a yellow one marks a result that may still be replaced.
     well together: recycling keeps one bad test from cascading onto its worker during the run,
     while retries give the tests that did fail a contention-free second chance.
 
-## Memory Pressure on macOS
+## Choosing the Number of Jobs
 
-On memory-constrained macOS machines (notably CI runners), requesting more jobs than the default can make the test suite take much longer than expected, sometimes enough to time out the job.
-This often manifests as per-test init times (shown with `--verbose`) steadily increasing over the run, likely because macOS compresses memory under pressure and each garbage collection gets slower. GC % being higer than usual can also be an indication that you're requesting too many jobs or that the max RSS threshold is too high.
-Prefer the default `--jobs` value, which accounts for available memory, and lower the `JULIA_TEST_MAXRSS_MB` environment variable so that workers get recycled sooner. See [issue #124](https://github.com/JuliaTesting/ParallelTestRunner.jl/issues/124) for more details.
+By default, ParallelTestRunner.jl runs with one worker per 2 GiB of available memory, capped by the number of effective CPU threads.
+For less demanding test suites, pass a smaller `memory_per_worker` (in bytes) to [`runtests`](@ref) to get more parallelism; for memory-hungry ones, pass a larger value.
+Since it only adjusts the default, the number of jobs still adapts to the memory of each machine running the tests.
+
+```julia
+runtests(MyPackage, ARGS; memory_per_worker = 2^30) # 1 GiB per worker
+```
+
+Workers are recycled once they exceed the `JULIA_TEST_MAXRSS_MB` environment variable (or the `max_worker_rss` keyword).
+Lowering it keeps memory pressure in check, which can make it possible to run more workers than would otherwise be safe.
+
+The number of jobs can also be forced with `--jobs=N` or the `PTR_NUM_JOBS` environment variable, but this bypasses the memory check entirely.
+
+More jobs is not always faster: once the machine runs out of memory the whole suite slows down, sometimes enough to time out a CI job.
+This has been especially problematic on low-memory macOS machines (notably CI runners), where memory compression makes each garbage collection slower.
+If you notice per-test init times (shown with `--verbose`) steadily increasing over the run and/or a higher-than-usual GC %, it might be worth tweaking the job selection behaviour.
+See [issue #124](https://github.com/JuliaTesting/ParallelTestRunner.jl/issues/124) for more details.
 
 ## Custom Workers
 
@@ -363,4 +377,4 @@ function jltest {
 
 1. **Only use `retries` for worker contention-related failures**: Not all intermittent failures are caused by parallel worker resource contention. Ensure you aren't masking real test failures when using this feature.
 
-1. **Don't request too many jobs on low-memory macOS machines**: tests can take much longer than expected; see [Memory Pressure on macOS](@ref).
+1. **Don't request too many jobs on low-memory machines**: tests can take much longer than expected, especially on macOS; see [Choosing the Number of Jobs](@ref).

docs/src/index.md

@@ -107,6 +107,7 @@ Pkg.test("MyPackage"; test_args=`--verbose --jobs=4 integration`)
 
 Tests run concurrently in isolated worker processes, each inside own module.
 `ParallelTestRunner` records historical tests duration for each package, so that in subsequent runs long-running tests are executed first, to improve load balancing.
+See [Choosing the Number of Jobs](@ref) for how the number of workers is chosen.
 
 ### Serial Test Support
 
@@ -130,6 +131,7 @@ See [Failure Handling](@ref) in the advanced usage guide for details.
 The test runner provides real-time output showing:
 - Test name and worker assignment, with the worker shown in yellow when it is about to be recycled
 - Execution time
+- Init time (with `--verbose`), i.e. the time spent before the test started
 - GC time and percentage
 - Memory allocation
 - RSS (Resident Set Size) memory usage, shown in yellow once it exceeds the RSS threshold

src/ParallelTestRunner.jl

@@ -160,6 +160,14 @@ function memory_usage(rec::AbstractTestRecord)
     return parent(rec).rss
 end
 
+function init_time(rec::AbstractTestRecord)
+    base = parent(rec)
+    return base.total_time - base.time
+end
+
+# the user is warned once a warm worker's init time exceeds this multiple of the cold-start cost
+const SLOW_INIT_FACTOR = 2
+
 function Base.getindex(rec::AbstractTestRecord)
     return parent(rec).value
 end
@@ -211,14 +219,14 @@ function print_header(::Type{<:AbstractTestRecord}, ctx::TestIOContext, testgrou
         name_pad_str = " "^(ctx.name_align + textwidth(testgroupheader) - 3) * " │ "
         init_str = ctx.verbose ? "   Init   │" : ""
         compile_str = VERSION >= v"1.11" && ctx.verbose ? " Compile │" : ""
-        header_top_str = styled"{ptr_default:$name_pad_str  Test   │$init_str$compile_str ──────────────── CPU ──────────────── │}\n"
+        header_top_str = "$name_pad_str  Test   │$init_str$compile_str ──────────────── CPU ──────────────── │\n"
         print(ctx.stdout, header_top_str)
 
         # header bottom
         workerheaderstr = lpad(workerheader, ctx.name_align - textwidth(testgroupheader) + 1)
         init_time_str = ctx.verbose ? " time (s) │" : ""
         comp_time_str = VERSION >= v"1.11" && ctx.verbose ? "   (%)   │" : ""
-        bottom_header_str = styled"{ptr_default:$testgroupheader$workerheaderstr │ time (s) │$init_time_str$comp_time_str GC (s) │ GC % │ Alloc (MB) │ RSS (MB) │}\n"
+        bottom_header_str = "$testgroupheader$workerheaderstr │ time (s) │$init_time_str$comp_time_str GC (s) │ GC % │ Alloc (MB) │ RSS (MB) │\n"
         print(ctx.stdout, bottom_header_str)
         flush(ctx.stdout)
     finally
@@ -230,7 +238,9 @@ function print_test_started(::Type{<:AbstractTestRecord}, wrkr, test, ctx::TestI
     lock(ctx.lock)
     try
         padded_wrkr = lpad("($wrkr)", ctx.name_align - textwidth(test) + 1, " ")
-        out_str = styled"{ptr_default:$(test)$padded_wrkr │}{ptr_light:$(\" \"^ctx.elapsed_align) started at $(now())}\n"
+        # StyledStrings 1.0.3 (used on Julia 1.10) cannot print a styled string whose leading
+        # unstyled text ends in a multi-byte character (e.g. `│`), so style it explicitly
+        out_str = styled"{default:$(test)$padded_wrkr │}{ptr_light:$(\" \"^ctx.elapsed_align) started at $(now())}\n"
         print(ctx.stdout, out_str)
         flush(ctx.stdout)
     finally
@@ -243,7 +253,7 @@ function print_test_finished(record::AbstractTestRecord, wrkr, test, ctx::TestIO
     lock(ctx.lock)
     try
         padded_wrkr = lpad("($wrkr)", ctx.name_align - textwidth(test) + 1, " ")
-        wrkr_face = ctx.recycled[] ? :ptr_warn : :ptr_default
+        wrkr_face = ctx.recycled[] ? :ptr_warn : :default
 
         time_str = @sprintf("%7.2f", base.time)
         padded_time = lpad(time_str, ctx.elapsed_align, " ")
@@ -275,11 +285,12 @@ function print_test_finished(record::AbstractTestRecord, wrkr, test, ctx::TestIO
         padded_alloc = lpad(alloc_str, ctx.alloc_align, " ")
 
         mem_use = memory_usage(record)
-        mem_face = mem_use > ctx.max_worker_rss ? :ptr_warn : :ptr_default
+        mem_face = mem_use > ctx.max_worker_rss ? :ptr_warn : :default
         rss_str = @sprintf("%5.2f", mem_use / 2^20)
         padded_rss = lpad(rss_str, ctx.rss_align, " ")
 
-        out_str = styled"{ptr_default:$test{$wrkr_face:$padded_wrkr} │ $padded_time │ $padded_init_time$padded_comp_time$padded_gc │ $padded_percent │ $padded_alloc │ {$mem_face:$padded_rss} │\n}"
+        # see `print_test_started` for why `test` is styled explicitly
+        out_str = styled"{default:$test}{$wrkr_face:$padded_wrkr} │ $padded_time │ $padded_init_time$padded_comp_time$padded_gc │ $padded_percent │ $padded_alloc │ {$mem_face:$padded_rss} │\n"
         print(ctx.stdout, out_str)
         flush(ctx.stdout)
     finally
@@ -533,13 +544,11 @@ end
 const DEFAULT_MEMORY_PER_WORKER = 2 * Int64(2)^30
 
 """
-    default_njobs(; memory_per_worker = 2 * 2^30,
-                    _cpu_threads = $(@static isdefined(Sys, :EFFECTIVE_CPU_THREADS) ? Symbol("Sys.EFFECTIVE_CPU_THREADS") : Symbol("Sys.CPU_THREADS")),
-                    _free_memory = ParallelTestRunner.available_memory())
+    default_njobs(; memory_per_worker = 2 * 2^30)
 
-*Internal* function used to determine the default number of parallel jobs. Calculated as the number of CPU threads,
-clamped such that each worker can be assumed to use `memory_per_worker` bytes of the
-available system memory.
+A function used to determine the default number of parallel jobs. Calculated as the
+number of CPU threads, clamped such that each worker can be assumed to use `memory_per_worker`
+bytes of the available system memory.
 """
 function default_njobs(;
         memory_per_worker = DEFAULT_MEMORY_PER_WORKER,
@@ -1089,7 +1098,7 @@ function runtests(mod::Module, args::ParsedArgs;
         for test in sorted_tests
             failed = test in historical_failures
             bullet = failed ? "×" : "-"
-            face = failed ? :ptr_error : :ptr_default
+            face = failed ? :ptr_error : :default
             line = rstrip(" $bullet $(rpad(test, name_align))  $(lpad(duration_strs[test], duration_align))")
             println(stdout, styled"{$face:$line}")
         end
@@ -1184,6 +1193,10 @@ function _runtests(mod::Module, args::ParsedArgs;
     t0 = time()
     results = Lockable([])
     running_tests = Lockable(Dict{String, Float64}())  # test => start_time
+    # init time of a test on a freshly spawned worker, i.e. the cost of recycling one
+    cold_init_time = Threads.Atomic{Float64}(Inf)
+    # the slow init warning is only printed once per run
+    slow_init_warned = Threads.Atomic{Bool}(false)
 
     worker_tasks = Task[]
 
@@ -1317,6 +1330,7 @@ function _runtests(mod::Module, args::ParsedArgs;
     # Message types for the printer channel
     # (:started, test_name, worker_id)
     # (:finished, test_name, worker_id, record, recycled)
+    # (:slow_init, test_name, init_time, cold_init_time)
     # (:crashed, test_name, worker_id, test_time)
     # (:retry, tests_n, retry_n)
     # (:nonpass_face, face)
@@ -1353,6 +1367,22 @@ function _runtests(mod::Module, args::ParsedArgs;
                             print_test_finished(record, wrkr, test_name, io_ctx)
                         end
 
+                    elseif msg_type === :slow_init
+                        test_name, init_t, cold_t = msg[2], msg[3], msg[4]
+
+                        clear_status()
+                        lock(io_ctx.lock)
+                        try
+                            msg_str = styled"""
+                            {ptr_warn,bold:Warning:} {ptr_light:Init time of `$test_name` ({default:$(round(init_t; digits=2))s}) was much longer than usual ({default:$(round(cold_t; digits=2))s} on a freshly spawned worker).
+                            This usually means too many workers for the available memory; see the docs for more information.}
+                            """
+                            print(io_ctx.stderr, msg_str)
+                            flush(io_ctx.stderr)
+                        finally
+                            unlock(io_ctx.lock)
+                        end
+
                     elseif msg_type === :crashed
                         test_name, wrkr = msg[2], msg[3]
 
@@ -1365,7 +1395,7 @@ function _runtests(mod::Module, args::ParsedArgs;
                         clear_status()
                         lock(io_ctx.lock)
                         try
-                            println(io_ctx.stdout, styled"{ptr_default:Retrying $tests_n failed test$(tests_n > 1 ? \"s\" : \" \") ($retry_n)}")
+                            println(io_ctx.stdout, "Retrying $tests_n failed test$(tests_n > 1 ? "s" : " ") ($retry_n)")
                             flush(io_ctx.stdout)
                         finally
                             unlock(io_ctx.lock)
@@ -1474,9 +1504,11 @@ function _runtests(mod::Module, args::ParsedArgs;
                               wrkr = p
                           end
                           # if a worker failed, spawn a new one
+                          fresh_worker = false
                           if wrkr === nothing || !Malt.isrunning(wrkr)
                               wrkr = p = addworker(; init_worker_code, io_ctx.color,
                                                    exename, exeflags, env)
+                              fresh_worker = true
                           end
 
                           # run the test
@@ -1507,13 +1539,23 @@ function _runtests(mod::Module, args::ParsedArgs;
 
                           # act on the results
                           if result isa AbstractTestRecord
+                              if fresh_worker
+                                  Threads.atomic_min!(cold_init_time, init_time(result))
+                              end
+                              # the pre-test full GC has become much slower than spawning a
+                              # new worker, which typically means there are too many workers
+                              # for the available memory (e.g. macOS memory pressure, #124)
+                              slow_init = wrkr === p && !fresh_worker && init_time(result) > SLOW_INIT_FACTOR * cold_init_time[]
                               # recycle a pool worker so future tests start with a smaller working
                               # set, or so that a failing test that may have left the worker in a
                               # bad state (e.g. a wedged GPU driver) cannot poison later tests
                               # (custom workers are stopped after every test regardless)
                               recycle = wrkr === p && (memory_usage(result) > max_worker_rss ||
                                                        ((recycle_on_failure || retry_mode) && anynonpass(result[])))
                               put!(printer_channel, (:finished, test, worker_id(wrkr), result, recycle))
+                              if slow_init && !Threads.atomic_xchg!(slow_init_warned, true)
+                                  put!(printer_channel, (:slow_init, test, init_time(result), cold_init_time[]))
+                              end
                               if anynonpass(result[]) && args.quickfail !== nothing
                                   stop_work()
                                   return
@@ -1662,7 +1704,7 @@ function _runtests(mod::Module, args::ParsedArgs;
             testface = if result isa Exception || anynonpass(result[])
                 :ptr_error
             else
-                :ptr_default
+                :default
             end
             println(io_ctx.stdout, styled"\nOutput generated during execution of '{$testface:$testname}':")
             lines = collect(eachline(IOBuffer(output)))
@@ -1798,7 +1840,6 @@ runtests(mod::Module, ARGS::Array{String}; kwargs...) = runtests(mod, parse_args
 
 # register faces used in printing
 function __init__()
-    addface!(:ptr_default => Face(inherit=:default))
     addface!(:ptr_warn => Face(inherit=:yellow))
     addface!(:ptr_error => Face(inherit=:red))
     addface!(:ptr_light => Face(inherit=:light))

test/workers.jl

@@ -114,6 +114,11 @@ end
 end
 
 # Issue <https://github.com/JuliaTesting/ParallelTestRunner.jl/issues/106>.
+# The cold init time of a worker varies between machines, so the "slow init warning"
+# testset below needs a measurement of it: rather than spending a worker on a
+# dedicated run, take it from the verbose output of the following testset.
+cold_init_time = Ref(NaN)
+
 @testset "default workers reused and stopped at end" begin
     # Use default workers (no test_worker) so the framework creates and should stop them.
     # More tests than workers so that workers are reused, and so that some tasks finish
@@ -152,6 +157,10 @@ end
     @test contains(str, "SUCCESS")
     # Make sure we didn't spawn more workers than expected: the same workers ran all tests.
     @test ParallelTestRunner.ID_COUNTER[] == old_id_counter + njobs
+    # The first test on each worker had a cold init: the slowest init is an upper bound.
+    init_times = [parse(Float64, m[1]) for m in eachmatch(r"^t\d\s+\(\d+\) │\s+[\d.]+ │\s+([\d.]+) │"m, str)]
+    @test length(init_times) == length(testsuite)
+    cold_init_time[] = maximum(init_times)
     if before < 0
         # Counting child PIDs not supported on this platform
         @test_skip false
@@ -195,6 +204,41 @@ end
     @test ParallelTestRunner.ID_COUNTER[] == old_id_counter + length(testsuite)
 end
 
+@testset "slow init warning" begin
+    init_worker_code = :( const slow_init_counter = Ref(0) )
+    args = ["--verbose", "--jobs=1"]
+
+    # `cold_init_time` was measured by the "default workers reused and stopped at end"
+    # testset above; if that testset failed before recording it, measure it here instead
+    # so that this testset stays independent.
+    if isnan(cold_init_time[])
+        io = IOBuffer()
+        runtests(ParallelTestRunner, args; testsuite=Dict("cold" => :( @test true )), init_worker_code, stdout=io, stderr=io)
+        cold_init_time[] = parse(Float64, match(r"^cold\s+\(\d+\) │\s+[\d.]+ │\s+([\d.]+) │"m, String(take!(io)))[1])
+    end
+
+    # every test after the first on the worker is slow to init
+    slow_init = ceil(Int, 1.5 * ParallelTestRunner.SLOW_INIT_FACTOR * cold_init_time[]) + 1
+    init_code = quote
+        Main.slow_init_counter[] += 1
+        Main.slow_init_counter[] > 1 && sleep($slow_init)
+    end
+    testsuite = Dict("a" => :( @test true ), "b" => :( @test true ), "c" => :( @test true ))
+    io = IOBuffer()
+    ioc = IOContext(io, :color => true)
+    old_id_counter = ParallelTestRunner.ID_COUNTER[]
+    runtests(ParallelTestRunner, args; testsuite, init_worker_code, init_code, stdout=ioc, stderr=ioc)
+    str = String(take!(io))
+    @test contains(str, "SUCCESS")
+    # a slow init does not recycle the worker
+    @test ParallelTestRunner.ID_COUNTER[] == old_id_counter + 1
+    # both tests after the first are slow, but the warning is only printed once, and it is
+    # the only thing printed in yellow
+    @test count("Warning:", str) == 1
+    @test count("\e[33m", str) == 1
+    @test contains(str, r"Init time of `[abc]`")
+end
+
 @testset "recycle_on_failure" begin
     # Call `_runtests` so that we can enforce a run order, and use a single job, so that
     # all tests share the same pool slot: a test only gets a new worker if the previous one