Diff Coverage

Diff: origin/master...HEAD, staged and unstaged changes

Source File Diff Coverage (%) Missing Lines
hyper_parallel/compile/compiler.py 100%  
hyper_parallel/compile/pass_config.py 100%  
hyper_parallel/compile/passes/parallel/fsdp_pass.py 100%  
hyper_parallel/compile/tracer/graph_tracer.py 94.1% 664
hyper_parallel/compile/trainer.py 100%  
hyper_parallel/core/utils/communication.py 39.2% 182,191,197-199,390,396,408-414,417-419,421-425,482-483,501,506-509,512-513
hyper_parallel/distributed/compile.py 100%  
hyper_parallel/trainer/base.py 44.0% 536,750-752,759,764,769-770,773,777-779,782-783
hyper_parallel/trainer/config/trainer.py 100%  
hyper_parallel/trainer/runtime/metrics.py 6.7% 96,99,103,105,112,115-117,120-125
hyper_parallel/trainer/text_trainer.py 0.0% 68,283
hyper_parallel/compile/tracer/graph_tracer.py
660
661
662
663
664
665
            if num_loss_outputs
            else list(outputs[1:])
        )
    else:
        loss, grads, loss_dict = outputs, [], {}
    return loss, grads, loss_dict
hyper_parallel/core/utils/communication.py
178
179
180
181
182
183
184
185
186
    ``TracingContext.get()`` raises ``RuntimeError`` when no trace is active,
    so we catch it and return False.
    """
    if torch.compiler.is_compiling():
        return True
    try:
        return torch._guards.TracingContext.get() is not None  # pylint: disable=protected-access
    except Exception:  # pylint: disable=W0718
        return False
187
188
189
190
191
192
193
194
195


def _reduce_op_to_str(reduce_op: Any) -> str:
    """Map a ``dist.ReduceOp`` to the string name ``_functional_collectives`` expects."""
    op_str = {
        dist.ReduceOp.SUM: 'sum',
        dist.ReduceOp.PRODUCT: 'product',
        dist.ReduceOp.MAX: 'max',
        dist.ReduceOp.MIN: 'min',
193
194
195
196
197
198
199
200
201
202
203
        dist.ReduceOp.PRODUCT: 'product',
        dist.ReduceOp.MAX: 'max',
        dist.ReduceOp.MIN: 'min',
    }.get(reduce_op, 'sum')
    if hasattr(dist.ReduceOp, 'AVG') and reduce_op == dist.ReduceOp.AVG:
        op_str = 'avg'
    return op_str


class _TorchContiguousGrad(torch.autograd.Function):  # pylint: disable=abstract-method
    """Autograd identity that materializes gradients before upstream collectives."""
386
387
388
389
390
391
392
393
394
    if rank_list is not None:
        group_ranks = dist.get_process_group_ranks(group)
        if tuple(rank_list) != tuple(group_ranks):
            # Rank subset gathering (not common path) — same for both modes.
            output = [
                _TorchContiguousGrad.apply(tensor)
                for tensor in dist_func.all_gather(data, group=group)
            ]
            rank_to_idx = {int(rank): idx for idx, rank in enumerate(group_ranks)}
392
393
394
395
396
397
398
399
400
                for tensor in dist_func.all_gather(data, group=group)
            ]
            rank_to_idx = {int(rank): idx for idx, rank in enumerate(group_ranks)}
            output = [output[rank_to_idx[int(rank)]] for rank in rank_list]
            return torch.cat(output, dim=concat_dim)
    if not _is_tracing_or_compiling():
        # Eager: original implementation using torch.distributed.nn.functional.
        output = [
            _TorchContiguousGrad.apply(tensor)
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
    # Compile: use functional collective API for torch.compile compatibility.
    # all_gather_single only reliably supports gather_dim=0; for other dims we
    # gather on dim 0 then reshape to avoid _maybe_view_chunk_cat view() bugs
    # in Dynamo fake tensor tracing.
    group_size = dist.get_world_size(group)
    import torch.distributed._functional_collectives as fc  # pylint: disable=C0415
    gathered = fc.all_gather_single(data, 0, group)
    if hasattr(gathered, 'wait'):
        gathered = gathered.wait()
    if concat_dim == 0:
        return gathered
    # Reshape from [group_size * d0, d1, ...] to [d0 * group_size, d1, ...]
    # by moving the group_size factor from dim 0 to concat_dim.
    orig_shape = list(data.shape)
    gathered_shape = [orig_shape[0] * group_size] + orig_shape[1:]
    gathered = gathered.view(gathered_shape)
    # Move group_size from dim 0 to concat_dim via unflatten + movedim + flatten
    unflattened = gathered.unflatten(0, (group_size, orig_shape[0]))
    moved = torch.movedim(unflattened, 0, concat_dim)
    result_shape = list(orig_shape)
    result_shape[concat_dim] *= group_size
    return moved.reshape(result_shape).contiguous()


def differentiable_all_to_all_single(input_tensor: Tensor, input_splits: Sequence[int],
                                     output_splits: Sequence[int], group) -> Tensor:
478
479
480
481
482
483
484
485
486
487
    if not _is_tracing_or_compiling():
        return dist_func.all_reduce(data, op=reduce_op, group=group)
    # Compile: use functional collective API for torch.compile compatibility.
    # fc.all_reduce expects a string reduceOp and a ProcessGroup.
    import torch.distributed._functional_collectives as fc  # pylint: disable=C0415
    return fc.all_reduce(data, _reduce_op_to_str(reduce_op), group)


def differentiable_reduce_scatter(data: Tensor, dev_num: int, axis: int,
                                  op: Union[str, Any], group) -> Tensor:
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
            output_tensor, input_tuple, op=reduce_op, group=group
        )
        # 'avg' maps to SUM in _OP_MAP, so the division stays manual.
        if op == 'avg':
            output_tensor = output_tensor / dev_num
        return output_tensor

    # Compile: use functional collective API for torch.compile compatibility.
    # fc.reduce_scatter_single expects a string reduceOp, scatter_dim, and group.
    import torch.distributed._functional_collectives as fc  # pylint: disable=C0415
    result = fc.reduce_scatter_single(data, _reduce_op_to_str(reduce_op), axis, group)
    if hasattr(result, 'wait'):
        result = result.wait()
    # 'avg' maps to SUM in _OP_MAP, so the division stays manual.
    if op == 'avg':
        result = result / dev_num
    return result


def differentiable_variable_all_gather(
        input_tensor: Tensor, output_splits: Sequence[int], group) -> Tensor:
hyper_parallel/trainer/base.py
532
533
534
535
536
537
538
539
540
            self.model_fwd_context = partial(loss_parallel, mesh=tp_mesh)
        else:
            self.model_fwd_context = nullcontext()
        self.model_bwd_context = nullcontext()
        self.graph_compiler: Optional[GraphCompiler] = None

    def _build_graph_compiler(self) -> None:
        """Build the optional joint-graph compiler for this trainer runtime."""
        if not self.config.compile.selects_graph_compiler():
746
747
748
749
750
751
752
753
754
755
756
            if channel_loss_callback is not None:
                channel_loss_callback.strip_model_inputs(micro_batch)
            self.model_integration.record_batch(micro_batch, loss_inputs)

            if self.graph_compiler is not None:
                if self.config.training.empty_cache_before_backward:
                    logger.warning(
                        "training.empty_cache_before_backward is ignored in graph compiler mode "
                        "because forward and backward execute in one joint graph"
                    )
                # The traced ``loss_dict`` values flow back as extra graph
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
                    )
                # The traced ``loss_dict`` values flow back as extra graph
                # outputs (emitted after the gradients), so the logging
                # metrics keep per-key parity with the eager path.
                loss, loss_dict = self.graph_compiler.forward_backward(
                    model_inputs=micro_batch,
                    labels=labels,
                )
            else:
                model_fwd_context = (
                    self.model_fwd_context()
                    if callable(self.model_fwd_context)
                    else self.model_fwd_context
                )
                with model_fwd_context:
                    outputs: ModelOutput = self.model(**micro_batch, use_cache=False)

                # with use_parallel_state("base"):
                loss, loss_dict = self.postforward(outputs, labels)
                # The loss graph owns everything required for backward. Releasing
                # the model output here avoids retaining large vocabulary logits
                # until the whole backward pass finishes.
                del outputs
                if self.config.training.empty_cache_before_backward:
                    empty_cache()

                # Backward pass
                with self.model_bwd_context:
                    loss.backward()

            del micro_batch
            return loss, loss_dict
hyper_parallel/trainer/runtime/metrics.py
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
        loss_name = key.split("_loss", maxsplit=1)[0]  # foundation/image_decoder/**

        cur_token_len = current_token_counts[f"{loss_name}_tokens"]
        if sequence_parallel:
            if _is_tracing_or_compiling():
                # Graph tracing: ``.item()`` is data-dependent and unsupported
                # by the static-shape tracer; aggregate on tensors instead.
                cur_token_len = differentiable_all_reduce(
                    cur_token_len, "sum", sequence_parallel_group
                )
            else:
                cur_token_len = all_reduce(cur_token_len.item(), op="sum", group=sequence_parallel_group)

        if _is_tracing_or_compiling():
            # Graph-traced variant of the aggregation below: identical math
            # expressed with tensor collectives so the static-shape tracer can
            # capture it (no ``.item()``, no data-dependent zero-token guard).
            # ``differentiable_all_reduce`` dispatches to functional collectives
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
            # capture it (no ``.item()``, no data-dependent zero-token guard).
            # ``differentiable_all_reduce`` dispatches to functional collectives
            # under tracing; the token-count tensors are baked as trace-time
            # constants, matching the tracer's fixed-shape contract.
            step_len = differentiable_all_reduce(
                step_token_counts[f"{loss_name}_tokens"], "sum", dp_cp_group
            )
            local_weighted_loss = cur_loss * cur_token_len
            backward_loss = local_weighted_loss / step_len * device_mesh.dp_size * device_mesh.cp_size
            global_weighted_loss = differentiable_all_reduce(
                local_weighted_loss.detach(), "sum", dp_cp_group
            )
            global_mean = global_weighted_loss / step_len
            cur_loss = backward_loss + global_mean - backward_loss.detach()
            if sequence_parallel:
                cur_loss = cur_loss / sequence_parallel_size
            loss_dict[key] = cur_loss
            continue

        all_reduced_len = all_reduce(
            step_token_counts[f"{loss_name}_tokens"].item(),
            op="sum",
hyper_parallel/trainer/text_trainer.py
64
65
66
67
68
69
70
71
72
        self.base._build_optimizer()
        self.base.model_integration.attach_optimizer(self.base.optimizer)
        self.base._build_lr_scheduler()
        self.base._build_training_context()
        self.base._build_graph_compiler()
        self.base._init_callbacks()

    def _build_model_assets(self) -> None:
        """Build tokenizer-backed assets for text training."""
279
280
281
282
283
284
285
286
287
        # Checkpoint and logging callbacks observe the number of completed
        # optimizer updates.
        self.base.state.global_step += 1
        grad_norm_value = float(grad_norm)
        self.base._end_model_integration_step(  # pylint: disable=protected-access
            {"loss": total_loss, "grad_norm": grad_norm_value}
        )
        self.on_step_end(
            loss=total_loss,