DataHelix: A CPU-GPU Co-Scheduling Data Plane for Multimodal Model Training

DataHelix: A CPU-GPU Co-Scheduling Data Plane for Multimodal Model Training Yuanhe Li1

Jiazhen Wu2

Dian Ding1

Guangtao Xue1

1 Shanghai Jiao Tong University

2 Xiamen University

Abstract

GPU Step 1

Multimodal model training breaks the loose coupling between data loading and model execution assumed by conventional training pipelines. Large per-sample variation in decoding cost, token/latent count, and tensor size causes both CPU preprocessing time and GPU compute time to vary substantially across data-parallel ranks (training processes). In synchronous training, these two sources of heterogeneity interact tightly: GPU execution determines how long the input pipeline has to prepare the next batch, while delayed preprocessing in any rank in turn stalls all GPUs at synchronization. As a result, optimizing CPU throughput or GPU load balance in isolation cannot eliminate pipeline bubbles. We present DataHelix, a metadata-first CPU–GPU coscheduling data plane for multimodal training. DataHelix jointly reasons about preprocessing demand and GPU execution progress, treating each GPU compute window as the readiness budget for the next batch and coordinating preprocessing across ranks to minimize deadline misses rather than CPU makespan. To avoid expensive data movement, placement is decided while samples remain lightweight descriptors, before decoding and collation amplify them into large tensors. Descriptor-level token-/latent-aware batching performs coarse pre-balancing, while cost-gated collective rebalancing corrects residual skew only when its expected benefit exceeds transfer cost. Across 11 video and multimodal workloads on 8 Ascend 910B accelerators, DataHelix reduces mean deadline-miss rate by 67.8% and mean p95 inputwait latency by 87.0%, while achieving 2.72× geometric-mean throughput over rank-local DataLoaders and 1.82× over a Ray actor pool under the same CPU budget.

1

Yiming Zhang1

Rank 0

GPU Step2 Delay Time

CPU Step 1

CPU Step 2 All Reduce

… GPU Step 1 Rank N

CPU Step 1

GPU Step2

CPU Step 2 (a) Rank-local DataLoader GPU Step 1

Rank 0

CPU Step 1

GPU Step2

CPU Step 2

… GPU Step 1 Rank N

CPU Step 1

GPU Step2

CPU Step 2 All Reduce (b) DataHelix

Figure 1. CPU–GPU coupling: preprocessing must meet perrank readiness deadlines set by GPU execution.

Equal sample count no longer implies equal preprocessing time, much less equal GPU compute time. The imbalance has a global consequence: in synchronous data-parallel training, the slowest rank’s preprocessing delay determines when all ranks advance, leaving faster ranks idle at the gradient synchronization barrier while a single straggler drags down cluster-wide GPU utilization—not merely a local CPU preprocessing issue but an architectural problem cascading into global GPU efficiency. Figure 2 shows cross-rank latency variation and long-tail waits in native synchronous Qwen-VL training with batch size 2 and eight ranks. Per-rank step-time p95 ranges from 15.849 s to 22.829 s, a 1.44× gap. Separately supplied step measurements provide a complementary view: in Step 50, Rank 3 waits 61.006 s for data, while Rank 6 spends 60.298 s in synchronization. These measurements motivate coordinating batch readiness across ranks to limit input stalls and synchronization waiting, rather than optimizing aggregate preprocessing throughput alone. Fundamentally, multimodal heterogeneity tightly couples CPU data production with GPU execution. While the GPU processes batch 𝑡, the input pipeline has only that compute window to prepare batch 𝑡 +1; conversely, if any rank falls behind, the resulting data stall propagates to all ranks through synchronization. GPU progress therefore determines the preprocessing budget, while preprocessing readiness determines

Introduction

Conventional training input pipelines assume sample count proxies computational cost, so equal sample distribution suffices for load balance. That holds for homogeneous text or image workloads but breaks down entirely in multimodal large language model (MLLM) training such as long-video understanding, VLM pretraining, and diffusion transformer (DiT) fine-tuning. Within one batch, preprocessing costs vary by orders of magnitude: CPU decoding latency scales superlinearly with video duration and resolution, spatial transformation cost grows with pixel count, and the resulting sequence lengths—visual tokens in autoregressive VLMs or spatial latents in DiT architectures—diverge by 10–100×. 1

Step time p95 (s)

Max / Min = 1.440× 25 22.673 20

18.290

Challenge 1: Preprocessing cost cannot be known before execution. Token and latent counts, CPU decoding latency, and output tensor sizes are unknown until preprocessing completes, yet decisions must be made at the descriptor stage before any tensor is materialized. A scheduler relying on static estimates risks systematic mis-scheduling; waiting for exact measurements adds unacceptable latency. Challenge 2: Placement must track synchronized consumption. Work stealing [3] and heterogeneous-cluster straggler mitigation [28] improve task completion. For training inputs, placement must also consider when each rank consumes its batch: a lightly loaded remote worker may still deliver too late after decoding and tensor transfer. A readiness miss can then propagate through synchronization, motivating per-rank deadline scoring alongside load and locality. Challenge 3: Data amplification makes postmaterialization migration prohibitively expensive. A sample occupies negligible storage as a descriptor but expands by orders of magnitude after decoding and collation: a short clip may take kilobytes at the index stage yet produce gigabytes of decoded RGB tensor. Deferring cross-node rebalancing until after materialization can cost more in transfer than re-decoding locally, and the benefit threshold is non-monotone in both sample size and network conditions. Challenge 4: Integrating with the GPU execution loop without introducing synchronization overhead. Deadline-aware scheduling requires real-time feedback from the GPU execution layer—the remaining compute window and per-rank progress—yet coupling the data pipeline to the training process risks extra synchronization barriers or compromised fault isolation. We present DataHelix, a CPU–GPU Co-Scheduling Data Plane for multimodal foundation model training that reframes the input system as a cluster-level data plane using GPU compute windows to set per-rank readiness targets. At its core, DataHelix decouples control from data movement: the Coordinator makes placement and routing decisions at the descriptor stage before any tensor is materialized, while DataServers execute preprocessing and deliver payloads. A deadline-aware scheduler routes requests to minimize per-rank deadline misses, token/latent-aware batching equalizes GPU compute load, and cost-gated rebalancing corrects residual skew using collective communication. An asynchronous pipeline overlaps preprocessing with GPU execution without extra barriers. Together these mechanisms preserve standard DataLoader compatibility while addressing the architectural gap between rank-local loading and deadline-aware scheduling. DataHelix has been seamlessly integrated into Pytorch and MindSpore [9]. We make the following contributions. • We identify CPU–GPU coupling as a key bottleneck in synchronous multimodal training and show that the relevant objective is per-rank data readiness along the GPU

22.451 22.792 22.829 22.738

20.573 15.849

15 10 5 0

0

1

2

3

4

5

6

7

Rank

(a) Per-rank step-time p95. Data wait

Compute

Step 42

Rank 3

12.106

Rank 6 0.043

1.371

Sync wait Step 50

1.165

0.001

11.877

Data wait: 281.5× (R3 / R6)

⋯

61.006

⋯ 0.125 1.305

0.713 0.001 60.298

Data wait: 488.0× (R3 / R6)

(b) Selected steps (s; widths not to scale).

