perf(scheduler): improve utilization and scale#810
Conversation
Signed-off-by: Eric W. Tramel <eric.tramel@gmail.com>
Signed-off-by: Eric W. Tramel <eric.tramel@gmail.com>
Signed-off-by: Eric W. Tramel <eric.tramel@gmail.com>
Persist resume metadata after the first durable row group and refresh it at completion. Resume already recovers progress from parquet files, avoiding repeated metadata scans and writes without adding a tuning knob. Add focused cadence coverage and document the resume contract.
Greptile SummaryThis PR improves async dataset scheduling throughput and large-run state handling. The main changes are:
|
| Filename | Overview |
|---|---|
| packages/data-designer-engine/src/data_designer/engine/dataset_builders/scheduling/completion.py | Adds compacted dropped-row state for checkpointed row groups while preserving terminal drop and completion queries. |
| packages/data-designer-engine/src/data_designer/engine/dataset_builders/async_scheduler.py | Waits for row-group workers to finish before checkpointing and compacts tracker state only after checkpoint success. |
| packages/data-designer-engine/src/data_designer/engine/dataset_builders/dataset_builder.py | Writes resume metadata after the first durable row group, refreshes it at completion, and updates already-complete resumes from disk. |
| packages/data-designer-engine/src/data_designer/engine/dataset_builders/scheduling/queue.py | Bounds fair-queue ordering state to active task groups instead of retaining historical heap entries. |
| packages/data-designer-engine/src/data_designer/engine/dataset_builders/scheduling/task_policies.py | Lets solo groups borrow the full resource capacity while preserving peer priority through borrow debt. |
Reviews (2): Last reviewed commit: "docs(scheduler): align checkpoint and bo..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| def is_dropped(self, row_group: int, row_index: int) -> bool: | ||
| dropped_mask = self._compacted_dropped.get(row_group) | ||
| if dropped_mask is not None: | ||
| byte_index, bit_index = divmod(row_index, 8) | ||
| return 0 <= byte_index < len(dropped_mask) and bool(dropped_mask[byte_index] & (1 << bit_index)) | ||
| return row_index in self._dropped.get(row_group, set()) |
There was a problem hiding this comment.
When a row group has been compacted, is_dropped() always reads _compacted_dropped before _dropped. If a late salvage or failure path calls drop_row() for the same compacted group, the new drop is added to _dropped but this changed branch keeps returning the old bitset, so the row can remain visible as not dropped while the tracker still reports the group complete.
Context Used: Do not suggest defensive coding patterns such as a... (source)
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/data-designer-engine/src/data_designer/engine/dataset_builders/scheduling/completion.py
Line: 138-143
Comment:
**Compacted Drops Shadow Later Drops**
When a row group has been compacted, `is_dropped()` always reads `_compacted_dropped` before `_dropped`. If a late salvage or failure path calls `drop_row()` for the same compacted group, the new drop is added to `_dropped` but this changed branch keeps returning the old bitset, so the row can remain visible as not dropped while the tracker still reports the group complete.
**Context Used:** Do not suggest defensive coding patterns such as a... ([source](greptile.json))
How can I resolve this? If you propose a fix, please make it concise.
Code Review: PR #810 —
|
2 participants
Summary
Improves async generation throughput, model-slot utilization, and large-run scheduler scaling with four targeted changes. The implementation is net +16 nonblank production LoC and adds no public API, dependency, feedback controller, or user tuning knob.
Changes
Performance vs.
mainEnd-to-end throughput and utilization
Full-job reservation measurements count each physical endpoint once, including shared-model endpoints. The reported unit is endpoint concurrent-service-capacity slot-hours, a GPU-hours proxy until the inference owner supplies the endpoint-to-GPU allocation.
mainWork-conserving admission removes speculative idle capacity while remaining practically equivalent for both slow-before-fast and fast-before-slow calibrated flows. It cannot preempt work already running: in the measured worst case, a late peer started up to 350 ms later when it arrived behind a 400 ms task; borrow debt still gives it the next released slot.
Large-run scheduler and memory scaling
mainThe pathological one-row-group-per-record case retains one compact terminal marker per group: 70.511 MiB at one million groups. With normal buffer sizes, retained completion state is effectively constant relative to record count.
Metadata checkpointing
mainThe shared-endpoint and hourglass cells are practically equivalent on wall time within a ±2% margin, while the heterogeneous fork/join cell improves both throughput and utilization.