API reference

ParallelManager

HPC experiment runtime for Julia.

Wraps ParamIO.jl and DataVault.jl with a unified run! that handles parallel dispatch, advisory locking, .done rollups, structured event logging, and retry. Designed to replace the recurring "glue" layer that every HPC research project re-invents.

Modules

Each file is one concern, one module-scope piece. They can be used independently.

Quick start

using ParamIO, DataVault, ParallelManager

spec  = ParamIO.load("config.toml")
keys  = ParamIO.expand(spec)
vault = DataVault.Vault("config.toml"; run="phase1")

ParallelManager.init_workers!(mode=:auto)
work_fn = key -> Dict{String,Any}("x" => compute(key))
ParallelManager.run!(work_fn, vault, keys)

Design constraints

1. work_fn is a pure function — no IO, no globals, no logging. All of that lives in the runtime.
2. There is no per-item println API. Use log_event for structured events. Per-item prints are the reason FiniteTemperature.jl used to generate 300 MB job logs.
3. The Manifest is monotonic — keys are only added, never removed. Re-computing the same root is considered a contract violation; branch into a new outdir instead.
4. Multi-master coordination is entirely delegated to DataVault.acquire_running!, which uses POSIX link() for atomic "create iff not exists" on NFS. No flock, no central service, no separate locks/ tree — .running itself is the lock.

See also

• ParamIO.canonical — the stable string form used by Manifest as a directory-safe key identity.
• DataVault.Vault, DataVault.save!, DataVault.is_done, DataVault.acquire_running! — the storage + lock layer run! delegates to.

atomic_write(write_fn, path)

Write path atomically: the target either does not exist or contains the complete output — never partially-written content.

Internally:

