platforms/interface.py: the Platform base class

    @classmethod
    def pre_register_and_update(
        cls, parser: FlexibleArgumentParser | None = None
    ) -> None:
        """Do some pre-registration or update action for the current platform.

        This function is called before global VllmConfig is initialized or cli
        arguments are parsed. It's used for out-of-tree platforms to register or
        update the configuration.

        For example, the out-of-tree quantization config can be imported and
        registered here dynamically.
        """
        pass

    @classmethod
    def apply_config_platform_defaults(cls, vllm_config: "VllmConfig") -> None:
        """Apply the platform-specific default values to the config.

        This function is called during the initialization of global VllmConfig, after
        parsing cli arguments.
        It can modify the defaults of the config according to the platform. For example,
        it can enable custom_ops based on the enabled features.

        The config is passed by reference, so it can be modified in place.
        """
        pass

    @classmethod
    def check_and_update_config(cls, vllm_config: "VllmConfig") -> None:
        """Check and update the configuration for the current platform.

        It can raise an exception if the configuration is not compatible with
        the current platform, or it can update the configuration to make it
        compatible with the current platform.

        The config is passed by reference, so it can be modified in place.
        """
        pass

    @classmethod
    def update_block_size_for_backend(cls, vllm_config: "VllmConfig") -> None:
        """Ensure block_size is compatible with the attention backend.
        For hybrid models, also aligns block_size with mamba page sizes.
        """
        from vllm.config.cache import CacheConfig

        cache_config = vllm_config.cache_config
        model_config = vllm_config.model_config

        # model_config may be None during testing.
        if not model_config:
            return

        backend_classes = cls._find_non_ssm_backends(vllm_config)
        if not backend_classes:
            return

        # Phase 1: Pick a block size every attention backend supports (skip if
        # user set --block-size). Models can mix backends with disjoint
        # preferences, and a size from the first alone later fails
        # select_common_block_size().
        if not cache_config.user_specified_block_size:
            preferred = cls._preferred_block_size_for_backends(
                backend_classes, CacheConfig.DEFAULT_BLOCK_SIZE, vllm_config
            )
            if preferred != CacheConfig.DEFAULT_BLOCK_SIZE:
                logger.info(
                    "Setting kv cache block size to %d for %s backend(s).",
                    preferred,
                    "/".join(b.get_name() for b in backend_classes),
                )
            cache_config.block_size = preferred

        # Phase 2: Align block/mamba sizes for hybrid models
        # (may override user settings).
        pre_block_size = cache_config.block_size
        if model_config.is_hybrid:
            cls._align_hybrid_block_size(vllm_config, backend_classes[0])

        # Phase 3: Align block/page sizes when multiple KV dtypes share the
        # block pool (e.g. nvfp4 primary + unquantized skip layers).
        # May override the user's --block-size.
        if cache_config.kv_cache_dtype_skip_layers:
            cls._align_heterogeneous_kv_block_size(vllm_config, backend_classes[0])
        cls._check_aligned_block_size(vllm_config, backend_classes, pre_block_size)

        from vllm.config.vllm import set_current_vllm_config
        from vllm.v1.attention.backend import MultipleOf

        # Backends may read the current config to decide a preference.
        with set_current_vllm_config(vllm_config):
            if len(backend_classes) == 1:
                return backend_classes[0].get_preferred_block_size(default_block_size)
            if all(b.supports_block_size(default_block_size) for b in backend_classes):
                return default_block_size
            # A backend declaring no sizes accepts any, so it contributes 1.
            per_backend_sizes = [
                [
                    s.base if isinstance(s, MultipleOf) else s
                    for s in b.get_supported_kernel_block_sizes()
                ]
                or [1]
                for b in backend_classes
            ]
            candidates = sorted(
                {math.lcm(*sizes) for sizes in itertools.product(*per_backend_sizes)}
            )
            for candidate in candidates:
                if all(b.supports_block_size(candidate) for b in backend_classes):
                    return candidate
            raise ValueError(
                "The attention backends share no supported KV cache block size ("
                + "; ".join(
                    f"{b.get_name()}: {b.get_supported_kernel_block_sizes()}"
                    for b in backend_classes
                )
                + ")."
            )

    @classmethod
    def device_id_to_physical_device_id(cls, device_id: int):
        """Map a vLLM-local logical device ID to a physical device ID.

        The input is a logical local ID (e.g. a local rank), NOT a visible
        device ordinal; for the latter use
        visible_device_id_to_physical_device_id(). The two coincide only
        when no logical-to-physical mapping is in effect.
        """
        if _assigned_physical_gpu_ids is not None:
            if device_id >= len(_assigned_physical_gpu_ids):
                raise IndexError(
                    f"device_id {device_id} is out of range for "
                    f"assigned_physical_gpu_ids {_assigned_physical_gpu_ids} "
                    f"({len(_assigned_physical_gpu_ids)} devices assigned)"
                )
            return _assigned_physical_gpu_ids[device_id]
        # Treat empty device control env var as unset. This is a valid
        # configuration in Ray setups where the engine is launched in
        # a CPU-only placement group located on a GPU node.
        if (
            cls.device_control_env_var in os.environ
            and os.environ[cls.device_control_env_var] != ""
        ):
            device_ids = os.environ[cls.device_control_env_var].split(",")
            physical_device_id = device_ids[device_id]
            return cls.device_control_id_to_physical_device_id(physical_device_id)
        else:
            return device_id

    @classmethod
    def import_kernels(cls) -> None:
        """Import any platform-specific C kernels."""
        try:
            import vllm._C  # noqa: F401
        except ImportError as e:
            logger.warning_once("Failed to import from vllm._C: %s", repr(e))
        with contextlib.suppress(ImportError):
            import vllm._moe_C_stable_libtorch  # noqa: F401

    @classmethod
    def register_triton_kernel_overrides(cls) -> None:
        """Override core Triton kernels with platform implementations.

        Platforms whose Triton kernels need replacement implementations
        (e.g. CPU fallbacks when the Triton-CPU backend is unavailable)
        override this method and call
        :func:`vllm.triton_utils.dispatcher.register_kernels`
        with their overrides. It runs during model runner init, before any
        kernel launch. The default implementation does nothing.
        """
        return

        device_capability = cls.get_device_capability()
        if device_capability is None:
            return MLAPrefillBackendEnum.FLASH_ATTN.get_class()

        if device_capability.major == 10:  # Blackwell
            if mla_selector_config.mla_dimensions == MLADimensions(
                qk_nope_head_dim=192,
                qk_rope_head_dim=64,
                v_head_dim=256,
            ):
                priorities = [
                    MLAPrefillBackendEnum.TRTLLM_RAGGED,
                    MLAPrefillBackendEnum.FLASH_ATTN,
                    MLAPrefillBackendEnum.FLASHINFER,
                    MLAPrefillBackendEnum.TOKENSPEED_MLA,
                ]
            else:
                priorities = [
                    MLAPrefillBackendEnum.FLASH_ATTN,
                    MLAPrefillBackendEnum.TRTLLM_RAGGED,
                    MLAPrefillBackendEnum.FLASHINFER,
                    MLAPrefillBackendEnum.TOKENSPEED_MLA,
                ]
        else:  # Hopper (SM90) and older
            priorities = [MLAPrefillBackendEnum.FLASH_ATTN]

        all_invalid_reasons: dict[str, list[str]] = {}

        for backend_enum in priorities:
            try:
                backend_cls = backend_enum.get_class()
                invalid_reasons = backend_cls.validate_configuration(
                    device_capability, mla_selector_config
                )
            except ImportError:
                invalid_reasons = ["ImportError"]
            if not invalid_reasons:
                return backend_cls
            all_invalid_reasons[backend_enum.name] = invalid_reasons

    def __getattr__(self, key: str):
        # Pickle checks dunder methods like __getstate__. If we return None
        # for them, pickle treats it like a real value and tries to call it.
        if key.startswith("__") and key.endswith("__"):
            raise AttributeError(key)

        device = getattr(torch, self.device_type, None)
        if device is not None and hasattr(device, key):
            attr = getattr(device, key)
            # NOTE: `hasattr(device, key)=True` can only avoid AttributeError,
            # but the value of this attr could be `None`.
            if attr is not None:
                return attr

        logger.warning_once(
            "Current platform %s does not have '%s' attribute.",
            self.device_type,
            key,
        )
        return None

    def get_global_graph_pool(self) -> Any:
        """Return the global graph pool for this platform."""
        cls = self.__class__
        if cls._global_graph_pool is None:
            cls._global_graph_pool = self.graph_pool_handle()
        return cls._global_graph_pool