Figure 2. Qwen-VL rank skew and long waits in native synchronous training (batch size 2, eight ranks). whether GPU execution can advance. Figure 8 makes this coupling visible in multimodal VLM training: under rank-local DataLoaders, data wait and synchronization wait account for a substantial fraction of the summed rank-0 component summaries. These summaries diagnose waiting behavior rather than measured end-to-end step latency. The relevant objective is thus not aggregate preprocessing throughput or CPU makespan, but keeping per-rank data production aligned with the GPU execution timeline. Prior work supplies important building blocks. DynaPipe [11] adapts microbatches and pipeline execution to variable sequence lengths. tf.data service [2] disaggregates preprocessing, Pecan [8] dynamically places workers across local and remote CPUs, and MegaScale-Data [30] coordinates multisource loading under heterogeneous preprocessing costs. Thus, remote execution and global coordination are established techniques; our focus is coupling input placement to each rank’s batch-readiness target. This coupling requires accounting for when and where data becomes expensive to move. FastFlow [25] already considers network bandwidth in offloading decisions. For heterogeneous multimodal batches, we jointly consider candidate-specific preprocessing and delivery costs against the consuming rank’s GPU window, make initial assignments before tensor materialization, and gate later corrections by their transfer cost. Coordinating these decisions calls for a CPU–GPU CoScheduling Data Plane: a globally coordinated substrate that uses the GPU compute window to set a per-step readiness target, decides placement and routing at the descriptor stage before any tensor is materialized, and jointly optimizes CPU allocation and inter-node transfer cost across all ranks. Realizing this vision surfaces four interrelated challenges. 2

execution timeline, rather than aggregate CPU throughput. We present DataHelix, a cluster-level CPU–GPU co-scheduling data plane that minimizes deadline misses across ranks. • We propose a metadata-first, deadline-aware placement architecture that combines descriptor-level cost estimates with runtime CPU, network, and GPU feedback to coordinate preprocessing before tensor materialization. • We design a two-tier load-balancing mechanism that combines token/latent-aware pre-balancing with cost-gated HCCL/NCCL-based residual correction, together with an asynchronous pipeline that overlaps preprocessing with GPU execution without extra barriers. • We evaluate DataHelix on an 8-card Ascend 910B server across 11 video and multimodal workloads, DataHelix reduces mean deadline-miss rate by 67.8% and mean p95 input-wait latency by 87.0%, while improving geometricmean throughput by 2.72× over rank-local DataLoaders and 1.82× over a Ray actor pool under the same CPU budget. Beyond controlled evaluation, DataHelix has been upstreamed into the data engine of a widely used opensource AI framework with millions of downloads and tens of thousands of community contributors, and reported production deployments across thousands of enterprise users provide further evidence of its practicality at scale.

2

3. Preprocess

4. Encoder

5. Tokens

6. Training

video + Metadata

sample frames + decode

Resize/Crop + Normalize

ViT/ Video Encoder

Visual/Text

LLM/DiT

GPU Pipeline

Figure 3. Data path of multimodal training. size approximates per-rank computation uniformly. In video, VLM, and DiT training, this assumption breaks down: a short clip may produce a handful of visual tokens, while a long or high-resolution sample may yield substantially more tokens, patches, or latent dimensions. Equal sample counts, therefore, translate to unequal GPU compute times, producing stragglers under synchronous training. CPU preprocessing cost exhibits the same heterogeneity—decode and transform latency varies with codec, frame count, resolution, and sampling strategy, and is no longer linear in sample count. Observation 2: Batch readiness depends on the GPU compute window. Plumber [14] and cedar [31] optimize pipeline throughput; Cachew [7] uses worker and trainingclient feedback to avoid data stalls. Our scheduling target is per-rank batch readiness: preparation and delivery of batch 𝑡 + 1 should fit within batch 𝑡’s GPU window to avoid a pipeline bubble: 𝑡 prep (batch𝑡 +1 ) + 𝑡 delivery (batch𝑡 +1 ) ≤ 𝑡 GPU (batch𝑡 ) Simple worker load balancing is insufficient: a DataServer with a short queue is not necessarily optimal if its I/O cost, output tensor size, or network path cannot satisfy the deadline. Observation 3: Post-materialization rebalancing incurs prohibitive communication cost. Video samples undergo severe data amplification along the preprocessing pipeline: a descriptor occupies a few hundred bytes, a raw compressed video occupies megabytes, and a decoded, normalized tensor may reach tens to hundreds of megabytes. This asymmetry has a direct implication for scheduling: placement and routing decisions made at the descriptor stage incur negligible communication overhead, whereas load balancing deferred until after materialization may generate inter-node transfer costs that negate any straggler reduction benefit. This asymmetry demands that placement decisions be finalized at the descriptor stage and that any post-materialization correction be gated by a cost model.

Background

Data Path of Multimodal Training

Framework loaders such as PyTorch DataLoader [21] and tf.data [18] expose the transformations in Fig. 3. Metadata probe produces lightweight descriptors: duration, resolution, frame count, file path, and sampling plan. Decode and resize/crop consume CPU time according to codec, frame count, and resolution; collate and delivery depend on materialized tensor size. The visual encoder and downstream LLM or DiT dominate GPU compute. Preprocessing studies [10] show why placement must account for these different resource demands. 2.2

2. Decode

CPU Pipeline

This section characterizes the multimodal training data path, analyzes limitations of existing input systems, and summarizes the problems DataHelix addresses. Unlike homogeneous image workloads, multimodal training exhibits pronounced per-sample heterogeneity in preprocessing cost, token count, and tensor volume, making the input pipeline a first-class determinant of GPU utilization and training throughput; large-scale production studies similarly show that data storage and ingestion can dominate training efficiency [32]. 2.1

1. Raw Data

2.3

Problem Statement and Design Goals

Based on the preceding analysis, we formalize the core problem: given a synchronous multi-GPU training job over heterogeneous multimodal samples, determine batch composition and DataServer placement at the descriptor stage—before any tensor is materialized—such that (1) each GPU rank receives an approximately equal token/latent compute load; (2) the full preprocessing pipeline for batch 𝑡 + 1 completes within the GPU compute window of batch 𝑡; (3) post-materialization rebalancing executes only when its

Limitations of Existing Input Systems

Observation 1: Sample count is no longer a valid proxy for workload. In conventional image classification, batch 3

Rank 0 Training GPU Ranks （GPUs） DataHelix Iterator

projected straggler reduction exceeds inter-node transfer cost; and (4) the system preserves standard DataLoader compatibility while reducing communication overhead and improving end-to-end throughput. These constraints map directly to three design goals: G1 addresses constraint (4) by decoupling control from data movement; G2 addresses constraint (2) through deadlineaware placement; and G3 jointly addresses constraints (1) and (3) via two-tier load balancing. G1 (High-performance data plane): Decouple control from data movement—the Coordinator handles only lightweight metadata while DataServers execute preprocessing and deliver payloads via shared memory, streaming, or peer transfer—eliminating serialization overhead on the critical path while preserving standard DataLoader compatibility. G2 (Deadline-aware CPU scheduling): Select DataServer placement by estimating end-to-end completion time per candidate, accounting for queue wait, I/O, CPU, output size, and network conditions, with all routing finalized at the descriptor stage before materialization. G3 (Two-tier GPU-side load balancing): Combine token/latent-aware batch construction—a zeromigration coarse pre-balance at the descriptor stage—with cost-gated post-materialization rebalancing (DDLB) that corrects residual rank skew using true sample statistics only when projected straggler reduction exceeds inter-node transfer cost.

3

Coordinator DataServer

Rank N-1

GPU

GPU

…

DataHelix Iterator

Data Flow

GPU Compute

DataHelix Iterator

LLM/DiT Training

DataHelix Plane Runtime Feedback Sec 3.5

Asynchronous Prefetch

Sec Metadata-First 3.2 Control Descriptor Intake

Sec Deadline-Aware 3.3 Placement Time Prediction

Sec Two-Tier GPU 3.4 Load Balancing Profile Grouping

Route Selection

Deadline Scoring

Load Distribution

Remote Execution

Tensor Bypass

Server Selection

Balance Correction

Batch Delivery

DataServer 0 (CPU) fetch

decode

transform

collate

Index Dispatch

DataServer M-1 (CPU)

…

fetch

decode

transform

collate

Local Storage Cache ｜ Shared memory ｜ Tensor Transport（gRPC）

Multimodal Data

Descriptors

Profiles

Raw Bytes

(IDs, paths, offsets)

(cost, bytes, tokens)

(compressed)

Figure 4. DataHelix design: metadata-only scheduling with DataServer preprocessing and tensor delivery. Data Plane

Training Rank Step

Rank/DS Registry

Local DataServer

GPU Step Consume Batch

DataHelix Design

