diff --git a/docs/src/advanced.md b/docs/src/advanced.md index 8cd8fbb8..3ad324eb 100644 --- a/docs/src/advanced.md +++ b/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). diff --git a/docs/src/index.md b/docs/src/index.md index eef97fec..ebd3c0f1 100644 --- a/docs/src/index.md +++ b/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 diff --git a/src/ParallelTestRunner.jl b/src/ParallelTestRunner.jl index e7f10cf3..9fc5a750 100644 --- a/src/ParallelTestRunner.jl +++ b/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,6 +1539,13 @@ 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 @@ -1514,6 +1553,9 @@ function _runtests(mod::Module, args::ParsedArgs; 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)) diff --git a/test/workers.jl b/test/workers.jl index 9297bee6..92d00018 100644 --- a/test/workers.jl +++ b/test/workers.jl @@ -114,6 +114,11 @@ end end # Issue . +# 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