API reference

The daisy package re-exports the public surface from internal modules. All names below are accessible directly as daisy.<name>.

Top-level helpers

daisy.run_blockwise(pipeline, multiprocessing=True, resources=None, progress=True, block_tracking=True, return_states=False, host=None, summary=True)

Run the given pipeline to completion.

pipeline must be a daisy.Pipeline (or a single Task, which is sugar for a singleton pipeline). The legacy form run_blockwise([task1, task2]) is supported by daisy.v1_compat, which converts a list of tasks into a Pipeline by reading each task’s v1.x-style upstream_tasks= kwarg.

Returns True only if every block of every task either completed successfully in this run or was skipped because a prior run already marked it done. Permanently failed or orphaned blocks cause this to return False.

With return_states=True, returns the task_id -> TaskState dict instead of the bool — per-task counters plus abandoned, abandon_reason and last_worker_error — so partial failures can be inspected programmatically without the lower-level Server class. Abandonment (worker restart cap exhausted) raises RuntimeError either way.

summary=False silences the execution summary that is otherwise printed on return, on KeyboardInterrupt and on abandonment. Each task’s figures (per-block and per-worker timing) are persisted and available afterwards from task.report() either way.

host is the address workers are told to connect to. Leave it unset and daisy detects one — this host’s resolved name, else the default-route interface, verified before use — and listens on every interface so workers on other machines can reach it. Set it when detection cannot know the answer: a multi-homed node with a specific fabric, a container whose external address differs from anything it can see, or a DNS name only the compute nodes resolve. Passing "127.0.0.1" keeps the scheduler private to this machine, which also means no remote worker can connect. DAISY_HOST sets the same value from the environment, for pipeline libraries that don’t thread the argument through.

daisy.set_done_marker_basedir(path)

Deprecated alias for set_tracking_basedir.

Renamed because the directory holds more than done state now — failure counts and per-block resource measurements live beside it.

Return type:

None

daisy.get_done_marker_basedir()

Deprecated alias for get_tracking_basedir.

Return type:

Path | None

Tasks and blocks

class daisy.Task(*args, num_workers=None, max_workers=None, upstream_tasks=None, **kwargs)

Bases: Task

v1.x-compatible Task — a Python subclass of _rs.Task that accepts the daisy 1.x kwargs num_workers=N (alias for max_workers=N) and upstream_tasks=[other_task, …] (declares per-task block-level dependencies, which v2 expresses via Pipeline composition).