step time

control flow

Scheduler Router

index

Worker Pool Delivery/Shared Memory Remote DataServer

index

Worker Pool ready batch payload flow

payload forward

Sender Fetcher

Coordinator

queue feedback

Control Plane

batch descriptors

Client Side

Dataloader Iterator

DataHelix turns the training input path from a collection of independent rank-local DataLoaders into a cluster-level data plane that uses GPU compute windows to set per-rank readiness targets. The core idea is to decide placement while a sample is still a lightweight descriptor and to keep heavyweight tensor payloads off the control path, so that CPU-side preprocessing can be scheduled to minimize per-rank deadline misses rather than global makespan, remote CPU sidecars can be exploited when they reduce deadline risk, and the training code still sees the ordinary DataLoader iterator interface. Figure 4 summarizes this philosophy. The Coordinator forms a metadata-only control loop that receives descriptor requests, uses cost and runtime feedback to choose an execution DataServer minimizing predicted deadline miss, and never touches decoded tensors. DataServers form the payload path: they execute dataset operations, exchange materialized batches via shared memory or HCCL/NCCL collectives when rebalancing is beneficial, and deliver batches toward the consuming rank’s readiness target. 3.1

Rank 1

Tensor Transport runtime feedback

Figure 5. Runtime architecture: descriptor routing, direct payload delivery, and scheduling feedback. load and runtime feedback but handles only metadata; DataServers execute preprocessing and deliver tensors directly. This separates iterator compatibility, placement control, and payload delivery without centralizing tensor traffic. Client Side. Each rank keeps the familiar DataLoader iterator interface: a background sender issues descriptor-level batch requests ahead of the training thread, while a fetcher receives completed batches, preserves iterator order, and stages ready results for GPU consumption. The programming model is unchanged—the loop still calls next(iterator)— yet dispatch and result collection are no longer serialized. Control Plane. The Coordinator maintains registered clients, local and remote sidecar DataServers, rank-toDataServer mappings, queue indices, and heartbeat state. For each descriptor request it picks the execution DataServer via the policy of Section 3.3; when that server is remote it attaches the target local DataServer to the descriptor so the materialized result reaches the right delivery endpoint. This gives the scheduler a global view of DataServer load and remote CPU availability without putting the Coordinator on the tensor-transfer path.

Coordinator–DataServer Architecture

Like tf.data service [2] and MegaScale-Data [30], DataHelix separates preprocessing from training. Figure 5 specializes this architecture for descriptor placement against per-rank readiness targets. The Coordinator observes DataServer 4

Data Plane. Each DataServer owns a worker pool running dataset fetch, video decode, image transforms, tokenization, collation, and exception propagation, with workers returning materialized results through shared-memory references. A local DataServer delivers directly to its co-located training rank, whereas a remote sidecar forwards the payload to the target local DataServer over the configured transport; when post-materialization rebalancing triggers, local DataServers use HCCL/NCCL collectives to redistribute at acceleratorinterconnect bandwidth rather than CPU network speed. Heavyweight tensors thus bypass the Coordinator and move only between DataServers and training ranks. This split has three advantages: placement decisions use cluster-level visibility rather than one rank-local queue; remote CPU capacity can be exploited when it reduces deadline risk while the common local path stays lightweight and free of a centralized payload bottleneck; and local DataServers leverage training-specific collective communication for highspeed rebalancing, cutting redistribution latency by 99% versus gRPC-based alternatives. 3.2

large buffers through the configured communication backend, with small tensors inlined and full gRPC serialization retained as fallback. The control path stays lightweight while decoded payloads use shared memory or peer transfer. 3.3

Deadline-Aware Placement

Latency-constrained scheduling is established in serving systems such as Clipper and Nexus [5, 24]. Here the target comes from the consuming training rank’s GPU window: each descriptor is routed before materialization by comparing predicted batch-ready times against that target. Time prediction. For each candidate DataServer 𝑖, DataHelix predicts when the requested batch would become ready if assigned there. The prediction is candidate-specific: a remote sidecar may have more idle CPU but adds forwarding cost, while a local DataServer avoids network transfer but can sit behind a deeper worker queue. Predicted finish time decomposes into CPU preprocessing, storage or input I/O, queueing delay, and materialized-payload transfer cost:

Metadata-First Control

𝑐𝑝𝑢

𝑇𝑖 = 𝐶𝑖

This layer optimizes decision timing and transfer volume. For video and multimodal samples, the cheapest moment to make a placement decision is before decode and collation, when the request is still a compact descriptor; once a tensor is materialized, load balancing may require moving tens or hundreds of megabytes across nodes. Metadata-first control therefore means more than sending metadata: it moves routing decisions before materialization and leaves tensor movement to the DataServer data plane only when needed. Descriptor intake. The client sends only sampler indices and batching descriptors to the Coordinator. The DataLoader object is assigned lazily to a DataServer the first time that server executes the client’s dataset, but the per-step request itself remains compact—important for multimodal workloads, where a video descriptor is a few bytes of indices and metadata while the decoded tensor can reach tens or hundreds of megabytes. Route selection. For each descriptor request the Coordinator selects a DataServer before tensor materialization— balancing across the client’s associated DataServers by default, scoring all candidates with the cost model of Section 3.3 when deadline-aware scheduling is on. If the selected server is not the requester’s local DataServer, the Coordinator records the local target in the descriptor so the result reaches the right delivery point. Tensor bypass. Batch payloads bypass the Coordinator. Locally, the DataServer delivers shared-memory metadata through a queue and the training process opens the referenced segment. Remotely, the execution DataServer extracts tensors from the Python batch structure, sends a compact structure description and tensor metadata, and transfers

𝑞𝑢𝑒𝑢𝑒

+ 𝐶𝑖𝑖𝑜 + 𝐶𝑖

𝑀𝑖 = max(0,𝑇𝑖 − 𝐷).

+ 𝐶𝑖𝑛𝑒𝑡 ,

(1)

These terms come from offline sample profiles and online heartbeat state: descriptor metadata such as frame count, resolution, input bytes, and locality hints give a 𝑐𝑝𝑢 pre-materialization estimate of 𝐶𝑖 , 𝐶𝑖𝑖𝑜 , and output size, 𝑞𝑢𝑒𝑢𝑒 while heartbeats refine 𝐶𝑖 and 𝐶𝑖𝑛𝑒𝑡 from current queue depth, recent task latency, and peer-link bandwidth. Deadline scoring. In synchronous training, however, lowest predicted finish time is not the only objective: a candidate helps only if it can make the next batch ready inside the current GPU compute window. Given a data deadline 𝐷, the same scoring rule computes the predicted miss 𝑀𝑖 in Equation 1. Work stealing [3] and LPT [6] provide established mechanisms for balancing work and reducing completion time. We specialize placement to synchronized consumption: the score combines predicted lateness, locality, and backpressure. When all candidates meet the readiness target, locality wins; when all miss, earliest-finish wins. This makes the cost of decoded-payload delivery part of the decision, rather than selecting a worker solely by queue length. Server selection. The final selection balances locality against deadline risk: if the local DataServer should meet the deadline, DataHelix keeps the request local unless a remote candidate offers enough predicted gain, whereas a local path likely to miss makes a remote sidecar attractive despite the transfer cost. Online heartbeat feedback closes the loop—as queues grow or peer links congest, predicted finish times rise and future descriptors shift away from overloaded paths. 5

Algorithm 1 Two-Tier GPU-Side Load Balancing.

coarse balance that changes which descriptors each rank requests rather than moving decoded tensors afterwards. The same interface supports microbatch fanout, splitting one logical batch into smaller descriptor requests merged before the iterator yields. Balance correction. Descriptor-level planning cannot remove all skew, because real decode cost, augmentation behavior, and tensor sizes differ from static estimates. DataHelix therefore adds a residual correction tier in the DataServer runtime—not merely a callback, but the DDLB execution path that buffers a small round of materialized samples, extracts lightweight features, gathers them across participating DataServers to turn an assignment decision into sends, receives, and keeps, and performs the exchange via the high-speed collectives of Section 3.5. The policy layer stays configurable—users supply the feature function and assignment policy while DataHelix handles round coordination and redistribution—and in the current evaluation a cost gate keeps the current assignment unless the projected straggler reduction justifies post-materialization transfer, leaving profile-aware grouping and coarse rank balancing to the data-side batch planner.

