Worker pool coordination across tasks#

How concurrent worker counts compose across tasks competing for shared resources. The mechanism is a per-task requires declaration plus a global resources budget consumed by the runner.

This replaces daisy 1.x’s per-task num_workers (which had no notion of cross-task resource sharing — each task got its own static pool, and 5 tasks each declaring num_workers=10 would happily spawn 50 workers regardless of the host’s actual capacity).

API#

Each task declares what one of its workers consumes:

extract = daisy.Task(..., requires={"cpu": 1}, max_workers=8)
predict = daisy.Task(..., requires={"gpu": 1, "cpu": 4}, max_workers=4)
relabel = daisy.Task(..., requires={"cpu": 1}, max_workers=8)

The runner is given a global budget:

daisy.run_blockwise(
    extract + predict + relabel,
    resources={"cpu": 32, "gpu": 8},
)

At any moment, the number of concurrent workers for a task is bounded by the smaller of:

  • its own max_workers cap,

  • the global budget after subtracting other tasks’ usage.

How it composes#

The allocator (daisy-core/src/resource_allocator.rs) maintains:

pub struct ResourceAllocator {
    budget: ResourceBudget,                  // global cap
    used:   HashMap<String, i64>,            // currently-consumed totals
    alive:  HashMap<String, usize>,          // worker count per task
}

try_allocate(&task) is the gate every worker spawn goes through. It checks that for every key in task.requires, used[k] + v ≤ budget[k]; on success, the per-key usage and the per-task alive counter both bump. release(&task) is the symmetric drop. Spawn paths in daisy-core/src/server.rs call try_allocate before launching a thread; the worker’s RAII ExitNotifier calls release on Drop.

A task with empty requires={} is treated as costing zero resources — it’s bounded only by max_workers. Useful for IO-bound tasks where you don’t want resource accounting at all.

Disjoint resources run in parallel#

Two tasks competing for the same key share that key’s pool:

budget = {"cpu": 4}
A.requires = {"cpu": 1}, A.max_workers = 8
B.requires = {"cpu": 1}, B.max_workers = 8

→ combined alive count for A and B never exceeds 4.

Two tasks declaring disjoint keys don’t compete:

budget = {"cpu": 4, "gpu": 2}
A.requires = {"cpu": 1}, A.max_workers = 4
B.requires = {"gpu": 1}, B.max_workers = 2

→ A and B both run at full max_workers simultaneously.

This is the test matrix in tests/test_resources.py:

  • test_max_workers_caps_concurrency_without_requires — empty requires, only max_workers matters

  • test_resource_budget_caps_one_task_below_max_workers — budget overrides max_workers when smaller

  • test_two_tasks_share_a_resource — combined peak ≤ shared budget

  • test_disjoint_resources_run_in_parallel — disjoint budgets don’t gate each other

  • test_requires_exceeds_budget_hard_errors — startup validation

  • test_chained_tasks_reassign_workers_when_upstream_drains — workers shift between dependent tasks as work moves through the DAG

Validation at startup#

ResourceAllocator::validate(&[Arc<Task>]) runs once before the run loop starts. For each task, for each requires key, it checks that the per-worker requirement fits in the global budget (v ≤ budget[k]). If not, it returns DaisyError::InvalidConfig with a message naming the offending task and key:

task "predict" requires {gpu: 8} but the global budget only has {gpu: 1}

This is hard-error rather than warn-and-skip: a task whose requires doesn’t fit will never spawn a worker, and silently never running is the worst possible failure mode.

Why workers are task-bound#

A more sophisticated design would have a global pool of fungible workers that the server dispatches across tasks based on where the work is. We don’t do that, on purpose:

  • Worker-function (0-arg) workers are stateful — they’re long-running subprocesses that load a model on startup and process blocks in a loop. Reassigning a “predict” worker to handle a “relabel” block would require it to switch models, which defeats the whole point of the lifecycle.

  • Process-function (1-arg) workers don’t have that constraint, but the protocol and worker threads are uniform across both shapes — making process-function workers fungible while keeping worker-function workers task-bound would split the implementation in two.

The resource-budget approach gets you most of the benefit (no over-subscription, fair sharing under contention) without the per-task statefulness penalty. The downside is that an idle “predict” worker doesn’t get reassigned to a busy “extract” task — it just exits when there’s no more “predict” work, freeing its resource share for “extract” to grow into.

Where to look#

  • daisy-core/src/resource_allocator.rs — ResourceBudget, ResourceAllocator::try_allocate/release/validate.

  • daisy-core/src/server.rs::rebalance_workers — calls try_allocate before each spawn; per-task spawn caps come from min(task.max_workers, allocator headroom).

  • tests/test_resources.py — full behavior matrix.