1. Writes to "$path.tmp.$(getpid()).$(rand(UInt32))" via write_fn(io).
2. Flushes + fsyncs the tmp file (best-effort on filesystems that don't support fsync via ccall).
3. mv(tmp, path; force=true) — POSIX rename(2) is atomic, including on NFS, so concurrent readers see either the old content or the new.

The parent directory of path is created if missing. If write_fn throws an exception, the tmp file is cleaned up and the target path is left untouched.

Returns

The path argument unchanged, so the function is chainable:

p = atomic_write("out/data.jld2") do io
    write(io, payload_bytes)
end
# p === "out/data.jld2"

Notes

• The tmp filename includes getpid() and a random UInt32 so two processes racing on the same path do not clobber each other's tmp.
• This helper is file-format agnostic: callers pass any IO-consuming closure. For JLD2 payloads, Manifest uses a similar path-level dance because JLD2.jldsave needs a path, not an IO.
• On a filesystem where fsync is not available, the ccall is caught and ignored; rename atomicity still holds.

See also: atomic_touch for empty-file markers.

atomic_touch(path)

Create an empty file at path atomically. Convenience wrapper over atomic_write for sentinel files like .done markers.

atomic_touch(joinpath(vault.outdir, "stage1", "done.marker"))

Semantics are identical to atomic_write(_ -> nothing, path) — the parent directory is created, the file appears atomically, and no tmp residue is left behind.

EventLog(path::AbstractString)

Append-only JSONL event log with a thread-safe per-EventLog lock.

Each call to log_event writes one JSON object as a single line. Concurrent writes from multiple tasks (within one master) are serialized through an internal ReentrantLock. Concurrent writes from multiple processes (separate masters) rely on POSIX O_APPEND atomicity, which is guaranteed for single write syscalls of length < PIPE_BUF (4 KiB); log_event composes each line as a single String and issues exactly one write(io, line) call to stay within that guarantee.

Fields

• path::String — target JSONL file. Parent directory is created lazily on first log_event.
• lock::ReentrantLock — protects appends from same-process races.

Event kinds used by run!

run! emits the following kind values (as strings in the JSON):

:key_start and :lock_busy are emitted at :debug level and are suppressed unless the EventLog is created with min_level=:debug (see RunOpts.log_level); their totals still appear in :stage_done. Downstream analysis (jq, DataFrame-based) can filter and aggregate over these kinds without ever parsing freeform text.

Example

log = EventLog("out/events.jsonl")
log_event(log, :stage_start; stage=:phase1, todo=3600)
# ... work ...
log_event(log, :stage_done; stage=:phase1, done=3600, err=0)

log_event(log, kind; kwargs...)

Append one JSON object to log.path with fields ts (ISO-8601 local time), kind (the Symbol converted to String), and any additional key/value pairs passed via kwargs.

The line is built in full (including the trailing newline) as a single String, then written with exactly one write(io, line) call inside an open(path, "a") block. This relies on POSIX O_APPEND atomicity so that cross-process writes do not tear each other's lines.

log_event(log, :key_done; stage=:phase1, key="N=8;J=1.0;#sample=1", secs=12.3)

produces one line like:

{"ts":"2026-04-13T14:23:51.123","kind":"key_done","stage":"phase1","key":"N=8;J=1.0;#sample=1","secs":12.3}

Returns nothing.

merge_event_logs(dir; output="events_merged.jsonl") -> String

Merge all per-master event log files (events_*.jsonl) in dir into a single sorted file. Returns the output path.

Each master writes to its own events_<host>_<pid>.jsonl; this function collects and sorts all lines by their ts field for post-hoc analysis.

Manifest(stage, root, complete)
Manifest(stage, root)

Rollup index for a single stage under a vault root. complete holds ParamIO.canonical strings of every key known to be finished.

Fields

• stage::Symbol — stage label, e.g. :phase1. Used to pick the sub-directory inside root.
• root::String — the vault root (typically vault.outdir) combined with the project name. See manifest_root for the convenience overload.
• complete::Set{String} — in-memory set of canonical key strings, rehydrated from the JLD2 file on load.

On-disk format

<root>/<stage>/manifest.jld2

JLD2 layout: one field complete::Vector{String}. A Vector is used (not a Set) so older JLD2 versions deserialize cleanly; load_manifest rehydrates to a Set{String} for O(1) lookup.

Invariants

• Monotonic. Once a key is added via add_complete! it is never removed by this package. Re-computing the same root with different work is considered a contract violation — branch into a new outdir.
• Atomic updates. save_manifest uses a tmp + rename dance so a crash mid-write cannot leave a half-written manifest on disk.
• Corrupt file = empty. If a manifest file exists but cannot be parsed by JLD2, load_manifest returns an empty Manifest. The worst outcome is re-running already-complete keys, which is idempotent thanks to the per-key .running lock and the is_done re-check in run!.

Typical use from inside run!

m = load_manifest(vault)
todo = todo_keys(m, collect(keys))
isempty(todo) && return :skip
for key in todo
    # ... do work ...
    add_complete!(m, key)
end
save_manifest(m)

See also

load_manifest, save_manifest, add_complete!, is_complete, todo_keys, manifest_path, manifest_root.

manifest_path(root, stage) -> String

Return the conventional on-disk location "$root/$stage/manifest.jld2". Pure function; does not touch the filesystem.

load_manifest(root, stage) -> Manifest
load_manifest(vault::DataVault.Vault) -> Manifest

Read manifest.jld2 if present; otherwise return an empty Manifest.

A corrupted manifest (any JLD2.load failure) is treated as empty. This prefers making progress over refusing to run — the runtime's is_done re-check after lock-acquire makes accidental re-processing harmless.

The second form is a convenience defined in Run.jl that derives (root, stage) from a DataVault.Vault (see manifest_root).

load_manifest(vault::DataVault.Vault) -> Manifest

Convenience overload of the two-argument load_manifest that derives (root, stage) from a DataVault.Vault:

load_manifest(manifest_root(vault), Symbol(vault.run))

save_manifest(m) -> String

Atomically write m.complete to manifest_path(m.root, m.stage) and return that path.

Internally uses a JLD2.jldsave to a sibling tmp path followed by mv(..., force=true), so a crash in the middle leaves either the old manifest (unchanged) or the new one — never a truncated JLD2.

m.complete is sorted on save so the on-disk Vector has a deterministic order, making jld2 files easier to diff for debugging.

add_complete!(m, key) -> Set{String}

Add ParamIO.canonical(key) to the in-memory m.complete set. Not persisted until save_manifest is called.

This is a one-way operation: the Manifest API does not expose a remove_complete! (see the monotonic invariant in the Manifest docstring).

is_complete(m, key) -> Bool

true if ParamIO.canonical(key) is in m.complete. This is the per-key check used by todo_keys; it never touches the filesystem.

todo_keys(m, keys) -> Vector{DataKey}

Return the subset of keys that are not yet in m.complete. This is the work list for the next run!; a fully-done run returns an empty vector, which triggers the early-skip path.

Cost is O(length(keys)) hash lookups — independent of filesystem state. On 3600 keys this takes ~6–12 ms in the benchmark.

manifest_root(vault) -> String

Return the directory under which run! and load_manifest look for this vault's manifest.jld2 — one manifest per (project, run).

The layout is:

<vault.outdir>/manifest/<project_name>/<vault.run>/manifest.jld2

Pure function; does not touch the filesystem.

merge_and_save_manifest!(m) -> String

Reload the on-disk manifest, merge its completed keys into m, then atomically persist. This prevents the "last writer wins" problem when multiple masters call save_manifest concurrently — without the merge, each master would overwrite the others' newly completed keys.

There is a small TOCTOU window between the load_manifest and the save_manifest; in the worst case a concurrent saver's keys are not included in this write, but they will be re-processed on the next invocation (safe due to the is_done re-check after lock-acquire).

init_workers!(; mode=:auto, master_blas=1, launch_timeout=300.0,
                worker_timeout=300, verbose=true) -> Symbol

Bootstrap worker processes / threads according to mode, and return the mode actually used (useful when mode=:auto).

Modes

Timeouts (relevant to :slurm / :distributed)

• launch_timeout::Real = 300.0 — seconds the master will wait for SlurmClusterManager.SlurmManager to produce worker addresses via srun. Large jobs (≳ 100 workers) with cold NFS package caches need a substantially larger value than the SlurmManager default (60s).
• worker_timeout::Integer = 300 — value exported as JULIA_WORKER_TIMEOUT so every freshly spawned Julia worker waits up to this many seconds for the master to send its first handshake message. The built-in Distributed default is 60s, which is too tight when 100+ workers race each other through _include_from_serialized on a shared depot.

Both defaults (300s) handle the 128-worker i8cpu case on ISSP System B comfortably. Set lower values only for local debugging.

Idempotent: calling multiple times with worker processes already present does not double-add. BLAS thread settings are always (re)applied.

detect_mode() -> Symbol

Inspect the environment to pick a default init_workers! mode:

• :slurm if SLURM_JOB_ID is present in ENV,
• :distributed if JULIA_SLURM_N_WORKERS > 0 (multi-worker without a Slurm job — e.g. a local addprocs smoke test driven by the same env var the batch scripts set),
• :threads if Threads.nthreads() > 1,
• :sequential otherwise.

This is what init_workers!(mode=:auto) delegates to. Callers rarely need to invoke detect_mode directly; it is public mainly for tests.

verify_workers!()

Probe each Distributed worker for hostname, Julia threads, BLAS threads, and CPU affinity. Prints a summary table and emits a @warn if any worker has BLAS.get_num_threads() > 1 (a common cause of OpenBLAS segfaults in multi-process Julia).

Ported from FiniteTemperature.jl Parallel/Slurm.jl::print_worker_identities.

RunOpts(; workers=:auto, max_attempts=3, stale_after=600.0,
         heartbeat_interval=60.0, stop_flag=nothing)

Execution options for run!.

Fields

• workers::Symbol = :auto — dispatch mode. :auto fans out over the Distributed WorkerPool(workers()) via pmap when nprocs() > 1, and runs sequentially otherwise. :sequential forces the sequential path even when worker processes are present (useful for debugging a serialization issue).
• max_attempts::Int = 3 — per-key retry budget. Set to 1 to disable retry (a failed work_fn is logged as :error instead of :gave_up).
• stale_after::Float64 = 600.0 — seconds before another master can reclaim a held lock as stale. Passed through to DataVault.acquire_running!.
• heartbeat_interval::Float64 = 60.0 — how often the per-lock heartbeat task refreshes DataVault's .running file. Enforced to be < stale_after (otherwise a live holder's lock could be reclaimed mid-work).
• log_level::Symbol = :info — event-log verbosity. At :info (default) the high-churn per-key :lock_busy and :key_start events are suppressed (their totals still ride in the :stage_done summary), keeping the JSONL log O(computed keys) instead of O(masters × keys) under multi-master contention. Set :debug to log them (e.g. to debug lock contention).
• stop_flag::Union{String,Nothing} = nothing — path to a sentinel file. When isfile(stop_flag) becomes true, run! and run_loop! stop dispatching new keys and return early. This is the infra equivalent of FiniteTemperature.jl's STOP_NOW_$JOB_ID mechanism, typically created by a SIGUSR1 signal handler in the batch script 60 s before Slurm kills the job.