Require: Sample set S, rank set R, cost profiles P, DDLB feature function 𝑓lb , DDLB assignment policy 𝑔lb Ensure: Per-rank assignment 𝐴 : S → R and redistribution plan Π // Tier 1: descriptor-level profile-aware assignment 1: G ← BucketByProfile(S, P) ⊲ group by (frame_count × resolution) 2: 𝜏𝑟 ← 0 ∀ 𝑟 ∈ R ⊲ per-rank token accumulator 3: for each profile group 𝐺 ∈ G do 4: for each sample 𝑠 ∈ 𝐺 do 5: 𝑟 ∗ ← arg min𝑟 ∈ R 𝜏𝑟 ⊲ greedy: route to least-loaded rank 6: 𝐴(𝑠) ← 𝑟 ∗ 7: 𝜏𝑟 ∗ ← 𝜏𝑟 ∗ + P [𝑠].token_estimate 8: end for 9: end for // Tier 2: policy-driven post-materialization correction 10: B ← MaterializeBuffer(𝐴) ⊲ small round of decoded samples 11: F ← AllGather(𝑓lb (B)) ⊲ user-defined lightweight features ˆ︁ ← 𝑔lb (F ) 12: 𝐴 ⊲ custom policy may rebalance or keep current assignment ˆ︁ 13: Π ← PlanRedistribution(𝐴, 𝐴) ⊲ derive sends, receives, and keeps ˆ︁ 14: 𝐴 ← 𝐴 15: return (𝐴, Π) 3.4

1 def sort_fn(features, args): 2 if projected_gain(features) < args.min_gain: 3 return keep_current_assignment(features) 4 return rebalance_assignment(features) 5 6 DistributedDataLoader.launch( 7 load_balance_config={ 8 "feature_fn": feature_fn, 9 "sort_fn": sort_fn, 10 "client_ranks": [[0, 1, 2, 3]], 11 "buffer_size": 64, 12 "args": policy_args, 13 } 14 )

Two-Tier GPU Load Balancing

DynaPipe [11] balances variable-length microbatch execution, while WLB-LLM and Zeppelin [4, 26] coordinate computation and communication across parallel training. We apply cost-aware balancing to the input path in two stages (Algorithm 1). Tier 1 assigns lightweight descriptors using token/profile estimates before decoding. Tier 2 uses actual sample properties to correct residual skew only when projected savings justify moving decoded tensors. The DDLB runtime handles feature collection, assignment planning, and HCCL/NCCL redistribution; feature extraction and assignment policies remain customizable. Profile grouping. Sample count is a weak proxy for GPU work in video and multimodal training, so DataHelix uses descriptor-level cost and profile information to build more representative batches before any tensor is materialized. For mixed video workloads the planner groups samples by profile—frame count and resolution—and preserves the intended mixture in the measured window, preventing a token-sorted prefix from containing mostly large clips. Load distribution. Within each global step the planner assigns profile-grouped samples to ranks while tracking perrank token totals and per-profile counts—a zero-migration

Listing 1. User-Defined DDLB function

3.5

High-Speed Redistribution via HCCL/NCCL

When post-materialization rebalancing triggers, the cost of cross-DataServer tensor transfer decides whether redistribution helps or hurts. Generic data planes use gRPC or TCP, which can take tens of seconds for gigabyte-scale decoded video tensors; DataHelix instead leverages HCCL/NCCL—the same collective infrastructure used for gradient synchronization—to redistribute samples among local DataServers at accelerator-interconnect bandwidth. Dual-backend architecture. The main DataServer group uses gRPC with a CPU-based backend (gloo) for controlplane coordination: heartbeats, descriptor routing, remote sidecar integration. The load-balancing subgroup, containing only local DataServers, uses HCCL (Ascend) or NCCL (NVIDIA) for high-speed tensor redistribution, each local DataServer bound to a unique accelerator device so transfers proceed over NVLink, PCIe, or HCCS rather than the CPU network stack. 6

time

DataSever Fetcher

t+1 t+2 t+3 t+4 t+1

t+5

t+2

t+3

t+4

t+1

t+2

t+3 t+4

prefetch batch

dispatch

DataHelix Interface

compute batch t+1 compute batch t+2

decode

Sec 4.2 Control Plane Coordinator

t+6 t+5

t+6

order

Data Plane

DataServer

Registration

Worker Pool

Global State

Request Routing

Result Handling

Request Routing

Sec 4.5

Figure 6. Asynchronous input preparation and ordered delivery overlap GPU execution (prefetch depth 4).

Sec 4.3

result request Sec 4.4 Async Iterator

Scheduler

feedback Cost Profile & Feedback

Cost Profiles

Online Feedback

notification & recovery Sec 4.6 Failure Handling Detection

Mitigation

Guarantee

Figure 7. Overview of DataHelix implementation.

Redistribution protocol. When Tier 2 rebalancing comˆ︁ the redistribution planner deterputes a new assignment 𝐴, mines which samples each DataServer must send and receive. Because local DataServers share a common address space within the LB subgroup and each binds to a distinct device, DataHelix uses collective all_to_all rather than pointto-point transfers: samples are packed into contiguous send buffers per destination rank and one collective completes the exchange in a single round. Impact. Redistributing 64 decoded video samples across 8 local DataServers takes roughly 74.5 seconds with gRPCbased one-by-one forwarding (the legacy fallback when HCCL initialization fails) but under 1 second with HCCL all_to_all—a 99% reduction. The gain is training-specific, exploiting infrastructure already present for gradient synchronization rather than general-purpose CPU network optimization, and the dual-backend design keeps remote sidecars on gRPC while enabling high-speed local rebalancing when the projected benefit justifies movement. 3.6

Dataset

register

t+6 t+5

DistributedDataLoader

User Code

data

Sender

compute batch t

task

Training

Batch delivery. A background fetcher thread consumes index responses and waits for the matching result from either the local DataServer queue or the gRPC data path. It preserves iterator order via task identifiers, optionally stages ready batches to the target device with non-blocking transfer, and places them in a client-side result queue. The training thread simply calls next(iterator) and receives an ordinary batch once the prefetcher finishes any required synchronization, so users keep the standard DataLoader programming model while DataHelix pipelines descriptor dispatch, remote execution, tensor transfer, and batch delivery behind the current step.

4

Implementation

We implement DataHelix as an extension of the DataLoader execution path of PyTorch and MindSpore, adding global scheduling and remote preprocessing without requiring any changes to model training code. The publicly shareable portion of the implementation is available in an anonymous repository [1].

Asynchronous Prefetch

This pipeline optimizes critical-path overlap while preserving iterator transparency. Even with good placement, the GPU can stall if descriptor dispatch, remote execution, tensor transfer, and batch assembly all start only after the training thread calls next(). DataHelix therefore splits request submission from result consumption: the sender issues future descriptors ahead, DataServers prepare them while the current GPU step runs, and the fetcher restores ordered delivery to the standard DataLoader interface (Figure 6). Index dispatch. A background sender thread reads the sampler, optionally splits a logical batch into microbatch descriptor requests, and sends them to the Coordinator subject to an in-flight limit. While the accelerator consumes batch 𝑡, descriptors for batch 𝑡 + 1 and beyond can already be queued at DataServers. Remote execution. Each DataServer enqueues assigned descriptors to its worker pool. Workers execute the dataset operation and return shared-memory-backed results to the dispatcher, which forwards them through the metadata-first tensor path when the request carries a remote target and delivers them locally otherwise.

4.1

Programming Interface

The DistributedDataLoader wrapper preserves the standard PyTorch DataLoader iterator interface. Optional arguments configure asynchronous prefetch, prefetch depth, microbatch fanout, and scheduler policy. Datasets can expose descriptor-based costs through estimate_cost(index); otherwise, DataHelix uses conservative defaults. 4.2

