Tour: Start-up

    --additional-config '{
        "afd": {
            "role": "attention",
            "connector": "P2pNcclAFDConnector",
            "host": "127.0.0.1",
            "port": 6269,
            "num_attention_ranks": 2,
            "num_ffn_ranks": 2
        }
    }' \
    --max-num-seqs 64 \
    --max-num-batched-tokens 64 \
    --enable-dbo \
    --dbo-decode-token-threshold 2 \
    --dbo-prefill-token-threshold 12 \
    --max-cudagraph-capture-size 64 \
    --compilation-config '{
        "cudagraph_mode": "FULL_DECODE_ONLY", "cudagraph_capture_sizes":[64]
    }' \

    try:
        import afd_plugin.compat.patches.async_dp_engine  # noqa: F401
        import afd_plugin.compat.patches.async_dp_forward_context  # noqa: F401
        import afd_plugin.compat.patches.config_validation  # noqa: F401
        import afd_plugin.compat.patches.engine_core  # noqa: F401
    except Exception:
        _logger.debug(
            "AFD plugin: compatibility patches could not be applied",
            exc_info=True,
        )

    from afd_plugin.model_executor.routing_simulator import (
        register_afd_balanced_routing_strategy,
    )

    register_afd_balanced_routing_strategy()

    try:
        from afd_plugin.v1.worker.dbo import register_dbo_yield_custom_op

        register_dbo_yield_custom_op()

def _select_afd_worker_for_auto(vllm_config: VllmConfig) -> None:
    afd_config = parse_optional_afd_config(vllm_config)
    if afd_config is None:
        return

    from vllm.platforms import current_platform

    platform_worker_qualname = vllm_config.parallel_config.worker_cls
    if not isinstance(platform_worker_qualname, str):
        raise ValueError(
            "platform worker_cls must be a qualname string before AFD automatic "
            f"selection, got {type(platform_worker_qualname).__name__}",
        )
    vllm_config.parallel_config.worker_cls = afd_worker_qualname_for_platform_default(
        afd_config.role,
        platform_worker_qualname,
        is_cuda=current_platform.is_cuda(),
        device_type=current_platform.device_type,
    )

    # ### PATCH START: AFD FFN EngineCore daemon initialization
    # FFN ranks are connector daemons, so stop EngineCore initialization after
    # executor construction instead of setting up KV cache and scheduler state.
    if _is_afd_ffn_config(vllm_config):
        _initialize_ffn_engine_core(
            self,
            core_module,
            vllm_config,
            executor_class,
            log_stats,
            executor_fail_callback,
        )
        return
    # ### PATCH END: AFD FFN EngineCore daemon initialization

        native_runner_cls = gpu_model_runner_v2.GPUModelRunner
        gpu_model_runner_v2.GPUModelRunner = AFDAttentionModelRunnerV2
        try:
            super().init_device()
        finally:
            gpu_model_runner_v2.GPUModelRunner = native_runner_cls
        # ### PATCH END: select and directly construct the AFD runner.

    def load_model(
        self,
        load_dummy_weights: bool = False,
        *args,
        **kwargs,
    ) -> None:
        super().load_model(load_dummy_weights, *args, **kwargs)
        if not self.connector.is_initialized:
            self.connector.init_afd_connector()

        afd_pg = init_afd_process_group(
            backend="nccl",
            init_method=f"tcp://{self.afd_config.host}:{self.afd_config.port}",
            world_size=self.ffn_size + self.attn_size,
            rank=self.world_rank,
            group_name="afd",
            timeout=timedelta(seconds=self.afd_config.afd_process_group_timeout_s),
        )

    def initialize_from_config(self, kv_cache_config: KVCacheConfig) -> None:
        """Skip KV cache allocation and start the FFN connector loop."""

        self.cache_config.num_gpu_blocks = kv_cache_config.num_blocks
        self.model_runner.initialize_kv_cache(kv_cache_config)
        self.model_runner.initialize_afd_connector()
        self.start_ffn_server_loop()

def _run_ffn_busy_loop(self, core_module: Any) -> None:
    started = False
    try:
        self.model_executor.collective_rpc("start_ffn_server_loop")
        started = True
        # Emit readiness only after the collective start RPC succeeded; the
        # worker connector loop may not be ready to receive before this point.
        core_module.logger.info(
            "AFD FFN EngineCore started; workers run connector loop."
        )