diff --git a/Project.toml b/Project.toml index f6afbd4..e973fbd 100644 --- a/Project.toml +++ b/Project.toml @@ -1,6 +1,6 @@ name = "SweepRunner" uuid = "be946ad2-3cb3-4b6e-8f7e-4a5ecc3c255b" -version = "0.6.7" +version = "0.6.8" authors = ["sota shimozono "] [deps] @@ -23,8 +23,9 @@ Aqua = "0.8" # there, and `save!` returning the digest it wrote (with `mark_done!(…; result)`) in 0.8.3. `artifact!` / `ArtifactBusy` # (what the deferral and `artifact_affinity` are built for) came in 0.8.2. The owner-stamped # `.running` API (`new_owner_token`, `running_owner`, and the three-argument `refresh_running!` / -# `clear_running!`) that the liveness reaper is built on arrived in 0.8.1. -DataVault = "0.8.4" +# `clear_running!`) that the liveness reaper is built on arrived in 0.8.1. 0.8.9 is the floor now: +# the key heartbeat is `start_heartbeat`'s child process, and a commit is `mark_done!(…, owner)`. +DataVault = "0.8.9" Dates = "1.11" Distributed = "1.11" JLD2 = "0.6" diff --git a/docs/src/architecture.md b/docs/src/architecture.md index b0cc750..cc46fcd 100644 --- a/docs/src/architecture.md +++ b/docs/src/architecture.md @@ -79,8 +79,10 @@ When you call `run!(work_fn, vault, keys)`, it does: — another master may have finished this key between our manifest read and lock acquisition. - Call `work_fn(key)` up to `opts.max_attempts` times, with a heartbeat - task refreshing `.running`. On success, `DataVault.save!` + - `DataVault.mark_done!` + `Manifest.add_complete!`. + child process (`DataVault.start_heartbeat`) refreshing `.running`. On + success, `DataVault.save!` + `DataVault.mark_done!(vault, key, owner)` + + `Manifest.add_complete!`; the owner form commits nothing if a sibling + reclaimed the key meanwhile. 5. `save_manifest(manifest)`. 6. Emit `:stage_done` and return the aggregate counts. @@ -100,11 +102,12 @@ moves on — no blocking, no waiting, no central queue. Two things keep this robust against crashes: -1. **Heartbeat + stale reclaim.** The live holder touches `heartbeat` on - a timer. If a holder dies, its `heartbeat` mtime stops advancing; the - next contender sees the lock as stale (by either heartbeat age or lock-dir - age, depending on whether the holder made it past initial write), and - reclaims via `mv lock lock.dead.X` + `rm -rf lock.dead.X`. +1. **Heartbeat + stale reclaim.** A child process of the live holder + rewrites the lock's `heartbeat_unix=` every `heartbeat_interval`, for as + long as the holder's pid lives — also while `work_fn` never yields, which + an in-process task did not survive. If the holder dies, the heartbeat + stops; after `stale_after` the next contender reclaims the lock under + DataVault's reclaim mutex (see DataVault's README, "ロックの規約"). 2. **Post-lock `is_done` re-check.** Even on the happy path, two masters can start the loop with overlapping `todo`. The re-check inside the locked critical section ensures the second master notices the work diff --git a/src/Run.jl b/src/Run.jl index ebbe96e..88d9f3e 100644 --- a/src/Run.jl +++ b/src/Run.jl @@ -31,9 +31,11 @@ Execution options for [`run!`](@ref). retry (a failed `work_fn` is logged as `:error` instead of `:gave_up`). - `stale_after::Float64 = 600.0` — seconds before another master can reclaim a held lock as stale. Passed through to `DataVault.acquire_running!`. -- `heartbeat_interval::Float64 = 60.0` — how often the per-lock heartbeat task - refreshes DataVault's `.running` file. Enforced to be `<` `stale_after` - (otherwise a live holder's lock could be reclaimed mid-work). +- `heartbeat_interval::Float64 = 60.0` — how often the per-lock heartbeat + (a child process, `DataVault.start_heartbeat`) refreshes DataVault's + `.running` file. It keeps beating while `work_fn` computes without yielding. + Enforced to be `<` `stale_after` (otherwise a live holder's lock could be + reclaimed mid-work). - `log_level::Symbol = :info` — event-log verbosity. At `:info` (default) the high-churn per-key `:lock_busy` and `:key_start` events are suppressed (their totals still ride in the `:stage_done` summary), keeping the JSONL log @@ -415,7 +417,7 @@ end Execute the per-key pipeline: atomic-acquire via `DataVault.acquire_running!`, re-check completion, run work_fn with a -background heartbeat task, and release on exit. Returns a +heartbeat child process, commit owner-checked, and release on exit. Returns a `(key, outcome)` pair suitable for aggregation by the caller. Outcome symbols: @@ -472,60 +474,25 @@ function _run_one_with_lock!( return (key, :already_done) end - # Background heartbeat task: periodically refresh `.running` so - # `DataVault.cleanup_stale` / sibling `acquire_running!` callers - # recognise us as alive. The `Threads.Atomic{Bool}` flag makes - # the stop signal thread-safe; a short sleep tick keeps finally - # cleanup responsive (the earlier fixed 60-s sleep would block the - # whole shutdown until the next heartbeat tick). - # `lost[]` is raised by the heartbeat task if it observes that a sibling reclaimed our lock (we - # stalled past `stale_after`). `_run_one_with_retry!` checks it before `save!`, and the - # `finally` below skips `clear_running!` when lost, so we neither commit on top of nor delete - # the lock now owned by the reclaiming master. The refresh is OWNER-CHECKED (DataVault 0.8.1), - # so a reclaim that has already happened is seen on the next beat; the previous - # existence-based form returned `true` against the reclaimer's own file. - hb_stop = Threads.Atomic{Bool}(false) - lost = Threads.Atomic{Bool}(false) - hb_task = Threads.@spawn begin - tick = 0.1 - elapsed = 0.0 - while !hb_stop[] - sleep(tick) - hb_stop[] && break - elapsed += tick - if elapsed >= opts.heartbeat_interval - # A `false` return (sibling reclaimed) OR a throw (un-refreshable - # lock, e.g. NFS hiccup) both mean "treat as lost": stop - # heartbeating and signal it, rather than silently dying. - alive = try - DataVault.refresh_running!(vault, key, tok) - catch - false - end - if !alive - lost[] = true - break - end - elapsed = 0.0 - end - end - end + # The heartbeat runs in a CHILD process (DataVault 0.8.9). A task here was starved by work that + # does not yield — a long BLAS call, a tight loop — under -t 1, -t 2 and -t 2,1 alike: it never + # beat, a sibling reclaimed the live key after `stale_after`, and this master committed it too. + # The child stops when this process dies or when the lock leaves our hands. + # + # Whether we still hold the key is asked where it matters, at commit: `_run_one_with_retry!` + # checks the owner before `save!`, and commits with the owner form of `mark_done!`, which + # refuses if a sibling reclaimed in between. The release below is owner-checked as well, so it + # can run unconditionally: it never deletes a reclaimer's lock. + hb = DataVault.start_heartbeat(vault, key, tok; interval=opts.heartbeat_interval) outcome = try - _run_one_with_retry!(work_fn, vault, key, kstr, stage, log, opts, lost) + _run_one_with_retry!(work_fn, vault, key, kstr, stage, log, opts, tok) finally - hb_stop[] = true - try - wait(hb_task) - catch - end - # Release the lock so the key is immediately retriable by a sibling - # without waiting for `stale_after` — BUT NOT if we lost it. When - # `lost[]`, the `.running` file is now the RECLAIMING master's, and - # `clear_running!` is owner-blind (`isfile && rm`), so clearing it would - # delete THEIR lock and re-open double-execution. On `:ok`, `mark_done!` - # already removed our `.running`; `clear_running!` is otherwise idempotent. - lost[] || DataVault.clear_running!(vault, key, tok) + DataVault.stop_heartbeat(hb) + # Release so a sibling can retry the key at once instead of after `stale_after`. On `:ok` + # the commit already released it; on a lost key the lock is the reclaimer's, and this is + # a no-op. + DataVault.clear_running!(vault, key, tok) end return (key, outcome) @@ -773,14 +740,15 @@ function _redispatch_deferred( end """ - _run_one_with_retry!(work_fn, vault, key, kstr, stage, log, opts, lost) -> Symbol + _run_one_with_retry!(work_fn, vault, key, kstr, stage, log, opts, tok) -> Symbol Execute `work_fn(key)` up to `opts.max_attempts` times. Returns: :ok — payload saved and mark_done! called :gave_up — all attempts failed, final `:gave_up` event logged :error — single-attempt config (`max_attempts == 1`) that failed once - :lock_busy — `lost[]` was set (a sibling reclaimed our lock); the result is - discarded before `save!` so the reclaiming master's result wins + :lock_busy — the lock is no longer `tok`'s (a sibling reclaimed it): the result is + discarded, before `save!` if the loss is already visible, else at the + owner-checked commit, so the reclaiming master's result wins """ function _run_one_with_retry!( work_fn, @@ -790,7 +758,7 @@ function _run_one_with_retry!( stage::Symbol, log::EventLog, opts::RunOpts, - lost::Threads.Atomic{Bool}, + tok::AbstractString, ) last_err = nothing for attempt in 1:opts.max_attempts @@ -802,7 +770,7 @@ function _run_one_with_retry!( "work_fn must return a Dict (got $(typeof(payload))). " * "Wrap scalars as e.g. Dict(\"value\" => x).", ) - if lost[] + if DataVault.running_owner(vault, key) != tok # A sibling master reclaimed our lock while work_fn ran; it now # owns this key. Discard our result rather than double-committing. log_event(log, :lock_lost; stage=stage, key=kstr, attempt=attempt) @@ -811,9 +779,15 @@ function _run_one_with_retry!( # The digest save! took before its rename goes into the marker, so `.done` names the # bytes this attempt wrote rather than whatever the file holds when someone looks. saved = DataVault.save!(vault, key, payload) - DataVault.mark_done!( - vault, key; result=saved, observation=_observation_token(vault) + # Owner-checked: a reclaim between the check above and here is refused, and nothing is + # committed. The file save! wrote is then the reclaimer's to overwrite. + committed = DataVault.mark_done!( + vault, key, tok; result=saved, observation=_observation_token(vault) ) + if !committed + log_event(log, :lock_lost; stage=stage, key=kstr, attempt=attempt) + return :lock_busy + end log_event( log, :key_done; diff --git a/test/run/test_double_run_safety.jl b/test/run/test_double_run_safety.jl index 06e369a..5c4b556 100644 --- a/test/run/test_double_run_safety.jl +++ b/test/run/test_double_run_safety.jl @@ -54,12 +54,14 @@ end try v = DataVault.Vault(FIXTURE_CFG; run="phase1", outdir=outdir) key = ParamIO.expand(v.spec)[1] - lost = Threads.Atomic{Bool}(true) # simulate: heartbeat saw a reclaim + # Simulate a reclaim: the lock on disk belongs to a sibling, not to `tok`. + tok = DataVault.new_owner_token() + @test DataVault.acquire_running!(v, key, DataVault.new_owner_token()) === :ok log = SweepRunner.EventLog(joinpath(outdir, "ev.jsonl")) ran = Ref(false) wf = k -> (ran[]=true; Dict{String,Any}("x" => 1)) outcome = SweepRunner._run_one_with_retry!( - wf, v, key, ParamIO.canonical(key), :phase1, log, RunOpts(), lost + wf, v, key, ParamIO.canonical(key), :phase1, log, RunOpts(), tok ) @test ran[] # work_fn ran... @test outcome === :lock_busy # ...but its result was discarded diff --git a/test/run/test_run_adversarial.jl b/test/run/test_run_adversarial.jl index 3d20963..f7296c0 100644 --- a/test/run/test_run_adversarial.jl +++ b/test/run/test_run_adversarial.jl @@ -10,6 +10,7 @@ # (exactly the situation the try-finally fix in #8 created a regression for) # ───────────────────────────────────────────────────────────────────────────── +using Printf: @sprintf using SweepRunner, Test, DataVault, ParamIO, JSON3, JLD2 using Distributed using Dates @@ -90,11 +91,20 @@ end # exactly the crashed-run recovery path. DataVault.mark_running!(v, target) running_path = DataVault._running_file(v, target) + # Both heartbeat lines, as a holder that stopped beating leaves them: age is read from + # `heartbeat_unix=` (DataVault 0.8.9), `heartbeat=` by older readers. old_str = Dates.format(Dates.now() - Dates.Second(7200), "yyyy-mm-ddTHH:MM:SS") + old_unix = @sprintf("%017.6f", time() - 7200) lines = readlines(running_path) open(running_path, "w") do io for line in lines - println(io, startswith(line, "heartbeat=") ? "heartbeat=$old_str" : line) + if startswith(line, "heartbeat_unix=") + println(io, "heartbeat_unix=", old_unix) + elseif startswith(line, "heartbeat=") + println(io, "heartbeat=", old_str) + else + println(io, line) + end end end @test DataVault.is_running(v, target) diff --git a/test/run/test_run_busy_holder.jl b/test/run/test_run_busy_holder.jl new file mode 100644 index 0000000..5d60aa3 --- /dev/null +++ b/test/run/test_run_busy_holder.jl @@ -0,0 +1,99 @@ +# A master whose work_fn never yields keeps its key. +# +# The key heartbeat used to be a task in the master. Under a work_fn that does not yield — a long +# BLAS call, a tight loop — it never beat, with -t 1, -t 2 and -t 2,1 alike; a sibling reclaimed +# the live key after `stale_after`, and the master committed it as well. It is now DataVault's +# heartbeat child. The master here runs in its own process (a busy loop in THIS process would +# stop the sibling too) and the sibling polls from this one. + +using SweepRunner, Test, DataVault, ParamIO + +const _BH_CFG = joinpath(@__DIR__, "fixtures", "study.toml") + +function _bh_master_script(outdir, ready, busy_s) + return """ + using SweepRunner, DataVault, ParamIO + v = DataVault.Vault($(repr(_BH_CFG)); run="busy", outdir=$(repr(outdir))) + k = DataVault.keys(v)[1] + work = key -> begin + touch($(repr(ready))) + s = let t = time() + $(busy_s), x = 0.0 + while time() < t; x += sin(x) + 1e-9; end # never yields + x + end + Dict{String,Any}("x" => s) + end + r = run!(work, v, [k]; opts=RunOpts(workers=:sequential, stale_after=1.0, + heartbeat_interval=0.2)) + print(r.done, " ", time()) + """ +end + +@testset "run!: a work_fn that never yields keeps its key against a sibling" begin + for threads in (1, 2) + outdir = mktempdir() + try + ready = joinpath(outdir, "ready") + out = IOBuffer() + exe = Base.julia_cmd() + script = _bh_master_script(outdir, ready, 4.0) + cmd = `$exe --startup-file=no -t $threads --project=$(Base.active_project()) -e $script` + p = run(pipeline(cmd; stdout=out, stderr=devnull); wait=false) + t0 = time() + while !isfile(ready) && process_running(p) && time() - t0 < 180 + sleep(0.05) + end + @test isfile(ready) + + v = DataVault.Vault(_BH_CFG; run="busy", outdir=outdir) + k = DataVault.keys(v)[1] + t_ready = mtime(ready) # when work_fn began, not when we noticed + got_at = Float64[] + tries = 0 + while process_running(p) && !DataVault.is_done(v, k) + tok = DataVault.new_owner_token() + r = DataVault.acquire_running!(v, k, tok; stale_after=1.0) + tries += 1 + if r !== :busy + push!(got_at, time()) + DataVault.clear_running!(v, k, tok) + end + sleep(0.1) + end + wait(p) + done, t_end = split(String(take!(out))) + + # The master really computed for several stale_after while the sibling tried. + @test time() - t_ready >= 3.5 + @test tries >= 25 + # The sibling never got the key while the master held it (only after `run!` ended). + @test count(<(parse(Float64, t_end)), got_at) == 0 + @test done == "1" + @test DataVault.is_done(v, k) + finally + rm(outdir; recursive=true, force=true) + end + end +end + +@testset "run!: a master whose key was reclaimed does not commit it" begin + # The loss is made to happen mid-work_fn: the sibling reclaims with stale_after=0. The master + # must return the key as `:lock_busy`, commit nothing, and leave the sibling's lock alone. + outdir = mktempdir() + try + v = DataVault.Vault(_BH_CFG; run="lost", outdir=outdir) + k = DataVault.keys(v)[1] + sib = DataVault.new_owner_token() + work = + key -> begin + @assert DataVault.acquire_running!(v, key, sib; stale_after=0.0) === :reclaimed + Dict{String,Any}("x" => 1) + end + r = run!(work, v, [k]; opts=RunOpts(workers=:sequential)) + @test r.done == 0 + @test !DataVault.is_done(v, k) + @test DataVault.running_owner(v, k) == sib + finally + rm(outdir; recursive=true, force=true) + end +end diff --git a/test/run/test_run_deadline.jl b/test/run/test_run_deadline.jl index 14a0107..8c466bd 100644 --- a/test/run/test_run_deadline.jl +++ b/test/run/test_run_deadline.jl @@ -52,8 +52,16 @@ end keys = ParamIO.expand(v.spec) @test length(keys) > 1 started = Ref(0) - deadline = time() + 0.3 - work = k -> (started[] += 1; sleep(0.6); Dict{String,Any}("x" => 1)) + # The first key runs until the deadline has passed, so the test does not depend on how long + # `run!` takes to hand it out — only that it is under the margin. A 0.3 s margin was not: + # a run-start observation takes most of a second, and on a hosted runner no key was handed + # out at all (started == 0). + deadline = time() + 5.0 + work = k -> begin + started[] += 1 + sleep(max(0.0, deadline - time()) + 0.1) + Dict{String,Any}("x" => 1) + end r = run!(work, v, keys; opts=RunOpts(workers=:sequential, deadline=deadline)) @test started[] >= 1 # the first key ran @test started[] < length(keys) # later keys were not handed out diff --git a/test/run/test_run_loop_busy.jl b/test/run/test_run_loop_busy.jl index 75e8500..02449a7 100644 --- a/test/run/test_run_loop_busy.jl +++ b/test/run/test_run_loop_busy.jl @@ -32,9 +32,12 @@ end @testset "run_loop!: a dead holder's key is waited out and completed" begin with_tail() do v, ks + # t0 BEFORE the marker, so `elapsed` bounds the marker's own age from above: the key must not + # be reclaimed before `stale_after` of it, and the heartbeat is exact (DataVault 0.8.9's + # `heartbeat_unix=`), so no allowance is due. + t0 = time() DataVault.mark_running!(v, ks[1]) # killed mid-key: the marker outlived the process n = Ref(0) - t0 = time() r = run_loop!( k -> (n[] += 1; Dict{String,Any}("x" => 1)), v, @@ -48,7 +51,10 @@ end @test n[] == 1 # it ran the key @test DataVault.is_done(v, ks[1]) # and the campaign is complete @test r.done == 1 - @test elapsed >= 3.0 # it waited stale_after out rather than giving up + # It waited stale_after out rather than giving up (the bug gave up after two empty rounds, + # ~1 s). Before DataVault 0.8.9 the heartbeat had whole seconds, the lock read up to 1 s + # older than it was, and this failed at 2.85 s and 2.90 s on hosted runners. + @test elapsed >= 3.0 end end