Control Plane

The Coordinator maintains global routing state in a gRPC service. Clients register their rank ID and dataloader object; DataServers register their address, rank, and heartbeat state. Requests use least-request routing by default or the cost-aware scheduler in Sec. 3.3. For remote execution, each request carries the requesting rank’s local DataServer as its delivery target. The execution DataServer forwards the materialized batch there, and the client response identifies this target, preserving local queue/shared-memory delivery. 7

4.3

Data Plane

under the same CPU budget, testing whether a generic remote execution substrate can match DataHelix with equal resources. We begin with a latency breakdown to show the basic benefits and trade-offs of the data plane before presenting aggregate performance. We additionally examine whether the performance advantage persists as per-GPU batch size increases (Figure 13).

Each DataServer dispatches a global task ID, serialized dataloader reference, auto-collation flag, and index list to its fixed worker pool. Workers use the ordinary dataset fetch path and serialize results into shared memory for reconstruction or forwarding. Remote forwarding sends batch-structure metadata and tensor descriptors first, with large tensors sent through the communication backend when enabled; small tensors and fallback cases travel inline. The target DataServer reconstructs batches into the client’s local result queue. Delivery follows client task IDs regardless of worker completion order; worker errors propagate as iterator exceptions. 4.4

5.1

Platform. Main experiments use one 8-card Ascend 910B server; multi-node experiments use 4 × 2 A100 GPUs. The native baseline launches eight active training ranks with ranklocal DataLoader workers; DataHelix also uses eight active training ranks but decouples data preparation into eight local DataServers plus a variable number of remote sidecar DataServers (denoted rd), sweeping rd8, rd16, and rd24. DataHelix runs were validated to use the intended communication contract: local delivery uses the shared-memory/queue path, while remote tensor payloads use the communication path rather than a large gRPC blob fallback. Equal CPU budget. For the Ray baseline, we use the same e2e workload configuration and launch 96 Ray actors/CPU slots. DataHelix rd16 uses 8 local and 16 remote DataServers with 4 CPU cores per DataServer, giving the same total budget of (8 + 16) × 4 = 96 CPU cores. Thus, the Ray comparison controls for provisioned CPU capacity while comparing the two data-pipeline designs on matched workloads. Equal budget refers to allocated CPU resources; CPU utilization and achieved throughput remain outcomes of each system. Workloads. Table 1 summarizes the two workload classes. Video Training covers video model training on Kinetics and OpenVid, including Kinetics small, mid, large, and mixed decode profiles, where mixed alternates 16-frame 224p, 32frame 336p, and 64-frame 448p samples. Multimodal SFT/PT covers video-text fine-tuning or pretraining with Qwen-VLstyle models over Kinetics and OpenVid. The class label only groups modalities: results are reported per concrete workload rather than averaged within a class. Metrics. We report aggregate throughput (samples/s or videos/s), step latency, input-wait latency, DDL miss rate, scheduler overhead, dispatch breakdown, and crossDataServer transfer volume. DDL miss rate is the fraction of measured steps whose input wait exceeds the configured data deadline. NPU utilization serves only as a secondary sanity check, since sampling can be sparse in short runs. For the 11-workload baseline summary, throughput uses a geometric mean across workloads; latency and DDL-miss metrics use arithmetic means. Each main-study cell uses 20 warmup and 100 measured steps, and the ablation follows the same convention with Kinetics mixed balanced across small, mid, and large samples.

Async Iterator Pipeline

The pipeline in Sec. 3.6 uses an index-sender thread bounded by the inflight limit and a fetcher thread that places ready batches into the iterator queue. Training blocks only when this queue is empty. Microbatch fanout stays within the client: it splits each logical batch into ordinary index requests and waits for all parts before yielding the merged batch, requiring no special Coordinator or DataServer protocol. Variable-shape video tensors are padded before concatenation. 4.5

Cost Profiles and Feedback

Offline video profiles record decode profiles, estimated CPU time, encoded input and decoded output sizes, and locality in a normalized JSONL table. The Coordinator aggregates per-sample entries for each request. DataServer heartbeats report worker capacity, task load and latency, and peer-link bandwidth, latency, and inflight traffic to update queueing and network estimates. Assignment telemetry records candidate scores, placement decisions, reasons, and scheduling overhead for evaluating dispatch, cross-DataServer transfer, and scheduler overhead. 4.6

Failure Handling

Worker exceptions reach the training iterator so corrupt videos or missing samples fail the responsible step instead of hanging the fetcher. Successful benchmark runs write metrics, synchronize ranks, and write the rank-0 summary before shutting down services, forcibly terminating them if graceful shutdown times out; this cleanup does not affect the measured data path.

5

Experimental Setup

Evaluation

We evaluate DataHelix along three axes: a main study comparing a tuned native PyTorch DataLoader against DataHelix data plane over the complete workload matrix; a controlled ablation isolating transport-stack separation, asynchronous prefetch with token-aware batching, and costaware DataServer scheduling; and a Ray actor pool baseline 8

Table 1. Evaluation workload classes and concrete workloads. Data-only experiments measure the data plane in isolation; e2e experiments additionally include model forward, backward, and optimizer updates when a paired training model is available. Workload

Component summary (s)

Data wait Native 4 3 2

Configuration

Compute Ray

1.39s 1.28s

Kine-L

Kine-mixed

OpenVid-M

5.3 Figure 8. Qwen-VL rank-0 latency summaries. Compute includes gradient sync; bar totals sum components, not measured end-to-end step latencies. 5.2

Kine-M

than DataHelix rd16 (0.055s), whereas DataHelix has lower synchronization wait (0.270s versus 0.386s) and a lower summed component time (1.280s versus 1.391s). The basic benefit is therefore workload-dependent: DataHelix reduces waiting on Kinetics-M and offers a different balance of latency components on OpenVid-M, without improving every component relative to Ray. These rank-0 summaries diagnose component-level behavior; they are not cross-rank critical-path measurements. The bar-top labels sum the displayed component summaries and should not be interpreted as measured end-to-end step latency or used to infer throughput. We next assess end-to-end throughput and deadline compliance across the workload matrix.

3.05s

Kinetics-M

40

Figure 9. Data-only throughput on Kinetics profiles.

1 0

60

Kine-S

Sync wait DataHelix (rd16)

1.73s

80

0

3.97s

1.62s

DataHelix rd24

20

S/M/L/Mixed video-text data; Kinetics Qwen-VL-3B SFT/PT Mixed video-text data; Qwen-VL-3B OpenVid SFT/PT

Multimodal SFT/PT

DataHelix rd16

100

S/M/L/Mixed video profiles; Kinetics VideoMAE / Wan-1.3B Mixed video profiles; VideoMAE / OpenVid Wan-1.3B

Video Training

DataHelix rd8

Data-only Kinetics throughput

samples/s

Class

Native

Main Performance Results

Figure 9 isolates the data plane by replacing model training with a fixed synthetic window. On Kinetics, DataHelix improves median data-preparation throughput by 2.85× at rd8, 7.45× at rd16, and 2.22× at rd24, peaking at 8.24× on Kinetics-mid: rank-local workers leave CPU parallelism unused, while a cluster-level DataServer pool absorbs decode, resize, and collation. The result is not monotonic in rd—rd16→rd24 often lowers throughput, as extra remote workers add contention that offsets the CPU capacity. Figure 10 shows end-to-end throughput across the sweep. Restricting each setting to the 10 workloads with real Native and DataHelix results at rd8, rd16, and rd24, the median speedup is 2.53× at rd8, 2.53× at rd16, and 2.10× at rd24, peaking at 8.47× on Kinetics-mixed with VideoMAE at rd16. At rd16 Kinetics gains most (median 3.29×) while OpenVid reaches 2.21× (maximum 2.61×)—expected, since Kinetics decode and resolution profiles create more CPU-side skew for remote preprocessing to hide. Remote DataServer scaling. The sweep shows rd8 already produces broad e2e gains and rd16 usually gives the best balance, while rd24 can regress: OpenVid-mixed with Wan reaches 2.21× at rd16 but falls below the native baseline at rd24. A data plane should therefore not simply maximize

