reference/code/Diffulex-mbd-lms/diffulex/config.py

import os

from dataclasses import dataclass, field
from transformers import AutoConfig
from diffulex.distributed.parallel_state import get_world_size
from diffulex.logger import get_logger

logger = get_logger(__name__)
SUPPORTED_PAGE_BLOCK_SIZES = (4, 8, 16, 32)
EDIT_SAMPLING_MODEL_NAMES = {
    "llada2",
    "llada2_moe",
    "llada2_mini",
    "llada2dot1_mini",
    "llada2_mini_dmax",
}
DMAX_MODEL_NAMES = EDIT_SAMPLING_MODEL_NAMES


@dataclass
class DecodingThresholds:
    add_block_threshold: float  # whether add a new block
    semi_complete_threshold: float  # whether unleash the decoding of the next block
    accept_threshold: float  # whether the token should be decoded (M2T)
    edit_threshold: float = 0.0  # whether a filled token should be edited (T2T)
    remask_threshold: float = 0.4  # whether a filled token should be remasked
    token_stability_threshold: float = 0.0  # whether decoded tokens are stable enough to add a new block


@dataclass
class Config:
    model: str
    lora_path: str = ""
    model_name: str = "dream"
    decoding_strategy: str = "d2f"  # "d2f", "multi_bd"

    # Sampling Harness
    hf_config: AutoConfig | None = None
    tokenizer_vocab_size: int | None = None
    eos: int = -1
    mask_token_id: int = 151666
    block_size: int = 32
    buffer_size: int = 4
    multi_block_prefix_full: bool = True
    token_merge_mode: str = "dmax_topk"  # "dmax_topk" or "iter_smooth_topk"
    token_merge_top_k: int = 1
    token_merge_renormalize: bool = True
    token_merge_weight: float = 1.0
    sampling_mode: str = "naive"  # "naive" or "edit"
    max_post_edit_steps: int = 16  # max refinement steps after all masks filled (JointThreshold)
    penalty_lambda: float = 0.0  # repetition penalty coefficient (JointThreshold)
    use_lora: bool = False
    pre_merge_lora: bool = False
    max_num_batched_tokens: int = 4096
    max_num_reqs: int = 128
    max_model_len: int = 2048
    gpu_memory_utilization: float = 0.9
    decoding_thresholds: DecodingThresholds | dict | None = None
    # TODO: Should be deprecated in the future
    add_block_threshold: float | None = None
    semi_complete_threshold: float | None = None
    accept_threshold: float | None = None
    remask_threshold: float | None = None
    token_stability_threshold: float | None = None
    # Truncation
    auto_max_nfe_warmup_steps: int = 8
    auto_max_nfe_tpf_floor: float = 1.0

    # Parallelism
    data_parallel_size: int = 1
    tensor_parallel_size: int = 2
    expert_parallel_size: int = 1
    master_addr: str = "localhost"
    master_port: int = 2333
    distributed_timeout_seconds: int = 600
    shm_name: str = "diffulex_shm"
    device_start: int = 0
    device_ids: list[int] = field(default_factory=lambda: [])

    # CUDA Graph
    enforce_eager: bool = False
    attn_impl: str = "triton"  # "triton" or "naive"
    enable_prefill_cudagraph: bool = True
    enable_full_static_runner: bool = True
    prefill_cudagraph_max_len: int = 0
    enable_torch_compile: bool = True
    enable_cudagraph_torch_compile: bool = False
    torch_compile_mode: str = "reduce-overhead"

    # MoE
    moe_dispatcher_backend: str = "standard"  # "standard", "naive", or "deepep"
    moe_gemm_impl: str = "triton"  # "triton", "vllm", or "naive"
    deepep_mode: str = "auto"  # "normal", "low_latency", or "auto"
    deepep_num_max_dispatch_tokens_per_rank: int = 256

    # KV Cache Page Table
    page_size: int = 32
    enable_prefix_caching: bool = True
    num_pages: int = -1
    k_cache_hdim_split_factor_x: int = 8
    kv_cache_layout: str = "unified"  # "unified" or "distinct"

    def _validate_sampling_mode(self) -> None:
        if self.sampling_mode == "edit" and self.model_name not in EDIT_SAMPLING_MODEL_NAMES:
            allowed = ", ".join(sorted(EDIT_SAMPLING_MODEL_NAMES))
            raise ValueError(
                f"sampling_mode='edit' is only supported for model_name in {{{allowed}}}, "
                f"got: {self.model_name}"
            )

        if self.model_name == "llada2dot1_mini" and self.sampling_mode != "edit":
            raise ValueError("model_name='llada2dot1_mini' requires sampling_mode='edit'.")

        if self.decoding_strategy == "dmax":
            allowed = ", ".join(sorted(DMAX_MODEL_NAMES))
            if self.model_name not in DMAX_MODEL_NAMES:
                raise ValueError(
                    f"decoding_strategy='dmax' is only supported for model_name in {{{allowed}}}, "
                    f"got: {self.model_name}"
                )
            if self.sampling_mode != "edit":
                raise ValueError("decoding_strategy='dmax' requires sampling_mode='edit'.")

    def __post_init__(self):
        if not os.path.isdir(self.model):
            raise ValueError(f"model must be an existing directory, got: {self.model}")

        if self.decoding_strategy == "d2f":
            if not self.multi_block_prefix_full:
                logger.warning("Forcing multi_block_prefix_full=True for decoding_strategy=d2f.")
            if self.enable_prefix_caching:
                logger.warning("Disabling prefix caching for decoding_strategy=d2f.")
            self.multi_block_prefix_full = True
            self.enable_prefix_caching = False
        elif self.decoding_strategy in ("multi_bd", "dmax"):
            if self.multi_block_prefix_full:
                logger.warning(f"Forcing multi_block_prefix_full=False for decoding_strategy={self.decoding_strategy}.")
            self.multi_block_prefix_full = False
            if self.enable_prefix_caching:
                logger.info(f"Enabling prefix caching for decoding_strategy={self.decoding_strategy}.")

        if self.page_size not in SUPPORTED_PAGE_BLOCK_SIZES:
            raise ValueError(
                f"page_size must be one of {SUPPORTED_PAGE_BLOCK_SIZES}, got: {self.page_size}"
            )

        if self.block_size not in SUPPORTED_PAGE_BLOCK_SIZES:
            raise ValueError(
                f"block_size must be one of {SUPPORTED_PAGE_BLOCK_SIZES}, got: {self.block_size}"
            )

        if self.block_size > self.page_size:
            raise ValueError(
                "block_size must be <= page_size, "
                f"got: block_size={self.block_size}, page_size={self.page_size}"
            )

        if not 1 <= self.tensor_parallel_size <= 8:
            raise ValueError(
                "tensor_parallel_size must be in [1, 8], "
                f"got: {self.tensor_parallel_size}"
            )

        if not 1 <= self.data_parallel_size <= 1024:
            raise ValueError(
                "data_parallel_size must be in [1, 1024], "
                f"got: {self.data_parallel_size}"
            )

        if not 1 <= self.expert_parallel_size <= 32768:
            raise ValueError(
                "expert_parallel_size must be in [1, 32768], "
                f"got: {self.expert_parallel_size}"
            )


        if self.token_merge_top_k <= 0:
            raise ValueError(f"token_merge_top_k must be positive, got: {self.token_merge_top_k}")
        
        if self.token_merge_mode not in ("dmax_topk", "iter_smooth_topk"):
            raise ValueError(
                "token_merge_mode must be one of {'dmax_topk', 'iter_smooth_topk'}, "
                f"got: {self.token_merge_mode}"
            )
            
        if self.token_merge_weight < 0:
            raise ValueError(f"token_merge_weight must be non-negative, got: {self.token_merge_weight}")
        
        if self.sampling_mode not in ("naive", "edit"):
            raise ValueError(
                "sampling_mode must be one of {'naive', 'edit'}, "
                f"got: {self.sampling_mode}"
            )
        self._validate_sampling_mode()
        if self.kv_cache_layout not in {"unified", "distinct"}:
            raise ValueError(
                "kv_cache_layout must be one of {'unified', 'distinct'}, "
                f"got: {self.kv_cache_layout}"
            )
        if self.attn_impl not in {"triton", "naive"}:
            raise ValueError(
                "attn_impl must be one of {'triton', 'naive'}, "
                f"got: {self.attn_impl}"
            )
        if self.moe_dispatcher_backend not in {"standard", "naive", "deepep"}:
            raise ValueError(
                "moe_dispatcher_backend must be one of {'standard', 'naive', 'deepep'}, "
                f"got: {self.moe_dispatcher_backend}"
            )
        if self.moe_gemm_impl not in {"triton", "vllm", "vllm_modular", "naive"}:
            raise ValueError(
                "moe_gemm_impl must be one of {'triton', 'vllm', 'vllm_modular', 'naive'}, "
                f"got: {self.moe_gemm_impl}"
            )
        if self.deepep_mode not in {"normal", "low_latency", "auto"}:
            raise ValueError(
                "deepep_mode must be one of {'normal', 'low_latency', 'auto'}, "
                f"got: {self.deepep_mode}"
            )
        if self.deepep_num_max_dispatch_tokens_per_rank <= 0:
            raise ValueError(
                "deepep_num_max_dispatch_tokens_per_rank must be positive, "
                f"got: {self.deepep_num_max_dispatch_tokens_per_rank}"
            )
        if self.expert_parallel_size != 1 or self.moe_dispatcher_backend != "standard":
            raise ValueError(
                "MoE A2A backends are currently unsupported. Require expert_parallel_size == 1 "
                "and moe_dispatcher_backend == 'standard', "
                f"got expert_parallel_size={self.expert_parallel_size}, "
                f"moe_dispatcher_backend={self.moe_dispatcher_backend}."
            )
            
        if not (isinstance(self.master_port, int) and 0 < self.master_port < 65536):
            raise ValueError(
                "master_port must be an int in (0, 65536), "
                f"got: {self.master_port}"
            )

        if not (
            isinstance(self.distributed_timeout_seconds, int)
            and self.distributed_timeout_seconds > 0
        ):
            raise ValueError(
                "distributed_timeout_seconds must be a positive int, "
                f"got: {self.distributed_timeout_seconds}"
            )
            
            
        if not (isinstance(self.device_start, int) and self.device_start >= 0):
            raise ValueError(
                "device_start must be a non-negative int, "
                f"got: {self.device_start}"
            )

        # LoRA validation
        if self.use_lora:
            if not self.lora_path:
                raise ValueError("lora_path must be provided when use_lora is True")

            if not os.path.exists(self.lora_path):
                logger.warning(f"LoRA path {self.lora_path} does not exist")

        self.hf_config = AutoConfig.from_pretrained(self.model, trust_remote_code=True)
        for name in (
            "attn_impl",
            "moe_dispatcher_backend",
            "moe_gemm_impl",
            "deepep_mode",
            "deepep_num_max_dispatch_tokens_per_rank",
            "expert_parallel_size",
            "tensor_parallel_size",
            "data_parallel_size",
            "mask_token_id",
        ):
            setattr(self.hf_config, name, getattr(self, name))
        cfg_max_model_len = (
            self.hf_config.max_position_embeddings
            if hasattr(self.hf_config, "max_position_embeddings")
            else self.hf_config.max_sequence_length
        )
        self.max_model_len = min(self.max_model_len, cfg_max_model_len)
        
        if self.max_num_batched_tokens < self.max_model_len:
            raise ValueError(
                "max_num_batched_tokens must be >= max_model_len after HF config clamp, "
                f"got max_num_batched_tokens={self.max_num_batched_tokens}, "
                f"max_model_len={self.max_model_len}"
            )
        if self.prefill_cudagraph_max_len < 0:
            raise ValueError(
                "prefill_cudagraph_max_len must be non-negative, "
                f"got: {self.prefill_cudagraph_max_len}"
            )
        if self.auto_max_nfe_warmup_steps <= 0:
            raise ValueError(
                "auto_max_nfe_warmup_steps must be positive, "
                f"got: {self.auto_max_nfe_warmup_steps}"
            )
        if self.auto_max_nfe_tpf_floor <= 0:
            raise ValueError(
                "auto_max_nfe_tpf_floor must be positive, "
                f"got: {self.auto_max_nfe_tpf_floor}"
            )

        if not self.device_ids:
            import torch
            # When CUDA_VISIBLE_DEVICES is set, PyTorch maps physical devices to logical device 0, 1, ...
            # So we should use logical device indices (0, 1, ...) instead of physical device IDs
            self.device_ids = list(range(torch.cuda.device_count()))
            logger.info(f"Using CUDA devices: {self.device_ids}")

        required_world_size = get_world_size(
            self.tensor_parallel_size,
            self.expert_parallel_size,
            self.data_parallel_size,
        )
        if len(self.device_ids) < required_world_size:
            raise ValueError(
                "Requested parallel world size exceeds available CUDA devices, "
                f"required={required_world_size}, device_ids={self.device_ids}."
            )

        # Build decoding_thresholds: dict or flat keys -> DecodingThresholds
        d = self.__dict__
        if isinstance(self.decoding_thresholds, dict):
            for key in (
                "add_block_threshold",
                "semi_complete_threshold",
                "accept_threshold",
                "edit_threshold",
                "remask_threshold",
                "token_stability_threshold",
            ):
                if d.get(key) is not None:
                    self.decoding_thresholds[key] = d[key]
            if "edit_threshold" not in self.decoding_thresholds:
                self.decoding_thresholds["edit_threshold"] = 0.0
            if "remask_threshold" not in self.decoding_thresholds:
                self.decoding_thresholds["remask_threshold"] = 0.4
            if "token_stability_threshold" not in self.decoding_thresholds:
                self.decoding_thresholds["token_stability_threshold"] = 0.0
            self.decoding_thresholds = DecodingThresholds(**self.decoding_thresholds)
        elif self.decoding_thresholds is None:
            add_block_threshold = d.get("add_block_threshold")
            semi_complete_threshold = d.get("semi_complete_threshold")
            accept_threshold = d.get("accept_threshold")
            edit_threshold = d.get("edit_threshold")
            remask_threshold = d.get("remask_threshold")
            token_stability_threshold = d.get("token_stability_threshold")
            self.decoding_thresholds = DecodingThresholds(
                add_block_threshold=0.1 if add_block_threshold is None else add_block_threshold,
                semi_complete_threshold=0.9 if semi_complete_threshold is None else semi_complete_threshold,
                accept_threshold=0.9 if accept_threshold is None else accept_threshold,
                edit_threshold=0.0 if edit_threshold is None else edit_threshold,
                remask_threshold=0.4 if remask_threshold is None else remask_threshold,
                token_stability_threshold=0.0
                if token_stability_threshold is None
                else token_stability_threshold,
            )

        if not 0.0 <= self.decoding_thresholds.accept_threshold <= 1.0:
            raise ValueError(
                "decoding_thresholds.accept_threshold must be in [0, 1], "
                f"got: {self.decoding_thresholds.accept_threshold}"
            )
        if not 0.0 <= self.decoding_thresholds.edit_threshold <= 1.0:
            raise ValueError(
                "decoding_thresholds.edit_threshold must be in [0, 1], "
                f"got: {self.decoding_thresholds.edit_threshold}"
            )
        if not 0.0 <= self.decoding_thresholds.remask_threshold <= 1.0:
            raise ValueError(
                "decoding_thresholds.remask_threshold must be in [0, 1], "
                f"got: {self.decoding_thresholds.remask_threshold}"
            )
        if not 0.0 <= self.decoding_thresholds.token_stability_threshold <= 1.0:
            raise ValueError(
                "decoding_thresholds.token_stability_threshold must be in [0, 1], "
                f"got: {self.decoding_thresholds.token_stability_threshold}"
            )
        if not isinstance(self.max_post_edit_steps, int) or self.max_post_edit_steps < 0:
            raise ValueError(
                "max_post_edit_steps must be a non-negative int, "
                f"got: {self.max_post_edit_steps}"
            )
        if not isinstance(self.penalty_lambda, (int, float)) or self.penalty_lambda < 0:
            raise ValueError(
                "penalty_lambda must be a non-negative float, "
                f"got: {self.penalty_lambda}"
            )
        self.accept_threshold = self.decoding_thresholds.accept_threshold
        self.remask_threshold = self.decoding_thresholds.remask_threshold

    @property
    def kv_cache_page_size(self) -> int:
        """Alias for page_size, used by engine/model_runner."""
        return self.page_size

    @property
    def add_new_block_threshold(self) -> float:
        return self.decoding_thresholds.add_block_threshold

    @property
    def complete_threshold(self) -> float:
        return self.decoding_thresholds.semi_complete_threshold