Example

opts = RunOpts(max_attempts=5, stale_after=900.0, heartbeat_interval=30.0,
               stop_flag="/path/to/STOP_NOW_12345")
ParallelManager.run!(work_fn, vault, keys; opts)

run!(work_fn, vault, keys; opts=RunOpts(), load=nothing) -> NamedTuple

Run work_fn(key) -> Dict for every key in keys, persisting through vault. Writes a structured JSONL event log at joinpath(vault.outdir, "events_<hostname>_<pid>.jsonl") — one file per master, so concurrent masters never contend on a single log.

load names the module(s) the worker processes need beyond the always-loaded seam (ParamIO/DataVault/ParallelManager) — typically the package or module that defines work_fn and the types it touches. Accepts a Module, Symbol, String, or a collection of them (e.g. load=MyModel or load=[MyModel, Statistics]). Under :distributed/:slurm, run! usings these in Main on every worker before fan-out, so a compute script no longer has to hand-roll the for w in workers(); remotecall_fetch(…, :(using …)); end broadcast. It is a no-op on the master (nprocs() == 1) and idempotent, so it is safe even when a project still broadcasts by hand.

Early skip (todo 10): on startup a stage-level Manifest is loaded. Keys already in the manifest are skipped — when all keys are done, the second run-through takes O(1) filesystem operations regardless of length(keys).

