Diff Coverage

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

Source File Diff Coverage (%) Missing Lines
hyper_parallel/platform/torch/chunk_loss.py 84.3% 59-60,64-66,145,209,211,216,221,223
hyper_parallel/trainer/base.py 20.5% 181-184,188-193,198-200,227,1268,1282,1285-1287,1289-1291,1320-1324,1326-1331,1333-1334,1337,1339-1343,1348,1351-1356,1358,1361-1362,1364-1365,1372,1377-1382,1460,1482
hyper_parallel/trainer/config.py 100%  
hyper_parallel/platform/torch/chunk_loss.py
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
            direct_backward: bool,
        ) -> tuple:
            """Compute and retain the accumulated chunk gradients."""
            if torch.is_tensor(loss_scale):
                if loss_scale.numel() != 1:
                    raise ValueError(
                        "loss_scale must be a scalar, "
                        f"got shape {tuple(loss_scale.shape)}"
                    )
                if loss_scale.requires_grad:
                    raise ValueError("loss_scale must not require gradients")
                chunk_loss_scale = loss_scale.to(
                    device=hidden_states.device, dtype=torch.float32,
                )
            else:
                chunk_loss_scale = float(loss_scale)
141
142
143
144
145
146
147
148
149
            """Replay the private input and output-head gradient buffers."""
            del grad_loss_sum
            grad_hidden, grad_weight = ctx.saved_tensors
            if grad_scaled_loss is None:
                return None, None, None, None, None, None, None

            # The Trainer passes the complete scale into ``forward`` and calls
            # backward directly on ``scaled_loss_sum``. Returning the private
            # buffers for that unit upstream gradient avoids an NPU scalar-mul
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
    Raises:
        ValueError: If an input shape or option is invalid.
    """
    if chunk_size <= 0:
        raise ValueError(f"chunk_size must be positive, got {chunk_size}")
    if hidden_states.dim() != 3:
        raise ValueError(
            "hidden_states must have shape [batch, sequence, hidden], "
            f"got {tuple(hidden_states.shape)}"
        )
    if targets.dim() != 2 or hidden_states.shape[:2] != targets.shape:
        raise ValueError(
            "targets must match hidden_states batch/sequence dimensions, "
            f"got hidden={tuple(hidden_states.shape)} and targets={tuple(targets.shape)}"
        )
    if hidden_states.size(1) == 0:
        raise ValueError("hidden_states sequence dimension must be non-empty")
    if head_weight.dim() != 2 or head_weight.size(1) != hidden_states.size(2):
        raise ValueError(
            "head_weight must have shape [vocab, hidden] matching hidden_states, "
            f"got hidden={tuple(hidden_states.shape)} and weight={tuple(head_weight.shape)}"
        )
hyper_parallel/trainer/base.py
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
        return self._output_loss_implementation() == "chunk_loss"

    def _validate_output_loss_configuration(self) -> None:
        """Reject unsupported output-loss combinations before model build."""
        implementation = self._output_loss_implementation()
        valid_implementations = {"eager", "chunk_loss"}
        if implementation not in valid_implementations:
            raise ValueError(
                "train.output_loss.implementation must be one of "
                f"{sorted(valid_implementations)}, got {implementation!r}."
            )
        if implementation == "eager":
            return
        tp_size = self._parallel_dim_size("tp")
        pp_size = self._parallel_dim_size("pp")
        if tp_size != 1 or pp_size != 1:
            raise ValueError(
                "Chunk Loss version one requires "
                "train.accelerator.tp=1 and pp=1, "
                f"got tp={tp_size}, pp={pp_size}."
            )
        chunk_size = int(self.args.train.output_loss.chunk_size)
        if chunk_size <= 0:
            raise ValueError(
                "train.output_loss.chunk_size must be positive for "
                f"chunk_loss, got {chunk_size}."
            )
223
224
225
226
227
228
229
230
231

        self.parallel_dims = ParallelDims.from_config(
            self.args.train.accelerator, world_size=platform.get_world_size(),
        )
        self._validate_output_loss_configuration()
        logger.info_rank0("ParallelDims: %s", self.parallel_dims.summary())
        # Mixed precision lives in FSDP2's MixedPrecisionPolicy, so a
        # low-precision run needs a dp_shard axis (size-1 is enough) for the
        # FSDP wrap to exist — see ``build_mesh``'s force_dp_shard contract.
1264
1265
1266
1267
1268
1269
1270
1271
1272
        shifted_labels = micro_batch.pop("labels", None) if labels_are_shifted else None
        if labels_are_shifted and shifted_labels is None:
            raise ValueError("CP-shifted loss marker is set but labels are missing.")
        if labels_are_shifted and self._chunk_loss_enabled():
            micro_batch["loss_targets"] = shifted_labels
        return micro_batch, labels_are_shifted, shifted_labels

    def _compute_micro_loss(
        self,
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
        """Return mean loss and summed loss for one micro-batch."""
        if not labels_are_shifted:
            loss = outputs["loss"] if isinstance(outputs, dict) else outputs.loss
            if isinstance(outputs, dict) and outputs.get("loss_sum") is not None:
                return loss, outputs["loss_sum"].detach()
            return loss, loss.detach() * max(micro_batch_tokens, 1)

        if isinstance(outputs, dict) and outputs.get("connected_loss_sum") is not None:
            loss_sum = outputs["connected_loss_sum"]
            return loss_sum / max(micro_batch_tokens, 1), loss_sum

        if isinstance(outputs, dict) and outputs.get("loss_sum") is not None:
            loss_sum = outputs["loss_sum"]
            return loss_sum / max(micro_batch_tokens, 1), loss_sum

        logits = outputs["logits"] if isinstance(outputs, dict) else outputs.logits
        target_device = logits.device if hasattr(logits, "device") else self.device
        shifted_labels = shifted_labels.to(target_device, non_blocking=True)
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
        forward; otherwise backward would scale another full-size dWeight.
        """
        if not self._chunk_loss_enabled():
            return None
        if labels_are_shifted:
            if micro_batch.get("loss_targets") is None:
                return None
        elif micro_batch.get("labels") is None:
            return None

        def _model_loss_normalizer():
            num_items = micro_batch.get("num_items_in_batch")
            if num_items is not None:
                if torch.is_tensor(num_items):
                    return num_items.to(dtype=torch.float32).clamp_min(1)
                return float(max(num_items, 1))

            labels = micro_batch["labels"]
            targets = torch.nn.functional.pad(
                labels, (0, 1), value=-100,
            )[..., 1:]
            return targets.ne(-100).sum().to(dtype=torch.float32).clamp_min(1)

        agg = self.args.train.optimizer.loss_aggregation
        cp_size = self._cp_size()
        cp_rank_average = agg == "rank_average" and cp_size > 1
        if agg == "rank_average" and not cp_rank_average:
            normalizer = (
                float(max(micro_batch_tokens, 1))
                if labels_are_shifted
                else _model_loss_normalizer()
            )
            rank_scale = getattr(
                self.model, "hp_rank_average_loss_scale_size", 1,
            )
            loss_scale = 1.0 / normalizer
            if num_micro > 1:
                loss_scale = loss_scale / num_micro
            if rank_scale != 1:
                loss_scale = loss_scale / rank_scale
            return loss_scale

        loss_scale_size = getattr(
            self.model, "hp_token_loss_scale_size", self.parallel_dims.dp_size,
        )
        if labels_are_shifted:
            loss_scale = loss_scale_size / float(max(global_tokens, 1))
        else:
            normalizer = _model_loss_normalizer()
            loss_scale = (
                float(micro_batch_tokens)
                / float(max(global_tokens, 1))
                * loss_scale_size
                / normalizer
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
                * loss_scale_size
                / normalizer
            )

        tp_loss_scale_size = getattr(
            self.model,
            "hp_loss_tp_scale_size",
            max(1, self._parallel_dim_size("tp")),
        )
        if tp_loss_scale_size != 1:
            loss_scale = loss_scale / tp_loss_scale_size
        ep_loss_scale_size = getattr(self.model, "hp_loss_ep_scale_size", 1)
        if ep_loss_scale_size != 1:
            loss_scale = loss_scale / ep_loss_scale_size
        return loss_scale

    def _scale_loss_for_backward(
        self,
        loss,
1456
1457
1458
1459
1460
1461
1462
1463
1464
            global_tokens,
            num_micro,
        )
        if output_loss_scale is not None:
            micro_batch["output_loss_scale"] = output_loss_scale

        # Forward (with training context for activation offload)
        with self.model_fwd_context:
            outputs = self.model(**micro_batch, use_cache=False)
1478
1479
1480
1481
1482
1483
1484
1485
1486
            if isinstance(outputs, dict)
            else getattr(outputs, "backward_loss", None)
        )
        if backward_loss is not None:
            scaled_loss = backward_loss
        else:
            scaled_loss = self._scale_loss_for_backward(
                loss,
                loss_sum,