Latency Component Breakdown

We first examine how the data plane changes latency in Qwen-VL training. We decompose rank-0 component summaries into data wait (request to local batch ready), compute (forward-backward, optimizer, and gradient synchronization), and synchronization wait, comparing the rank-local Native DataLoader, Ray, and DataHelix rd16 on Kinetics-M and OpenVid-M. Figure 8 reports the three-component summaries for Native, Ray, and DataHelix rd16. On Kinetics-M, DataHelix rd16 reduces data wait from 0.147s to 0.037s and synchronization wait from 2.532s to 0.295s relative to Native, reducing the summed component time from 3.968s to 1.618s. Ray is intermediate on this workload (0.117s data wait, 1.652s synchronization wait, and 3.053s summed time), showing that matched CPU capacity alone does not reproduce the component profile of DataHelix. On OpenVid-M, the comparison exposes a trade-off rather than a uniform latency win. Ray has lower data wait (0.037s) 9

Native rd0

DataHelix rd8

DataHelix rd16

DataHelix rd24

E2E training throughput 60

samples/s

50 40 30 20 10 0 Kine-L Qwen-VL

Kine-L VideoMAE

Kine-M Qwen-VL

Kine-M VideoMAE

Kine-S Qwen-VL

Kine-S Kine-mixed Kine-mixed OpenVid VideoMAE Qwen-VL VideoMAE Qwen-VL

OpenVid VideoMAE

OpenVid Wan

Figure 10. End-to-end throughput across workloads and remote DataServer counts. Table 2. Aggregate end-to-end comparison across 11 workloads. Throughput uses a geometric mean; other metrics use arithmetic means. Ratios are relative to Native. Bold/underline values denote the best/2nd -best results, respectively. Metric

Native

Ray actor96

Tput. GM↑ 6.86 (1.00×) 10.24 (1.49×) Wait p95↓ 70.848s (1.00×) 7.902s (0.11×) Ready p95↓ 0.219s (1.00×) 0.311s (1.42×) DDL miss↓ 0.506 (1.00×) 0.554 (1.10×)

Multi-node Comparison. Figure 12 reports a follow-up multi-node comparison on Kinetics-mixed and OpenVidmixed workloads. Across the 6 matched workloads, DataHelix improves throughput over Native by 1.36× geometric mean while reducing the average DDL miss rate from 37.5% to 21.2%, indicating that the control/data-plane split remains useful when remote DataServers sit on separate machines rather than being emulated as sidecars.

DataHelix rd16 18.66 (2.72×) 9.216s (0.13×) 0.129s (0.59×) 0.163 (0.32×)

5.4

Batch-Size Scaling

We vary per-GPU batch size over {2, 4, 8} and micro-batch size over {1, 2}, using 40 warmup and 100 measured steps per configuration. The throughput heatmap in Figure 13(a) shows that DataHelix sustains its throughput advantage as batches grow. Increasing batch size from 2 to 8 retains 96.7–96.9% of its throughput across the two micro-batch settings, compared with 90.6–93.4% for Native. Its throughput advantage consequently grows from 2.50–2.58× at batch size 2 to approximately 2.67× at batch size 8. Figure 13(b,c) shows that larger batches increase DDL misses and input-wait tails for both systems, but DataHelix remains substantially better in all six configurations. At batch size 8, its DDL miss rate is 37.8–42.1%, versus Native’s 78.3– 81.1%, and its input-wait p95 is 23.03–25.73 s, versus 74.58– 82.48 s. Thus, over the tested batch-size range, DataHelix maintains nearly stable throughput while limiting deadline misses and tail waiting relative to Native.

remote workers; it must account for deadline slack, queue depth, payload size, and network contention. Performance under equal CPU resources. Figure 11 combines the two end-to-end baseline comparisons over 11 matched video and multimodal workloads across Native, Ray, and DataHelix rd16 (the Kinetics-mixed/Qwen-VL row uses the user-confirmed historical supplement documented with the paper data); Table 2 summarizes the set. Under the same 96-core budget, DataHelix rd16 gains 1.82× geometricmean throughput over Ray rd96, while relative to Native Ray reaches 1.49× and DataHelix 2.72×. The advantage over Ray persists after matching CPU capacity, so more CPU alone does not reproduce DataHelix’s performance. The equal-budget advantage extends to batch readiness and deadline compliance: DataHelix reduces mean batchready p95 from Ray’s 0.311 s to 0.129 s and the mean DDL miss rate from 0.554 to 0.163. Input-wait tails trade off differently—Ray reaches a lower mean input-wait p95 of 7.902 s than DataHelix’s 9.216 s, though both beat Native’s 70.848 s. The comparison thus establishes higher throughput and better deadline compliance at equal CPU budget without implying that every latency metric improves; the ablation below attributes the combined effect to individual components.

5.5

Ablation Study

The main study measures the full system; a featureincremental ablation attributes the improvement to individual mechanisms. Its cumulative x-axis is: Base DS pool (eight active training ranks with local and remote DataServers under least-request routing), +Metadata transport (Queue/SHM local delivery, metadata-first remote forwarding, explicit 10

Native

Throughput

Ray rd96

Input wait p95

DataHelix rd16

Batch ready latency p95

DDL miss rate

Kine-L Qwen-VL Kine-L VideoMAE Kine-M Qwen-VL Kine-M VideoMAE Kine-S Qwen-VL Kine-S VideoMAE Kine-mixed Qwen-VL Kine-mixed VideoMAE OpenVid Qwen-VL OpenVid VideoMAE OpenVid Wan 0

20

40

60

samples/s

0

50

seconds

100

0.0

0.2

seconds

0.4

0.00

0.25

0.50

fraction

0.75

1.00

Figure 11. E2E baseline comparison across Native, Ray, and DataHelix. Native

Table 3. End-to-end ablation. Bold/underline: best/secondbest per workload, before rounding.

DataHelix

Throughput

DDL miss rate

Kine Qwen-VL Kine VideoMAE Kine Wan OpenVid Qwen-VL OpenVid VideoMAE OpenVid Wan

Feature

0

1

2

3

samples/s

0.0

0.2

0.4

fraction

Figure 12. Multi-node throughput and DDL misses: Native versus DataHelix on mixed-profile workloads.

Native DataHelix

Micro-batch size

(a) Throughput ↑

(b) DDL miss rate ↓

4.19

4.01

1 46.8% 59.5% 78.3% 1 33.02 53.53 74.58

2 4.25

4.18

3.97

2 47.5% 57.7% 81.1% 2 34.73 63.50 82.48

4

8

2

4

8

2

4

8

1 11.06 10.75 10.72

1 15.1% 22.2% 37.8% 1 7.27

11.28 23.03

2 10.96 10.85 10.60

2 17.6% 22.1% 42.1% 2 7.55

13.26 25.73

2

4

8

2

Per-GPU batch size 0

4

8

samples/s

4

8

2

Per-GPU batch size 12 0

50

%

Video: Kinetics / Video-1B Base DS 8.83 +Transport 8.03 +Prefetch/token 9.43 +Cost/DDL 9.87

7.792 7.908 7.877 7.430

18.9 20.2 18.9 16.5

199.6 203.5 178.5 233.0

Video: OpenVid / Video-1B Base DS 9.80 +Transport 9.53 +Prefetch/token 11.07 +Cost/DDL 11.21

6.075 5.695 5.271 4.727

27.0 24.3 19.6 17.5

131.8 131.5 121.1 183.5

Multimodal: Kinetics / Qwen-VL-3B Base DS 9.70 4.889 +Transport 10.82 4.094 +Prefetch/token 10.73 1.805 +Cost/DDL 11.13 2.267

12.9 11.9 7.9 6.4

96.0 88.8 106.4 122.6

Multimodal: OpenVid / Qwen-VL-3B Base DS 10.76 2.687 2.769 +Transport 12.84 +Prefetch/token 13.27 1.955 +Cost/DDL 12.64 0.792

7.5 7.7 6.0 5.2

65.8 62.8 58.0 78.9

(c) Input wait p95 ↓

1 4.43

2

Tput.↑ Input wait↓ DDL miss↓ Cross-DS↓ (samples/s) p95 (s) (%) (GB/100 steps)

100 0

4

8

Per-GPU batch size 30

s

60

90

Prefetch/token batching improves video throughput. Table 3 shows that prefetch/token batching produces the largest throughput increment for both video workloads, e.g., 8.03→9.43 samples/s on W2 Kinetics. Multimodal gains differ: W3 OpenVid benefits most from transport, while W3 Kinetics gains little throughput from prefetch despite a sharply shorter input-wait tail. These differences reflect the balance between preprocessing, payload movement, and training compute. Cost/DDL scheduling targets deadline behavior and tail latency. Compared with prefetch/token batching, Cost/DDL reduces DDL misses on all four workloads and

Figure 13. Batch-size scaling. Darker is better; each metric shares a color scale across methods. task identity, CommunicationBackend tensor transfer), +Prefetch/token batch (token/profile-aware batch planning, asynchronous prefetch, batch-internal microbatch fanout), and +Cost/DDL scheduler (cost-profile and deadline-aware placement with runtime feedback). The reported ablation uses eight active ranks and evaluates W2 Video Training and W3 Multimodal SFT/PT on Kinetics-mixed and OpenVid-mixed. 11

