Diff Coverage

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

Source File Diff Coverage (%) Missing Lines
hyper_parallel/__init__.py 100%  
hyper_parallel/core/distributed_checkpoint/__init__.py 100%  
hyper_parallel/core/distributed_checkpoint/state_dict.py 75.0% 44,65
hyper_parallel/core/fully_shard/hsdp_param.py 100%  
hyper_parallel/platform/mindspore/platform.py 66.7% 1696,1702
hyper_parallel/platform/platform.py 66.7% 1017,1031
hyper_parallel/platform/torch/fully_shard/state_dict_utils.py 8.8% 312,328-334,339-342,346-347,359-362,364-367,371,373-375,379,381-382,389-391,422-423,425-427,429-435,437-441,443-444,446,450,457-461,470-474,484-486,488-494,496,499-500,502-504,513-518,520-522,526-527,556,558-560,562-563,565-568,570-574,577-578,580,584,589-590,592,601-605,619-623,627,631-632,641-643,645-647,651,653-665,678,680-683,685-690,692-697,699,717,719-722,725,727-728,730-731,733,740,770-774,776,778-779,781-784,791-792,796-798,800,804-806,808-809,814,816,822,856,858-863,865-866,868-876,878-884,886-887,889-893,895-896,898-899,924,926-931,933-934,936-939,959-961,963,965-966,968-969,971-982,984,986-987,989-991,993,1009-1012,1019-1022,1028-1029,1040-1043,1045-1049,1054-1057,1071-1073,1075-1079,1081,1083-1086,1101-1104,1106-1109,1111,1113,1115-1117,1131-1140,1145,1149,1175-1176,1178,1180-1181,1186-1190,1192-1193,1196,1201,1226,1228-1230,1232-1235,1241-1245,1247,1257-1261,1267,1284-1285,1287-1292,1296-1297,1299-1303,1307-1308,1310,1320-1321,1323-1328,1330,1333-1334,1336-1337,1339-1340,1342-1343,1348,1371-1372,1374-1375,1377,1410-1412,1414-1418,1420-1422,1424-1425,1429,1436-1439,1447-1450,1458-1466,1477,1481-1485,1487-1489,1491-1492,1494-1496,1498-1500,1502,1504,1512,1516-1524,1526-1527,1529-1530,1552-1554,1556-1557,1559-1560,1563,1567-1569,1572-1579,1581-1585,1587,1605,1609-1610,1612-1617,1619-1621,1623-1626,1628-1629,1631,1635,1637-1640,1642,1667,1669-1673,1675-1676,1678-1679,1683,1692-1693,1695-1699,1701-1705,1707,1712
hyper_parallel/platform/torch/platform.py 50.0% 1039,1042,1047,1050
hyper_parallel/core/distributed_checkpoint/state_dict.py
40
41
42
43
44
45
46
47
48

    Returns:
        dict: Optimizer state dict with FQN-based keys.
    """
    return platform.get_optim_state_dict(model, optimizer, options=options)


def set_optim_state_dict(
    model: Any,
61
62
63
64
65
        optim_state_dict: The optimizer state dict to load.
        options: Optional configuration (full_state_dict, cpu_offload,
            strict, broadcast_from_rank0, etc.).
    """
    platform.set_optim_state_dict(model, optimizer, optim_state_dict, options=options)
hyper_parallel/platform/mindspore/platform.py
1692
1693
1694
1695
1696
1697
1698
1699
1700
        )

    @staticmethod
    def get_optim_state_dict(model, optimizer, *, options=None):
        raise NotImplementedError(
            "issue240 optimizer state_dict adapter is currently supported only on Torch backend"
        )

    @staticmethod
1698
1699
1700
1701
1702
1703
1704
1705
1706
        )

    @staticmethod
    def set_optim_state_dict(model, optimizer, optim_state_dict, *, options=None):
        raise NotImplementedError(
            "issue240 optimizer state_dict adapter is currently supported only on Torch backend"
        )

    @staticmethod
hyper_parallel/platform/platform.py
1013
1014
1015
1016
1017
1018
1019
1020
1021

        Returns:
            dict: The optimizer state dictionary with FQN-based keys.
        """
        raise NotImplementedError(
            "Platform subclasses must implement get_optim_state_dict"
        )

    @staticmethod
1027
1028
1029
1030
1031
1032
1033
1034
1035
            optimizer: The optimizer instance.
            optim_state_dict: The optimizer state dict to load.
            options: Optional configuration for state dict loading.
        """
        raise NotImplementedError(
            "Platform subclasses must implement set_optim_state_dict"
        )

    @staticmethod
hyper_parallel/platform/torch/fully_shard/state_dict_utils.py
308
309
310
311
312
313
314
315
316
    """Raised when the requested configuration is not supported."""


def _is_dtensor_param(param: Any) -> bool:
    return isinstance(param, DTensor)


