Per-worker time tracking and the run report

CHANGELOG.md

@@ -65,6 +65,41 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Per-worker time tracking and the run report.** Every run now produces
+  a `daisy.RunReport` (`Server.last_report`, and `daisy_run_report` on the
+  `KeyboardInterrupt` / abandonment error) with one `TaskReport` per task,
+  read back per task with `task.report()`: `blocks` (count, failures, and `mean ± std`
+  for wall, CPU, peak RSS, IO, plus the per-block `overhead` — server round
+  trip minus the worker's own wall time — and the wall-time trend) and
+  `workers` (count, `startup`, `idle`, `compute`, `teardown` as
+  distributions, exit kinds, and a row per worker under `by_id`).
+  `TaskReport.worker(id)`, `TaskReport.block(block | grid coord | id)` and
+  `TaskReport.block_records()` drill down; the latter two read the tracking
+  store (`Task.block_record`, `Task.block_records`). `blocks.skipped`
+  separates a resumed run's skipped blocks from executed ones.
+  `Task.run_blockwise(**kwargs)` runs a task on its own and `Task.report()`
+  returns its section of its most recent run's report, read from the
+  persisted `runs.jsonl`, so it works in a later process and after an
+  interrupted or abandoned run. `Task.tracking_path` is the resolved store
+  directory (None without a persistent store).
+  Worker timing is stamped by the server from the lifecycle events it
+  already sees — spawn, first acquire, each dispatch and return, the
+  shutdown reply, the spawn thread's exit — so it needs no protocol change,
+  is always collected (O(workers) memory), and covers startup and teardown,
+  which a worker cannot report about itself. The "Resource Utilization"
+  panel prints what needs no tracking store for every distributed task —
+  the scheduler's per-block `round trip` and the worker line plus a
+  per-worker table (ten rows, then a pointer) — and the in-worker
+  measurements (`wall`, `cpu`, `peak RSS`, IO, `overhead`, `trend`) only
+  for tasks with `resource_tracking=True`.
+  Every task's report is appended once per run to a `runs.jsonl`: in its
+  tracking store when it has one, else under `<log basedir>/<task_id>/`
+  next to its worker logs. With `resource_tracking=True` the tracking store
+  also gains `worker_id`, `sent_at` and `returned_at` arrays (who ran each
+  block and when, scheduler clock).
+  **Removed**: `Server.last_tracking_summary` and the `{"per_task": {...}}`
+  dict it held; `daisy_run_stats` on the partial-run errors (now
+  `daisy_run_report`).
 - `Task(skip_mask=...)`: a caller-provided mask over the block grid (a
   uint8 or bool array at `Task.block_grid_shape`, nonzero = skip). Masked
   blocks are reported as skipped and release their downstream dependencies,
@@ -349,6 +384,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **Block measurements cover the whole worker process.** `cpu_seconds` and
+  `io_*_bytes` were counted per thread (`RUSAGE_THREAD`,
+  `/proc/self/task/<tid>/io`), so work a block handed to other threads was
+  missed. zarr's IO threads are one example: a 32 MB zarr write was recorded
+  as 0 bytes. They now come from `RUSAGE_SELF` and `/proc/self/io`.
+- **`peak_rss_bytes` is the peak while the block ran.** It was read from
+  `ru_maxrss`, which survives fork+exec, so every spawned worker reported
+  its driver's peak (6.1 GB for every task of a run whose driver held the
+  volume). On Linux the high-water mark is now reset at block start and
+  `VmHWM` is read at the end. Elsewhere it falls back to `ru_maxrss`.
 - **The bundled benchmarks no longer compare daisy against itself.** The
   `gerbera -> daisy` rename left both arms of every comparison pointing at
   the same package: each script defined `bench_daisy` twice (the second

MIGRATION.md

@@ -45,7 +45,8 @@ A few things v2 added that have no 1.x equivalent:
 
 - **Block tracking as a built-in** (`Task(tracking_path=…)` or via `set_tracking_basedir(…)`; `done_marker_path` / `set_done_marker_basedir` are deprecated aliases). A per-task Zarr v3 group recording which blocks completed (used to skip already-done work on resume) and how many times each failed. `Task.reset()` clears it for an explicit re-run. Tracking is **opt-in**: without an explicit path or a basedir set, nothing is tracked or skipped. The recommended pattern is one `set_tracking_basedir("<stable absolute path>")` call at pipeline start. Note that tracking is keyed by task_id + block geometry, *not* by your code — rerunning changed code against existing tracking resumes silently, so use distinct task_ids per pipeline version or `Task.reset()` after code changes. It is also bound to the exact task layout: changing any ROI (including growing `total_roi` for a dataset that acquired more data) invalidates it entirely; in-place migration for grown volumes is a planned enhancement (see `docs/source/design/DONE_MARKERS.md`).
 - **Skip masks** (`Task(skip_mask=…)`): a uint8 or bool array with one cell per block (`Task.block_grid_shape`; cell `(0, 0, …)` is the first block), nonzero = skip. Masked blocks count as skipped and release their dependents, but are not recorded as done, so the mask is a per-run input: a later run with a different mask (or none) runs them. Composes with tracking: done blocks stay done, masked blocks stay undone. Use it for blocks you know carry no work (far from a predicted surface, outside an acquired region) instead of a `check_function`, which runs a Python callable per block and cannot be seeded ahead of a run.
-- **Per-block resource tracking** (`Task(resource_tracking=True)`, needs a tracking path). Each block reports its wall time, CPU time, peak RSS and IO bytes, measured inside whoever ran it — so thread, subprocess and external cluster workers all report the same numbers. Values land in the tracking group as sibling arrays indexed the same way as `done`, readable with plain `zarr.open_group(...)`, and the run prints an agglomerated **Resource Utilization** summary. daisy measures the whole block body automatically; `with daisy.profile_block(block):` narrows the scope if you want to exclude your own setup. Programmatic access: `daisy.Server().last_tracking_summary`. 1.x had no equivalent.
+- **Per-block resource tracking** (`Task(resource_tracking=True)`, needs a tracking path). Each block reports its wall time, CPU time, peak RSS and IO bytes, measured inside whoever ran it — so thread, subprocess and external cluster workers all report the same numbers. Values land in the tracking group as sibling arrays indexed the same way as `done`, readable with plain `zarr.open_group(...)`, and the run prints an agglomerated **Resource Utilization** summary. daisy measures the whole block body automatically; `with daisy.profile_block(block):` narrows the scope if you want to exclude your own setup. Programmatic access: `task.report()` (a `TaskReport` with `mean ± std` per metric, read from the persisted `runs.jsonl`). 1.x had no equivalent.
+- **Per-worker time tracking** (always on). The scheduler stamps every worker's lifecycle — spawn → first acquire (**startup**, including cluster queue time), the gap before each block (**idle**), each round trip (**compute**), and shutdown reply → exit (**teardown**) — and the run report carries them per task (`task.report().workers`, distributions over workers and blocks) and per worker (`.workers.by_id[i]`, or `.worker(i)`). Printed in every summary along with the scheduler's per-block round trip (neither needs a tracking store); `resource_tracking=True` adds the in-worker measurements to the panel and persists `worker_id` / `sent_at` / `returned_at` per block in the tracking group and the report is appended to `runs.jsonl` (in the tracking store, else next to the worker logs) so `task.report()` reads it back in any later process; `task.report().block(block)` reads one block's record back. 1.x had no equivalent.
 - **`block_tracking` flag on `run_blockwise`** to disable done-marker tracking entirely for a run.
 - **`JsonProgressObserver`** — line-delimited JSON progress events, useful for piping into log aggregators or external dashboards.
 - **Per-task resource budgets** (`requires` per task + global `resources` dict) — coordinate worker concurrency across tasks competing for shared resources (CPU, GPU, etc.).

daisy-core/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "daisy-core"
-version = "2.0.0"
+version = "2.0.1"
 edition = "2024"
 description = "Blockwise task scheduler for processing large volumetric data — Rust core for the daisy Python package"
 license = "MIT"

daisy-core/src/block_bookkeeper.rs

@@ -43,33 +43,34 @@ impl BlockBookkeeper {
         client_addr: SocketAddr,
         timeout: Option<Duration>,
         worker_id: Option<u64>,
+        sent_at: Instant,
     ) {
         debug!(block_id = %block.block_id, %client_addr, "block sent to client");
         self.sent_blocks.insert(
             block.block_id.clone(),
             BlockLog {
                 block,
                 client_addr,
-                time_sent: Instant::now(),
+                time_sent: sent_at,
                 timeout,
                 worker_id,
             },
         );
     }
 
-    /// Record that a block was returned by a client. Returns the time
-    /// the block spent in the worker (`Instant::now() - time_sent`)
-    /// when the return is valid, otherwise `None`.
+    /// Record that a block was returned by a client. Returns the instant
+    /// the block was sent and the id of the worker that held it when the
+    /// return is valid, otherwise `None`.
     pub fn notify_block_returned(
         &mut self,
         block: &Block,
         client_addr: SocketAddr,
-    ) -> Option<std::time::Duration> {
+    ) -> Option<(Instant, Option<u64>)> {
         if let Some(log) = self.sent_blocks.get(&block.block_id) {
             if log.client_addr == client_addr {
-                let elapsed = log.time_sent.elapsed();
+                let sent = (log.time_sent, log.worker_id);
                 self.sent_blocks.remove(&block.block_id);
-                return Some(elapsed);
+                return Some(sent);
             }
             debug!(
                 block_id = %block.block_id,