input-wait p95 on three. For W3 OpenVid, DDL misses fall from 6.0% to 5.2% and input-wait p95 from 1.955 to 0.792 s. W3 Kinetics remains the exception: p95 rises from 1.805 to 2.267 s despite fewer misses. The scheduler therefore targets deadline compliance and tail latency, with workload-dependent throughput effects. The trade-off is higher cross-DataServer movement. Cost/DDL increases Cross-DS transfer on all four workloads, e.g., 178.5→233.0 GB/100 steps on W2 Kinetics. The weights of transfer cost, remote-placement gain, and DDL miss penalty control this trade-off between communication and deadline compliance.

6

Related Work

6.1

ML Training Data Pipelines

using token/profile costs rather than sample counts, but their scheduling objects are model workloads. Our two-tier scheme (Sec. 3.4) applies this principle before preprocessing and couples residual rank balancing to the cost of moving decoded inputs. The prefetch/token-batching ablation (Sec. 5.5) evaluates the combined input-side mechanism. 6.3

Work stealing [3] balances task execution, and Ray [17] supplies distributed task and actor abstractions. DataHelix specializes placement with descriptor costs and rank readiness; the CPU-matched Ray actor-pool comparison (Sec. 5.3) tests the integrated design at equal provisioned capacity. Latency-aware scheduling is also established: Clipper [5] and Nexus [24] batch inference under latency objectives, while Orca [27] schedules generative inference at iteration granularity. Pollux [23] instead co-adapts training configurations and cluster allocations for goodput. Our deadline is an input-readiness target within a training job, with overruns causing stalls. The Cost/DDL ablation (Sec. 5.5) makes its consequence explicit: fewer deadline misses can require more cross-DataServer traffic, and need not improve every latency metric.

Disaggregation and placement. tf.data service [2] disaggregates preprocessing and coordinates reads to mitigate stragglers from unequal input sizes, while GoldMiner [29] elastically scales preprocessing workers. FastFlow [25] profiles local/remote execution and network limits to choose offloading decisions; Pecan [8] combines dynamic worker placement with transformation reordering to reduce cost. MegaScale-Data [30] addresses multisource orchestration and parallelism-aware data delivery. These systems establish the value of global coordination, which our Coordinator– DataServer architecture also uses (Sec. 3.1). Our placement decision operates on each descriptor request, combining predicted preparation and decoded-payload delivery time with the consuming rank’s readiness target (Sec. 3.3). Pipeline efficiency and readiness. CoorDL [16] reduces redundant I/O and preprocessing through coordination and caching; Cachew [7] combines caching with feedback-driven worker scaling. Plumber [14] tunes parallelism, prefetching, and caching, while cedar [31] optimizes operator execution across local and distributed resources. FFCV [15], DALI [20], and FusionFlow [12] accelerate data loading and transformations, with FusionFlow explicitly sharing preprocessing across CPUs and GPUs. These operator and resource choices affect the costs in Eq. 1; our scheduler scores their sum against each rank’s readiness target. MinatoLoader [19] tackles sample-time variability by constructing batches from readily prepared samples on one multi-GPU server. Our asynchronous iterator preserves task order (Sec. 3.6), addressing readiness through advance dispatch and placement rather than completion-driven batch construction. 6.2

Scheduling Objectives and Granularity

7

Conclusion

This paper presents DataHelix, a CPU–GPU co-scheduling data plane that moves placement decisions before tensor materialization. Its PyTorch-compatible implementation combines descriptor-level planning, metadata-first routing, DataServer-based preprocessing and delivery, asynchronous prefetching, and cost/DDL-aware placement, allowing heterogeneous input work to be scheduled against the GPU step window rather than only average worker throughput. Beyond the benchmarks above, our design has been merged into the data engine of a widely-used open-source AI framework, integrating directly with its existing dataset interfaces and model toolkits.

References [1] Anonymous. 2026. DataHelix: Anonymous Code Repository. https:// anonymous.4open.science/status/datahelix-2601. Publicly shareable portion of the implementation. [2] Andrew Audibert, Yang Chen, Dan Graur, Ana Klimovic, Jiri Simsa, and Chandramohan A. Thekkath. 2022. tf.data service: A Case for Disaggregating ML Input Data Processing. arXiv preprint arXiv:2210.14826 (2022). https://doi.org/10.48550/arXiv.2210.14826 [3] Robert D Blumofe and Charles E Leiserson. 1999. Scheduling multithreaded computations by work stealing. J. ACM 46, 5 (1999), 720–748. [4] Chang Chen, Tiancheng Chen, Jiangfei Duan, Qianchao Zhu, Zerui Wang, Qinghao Hu, Peng Sun, Xiuhong Li, Chao Yang, and Torsten Hoefler. 2026. Zeppelin: Balancing variable-length workloads in data parallel large model training. In Proceedings of the 21st European Conference on Computer Systems. 1879–1893. [5] Daniel Crankshaw, Xin Wang, Guilio Zhou, Michael J Franklin, Joseph E Gonzalez, and Ion Stoica. 2017. Clipper: A Low-Latency

Heterogeneity-aware Batch Scheduling

Sequence packing [13] and length-based dataset decomposition [22] reduce padding or attention work. DynaPipe [11] jointly plans variable-length microbatches and pipeline execution; WLB-LLM [26] and Zeppelin [4] balance model computation and communication through packing, sharding, or sequence partitioning. These approaches motivate 12

