Skip to content

Commit db85cda

Browse files
AkshitaBclaudeTianhuaTao
committed
Add Transformer.prepare_cp_sequence_inputs (CP precursor)
Pull forward olmo-ddp's context-parallel sequence-sharding helper onto the base Transformer. It shards input_ids/labels via the existing _cp_load_balancer.batch_shard and returns the original sequence length so pipeline stages can rebuild RoPE buffers consistently. Self-contained (core already has the CP load balancer); full CP train wiring lands with the DDP train module. The rest of olmo-ddp's shared transformer/block.py + model.py diff is intentionally not taken: the shared-block norm split (#687) is rejected in favor of core's single layer_norm, weight-tying is kept, and the v1 MoEHybrid overlap isn't on the OLMoDDP path. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Co-Authored-By: Tianhua Tao <taotianhua@outlook.com>
1 parent 34a8084 commit db85cda

1 file changed

Lines changed: 48 additions & 0 deletions

File tree

src/olmo_core/nn/transformer/model.py

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -261,6 +261,54 @@ def get_rope_buffers(
261261
rope_buffers[int(key)] = None
262262
return rope_buffers
263263

264+
def prepare_cp_sequence_inputs(
265+
self,
266+
input_ids: torch.Tensor,
267+
labels: Optional[torch.Tensor] = None,
268+
*,
269+
ignore_index: int = -100,
270+
) -> Tuple[torch.Tensor, Optional[torch.Tensor], int]:
271+
"""
272+
Shard sequence inputs for context parallelism before entering a pipeline schedule.
273+
274+
Pipeline-parallel stages exchange hidden activations with already-local CP sequence
275+
lengths. The model still needs the original full sequence length later so each stage can
276+
build and shard its own RoPE buffers consistently.
277+
278+
:returns: The CP-sharded ``input_ids``, the CP-sharded ``labels`` (or ``None``), and the
279+
original (pre-shard) sequence length.
280+
"""
281+
cp_load_balancer = self._cp_load_balancer
282+
if cp_load_balancer is None:
283+
raise OLMoConfigurationError(
284+
"prepare_cp_sequence_inputs() requires context parallelism to be applied first"
285+
)
286+
if input_ids.dim() < 2:
287+
raise ValueError("'input_ids' must have a batch and sequence dimension")
288+
if labels is not None and labels.shape[:2] != input_ids.shape[:2]:
289+
raise ValueError(
290+
f"'labels' shape {tuple(labels.shape)} does not match input batch/sequence "
291+
f"shape {tuple(input_ids.shape[:2])}"
292+
)
293+
294+
original_seq_len = input_ids.shape[1]
295+
inputs = [input_ids]
296+
seq_dims = [1]
297+
pad_values: List[Union[int, float]] = [0]
298+
299+
if labels is not None:
300+
inputs.append(labels)
301+
seq_dims.append(1)
302+
pad_values.append(ignore_index)
303+
304+
sharded_inputs = cp_load_balancer.batch_shard(
305+
inputs=inputs,
306+
seq_dims=seq_dims,
307+
pad_values=pad_values,
308+
)
309+
sharded_labels = sharded_inputs[1] if labels is not None else None
310+
return sharded_inputs[0], sharded_labels, original_seq_len
311+
264312
@torch.no_grad()
265313
def init_weights(
266314
self,

0 commit comments

Comments
 (0)