diff --git a/docs/protocol-notes.md b/docs/protocol-notes.md index 480a4b2..614db45 100644 --- a/docs/protocol-notes.md +++ b/docs/protocol-notes.md @@ -104,7 +104,10 @@ source are never read. - **Finalizers never do I/O**: a dropped `MySQL.Handle` enqueues its `ReapEntry` (CAS `:live → :pending`); the reaper (0.5 s timer, `reap_now!`, `atexit`) removes entries under `REAPER_LOCK`, then closes each transport after releasing the lock. `close!` retires the - entry first so a later finalizer is a no-op. Reseau's own poll-FD finalizer is the + entry first so a later finalizer is a no-op. The reaper task runs on the default pool and + catches nothing around its timer wait, so a Ctrl-C that lands in it leaves the task as an + `InterruptException` (which Base hands to the REPL); `ensure_reaper!` starts a new task on + the next connect when the old one is done. Reseau's own poll-FD finalizer is the last-resort fd reclaimer. - **RSA-OAEP through OpenSSL_jll's libcrypto** (`crypto.jl`): explicit SHA-1 OAEP + MGF1, `k - 42` plaintext cap, every handle freed in `finally`, 20k-iteration leak test. The diff --git a/src/reaper.jl b/src/reaper.jl index 4af60b3..b9e2e5c 100644 --- a/src/reaper.jl +++ b/src/reaper.jl @@ -36,6 +36,7 @@ const REAPER_LOCK = Threads.SpinLock() const REAPER_QUEUE = Ref{Union{Nothing, ReapEntry}}(nothing) const REAPER_QUEUE_LENGTH = Ref(0) const REAPER_TIMER = Ref{Union{Nothing, Timer}}(nothing) +const REAPER_TASK = Ref{Union{Nothing, Task}}(nothing) const REAPER_INTERVAL_S = 0.5 const REAPER_STATS = Ref((enqueued=0, closed=0)) @@ -127,15 +128,15 @@ const REAPER_SETUP_LOCK = ReentrantLock() # `Timer(callback)`: Base dispatches timer callbacks from an internal closure task, which # a trimmed executable cannot run, while `wait(::Timer)` is woken from the event loop's C # side. The task body and the atexit hook are `@noinline` with static call edges below. +# +# Nothing is caught around `wait`: a SIGINT is thrown into whichever task is resident on +# thread 1, and when that is the reaper the InterruptException must leave the task so that +# Base's task_done_hook can hand it to the REPL. `ensure_reaper!` replaces a dead reaper. @noinline function reaper_loop() t = REAPER_TIMER[] t === nothing && return nothing while isopen(t) - try - wait(t) - catch - break # timer closed - end + wait(t) try reap_now!() catch err @@ -147,23 +148,30 @@ end @noinline reaper_atexit() = return (try; reap_now!(); catch; end; nothing) -# Starts the timer once. A ReentrantLock (not the finalizer-safe spinlock) because creating a -# Timer and registering the atexit hook may yield. +# Starts the timer once and a reaper task whenever none is alive. A ReentrantLock (not the +# finalizer-safe spinlock) because creating a Timer and registering the atexit hook may yield. function ensure_reaper!() lock(REAPER_SETUP_LOCK) try - REAPER_TIMER[] === nothing || return nothing - REAPER_TIMER[] = Timer(REAPER_INTERVAL_S; interval=REAPER_INTERVAL_S) + task = REAPER_TASK[] + task === nothing || istaskdone(task) || return nothing + if REAPER_TIMER[] === nothing + REAPER_TIMER[] = Timer(REAPER_INTERVAL_S; interval=REAPER_INTERVAL_S) + atexit(reaper_atexit) + end if TRIM_CALL_EDGE[] reaper_loop() reaper_atexit() end task = Task(reaper_loop) task.sticky = false - # reaper_loop handles timer shutdown and logs reclamation errors itself. - # Base.errormonitor would add an untrimmable dynamic error-display task. + # The default pool: when an interactive pool exists the user's tasks run there, and + # a Ctrl-C goes to the task resident on thread 1, which must not be the reaper. + Threads._spawn_set_thrpool(task, :default) + # reaper_loop logs reclamation errors itself. Base.errormonitor would add an + # untrimmable dynamic error-display task. + REAPER_TASK[] = task schedule(task) - atexit(reaper_atexit) finally unlock(REAPER_SETUP_LOCK) end diff --git a/test/protocol/native_tests.jl b/test/protocol/native_tests.jl index 6fb202c..dc9c902 100644 --- a/test/protocol/native_tests.jl +++ b/test/protocol/native_tests.jl @@ -476,6 +476,56 @@ end @test all(ref -> ref.value === nothing, refs) end +@testset "reaper: an interrupt leaves the task and the next registration restarts it" begin + N.ensure_reaper!() + reaper = N.REAPER_TASK[] + @test reaper isa Task + # Off the interactive pool, where the user's work runs when one exists, so a Ctrl-C + # (thrown into the task resident on thread 1) does not land in the reaper. + @test Threads.threadpool(reaper) === :default + timer = N.REAPER_TIMER[] + # Under the timer's lock, a task linked in the timer's wait queue is parked in `wait(timer)`. + # The queue holds the tasks themselves on Julia <= 1.13 and WaitEntry records with an + # atomic `task` field from 1.14 on. + in_waitq(t::Task, c) = any(w -> (w isa Task ? w : (@atomic :monotonic w.task)) === t, c.waitq) + # Throw an InterruptException into the reaper while it is parked in `wait(timer)`, as a + # Ctrl-C does when the reaper is the task resident on thread 1. The timer's lock keeps + # the timer tick from waking the reaper between the check and the throw. + injected = false + deadline = time() + 15 + while !injected && time() < deadline + @lock timer.cond begin + if in_waitq(reaper, timer.cond) + schedule(reaper, InterruptException(); error=true) + injected = true + end + end + injected || sleep(0.01) + end + @test injected + deadline = time() + 15 + while !istaskdone(reaper) && time() < deadline + sleep(0.01) + end + # The interrupt leaves the task, so Base's task_done_hook can hand it to the REPL. + @test istaskfailed(reaper) + @test reaper.result isa InterruptException + N.ensure_reaper!() + fresh = N.REAPER_TASK[] + @test fresh isa Task && fresh !== reaper && !istaskdone(fresh) + counter = P.FaultTransport(IOBuffer()) + entry = N.ReapEntry(counter) + while (@atomic entry.state) == :live + N.enqueue_from_finalizer!(entry) || yield() + end + deadline = time() + 15 + while (@atomic entry.state) != :closed && time() < deadline + sleep(0.05) + end + @test (@atomic entry.state) == :closed + @test (@atomic counter.close_count) == 1 +end + @testset "no descriptor growth across connect/close cycles" begin Sys.iswindows() && return multi_accept_server() do port