def _is_replicate_group_root(
    mesh: DeviceMesh,
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
    its global rank is not 0 (e.g. PP stage-1 ranks in a 3-D mesh).

    If there are no Replicate dimensions, falls back to ``dist.get_rank() == 0``.
    """
    coord = mesh.get_coordinate()
    if coord is None:
        return False
    replicate_dims = [i for i, p in enumerate(placements) if isinstance(p, Replicate)]
    if not replicate_dims:
        return dist.get_rank() == 0
    return all(coord[d] == 0 for d in replicate_dims)


def _param_to_fqn(model: nn.Module) -> Dict[nn.Parameter, str]:
    """Build a mapping from parameter object to its fully-qualified name."""
    param_to_name: Dict[nn.Parameter, str] = {}
    for name, param in model.named_parameters():
        if param in param_to_name:
            raise ValueError(
                f"Parameter {name} shares the same object as "
                f"{param_to_name[param]}. Duplicate parameter objects are not supported."
            )
        param_to_name[param] = name
    return param_to_name


def _build_id_to_fqn(
    optimizer: torch.optim.Optimizer,
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386

    Returns:
        Tuple of (saved_id_to_fqn, fqn_to_saved_id).
    """
    param_to_name = _param_to_fqn(model)
    raw_sd = optimizer.state_dict()
    saved_id_to_fqn: Dict[int, str] = {}
    saved_id_to_param: Dict[int, nn.Parameter] = {}

    for runtime_group, saved_group in zip(optimizer.param_groups, raw_sd["param_groups"]):
        for parameter, saved_id in zip(runtime_group["params"], saved_group["params"]):
            if saved_id in saved_id_to_param and saved_id_to_param[saved_id] is not parameter:
                raise ValueError(
                    f"saved_id {saved_id} maps to different parameters; "
                    f"this is not supported."
                )
            saved_id_to_param[saved_id] = parameter

    for saved_id, param in saved_id_to_param.items():
        if param not in param_to_name:
            raise ValueError(
                f"Parameter with saved_id={saved_id} not found in "
                f"model.named_parameters(). Ensure the model matches the optimizer."
            )
        saved_id_to_fqn[saved_id] = param_to_name[param]

    fqn_to_saved_id = {v: k for k, v in saved_id_to_fqn.items()}
    return saved_id_to_fqn, fqn_to_saved_id


def _get_param_dtensor_info(
    param: nn.Parameter,
385
386
387
388
389
390
391
392
393
394
395
def _get_param_dtensor_info(
    param: nn.Parameter,
) -> Optional[Tuple[DeviceMesh, Sequence]]:
    """Return (device_mesh, placements) if param is DTensor, else None."""
    if _is_dtensor_param(param):
        return param.device_mesh, param.placements
    return None


def _convert_state_tensor(
    tensor: torch.Tensor,
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454

    Returns:
        Converted tensor, or None if this rank should not keep the result.
    """
    dtensor_info = _get_param_dtensor_info(param)
    is_dtensor_value = isinstance(tensor, DTensor)

    if full_state_dict and dtensor_info is not None:
        if is_dtensor_value:
            full_t = tensor.full_tensor()
        else:
            mesh, placements = dtensor_info
            full_t = DTensor.from_local(tensor, mesh, placements).full_tensor()
        if cpu_offload:
            if not is_rank0:
                return None
            return full_t.cpu()
        return full_t

    if is_dtensor_value:
        local_t = tensor.to_local()
        if cpu_offload:
            return local_t.cpu()
        return local_t

    if cpu_offload:
        return tensor.cpu()

    return tensor


def _is_scalar_state(key: str) -> bool:
    return key in _SCALAR_STATE_KEYS


def _convert_state_scalar(
    tensor: torch.Tensor,
453
454
455
456
457
458
459
460
461
462
463
464
465
def _convert_state_scalar(
    tensor: torch.Tensor,
    cpu_offload: bool,
) -> torch.Tensor:
    if isinstance(tensor, DTensor):
        tensor = tensor.to_local()
    if cpu_offload:
        return tensor.cpu()
    return tensor


def _determine_is_root(
    full_state_dict: bool,
466
467
468
469
470
471
472
473
474
475
476
477
478
    cpu_offload: bool,
    dtensor_info: Optional[Tuple[Any, Any]],
) -> bool:
    """Determine if current rank is the replicate-group root."""
    if full_state_dict and cpu_offload and dtensor_info is not None:
        return _is_replicate_group_root(dtensor_info[0], dtensor_info[1])
    if full_state_dict and cpu_offload:
        return (not dist.is_initialized()) or (dist.get_rank() == 0)
    return True


def _convert_state_entries(
    state: Dict[str, Any],
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
    full_state_dict: bool,
    cpu_offload: bool,
) -> Dict[str, Any]:
    """Convert optimizer state entries for a single parameter."""
    dtensor_info = _get_param_dtensor_info(param)
    is_root = _determine_is_root(full_state_dict, cpu_offload, dtensor_info)
    skip_non_root = full_state_dict and cpu_offload and not is_root

    converted: Dict[str, Any] = {}
    for key, value in state.items():
        if isinstance(value, torch.Tensor):
            if _is_scalar_state(key):
                result = _convert_state_scalar(value, cpu_offload)
                if not skip_non_root:
                    converted[key] = result
            else:
                result = _convert_state_tensor(
                    value, param, full_state_dict, cpu_offload, is_root,
                )
                if result is not None:
                    converted[key] = result
        else:
            if not skip_non_root:
                converted[key] = value
    return converted


def _build_result_param_groups(
    optimizer: torch.optim.Optimizer,
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
    raw_sd: Dict[str, Any],
    saved_id_to_fqn: Dict[int, str],
) -> List[Dict[str, Any]]:
    """Build result param_groups with FQN keys from saved id-based groups."""
    result_param_groups: List[Dict[str, Any]] = []
    for saved_group in raw_sd["param_groups"]:
        pg: Dict[str, Any] = {}
        for k, v in saved_group.items():
            if k == "params":
                pg["params"] = [saved_id_to_fqn[sid] for sid in v]
            else:
                pg[k] = v
        result_param_groups.append(pg)
    return result_param_groups


def _check_chained_optimizer(optimizer: Any) -> None:
    if type(optimizer).__name__ == "ChainedOptimizer":
        raise ValueError(
            "ChainedOptimizer is not supported by get_optim_state_dict / "
            "set_optim_state_dict. Use ChainedOptimizer.state_dict() and "
            "ChainedOptimizer.load_state_dict() instead, which handle "
            "multi-optimizer merging internally."
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596

    Returns:
        Optimizer state dict with FQN-based keys.
    """
    _check_chained_optimizer(optimizer)

    full_state_dict = getattr(options, "full_state_dict", False)
    cpu_offload = getattr(options, "cpu_offload", False)
    flatten = getattr(options, "flatten_optimizer_state_dict", False)

    raw_sd = optimizer.state_dict()
    saved_id_to_fqn, _ = _build_id_to_fqn(optimizer, model)

    param_by_id: Dict[int, nn.Parameter] = {}
    for runtime_group, saved_group in zip(optimizer.param_groups, raw_sd["param_groups"]):
        for parameter, saved_id in zip(runtime_group["params"], saved_group["params"]):
            param_by_id[saved_id] = parameter

    result_state: Dict[str, Dict[str, Any]] = {}
    for saved_id, state in raw_sd["state"].items():
        fqn = saved_id_to_fqn[saved_id]
        param = param_by_id[saved_id]
        converted = _convert_state_entries(
            state, param, full_state_dict, cpu_offload,
        )
        if converted:
            result_state[fqn] = converted

    result_param_groups = _build_result_param_groups(
        optimizer, raw_sd, saved_id_to_fqn,
    )

    result: Dict[str, Any] = {
        "state": result_state,
        "param_groups": result_param_groups,
    }

    if flatten:
        result = _flatten_optim_state_dict(result)

    return result


def _check_strict_fqns(
    source_fqns: set,
597
598
599
600
601
602
603
604
605
606
607
608
    target_fqns: set,
    strict: bool,
) -> None:
    """Validate FQN compatibility under strict mode."""
    if not strict:
        return
    extra_fqns = source_fqns - target_fqns
    if extra_fqns:
        raise ValueError(
            f"strict=True but checkpoint contains FQNs not in target "
            f"optimizer: {sorted(extra_fqns)}"
        )
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
    cpu_offload: bool,
    stores_dtensor: bool,
) -> Dict[str, Any]:
    """Convert and load state values for a single FQN."""
    converted: Dict[str, Any] = {}
    for key, value in source_state.items():
        if isinstance(value, torch.Tensor):
            if _is_scalar_state(key):
                converted[key] = _convert_input_scalar_to_target(
                    value, param, cpu_offload, stores_dtensor,
                )
            else:
                converted[key] = _convert_input_tensor_to_target(
                    value, param, full_state_dict, cpu_offload, stores_dtensor,
                )
        else:
            converted[key] = value
    return converted


def _merge_param_groups(
    source_param_groups: List[Dict[str, Any]],
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
    target_raw_sd: Dict[str, Any],
    target_fqn_to_saved_id: Dict[str, int],
) -> None:
    """Merge source param_group fields into target param_groups in-place."""
    if not source_param_groups:
        return
    target_saved_id_to_fqn = {v: k for k, v in target_fqn_to_saved_id.items()}

    target_pg_by_fqns: Dict[frozenset, Dict[str, Any]] = {}
    for saved_group in target_raw_sd["param_groups"]:
        fqns_in_group = frozenset(
            target_saved_id_to_fqn.get(sid, "")
            for sid in saved_group.get("params", [])
        )
        target_pg_by_fqns[fqns_in_group] = saved_group

    for source_pg in source_param_groups:
        source_fqn_set = frozenset(source_pg.get("params", []))
        matched_target_pg = None
        for target_fqn_set, target_pg in target_pg_by_fqns.items():
            if source_fqn_set & target_fqn_set:
                matched_target_pg = target_pg
                break
        if matched_target_pg is None:
            continue
        for k, v in source_pg.items():
            if k == "params":
                continue
            matched_target_pg[k] = v


def _build_target_fqn_mappings(
    optimizer: torch.optim.Optimizer,
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703

    Returns:
        Tuple of (fqn_to_saved_id, dtensor_state_ids).
    """
    target_raw_sd = optimizer.state_dict()

    target_id_to_param: Dict[int, nn.Parameter] = {}
    for runtime_group, saved_group in zip(optimizer.param_groups, target_raw_sd["param_groups"]):
        for parameter, saved_id in zip(runtime_group["params"], saved_group["params"]):
            target_id_to_param[saved_id] = parameter

    fqn_to_saved_id: Dict[str, int] = {}
    for saved_id, param in target_id_to_param.items():
        for name, p in model.named_parameters():
            if p is param:
                fqn_to_saved_id[name] = saved_id
                break

    dtensor_state_ids: set = set()
    for saved_id, state in target_raw_sd["state"].items():
        for value in state.values():
            if isinstance(value, DTensor):
                dtensor_state_ids.add(saved_id)
                break

    return fqn_to_saved_id, dtensor_state_ids


def _load_fqn_states(
    optim_state_dict: Dict[str, Any],
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737

    Returns:
        Updated state dict with saved_id keys.
    """
    new_state: Dict[int, Dict[str, Any]] = dict(target_raw_sd["state"])

    for fqn, source_state in optim_state_dict.get("state", {}).items():
        if fqn not in target_fqn_to_saved_id:
            if strict:
                raise ValueError(
                    f"strict=True but FQN '{fqn}' not found in target optimizer."
                )
            continue

        target_saved_id = target_fqn_to_saved_id[fqn]
        param = param_by_fqn[fqn]

        if target_saved_id not in new_state:
            new_state[target_saved_id] = {}

        new_state[target_saved_id].update(
            _load_state_values(
                source_state, param, full_state_dict, cpu_offload,
                target_saved_id in dtensor_state_ids,
            )
736
737
738
739
740
741
742
743
744
                target_saved_id in dtensor_state_ids,
            )
        )

    return new_state


def set_optim_state_dict(
    model: nn.Module,
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
                broadcast from the replicate-group root to all ranks within
                the same replicate subgroup.  In PP+HSDP, each PP stage
                broadcasts independently within its own HSDP subgroup.
    """
    full_state_dict = getattr(options, "full_state_dict", False)
    cpu_offload = getattr(options, "cpu_offload", False)
    flatten = getattr(options, "flatten_optimizer_state_dict", False)
    strict = getattr(options, "strict", True)
    broadcast_from_rank0 = getattr(options, "broadcast_from_rank0", False)

    _check_chained_optimizer(optimizer)

    if flatten:
        optim_state_dict = _unflatten_optim_state_dict(optim_state_dict, model, strict=strict)

    if full_state_dict and cpu_offload and not broadcast_from_rank0:
        has_any_state = bool(optim_state_dict.get("state", {}))
        if not has_any_state:
            raise ValueError(
                "Received empty state dict with full_state_dict=True "
                "and cpu_offload=True but broadcast_from_rank0=False. "
                "Set broadcast_from_rank0=True to allow the replicate-group "
                "root to broadcast."
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
                "Set broadcast_from_rank0=True to allow the replicate-group "
                "root to broadcast."
            )

    if broadcast_from_rank0:
        optim_state_dict = _broadcast_state_from_rank0(
            optim_state_dict, model, full_state_dict, cpu_offload,
        )

    param_by_fqn: Dict[str, nn.Parameter] = {}
    for name, param in model.named_parameters():
        param_by_fqn[name] = param

    target_fqn_to_saved_id, dtensor_state_ids = _build_target_fqn_mappings(
        optimizer, model, param_by_fqn,
    )

    source_fqns = set(optim_state_dict.get("state", {}).keys())
    target_fqns = set(target_fqn_to_saved_id.keys())
    _check_strict_fqns(source_fqns, target_fqns, strict)

    target_raw_sd = optimizer.state_dict()
    new_state = _load_fqn_states(
        optim_state_dict, target_raw_sd, target_fqn_to_saved_id,
        param_by_fqn, dtensor_state_ids, full_state_dict, cpu_offload, strict,
    )

    target_raw_sd["state"] = new_state

    _merge_param_groups(
        optim_state_dict.get("param_groups", []),
        target_raw_sd,
        target_fqn_to_saved_id,
    )
818
819
820
821
822
823
824
825
826
        target_raw_sd,
        target_fqn_to_saved_id,
    )

    optimizer.load_state_dict(target_raw_sd)


def _convert_input_tensor_to_target(
    tensor: torch.Tensor,
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903

    Returns:
        Tensor or DTensor matching the target optimizer's expected format.
    """
    dtensor_info = _get_param_dtensor_info(param)

    if not dtensor_info:
        target_device = param.data.device
        result = tensor
        if cpu_offload and result.device != target_device:
            result = result.to(target_device)
        return result

    mesh, placements = dtensor_info
    target_device = param.to_local().device

    if isinstance(tensor, DTensor):
        if tensor.device_mesh == mesh and tensor.placements == placements:
            if stores_dtensor:
                return tensor
            return tensor.to_local()
        redistributed = tensor.redistribute(mesh, placements)
        if stores_dtensor:
            return redistributed
        return redistributed.to_local()

    if full_state_dict:
        if cpu_offload:
            tensor = tensor.to(target_device)
        dt = distribute_tensor(tensor, mesh, placements)
        if stores_dtensor:
            return dt
        return dt.to_local()

    if cpu_offload and tensor.device != target_device:
        tensor = tensor.to(target_device)

    if stores_dtensor:
        if tensor.shape == param.to_local().shape:
            return DTensor.from_local(tensor, mesh, placements)
        dt = distribute_tensor(tensor, mesh, placements)
        return dt

    if tensor.shape == param.to_local().shape:
        return tensor

    dt = distribute_tensor(tensor, mesh, placements)
    return dt.to_local()


def _convert_input_scalar_to_target(
    tensor: torch.Tensor,
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943

    Returns:
        Plain tensor or replicated DTensor matching the target's expected format.
    """
    dtensor_info = _get_param_dtensor_info(param)

    if not dtensor_info or not stores_dtensor:
        target_device = param.data.device if not dtensor_info else param.to_local().device
        result = tensor
        if cpu_offload and result.device != target_device:
            result = result.to(target_device)
        return result

    mesh, placements = dtensor_info
    target_device = param.to_local().device

    scalar_placements = [Replicate()] * mesh.ndim
    if cpu_offload and tensor.device != target_device:
        tensor = tensor.to(target_device)
    return DTensor.from_local(tensor, mesh, scalar_placements)


def _get_broadcast_groups(
    model: nn.Module,
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
            pg: ProcessGroup for broadcast within the group
            src_rank: global rank of the replicate-group root
            is_root: True if current rank is the root of this group
    """
    param_by_fqn: Dict[str, nn.Parameter] = {}
    for name, param in model.named_parameters():
        param_by_fqn[name] = param

    model_fqns = list(param_by_fqn.keys())

    if not model_fqns:
        return []

    first_param = param_by_fqn[model_fqns[0]]
    dtensor_info = _get_param_dtensor_info(first_param)

    if dtensor_info is not None:
        mesh, placements = dtensor_info
        is_root = _is_replicate_group_root(mesh, placements)
        replicate_dims = [i for i, p in enumerate(placements) if isinstance(p, Replicate)]
        if replicate_dims:
            pg = mesh.get_group(replicate_dims[0])
            coord = mesh.get_coordinate()
            if coord is not None:
                root_coord = list(coord)
                for d in replicate_dims:
                    root_coord[d] = 0
                src_rank = int(mesh.mesh[tuple(root_coord)])
            else:
                src_rank = 0
        else:
            pg = dist.group.WORLD
            src_rank = 0
    else:
        is_root = dist.get_rank() == 0
        pg = dist.group.WORLD
        src_rank = 0

    return [{
        "fqns": model_fqns,
        "pg": pg,
        "src_rank": src_rank,
        "is_root": is_root,
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
    pg: dist.ProcessGroup,
    src_rank: int,
) -> List[str]:
    """Broadcast FQN list from root to all ranks in the replicate group."""
    fqn_list = fqns if is_root else []
    obj = [fqn_list]
    dist.broadcast_object_list(obj, src=src_rank, group=pg)
    return obj[0]


def _build_state_schema(
    state: Dict[str, Any],
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
def _build_state_schema(
    state: Dict[str, Any],
) -> Dict[str, Any]:
    """Build schema dict describing the types and shapes of state entries."""
    schema: Dict[str, Any] = {}
    for key, value in state.items():
        if isinstance(value, torch.Tensor):
            schema[key] = {
                "shape": tuple(value.shape),
                "dtype": str(value.dtype),
                "is_scalar": _is_scalar_state(key),
            }
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
                "dtype": str(value.dtype),
                "is_scalar": _is_scalar_state(key),
            }
        else:
            schema[key] = {"type": type(value).__name__, "value": value}
    return schema


def _broadcast_schema_per_fqn(
    fqn_list: List[str],
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
    pg: dist.ProcessGroup,
    src_rank: int,
) -> Dict[str, Dict[str, Any]]:
    """Broadcast schema for each FQN from root to all ranks."""
    fqn_schema: Dict[str, Dict[str, Any]] = {}
    for fqn in fqn_list:
        if is_root and fqn in optim_state_dict.get("state", {}):
            schema = _build_state_schema(optim_state_dict["state"][fqn])
        else:
            schema = {}
        schema_list = [schema] if is_root else [None]
        dist.broadcast_object_list(schema_list, src=src_rank, group=pg)
        fqn_schema[fqn] = schema_list[0]
    return fqn_schema


def _get_local_device() -> torch.device:
    """Get the device for the current local rank."""
    local_rank = dist.get_rank() if dist.is_initialized() else 0
    if torch.npu.is_available():
        return torch.device(f"npu:{local_rank}")
    return torch.device(f"cuda:{local_rank}")


def _broadcast_scalar_entry(
    key: str,
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
    pg: dist.ProcessGroup,
    cpu_offload: bool,
) -> torch.Tensor:
    """Broadcast a scalar state entry and return the result tensor."""
    dtype_str = info.get("dtype", "torch.float32")
    dtype = _resolve_dtype(dtype_str)
    device = _get_local_device()

    if is_root and fqn in optim_state_dict.get("state", {}):
        scalar_val = optim_state_dict["state"][fqn][key]
        t = scalar_val.clone() if isinstance(scalar_val, torch.Tensor) else torch.tensor(scalar_val)
        if t.dim() == 0:
            t = t.reshape(1)
    else:
        t = torch.zeros(1, dtype=dtype)

    if t.device.type == "cpu":
        t = t.to(device)
    dist.broadcast(t, src=src_rank, group=pg)
    return t.reshape(()).cpu() if cpu_offload else t.reshape(())


def _broadcast_tensor_entry(
    key: str,
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
    full_state_dict: bool,
    cpu_offload: bool,
) -> torch.Tensor:
    """Broadcast a tensor state entry and return the result tensor."""
    shape = info.get("shape", ())
    dtype_str = info.get("dtype", "torch.float32")
    dtype = _resolve_dtype(dtype_str)
    device = _get_local_device()

    if is_root and fqn in optim_state_dict.get("state", {}):
        src_tensor = optim_state_dict["state"][fqn][key]
        if src_tensor.device.type == "cpu":
            src_tensor = src_tensor.to(device)
    else:
        src_tensor = torch.zeros(shape, dtype=dtype, device=device)

    dist.broadcast(src_tensor, src=src_rank, group=pg)

    if full_state_dict and cpu_offload:
        return src_tensor.cpu() if is_root else src_tensor
    return src_tensor.cpu() if cpu_offload else src_tensor


def _broadcast_tensor_data_per_fqn(
    fqn_list: List[str],
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
    full_state_dict: bool,
    cpu_offload: bool,
) -> Dict[str, Dict[str, Any]]:
    """Broadcast tensor data for each FQN from root to all ranks."""
    result_state: Dict[str, Dict[str, Any]] = {}
    for fqn in fqn_list:
        schema = fqn_schema[fqn]
        if not schema:
            continue
        result_state[fqn] = {}
        for key, info in schema.items():
            is_scalar = info.get("is_scalar", False)
            if is_scalar:
                result_state[fqn][key] = _broadcast_scalar_entry(
                    key, fqn, info, optim_state_dict,
                    is_root, src_rank, pg, cpu_offload,
                )
            else:
                result_state[fqn][key] = _broadcast_tensor_entry(
                    key, fqn, info, optim_state_dict,
                    is_root, src_rank, pg, full_state_dict, cpu_offload,
                )
    return result_state


def _broadcast_state_from_rank0(
    optim_state_dict: Dict[str, Any],
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
    subgroup), so different PP stages broadcast independently without
    cross-stage interference.  This avoids the problem where global
    rank 0's FQN list does not include other stages' parameters.
    """
    if not dist.is_initialized():
        return optim_state_dict

    groups = _get_broadcast_groups(model)

    if not groups:
        return {
            "state": {},
            "param_groups": optim_state_dict.get("param_groups", []),
        }

    group = groups[0]
    pg = group["pg"]
    src_rank = group["src_rank"]
    is_root = group["is_root"]
    fqns = group["fqns"]

    fqn_list = _broadcast_fqn_list(is_root, fqns, pg, src_rank)
    fqn_schema = _broadcast_schema_per_fqn(
        fqn_list, optim_state_dict, is_root, pg, src_rank,
    )
    result_state = _broadcast_tensor_data_per_fqn(
        fqn_list, fqn_schema, optim_state_dict,
        is_root, src_rank, pg, full_state_dict, cpu_offload,
    )

    return {
        "state": result_state,
        "param_groups": optim_state_dict.get("param_groups", []),
    }
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
    optimizer state dicts using ``no_dist=True`` (see P6 test pattern),
    rather than using a coordinated DCP save with flatten format across
    all ranks.
    """
    flat: Dict[str, Any] = {}

    for fqn, state in state_dict.get("state", {}).items():
        for key, value in state.items():
            flat[f"state.{fqn}.{key}"] = value

    for group_idx, group in enumerate(state_dict.get("param_groups", [])):
        params_list = group.get("params", [])
        if not params_list:
            raise UnsupportedConfigurationError(
                f"Cannot flatten param_groups[{group_idx}]: empty 'params' list. "
                f"Empty param groups cannot be represented in flatten format "
                f"because there is no FQN to prefix the group fields. "
                f"Use non-flatten format or provide a stable group_name."
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
                f"Empty param groups cannot be represented in flatten format "
                f"because there is no FQN to prefix the group fields. "
                f"Use non-flatten format or provide a stable group_name."
            )
        for fqn in params_list:
            for key, value in group.items():
                if key == "params":
                    continue
                flat[f"param_group.{fqn}.{key}"] = value

    return flat


def _check_inconsistent_pg_fields(
    existing: Dict[str, Any],
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
    fqn: str,
    strict: bool,
) -> None:
    """Check and raise/warn on inconsistent param_group fields."""
    common_keys = set(existing.keys()) & set(incoming.keys())
    for k in common_keys:
        if existing[k] != incoming[k]:
            if strict:
                raise ValueError(
                    f"strict=True but param_group field '{k}' is "
                    f"inconsistent within the same group: "
                    f"existing={existing[k]!r}, "
                    f"incoming(from {fqn})={incoming[k]!r}"
1263
1264
1265
1266
1267
1268
1269
1270
1271
                    f"inconsistent within the same group: "
                    f"existing={existing[k]!r}, "
                    f"incoming(from {fqn})={incoming[k]!r}"
                )
            logger.warning(
                "param_group field '%s' is inconsistent within the "
                "same group: existing=%r, incoming(from %s)=%r. "
                "Keeping existing value (strict=False).",
                k, existing[k], fqn, incoming[k],
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314

    Returns:
        Tuple of (state, param_group_fields).
    """
    state: Dict[str, Dict[str, Any]] = {}
    param_group_fields: Dict[str, Dict[str, Any]] = {}

    for key, value in flat_dict.items():
        if key.startswith("state."):
            remainder = key[len("state."):]
            fqn = _match_fqn_from_remainder(remainder, known_fqns_sorted)
            if fqn is None:
                raise ValueError(
                    f"Cannot match FQN from flat key '{key}'. "
                    f"Known FQNs: {known_fqns_sorted}"
                )
            state_key = remainder[len(fqn) + 1:]
            state.setdefault(fqn, {})[state_key] = value

        elif key.startswith("param_group."):
            remainder = key[len("param_group."):]
            fqn = _match_fqn_from_remainder(remainder, known_fqns_sorted)
            if fqn is None:
                raise ValueError(
                    f"Cannot match FQN from flat key '{key}'. "
                    f"Known FQNs: {known_fqns_sorted}"
                )
            field_name = remainder[len(fqn) + 1:]
            param_group_fields.setdefault(fqn, {})[field_name] = value

    return state, param_group_fields


def _assemble_param_groups(
    known_fqns: List[str],
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
    param_group_fields: Dict[str, Dict[str, Any]],
    strict: bool,
) -> List[Dict[str, Any]]:
    """Assemble param_groups list from FQN-based state and fields."""
    param_groups: List[Dict[str, Any]] = []
    current_group: Dict[str, Any] = {"params": []}

    for fqn in known_fqns:
        if fqn in param_group_fields or fqn in state:
            fields = param_group_fields.get(fqn, {})
            if not current_group["params"]:
                current_group["params"].append(fqn)
                current_group.update(fields)
            else:
                existing_fields = {
                    k: v for k, v in current_group.items() if k != "params"
                }
                if existing_fields == fields or not fields:
                    current_group["params"].append(fqn)
                else:
                    _check_inconsistent_pg_fields(existing_fields, fields, fqn, strict)
                    current_group["params"].append(fqn)

    if current_group["params"]:
        param_groups.append(current_group)

    if not param_groups:
        raise UnsupportedConfigurationError(
            "Cannot unflatten: empty param_group in flatten format. "
            "Provide a stable group_name or use non-flatten format."
        )

    return param_groups


def _unflatten_optim_state_dict(
    flat_dict: Dict[str, Any],
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380

    Returns:
        The nested state dict with FQN keys.
    """
    known_fqns = [name for name, _ in model.named_parameters()]
    known_fqns_sorted = sorted(known_fqns, key=len, reverse=True)

    state, param_group_fields = _parse_unflatten_entries(flat_dict, known_fqns_sorted)
    param_groups = _assemble_param_groups(known_fqns, state, param_group_fields, strict)

    return {
        "state": state,
        "param_groups": param_groups,
    }
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432

    Returns:
        State dict template with empty tensors of correct shape/dtype.
    """
    full_state_dict = getattr(options, "full_state_dict", False)
    cpu_offload = getattr(options, "cpu_offload", False)
    flatten = getattr(options, "flatten_optimizer_state_dict", False)

    try:
        metadata = storage_reader.load_metadata()
    except FileNotFoundError:
        rank = dist.get_rank() if dist.is_initialized() else 0
        metadata = storage_reader.load_metadata(rank=rank)

    param_by_fqn: Dict[str, nn.Parameter] = {}
    for name, param in model.named_parameters():
        param_by_fqn[name] = param

    if flatten:
        return _build_flatten_template_from_metadata(
            metadata, param_by_fqn, full_state_dict, cpu_offload,
        )

    return _build_nested_template_from_metadata(
        metadata, optimizer, param_by_fqn, full_state_dict, cpu_offload,
    )

1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443


def _resolve_dtype(dtype_str: str) -> torch.dtype:
    """Resolve a dtype string like 'torch.float32' to torch.dtype."""
    try:
        return getattr(torch, dtype_str.replace("torch.", ""))
    except AttributeError:
        return torch.float32


def _match_fqn_from_remainder(
    remainder: str,
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
    remainder: str,
    known_fqns_sorted: List[str],
) -> Optional[str]:
    """Match a remainder string against known FQNs using longest-prefix match."""
    for fqn in known_fqns_sorted:
        if remainder == fqn or remainder.startswith(fqn + "."):
            return fqn
    return None


def _build_saved_id_to_fqn(
    optimizer: torch.optim.Optimizer,
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
    optimizer: torch.optim.Optimizer,
    param_by_fqn: Dict[str, nn.Parameter],
) -> Dict[int, str]:
    """Build mapping from optimizer saved IDs to FQNs."""
    raw_sd = optimizer.state_dict()
    saved_id_to_fqn: Dict[int, str] = {}
    for runtime_group, saved_group in zip(optimizer.param_groups, raw_sd["param_groups"]):
        for parameter, saved_id in zip(runtime_group["params"], saved_group["params"]):
            for name, p in param_by_fqn.items():
                if p is parameter:
                    saved_id_to_fqn[saved_id] = name
                    break
    return saved_id_to_fqn


def _parse_nested_state_entry(
    meta_key: str,
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
    full_state_dict: bool,
    cpu_offload: bool,
) -> Optional[Tuple[str, str, torch.Tensor]]:
    """Parse a state.* metadata entry and return (fqn, state_key, tensor) or None."""
    from hyper_parallel.core.distributed_checkpoint.metadata import (
        TensorStorageMetadata,
    )

    remainder = meta_key[len("state."):]
    fqns_sorted = sorted(param_by_fqn.keys(), key=len, reverse=True)
    matched_fqn = _match_fqn_from_remainder(remainder, fqns_sorted)
    if matched_fqn is None:
        return None

    state_key = remainder[len(matched_fqn) + 1:] if len(remainder) > len(matched_fqn) else None
    if state_key is None:
        return None

    if not isinstance(meta_val, TensorStorageMetadata):
        return None

    dtype = _resolve_dtype(meta_val.properties.dtype)
    global_shape = meta_val.size
    param = param_by_fqn.get(matched_fqn)

    if _is_scalar_state(state_key):
        device = torch.device("cpu")
        t = torch.zeros(global_shape, dtype=dtype, device=device)
    else:
        t = _create_empty_state_tensor(param, global_shape, dtype, full_state_dict, cpu_offload)

    return matched_fqn, state_key, t


def _parse_nested_param_group_entry(
    meta_key: str,
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
    meta_key: str,
    meta_val: Any,
) -> Optional[Tuple[int, str, torch.Tensor]]:
    """Parse a param_group.* metadata entry and return (group_idx, field_name, tensor) or None."""
    from hyper_parallel.core.distributed_checkpoint.metadata import (
        TensorStorageMetadata,
    )

    remainder = meta_key[len("param_group."):]
    parts = remainder.split(".", 1)
    if len(parts) < 2:
        return None
    try:
        group_idx = int(parts[0])
    except ValueError:
        return None
    field_name = parts[1]

    if not isinstance(meta_val, TensorStorageMetadata):
        return None

    dtype = _resolve_dtype(meta_val.properties.dtype)
    return group_idx, field_name, torch.zeros(meta_val.size, dtype=dtype)


def _build_nested_template_from_metadata(
    metadata: Any,
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
            "state": { "<FQN>": { "<state_key>": empty_tensor, ... }, ... },
            "param_groups": [ { "params": [...], "<field>": value, ... }, ... ]
        }
    """
    state: Dict[str, Dict[str, Any]] = {}
    param_groups_raw: Dict[int, Dict[str, Any]] = {}
    param_groups_fqns: Dict[int, List[str]] = {}

    saved_id_to_fqn = _build_saved_id_to_fqn(optimizer, param_by_fqn)
    raw_sd = optimizer.state_dict()

    for group_idx, saved_group in enumerate(raw_sd["param_groups"]):
        param_groups_raw[group_idx] = {
            k: v for k, v in saved_group.items() if k != "params"
        }
        param_groups_fqns[group_idx] = [
            saved_id_to_fqn[sid] for sid in saved_group["params"]
        ]

    for meta_key, meta_val in metadata.state_dict_metadata.items():
        if meta_key.startswith("state."):
            result = _parse_nested_state_entry(
                meta_key, meta_val, param_by_fqn, full_state_dict, cpu_offload,
            )
            if result is not None:
                fqn, state_key, t = result
                state.setdefault(fqn, {})[state_key] = t
        elif meta_key.startswith("param_group."):
            result = _parse_nested_param_group_entry(meta_key, meta_val)
            if result is not None:
                group_idx, field_name, t = result
                param_groups_raw.setdefault(group_idx, {})[field_name] = t

    result_param_groups: List[Dict[str, Any]] = []
    for group_idx in sorted(param_groups_fqns.keys()):
        pg: Dict[str, Any] = {"params": param_groups_fqns[group_idx]}
        pg.update(param_groups_raw.get(group_idx, {}))
        result_param_groups.append(pg)

    return {
        "state": state,
        "param_groups": result_param_groups,
    }
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
    Flatten checkpoint keys look like:
        state.<FQN>.<state_key>  -> TensorStorageMetadata
        param_group.<FQN>.<field> -> TensorStorageMetadata or BytesStorageMetadata
    """
    from hyper_parallel.core.distributed_checkpoint.metadata import (
        TensorStorageMetadata,
    )

    flat: Dict[str, Any] = {}
    fqns_sorted = sorted(param_by_fqn.keys(), key=len, reverse=True)

    for meta_key, meta_val in metadata.state_dict_metadata.items():
        if meta_key.startswith("state."):
            remainder = meta_key[len("state."):]
            matched_fqn = _match_fqn_from_remainder(remainder, fqns_sorted)
            if matched_fqn is None:
                continue

            state_key = remainder[len(matched_fqn) + 1:] if len(remainder) > len(matched_fqn) else None
            if state_key is None:
                continue

            if isinstance(meta_val, TensorStorageMetadata):
                dtype = _resolve_dtype(meta_val.properties.dtype)
                global_shape = meta_val.size
                param = param_by_fqn.get(matched_fqn)

                if _is_scalar_state(state_key):
                    t = torch.zeros(global_shape, dtype=dtype)
                else:
                    t = _create_empty_state_tensor(
                        param, global_shape, dtype,
                        full_state_dict, cpu_offload,
                    )
                flat[meta_key] = t

        elif meta_key.startswith("param_group."):
            if isinstance(meta_val, TensorStorageMetadata):
                dtype = _resolve_dtype(meta_val.properties.dtype)
                flat[meta_key] = torch.zeros(meta_val.size, dtype=dtype)

    return flat


def _create_empty_state_tensor(
    param: Optional[nn.Parameter],
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687

    Returns:
        Empty tensor with correct shape, dtype, and device.
    """
    dtensor_info = _get_param_dtensor_info(param) if param is not None else None

    if dtensor_info is not None:
        mesh, placements = dtensor_info
        if full_state_dict:
            shape = global_shape
            device = torch.device("cpu") if cpu_offload else param.to_local().device
        else:
            shape = param.to_local().shape
            device = torch.device("cpu") if cpu_offload else param.to_local().device
    else:
        shape = global_shape
        device = torch.device("cpu") if cpu_offload else (
            param.data.device if param is not None else torch.device("cpu")
        )

    return torch.zeros(shape, dtype=dtype, device=device)


def _infer_state_keys(optimizer: torch.optim.Optimizer) -> List[str]:
    """Infer expected state keys from optimizer defaults.
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712

    Supports Adam, AdamW, and SGD. Returns keys that would appear
    after at least one optimizer.step() call.
    """
    opt_cls = type(optimizer)
    opt_name = opt_cls.__name__.lower()

    if "adam" in opt_name:
        keys = ["step", "exp_avg", "exp_avg_sq"]
        if optimizer.defaults.get("amsgrad", False):
            keys.append("max_exp_avg_sq")
        return keys

    if "sgd" in opt_name:
        keys = []
        if optimizer.defaults.get("momentum", 0) != 0:
            keys.append("momentum_buffer")
        return keys

    logger.warning(
        "Cannot infer state keys for optimizer type '%s'. "
        "DCP load template may be incomplete.",
        opt_cls.__name__,
    )
    return []
hyper_parallel/platform/torch/platform.py
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054

    @staticmethod
    def get_optim_state_dict(model, optimizer, *, options=None):
        # pylint: disable=C0415
        from hyper_parallel.platform.torch.fully_shard.state_dict_utils import (
            get_optim_state_dict as _get_optim_state_dict,
        )
        return _get_optim_state_dict(model, optimizer, options=options)

    @staticmethod
    def set_optim_state_dict(model, optimizer, optim_state_dict, *, options=None):
        # pylint: disable=C0415
        from hyper_parallel.platform.torch.fully_shard.state_dict_utils import (
            set_optim_state_dict as _set_optim_state_dict,
        )
        _set_optim_state_dict(model, optimizer, optim_state_dict, options=options)

    @staticmethod
    def save_checkpoint(cell: Module, file_path: str, ckpt_format: str = "safetensors") -> None:
        if ckpt_format == "safetensors":