Diff Coverage

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

Source File Diff Coverage (%) Missing Lines
hyper_parallel/core/activation_memory/_backend.py 100%  
hyper_parallel/core/activation_memory/api.py 100%  
hyper_parallel/core/activation_memory/checkpoint.py 58.2% 471,475-476,478,493-495,502-503,508-511,513-515,518-519,527,543,551,558,561
hyper_parallel/core/activation_memory/compile_adapter.py 100%  
hyper_parallel/core/activation_memory/sac.py 50.0% 124
hyper_parallel/core/activation_memory/wrapper.py 100%  
hyper_parallel/distributed/activation_checkpoint.py 71.6% 315,335-342,539-540,692,698-700,702,787,791,906
hyper_parallel/distributed/attention_swap.py 100%  
hyper_parallel/core/activation_memory/checkpoint.py
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
        """Resolve contexts and capture RNG state before the forward pass runs."""
        device_type = _infer_device_type(*args)
        device_module = _get_device_module(device_type)
        forward_context, recompute_context = _resolve_contexts(context_fn)
        device_autocast_kwargs, cpu_autocast_kwargs = _get_autocast_kwargs(device_type)

        # Device RNG is captured only when the device is already initialized;
        # otherwise the device indexes its state lazily at first use.
        device_initialized = preserve_rng_state and getattr(device_module, "_initialized", False)
        device_ids, device_states = _get_device_states(device_type, *args) if device_initialized else ([], [])

        return cls(
            device_type=device_type,
            device_module=device_module,
            forward_context=forward_context,
            recompute_context=recompute_context,
489
490
491
492
493
494
495
496
497
498
        )

    def check_device_state_preserved(self, preserve_rng_state: bool) -> None:
        """Fail when the device was first initialized inside the checkpointed forward."""
        device_initialized = getattr(self.device_module, "_initialized", False)
        if preserve_rng_state and not self.device_was_initialized and device_initialized:
            raise RuntimeError(
                "The device state was initialized inside a Hyper checkpoint forward, so its initial RNG state "
                "could not be preserved. Initialize the device before entering checkpoint."
            )
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
            )

    def recompute_fn(self, function: Callable, *inputs: Any) -> None:
        """Restore execution state and rerun the checkpointed ``function``."""
        function_kwargs, *function_args = inputs
        with torch.random.fork_rng(
            devices=self.device_ids,
            enabled=self.cpu_state is not None,
            device_type=self.device_type,
        ):
            if self.cpu_state is not None:
                torch.set_rng_state(self.cpu_state)
                if self.device_was_initialized:
                    _set_device_states(self.device_type, self.device_ids, self.device_states)

            device_autocast_context = contextlib.nullcontext()
            if self.device_autocast_kwargs is not None:
                device_autocast_context = torch.amp.autocast(
                    device_type=self.device_type, **self.device_autocast_kwargs
                )
            with device_autocast_context, torch.amp.autocast("cpu", **self.cpu_autocast_kwargs), self.recompute_context:
                function(*function_args, **function_kwargs)


def _resolve_contexts(context_fn: Callable) -> Tuple[Any, Any]:
    """Validate and unpack the (forward, recompute) context tuple from ``context_fn``."""
523
524
525
526
527
528
529
530
531
    """Validate and unpack the (forward, recompute) context tuple from ``context_fn``."""
    contexts = context_fn()
    if not isinstance(contexts, tuple) or len(contexts) != 2:
        raise ValueError("context_fn must return a (forward_context, recompute_context) tuple.")
    return contexts[0], contexts[1]


