diff --git a/.buildkite/pipeline-julia.yml b/.buildkite/pipeline-julia.yml index 284569e90..67a28348c 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()), PackageSpec(path="lib/TimespanLogging")]); Pkg.instantiate()' julia --project=test/openclenv test/run_mpi.jl 2 2 test/mpi_opencl.jl - label: Julia 1 - TimespanLogging @@ -153,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 @@ -163,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/.buildkite/pipeline.yml b/.buildkite/pipeline.yml index b61fbc9b4..90a7c45b1 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()), 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) @@ -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()), 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) @@ -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()), 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) @@ -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()), 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 5f9b2a108..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) @@ -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()), 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.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 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/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" 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..a8d3dadca --- /dev/null +++ b/lib/TimespanLogging/src/types.jl @@ -0,0 +1,116 @@ +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!(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 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..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) - @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=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) - @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=logged_options(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 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