online prediction serving system. In 14th USENIX Symposium on Networked Systems Design and Implementation (NSDI 17). 613–627. [6] Ronald L Graham. 1969. Bounds on multiprocessing timing anomalies. SIAM J. Appl. Math. 17, 2 (1969), 416–429. [7] Dan Graur, Damien Aymon, Dan Kluser, Tanguy Albrici, Chandramohan A Thekkath, and Ana Klimovic. 2022. Cachew: Machine learning input data processing as a service. In 2022 USENIX annual technical conference (USENIX ATC 22). 689–706. [8] Dan Graur, Oto Mraz, Muyu Li, Sepehr Pourghannad, Chandramohan A Thekkath, and Ana Klimovic. 2024. Pecan: Cost-Efficient ML Data Preprocessing with Automatic Transformation Ordering and Hybrid Placement. In 2024 USENIX Annual Technical Conference (USENIX ATC 24). 649–665. [9] Huawei Technologies. 2020. MindSpore: An All-Scenario AI Computing Framework. https://www.mindspore.cn. [10] Alexander Isenko, Ruben Mayer, Jeffrey Jedele, and Hans-Arno Jacobsen. 2022. Where is my training bottleneck? hidden trade-offs in deep learning preprocessing pipelines. In Proceedings of the 2022 International Conference on Management of Data. 1825–1839. [11] Chenyu Jiang, Zhen Jia, Shuai Zheng, Yida Wang, and Chuan Wu. 2024. DynaPipe: Optimizing multi-task training through dynamic pipelines. In Proceedings of the Nineteenth European Conference on Computer Systems. 542–559. [12] Taeyoon Kim, ChanHo Park, Mansur Mukimbekov, Heelim Hong, Minseok Kim, Ze Jin, Changdae Kim, Ji-Yong Shin, and Myeongjae Jeon. 2023. FusionFlow: Accelerating Data Preprocessing for Machine Learning with CPU-GPU Cooperation. Proceedings of the VLDB Endowment 17, 4 (2023), 863–876. https://doi.org/10.14778/3636218.3636238 [13] Mario Michael Krell, Matej Kosec, Sergio P Perez, and Andrew Fitzgibbon. 2021. Efficient sequence packing without cross-contamination: Accelerating large language models without impacting performance. arXiv preprint arXiv:2107.02027 (2021). [14] Michael Kuchnik, Ana Klimovic, Jiri Simsa, Virginia Smith, and George Amvrosiadis. 2021. Plumber: Diagnosing and Removing Performance Bottlenecks in Machine Learning Data Pipelines. arXiv preprint arXiv:2111.04131 (2021). https://doi.org/10.48550/arXiv.2111.04131 [15] Guillaume Leclerc, Andrew Ilyas, Logan Engstrom, Sung Min Park, Hadi Salman, and Aleksander Madry. 2023. FFCV: Accelerating Training by Removing Data Bottlenecks. arXiv preprint arXiv:2306.12517 (2023). https://doi.org/10.48550/arXiv.2306.12517 [16] Jayashree Mohan, Amar Phanishayee, Ashish Raniwala, and Vijay Chidambaram. 2020. Analyzing and mitigating data stalls in DNN training. arXiv preprint arXiv:2007.06775 (2020). [17] Philipp Moritz, Robert Nishihara, Stephanie Wang, Alexey Tumanov, Richard Liaw, Eric Liang, Melih Elibol, Zongheng Yang, William Paul, Michael I Jordan, et al. 2018. Ray: A distributed framework for emerging AI applications. In 13th USENIX symposium on operating systems design and implementation (OSDI 18). 561–577. [18] Derek G Murray, Jiri Simsa, Ana Klimovic, and Ihor Indyk. 2021. tf. data: A machine learning data processing framework. arXiv preprint arXiv:2101.12127 (2021). [19] Rahma Nouaji, Stella Bitchebe, Ricardo Macedo, and Oana Balmau. 2026. MinatoLoader: Accelerating Machine Learning Training Through Efficient Data Preprocessing. In Proceedings of the 21st European Conference on Computer Systems. 2037–2053. [20] NVIDIA. 2019. NVIDIA DALI: GPU-accelerated data loading and preprocessing library. https://github.com/NVIDIA/DALI.

[21] Adam Paszke, Sam Gross, Francisco Massa, Adam Lerer, James Bradbury, Gregory Chanan, Trevor Killeen, Zeming Lin, Natalia Gimelshein, Luca Antiga, et al. 2019. Pytorch: An imperative style, high-performance deep learning library. Advances in neural information processing systems 32 (2019). [22] Hadi Pouransari, Chun-Liang Li, Jen-Hao R Chang, Pavan K Vasu, Cem Koc, Vaishaal Shankar, and Oncel Tuzel. 2024. Dataset decomposition: Faster llm training with variable sequence length curriculum. Advances in Neural Information Processing Systems 37 (2024), 36121– 36147. [23] Aurick Qiao, Sang Keun Choe, Suhas Jayaram Subramanya, Willie Neiswanger, Qirong Ho, Hao Zhang, Gregory R Ganger, and Eric P Xing. 2021. Pollux: Co-adaptive cluster scheduling for goodputoptimized deep learning. In 15th USENIX Symposium on Operating Systems Design and Implementation (OSDI 21). [24] Haichen Shen, Lequn Chen, Yuchen Jin, Liangyu Zhao, Bingyu Kong, Matthai Philipose, Arvind Krishnamurthy, and Ravi Sundaram. 2019. Nexus: A GPU cluster engine for accelerating DNN-based video analysis. In Proceedings of the 27th ACM Symposium on Operating Systems Principles. 322–337. [25] Taegeon Um, Byungsoo Oh, Byeongchan Seo, Minhyeok Kweun, Goeun Kim, and Woo-Yeon Lee. 2023. Fastflow: Accelerating deep learning model training with smart offloading of input data pipeline. Proceedings of the VLDB Endowment 16, 5 (2023), 1086–1099. [26] Zheng Wang, Anna Cai, Xinfeng Xie, Zaifeng Pan, Yue Guan, Weiwei Chu, Jie Wang, Shikai Li, Jianyu Huang, Chris Cai, et al. 2025. WLBLLM: Workload-Balanced 4D Parallelism for Large Language Model Training. In 19th USENIX Symposium on Operating Systems Design and Implementation (OSDI 25). 785–801. [27] Gyeong-In Yu, Joo Seong Jeong, Geon-Woo Kim, Soojeong Kim, and Byung-Gon Chun. 2022. Orca: A distributed serving system for Transformer-Based generative models. In 16th USENIX symposium on operating systems design and implementation (OSDI 22). 521–538. [28] Matei Zaharia, Andy Konwinski, Anthony D Joseph, Randy H Katz, and Ion Stoica. 2008. Improving MapReduce performance in heterogeneous environments. In 8th USENIX Symposium on Operating Systems Design and Implementation (OSDI 08). 29–42. [29] Hanyu Zhao, Zhi Yang, Yu Cheng, Chao Tian, Shiru Ren, Wencong Xiao, Man Yuan, Langshi Chen, Kaibo Liu, Yang Zhang, et al. 2023. Goldminer: Elastic scaling of training data pre-processing pipelines for deep learning. Proceedings of the ACM on Management of Data 1, 2 (2023), 1–25. [30] Juntao Zhao, Qi Lu, Wei Jia, Borui Wan, Lei Zuo, Junda Feng, Jianyu Jiang, Yangrui Chen, Shuaishuai Cao, Jialing He, et al. 2026. MegaScaleData: Scaling Dataloader for Multisource Large Foundation Model Training. In Proceedings of the 21st European Conference on Computer Systems. 1–18. [31] Mark Zhao, Emanuel Adamiak, and Christos Kozyrakis. 2024. cedar: Optimized and Unified Machine Learning Input Data Pipelines. Proceedings of the VLDB Endowment 18, 2 (2024), 488–502. https: //doi.org/10.14778/3705829.3705861 [32] Mark Zhao, Niket Agarwal, Aarti Basant, Bugra Gedik, Satadru Pan, Mustafa Ozdal, Rakesh Komuravelli, Jerry Pan, Tianshu Bao, Haowei Lu, Sundaram Narayanan, Jack Langman, Kevin Wilfong, Harsha Rastogi, Carole-Jean Wu, Christos Kozyrakis, and Parik Pol. 2022. Understanding Data Storage and Ingestion for Large-Scale Deep Recommendation Model Training. In Proceedings of the 49th Annual International Symposium on Computer Architecture. https: //doi.org/10.1145/3470496.3533044

13

Related documents

• HOCCL: Offloading Collective Communication from GPU Cores to Accelerate Distributed Training#1119407
• HOCCL: Offloading Collective Communication from GPU Cores to Accelerate Distributed Training#1108656
• MeanField Surrogate Modeling for Scalable Runtime Scheduling of Concurrent Heterogeneous AI Inference on Shared GPUs#822511
• Agentic CPU-GPU Scheduling for Heterogeneous AI Workloads#405626
• CONFERM: Recurrence-Aware Temporal Mapping for Multi-Cycle Multi-Context CGRAs#1157323