def _checkpoint_without_reentrant_generator(
    function: Callable,
539
540
541
542
543
544
545
546
547
    """Set up eager checkpoint state around the caller's forward execution."""
    metadata_fn = _resolve_metadata_fn(determinism_check)
    state = _ExecutionState.capture(context_fn, preserve_rng_state, args)

    frame = _CheckpointFrame(partial(state.recompute_fn, function), early_stop, metadata_fn)
    dummy = torch.empty((0,), requires_grad=True)
    frame.input_saver = _NoopSaveInputs.apply(dummy, kwargs, *args)

    if frame.input_saver.grad_fn is None:
547
548
549
550
551
552
553
554
555
    if frame.input_saver.grad_fn is None:
        yield
        return

    if _RECOMPUTE_SESSION.get() is not None:
        raise CheckpointError("Nested checkpoint is not supported during scheduled recomputation.")

    collector = _RECOMPUTE_COLLECTOR.get()
    if collector is not None:
554
555
556
557
558
559
560
561
562
563
564
565
    collector = _RECOMPUTE_COLLECTOR.get()
    if collector is not None:
        collector.append(frame)
    try:
        with _create_checkpoint_hooks(frame), state.forward_context:
            yield
        frame.forward_completed = True
        state.check_device_state_preserved(preserve_rng_state)
    except BaseException:
        if collector is not None and frame in collector:
            collector.remove(frame)
        raise
hyper_parallel/core/activation_memory/sac.py
120
121
122
123
124
125
126
127
    return CheckpointPolicy.MUST_SAVE if b else CheckpointPolicy.PREFER_RECOMPUTE


def _make_swap_entry(x, *, has_alias, funcname, group_swap, cpu_pool):
    return _SwapCacheEntry(
        _maybe_detach(x, has_alias), funcname, group_swap=group_swap, cpu_pool=cpu_pool
    )

hyper_parallel/distributed/activation_checkpoint.py
311
312
313
314
315
316
317
318
319
    recomputation may need to unshard them inside that region. These allocation,
    copy and collective operators manage parameters rather than model
    activations, and therefore must not be matched against the forward replay.
    """
    ignore_sac_ops(_FSDP_SAC_IGNORED_OPS)


def compile_selective_checkpoint_policy(
    ctx: Any,
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346

    Returns:
        The checkpoint policy for ``func``.
    """
    del ctx, args, kwargs
    if func in _SELECTIVE_AC_FORCE_RECOMPUTE_OPS:
        return CheckpointPolicy.MUST_RECOMPUTE
    if func in _SELECTIVE_AC_MATMUL_OPS:
        return CheckpointPolicy.MUST_SAVE
    if func in _SELECTIVE_AC_MUST_SAVE_OPS:
        return CheckpointPolicy.MUST_SAVE
    return CheckpointPolicy.MUST_RECOMPUTE


def _make_selective_checkpoint_policy_fn() -> Callable:
    """Create an isolated eager selective activation checkpointing policy."""
535
536
537
538
539
540
541
542
543
544
    running fully eager and memory-unbounded.
    """
    if wrapped_count != 0 or not containers:
        return
    layer_count = sum(len(container.blocks) for container in containers)
    logger.warning(
        "%s activation checkpointing wrapped no module on %d layer(s) in %s; the "
        "model is running without activation checkpointing. Expected submodule "
        "names may not match this architecture, or the layers are already wrapped.",
        activation_checkpoint.capitalize(),
688
689
690
691
692
693
694
695
696
            sub_config is not config and not hasattr(sub_config, "use_cache")
        ):
            continue
        if getattr(sub_config, "use_cache", None) is not False:
            _try_disable_use_cache(sub_config)
    return False


def _try_disable_use_cache(sub_config: Any) -> None:
694
695
696
697
698
699
700
701
702
703
704
705
706


def _try_disable_use_cache(sub_config: Any) -> None:
    """Best-effort disable of ``use_cache`` on one config object."""
    try:
        sub_config.use_cache = False
    except Exception:  # pylint: disable=broad-exception-caught
        # Configuration objects may reject assignment with custom errors.
        pass


def _validate_activation_checkpoint_config(
    activation_checkpoint: Optional[str],
783
784
785
786
787
788
789
790
791
792
793
794
795
    Returns:
        The number of layers, or submodules in the KV-shared fallback, wrapped.
    """
    if has_kv_sharing:
        logger.warning(
            "Selective activation checkpointing is not supported for KV-shared models; "
            "falling back to submodule activation checkpointing."
        )
        return apply_submodule_checkpointing(
            ac_layers,
            has_kv_sharing,
            enable_compile=enable_compile,
            swap_inputs=swap_inputs,
902
903
904
905
906
907
908
909
910
    ac_layers = _flatten_layer_container_infos(containers)
    has_kv_sharing = _detect_kv_sharing_and_maybe_disable_cache(model)

    if hasattr(model, "gradient_checkpointing_disable"):
        model.gradient_checkpointing_disable()

    if activation_checkpoint == "selective":
        wrapped_count = _apply_selective_checkpointing(
            containers,