From e8b7f47f116a77c244a94e35c6cfec64638777e1 Mon Sep 17 00:00:00 2001 From: Julian P Samaroo Date: Fri, 11 Sep 2026 08:56:55 -0700 Subject: [PATCH 1/7] Make TimespanLogging cheap enough to leave on MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Record into per-thread typed chunk lists and run consumers at collect time, so Datadeps logging-on is ~1.2× instead of ~8×. Categories are Log* types (generated as_old_id, lazy IDs so Dagger precompile does not mutate TimespanLogging), with per-category max_chunks via Preferences. Co-authored-by: Cursor --- AGENTS.md | 30 ++ Project.toml | 2 +- lib/TimespanLogging/Project.toml | 2 +- lib/TimespanLogging/bench/compare_datadeps.jl | 98 ++++ lib/TimespanLogging/bench/workload.jl | 69 +++ lib/TimespanLogging/src/TimespanLogging.jl | 19 +- lib/TimespanLogging/src/buffer.jl | 143 +++++ lib/TimespanLogging/src/category.jl | 185 +++++++ lib/TimespanLogging/src/collect.jl | 154 ++++++ lib/TimespanLogging/src/compat.jl | 305 +++++++++++ lib/TimespanLogging/src/core.jl | 508 ------------------ lib/TimespanLogging/src/emit.jl | 101 ++++ lib/TimespanLogging/src/runtime.jl | 157 ++++++ lib/TimespanLogging/src/types.jl | 104 ++++ lib/TimespanLogging/test/compat_api.jl | 14 + lib/TimespanLogging/test/runtests.jl | 323 ++++++++++- src/Dagger.jl | 2 +- src/datadeps/aliasing.jl | 4 +- src/datadeps/hierarchical.jl | 8 +- src/datadeps/queue.jl | 8 +- src/datadeps/remainders.jl | 16 +- src/sch/Sch.jl | 46 +- src/submission.jl | 4 +- src/utils/logging-categories.jl | 19 + src/utils/logging.jl | 34 +- 25 files changed, 1780 insertions(+), 575 deletions(-) create mode 100644 lib/TimespanLogging/bench/compare_datadeps.jl create mode 100644 lib/TimespanLogging/bench/workload.jl create mode 100644 lib/TimespanLogging/src/buffer.jl create mode 100644 lib/TimespanLogging/src/category.jl create mode 100644 lib/TimespanLogging/src/collect.jl create mode 100644 lib/TimespanLogging/src/compat.jl delete mode 100644 lib/TimespanLogging/src/core.jl create mode 100644 lib/TimespanLogging/src/emit.jl create mode 100644 lib/TimespanLogging/src/runtime.jl create mode 100644 lib/TimespanLogging/src/types.jl create mode 100644 lib/TimespanLogging/test/compat_api.jl create mode 100644 src/utils/logging-categories.jl diff --git a/AGENTS.md b/AGENTS.md index ba7eaaff9..5f18e0547 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -294,3 +294,33 @@ lesson. and died in `ipc_export(::Matrix)`. Stamp the result from `value_memory_space`, and do not select IPC unless the chunktype is a GPU array (`ipc_type_eligible`). Space-only `ipc_eligible` is not enough. + +27. **Log emitters in tests must be count-bounded, and chunk lists must be + memory-bounded.** A `while !stop[]` logger on every default thread will + starve the task that flips `stop` (lesson 12) and allocate slabs until + the machine OOMs — measured at 250GB+ virtual across leftover + `Pkg.test` children after only the parent shell was killed. Use a + fixed `for i in 1:N` (N on the order of a few chunks), cap published + slabs (`MAX_CHUNKS`), and when killing a hung Julia test kill the + whole process group (`kill -- -$PGID`), not just the `julia -e` + wrapper. `Pkg.test` spawns a child that keeps running if you only + SIGTERM the wrapper. + +28. **`nworkers() == 1` means "this process", not "there are workers".** + Without `addprocs`, `workers() == [1]`. `remotecall_wait` / + `remotecall_fetch` to `myid()` deadlocks — the Distributed waiter + never runs on the calling task. Gate broadcasts with + `length(procs()) > 1` (then `workers()` are remote only; Dagger does + not import `nprocs`). `_map_workers` already calls `f()` + locally when `p == myid()`; do not reintroduce a self-remotecall + around `enable_logging!` / `get_logs!`. + +29. **Do not mutate TimespanLogging globals at another package's toplevel.** + `@logcategory` used to call `register_category!` while Dagger was + precompiling. Those writes land in TimespanLogging's arrays in the + *precompile process* and are discarded when TimespanLogging loads from + its own image; Dagger's baked `const` IDs then disagree with a fresh + runtime registry (or collide with MemPool/tests that register later). + Category IDs are assigned lazily on first `category_id` use. Keep it + that way — a `const ID = register_category!(...)` at Dagger toplevel + is not safe. diff --git a/Project.toml b/Project.toml index 3e78bbf9c..958c80843 100644 --- a/Project.toml +++ b/Project.toml @@ -124,6 +124,6 @@ ScopedValues = "1.1" Statistics = "1" StatsBase = "0.28, 0.29, 0.30, 0.31, 0.32, 0.33, 0.34" TaskLocalValues = "0.1" -TimespanLogging = "0.1.1" +TimespanLogging = "0.2" julia = "1.10" oneAPI = "1, 2" diff --git a/lib/TimespanLogging/Project.toml b/lib/TimespanLogging/Project.toml index 118c9a573..c902402b9 100644 --- a/lib/TimespanLogging/Project.toml +++ b/lib/TimespanLogging/Project.toml @@ -1,7 +1,7 @@ name = "TimespanLogging" uuid = "a526e669-04d3-4846-9525-c66122c55f63" authors = ["Julian P Samaroo "] -version = "0.1.1" +version = "0.2.0" [deps] Distributed = "8ba89e20-285c-5b6f-9357-94700520ee1b" diff --git a/lib/TimespanLogging/bench/compare_datadeps.jl b/lib/TimespanLogging/bench/compare_datadeps.jl new file mode 100644 index 000000000..853de5b02 --- /dev/null +++ b/lib/TimespanLogging/bench/compare_datadeps.jl @@ -0,0 +1,98 @@ +# Compare Datadeps matmul / cholesky: master vs this worktree, logging on vs off. +# Run with: timeout 600 julia --startup-file=no lib/TimespanLogging/bench/compare_datadeps.jl + +const WORKTREE = abspath(joinpath(@__DIR__, "..", "..", "..")) +# Same Dagger commit as this worktree, old TimespanLogging, Manifest +# pinned to MemPool 0.4.18 (the live `Dagger` checkout's Manifest still +# has 0.4.15, which cannot load current Dagger). +const MASTER = get(ENV, "DAGGER_TSL_BASELINE", "/tmp/Dagger-tsl-baseline") +const WORKLOAD = joinpath(@__DIR__, "workload.jl") +const JULIA = Base.julia_cmd().exec[1] +const NTHREADS = "4" +const PROC_TIMEOUT = 240 + +function run_one(label::String, dagger_path::String, logging::Bool) + isdir(dagger_path) || error("missing Dagger tree: $dagger_path") + script = """ + include($(repr(WORKLOAD))) + results = run_workload($(logging)) + print_results($(repr(label)), $(logging), results) + """ + cmd = `timeout --kill-after=15 $(PROC_TIMEOUT) $(JULIA) --project=$(dagger_path) -t $(NTHREADS) --startup-file=no -e $(script)` + println(stderr, "running $(label) logging=$(logging) ...") + flush(stderr) + return read(cmd, String) +end + +function parse_results(text::String) + rows = NamedTuple[] + for line in split(text, '\n') + startswith(line, "RESULT ") || continue + fields = Dict{String,String}() + for part in split(line[8:end], ' ') + k, v = split(part, '='; limit=2) + fields[k] = v + end + push!(rows, ( + tree=fields["tree"], + logging=fields["logging"] == "true", + op=Symbol(fields["op"]), + time=parse(Float64, fields["time"]), + allocs=parse(Int, fields["allocs"]), + bytes=parse(Int, fields["bytes"]), + )) + end + return rows +end + +function fmt_bytes(b) + b < 1024 && return "$(b) B" + b < 1024^2 && return string(round(b / 1024; digits=1), " KiB") + return string(round(b / 1024^2; digits=2), " MiB") +end + +function main() + configs = ( + ("master", MASTER, false), + ("master", MASTER, true), + ("worktree", WORKTREE, false), + ("worktree", WORKTREE, true), + ) + rows = NamedTuple[] + for (label, path, logging) in configs + try + text = run_one(label, path, logging) + print(text) + append!(rows, parse_results(text)) + catch err + @error "config failed" label logging exception=err + end + end + println() + println("Datadeps logging overhead (min of $(3) measured runs; N=512, B=64)") + println(rpad("tree", 10), rpad("log", 6), rpad("op", 10), + lpad("time (s)", 10), lpad("allocs", 12), lpad("bytes", 12)) + println(repeat("-", 60)) + for r in rows + println(rpad(r.tree, 10), rpad(string(r.logging), 6), rpad(string(r.op), 10), + lpad(string(round(r.time; digits=4)), 10), + lpad(string(r.allocs), 12), + lpad(fmt_bytes(r.bytes), 12)) + end + println() + println("logging-on / logging-off") + for op in (:matmul, :cholesky) + for tree in ("master", "worktree") + off = findfirst(r -> r.tree == tree && !r.logging && r.op == op, rows) + on = findfirst(r -> r.tree == tree && r.logging && r.op == op, rows) + (off === nothing || on === nothing) && continue + a, b = rows[off], rows[on] + println(rpad(tree, 10), rpad(string(op), 10), + " time×", round(b.time / a.time; digits=2), + " allocs×", round(b.allocs / a.allocs; digits=2), + " bytes×", round(b.bytes / a.bytes; digits=2)) + end + end +end + +main() diff --git a/lib/TimespanLogging/bench/workload.jl b/lib/TimespanLogging/bench/workload.jl new file mode 100644 index 000000000..abcba2ace --- /dev/null +++ b/lib/TimespanLogging/bench/workload.jl @@ -0,0 +1,69 @@ +# Bounded Datadeps workload used by compare_datadeps.jl. +# Do not fetch_logs! here — we want emit-path overhead only. + +using Dagger +using LinearAlgebra + +const N = 512 +const B = 64 +const WARMUP = 3 +const RUNS = 3 + +function measure(f; warmup=WARMUP, runs=RUNS) + for _ in 1:warmup + f() + end + GC.gc() + best_t = Inf + best_a = typemax(Int) + best_b = typemax(Int) + for _ in 1:runs + t0 = time_ns() + before = Base.gc_num() + f() + dt = (time_ns() - t0) / 1e9 + diff = Base.GC_Diff(Base.gc_num(), before) + best_t = min(best_t, dt) + best_a = min(best_a, Base.gc_alloc_count(diff)) + best_b = min(best_b, Int(diff.allocd)) + end + return (time=best_t, allocs=best_a, bytes=best_b) +end + +function setup_logging!(on::Bool) + if on + Dagger.enable_logging!(;all_task_deps=true) + else + Dagger.disable_logging!() + end + return nothing +end + +function run_workload(logging::Bool) + setup_logging!(logging) + A = rand(Blocks(B, B), Float64, N, N) + C = zeros(Blocks(B, B), Float64, N, N) + wait(A); wait(C) + matmul = measure() do + mul!(C, A, A) + end + G = rand(Blocks(B, B), Float64, N, N) + S = G * G' + wait(S) + chol = measure() do + wait(cholesky(S).factors) + end + return (matmul=matmul, cholesky=chol) +end + +function print_results(label, logging, results) + for (op, r) in ((:matmul, results.matmul), (:cholesky, results.cholesky)) + println("RESULT tree=", label, + " logging=", logging, + " op=", op, + " time=", r.time, + " allocs=", r.allocs, + " bytes=", r.bytes) + end + flush(stdout) +end diff --git a/lib/TimespanLogging/src/TimespanLogging.jl b/lib/TimespanLogging/src/TimespanLogging.jl index 0b2b3b9b1..442799ef8 100644 --- a/lib/TimespanLogging/src/TimespanLogging.jl +++ b/lib/TimespanLogging/src/TimespanLogging.jl @@ -1,6 +1,23 @@ module TimespanLogging -include("core.jl") +export enable!, disable!, reset!, steal_typed, steal_legacy, steal_all_old_events +export EventRecord, LegacyEvent, LogCategory, category_id, category_symbol, event_type +export CHUNK_CAPACITY, MAX_CHUNKS, max_chunks, NoOpLog, ActiveLog, LocalEventLog, MultiEventLog + +include("types.jl") +include("category.jl") +include("buffer.jl") +include("runtime.jl") +include("emit.jl") +include("collect.jl") +include("compat.jl") include("extras.jl") +function __init__() + GC_PLACEHOLDER[] = Base.gc_num() + n = max(Threads.maxthreadid(), 1) + THREAD_STATES[] = Vector{Union{ThreadState,Nothing}}(nothing, n) + return nothing +end + end # module diff --git a/lib/TimespanLogging/src/buffer.jl b/lib/TimespanLogging/src/buffer.jl new file mode 100644 index 000000000..cbd78efba --- /dev/null +++ b/lib/TimespanLogging/src/buffer.jl @@ -0,0 +1,143 @@ +""" +Chunk capacity for per-thread log buffers. Full chunks are linked and +published; the open chunk holds a partial tail. Tests that overflow this +capacity exercise the publish path. +""" +const CHUNK_CAPACITY = 256 + +""" +Default hard cap on published slabs per `ChunkList` (preference +`max_chunks`, default 1024). Override one category with +`max_chunks_` (e.g. `max_chunks_compute = 4096`). +1024 × 256 = 262144 events per thread per list; further events overwrite +the open chunk and increment `dropped`. +""" +const MAX_CHUNKS = Int(@load_preference("max_chunks", 1024)) + +function max_chunks(::Type{C}) where C <: LogCategory + key = string("max_chunks_", category_symbol(C)) + v = load_preference(TimespanLogging, key, nothing) + return v === nothing ? MAX_CHUNKS : Int(v) +end + +mutable struct LogChunk{E} + const events::Vector{E} + len::Int + next::Union{LogChunk{E}, Nothing} +end + +function LogChunk{E}(cap::Int=CHUNK_CAPACITY) where E + return LogChunk{E}(Vector{E}(undef, cap), 0, nothing) +end + +""" + ChunkList{E} + +Single-writer growable list of event slabs. The writer fills `open`; when +it is full the slab is prepended to `published` and a new `open` is +allocated. `lock` is a per-list spinlock: only the owning thread writes, +and `steal!` takes it briefly during `get_logs!`. +""" +mutable struct ChunkList{E} + open::LogChunk{E} + published::Union{LogChunk{E}, Nothing} + npublished::Int + dropped::Int + lock::Threads.SpinLock + const max_chunks::Int +end + +function ChunkList{E}(max_chunks::Int=MAX_CHUNKS) where E + return ChunkList{E}(LogChunk{E}(), nothing, 0, 0, Threads.SpinLock(), max_chunks) +end + +@inline function push_event!(list::ChunkList{E}, ev::E) where E + @lock list.lock begin + chunk = list.open + n = chunk.len + if n == length(chunk.events) + if list.npublished >= list.max_chunks + # Bound memory: reuse the open slab instead of allocating. + list.dropped += n + n = 0 + else + chunk.next = list.published + list.published = chunk + list.npublished += 1 + chunk = LogChunk{E}() + list.open = chunk + n = 0 + end + end + @inbounds chunk.events[n + 1] = ev + chunk.len = n + 1 + end + return nothing +end + +""" + steal!(list) -> (open, published) + +Detach the current chain in O(1) (plus a new empty open chunk) so the +writer can continue. `published` is newest-first. +""" +function steal!(list::ChunkList{E}) where E + @lock list.lock begin + open = list.open + pub = list.published + list.published = nothing + list.npublished = 0 + list.open = LogChunk{E}() + return (open, pub) + end +end + +function chunk_count(open::LogChunk, pub::Union{LogChunk, Nothing}) + n = 1 + while pub !== nothing + n += 1 + pub = pub.next + end + return n +end + +function event_count(open::LogChunk, pub::Union{LogChunk, Nothing}) + n = open.len + while pub !== nothing + n += pub.len + pub = pub.next + end + return n +end + +""" +Append stolen chunks to `out` in chronological order (oldest first). +""" +function collect_events!(out::Vector{E}, open::LogChunk{E}, pub::Union{LogChunk{E}, Nothing}) where E + # `published` is newest-first; reverse onto a small stack. + npub = 0 + p = pub + while p !== nothing + npub += 1 + p = p.next + end + if npub > 0 + stack = Vector{LogChunk{E}}(undef, npub) + p = pub + i = npub + while p !== nothing + @inbounds stack[i] = p + i -= 1 + p = p.next + end + for c in stack + if c.len > 0 + append!(out, @view c.events[1:c.len]) + end + end + end + if open.len > 0 + append!(out, @view open.events[1:open.len]) + end + return out +end diff --git a/lib/TimespanLogging/src/category.jl b/lib/TimespanLogging/src/category.jl new file mode 100644 index 000000000..8e70b82a9 --- /dev/null +++ b/lib/TimespanLogging/src/category.jl @@ -0,0 +1,185 @@ +""" + LogCategory + +Abstract supertype of statically declared log categories. Each concrete +category has a runtime-assigned `category_id`, a concrete `id_type`, and a +`category_symbol` used when projecting events back to the legacy `Event` +shape at collect time. +""" +abstract type LogCategory end + +const _CATEGORY_LOCK = Threads.SpinLock() +const _CATEGORY_IDS = Dict{Type,UInt8}() +const _CATEGORY_TYPES = Type[] +const _CATEGORY_SYMBOLS = Symbol[] + +function register_category!(::Type{C}, name::Symbol) where C <: LogCategory + @lock _CATEGORY_LOCK begin + haskey(_CATEGORY_IDS, C) && return _CATEGORY_IDS[C] + length(_CATEGORY_TYPES) < 64 || error("TimespanLogging supports at most 64 categories") + id = UInt8(length(_CATEGORY_TYPES)) + push!(_CATEGORY_TYPES, C) + push!(_CATEGORY_SYMBOLS, name) + _CATEGORY_IDS[C] = id + return id + end +end + +""" + category_id(::Type{C}) -> UInt8 + +Runtime slot for `C`. Assigned lazily on first use so declaring a category +at another package's toplevel does not mutate TimespanLogging during that +package's precompilation (those mutations would be discarded when +TimespanLogging loads from its own image). +""" +function category_id(::Type{C}) where C <: LogCategory + haskey(_CATEGORY_IDS, C) && return _CATEGORY_IDS[C] + return register_category!(C, category_symbol(C)) +end + +category_symbol(::Type{C}) where C <: LogCategory = + error("unregistered log category $C") + +id_type(::Type{C}) where C <: LogCategory = Any +data_type(::Type{C}) where C <: LogCategory = Any + +""" + @logcategory Name as=:symbol id=(field::T, ...) [data=T] [old_data=...] + +Declare a `LogCategory` and its concrete id struct (`NameId`). `data` defaults +to `Any`. Use `data=Nothing` for heartbeat events so the event is `isbits` +and the per-thread buffer stays allocation-free. + +`as_old_id` is generated from the id fields (a NamedTuple with the same +names). `old_data` is optional and only needed when start/finish pass a +bare value that legacy consumers expect wrapped: + +- `old_data=:data` → `(;data=data)` when `data` is not already a NamedTuple +- `old_data=(:f, :result)` → `(;f=data, result=data)` likewise +""" +macro logcategory(name::Symbol, args...) + sym = Symbol(lowercase(String(name))) + id_fields = Tuple{Symbol,Any}[] + data_ty = :Any + old_data = nothing + for arg in args + if Meta.isexpr(arg, :(=)) + lhs, rhs = arg.args[1], arg.args[2] + if lhs === :as + rhs isa QuoteNode || error("@logcategory as= must be a Symbol") + sym = rhs.value + elseif lhs === :id + id_fields = _parse_fields(rhs) + elseif lhs === :data + data_ty = rhs + elseif lhs === :old_data + old_data = rhs + else + error("@logcategory: unknown option $lhs") + end + else + error("@logcategory: expected keyword assignments, got $arg") + end + end + id_struct = Symbol(name, :Id) + id_defs = Expr[] + nt_kws = Expr[] + for (fname, fty) in id_fields + push!(id_defs, :($(esc(fname))::$(esc(fty)))) + push!(nt_kws, Expr(:kw, fname, :(id.$(fname)))) + end + old_id_body = Expr(:tuple, Expr(:parameters, nt_kws...)) + old_data_def = _gen_as_old_data(name, old_data) + quote + struct $(esc(name)) <: $(TimespanLogging).LogCategory end + struct $(esc(id_struct)) + $(id_defs...) + end + $(TimespanLogging).category_symbol(::Type{$(esc(name))}) = $(QuoteNode(sym)) + $(TimespanLogging).id_type(::Type{$(esc(name))}) = $(esc(id_struct)) + $(TimespanLogging).data_type(::Type{$(esc(name))}) = $(esc(data_ty)) + $(TimespanLogging).as_old_id(::Type{$(esc(name))}, id::$(esc(id_struct))) = $old_id_body + $old_data_def + $(esc(name)) + end +end + +function _parse_fields(rhs) + if rhs === :nothing || rhs == :(()) + return Tuple{Symbol,Any}[] + end + args = Meta.isexpr(rhs, :tuple) ? rhs.args : [rhs] + fields = Tuple{Symbol,Any}[] + for a in args + Meta.isexpr(a, :(::)) || error("@logcategory id= expected name::T, got $a") + push!(fields, (a.args[1], a.args[2])) + end + return fields +end + +function _old_data_keys(old_data) + old_data === nothing && return Symbol[] + if old_data isa QuoteNode + return Symbol[old_data.value] + elseif old_data isa Symbol + return Symbol[old_data] + elseif Meta.isexpr(old_data, :tuple) + keys = Symbol[] + for a in old_data.args + if a isa QuoteNode + push!(keys, a.value) + elseif a isa Symbol + push!(keys, a) + else + error("@logcategory old_data= expected Symbol or tuple of Symbols, got $a") + end + end + return keys + else + error("@logcategory old_data= expected Symbol or tuple of Symbols, got $old_data") + end +end + +function _gen_as_old_data(name, old_data) + keys = _old_data_keys(old_data) + isempty(keys) && return nothing + nt_kws = [Expr(:kw, k, :data) for k in keys] + nt = Expr(:tuple, Expr(:parameters, nt_kws...)) + quote + function $(TimespanLogging).as_old_data(::Type{$(esc(name))}, data) + data isa NamedTuple && return data + data === nothing && return data + return $nt + end + end +end + +""" + EventRecord{Cat, Id, D} + +Typed record stored in a per-category chunk list. `phase` is `0x00` for +start and `0x01` for finish. When `Id` and `D` are `isbits`, the record +itself is `isbits` and occupies a packed slot in the chunk. +""" +struct EventRecord{Cat<:LogCategory, Id, D} + phase::UInt8 + timestamp::UInt64 + id::Id + data::D +end + +event_type(::Type{C}) where C <: LogCategory = + EventRecord{C, id_type(C), data_type(C)} + +@inline function adapt_id(::Type{C}, id) where C + T = id_type(C) + id isa T && return id + return convert(T, id) +end + +@inline function adapt_data(::Type{C}, data) where C + T = data_type(C) + data isa T && return data + return convert(T, data) +end diff --git a/lib/TimespanLogging/src/collect.jl b/lib/TimespanLogging/src/collect.jl new file mode 100644 index 000000000..819429122 --- /dev/null +++ b/lib/TimespanLogging/src/collect.jl @@ -0,0 +1,154 @@ +as_old_id(::Type{C}, id) where C <: LogCategory = id +as_old_data(::Type{C}, data) where C <: LogCategory = data + +function as_old_event(ev::LegacyEvent) + phase = ev.phase == 0x00 ? :start : :finish + return Event{phase}(ev.category, ev.id, ev.timeline, ev.timestamp, ev.gc_num, EMPTY_PROF) +end + +function as_old_event(ev::EventRecord{C}) where C <: LogCategory + phase = ev.phase == 0x00 ? :start : :finish + return Event{phase}(category_symbol(C), + as_old_id(C, ev.id), + as_old_data(C, ev.data), + ev.timestamp, + GC_PLACEHOLDER[], + EMPTY_PROF) +end + +function _steal_typed!(olds::Vector{Event}, buf::ChunkList{E}) where E + open, pub = steal!(buf) + n = event_count(open, pub) + n == 0 && return olds + recs = Vector{E}(undef, n) + empty!(recs) + collect_events!(recs, open, pub) + for ev in recs + push!(olds, as_old_event(ev)) + end + return olds +end + +""" + steal_all_old_events() -> Vector{Event} + +Steal every thread's legacy and typed buffers on this process and project +them into the legacy `Event` shape, sorted by timestamp. +""" +function steal_all_old_events() + olds = Event[] + states = THREAD_STATES[] + for i in 1:length(states) + s = states[i] + s === nothing && continue + open, pub = steal!(s.legacy) + if event_count(open, pub) > 0 + recs = LegacyEvent[] + collect_events!(recs, open, pub) + for ev in recs + push!(olds, as_old_event(ev)) + end + end + for j in 1:length(s.typed) + isassigned(s.typed, j) || continue + buf = s.typed[j] + buf === nothing && continue + _steal_typed!(olds, buf) + end + end + sort!(olds, by=e -> e.timestamp) + return olds +end + +""" + steal_typed(::Type{C}) -> Vector{EventRecord} + +Steal only category `C` from every thread (tests / typed consumers). +""" +function steal_typed(::Type{C}) where C <: LogCategory + E = event_type(C) + out = E[] + states = THREAD_STATES[] + for i in 1:length(states) + s = states[i] + s === nothing && continue + id = Int(category_id(C)) + 1 + id > length(s.typed) && continue + isassigned(s.typed, id) || continue + buf = s.typed[id] + buf === nothing && continue + open, pub = steal!(buf::ChunkList{E}) + collect_events!(out, open, pub) + end + sort!(out, by=e -> e.timestamp) + return out +end + +function steal_legacy() + out = LegacyEvent[] + states = THREAD_STATES[] + for i in 1:length(states) + s = states[i] + s === nothing && continue + open, pub = steal!(s.legacy) + collect_events!(out, open, pub) + end + sort!(out, by=e -> e.timestamp) + return out +end + +function consume_events(events::Vector{Event}, + consumers::Dict{Symbol,Any}, + aggregators::Dict{Symbol,Any}=Dict{Symbol,Any}()) + result = Dict{Symbol,Vector}() + n = length(events) + for name in keys(consumers) + result[name] = Vector{Any}(undef, n) + end + for (i, ev) in enumerate(events) + for (name, c) in consumers + result[name][i] = try + c(ev) + catch err + @error "Error during event consumption:" exception=(err, catch_backtrace()) + nothing + end + end + end + for (name, agg) in aggregators + try + agg(result) + catch err + @error "Error during log aggregation:" exception=(err, catch_backtrace()) + end + end + return result +end + +function _get_logs_local(consumers::Dict{Symbol,Any}, + aggregators::Dict{Symbol,Any}) + events = steal_all_old_events() + return consume_events(events, consumers, aggregators) +end + +function _map_workers(f, wkrs) + result = Dict{Int,Any}() + @sync for p in wkrs + if p == myid() + result[p] = f() + else + @async result[p] = remotecall_fetch(f, p) + end + end + return result +end + +function get_logs!(::ActiveLog; only_local=false) + wkrs = only_local ? Int[myid()] : procs() + consumers = INSTALLED_CONSUMERS[] + aggregators = INSTALLED_AGGREGATORS[] + raw = _map_workers(wkrs) do + TimespanLogging._get_logs_local(consumers, aggregators) + end + return Dict{Int,Dict{Symbol,Vector}}(p => v for (p, v) in raw) +end diff --git a/lib/TimespanLogging/src/compat.jl b/lib/TimespanLogging/src/compat.jl new file mode 100644 index 000000000..b0007dd9d --- /dev/null +++ b/lib/TimespanLogging/src/compat.jl @@ -0,0 +1,305 @@ +struct FilterLog + f::Function + inner_chan::Any +end + +function write_event(c::FilterLog, event) + if c.f(event) + write_event(c.inner_chan, event) + end +end + +get_logs!(f::FilterLog; kwargs...) = get_logs!(f.inner_chan; kwargs...) + +function write_event(io::IO, event::Event) + serialize(io, event) +end + +function write_event(chan::Union{RemoteChannel, Channel}, event::Event) + put!(chan, event) +end + +function write_event(arr::AbstractArray, event::Event) + push!(arr, event) +end + +const event_log_lock = Threads.ReentrantLock() + +""" + LocalEventLog + +Compatibility sink. Events are recorded into per-thread buffers; `get_logs!` +steals them and optionally pairs start/finish into `Timespan`s. +""" +struct LocalEventLog end + +function write_event(::LocalEventLog, event::Event) + # Direct write_event is rare; route through the TLS legacy list so we + # do not reintroduce a process-wide lock on the common path. + phase = event isa Event{:start} ? 0x00 : 0x01 + _emit_legacy(phase, event.category, event.id, event.timeline) + return nothing +end + +function get_logs!(::LocalEventLog; raw=false, only_local=false) + wkrs = only_local ? Int[myid()] : procs() + fetched = _map_workers(wkrs) do + TimespanLogging.steal_all_old_events() + end + logs = Dict{Int,Vector{Event}}(p => v for (p, v) in fetched) + if raw + return logs + else + spans = build_timespans(vcat(values(logs)...)).completed + return convert(Vector{Timespan}, spans) + end +end +get_logs!(l::LocalEventLog, raw::Bool; kwargs...) = get_logs!(l; raw=raw, kwargs...) + +mutable struct MultiEventLogState + consumers::Dict{Symbol,Any} + consumer_logs::Dict{Symbol,Vector} + aggregators::Dict{Symbol,Any} +end +MultiEventLogState() = MultiEventLogState(Dict{Symbol,Any}(), + Dict{Symbol,Vector}(), + Dict{Symbol,Any}()) + +const MultiEventLogState_PLS = Dict{UInt64,MultiEventLogState}() + +""" + MultiEventLog + +Compatibility sink. Recording is per-thread; consumers and aggregators run +once at `get_logs!` on the stolen batch (not on every emit). +""" +struct MultiEventLog + uid::UInt64 + consumers::Dict{Symbol,Any} + aggregators::Dict{Symbol,Any} +end +MultiEventLog() = MultiEventLog(rand(UInt64), Dict{Symbol,Any}(), Dict{Symbol,Any}()) + +function Base.setindex!(ml::MultiEventLog, c, name::Symbol) + ml.consumers[name] = c +end + +function get_state(ml::MultiEventLog) + @lock event_log_lock begin + mls = get!(() -> MultiEventLogState(), MultiEventLogState_PLS, ml.uid) + for name in keys(ml.consumers) + if !haskey(mls.consumers, name) + mls.consumers[name] = init_similar(ml.consumers[name]) + mls.consumer_logs[name] = Any[] + end + end + for name in keys(ml.aggregators) + if !haskey(mls.aggregators, name) + mls.aggregators[name] = init_similar(ml.aggregators[name]) + end + end + mls + end +end + +"Creates a copy of `x` with the same configuration, but fresh/empty data." +init_similar(x) = x + +function write_event(::MultiEventLog, event::Event) + phase = event isa Event{:start} ? 0x00 : 0x01 + _emit_legacy(phase, event.category, event.id, event.timeline) + return nothing +end + +function get_logs!(ml::MultiEventLog; only_local=false) + wkrs = only_local ? Int[myid()] : procs() + fetched = _map_workers(wkrs) do + mls = get_state(ml) + events = TimespanLogging.steal_all_old_events() + TimespanLogging.consume_events(events, mls.consumers, mls.aggregators) + end + return Dict{Int,Dict{Symbol,Vector}}(p => v for (p, v) in fetched) +end + +# Profile-enabled finish (rare). Still records a legacy event; attaches +# profiler samples via a second legacy event on the timeline if needed. +const prof_refcount = Ref{Threads.Atomic{Int}}(Threads.Atomic{Int}(0)) +const prof_lock = Threads.ReentrantLock() +const prof_tasks = IdDict{Any, Vector{Task}}() + +function prof_task_put!(id, task::Task=Base.current_task()) + @lock prof_lock push!(get!(()->Task[], prof_tasks, id), task) +end +function prof_tasks_take!(id) + @lock prof_lock begin + if haskey(prof_tasks, id) + pop!(prof_tasks, id) + else + Task[] + end + end +end + +function _timespan_finish_profile(sink, category, @nospecialize(id), @nospecialize(tl), tasks) + time = time_ns() + gcn = gc_num() + prof = UInt[] + lidict = Dict{UInt64, Vector{Base.StackTraces.StackFrame}}() + tasks === nothing && (tasks = prof_tasks_take!(id)) + GC.@preserve tasks begin + @lock prof_lock begin + prof_done = Threads.atomic_sub!(prof_refcount[], 1) == 1 + if prof_done + Profile.stop_timer() + end + prof = @static if VERSION >= v"1.8-" + Profile.fetch(;include_meta=true) + else + Profile.fetch() + end + prof = tasks !== nothing ? filter_profile_data(prof, tasks) : prof + lidict = Profile.getdict(prof) + if prof_done + Profile.clear() + end + end + ev = Event(:finish, category, id, tl, time, gcn, ProfilerResult(prof, lidict, tasks)) + write_event(sink, ev) + end + return nothing +end + +function timespan_start(ctx, category::Symbol, @nospecialize(id), @nospecialize(tl), ::Val{:profile}) + sink = log_sink(ctx) + isa(sink, NoOpLog) && return + if profile(ctx, category, id, tl) && Threads.atomic_add!(prof_refcount[], 1) == 0 + @lock prof_lock Profile.start_timer() + end + _emit_legacy(0x00, category, id, tl) + return nothing +end + +@static if VERSION >= v"1.8-" + function filter_profile_data(prof, tasks::Vector{UInt}) + newprof = UInt[] + startidx = 1 + for i in 1:length(prof) + if prof[i] == 0 + if (i > 2 && prof[i-2] == 0) || + (i > 3 && prof[i-3] == 0) || + (i > 4 && prof[i-4] == 0) + continue + end + task = prof[i - 3] + if task in tasks + append!(newprof, prof[startidx:i]) + end + startidx = i+1 + end + end + newprof + end + filter_profile_data(prof, tasks::Vector{Task}) = + filter_profile_data(prof, map(x->UInt(Base.pointer_from_objref(x)), tasks)) +else + filter_profile_data(prof, tasks) = prof +end + +# Start/finish matching used by LocalEventLog and visualization helpers. + +mutable struct State + start_events::Dict + finish_events::Dict + completed::Vector + start_time::Timestamp + finish_time::Timestamp +end +State() = State(Dict(), Dict(), Any[], 0, 0) + +function add_span(state, tl, category, span) + push!(state.completed, span) + if state.start_time == 0 + state.start_time = span.start + else + state.start_time = min(span.start, state.start_time) + end + if state.finish_time == 0 + state.finish_time = span.finish + else + state.finish_time = max(span.finish, state.finish_time) + end + state +end + +function next_state(state::State, event::Event{:start}) + key = (event.category, event.id) + if haskey(state.finish_events, key) + span = make_timespan(event, pop!(state.finish_events, key)) + add_span(state, event.timeline, event.category, span) + else + state.start_events[key] = event + end + state +end + +function next_state(state::State, event::Event{:finish}) + key = (event.category, event.id) + if haskey(state.start_events, key) + span = make_timespan(pop!(state.start_events, key), event) + add_span(state, event.timeline, event.category, span) + else + state.finish_events[key] = event + end + state +end +next_state(state::State, events::AbstractArray) = + foldl(next_state, events, init=state) + +function mix_samples(a, b) + ProfilerResult(vcat(a.samples, b.samples), + merge(a.lineinfo, b.lineinfo), + unique(vcat(a.tasks, b.tasks))) +end + +function build_timespans(events) + next_state(State(), events) +end + +function add_gc_diff(x, y) + Base.GC_Diff( + x.allocd + y.allocd, + x.malloc + y.malloc, + x.realloc + y.realloc, + x.poolalloc + y.poolalloc, + x.bigalloc + y.bigalloc, + x.freecall + y.freecall, + x.total_time + y.total_time, + x.pause + y.pause, + x.full_sweep + y.full_sweep + ) +end + +function aggregate_events(xs) + gc_diff = reduce(add_gc_diff, map(x -> x.gc_diff, xs)) + time_spent = sum(map(x -> x.finish - x.start, xs)) + profiler_samples = treereduce(mix_samples, map(x->x.profiler_samples, xs)) + time_spent, gc_diff, profiler_samples +end + +function summarize_events(time_spent, gc_diff, profiler_samples) + Base.time_print(time_spent, gc_diff.allocd, gc_diff.total_time, Base.gc_alloc_count(gc_diff)) + if !isempty(profiler_samples.samples) + Profile.print(profiler_samples.samples, profiler_samples.lineinfo) + end +end + +summarize_events(xs) = summarize_events(aggregate_events(xs)...) + +# `timespan_start` used to start the profile timer. Keep that when profile() +# is true by wrapping the exported method. +function _maybe_start_profile(ctx, category, id, tl) + if profile(ctx, category, id, tl) && Threads.atomic_add!(prof_refcount[], 1) == 0 + @lock prof_lock Profile.start_timer() + end + return nothing +end diff --git a/lib/TimespanLogging/src/core.jl b/lib/TimespanLogging/src/core.jl deleted file mode 100644 index 815345671..000000000 --- a/lib/TimespanLogging/src/core.jl +++ /dev/null @@ -1,508 +0,0 @@ -import Preferences: @load_preference, @set_preferences! -if @load_preference("distributed-package") == "DistributedNext" - using DistributedNext -else - using Distributed -end - -import Profile -import Base.gc_num - -export timespan_start, timespan_finish - -const Timestamp = UInt64 - -struct ProfilerResult - samples::Vector{UInt} - lineinfo::AbstractDict - tasks::Vector{UInt} -end -ProfilerResult(samples, lineinfo, tasks::Vector{Task}) = - ProfilerResult(samples, lineinfo, map(Base.pointer_from_objref, tasks)) -ProfilerResult(samples, lineinfo, tasks::Nothing) = - ProfilerResult(samples, lineinfo, map(Base.pointer_from_objref, UInt[])) - -""" - set_distributed_package!(value[="Distributed|DistributedNext"]) - -Set a [preference](https://github.com/JuliaPackaging/Preferences.jl) for using -either the Distributed.jl stdlib or DistributedNext.jl. You will need to restart -Julia after setting a new preference. -""" -function set_distributed_package!(value) - @set_preferences!("distributed-package" => value) - @info "TimespanLogging.jl preference has been set, restart your Julia session for this change to take effect!" -end - -""" - Timespan - -Identifies space (category, id) and time (timeline, start, finish). It also -tracks GC allocations and profiling samples. -""" -struct Timespan - category::Symbol - id::Any - timeline::Any - start::Timestamp - finish::Timestamp - gc_diff::Base.GC_Diff - profiler_samples::ProfilerResult -end - -"An event generated by `timespan_start` or `timespan_finish`." -struct Event{phase} - category::Symbol - id::Any - timeline::Any - timestamp::Timestamp - gc_num::Base.GC_Num - profiler_samples::ProfilerResult -end - -@inline Event(phase::Symbol, category::Symbol, - @nospecialize(id), @nospecialize(tl), - time, gc_num, prof) = - Event{phase}(category, id, tl, time, gc_num, prof) - -""" - make_timespan(start::Event, finish::Event) -> Timespan - -Creates a `Timespan` given the start and finish `Event`s. -""" -function make_timespan(start::Event, finish::Event) - @assert start.category == finish.category - @assert start.id == finish.id - - Timespan(start.category, - start.id, - finish.timeline, - start.timestamp, - finish.timestamp, - Base.GC_Diff(finish.gc_num,start.gc_num), - mix_samples(start.profiler_samples, finish.profiler_samples)) -end - -get_logs!(ctx; kwargs...) = get_logs!(log_sink(ctx); kwargs...) - -""" - NoOpLog - -Disables event logging entirely. -""" -struct NoOpLog end - -function write_event(::NoOpLog, event::Event) -end - -get_logs!(::NoOpLog) = nothing - -struct FilterLog - f::Function - inner_chan::Any -end - -function write_event(c::FilterLog, event) - if c.f(event) - write_event(c.inner_chan, event) - end -end - -get_logs!(f::FilterLog; kwargs...) = get_logs!(f.inner_chan; kwargs...) - -function write_event(io::IO, event::Event) - serialize(io, event) -end - -function write_event(chan::Union{RemoteChannel, Channel}, event::Event) - put!(chan, event) -end - -function write_event(arr::AbstractArray, event::Event) - push!(arr, event) -end - -const event_log_lock = Threads.ReentrantLock() - -""" - LocalEventLog - -Stores events in a process-local array. Accessing the logs is all-or-nothing; -if multiple consumers call `get_logs!`, they will get different sets of logs. -""" -struct LocalEventLog end - -const _local_event_log = Any[] - -function write_event(::LocalEventLog, event::Event) - lock(event_log_lock) do - write_event(_local_event_log, event) - end -end - -""" - get_logs!(::LocalEventLog, raw=false; only_local=false) -> Union{Vector{Timespan},Vector{Event}} - -Get the logs from each process' local event log, clearing it in the process. -Set `raw` to `true` to get potentially unmatched `Event`s; the default is to -return only matched events as `Timespan`s. If `only_local` is set `true`, only -process-local logs will be fetched; the default is to fetch logs from all -processes. -""" -function get_logs!(::LocalEventLog; raw=false, only_local=false) - logs = Dict() - wkrs = only_local ? myid() : procs() - # FIXME: Log this logic - @sync for p in wkrs - @async logs[p] = remotecall_fetch(p) do - lock(event_log_lock) do - log = copy(_local_event_log) - empty!(_local_event_log) - log - end - end - end - if raw - return logs - else - spans = build_timespans(vcat(values(logs)...)).completed - return convert(Vector{Timespan}, spans) - end -end -get_logs!(l::LocalEventLog, raw::Bool; kwargs...) = get_logs!(l; raw=raw, kwargs...) - -mutable struct MultiEventLogState - consumers::Dict{Symbol,Any} - consumer_logs::Dict{Symbol,Vector} - aggregators::Dict{Symbol,Any} -end -MultiEventLogState() = MultiEventLogState(Dict{Symbol,Any}(), - Dict{Symbol,Vector}(), - Dict{Symbol,Any}()) - -const MultiEventLogState_PLS = Dict{UInt64,MultiEventLogState}() - -""" - MultiEventLog - -Processes events immediately, generating multiple log streams. Multiple -consumers may register themselves in the `MultiEventLog`, and when accessed, -log events will be provided to all consumers. A consumer is simply a function -or callable struct which will be called with an event when it's generated. The -return value of the consumer will be pushed into a log stream dedicated to that -consumer. Errors thrown by consumers will be caught and rendered, but will not -otherwise interrupt consumption by other consumers, or future consumption -cycles. An error will result in `nothing` being appended to that consumer's -log. -""" -struct MultiEventLog - uid::UInt64 - consumers::Dict{Symbol,Any} - aggregators::Dict{Symbol,Any} -end -MultiEventLog() = MultiEventLog(rand(UInt64), Dict{Symbol,Any}(), Dict{Symbol,Any}()) - -function Base.setindex!(ml::MultiEventLog, c, name::Symbol) - ml.consumers[name] = c -end - -function get_state(ml::MultiEventLog) - lock(event_log_lock) do - mls = get!(()->MultiEventLogState(), MultiEventLogState_PLS, ml.uid) - max_length = reduce(max, map(length, values(mls.consumer_logs)); init=0) - for name in keys(ml.consumers) - if !haskey(mls.consumers, name) - mls.consumers[name] = init_similar(ml.consumers[name]) - mls.consumer_logs[name] = Vector{Any}(fill(nothing, max_length)) - end - end - for name in keys(ml.aggregators) - if !haskey(mls.aggregators, name) - mls.aggregators[name] = init_similar(ml.aggregators[name]) - end - end - # FIXME: Remove deleted consumers and aggregators - mls - end -end - -"Creates a copy of `x` with the same configuration, but fresh/empty data." -init_similar(x) = x - -function write_event(ml::MultiEventLog, event::Event) - mls = get_state(ml) - lock(event_log_lock) do - for name in keys(mls.consumers) - cevent = try - mls.consumers[name](event) - catch err - @error "Error during event consumption:" exception=(err,catch_backtrace()) - nothing - end - push!(mls.consumer_logs[name], cevent) - end - for name in keys(mls.aggregators) - try - mls.aggregators[name](mls.consumer_logs) - catch err - @error "Error during log aggregation:" exception=(err,catch_backtrace()) - nothing - end - end - end -end - -function get_logs!(ml::MultiEventLog; only_local=false) - logs = Dict{Int,Dict{Symbol,Vector}}() - wkrs = only_local ? myid() : procs() - # FIXME: Log this logic - @sync for p in wkrs - @async begin - logs[p] = remotecall_fetch(p, ml) do ml - mls = get_state(ml) - lock(event_log_lock) do - sublogs = Dict{Symbol,Vector}() - for name in keys(mls.consumers) - sublogs[name] = mls.consumer_logs[name] - mls.consumer_logs[name] = [] - end - sublogs - end - end - end - end - return logs -end - -# Core logging operations - -empty_prof() = ProfilerResult(UInt[], Dict{UInt64, Vector{Base.StackTraces.StackFrame}}(), UInt[]) - -const prof_refcount = Ref{Threads.Atomic{Int}}(Threads.Atomic{Int}(0)) -const prof_lock = Threads.ReentrantLock() -const prof_tasks = IdDict{Any, Vector{Task}}() - -function prof_task_put!(id, task::Task=Base.current_task()) - lock(prof_lock) do - push!(get!(()->Task[], prof_tasks, id), task) - end -end -function prof_tasks_take!(id) - lock(prof_lock) do - if haskey(prof_tasks, id) - pop!(prof_tasks, id) - else - Task[] - end - end -end - -log_sink(ctx) = NoOpLog() -profile(ctx, category, id, tl) = false - -""" - timespan_start(ctx, category::Symbol, id, tl) - -Generates an `Event{:start}` which denotes the start of an event. The event is -categorized by `category`, and uniquely identified by `id`; these two must be -the same passed to `timespan_finish` to close the event. `tl` is the "timeline" -of the event, which is just an arbitrary payload attached to the event. -""" -function timespan_start(ctx, category::Symbol, @nospecialize(id), @nospecialize(tl)) - sink = log_sink(ctx) - isa(sink, NoOpLog) && return - do_profile = profile(ctx, category, id, tl) - if do_profile && Threads.atomic_add!(prof_refcount[], 1) == 0 - lock(prof_lock) do - Profile.start_timer() - end - end - ev = Event(:start, category, id, tl, time_ns(), gc_num(), empty_prof()) - write_event(sink, ev) - nothing -end - -""" - timespan_finish(ctx, category::Symbol, id, tl) - -Generates an `Event{:finish}` which denotes the end of an event. The event is -categorized by `category`, and uniquely identified by `id`; these two must be -the same as previously passed to `timespan_start`. `tl` is the "timeline" of -the event, which is just an arbitrary payload attached to the event. -""" -function timespan_finish(ctx, category::Symbol, @nospecialize(id), @nospecialize(tl); tasks=prof_tasks_take!(id)) - sink = log_sink(ctx) - isa(sink, NoOpLog) && return - do_profile = profile(ctx, category, id, tl) - time = time_ns() - gcn = gc_num() - prof = UInt[] - lidict = Dict{UInt64, Vector{Base.StackTraces.StackFrame}}() - GC.@preserve tasks begin - if do_profile - lock(prof_lock) do - prof_done = Threads.atomic_sub!(prof_refcount[], 1) == 1 - if prof_done - Profile.stop_timer() - end - prof = @static if VERSION >= v"1.8-" - Profile.fetch(;include_meta=true) - else - Profile.fetch() - end - prof = tasks !== nothing ? filter_profile_data(prof, tasks) : prof - lidict = Profile.getdict(prof) - if prof_done - Profile.clear() - end - end - end - ev = Event(:finish, category, id, tl, time, gcn, ProfilerResult(prof, lidict, tasks)) - write_event(sink, ev) - end - nothing -end - -@static if VERSION >= v"1.8-" - function filter_profile_data(prof, tasks::Vector{UInt}) - newprof = UInt[] - startidx = 1 - for i in 1:length(prof) - if prof[i] == 0 - if (i > 2 && prof[i-2] == 0) || - (i > 3 && prof[i-3] == 0) || - (i > 4 && prof[i-4] == 0) - # XXX: Somehow we can get truncated frames? - continue - end - task = prof[i - 3] - if task in tasks - append!(newprof, prof[startidx:i]) - end - startidx = i+1 - end - end - newprof - end - filter_profile_data(prof, tasks::Vector{Task}) = - filter_profile_data(prof, map(x->UInt(Base.pointer_from_objref(x)), tasks)) -else - filter_profile_data(prof, tasks) = prof -end - -""" -Overall state used during visualization -""" -mutable struct State - start_events::Dict # (category, id) => Event - finish_events::Dict # (category, id) => Event - #completed::Dict # timeline => category => Array - completed::Vector - start_time::Timestamp - finish_time::Timestamp -end -State() = State(Dict(), Dict(), Any[], 0, 0) - -""" -Add a Timespan to a given State under `tl` (timeline) -and `category`. -""" -function add_span(state, tl, category, span) - push!(state.completed, span) - if state.start_time == 0 - state.start_time = span.start - else - state.start_time = min(span.start, state.start_time) - end - if state.finish_time == 0 - state.finish_time = span.finish - else - state.finish_time = max(span.finish, state.finish_time) - end - state -end - -"""When building state for real-time visualization, - use next_state to progress gantt state.""" -function next_state(state::State, event::Event{:start}) - key = (event.category, event.id) - if haskey(state.finish_events, key) # finish event reached before start - span = make_timespan(event, pop!(state.finish_events, key)) - add_span(state, event.timeline, event.category, span) - else - state.start_events[key] = event - end - state -end - -function next_state(state::State, event::Event{:finish}) - key = (event.category, event.id) - if haskey(state.start_events, key) - span = make_timespan(pop!(state.start_events, key), event) - add_span(state, event.timeline, event.category, span) - else - state.finish_events[key] = event - end - state -end -next_state(state::State, events::AbstractArray) = - foldl(next_state, events, init=state) - -# util - -function pushkey(dict, key, thing) - if haskey(dict, key) - push!(dict[key],thing) - else - dict[key] = Any[thing] - end -end - -function pushkey(dict, key1, args...) - if haskey(dict, key1) - pushkey(dict[key1], args...) - else - dict[key1] = Dict() - pushkey(dict[key1], args...) - end -end - -function mix_samples(a,b) - ProfilerResult(vcat(a.samples, b.samples), - merge(a.lineinfo, b.lineinfo), - unique(vcat(a.tasks, b.tasks))) -end - -function build_timespans(events) - next_state(State(), events) -end - -function add_gc_diff(x,y) - Base.GC_Diff( - x.allocd + y.allocd, - x.malloc + y.malloc, - x.realloc + y.realloc, - x.poolalloc + y.poolalloc, - x.bigalloc + y.bigalloc, - x.freecall + y.freecall, - x.total_time + y.total_time, - x.pause + y.pause, - x.full_sweep + y.full_sweep - ) -end - -function aggregate_events(xs) - gc_diff = reduce(add_gc_diff, map(x -> x.gc_diff, xs)) - time_spent = sum(map(x -> x.finish - x.start, xs)) - profiler_samples = treereduce(mix_samples, map(x->x.profiler_samples, xs)) - time_spent, gc_diff, profiler_samples -end - -function summarize_events(time_spent, gc_diff, profiler_samples) - Base.time_print(time_spent, gc_diff.allocd, gc_diff.total_time, Base.gc_alloc_count(gc_diff)) - if !isempty(profiler_samples.samples) - Profile.print(profiler_samples.samples, profiler_samples.lineinfo) - end -end - -summarize_events(xs) = summarize_events(aggregate_events(xs)...) diff --git a/lib/TimespanLogging/src/emit.jl b/lib/TimespanLogging/src/emit.jl new file mode 100644 index 000000000..6a47a49d7 --- /dev/null +++ b/lib/TimespanLogging/src/emit.jl @@ -0,0 +1,101 @@ +@inline function _gc_snapshot() + return CAPTURE_GC[] ? Base.gc_num() : GC_PLACEHOLDER[] +end + +@inline function _emit_legacy(phase::UInt8, category::Symbol, @nospecialize(id), @nospecialize(tl)) + st = thread_state() + ev = LegacyEvent(phase, time_ns(), category, id, tl, _gc_snapshot()) + push_event!(st.legacy, ev) + return nothing +end + +@inline function _emit(::Type{C}, phase::UInt8, id, data) where C <: LogCategory + st = thread_state() + buf = typed_buffer(C, st) + ev = EventRecord{C, id_type(C), data_type(C)}( + phase, time_ns(), adapt_id(C, id), adapt_data(C, data)) + push_event!(buf, ev) + return nothing +end + +""" + @logstart ctx Category id data + +Record a start event when `ctx`'s log sink is not `NoOpLog` (and the +category bit is enabled, if a filter is installed). `id` and `data` are +only evaluated when logging is on. +""" +macro logstart(ctx, cat, id, data) + quote + if $(TimespanLogging).logging_enabled($(esc(ctx)), $(esc(cat))) + $(TimespanLogging)._emit($(esc(cat)), 0x00, $(esc(id)), $(esc(data))) + end + nothing + end +end + +""" + @logfinish ctx Category id data +""" +macro logfinish(ctx, cat, id, data) + quote + if $(TimespanLogging).logging_enabled($(esc(ctx)), $(esc(cat))) + $(TimespanLogging)._emit($(esc(cat)), 0x01, $(esc(id)), $(esc(data))) + end + nothing + end +end + +# Category-only form (no ctx): gated solely by `enable!` bits. +macro logstart(cat, id, data) + quote + if $(TimespanLogging).category_enabled($(esc(cat))) + $(TimespanLogging)._emit($(esc(cat)), 0x00, $(esc(id)), $(esc(data))) + end + nothing + end +end + +macro logfinish(cat, id, data) + quote + if $(TimespanLogging).category_enabled($(esc(cat))) + $(TimespanLogging)._emit($(esc(cat)), 0x01, $(esc(id)), $(esc(data))) + end + nothing + end +end + +""" + timespan_start(ctx, category::Symbol, id, tl) + +Legacy emit path. When the sink is `NoOpLog`, this is a no-op (and +`@maybelog` already skipped constructing `id`/`tl`). Otherwise the event +is appended to the calling thread's legacy chunk list — no process-wide +lock, no consumer dispatch, no `ProfilerResult` allocation. +""" +function timespan_start(ctx, category::Symbol, @nospecialize(id), @nospecialize(tl)) + sink = log_sink(ctx) + isa(sink, NoOpLog) && return + _maybe_start_profile(ctx, category, id, tl) + _emit_legacy(0x00, category, id, tl) + return nothing +end + +""" + timespan_finish(ctx, category::Symbol, id, tl; tasks=nothing) + +Legacy finish path. Profiling (`profile(ctx, ...)`) still uses the old +`Profile.fetch` machinery; the common path only stores a typed-enough +legacy record. +""" +function timespan_finish(ctx, category::Symbol, @nospecialize(id), @nospecialize(tl); + tasks=nothing) + sink = log_sink(ctx) + isa(sink, NoOpLog) && return + if profile(ctx, category, id, tl) + _timespan_finish_profile(sink, category, id, tl, tasks) + return nothing + end + _emit_legacy(0x01, category, id, tl) + return nothing +end diff --git a/lib/TimespanLogging/src/runtime.jl b/lib/TimespanLogging/src/runtime.jl new file mode 100644 index 000000000..805ab782b --- /dev/null +++ b/lib/TimespanLogging/src/runtime.jl @@ -0,0 +1,157 @@ +# A zeroed-looking GC_Num captured at init so unsampled events still have a +# valid field without calling `gc_num()` on the emit path. +const GC_PLACEHOLDER = Ref{Base.GC_Num}() + +""" + LegacyEvent + +Record produced by the `timespan_start(ctx, ::Symbol, ...)` compatibility +API. Ids and timelines stay `Any` (call sites still build NamedTuples); +the win on this path is lock-free TLS storage and deferred consumers. +""" +struct LegacyEvent + phase::UInt8 + timestamp::UInt64 + category::Symbol + id::Any + timeline::Any + gc_num::Base.GC_Num +end + +mutable struct ThreadState + const tid::Int + const legacy::ChunkList{LegacyEvent} + # One ChunkList per registered category, created lazily. + const typed::Vector{Any} +end + +function ThreadState(tid::Int) + return ThreadState(tid, ChunkList{LegacyEvent}(), Any[nothing for _ in 1:64]) +end + +const THREAD_STATES_LOCK = Threads.SpinLock() +const THREAD_STATES = Ref{Vector{Union{ThreadState,Nothing}}}(Union{ThreadState,Nothing}[nothing]) + +# Bitset of enabled typed categories. `typemax(UInt64)` enables all. +# `0` means "no category filter": emit is gated only by the call-site sink. +const ENABLED_BITS = Threads.Atomic{UInt64}(0) +const CAPTURE_GC = Threads.Atomic{Bool}(false) + +# Consumers / aggregators installed by `enable!` for `ActiveLog`. +const INSTALLED_CONSUMERS = Ref{Dict{Symbol,Any}}(Dict{Symbol,Any}()) +const INSTALLED_AGGREGATORS = Ref{Dict{Symbol,Any}}(Dict{Symbol,Any}()) + +function _ensure_thread_states!(tid::Int) + states = THREAD_STATES[] + if tid <= length(states) + return states + end + @lock THREAD_STATES_LOCK begin + states = THREAD_STATES[] + if tid > length(states) + newlen = max(tid, Threads.maxthreadid()) + newv = Vector{Union{ThreadState,Nothing}}(nothing, newlen) + copyto!(newv, states) + THREAD_STATES[] = newv + states = newv + end + end + return states +end + +@inline function thread_state() + tid = Threads.threadid() + states = _ensure_thread_states!(tid) + s = @inbounds states[tid] + if s === nothing + s = ThreadState(tid) + @inbounds states[tid] = s + end + return s::ThreadState +end + +function typed_buffer(::Type{C}, st::ThreadState) where C <: LogCategory + id = Int(category_id(C)) + 1 + typed = st.typed + buf = typed[id] + if buf === nothing + E = event_type(C) + buf = ChunkList{E}(max_chunks(C)) + typed[id] = buf + end + return buf::ChunkList{event_type(C)} +end + +""" + enable!(; categories=nothing, capture_gc=false, consumers=..., aggregators=...) + +Install the process-local runtime. `categories` is `nothing` (all bits) or +an iterator of `LogCategory` types. Broadcasts to `workers()` when called +from worker 1. +""" +function enable!(; categories=nothing, + capture_gc::Bool=false, + consumers::Dict{Symbol,Any}=Dict{Symbol,Any}(), + aggregators::Dict{Symbol,Any}=Dict{Symbol,Any}()) + bits = UInt64(0) + if categories === nothing + bits = typemax(UInt64) + else + for C in categories + bits |= UInt64(1) << category_id(C) + end + end + Threads.atomic_xchg!(ENABLED_BITS, bits) + Threads.atomic_xchg!(CAPTURE_GC, capture_gc) + INSTALLED_CONSUMERS[] = consumers + INSTALLED_AGGREGATORS[] = aggregators + return nothing +end + +function disable!() + Threads.atomic_xchg!(ENABLED_BITS, UInt64(0)) + Threads.atomic_xchg!(CAPTURE_GC, false) + INSTALLED_CONSUMERS[] = Dict{Symbol,Any}() + INSTALLED_AGGREGATORS[] = Dict{Symbol,Any}() + return nothing +end + +""" + reset!() + +Disable logging and drop every thread buffer. For tests. +""" +function reset!() + disable!() + @lock THREAD_STATES_LOCK begin + states = THREAD_STATES[] + for i in 1:length(states) + states[i] = nothing + end + end + return nothing +end + +@inline category_enabled(::Type{C}) where C <: LogCategory = begin + bits = ENABLED_BITS[] + bits == 0 && return false + return (bits & (UInt64(1) << category_id(C))) != 0 +end + +@inline logging_enabled() = ENABLED_BITS[] != 0 + +@inline function logging_enabled(ctx) + sink = log_sink(ctx) + return !(sink isa NoOpLog) +end + +@inline function logging_enabled(ctx, ::Type{C}) where C <: LogCategory + log_sink(ctx) isa NoOpLog && return false + bits = ENABLED_BITS[] + # No filter installed: any non-NoOp sink records every category. + bits == 0 && return true + return (bits & (UInt64(1) << category_id(C))) != 0 +end + +log_sink(ctx) = NoOpLog() +profile(ctx, category, id, tl) = false diff --git a/lib/TimespanLogging/src/types.jl b/lib/TimespanLogging/src/types.jl new file mode 100644 index 000000000..6ef8ae6c5 --- /dev/null +++ b/lib/TimespanLogging/src/types.jl @@ -0,0 +1,104 @@ +import Preferences: @load_preference, @set_preferences!, load_preference +if @load_preference("distributed-package") == "DistributedNext" + using DistributedNext +else + using Distributed +end + +import Profile +import Base.gc_num + +export timespan_start, timespan_finish, @logstart, @logfinish, @logcategory + +const Timestamp = UInt64 + +struct ProfilerResult + samples::Vector{UInt} + lineinfo::AbstractDict + tasks::Vector{UInt} +end +ProfilerResult(samples, lineinfo, tasks::Vector{Task}) = + ProfilerResult(samples, lineinfo, map(Base.pointer_from_objref, tasks)) +ProfilerResult(samples, lineinfo, tasks::Nothing) = + ProfilerResult(samples, lineinfo, UInt[]) + +const EMPTY_PROF = ProfilerResult(UInt[], Dict{UInt64, Vector{Base.StackTraces.StackFrame}}(), UInt[]) +empty_prof() = EMPTY_PROF + +""" + set_distributed_package!(value[="Distributed|DistributedNext"]) + +Set a [preference](https://github.com/JuliaPackaging/Preferences.jl) for using +either the Distributed.jl stdlib or DistributedNext.jl. You will need to restart +Julia after setting a new preference. +""" +function set_distributed_package!(value) + @set_preferences!("distributed-package" => value) + @info "TimespanLogging.jl preference has been set, restart your Julia session for this change to take effect!" +end + +""" + Timespan + +Identifies space (category, id) and time (timeline, start, finish). It also +tracks GC allocations and profiling samples. +""" +struct Timespan + category::Symbol + id::Any + timeline::Any + start::Timestamp + finish::Timestamp + gc_diff::Base.GC_Diff + profiler_samples::ProfilerResult +end + +"An event generated by `timespan_start` or `timespan_finish`." +struct Event{phase} + category::Symbol + id::Any + timeline::Any + timestamp::Timestamp + gc_num::Base.GC_Num + profiler_samples::ProfilerResult +end + +@inline Event(phase::Symbol, category::Symbol, + @nospecialize(id), @nospecialize(tl), + time, gc_num, prof) = + Event{phase}(category, id, tl, time, gc_num, prof) + +""" + make_timespan(start::Event, finish::Event) -> Timespan +""" +function make_timespan(start::Event, finish::Event) + @assert start.category == finish.category + @assert start.id == finish.id + + Timespan(start.category, + start.id, + finish.timeline, + start.timestamp, + finish.timestamp, + Base.GC_Diff(finish.gc_num, start.gc_num), + mix_samples(start.profiler_samples, finish.profiler_samples)) +end + +""" + NoOpLog + +Disables event logging entirely. +""" +struct NoOpLog end + +""" + ActiveLog + +Token sink used by `enable_logging!`. Events are recorded into per-thread +chunk lists and consumed at `get_logs!`. +""" +struct ActiveLog end + +get_logs!(ctx; kwargs...) = get_logs!(log_sink(ctx); kwargs...) +write_event(::NoOpLog, event::Event) = nothing +get_logs!(::NoOpLog) = nothing diff --git a/lib/TimespanLogging/test/compat_api.jl b/lib/TimespanLogging/test/compat_api.jl new file mode 100644 index 000000000..14bb9c52f --- /dev/null +++ b/lib/TimespanLogging/test/compat_api.jl @@ -0,0 +1,14 @@ +@testset "Legacy Context helpers" begin + reset!() + ctx = NullContext() + @test TimespanLogging.get_logs!(ctx) == TimespanLogging.get_logs!(NoOpLog()) + timespan_start(ctx, :compute, 1, 2) + timespan_finish(ctx, :compute, 1, 2) + + ctx = Ctx(LocalEventLog(), true) + timespan_start(ctx, :compute, 1, 2) + timespan_finish(ctx, :compute, 1, 2) + logs = TimespanLogging.get_logs!(ctx.log_sink; raw=true) + @test length(logs[1]) == 2 + @test typeof(TimespanLogging.get_logs!(ctx)) == typeof(TimespanLogging.get_logs!(ctx.log_sink)) +end diff --git a/lib/TimespanLogging/test/runtests.jl b/lib/TimespanLogging/test/runtests.jl index 4af73c367..6fad96d8e 100644 --- a/lib/TimespanLogging/test/runtests.jl +++ b/lib/TimespanLogging/test/runtests.jl @@ -1,31 +1,314 @@ using TimespanLogging -import TimespanLogging: NoOpLog, LocalEventLog, MultiEventLog +import TimespanLogging: NoOpLog, LocalEventLog, MultiEventLog, ActiveLog +import TimespanLogging: Event, EventRecord, LegacyEvent, LogCategory +import TimespanLogging: enable!, disable!, reset!, steal_typed, steal_legacy +import TimespanLogging: steal_all_old_events, CHUNK_CAPACITY, category_id +import TimespanLogging: Events using Test -@testset "Contexts" begin - struct NullContext end - ctx = NullContext() - @test TimespanLogging.log_sink(ctx) == NoOpLog() - @test TimespanLogging.profile(ctx, 1, 2, 3) == false +TimespanLogging.@logcategory BenchTick as=:bench_tick id=(n::Int,) data=Nothing +TimespanLogging.@logcategory BenchPay as=:bench_pay id=(n::Int,) data=Any +TimespanLogging.@logcategory PairCat as=:pair id=(key::UInt,) data=Any - timespan_start(ctx, :compute, 1, 2) - timespan_finish(ctx, :compute, 1, 2) +struct NullContext +end + +struct Ctx + log_sink + profile::Bool +end +TimespanLogging.log_sink(ctx::Ctx) = ctx.log_sink +TimespanLogging.profile(ctx::Ctx, xs...) = ctx.profile + +function measure_allocs(f) + f() # warmup caller-side + GC.gc() + before = Base.gc_num() + f() + diff = Base.GC_Diff(Base.gc_num(), before) + return (allocs=Base.gc_alloc_count(diff), bytes=Int(diff.allocd)) +end + +@testset verbose=true "TimespanLogging" begin + reset!() + + @testset "NoOp sink" begin + ctx = NullContext() + @test TimespanLogging.log_sink(ctx) == NoOpLog() + timespan_start(ctx, :compute, 1, 2) + timespan_finish(ctx, :compute, 1, 2) + @test TimespanLogging.get_logs!(ctx) === nothing + @test isempty(steal_legacy()) + end + + @testset "Legacy Symbol API + LocalEventLog" begin + reset!() + ctx = Ctx(LocalEventLog(), false) + timespan_start(ctx, :compute, 1, 2) + timespan_finish(ctx, :compute, 1, 2) + raw = TimespanLogging.get_logs!(ctx.log_sink; raw=true) + @test length(raw[1]) == 2 + @test raw[1][1] isa Event{:start} + @test raw[1][2] isa Event{:finish} + @test raw[1][1].category === :compute + @test raw[1][1].id == 1 + @test raw[1][2].timeline == 2 + # second fetch is empty + raw2 = TimespanLogging.get_logs!(ctx.log_sink; raw=true) + @test isempty(raw2[1]) + + timespan_start(ctx, :compute, :a, nothing) + timespan_finish(ctx, :compute, :a, :done) + spans = TimespanLogging.get_logs!(ctx.log_sink) + @test spans isa Vector{TimespanLogging.Timespan} + @test length(spans) == 1 + @test spans[1].category === :compute + @test spans[1].id === :a + end + + @testset "MultiEventLog consumers run at collect" begin + reset!() + ml = MultiEventLog() + ml[:core] = Events.CoreMetrics() + ml[:id] = Events.IDMetrics() + ctx = Ctx(ml, false) + timespan_start(ctx, :move, (;thunk_id=3), (;data=1)) + timespan_finish(ctx, :move, (;thunk_id=3), (;data=2)) + logs = TimespanLogging.get_logs!(ml) + @test haskey(logs, 1) + @test length(logs[1][:core]) == 2 + @test logs[1][:core][1].kind === :start + @test logs[1][:core][2].kind === :finish + @test logs[1][:core][1].category === :move + @test logs[1][:id][1].thunk_id == 3 + logs2 = TimespanLogging.get_logs!(ml) + @test isempty(logs2[1][:core]) + end + + @testset "Typed categories" begin + reset!() + @test category_id(BenchTick) isa UInt8 + @test TimespanLogging.category_symbol(BenchTick) === :bench_tick + @test isbitstype(EventRecord{BenchTick, BenchTickId, Nothing}) + + enable!(categories=[BenchTick, BenchPay]) + @logstart BenchTick BenchTickId(1) nothing + @logfinish BenchTick BenchTickId(1) nothing + @logstart BenchPay BenchPayId(7) "hello" + @logfinish BenchPay BenchPayId(7) "world" + + ticks = steal_typed(BenchTick) + pays = steal_typed(BenchPay) + @test length(ticks) == 2 + @test ticks[1].phase == 0x00 + @test ticks[2].phase == 0x01 + @test ticks[1].id.n == 1 + @test ticks[1].timestamp <= ticks[2].timestamp + @test length(pays) == 2 + @test pays[1].data == "hello" + @test pays[2].data == "world" + + # Disabled category does not record + @logstart PairCat PairCatId(UInt(1)) :x + @test isempty(steal_typed(PairCat)) + + disable!() + @logstart BenchTick BenchTickId(99) nothing + @test isempty(steal_typed(BenchTick)) + end - @test TimespanLogging.get_logs!(ctx) == TimespanLogging.get_logs!(NoOpLog()) + @testset "@logstart with ctx gates on NoOpLog" begin + reset!() + ctx_off = Ctx(NoOpLog(), false) + ctx_on = Ctx(LocalEventLog(), false) + @logstart ctx_off BenchTick BenchTickId(1) nothing + @logstart ctx_on BenchTick BenchTickId(2) nothing + ticks = steal_typed(BenchTick) + @test length(ticks) == 1 + @test ticks[1].id.n == 2 + end + + @testset "Chunk overflow publishes full slabs" begin + reset!() + enable!(categories=[BenchTick]) + N = CHUNK_CAPACITY * 3 + 17 + for i in 1:N + @logstart BenchTick BenchTickId(i) nothing + end + st = TimespanLogging.thread_state() + buf = TimespanLogging.typed_buffer(BenchTick, st) + # Peek without steal: open + published should cover N events + nchunk = @lock buf.lock TimespanLogging.chunk_count(buf.open, buf.published) + @test nchunk == 4 # 3 full + partial + evs = steal_typed(BenchTick) + @test length(evs) == N + @test [e.id.n for e in evs] == 1:N + @test all(e.phase == 0x00 for e in evs) + @test isempty(steal_typed(BenchTick)) + end + + @testset "Start/finish pairing across threads" begin + reset!() + enable!(categories=[PairCat]) + nwriters = min(Threads.nthreads(), 4) + nper = 64 + @sync for t in 1:nwriters + Threads.@spawn begin + for i in 1:nper + key = UInt(t) << 32 | UInt(i) + @logstart PairCat PairCatId(key) :start + @logfinish PairCat PairCatId(key) :finish + end + end + end + evs = steal_typed(PairCat) + @test length(evs) == nwriters * nper * 2 + starts = Dict{UInt,Int}() + finishes = Dict{UInt,Int}() + for e in evs + if e.phase == 0x00 + starts[e.id.key] = get(starts, e.id.key, 0) + 1 + else + finishes[e.id.key] = get(finishes, e.id.key, 0) + 1 + end + end + @test length(starts) == nwriters * nper + @test starts == finishes + @test all(==(1), values(starts)) + end - struct Context - log_sink - profile::Bool + @testset "Concurrent steal does not drop or duplicate" begin + reset!() + enable!(categories=[BenchTick]) + nper = 128 + stolen = Threads.Atomic{Int}(0) + @sync begin + Threads.@spawn begin + for i in 1:nper + @logstart BenchTick BenchTickId(i) nothing + end + end + Threads.@spawn begin + # Fixed iteration budget — never wait on a condition that + # another sticky spinner might prevent from running. + for _ in 1:16 + evs = steal_typed(BenchTick) + Threads.atomic_add!(stolen, length(evs)) + stolen[] >= nper && break + sleep(0.001) + end + end + end + Threads.atomic_add!(stolen, length(steal_typed(BenchTick))) + @test stolen[] == nper end - ctx = Context(LocalEventLog(), true) - TimespanLogging.log_sink(ctx) = ctx.log_sink - TimespanLogging.profile(ctx, xs...) = ctx.profile - timespan_start(ctx, :compute, 1, 2) - timespan_finish(ctx, :compute, 1, 2) + @testset "Collect-time consumers are type-stable per event" begin + reset!() + enable!(categories=[BenchPay]; + consumers=Dict{Symbol,Any}(:core => Events.CoreMetrics(), + :id => Events.IDMetrics())) + ctx = Ctx(ActiveLog(), false) + @logstart ctx BenchPay BenchPayId(4) :payload + @logfinish ctx BenchPay BenchPayId(4) :done + logs = TimespanLogging.get_logs!(ActiveLog()) + @test length(logs[1][:core]) == 2 + @test logs[1][:core][1].category === :bench_pay + @test logs[1][:id][1] isa NamedTuple + @test logs[1][:id][1].n == 4 + end + + @testset "generated as_old_id and old_data" begin + reset!() + TimespanLogging.@logcategory LogWrap as=:wrap id=(n::Int,) old_data=:payload + enable!(categories=[LogWrap]) + @logstart LogWrap LogWrapId(3) "x" + evs = steal_all_old_events() + @test length(evs) == 1 + @test evs[1].id == (;n=3) + @test evs[1].timeline == (;payload="x") + @logstart LogWrap LogWrapId(4) (;payload="kept") + evs = steal_all_old_events() + @test evs[1].timeline == (;payload="kept") + end + + @testset "per-list max_chunks drops overflow" begin + list = TimespanLogging.ChunkList{Int}(1) + N = CHUNK_CAPACITY * 2 + 10 + for i in 1:N + TimespanLogging.push_event!(list, i) + end + @test list.dropped == CHUNK_CAPACITY + open, pub = TimespanLogging.steal!(list) + @test TimespanLogging.event_count(open, pub) == CHUNK_CAPACITY + 10 + end - logs = TimespanLogging.get_logs!(ctx.log_sink; raw=true) - @test length(logs[1]) == 2 + @testset "as_old_event projection" begin + reset!() + enable!(categories=[BenchTick]) + @logstart BenchTick BenchTickId(5) nothing + evs = steal_all_old_events() + @test length(evs) == 1 + @test evs[1] isa Event{:start} + @test evs[1].category === :bench_tick + @test evs[1].id isa NamedTuple + @test evs[1].id.n == 5 + end + + @testset "Disabled emit does not evaluate payloads" begin + reset!() + evaluated = Ref(false) + payload() = (evaluated[] = true; BenchTickId(1)) + ctx = Ctx(NoOpLog(), false) + @logstart ctx BenchTick payload() nothing + @test !evaluated[] + # The Symbol API is a function, so arguments are evaluated; Dagger + # uses `@maybelog` (or `@logstart`) to skip construction. + end + + @testset "Allocations: disabled and typed heartbeat" begin + reset!() + ctx = Ctx(NoOpLog(), false) + # Disabled Symbol path + function disabled_legacy() + timespan_start(ctx, :compute, 1, nothing) + timespan_finish(ctx, :compute, 1, nothing) + end + disabled_legacy() + a = measure_allocs(disabled_legacy) + @test a.allocs == 0 + + enable!(categories=[BenchTick]) + # Warm TLS / first chunk + for i in 1:32 + @logstart BenchTick BenchTickId(i) nothing + @logfinish BenchTick BenchTickId(i) nothing + end + steal_typed(BenchTick) + + function typed_heartbeat() + @logstart BenchTick BenchTickId(1) nothing + @logfinish BenchTick BenchTickId(1) nothing + end + a = measure_allocs(typed_heartbeat) + @test a.allocs == 0 + steal_typed(BenchTick) + disable!() + end + + @testset "enable! / disable! bitset filter" begin + reset!() + enable!(categories=[BenchTick]) + @test TimespanLogging.category_enabled(BenchTick) + @test !TimespanLogging.category_enabled(BenchPay) + @logstart BenchTick BenchTickId(1) nothing + @logstart BenchPay BenchPayId(1) :x + @test length(steal_typed(BenchTick)) == 1 + @test isempty(steal_typed(BenchPay)) + enable!(categories=[BenchTick, BenchPay]) + @logstart BenchPay BenchPayId(2) :y + @test length(steal_typed(BenchPay)) == 1 + end - @test typeof(TimespanLogging.get_logs!(ctx)) == typeof(TimespanLogging.get_logs!(ctx.log_sink)) + include("compat_api.jl") end diff --git a/src/Dagger.jl b/src/Dagger.jl index aa7ba9369..3c19ac4a1 100644 --- a/src/Dagger.jl +++ b/src/Dagger.jl @@ -27,7 +27,7 @@ if !isdefined(Base, :get_extension) end import TimespanLogging -import TimespanLogging: timespan_start, timespan_finish +import TimespanLogging: timespan_start, timespan_finish, @logstart, @logfinish import Adapt diff --git a/src/datadeps/aliasing.jl b/src/datadeps/aliasing.jl index a8ba0d2ab..13a3ea61f 100644 --- a/src/datadeps/aliasing.jl +++ b/src/datadeps/aliasing.jl @@ -1225,7 +1225,7 @@ function generate_slot!(state::DataDepsState, dest_space, data) ctx = Sch.eager_context() logging = !(ctx.log_sink isa TimespanLogging.NoOpLog) id = logging ? rand(Int) : 0 - logging && timespan_start(ctx, :move, (;thunk_id=0, id, position=ArgPosition(), processor=to_proc), (;f=nothing, data)) + logging && @logstart ctx LogMove LogMoveId(0, ArgPosition(), to_proc, id) data tid = something(DATADEPS_CURRENT_TASK[], (;uid=0)).uid data_chunk = if slot_is_already_in_place(data, orig_space, dest_space) # Nothing to move: the slot for data already in `dest_space` is the data @@ -1240,7 +1240,7 @@ function generate_slot!(state::DataDepsState, dest_space, data) remotecall_endpoint_toplevel(move_rewrap, current_acceleration(), aliased_object_cache, from_proc, to_proc, orig_space, dest_space, data) end end - logging && timespan_finish(ctx, :move, (;thunk_id=0, id, position=ArgPosition(), processor=to_proc), (;f=nothing, data=data_chunk)) + logging && @logfinish ctx LogMove LogMoveId(0, ArgPosition(), to_proc, id) data_chunk @assert memory_space(data_chunk) == dest_space "space mismatch! $dest_space (dest) != $(memory_space(data_chunk)) (actual) ($(typeof(data)) (data) vs. $(typeof(data_chunk)) (chunk)), spaces ($orig_space -> $dest_space)" dest_space_args[data] = data_chunk state.remote_arg_to_original[data_chunk] = data diff --git a/src/datadeps/hierarchical.jl b/src/datadeps/hierarchical.jl index 43e2128f2..9c558d7af 100644 --- a/src/datadeps/hierarchical.jl +++ b/src/datadeps/hierarchical.jl @@ -1440,9 +1440,11 @@ function _hierarchical_copy_from!(state::DataDepsState, arg_w::ArgumentWrapper, @dagdebug nothing :spawn_datadeps "Skipped copy-from (up-to-date): $origin_space" arg = arg_w.arg ctx = Sch.eager_context() - id = rand(UInt) - @maybelog ctx timespan_start(ctx, :datadeps_copy_skip, (;id), (;)) - @maybelog ctx timespan_finish(ctx, :datadeps_copy_skip, (;id), (;thunk_id=0, from_space=origin_space, to_space=origin_space, arg_w, from_arg=arg, to_arg=arg)) + if !(ctx.log_sink isa TimespanLogging.NoOpLog) + id = rand(UInt) + @logstart ctx LogDatadepsCopySkip LogDatadepsCopySkipId(id) nothing + @logfinish ctx LogDatadepsCopySkip LogDatadepsCopySkipId(id) (;thunk_id=0, from_space=origin_space, to_space=origin_space, arg_w, from_arg=arg, to_arg=arg) + end end return end diff --git a/src/datadeps/queue.jl b/src/datadeps/queue.jl index 2db311336..0dab5ddd7 100644 --- a/src/datadeps/queue.jl +++ b/src/datadeps/queue.jl @@ -190,8 +190,8 @@ function distribute_tasks!(queue::DataDepsTaskQueue) ctx = Sch.eager_context() if !(ctx.log_sink isa TimespanLogging.NoOpLog) id = rand(UInt) - timespan_start(ctx, :datadeps_copy_skip, (;id), (;)) - timespan_finish(ctx, :datadeps_copy_skip, (;id), (;thunk_id=0, from_space=origin_space, to_space=origin_space, arg_w, from_arg=arg, to_arg=arg)) + @logstart ctx LogDatadepsCopySkip LogDatadepsCopySkipId(id) nothing + @logfinish ctx LogDatadepsCopySkip LogDatadepsCopySkipId(id) (;thunk_id=0, from_space=origin_space, to_space=origin_space, arg_w, from_arg=arg, to_arg=arg) end end end @@ -413,11 +413,11 @@ function distribute_task!(queue::DataDepsTaskQueue, state::DataDepsState, all_pr new_spec.options.occupancy = Dict(Any=>0) end ctx = Sch.eager_context() - @maybelog ctx timespan_start(ctx, :datadeps_execute, (;thunk_id=task.uid), (;)) + @logstart ctx LogDatadepsExecute LogDatadepsExecuteId(task.uid) nothing enqueue!(queue.upper_queue, DTaskPair(new_spec, task)) # N.B. `task_arg_ws`/`remote_args` are per-task scratch buffers, so the # logged payload snapshots them (only evaluated when logging is enabled) - @maybelog ctx timespan_finish(ctx, :datadeps_execute, (;thunk_id=task.uid), (;space=our_space, deps=logged_task_args(deps_vec, task_arg_ws), args=copy(remote_args))) + @logfinish ctx LogDatadepsExecute LogDatadepsExecuteId(task.uid) (;space=our_space, deps=logged_task_args(deps_vec, task_arg_ws), args=copy(remote_args)) # Reclaim the syncdeps set when the (synchronous) submission above has # already consumed it — see `syncdeps_consumed` for the guard rationale. diff --git a/src/datadeps/remainders.jl b/src/datadeps/remainders.jl index 6a1d947b2..9566b6537 100644 --- a/src/datadeps/remainders.jl +++ b/src/datadeps/remainders.jl @@ -365,9 +365,9 @@ function enqueue_remainder_copy_to!(state::DataDepsState, dest_space::MemorySpac ctx = Sch.eager_context() logging = !(ctx.log_sink isa TimespanLogging.NoOpLog) id = logging ? rand(UInt) : UInt(0) - logging && timespan_start(ctx, :datadeps_copy, (;id), (;)) + logging && @logstart ctx LogDatadepsCopy LogDatadepsCopyId(id) nothing copy_task = Dagger.@spawn scope=dest_scope exec_scope=dest_scope syncdeps=remainder_syncdeps meta=true tag=datadeps_task_tag() Dagger.move!(remainder_aliasing, dest_space, source_space, arg_dest, arg_source) - logging && timespan_finish(ctx, :datadeps_copy, (;id), (;thunk_id=copy_task.uid, from_space=source_space, to_space=dest_space, arg_w, from_arg=arg_source, to_arg=arg_dest)) + logging && @logfinish ctx LogDatadepsCopy LogDatadepsCopyId(id) (;thunk_id=copy_task.uid, from_space=source_space, to_space=dest_space, arg_w, from_arg=arg_source, to_arg=arg_dest) # This copy task reads the sources and writes to the target for ainfo in source_ainfos @@ -424,9 +424,9 @@ function enqueue_remainder_copy_from!(state::DataDepsState, dest_space::MemorySp ctx = Sch.eager_context() logging = !(ctx.log_sink isa TimespanLogging.NoOpLog) id = logging ? rand(UInt) : UInt(0) - logging && timespan_start(ctx, :datadeps_copy, (;id), (;)) + logging && @logstart ctx LogDatadepsCopy LogDatadepsCopyId(id) nothing copy_task = Dagger.@spawn scope=dest_scope exec_scope=dest_scope syncdeps=remainder_syncdeps meta=true tag=datadeps_task_tag() Dagger.move!(remainder_aliasing, dest_space, source_space, arg_dest, arg_source) - logging && timespan_finish(ctx, :datadeps_copy, (;id), (;thunk_id=copy_task.uid, from_space=source_space, to_space=dest_space, arg_w, from_arg=arg_source, to_arg=arg_dest)) + logging && @logfinish ctx LogDatadepsCopy LogDatadepsCopyId(id) (;thunk_id=copy_task.uid, from_space=source_space, to_space=dest_space, arg_w, from_arg=arg_source, to_arg=arg_dest) # This copy task reads the sources and writes to the target for ainfo in source_ainfos @@ -463,9 +463,9 @@ function enqueue_copy_to!(state::DataDepsState, dest_space::MemorySpace, arg_w:: ctx = Sch.eager_context() logging = !(ctx.log_sink isa TimespanLogging.NoOpLog) id = logging ? rand(UInt) : UInt(0) - logging && timespan_start(ctx, :datadeps_copy, (;id), (;)) + logging && @logstart ctx LogDatadepsCopy LogDatadepsCopyId(id) nothing copy_task = Dagger.@spawn scope=dest_scope exec_scope=dest_scope syncdeps=copy_syncdeps meta=true tag=datadeps_task_tag() Dagger.move!(dep_mod, dest_space, source_space, arg_dest, arg_source) - logging && timespan_finish(ctx, :datadeps_copy, (;id), (;thunk_id=copy_task.uid, from_space=source_space, to_space=dest_space, arg_w, from_arg=arg_source, to_arg=arg_dest)) + logging && @logfinish ctx LogDatadepsCopy LogDatadepsCopyId(id) (;thunk_id=copy_task.uid, from_space=source_space, to_space=dest_space, arg_w, from_arg=arg_source, to_arg=arg_dest) # This copy task reads the source and writes to the target add_reader!(state, arg_w, source_space, source_ainfo, copy_task, write_num) @@ -498,9 +498,9 @@ function enqueue_copy_from!(state::DataDepsState, dest_space::MemorySpace, arg_w ctx = Sch.eager_context() logging = !(ctx.log_sink isa TimespanLogging.NoOpLog) id = logging ? rand(UInt) : UInt(0) - logging && timespan_start(ctx, :datadeps_copy, (;id), (;)) + logging && @logstart ctx LogDatadepsCopy LogDatadepsCopyId(id) nothing copy_task = Dagger.@spawn scope=dest_scope exec_scope=dest_scope syncdeps=copy_syncdeps meta=true tag=datadeps_task_tag() Dagger.move!(dep_mod, dest_space, source_space, arg_dest, arg_source) - logging && timespan_finish(ctx, :datadeps_copy, (;id), (;thunk_id=copy_task.uid, from_space=source_space, to_space=dest_space, arg_w, from_arg=arg_source, to_arg=arg_dest)) + logging && @logfinish ctx LogDatadepsCopy LogDatadepsCopyId(id) (;thunk_id=copy_task.uid, from_space=source_space, to_space=dest_space, arg_w, from_arg=arg_source, to_arg=arg_dest) # This copy task reads the source and writes to the target add_reader!(state, arg_w, source_space, source_ainfo, copy_task, write_num) diff --git a/src/sch/Sch.jl b/src/sch/Sch.jl index 0522d5021..7878c92e7 100644 --- a/src/sch/Sch.jl +++ b/src/sch/Sch.jl @@ -20,6 +20,10 @@ import ..Dagger: DepNode, deps_push!, deps_seal! import ..Dagger: order, dependents, noffspring, istask, inputs, unwrap_weak, unwrap_weak_checked, wrap_weak, tochunk, timespan_start, timespan_finish, procs, move, chunktype, default_enabled, processor, get_processors, get_parent, execute!, rmprocs!, task_processor, constrain, cputhreadtime, maybe_take_or_alloc! import ..Dagger: datasize, root_worker_id, is_local_processor, fire_order_key, short_name, select_processors_uniform!, processor_order_key, current_acceleration, set_task_acceleration!, scheduling_ignore_capacity, scheduling_task_occupancy, schedule_argument_move, argument_move_may_inline, sched_move, bind_moved_argument import ..Dagger: @dagdebug, @safe_lock_spin1, @maybelog, @take_or_alloc! +import ..Dagger: LogCompute, LogComputeId, LogMove, LogMoveId, LogTake, LogTakeId +import ..Dagger: LogProcRunWait, LogProcRunWaitId, LogProcRunFetch, LogProcRunFetchId +import ..Dagger: LogEnqueue, LogEnqueueId, LogSchedule, LogScheduleId, LogFire, LogFireId, LogFinish, LogFinishId +import TimespanLogging: @logstart, @logfinish import DataStructures: PriorityQueue import ..Dagger: ReusableCache, ReusableLinkedList, ReusableDict @@ -604,9 +608,9 @@ function handle_result!(ctx, state::ComputeState, pid, proc, thunk_id, res, meta end end - @maybelog ctx timespan_start(ctx, :finish, (;uid=state.uid, thunk_id), (;thunk_id, result=res)) + @logstart ctx LogFinish LogFinishId(state.uid, thunk_id) (;thunk_id, result=res) finish_task!(ctx, state, node, thunk_failed, ready) - @maybelog ctx timespan_finish(ctx, :finish, (;uid=state.uid, thunk_id), (;thunk_id, result=res)) + @logfinish ctx LogFinish LogFinishId(state.uid, thunk_id) (;thunk_id, result=res) return true end proceed || return @@ -647,10 +651,10 @@ function scheduler_run(ctx, state::ComputeState, d::Thunk, options::SchedulerOpt while state.running_count[] > 0 check_workers_available(ctx, options) - @maybelog ctx timespan_start(ctx, :take, (;uid=state.uid), nothing) + @logstart ctx LogTake LogTakeId(state.uid) nothing @dagdebug nothing :take "Waiting for results" tresult = take!(state.chan) # get result of completed thunk - @maybelog ctx timespan_finish(ctx, :take, (;uid=state.uid), nothing) + @logfinish ctx LogTake LogTakeId(state.uid) nothing if tresult isa RescheduleSignal continue end @@ -906,7 +910,7 @@ concurrently across threads. return (true, procs_filt) end @dagdebug task :schedule "Scheduling task" - @maybelog ctx timespan_start(ctx, :schedule, (;uid=state.uid, thunk_id=task.id), (;thunk_id=task.id)) + @logstart ctx LogSchedule LogScheduleId(state.uid, task.id) (;thunk_id=task.id) if has_result(state, task) if (@atomic task.errored) @@ -928,7 +932,7 @@ concurrently across threads. # nor `finish_task!` on this path — release that credit now so the # counter doesn't leak (which would otherwise hang the scheduler). Threads.atomic_sub!(state.running_count, 1) - @maybelog ctx timespan_finish(ctx, :schedule, (;uid=state.uid, thunk_id=task.id), (;thunk_id=task.id)) + @logfinish ctx LogSchedule LogScheduleId(state.uid, task.id) (;thunk_id=task.id) return (true, procs_filt) end @@ -993,7 +997,7 @@ concurrently across threads. # entered `ready_out` (see comment at the other `set_failed!` # call sites in this function for why this is necessary). Threads.atomic_sub!(state.running_count, 1) - @maybelog ctx timespan_finish(ctx, :schedule, (;uid=state.uid, thunk_id=task.id), (;thunk_id=task.id)) + @logfinish ctx LogSchedule LogScheduleId(state.uid, task.id) (;thunk_id=task.id) end return end @@ -1083,7 +1087,7 @@ concurrently across threads. # `finish_task!` — release that credit now to avoid leaking it. Threads.atomic_sub!(state.running_count, 1) end - @maybelog ctx timespan_finish(ctx, :schedule, (;uid=state.uid, thunk_id=task.id), (;thunk_id=task.id)) + @logfinish ctx LogSchedule LogScheduleId(state.uid, task.id) (;thunk_id=task.id) finally unlock(state.lock) end @@ -1495,7 +1499,7 @@ function (ets::FireTaskSpec)() chan = ets.return_chan pid = Dagger.root_worker_id(proc) - @maybelog ctx timespan_start(ctx, :fire, (;uid, worker=pid), nothing) + @logstart ctx LogFire LogFireId(uid, pid) nothing try if pid == myid() do_tasks(proc, chan, tasks) @@ -1510,7 +1514,7 @@ function (ets::FireTaskSpec)() put!(chan, TaskResult(pid, proc, thunk_id, CapturedException(err, bt), nothing)) end finally - @maybelog ctx timespan_finish(ctx, :fire, (;uid, worker=pid), nothing) + @logfinish ctx LogFire LogFireId(uid, pid) nothing end return end @@ -1724,12 +1728,12 @@ function start_processor_runner!(istate::ProcessorInternalState, uid::UInt64, re # Wait for new tasks if !work_to_do @dagdebug nothing :processor "Waiting for tasks" - @maybelog ctx timespan_start(ctx, :proc_run_wait, (;uid, worker=wid, processor=to_proc), nothing) + @logstart ctx LogProcRunWait LogProcRunWaitId(uid, wid, to_proc) nothing wait(istate.reschedule) @static if VERSION >= v"1.9" reset(istate.reschedule) end - @maybelog ctx timespan_finish(ctx, :proc_run_wait, (;uid, worker=wid, processor=to_proc), nothing) + @logfinish ctx LogProcRunWait LogProcRunWaitId(uid, wid, to_proc) nothing if istate.done[] return end @@ -1737,7 +1741,7 @@ function start_processor_runner!(istate::ProcessorInternalState, uid::UInt64, re # Fetch a new task to execute @dagdebug nothing :processor "Trying to dequeue" - @maybelog ctx timespan_start(ctx, :proc_run_fetch, (;uid, worker=wid, processor=to_proc), nothing) + @logstart ctx LogProcRunFetch LogProcRunFetchId(uid, wid, to_proc) nothing # N.B. Results are returned from the locked block rather than # assigned to captured outer locals (which would Core.Box them on # every wakeup) @@ -1757,7 +1761,7 @@ function start_processor_runner!(istate::ProcessorInternalState, uid::UInt64, re return (queue_result, length(queue) > 0) end if task_and_occupancy === nothing - @maybelog ctx timespan_finish(ctx, :proc_run_fetch, (;uid, worker=wid, processor=to_proc), nothing) + @logfinish ctx LogProcRunFetch LogProcRunFetchId(uid, wid, to_proc) nothing @dagdebug nothing :processor "Failed to dequeue" @@ -1827,7 +1831,7 @@ function start_processor_runner!(istate::ProcessorInternalState, uid::UInt64, re task, task_occupancy = task_and_occupancy thunk_id = task.thunk_id time_util = task.est_time_util - @maybelog ctx timespan_finish(ctx, :proc_run_fetch, (;uid, worker=wid, processor=to_proc), (;thunk_id, proc_occupancy=proc_occupancy[], task_occupancy)) + @logfinish ctx LogProcRunFetch LogProcRunFetchId(uid, wid, to_proc) (;thunk_id, proc_occupancy=proc_occupancy[], task_occupancy) @dagdebug thunk_id :processor "Dequeued task" # Skip tasks cancelled by the fallback (which may have fired @@ -2099,7 +2103,7 @@ function do_tasks(to_proc, return_queue, tasks) for task in tasks thunk_id = task.thunk_id occupancy = task.est_occupancy - @maybelog ctx timespan_start(ctx, :enqueue, (;uid, processor=to_proc, thunk_id), nothing) + @logstart ctx LogEnqueue LogEnqueueId(uid, to_proc, thunk_id) nothing # Skip tasks cancelled by the fallback before do_tasks ran. # The fallback marks thunk IDs in states.pre_cancelled so we don't @@ -2128,7 +2132,7 @@ function do_tasks(to_proc, return_queue, tasks) end should_launch || continue push!(queue, task => occupancy) - @maybelog ctx timespan_finish(ctx, :enqueue, (;uid, processor=to_proc, thunk_id), nothing) + @logfinish ctx LogEnqueue LogEnqueueId(uid, to_proc, thunk_id) nothing @dagdebug thunk_id :processor "Enqueued task" end end @@ -2191,7 +2195,7 @@ function move_one_argument!(arg, mctx::MoveCtx) f = mctx.f value = Dagger.value(arg) position = arg.pos - @maybelog ctx timespan_start(ctx, :move, (;thunk_id, position, processor=to_proc), (;f, data=value)) + @logstart ctx LogMove LogMoveId(thunk_id, position, to_proc, nothing) value #= FIXME: This isn't valid if x is written to (formerly used transfer_time/transfer_size stats) x = if x isa Chunk value = lock(TASK_SYNC) do @@ -2247,7 +2251,7 @@ function move_one_argument!(arg, mctx::MoveCtx) @dagdebug thunk_id :move "Moved argument @ $position to $to_proc: $(typeof(value)) -> $(typeof(bound))" end arg.value = bound - @maybelog ctx timespan_finish(ctx, :move, (;thunk_id, position, processor=to_proc), (;f, data=Dagger.value(arg)); tasks=[Base.current_task()]) + @logfinish ctx LogMove LogMoveId(thunk_id, position, to_proc, nothing) Dagger.value(arg) return end @@ -2392,7 +2396,7 @@ Executes a single task specified by `task` on `to_proc`. =# real_time_util[] += est_time_util - @maybelog ctx timespan_start(ctx, :compute, (;thunk_id, processor=to_proc), (;f)) + @logstart ctx LogCompute LogComputeId(thunk_id, to_proc) f # Start counting time and GC allocations threadtime_start = cputhreadtime() @@ -2455,7 +2459,7 @@ Executes a single task specified by `task` on `to_proc`. threadtime = cputhreadtime() - threadtime_start # FIXME: This is not a realistic measure of max. required memory #gc_allocd = min(max(UInt64(Base.gc_num().allocd) - UInt64(gcnum_start.allocd), UInt64(0)), UInt64(1024^4)) - @maybelog ctx timespan_finish(ctx, :compute, (;thunk_id, processor=to_proc), (;f, result=result_meta)) + @logfinish ctx LogCompute LogComputeId(thunk_id, to_proc) (;f, result=result_meta) lock(TASK_SYNC) do real_time_util[] -= est_time_util diff --git a/src/submission.jl b/src/submission.jl index aa8eb7b3f..9fd8fae08 100644 --- a/src/submission.jl +++ b/src/submission.jl @@ -105,7 +105,7 @@ eager_submit_internal!(ctx, state, task, tid, payload::Tuple{<:AnyPayload}) = # Eager DTask uid and Sch thunk id are the same value. id = Int(uid) - @maybelog ctx timespan_start(ctx, :add_thunk, (;thunk_id=id), (;f=fargs[1], args=fargs[2:end], options, uid)) + @logstart ctx LogAddThunk LogAddThunkId(id) (;f=fargs[1], args=fargs[2:end], options, uid) # Keep the *values* of the original arguments alive across edge-wiring: the # loop below replaces `fargs` entries holding a `DTask`/`ThunkID`/`Chunk` @@ -292,7 +292,7 @@ eager_submit_internal!(ctx, state, task, tid, payload::Tuple{<:AnyPayload}) = Sch.schedule_ready!(state, ready) @assert options.syncdeps === nothing || all(dep->dep isa Dagger.ThunkSyncdep && dep.thunk isa Dagger.WeakThunk, options.syncdeps) - @maybelog ctx timespan_finish(ctx, :add_thunk, (;thunk_id=id), (;f=fargs[1], args=fargs[2:end], options, uid)) + @logfinish ctx LogAddThunk LogAddThunkId(id) (;f=fargs[1], args=fargs[2:end], options, uid) return thunk_id end diff --git a/src/utils/logging-categories.jl b/src/utils/logging-categories.jl new file mode 100644 index 000000000..297e42688 --- /dev/null +++ b/src/utils/logging-categories.jl @@ -0,0 +1,19 @@ +# Statically declared log categories for Dagger's hot paths. Call sites use +# `@logstart` / `@logfinish` so ids are concrete structs rather than +# NamedTuples. `as_old_id` is generated from the id fields; `old_data` +# wraps a bare start/finish payload into the NamedTuple shape that +# existing Events.* consumers expect. + +TimespanLogging.@logcategory LogCompute as=:compute id=(thunk_id::Int, processor::Any) old_data=(:f, :result) +TimespanLogging.@logcategory LogMove as=:move id=(thunk_id::Int, position::Any, processor::Any, id::Any) old_data=:data +TimespanLogging.@logcategory LogTake as=:take id=(uid::UInt,) data=Nothing +TimespanLogging.@logcategory LogProcRunWait as=:proc_run_wait id=(uid::UInt, worker::Int, processor::Any) data=Nothing +TimespanLogging.@logcategory LogProcRunFetch as=:proc_run_fetch id=(uid::UInt, worker::Int, processor::Any) old_data=:thunk_id +TimespanLogging.@logcategory LogDatadepsCopy as=:datadeps_copy id=(id::UInt,) +TimespanLogging.@logcategory LogDatadepsCopySkip as=:datadeps_copy_skip id=(id::UInt,) +TimespanLogging.@logcategory LogDatadepsExecute as=:datadeps_execute id=(thunk_id::UInt,) +TimespanLogging.@logcategory LogAddThunk as=:add_thunk id=(thunk_id::Int,) +TimespanLogging.@logcategory LogFinish as=:finish id=(uid::UInt, thunk_id::Int) +TimespanLogging.@logcategory LogEnqueue as=:enqueue id=(uid::UInt, processor::Any, thunk_id::Int) data=Nothing +TimespanLogging.@logcategory LogSchedule as=:schedule id=(uid::UInt, thunk_id::Int) +TimespanLogging.@logcategory LogFire as=:fire id=(uid::UInt, worker::Int) data=Nothing diff --git a/src/utils/logging.jl b/src/utils/logging.jl index a431bd9eb..7b0022f77 100644 --- a/src/utils/logging.jl +++ b/src/utils/logging.jl @@ -1,5 +1,7 @@ # Logging utilities +include("logging-categories.jl") + """ enable_logging!(;kwargs...) @@ -22,6 +24,7 @@ Extra events: - `gc_stats::Bool`: Enables GC allocation tracking per event - `lock_contend::Bool`: Enables lock contention counting per event - `compile_time::Bool`: Enables Julia compile-time tracking per event +- `mempool_fine::Bool`: Enables high-frequency MemPool events (storage RCU) """ function enable_logging!(;metrics::Bool=false, timeline::Bool=false, @@ -38,7 +41,8 @@ function enable_logging!(;metrics::Bool=false, linuxperf::String="", gc_stats::Bool=false, lock_contend::Bool=false, - compile_time::Bool=false) + compile_time::Bool=false, + mempool_fine::Bool=false) ml = TimespanLogging.MultiEventLog() ml[:core] = TimespanLogging.Events.CoreMetrics() ml[:id] = TimespanLogging.Events.IDMetrics() @@ -107,7 +111,31 @@ function enable_logging!(;metrics::Bool=false, if compile_time ml[:compile_time] = Dagger.Events.CompileTimeMetrics() end - Dagger.Sch.eager_context().log_sink = ml + _install_log_sink!(ml, gc_stats, mempool_fine) + return +end + +function _install_log_sink!(sink, capture_gc::Bool, mempool_fine::Bool=false) + Dagger.Sch.eager_context().log_sink = sink + if sink isa TimespanLogging.NoOpLog + TimespanLogging.disable!() + else + TimespanLogging.enable!(; capture_gc) + end + if isdefined(MemPool, :set_log_sink!) + MemPool.set_log_sink!(sink) + fine = mempool_fine && !(sink isa TimespanLogging.NoOpLog) + if isdefined(MemPool, :set_log_fine!) + MemPool.set_log_fine!(fine) + end + end + # `nworkers() == 1` with no extra workers, and `workers() == [1]`. A + # remotecall to self deadlocks (the Distributed waiter never runs). + if myid() == 1 && length(procs()) > 1 + @sync for w in workers() + @async remotecall_wait(_install_log_sink!, w, sink, capture_gc, mempool_fine) + end + end return end @@ -123,7 +151,7 @@ end Disables logging previously enabled with `enable_logging!`. """ function disable_logging!() - Dagger.Sch.eager_context().log_sink = TimespanLogging.NoOpLog() + _install_log_sink!(TimespanLogging.NoOpLog(), false) return end From 85ff1284aef9ae0d88ef15d5ce6e1e122013b4f6 Mon Sep 17 00:00:00 2001 From: Julian P Samaroo Date: Fri, 11 Sep 2026 12:18:35 -0700 Subject: [PATCH 2/7] submission: Snapshot LogAddThunk's syncdeps instead of aliasing the pooled Set LogAddThunk's logged payload held a live reference to the task's `Options` struct, whose `syncdeps` field datadeps' queue.jl nulls out and returns to a pool (`return_syncdeps_set!`) immediately after a synchronous submission consumes it. That was harmless under the old eager-consumer design, where `TaskDependencies` read `syncdeps` synchronously at the `timespan_finish` call site -- before the pool reclaimed it. Under the new deferred-consumer design ("Make TimespanLogging cheap enough to leave on", e8b7f47f), the event is only projected into the legacy `Event` shape and consumed at `fetch_logs!()` time, by which point the pooled `Set` this event still references has already been nulled and possibly handed to a later task. `ev.timeline.options.syncdeps` then reads back `nothing`, so `TaskDependencies` reports every task as having zero dependencies. This silently broke `test/datadeps.jl`'s dependency-graph reconstruction (`build_dataflow`/`test_dataflow`), which only ever consults task IDs, not data content, so nothing crashed -- flows just looked undominated. 146 of 1464 datadeps tests failed as a result; a minimal repro (`In(A)` then `Out(A)`, two tasks) reproduces it standalone and confirms the fix restores the same `taskdeps` result as the pre-e8b7f47f baseline. `TaskDependencies` is the only reader of `LogAddThunk`'s `options.syncdeps` anywhere in the tree, so the fix only needs to keep that one field alive: snapshot it into a bare `(;syncdeps=...)` NamedTuple at log time, replacing the live `Options` reference the logged payload used to carry. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GaPaEAUCJMwpQjFFPbU7Ck --- src/submission.jl | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/src/submission.jl b/src/submission.jl index 9fd8fae08..2f8102a4d 100644 --- a/src/submission.jl +++ b/src/submission.jl @@ -81,6 +81,19 @@ function eager_submit_internal!(payload::AnyPayload) tid = 0 return eager_submit_internal!(ctx, state, task, tid, payload) end +""" +`LogAddThunk`'s payload used to carry the live `Options` object by reference. +`options.syncdeps` is a pooled `Set` that queue.jl's datadeps path nulls out +and returns to the pool (`return_syncdeps_set!`) right after a *synchronous* +submission consumes it -- so a consumer reading `ev.timeline.options.syncdeps` +at collect time (rather than eagerly, at log time) would see `nothing` even +though the dependency really was there. `TaskDependencies` is the only reader +of this field (checked across the whole tree); snapshot just that field here, +at log time, instead of aliasing the mutable struct. +""" +logged_options(options::Options) = + (;syncdeps = options.syncdeps === nothing ? nothing : copy(options.syncdeps)) + eager_submit_internal!(ctx, state, task, tid, payload::Tuple{<:AnyPayload}) = eager_submit_internal!(ctx, state, task, tid, payload[1]) @reuse_scope function eager_submit_internal!(ctx, state, task, tid, payload::AnyPayload) @@ -105,7 +118,7 @@ eager_submit_internal!(ctx, state, task, tid, payload::Tuple{<:AnyPayload}) = # Eager DTask uid and Sch thunk id are the same value. id = Int(uid) - @logstart ctx LogAddThunk LogAddThunkId(id) (;f=fargs[1], args=fargs[2:end], options, uid) + @logstart ctx LogAddThunk LogAddThunkId(id) (;f=fargs[1], args=fargs[2:end], options=logged_options(options), uid) # Keep the *values* of the original arguments alive across edge-wiring: the # loop below replaces `fargs` entries holding a `DTask`/`ThunkID`/`Chunk` @@ -292,7 +305,7 @@ eager_submit_internal!(ctx, state, task, tid, payload::Tuple{<:AnyPayload}) = Sch.schedule_ready!(state, ready) @assert options.syncdeps === nothing || all(dep->dep isa Dagger.ThunkSyncdep && dep.thunk isa Dagger.WeakThunk, options.syncdeps) - @logfinish ctx LogAddThunk LogAddThunkId(id) (;f=fargs[1], args=fargs[2:end], options, uid) + @logfinish ctx LogAddThunk LogAddThunkId(id) (;f=fargs[1], args=fargs[2:end], options=logged_options(options), uid) return thunk_id end From 60ff24e53d47f6962cd188a2533743230c2cf82e Mon Sep 17 00:00:00 2001 From: Julian P Samaroo Date: Fri, 11 Sep 2026 17:15:04 -0700 Subject: [PATCH 3/7] CI: Develop local TimespanLogging in the GPU/MPI env jobs too The main Buildkite/GHA test matrix already devs lib/TimespanLogging into the root project, but the OpenCL, OpenCL+MPI, CUDA/ROCm/oneAPI/ Metal (plain and +MPI), and GHA mpi-cpu/mpi-opencl jobs instantiate a separate test/*env project and only `Pkg.develop` the Dagger checkout into it -- TimespanLogging stayed pinned to whatever's in the General registry there, since it's only a transitive dependency and isn't covered by Dagger's own [sources]. Add the same `Pkg.develop(path="lib/TimespanLogging")` call to each of those jobs so they all exercise the fast local TimespanLogging. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GaPaEAUCJMwpQjFFPbU7Ck --- .buildkite/pipeline-julia.yml | 3 ++- .buildkite/pipeline.yml | 9 +++++---- .github/workflows/CI.yml | 4 ++-- 3 files changed, 9 insertions(+), 7 deletions(-) diff --git a/.buildkite/pipeline-julia.yml b/.buildkite/pipeline-julia.yml index 284569e90..e0852ee1f 100644 --- a/.buildkite/pipeline-julia.yml +++ b/.buildkite/pipeline-julia.yml @@ -127,6 +127,7 @@ steps: arch: aarch64 env: CI_USE_OPENCL: "1" + command: "julia --project -e 'using Pkg; Pkg.develop(;path=\"lib/TimespanLogging\")'" # MPI × OpenCL SPMD datadeps suite. Runs on the same macOS/aarch64 agent as the # non-MPI OpenCL job, but drives test/mpi_opencl.jl under MPI.jl's bundled @@ -142,7 +143,7 @@ steps: os: macos arch: aarch64 command: | - julia --project=test/openclenv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.instantiate()' + julia --project=test/openclenv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.develop(path="lib/TimespanLogging"); Pkg.instantiate()' julia --project=test/openclenv test/run_mpi.jl 2 2 test/mpi_opencl.jl - label: Julia 1 - TimespanLogging diff --git a/.buildkite/pipeline.yml b/.buildkite/pipeline.yml index b61fbc9b4..e86589d05 100644 --- a/.buildkite/pipeline.yml +++ b/.buildkite/pipeline.yml @@ -1,6 +1,7 @@ .gputest: &gputest timeout_in_minutes: 60 if: build.message !~ /\[skip tests\]/ + command: "julia --project -e 'using Pkg; Pkg.develop(;path=\"lib/TimespanLogging\")'" # MPI × GPU jobs launch the SPMD datadeps suite (test/mpi_.jl) under # `mpiexec`. They use only the `julia` plugin (not `julia-test`, which would run @@ -74,7 +75,7 @@ steps: agents: queue: "cuda" command: | - julia --project=test/cudaenv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.instantiate()' + julia --project=test/cudaenv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.develop(path="lib/TimespanLogging"); Pkg.instantiate()' julia --project=test/cudaenv test/run_mpi.jl 2 2 test/mpi_cuda.jl - label: Julia 1.11 (ROCm, MPI) @@ -85,7 +86,7 @@ steps: agents: queue: "rocm" command: | - julia --project=test/rocmenv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.instantiate()' + julia --project=test/rocmenv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.develop(path="lib/TimespanLogging"); Pkg.instantiate()' julia --project=test/rocmenv test/run_mpi.jl 2 2 test/mpi_rocm.jl - label: Julia 1.11 (oneAPI, MPI) @@ -96,7 +97,7 @@ steps: agents: queue: "oneapi" command: | - julia --project=test/oneapienv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.instantiate()' + julia --project=test/oneapienv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.develop(path="lib/TimespanLogging"); Pkg.instantiate()' julia --project=test/oneapienv test/run_mpi.jl 2 2 test/mpi_oneapi.jl - label: Julia 1.11 (Metal, MPI) @@ -108,7 +109,7 @@ steps: agents: queue: "metal" command: | - julia --project=test/metalenv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.instantiate()' + julia --project=test/metalenv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.develop(path="lib/TimespanLogging"); Pkg.instantiate()' julia --project=test/metalenv test/run_mpi.jl 2 2 test/mpi_metal.jl - label: Julia 1.11 (Finch) diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml index 5f9b2a108..23bc7b537 100644 --- a/.github/workflows/CI.yml +++ b/.github/workflows/CI.yml @@ -437,7 +437,7 @@ jobs: # feature and is silently ignored on 1.10 (LTS), which would otherwise # resolve a stale registered Dagger lacking the MPI API (`accelerate!`). - name: Instantiate CPU MPI environment - run: julia --project=test/mpienv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.instantiate()' + run: julia --project=test/mpienv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.develop(path="lib/TimespanLogging"); Pkg.instantiate()' - name: Run CPU MPI tests run: julia --project=test/mpienv test/run_mpi.jl ${{ matrix.ranks }} 2 test/mpi.jl @@ -462,6 +462,6 @@ jobs: arch: x64 - uses: julia-actions/cache@v3 - name: Instantiate OpenCL MPI environment - run: julia --project=test/openclenv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.instantiate()' + run: julia --project=test/openclenv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.develop(path="lib/TimespanLogging"); Pkg.instantiate()' - name: Run OpenCL MPI tests run: julia --project=test/openclenv test/run_mpi.jl 2 2 test/mpi_opencl.jl From 954e6f517cea425038c1f0efa45d14c27a939127 Mon Sep 17 00:00:00 2001 From: Julian P Samaroo Date: Fri, 11 Sep 2026 17:15:04 -0700 Subject: [PATCH 4/7] test/logging: Poll for quiescence instead of a fixed sleep(1) The "Manual" MultiEventLog test drained the shared per-thread event buffers once, after a fixed sleep(1). get_logs! is destructive, and compute(ctx, c) returning doesn't guarantee all scheduler teardown activity has finished landing its events -- on a loaded/oversubscribed runner (e.g. CI's shared vCPUs), that trailing activity can still be in flight past the 1-second mark, dropping events the test then expects to see. Poll and merge across drains instead, requiring a short quiet period (no newly-drained events) after the expected categories show up, bounded by an overall deadline. Verified the naive version of this fix (break as soon as the expected categories appear, without the quiet-period check) actively reproduces a related failure -- trailing events land on the *next* drain instead, which the test's own "should be empty after we're done" check then catches. That's direct evidence of real trailing async activity after compute() returns, and the likely root cause of the "Julia 1 - ubuntu-latest" flake on PR JuliaParallel/Dagger.jl#751 (not locally reproducible even after 10 repeated runs, including one with the CPU count pinned to match the runner). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GaPaEAUCJMwpQjFFPbU7Ck --- test/logging.jl | 36 ++++++++++++++++++++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/test/logging.jl b/test/logging.jl index 92f1dab32..7378ead22 100644 --- a/test/logging.jl +++ b/test/logging.jl @@ -132,9 +132,41 @@ end b = delayed(sum)(X) c = delayed(+)(a,b) compute(ctx, c) - sleep(1) - logs = TimespanLogging.get_logs!(ml) + # `get_logs!` destructively drains the shared per-thread event + # buffers, so a single fixed sleep() before one drain can race + # scheduler teardown activity that lands slightly late under CPU + # pressure (e.g. a busy, oversubscribed CI runner). Poll and + # merge across drains instead of sleep-then-drain-once, requiring + # a short quiet period (no newly-drained events) once the + # expected categories show up, so a slow-but-eventually- + # consistent run still passes without racing trailing activity. + logs = Dict{Int,Dict{Symbol,Vector}}() + deadline = time() + 10 + quiet_since = nothing + while true + added = 0 + for (w, cats) in TimespanLogging.get_logs!(ml) + dcats = get!(Dict{Symbol,Vector}, logs, w) + for (cat, v) in cats + added += length(v) + append!(get!(Vector{Any}, dcats, cat), v) + end + end + w1 = get(logs, 1, Dict{Symbol,Vector}()) + ready = haskey(w1, :core) && length(w1[:core]) > 1 && + haskey(w1, :esat) && + any(e -> haskey(e, :scheduler_init), w1[:esat]) && + any(e -> haskey(e, :finish), w1[:esat]) + if ready && added == 0 + quiet_since === nothing && (quiet_since = time()) + time() - quiet_since >= 0.2 && break + else + quiet_since = nothing + end + time() >= deadline && break + sleep(0.05) + end for w in keys(logs) len = length(logs[w][:core]) if w == 1 From 33eff72085ff64dbf57a9a5c21e87abd00e99f60 Mon Sep 17 00:00:00 2001 From: Julian P Samaroo Date: Sat, 12 Sep 2026 08:20:12 -0700 Subject: [PATCH 5/7] CI: Batch the two Pkg.develop calls in MPI/GPU env jobs Two sequential Pkg.develop calls (Dagger, then TimespanLogging) into the same test/*env project broke dependency resolution on Julia 1.10: the second call's targeted resolve no longer saw Dagger as pinned to its dev path and instead treated it as fixed to the registered 0.22.4, which only allows TimespanLogging 0.1.x -- unsatisfiable against Dagger's own TimespanLogging = "0.2" compat bound. Reproduced directly with a from-scratch depot on Julia 1.10: ERROR: Unsatisfiable requirements detected for package TimespanLogging [a526e669]: ... restricted to versions 0.2 by Dagger [d58978e5] -- no versions left Batching both packages into one Pkg.develop([...]) call resolves them together against a single dev-pinned Dagger and avoids the conflict; confirmed against the same from-scratch 1.10 depot. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GaPaEAUCJMwpQjFFPbU7Ck --- .buildkite/pipeline-julia.yml | 2 +- .buildkite/pipeline.yml | 8 ++++---- .github/workflows/CI.yml | 4 ++-- 3 files changed, 7 insertions(+), 7 deletions(-) diff --git a/.buildkite/pipeline-julia.yml b/.buildkite/pipeline-julia.yml index e0852ee1f..a035c48fc 100644 --- a/.buildkite/pipeline-julia.yml +++ b/.buildkite/pipeline-julia.yml @@ -143,7 +143,7 @@ steps: os: macos arch: aarch64 command: | - julia --project=test/openclenv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.develop(path="lib/TimespanLogging"); Pkg.instantiate()' + julia --project=test/openclenv -e 'using Pkg; Pkg.develop([PackageSpec(path=pwd()), PackageSpec(path="lib/TimespanLogging")]); Pkg.instantiate()' julia --project=test/openclenv test/run_mpi.jl 2 2 test/mpi_opencl.jl - label: Julia 1 - TimespanLogging diff --git a/.buildkite/pipeline.yml b/.buildkite/pipeline.yml index e86589d05..90a7c45b1 100644 --- a/.buildkite/pipeline.yml +++ b/.buildkite/pipeline.yml @@ -75,7 +75,7 @@ steps: agents: queue: "cuda" command: | - julia --project=test/cudaenv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.develop(path="lib/TimespanLogging"); Pkg.instantiate()' + julia --project=test/cudaenv -e 'using Pkg; Pkg.develop([PackageSpec(path=pwd()), PackageSpec(path="lib/TimespanLogging")]); Pkg.instantiate()' julia --project=test/cudaenv test/run_mpi.jl 2 2 test/mpi_cuda.jl - label: Julia 1.11 (ROCm, MPI) @@ -86,7 +86,7 @@ steps: agents: queue: "rocm" command: | - julia --project=test/rocmenv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.develop(path="lib/TimespanLogging"); Pkg.instantiate()' + julia --project=test/rocmenv -e 'using Pkg; Pkg.develop([PackageSpec(path=pwd()), PackageSpec(path="lib/TimespanLogging")]); Pkg.instantiate()' julia --project=test/rocmenv test/run_mpi.jl 2 2 test/mpi_rocm.jl - label: Julia 1.11 (oneAPI, MPI) @@ -97,7 +97,7 @@ steps: agents: queue: "oneapi" command: | - julia --project=test/oneapienv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.develop(path="lib/TimespanLogging"); Pkg.instantiate()' + julia --project=test/oneapienv -e 'using Pkg; Pkg.develop([PackageSpec(path=pwd()), PackageSpec(path="lib/TimespanLogging")]); Pkg.instantiate()' julia --project=test/oneapienv test/run_mpi.jl 2 2 test/mpi_oneapi.jl - label: Julia 1.11 (Metal, MPI) @@ -109,7 +109,7 @@ steps: agents: queue: "metal" command: | - julia --project=test/metalenv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.develop(path="lib/TimespanLogging"); Pkg.instantiate()' + julia --project=test/metalenv -e 'using Pkg; Pkg.develop([PackageSpec(path=pwd()), PackageSpec(path="lib/TimespanLogging")]); Pkg.instantiate()' julia --project=test/metalenv test/run_mpi.jl 2 2 test/mpi_metal.jl - label: Julia 1.11 (Finch) diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml index 23bc7b537..dc97d0954 100644 --- a/.github/workflows/CI.yml +++ b/.github/workflows/CI.yml @@ -437,7 +437,7 @@ jobs: # feature and is silently ignored on 1.10 (LTS), which would otherwise # resolve a stale registered Dagger lacking the MPI API (`accelerate!`). - name: Instantiate CPU MPI environment - run: julia --project=test/mpienv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.develop(path="lib/TimespanLogging"); Pkg.instantiate()' + run: julia --project=test/mpienv -e 'using Pkg; Pkg.develop([PackageSpec(path=pwd()), PackageSpec(path="lib/TimespanLogging")]); Pkg.instantiate()' - name: Run CPU MPI tests run: julia --project=test/mpienv test/run_mpi.jl ${{ matrix.ranks }} 2 test/mpi.jl @@ -462,6 +462,6 @@ jobs: arch: x64 - uses: julia-actions/cache@v3 - name: Instantiate OpenCL MPI environment - run: julia --project=test/openclenv -e 'using Pkg; Pkg.develop(PackageSpec(path=pwd())); Pkg.develop(path="lib/TimespanLogging"); Pkg.instantiate()' + run: julia --project=test/openclenv -e 'using Pkg; Pkg.develop([PackageSpec(path=pwd()), PackageSpec(path="lib/TimespanLogging")]); Pkg.instantiate()' - name: Run OpenCL MPI tests run: julia --project=test/openclenv test/run_mpi.jl 2 2 test/mpi_opencl.jl From 62e81d89ee258415663c1cd412e30df178d5bf72 Mon Sep 17 00:00:00 2001 From: Julian P Samaroo Date: Sat, 12 Sep 2026 12:11:36 -0700 Subject: [PATCH 6/7] CI: Fix resolver errors in the TimespanLogging and DaggerWebDash jobs Two more resolver bugs in the same family as 33eff720, both on a from-scratch checkout with no cached Manifest.toml: - "Julia 1 - TimespanLogging" called Pkg.instantiate() *before* Pkg.develop(path="lib/TimespanLogging"). Instantiate resolves the root project's declared TimespanLogging = "0.2" compat straight against the registry, which only has 0.1.0-0.1.1 published -- unsatisfiable, since the local 0.2.0 doesn't exist as a registered release yet. Fix: develop before instantiate, matching every other job's order. - "Julia 1 - DaggerWebDash" did three sequential Pkg.develop calls (Dagger, then TimespanLogging, then DaggerWebDash). Confirmed this hits the same class of bug as 33eff720 -- and NOT just on Julia 1.10: reproduced on 1.12 too with a from-scratch depot. Batching into one Pkg.develop([...]) call fixes the sequencing issue, but surfaces a real, separate conflict once TimespanLogging correctly resolves to its dev-pinned 0.2.0: DaggerWebDash's own compat pins `TimespanLogging = "0.1"`, which excludes it. DaggerWebDash's actual usage (MultiEventLog, LogWindow, Events.creation_hook/deletion_hook, init_similar) is all still present in TimespanLogging 0.2, so this is a compat-bound update, not an API fix: widen to "0.1, 0.2". Verified both against a from-scratch JULIA_DEPOT_PATH: the TimespanLogging job's fixed command runs its full suite (72/72), and DaggerWebDash's fixed command resolves, precompiles, and builds DaggerWebDash and its extensions cleanly (the one remaining test failure is an unrelated, pre-existing race between the D3Renderer's HTTP server startup and an immediate curl probe with no wait/retry -- last touched in 738ea299, well before this branch). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GaPaEAUCJMwpQjFFPbU7Ck --- .buildkite/pipeline-julia.yml | 4 ++-- .github/workflows/CI.yml | 4 ++-- lib/DaggerWebDash/Project.toml | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/.buildkite/pipeline-julia.yml b/.buildkite/pipeline-julia.yml index a035c48fc..67a28348c 100644 --- a/.buildkite/pipeline-julia.yml +++ b/.buildkite/pipeline-julia.yml @@ -154,7 +154,7 @@ steps: version: "1" - JuliaCI/julia-coverage#v1: codecov: true - command: "julia --project -e 'using Pkg; Pkg.instantiate(); Pkg.develop(;path=\"lib/TimespanLogging\"); Pkg.test(\"TimespanLogging\")'" + command: "julia --project -e 'using Pkg; Pkg.develop(;path=\"lib/TimespanLogging\"); Pkg.instantiate(); Pkg.test(\"TimespanLogging\")'" - label: Julia 1 - DaggerWebDash <<: *test @@ -164,7 +164,7 @@ steps: version: "1" - JuliaCI/julia-coverage#v1: codecov: true - command: "julia -e 'using Pkg; Pkg.develop(;path=pwd()); Pkg.develop(;path=\"lib/TimespanLogging\"); Pkg.develop(;path=\"lib/DaggerWebDash\"); include(\"lib/DaggerWebDash/test/runtests.jl\")'" + command: "julia -e 'using Pkg; Pkg.develop([PackageSpec(path=pwd()), PackageSpec(path=\"lib/TimespanLogging\"), PackageSpec(path=\"lib/DaggerWebDash\")]); include(\"lib/DaggerWebDash/test/runtests.jl\")'" - label: "Benchmarks (vs master)" timeout_in_minutes: 120 diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml index dc97d0954..89d7968dd 100644 --- a/.github/workflows/CI.yml +++ b/.github/workflows/CI.yml @@ -131,7 +131,7 @@ jobs: arch: x64 - uses: julia-actions/cache@v3 - name: Test TimespanLogging - run: julia --project=. -e 'using Pkg; Pkg.instantiate(); Pkg.develop(;path="lib/TimespanLogging"); Pkg.test("TimespanLogging")' + run: julia --project=. -e 'using Pkg; Pkg.develop(;path="lib/TimespanLogging"); Pkg.instantiate(); Pkg.test("TimespanLogging")' daggerwebdash: name: Julia 1 - DaggerWebDash @@ -146,7 +146,7 @@ jobs: arch: x64 - uses: julia-actions/cache@v3 - name: Test DaggerWebDash - run: julia -e 'using Pkg; Pkg.develop(;path=pwd()); Pkg.develop(;path="lib/TimespanLogging"); Pkg.develop(;path="lib/DaggerWebDash"); include("lib/DaggerWebDash/test/runtests.jl")' + run: julia -e 'using Pkg; Pkg.develop([PackageSpec(path=pwd()), PackageSpec(path="lib/TimespanLogging"), PackageSpec(path="lib/DaggerWebDash")]); include("lib/DaggerWebDash/test/runtests.jl")' benchmarks: name: Benchmarks (vs master) diff --git a/lib/DaggerWebDash/Project.toml b/lib/DaggerWebDash/Project.toml index a50d19e22..fd6f79317 100644 --- a/lib/DaggerWebDash/Project.toml +++ b/lib/DaggerWebDash/Project.toml @@ -25,5 +25,5 @@ Mux = "0.7, 1" ProfileSVG = "0.2" StructTypes = "1" Tables = "1" -TimespanLogging = "0.1" +TimespanLogging = "0.1, 0.2" julia = "1.6" From 1ab71b0080ffb51ddc75b6903d306eaf6bec7719 Mon Sep 17 00:00:00 2001 From: Julian P Samaroo Date: Sat, 12 Sep 2026 14:16:09 -0700 Subject: [PATCH 7/7] TimespanLogging: Restore get_logs!'s docstring The pre-rewrite version had a docstring directly on the LocalEventLog-specific get_logs! method; during the collect-time rewrite it got repurposed into a docstring on the LocalEventLog type itself, leaving the get_logs! generic function with no docs at all. Broke the Documenter build: Error: no docs found for 'get_logs!' in `@docs` block in docs/src/api-timespanlogging/functions.md:11-15 Add a docstring to the general ctx/sink-dispatching entry point, describing the current per-sink projection behavior instead of the old single-sink implementation. Verified docs/make.jl builds clean (no `no docs found` / no docs_block error). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GaPaEAUCJMwpQjFFPbU7Ck --- lib/TimespanLogging/src/types.jl | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/lib/TimespanLogging/src/types.jl b/lib/TimespanLogging/src/types.jl index 6ef8ae6c5..a8d3dadca 100644 --- a/lib/TimespanLogging/src/types.jl +++ b/lib/TimespanLogging/src/types.jl @@ -99,6 +99,18 @@ chunk lists and consumed at `get_logs!`. """ struct ActiveLog end +""" + get_logs!(ctx; kwargs...) + get_logs!(sink; kwargs...) + +Get the logs recorded by `ctx`'s (or `sink`'s) log sink, clearing them in the +process. Recording is always into per-thread buffers regardless of sink; +`get_logs!` is what steals and projects them into the sink's own log shape +(a `Vector{Timespan}` for `LocalEventLog`, a `Dict` of consumer name to +per-event values for `MultiEventLog`, `nothing` for `NoOpLog`). See the +sink's own docstring for sink-specific keyword arguments (e.g. `raw` and +`only_local` on `LocalEventLog`). +""" get_logs!(ctx; kwargs...) = get_logs!(log_sink(ctx); kwargs...) write_event(::NoOpLog, event::Event) = nothing get_logs!(::NoOpLog) = nothing