Passing either kwarg emits a DeprecationWarning. The upstream_tasks are stored on this Python subclass instance (Python subclasses of #[pyclass(subclass)] get their own __dict__); daisy._task._build_pipeline_from_tasks reads this attribute when bridging a v1.x list-of-tasks call into a v2 Pipeline.

class daisy.Block(total_roi, read_roi, write_roi, task_id=None, block_id=None)

Bases: object

One unit of work: a read ROI, a write ROI, an id and a status.

total_roi only feeds the derivation of the block id; pass block_id to override it (what daisy does for blocks received off the wire). ROIs may be funlib.geometry.Roi or anything with integer offset and shape.

property read_roi

funlib.geometry.Roi this block reads from.

property write_roi

funlib.geometry.Roi this block writes to.

property block_id

(task_id, spatial_id), as in daisy 1.x.

property status

One of the BlockStatus constants; set it to FAILED to mark the block failed without raising.

property stats

Resource measurement attached by profile_block, or None.

class daisy.BlockStatus

Bases: object

class daisy.TaskState

Bases: object

Python-facing wrapper around a TaskCounters snapshot. The Rust scheduler internally tracks tasks as a typestate enum (Running/Done/Abandoned), but observers and post-run consumers only care about the counter values, so we hand them a frozen snapshot at the FFI boundary.

abandon_reason

Human-readable abandonment reason, or None if not abandoned.

Type:

str | None

abandoned

True iff the task was abandoned (restart cap exhausted, or an upstream task was abandoned) rather than run to completion.

Type:

bool

first_worker_error

The first worker error observed for this task (usually the root cause; later errors are often echoes), or None.

Type:

str | None

last_worker_error

The most recent worker error observed for this task (spawn function failure, dirty worker exit, or a reported block failure), or None if no worker ever errored.

Type:

str | None

timeout_reclaim_count

Block attempts reclaimed for exceeding the block timeout.

Type:

int

timeout_secs

The task’s configured block timeout in seconds.

Type:

float | None

Geometry

daisy.Roi and daisy.Coordinate are funlib.geometry’s classes, as in 1.x. Every daisy entry point accepts them (or anything with integer offset and shape), and every ROI daisy hands back is one.

Scheduling

class daisy.Scheduler(tasks_or_pipeline)

Bases:

v1.x-compatible Scheduler factory. Accepts a Pipeline, a Task, or a list of tasks (with upstream_tasks= declarations honoured) and returns an _rs.Scheduler.

daisy v2 internalised the scheduler to consume a Pipeline only; this shim builds a Pipeline from the list-of-tasks form by walking the v1.x upstream side-table on each task.

Return type:

Scheduler

class daisy.BlockwiseDependencyGraph(task_id, read_roi, write_roi, read_write_conflict, fit, total_read_roi=None)

Bases: object

The blocks of one task and their conflict dependencies, without a scheduler. ROIs are funlib.geometry.Roi (or anything with integer offset and shape).

class daisy.DependencyGraph(native)

Bases: object

Block-level dependencies of a scheduled pipeline, from Scheduler.dependency_graph.

Server / Client

class daisy.Server(stop_event=None)

Bases: object

Coordinator. Thin shim over _rs._run_distributed_server that wraps process functions with per-worker logging context first.

Most users go through daisy.run_blockwise(…) and read a task’s figures back with Task.report(). Instantiate this class directly when you want the whole pipeline’s last_report or last_task_order after the run.

last_report

per-task block aggregates (resource figures only for tasks with resource_tracking=True) and per-worker startup / idle / compute / teardown timing.

Type:

This run’s daisy.RunReport

run_blockwise(pipeline, resources=None, progress=True, block_tracking=True, host=None, summary=True)

Run and return the task_id -> TaskState dict. Never raises for abandoned tasks (inspect the states instead). The execution summary is printed on return and on KeyboardInterrupt; summary=False silences it.

class daisy.Client(context=None)

Bases: object

Worker-side client. Reads connection info from the DAISY_CONTEXT env var (or accepts an explicit Context), connects to the scheduler over TCP, and yields blocks via a context manager that handles status bookkeeping and routes failure tracebacks into the per-worker log.

Every acquired block is watched automatically: when the task has a block timeout (Task(timeout=…), default 600s), a watchdog kills this worker process if the block is still unreleased after that long — true preemption, even for code stuck inside C. This covers every loop built on this Client, including hand-written cluster workers; the server reclaims and retries the block on the same deadline.

The connection also runs TCP keepalive, so a worker waiting in acquire_block() notices a server whose host has vanished (cloud scale-down — no RST will ever arrive) as a connection error within a few minutes, instead of blocking forever on a half-open socket. Together with the block watchdog, this bounds an orphaned worker’s lifetime at roughly the block timeout plus keepalive detection.

A server that is already gone at construction time (connection refused) is treated as “the run has ended”: construction does NOT raise — a WARNING is logged, connected is False, and acquire_block() yields None immediately, so the canonical worker loop exits cleanly. This is the normal fate of straggler cluster jobs that start after the run drained; it must not register as a worker failure. A missing or malformed DAISY_CONTEXT still raises (KeyError / ValueError) — that’s a configuration error, not a race. Mid-run connection losses also still raise.

The acquire_block context manager is implemented in Python because it integrates with daisy.logging (the “logging” carve-out for the otherwise all-Rust runtime).

property connected: bool

False if the server was already gone at construction, or after disconnect().

class daisy.Context(**kwargs)

Bases: object

Dict-like key→string store with to_env/from_env helpers. Construct from kwargs (Context(hostname=”…”, port=12345)) or from Context.from_env() to read back the env var on the worker side. Keys and values are always coerced to str.

static from_env()

Reconstruct from the DAISY_CONTEXT env var. Raises KeyError if the variable is unset.

static from_env_string(env)

Parse a key=value:key=value context string (the to_env encoding) without touching the process environment.

to_env()

Encode as key=value:key=value for the DAISY_CONTEXT env var. Keys are emitted in insertion order isn’t preserved by HashMap; callers that depend on a specific order should not rely on the output ordering.

Return type:

str

Run report

class daisy.RunReport(tasks)

Bases: object

Reports for every task of one run, keyed by task id.

Parameters:

tasks (dict)

property measured: dict

The tasks that measured their blocks (resource_tracking=True).

property timed: dict

a worker timeline, a per-block round trip, or resource measurements. A serial task without resource tracking has none of these.

Type:

The tasks with anything to show

render(task_order=None, max_workers=10)

The Resource Utilization panel, one section per task in task_order. What needs no tracking store — the per-block round trip and the per-worker startup / idle / compute / teardown — is always shown; the in-worker measurements (wall, CPU, RSS, IO, overhead, trend) only for tasks with resource_tracking=True. Empty when there is nothing at all to show (a serial run without resource tracking) — a table of zeros is worse than no table.

Return type:

str

Parameters:

max_workers (int)

class daisy.TaskReport(task_id, blocks, workers, started_at=0.0, finished_at=0.0, tracked=False)

Bases: object

One task’s report. blocks and workers are the summaries; worker(id) and block(...) drill down.

property elapsed: float

Run wall time in seconds, start to report.

block(block)

Look a block up in the task’s tracking store.

block is a daisy.Block, a grid coordinate (tuple of ints, one per axis, anchored at total_roi) or a spatial block id (int; found by scanning the grid, so prefer the other two forms on large tasks). None when the task has no store or the block is not in it. Available on reports returned by a run in this process; a report rebuilt from JSON has no task to read the store through.

Return type:

BlockRecord | None

block_records()

Every block the tracking store has recorded (done, or with a failure count) as BlockRecord`s sorted by spatial id: the per-block counterpart of ``workers.by_id`. Per-block timing (round_trip, worker_id, sent_at, returned_at) and resource figures need resource_tracking=True; a plain done-marker store yields done and failures only. Walks the whole grid, so it is O(blocks); for very large grids read the zarr arrays in the store directly. Empty without a store; needs the task attached, like block.

Return type:

list

render(max_workers=10)

The task’s section of the Resource Utilization panel.

Return type:

str

Parameters:

max_workers (int)

class daisy._report.BlockSummary(count=0, skipped=0, failures=0, has_stats=False, round_trip=Stat(), wall=Stat(), cpu=Stat(), peak_rss=Stat(), io_read=Stat(), io_write=Stat(), overhead=Stat(), trend_slope_ms=0.0)

Bases: object

Per-block figures for one task over this run.

round_trip (dispatch → return, measured by the scheduler) needs no tracking store and is there for every distributed task. The remaining Stat fields are measured inside the worker and are empty unless the task set resource_tracking=True; overhead is the round trip minus the block’s own wall time.

Parameters:

• count (int)

• skipped (int)

• failures (int)

• has_stats (bool)

• round_trip (Stat)

• wall (Stat)

• cpu (Stat)

• peak_rss (Stat)

• io_read (Stat)

• io_write (Stat)

• overhead (Stat)

• trend_slope_ms (float)

skipped: int = 0

blocks this run skipped via done markers or a skip mask; they are included in count but contributed no timing samples

class daisy._report.WorkerSummary(count=0, unmanaged_clients=0, startup=Stat(), idle=Stat(), compute=Stat(), teardown=Stat(), exits=dict(), by_id=dict())

Bases: object

The workers daisy spawned for one task. startup and teardown have one sample per worker; idle and compute pool every worker’s per-block samples.

Parameters:

• count (int)

• unmanaged_clients (int)

• startup (Stat)

• idle (Stat)

• compute (Stat)

• teardown (Stat)

• exits (dict)

• by_id (dict)

class daisy._report.WorkerRecord(worker_id, startup=None, blocks_ok=0, blocks_failed=0, idle=Stat(), compute=Stat(), final_wait=None, teardown=None, exit=None)

Bases: object

One worker’s lifecycle. Durations in seconds; None where the phase never happened (no startup for a worker that never connected, no teardown for one that crashed or was cut off).

Parameters:

• worker_id (int)

• startup (float | None)

• blocks_ok (int)

• blocks_failed (int)

• idle (Stat)

• compute (Stat)

• final_wait (float | None)

• teardown (float | None)

• exit (str | None)

exit: str | None = None

"clean", "error", "timeout", "interrupted" or None if still running when the report was built.

class daisy._report.BlockRecord(block_id, done, failures, stats, worker_id, sent_at, returned_at)

Bases: object

What the tracking store holds for one block.

Parameters:

• block_id (tuple)

• done (bool)

• failures (int)

• stats (BlockStats | None)

• worker_id (int | None)

• sent_at (float | None)

• returned_at (float | None)

stats: BlockStats | None

daisy.BlockStats or None (no resource tracking, or never ran)

worker_id: int | None

worker that completed it, or None (unmanaged client / never ran)

sent_at: float | None

seconds since the UNIX epoch on the scheduler’s clock, or None

property round_trip: float | None

Dispatch to return in seconds, from the scheduler’s stamps.

class daisy._report.Stat(n=0, mean=0.0, std=0.0, min=0.0, max=0.0, total=0.0)

Bases: object

A measured distribution: sample count, mean, sample standard deviation, min, max and total. Empty (n == 0) when nothing was measured — it then formats as —.

Parameters:

• n (int)

• mean (float)

• std (float)

• min (float)

• max (float)

• total (float)

format(unit='s')

mean ± std in seconds (unit="s") or bytes (unit="B"); just the mean when there is a single sample.

Return type:

str

Parameters:

unit (str)

Progress observers

class daisy.JsonProgressObserver(path=None, stream=None)

Bases: object

Streaming JSON observer for monitoring/dashboards.

Emits one JSON object per task per state-changing event, with a monotonic timestamp and the full counter snapshot. Pipes cleanly into log aggregators (fluentd, journald), JSON-line tools (jq -c), or anything that reads line-delimited JSON.

Usage:

# to a file
obs = JsonProgressObserver(path="/tmp/progress.jsonl")
daisy.run_blockwise(tasks, progress=obs)

# to stdout
daisy.run_blockwise(tasks, progress=JsonProgressObserver())

# to an existing file-like object
with open("progress.jsonl", "w") as f:
    daisy.run_blockwise(tasks, progress=JsonProgressObserver(stream=f))

One line per task per call. With three tasks and an on_progress rate of thousands per second, expect a high-volume sink — the observer is the right level of granularity for dashboards but not for tail -f on a slow run.

Output schema (one line, pretty-printed for readability):

{
  "t": 1745781234.123,        # time.time() at emit
  "event": "progress",        # "start" | "progress" | "finish"
  "task": "extract",
  "total": 100000,
  "ready": 99,
  "processing": 4,
  "completed": 4321,
  "skipped": 0,
  "failed": 2,
  "orphaned": 0,
  "restarts": 1,              # worker_restart_count
  "failures": 1               # worker_failure_count (raw deaths)
}