Contract:

• work_fn is expected to be a pure function: given a DataKey, return a Dict payload to persist via DataVault.save!.
• Exceptions in work_fn are caught and logged; the corresponding key's .done file is not written, so re-runs will pick it up.
• The stage label used for logging is Symbol(vault.run).
• Manifest is monotonic: saved at end-of-stage with every newly completed key.

Parallel dispatch

If nprocs() > 1 (i.e. init_workers!(mode=:distributed|:slurm) has added worker processes), run! automatically fans out over the Distributed WorkerPool(workers()) via pmap. Each worker runs the per-key lock-acquire → work_fn → DataVault.save! → mark_done! pipeline independently. All filesystem operations (the .running lock, atomic JLD2 write, JSONL event log) are already NFS-safe, so concurrent workers inside one master are structurally consistent with multi-master operation.

If only the master is active (nprocs() == 1), run! falls back to the sequential loop from todo 11. This means the same compute.jl script is valid in three modes:

1. No init_workers! call at all → sequential on the master.
2. init_workers!(mode=:distributed) with addprocs(n) → local pmap fan-out.
3. init_workers!(mode=:slurm) inside a SLURM job → cluster fan-out.

Multi-master locking (several separate julia processes writing to the same vault) continues to work underneath either path because the lock layer (DataVault's .running) uses POSIX link() / atomic rename only.

run_loop!(work_fn, vault, keys; opts=RunOpts(),
          max_empty_rounds=3, idle_sleep=30.0, load=nothing)

Work-stealing loop that repeatedly calls run! until there is no more work to do. This is the infra equivalent of FiniteTemperature.jl's _work_loop driver.

The loop exits when:

• max_empty_rounds consecutive rounds produce zero new completions, or
• opts.stop_flag is raised (graceful shutdown).

Default parameters (max_empty_rounds=3, idle_sleep=30.0) are the battle-tested values from FiniteTemperature.jl.

load is forwarded verbatim to every run! call (see its docstring) — name the work module(s) the workers need and the loop handles the per-round broadcast.