Parallelism
Parallelism here means how many workers run the same job shape, not “turn on sklearn threads”. Almost every heavy mode shares one plan: runtime.parallelism.backend + workers (+ Ray address). What differs is the work unit — a feed shard, a shuffle bucket, a vector part, a probe candidate.
The step never picks a backend in Python. DRTML / UI parallelism_controls do.
Two channels (do not mix them)
Section titled “Two channels (do not mix them)”| Namespace | Meaning |
|---|---|
runtime.parallelism.* |
Execution workers: inprocess / processes / ray |
compute.parallelism.* |
Threads inside an algorithm call (BLAS-style), orthogonal to dispatch |
Two fan-out tracks (plus actors)
Section titled “Two fan-out tracks (plus actors)”| Track | Module | Work unit |
|---|---|---|
| Feed multi-worker | partition_loop/parallel/ |
primary feed shards |
| Offline map | processing/job_map.py |
reduce buckets, assembly shards, probe evaluate / pipeline |
| Actors (not map) | embed PS, sticky cluster fit | stateful W / shard actors |
job_map exposes map_jobs, pipeline_jobs, and assign_round_robin. Reduce and row-assembly workers open child sessions via processing/job_session.open_job_worker_session.
Parallel transport registry
Section titled “Parallel transport registry”Backends resolve through partition_loop/parallel/backend_registry.py:
- builtins:
inprocess(checkpoint/resume),processes,ray; - alias:
threads→inprocess; - extension:
register_parallel_backend/ParallelBackend(public viadrtoller.framework.ext).
ParallelPlan.from_params uses resolve_parallel_backend_name + get_parallel_backend. Checkpoint/resume still requires inprocess (ParallelPlan.require_inprocess).
Extra Ray fields on the same plan:
| Param / env | Role |
|---|---|
runtime.parallelism.num_gpus |
GPU reservation per partition Ray actor |
runtime.parallelism.cpus_per_worker |
vCPU advertised per KubeRay worker pod (default 4; UI + live CR patch) |
runtime.parallelism.max_nodes |
Cluster node cap including head (default 23; maxReplicas = max_nodes − 1) |
runtime.parallelism.inflate_timeout_s |
Sticky Ray: wait for worker CPUs before vector load (default 900) |
runtime.parallelism.inflate_poll_s |
Sticky Ray: poll interval while inflating (default 5) |
runtime.parallelism.require_object_store / DRTOLLER_REQUIRE_OBJECT_STORE |
Refuse host-local parquet stores under multi-worker / remote Ray |
runtime.parallelism.schedule |
static (fixed shard bags) or pull (workers lease the next shard). Default static; UI on processes / ray |
runtime.parallelism.deflate_idle |
Ray + pull only: shrink the cluster as workers finish. Default false |
workers is the actor count. For map-style jobs, pods ≈ ceil(actors / cpus_per_worker), capped by maxReplicas. Inflate waits for packed worker CPUs; if the replica cap is hit, actors are capped to advertised CPUs. Sticky fit uses one actor per pod. Framework patches the live RayCluster at Ray task start; partition_loop pins replicas for the run. Sticky fit also inflates replicas before load — see Ray / KubeRay.
When ray_address is set and workers > 1 (or the require flag is on), parquet storage.stores must use backend: s3, not local. Details: Ray.
What a “worker” means per mode
Section titled “What a “worker” means per mode”partition_loop — primary feed shards. Each worker rebuilds session + processor context. With schedule: static, shards are bag-assigned up front. With schedule: pull, idle workers take the next shard from a shared queue so a few heavy shards do not leave the rest of the fleet idle. Driver merges manifests and metric deltas. Checkpoint/resume and run-global mutable registries need inprocess; otherwise write partials and reduce. Pull does not add multi-worker resume.
embed_train — not shard mapping. workers=1 → one actor owns full W; workers≥2 on Ray → parameter-server shards of W. See embed_train, Ray.
reduce_merge — shuffle buckets (or one flat group). Round-robin via job_map, private write lanes, driver merges manifests. See reduce_merge.
Row assembly — optional spill-parallel path (row_assembly/spill + parallel_jobs + shard_plan). Large inputs map over row-group slices, then SUM by key over shuffle buckets via the same reduce_merge workers as a reduce job. Assembly is not probe-specific glue.
Cluster fit / assign — vector parquet shards; sticky fit uses actors / cached matrices. On Ray, sticky inflates KubeRay workers before load (one actor per pod) and releases those actors before nested evaluate. See Streaming cluster.
Probe — evaluate jobs via job_map.map_jobs; interleaved prepare→evaluate via pipeline_jobs. Nested children split the worker budget. See Probe loop.
Memory policy
Section titled “Memory policy”Hot paths must not materialize a full run, full N×D matrix, or full keyspace in Python. Prefer streaming reads, vector batches, early-shuffle spill, and SQL aggregation.
Do not size RAM from on-disk parquet MB. Do not “fix” OOM by forcing inprocess / workers=1 unless the algorithm truly needs one address space — that trades capacity for a bug that still exists at larger scale.