[feature]对接SGLang — coordinator、deploy-MindIE-Motor-AtomGit

Mexamples/deployer/deploy.py+12-16

	@@ -16,7 +16,8 @@ from lib.generator import k8s_utils
16	from lib.generator.k8s_utils import (	16	from lib.generator.k8s_utils import (
17	get_baseline_config_from_configmap, exec_all_kubectl_multi, exec_all_kubectl_singer,	17	get_baseline_config_from_configmap, exec_all_kubectl_multi, exec_all_kubectl_singer,
18	create_motor_config_configmap, init_service_domain_name, get_deploy_mode_from_config,	18	create_motor_config_configmap, init_service_domain_name, get_deploy_mode_from_config,
19	- update_kv_pool_enabled_flag, update_kv_conductor_enabled_flag, set_user_config_path	19	+ update_kv_pool_enabled_flag, update_kv_conductor_enabled_flag, update_engine_type_flag,
		20	+ set_user_config_path
20	)	21	)
21	from lib.generator.controller import generate_yaml_controller	22	from lib.generator.controller import generate_yaml_controller
22	from lib.generator.coordinator import generate_yaml_coordinator	23	from lib.generator.coordinator import generate_yaml_coordinator
	@@ -27,6 +28,7 @@ from lib.generator.single_container import generate_yaml_single_container
27	from lib.generator.infer_service import (	28	from lib.generator.infer_service import (
28	generate_yaml_infer_service_set, init_infer_service_domain_name, update_infer_service_replicas_only	29	generate_yaml_infer_service_set, init_infer_service_domain_name, update_infer_service_replicas_only
29	)	30	)
		31	+from lib.generator.mf_store import generate_yaml_mf_store
30	from lib.config_validator import (	32	from lib.config_validator import (
31	validate_deploy_mode_consistency, validate_deploy_mode_value,	33	validate_deploy_mode_consistency, validate_deploy_mode_value,
32	validate_only_instance_changed, resolve_config_paths	34	validate_only_instance_changed, resolve_config_paths
	@@ -76,13 +78,7 @@ def handle_update_instance_num(user_config):
76	if os.path.exists(infer_output):	78	if os.path.exists(infer_output):
77	update_infer_service_replicas_only(infer_output, deploy_config)	79	update_infer_service_replicas_only(infer_output, deploy_config)
78	else:	80	else:
79	- init_service_domain_name(	81	+ init_service_domain_name(paths, deploy_config)
80	- paths["controller_input_yaml"],
81	- paths["coordinator_input_yaml"],
82	- paths["kv_pool_input_yaml"],
83	- paths["kv_conductor_input_yaml"],
84	- deploy_config
85	- )
86	if not os.path.exists(infer_input):	82	if not os.path.exists(infer_input):
87	raise FileNotFoundError(f"InferServiceSet template yaml not found: {infer_input}.")	83	raise FileNotFoundError(f"InferServiceSet template yaml not found: {infer_input}.")
88	init_infer_service_domain_name(infer_input, deploy_config)	84	init_infer_service_domain_name(infer_input, deploy_config)
	@@ -96,10 +92,7 @@ def handle_update_instance_num(user_config):
96		92
97	def deploy_services_multi_yaml(paths, user_config):	93	def deploy_services_multi_yaml(paths, user_config):
98	deploy_config = user_config[C.MOTOR_DEPLOY_CONFIG]	94	deploy_config = user_config[C.MOTOR_DEPLOY_CONFIG]
99	- init_service_domain_name(	95	+ init_service_domain_name(paths, deploy_config)
100	- paths["controller_input_yaml"], paths["coordinator_input_yaml"],
101	- paths["kv_pool_input_yaml"], paths["kv_conductor_input_yaml"], deploy_config
102	- )
103	generate_yaml_controller(	96	generate_yaml_controller(
104	paths["controller_input_yaml"], paths["controller_output_yaml"], user_config	97	paths["controller_input_yaml"], paths["controller_output_yaml"], user_config
105	)	98	)
	@@ -118,15 +111,16 @@ def deploy_services_multi_yaml(paths, user_config):
118	paths["kv_conductor_input_yaml"], paths["kv_conductor_output_yaml"],	111	paths["kv_conductor_input_yaml"], paths["kv_conductor_output_yaml"],
119	user_config, kv_conductor_config	112	user_config, kv_conductor_config
120	)	113	)
		114	+ if k8s_utils.g_mf_store_enabled:
		115	+ generate_yaml_mf_store(
		116	+ paths["mf_store_input_yaml"], paths["mf_store_output_yaml"], user_config
		117	+ )
121	exec_all_kubectl_multi(deploy_config, None, C.DEPLOY_MODE_MULTI_DEPLOYMENT_YAML)	118	exec_all_kubectl_multi(deploy_config, None, C.DEPLOY_MODE_MULTI_DEPLOYMENT_YAML)
122		119
123		120
124	def deploy_services_infer_service_set(paths, user_config):	121	def deploy_services_infer_service_set(paths, user_config):
125	deploy_config = user_config[C.MOTOR_DEPLOY_CONFIG]	122	deploy_config = user_config[C.MOTOR_DEPLOY_CONFIG]
126	- init_service_domain_name(	123	+ init_service_domain_name(paths, deploy_config)
127	- paths["controller_input_yaml"], paths["coordinator_input_yaml"],
128	- paths["kv_pool_input_yaml"], paths["kv_conductor_input_yaml"], deploy_config
129	- )
130	infer_input = paths["infer_service_input_yaml"]	124	infer_input = paths["infer_service_input_yaml"]
131	if not os.path.exists(infer_input):	125	if not os.path.exists(infer_input):
132	raise FileNotFoundError(	126	raise FileNotFoundError(
	@@ -153,6 +147,8 @@ def deploy_services(user_config, env_config_path):
153	deploy_config = user_config[C.MOTOR_DEPLOY_CONFIG]	147	deploy_config = user_config[C.MOTOR_DEPLOY_CONFIG]
154	update_kv_pool_enabled_flag(user_config)	148	update_kv_pool_enabled_flag(user_config)
155	update_kv_conductor_enabled_flag(user_config)	149	update_kv_conductor_enabled_flag(user_config)
		150	+ update_engine_type_flag(user_config)
		151	+
156	update_engine_base_name(user_config)	152	update_engine_base_name(user_config)
157		153
158	deploy_mode_arg = get_deploy_mode_from_config(deploy_config)	154	deploy_mode_arg = get_deploy_mode_from_config(deploy_config)

Mexamples/deployer/lib/constant.py+10-0

	@@ -51,6 +51,7 @@ CONTROLLER_SHELL_PATH = os.path.join(STARTUP_ROOT_PATH, "roles/controller.sh")
51	COORDINATOR_SHELL_PATH = os.path.join(STARTUP_ROOT_PATH, "roles/coordinator.sh")	51	COORDINATOR_SHELL_PATH = os.path.join(STARTUP_ROOT_PATH, "roles/coordinator.sh")
52	ENGINE_SHELL_PATH = os.path.join(STARTUP_ROOT_PATH, "roles/engine.sh")	52	ENGINE_SHELL_PATH = os.path.join(STARTUP_ROOT_PATH, "roles/engine.sh")
53	KV_POOL_SHELL_PATH = os.path.join(STARTUP_ROOT_PATH, "roles/kv_pool.sh")	53	KV_POOL_SHELL_PATH = os.path.join(STARTUP_ROOT_PATH, "roles/kv_pool.sh")
		54	+MF_STORE_SHELL_PATH = os.path.join(STARTUP_ROOT_PATH, "roles/mf_store.sh")
54	SINGLE_CONTAINER_SHELL_PATH = os.path.join(STARTUP_ROOT_PATH, "roles/all_combine_in_single_container.sh")	55	SINGLE_CONTAINER_SHELL_PATH = os.path.join(STARTUP_ROOT_PATH, "roles/all_combine_in_single_container.sh")
55	MOTOR_COMMON_ENV = "motor_common_env"	56	MOTOR_COMMON_ENV = "motor_common_env"
56	WEIGHT_MOUNT = "weight-mount"	57	WEIGHT_MOUNT = "weight-mount"
	@@ -62,6 +63,7 @@ DEFAULT_KV_POOL_PORT = 50088
62	KV_CONDUCTOR_CONFIG = "kv_conductor_config"	63	KV_CONDUCTOR_CONFIG = "kv_conductor_config"
63	KV_CONDUCTOR_PORT = "http_server_port"	64	KV_CONDUCTOR_PORT = "http_server_port"
64	KV_CONDUCTOR_SHELL_PATH = os.path.join(STARTUP_ROOT_PATH, "roles/kv_conductor.sh")	65	KV_CONDUCTOR_SHELL_PATH = os.path.join(STARTUP_ROOT_PATH, "roles/kv_conductor.sh")
		66	+DEFAULT_MF_STORE_PORT = 50089
65	STANDBY_CONFIG = "standby_config"	67	STANDBY_CONFIG = "standby_config"
66	MOTOR_CONTROLLER_CONFIG = "motor_controller_config"	68	MOTOR_CONTROLLER_CONFIG = "motor_controller_config"
67	MOTOR_COORDINATOR_CONFIG = "motor_coordinator_config"	69	MOTOR_COORDINATOR_CONFIG = "motor_coordinator_config"
	@@ -95,6 +97,8 @@ SERVICE_ID = "SERVICE_ID"
95	ENGINE_TYPE = "ENGINE_TYPE"	97	ENGINE_TYPE = "ENGINE_TYPE"
96	NORTH_PLATFORM = "NORTH_PLATFORM"	98	NORTH_PLATFORM = "NORTH_PLATFORM"
97	MODEL_NAME = "MODEL_NAME"	99	MODEL_NAME = "MODEL_NAME"
		100	+SECURITY_CONTEXT = "securityContext"
		101	+PRIVILEGED = "privileged"
98		102
99	HARDWARE_TYPE_800I_A2 = "800I_A2"	103	HARDWARE_TYPE_800I_A2 = "800I_A2"
100	HARDWARE_TYPE_800I_A3 = "800I_A3"	104	HARDWARE_TYPE_800I_A3 = "800I_A3"
	@@ -121,6 +125,11 @@ ENV_KVP_MASTER_SERVICE = "KVP_MASTER_SERVICE"
121	ENV_KV_POOL_PORT = "KV_POOL_PORT"	125	ENV_KV_POOL_PORT = "KV_POOL_PORT"
122	ENV_KV_POOL_EVICTION_HIGH_WATERMARK_RATIO = "KV_POOL_EVICTION_HIGH_WATERMARK_RATIO"	126	ENV_KV_POOL_EVICTION_HIGH_WATERMARK_RATIO = "KV_POOL_EVICTION_HIGH_WATERMARK_RATIO"
123	ENV_KV_POOL_EVICTION_RATIO = "KV_POOL_EVICTION_RATIO"	127	ENV_KV_POOL_EVICTION_RATIO = "KV_POOL_EVICTION_RATIO"
		128	+ENV_DISAGGREGATION_BOOTSTRAP_PORT = "DISAGGREGATION_BOOTSTRAP_PORT"
		129	+ENV_ASCEND_MF_STORE_URL = "ASCEND_MF_STORE_URL"
		130	+ENV_ASCEND_MF_STORE_PORT = "ASCEND_MF_STORE_PORT"
		131	+ENV_ASCEND_MF_TRANSFER_PROTOCOL = "ASCEND_MF_TRANSFER_PROTOCOL"
		132	+ENV_SGLANG_HOST_IP = "SGLANG_HOST_IP"
124		133
125	VOLUMES = "volumes"	134	VOLUMES = "volumes"
126	VOLUME_MOUNTS = "volumeMounts"	135	VOLUME_MOUNTS = "volumeMounts"
	@@ -143,3 +152,4 @@ JOB_NAME = "job-name"
143	ROLES = "roles"	152	ROLES = "roles"
144	SERVICES = "services"	153	SERVICES = "services"
145	KIND_KEY = "kind"	154	KIND_KEY = "kind"
		155	+

Mexamples/deployer/lib/generator/coordinator.py+7-0

	@@ -55,6 +55,13 @@ def modify_coordinator_deployment(deployment_data, user_config):
55	{C.NAME: C.ENV_COORDINATOR_SERVICE, C.VALUE: k8s_utils.g_coordinator_service}	55	{C.NAME: C.ENV_COORDINATOR_SERVICE, C.VALUE: k8s_utils.g_coordinator_service}
56	])	56	])
57		57
		58	+ disaggregation_bootstrap_port = user_config.get(C.MOTOR_ENGINE_PREFILL_CONFIG, {}).get(C.ENGINE_CONFIG, {}) \
		59	+ .get("disaggregation_bootstrap_port", "")
		60	+ if disaggregation_bootstrap_port:
		61	+ container[C.ENV].append(
		62	+ {C.NAME: C.ENV_DISAGGREGATION_BOOTSTRAP_PORT, C.VALUE: str(disaggregation_bootstrap_port)}
		63	+ )
		64	+
58	modify_coordinator_replicas(deployment_data, user_config)	65	modify_coordinator_replicas(deployment_data, user_config)
59	modify_log_mount(deployment_data, user_config, "mindie-motor-coordinator")	66	modify_log_mount(deployment_data, user_config, "mindie-motor-coordinator")
60		67

Mexamples/deployer/lib/generator/engine.py+24-0

	@@ -36,6 +36,24 @@ def build_engine_env_items(role, job_name, include_kv_pool=False):
36	]	36	]
37	if include_kv_pool and k8s_utils.g_kv_pool_enabled:	37	if include_kv_pool and k8s_utils.g_kv_pool_enabled:
38	env_items.append({C.NAME: C.ENV_KVP_MASTER_SERVICE, C.VALUE: k8s_utils.g_kv_pool_service})	38	env_items.append({C.NAME: C.ENV_KVP_MASTER_SERVICE, C.VALUE: k8s_utils.g_kv_pool_service})
		39	+ if k8s_utils.g_mf_store_enabled:
		40	+ deploy_config = user_config[C.MOTOR_DEPLOY_CONFIG]
		41	+ ascend_mf_store_url = f"tcp://{k8s_utils.g_mf_store_service}:{C.DEFAULT_MF_STORE_PORT}"
		42	+ hardware_type = deploy_config.get(C.HARDWARE_TYPE, C.HARDWARE_TYPE_800I_A2)
		43	+ ascend_mf_transfer_protocol = "device_rdma" if hardware_type == C.HARDWARE_TYPE_800I_A2 else "sdma"
		44	+ env_items.extend([
		45	+ {C.NAME: C.ENV_ASCEND_MF_STORE_URL, C.VALUE: ascend_mf_store_url},
		46	+ {C.NAME: C.ENV_ASCEND_MF_TRANSFER_PROTOCOL, C.VALUE: ascend_mf_transfer_protocol}
		47	+ ])
		48	+ if k8s_utils.g_engine_type == C.ENGINE_TYPE_SGLANG:
		49	+ env_items.append({
		50	+ C.NAME: C.ENV_SGLANG_HOST_IP,
		51	+ "valueFrom": {
		52	+ "fieldRef": {
		53	+ "fieldPath": "status.podIP"
		54	+ }
		55	+ }
		56	+ })
39	return env_items	57	return env_items
40		58
41		59
	@@ -128,6 +146,12 @@ def set_engine_weight_mount(deployment_data, container, deploy_config):
128	def modify_engine_yaml(deployment_data, user_config, index, node_type):	146	def modify_engine_yaml(deployment_data, user_config, index, node_type):
129	deploy_config = user_config[C.MOTOR_DEPLOY_CONFIG]	147	deploy_config = user_config[C.MOTOR_DEPLOY_CONFIG]
130	container = deployment_data[C.SPEC][C.TEMPLATE][C.SPEC][C.CONTAINERS][0]	148	container = deployment_data[C.SPEC][C.TEMPLATE][C.SPEC][C.CONTAINERS][0]
		149	+
		150	+ if k8s_utils.g_engine_type == C.ENGINE_TYPE_SGLANG:
		151	+ if C.SECURITY_CONTEXT not in container:
		152	+ container[C.SECURITY_CONTEXT] = {}
		153	+ container[C.SECURITY_CONTEXT][C.PRIVILEGED] = True
		154	+
131	container[C.IMAGE] = deploy_config[C.IMAGE_NAME]	155	container[C.IMAGE] = deploy_config[C.IMAGE_NAME]
132	job_name = f"{deploy_config[C.CONFIG_JOB_ID]}-{node_type}{index}-{generate_unique_id()}"	156	job_name = f"{deploy_config[C.CONFIG_JOB_ID]}-{node_type}{index}-{generate_unique_id()}"
133	set_engine_metadata(deployment_data, deploy_config, index, node_type, job_name)	157	set_engine_metadata(deployment_data, deploy_config, index, node_type, job_name)

Mexamples/deployer/lib/generator/k8s_utils.py+33-6

	@@ -23,6 +23,9 @@ g_kv_conductor_enabled = False
23	g_engine_base_name = "mindie-server"	23	g_engine_base_name = "mindie-server"
24	g_generate_yaml_list = []	24	g_generate_yaml_list = []
25	g_user_config_path = None	25	g_user_config_path = None
		26	+g_mf_store_service = "mf_store"
		27	+g_mf_store_enabled = False
		28	+g_engine_type = "vllm"
26		29
27		30
28	def set_user_config_path(path):	31	def set_user_config_path(path):
	@@ -50,6 +53,11 @@ def set_kv_conductor_service(service_name):
50	g_kv_conductor_service = service_name	53	g_kv_conductor_service = service_name
51		54
52		55
		56	+def set_mf_store_service(service_name):
		57	+ global g_mf_store_service
		58	+ g_mf_store_service = service_name
		59	+
		60	+
53	def set_engine_base_name(engine_name):	61	def set_engine_base_name(engine_name):
54	global g_engine_base_name	62	global g_engine_base_name
55	g_engine_base_name = engine_name	63	g_engine_base_name = engine_name
	@@ -77,6 +85,16 @@ def update_kv_conductor_enabled_flag(user_config):
77	g_kv_conductor_enabled = True	85	g_kv_conductor_enabled = True
78		86
79		87
		88	+def update_engine_type_flag(user_config):
		89	+ global g_engine_type
		90	+ global g_mf_store_enabled
		91	+ g_mf_store_enabled = False
		92	+
		93	+ g_engine_type = user_config.get(C.MOTOR_ENGINE_PREFILL_CONFIG, {}).get("engine_type", "")
		94	+ if g_engine_type == C.ENGINE_TYPE_SGLANG:
		95	+ g_mf_store_enabled = True
		96	+
		97	+
80	def get_deploy_mode_from_config(deploy_config):	98	def get_deploy_mode_from_config(deploy_config):
81	"""Read deploy_mode from motor_deploy_config; default infer_service_set; validate value."""	99	"""Read deploy_mode from motor_deploy_config; default infer_service_set; validate value."""
82	mode = deploy_config.get(C.DEPLOY_MODE_CONFIG_KEY, C.DEPLOY_MODE_INFER_SERVICE_SET)	100	mode = deploy_config.get(C.DEPLOY_MODE_CONFIG_KEY, C.DEPLOY_MODE_INFER_SERVICE_SET)
	@@ -88,12 +106,12 @@ def get_deploy_mode_from_config(deploy_config):
88	return mode	106	return mode
89		107
90		108
91	-def init_service_domain_name(controller_input_yaml, coordinator_input_yaml, kv_pool_input_yaml,	109	+def init_service_domain_name(paths, deploy_config):
92	- kv_conductor_input_yaml, deploy_config):	110	+ controller_data = load_yaml(paths["controller_input_yaml"], False)
93	- controller_data = load_yaml(controller_input_yaml, False)	111	+ coordinator_data = load_yaml(paths["coordinator_input_yaml"], False)
94	- coordinator_data = load_yaml(coordinator_input_yaml, False)	112	+ kv_pool_data = load_yaml(paths["kv_pool_input_yaml"], False)
95	- kv_pool_data = load_yaml(kv_pool_input_yaml, False)	113	+ kv_conductor_data = load_yaml(paths["kv_conductor_input_yaml"], False)
96	- kv_conductor_data = load_yaml(kv_conductor_input_yaml, False)	114	+ mf_store_data = load_yaml(paths["mf_store_input_yaml"], False)
97		115
98	controller_service_data = None	116	controller_service_data = None
99	for doc in controller_data:	117	for doc in controller_data:
	@@ -119,6 +137,12 @@ def init_service_domain_name(controller_input_yaml, coordinator_input_yaml, kv_p
119	kv_conductor_service_data = doc	137	kv_conductor_service_data = doc
120	break	138	break
121		139
		140	+ mf_store_service_data = None
		141	+ for doc in mf_store_data:
		142	+ if doc.get(C.KIND) == C.SERVICE:
		143	+ mf_store_service_data = doc
		144	+ break
		145	+
122	controller_name = controller_service_data[C.METADATA][C.NAME]	146	controller_name = controller_service_data[C.METADATA][C.NAME]
123	set_controller_service(f"{controller_name}.{deploy_config[C.CONFIG_JOB_ID]}.svc.cluster.local")	147	set_controller_service(f"{controller_name}.{deploy_config[C.CONFIG_JOB_ID]}.svc.cluster.local")
124	coordinator_name = coordinator_service_data[C.METADATA][C.NAME]	148	coordinator_name = coordinator_service_data[C.METADATA][C.NAME]
	@@ -127,6 +151,8 @@ def init_service_domain_name(controller_input_yaml, coordinator_input_yaml, kv_p
127	set_kv_pool_service(f"{kv_pool_name}.{deploy_config[C.CONFIG_JOB_ID]}.svc.cluster.local")	151	set_kv_pool_service(f"{kv_pool_name}.{deploy_config[C.CONFIG_JOB_ID]}.svc.cluster.local")
128	kv_conductor_name = kv_conductor_service_data[C.METADATA][C.NAME]	152	kv_conductor_name = kv_conductor_service_data[C.METADATA][C.NAME]
129	set_kv_conductor_service(f"{kv_conductor_name}.{deploy_config[C.CONFIG_JOB_ID]}.svc.cluster.local")	153	set_kv_conductor_service(f"{kv_conductor_name}.{deploy_config[C.CONFIG_JOB_ID]}.svc.cluster.local")
		154	+ mf_store_name = mf_store_service_data[C.METADATA][C.NAME]
		155	+ set_mf_store_service(f"{mf_store_name}.{deploy_config[C.CONFIG_JOB_ID]}.svc.cluster.local")
130		156
131		157
132	def run_cmd_get_output(args):	158	def run_cmd_get_output(args):
	@@ -241,6 +267,7 @@ def create_motor_config_configmap(job_id):
241	f"--from-file=./{C.STARTUP_ROOT_PATH}/roles/engine.sh "	267	f"--from-file=./{C.STARTUP_ROOT_PATH}/roles/engine.sh "
242	f"--from-file=./{C.STARTUP_ROOT_PATH}/roles/kv_pool.sh "	268	f"--from-file=./{C.STARTUP_ROOT_PATH}/roles/kv_pool.sh "
243	f"--from-file=./{C.STARTUP_ROOT_PATH}/roles/kv_conductor.sh "	269	f"--from-file=./{C.STARTUP_ROOT_PATH}/roles/kv_conductor.sh "
		270	+ f"--from-file=./{C.STARTUP_ROOT_PATH}/roles/mf_store.sh "
244	f"--from-file=./{C.STARTUP_ROOT_PATH}/roles/all_combine_in_single_container.sh "	271	f"--from-file=./{C.STARTUP_ROOT_PATH}/roles/all_combine_in_single_container.sh "
245	"--from-file=./probe/probe.sh "	272	"--from-file=./probe/probe.sh "
246	"--from-file=./probe/probe.py "	273	"--from-file=./probe/probe.py "

Aexamples/deployer/lib/generator/mf_store.py+41-0

@@ -0,0 +1,41 @@
	1	+# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
	2	+# MindIE is licensed under Mulan PSL v2.
	3	+# You can use this software according to the terms and conditions of the Mulan PSL v2.
	4	+# You may obtain a copy of Mulan PSL v2 at:
	5	+# http://license.coscl.org.cn/MulanPSL2
	6	+# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
	7	+# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
	8	+# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
	9	+# See the Mulan PSL v2 for more details.
	10	+
	11	+import lib.constant as C
	12	+from lib.utils import load_yaml, write_yaml, logger
	13	+from lib.generator import k8s_utils
	14	+
	15	+
	16	+def generate_yaml_mf_store(input_yaml, output_file, user_config):
	17	+ logger.info(f"Generating YAML from {input_yaml} to {output_file}")
	18	+ deploy_config = user_config[C.MOTOR_DEPLOY_CONFIG]
	19	+ data = load_yaml(input_yaml, False)
	20	+ deployment_data = data[0]
	21	+ deployment_data[C.METADATA][C.NAMESPACE] = deploy_config[C.CONFIG_JOB_ID]
	22	+
	23	+ container = deployment_data[C.SPEC][C.TEMPLATE][C.SPEC][C.CONTAINERS][0]
	24	+ container[C.IMAGE] = deploy_config[C.IMAGE_NAME]
	25	+
	26	+ if C.ENV not in container:
	27	+ container[C.ENV] = []
	28	+ container[C.ENV].append(
	29	+ {C.NAME: C.ENV_ASCEND_MF_STORE_PORT, C.VALUE: str(C.DEFAULT_MF_STORE_PORT)}
	30	+ )
	31	+
	32	+ service_data = data[1]
	33	+ service_data[C.METADATA][C.NAMESPACE] = deploy_config[C.CONFIG_JOB_ID]
	34	+ ports = service_data.get(C.SPEC, {}).get(C.PORTS, [])
	35	+ if not ports:
	36	+ raise ValueError(f"Missing required service ports in {input_yaml}.")
	37	+ ports[0][C.PORT] = C.DEFAULT_MF_STORE_PORT
	38	+ ports[0][C.TARGET_PORT] = C.DEFAULT_MF_STORE_PORT
	39	+
	40	+ write_yaml(data, output_file, False)
	41	+ k8s_utils.g_generate_yaml_list.append(output_file)

Mexamples/deployer/lib/utils.py+5-1

	@@ -210,6 +210,7 @@ def set_env_to_shell(user_config, env_config_path, deploy_mode):
210	update_shell_safely(C.SINGLE_CONTAINER_SHELL_PATH, env_config, "motor_engine_prefill_env", "set_prefill_env")	210	update_shell_safely(C.SINGLE_CONTAINER_SHELL_PATH, env_config, "motor_engine_prefill_env", "set_prefill_env")
211	update_shell_safely(C.SINGLE_CONTAINER_SHELL_PATH, env_config, "motor_engine_decode_env", "set_decode_env")	211	update_shell_safely(C.SINGLE_CONTAINER_SHELL_PATH, env_config, "motor_engine_decode_env", "set_decode_env")
212	update_shell_safely(C.SINGLE_CONTAINER_SHELL_PATH, env_config, "motor_kv_cache_pool_env", "set_kv_pool_env")	212	update_shell_safely(C.SINGLE_CONTAINER_SHELL_PATH, env_config, "motor_kv_cache_pool_env", "set_kv_pool_env")
		213	+ update_shell_safely(C.MF_STORE_SHELL_PATH, env_config, "motor_mf_store_env", "set_mf_store_env")
213	update_shell_safely(	214	update_shell_safely(
214	C.SINGLE_CONTAINER_SHELL_PATH, env_config, "motor_kv_conductor_env", "set_kv_conductor_env"	215	C.SINGLE_CONTAINER_SHELL_PATH, env_config, "motor_kv_conductor_env", "set_kv_conductor_env"
215	)	216	)
	@@ -219,6 +220,7 @@ def set_env_to_shell(user_config, env_config_path, deploy_mode):
219	update_shell_safely(C.ENGINE_SHELL_PATH, env_config, "motor_engine_prefill_env", "set_prefill_env")	220	update_shell_safely(C.ENGINE_SHELL_PATH, env_config, "motor_engine_prefill_env", "set_prefill_env")
220	update_shell_safely(C.ENGINE_SHELL_PATH, env_config, "motor_engine_decode_env", "set_decode_env")	221	update_shell_safely(C.ENGINE_SHELL_PATH, env_config, "motor_engine_decode_env", "set_decode_env")
221	update_shell_safely(C.KV_POOL_SHELL_PATH, env_config, "motor_kv_cache_pool_env", "set_kv_pool_env")	222	update_shell_safely(C.KV_POOL_SHELL_PATH, env_config, "motor_kv_cache_pool_env", "set_kv_pool_env")
		223	+ update_shell_safely(C.MF_STORE_SHELL_PATH, env_config, "motor_mf_store_env", "set_mf_store_env")
222	update_shell_safely(	224	update_shell_safely(
223	C.KV_CONDUCTOR_SHELL_PATH, env_config, "motor_kv_conductor_env", "set_kv_conductor_env"	225	C.KV_CONDUCTOR_SHELL_PATH, env_config, "motor_kv_conductor_env", "set_kv_conductor_env"
224	)	226	)
	@@ -240,5 +242,7 @@ def get_deploy_paths():
240	"infer_service_input_yaml": os.path.join(C.DEPLOY_YAML_ROOT_PATH, 'infer_service_template.yaml'),	242	"infer_service_input_yaml": os.path.join(C.DEPLOY_YAML_ROOT_PATH, 'infer_service_template.yaml'),
241	"infer_service_output_yaml": os.path.join(C.OUTPUT_ROOT_PATH, 'infer_service.yaml'),	243	"infer_service_output_yaml": os.path.join(C.OUTPUT_ROOT_PATH, 'infer_service.yaml'),
242	"single_container_input_yaml": os.path.join(C.DEPLOY_YAML_ROOT_PATH, 'single_container_template.yaml'),	244	"single_container_input_yaml": os.path.join(C.DEPLOY_YAML_ROOT_PATH, 'single_container_template.yaml'),
243	- "single_container_output_yaml": os.path.join(C.OUTPUT_ROOT_PATH, 'mindie_motor_single_container.yaml')	245	+ "single_container_output_yaml": os.path.join(C.OUTPUT_ROOT_PATH, 'mindie_motor_single_container.yaml'),
		246	+ "mf_store_input_yaml": os.path.join(C.DEPLOY_YAML_ROOT_PATH, 'mf_store_template.yaml'),
		247	+ "mf_store_output_yaml": os.path.join(C.OUTPUT_ROOT_PATH, 'mindie_motor_mf_store.yaml')
244	}	248	}

Mexamples/deployer/startup/boot.sh+3-0

	@@ -31,6 +31,9 @@ case "$ROLE" in
31	"kv_conductor")	31	"kv_conductor")
32	source "$SCRIPT_DIR/kv_conductor.sh"	32	source "$SCRIPT_DIR/kv_conductor.sh"
33	;;	33	;;
		34	+ "mf_store")
		35	+ source "$SCRIPT_DIR/mf_store.sh"
		36	+ ;;
34	*)	37	*)
35	echo "Error: Unknown ROLE=$ROLE"	38	echo "Error: Unknown ROLE=$ROLE"
36	echo "Valid roles: SINGLE_CONTAINER, prefill, decode, controller, coordinator, kv_pool, kv_conductor"	39	echo "Valid roles: SINGLE_CONTAINER, prefill, decode, controller, coordinator, kv_pool, kv_conductor"

Mexamples/deployer/startup/common.sh+44-0

	@@ -132,3 +132,47 @@ gen_kv_pool_config() {
132	python3 "$CONFIGMAP_PATH/mooncake_config.py" pool "$MOONCAKE_CONFIG_PATH" "$USER_CONFIG_PATH"	132	python3 "$CONFIGMAP_PATH/mooncake_config.py" pool "$MOONCAKE_CONFIG_PATH" "$USER_CONFIG_PATH"
133	fi	133	fi
134	}	134	}
		135	+
		136	+set_mf_store_env() {
		jason lyu3月18日这个环境变量函数只是sglang使用的，放在common.sh里不是很合适，建议在dpeloyer的generator/engine.py里走插入函数逻辑
		137	+ # convert ASCEND_MF_STORE_URL to IP
		138	+ if [ -n "$ASCEND_MF_STORE_URL" ]; then
		139	+ if [[ "$ASCEND_MF_STORE_URL" =~ ^(tcp://)?([^:/]+)(:([0-9]+))?$ ]]; then
		140	+ PROTO="${BASH_REMATCH[1]}"
		141	+ HOST="${BASH_REMATCH[2]}"
		142	+ PORT="${BASH_REMATCH[4]}"
		143	+
		144	+ if [[ ! "$HOST" =~ ^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$ ]]; then
		145	+ MAX_RETRY=5
		146	+ RETRY_INTERVAL=10
		147	+ RETRY_COUNT=0
		148	+ MF_STORE_POD_IP=""
		149	+ while [ $RETRY_COUNT -lt $MAX_RETRY ]; do
		150	+ MF_STORE_POD_IP=$(getent hosts "$HOST" \| awk '{print $1}' \| head -n1)
		151	+
		152	+ if [ -n "$MF_STORE_POD_IP" ]; then
		153	+ break
		154	+ fi
		155	+
		156	+ RETRY_COUNT=$((RETRY_COUNT+1))
		157	+ echo "resolve $HOST failed, retry $RETRY_COUNT/$MAX_RETRY ..."
		158	+ sleep $RETRY_INTERVAL
		159	+ done
		160	+
		161	+ if [ -z "$MF_STORE_POD_IP" ]; then
		162	+ echo "get pod ip error: $HOST"
		163	+ exit 1
		164	+ else
		165	+ echo "$HOST pod ip: $MF_STORE_POD_IP"
		166	+ export ASCEND_MF_STORE_URL="${PROTO}${MF_STORE_POD_IP}:${PORT}"
		167	+ fi
		168	+ else
		169	+ echo "HOST is already IP: $HOST"
		170	+ fi
		171	+ else
		172	+ echo "ASCEND_MF_STORE_URL format invalid: $ASCEND_MF_STORE_URL"
		173	+ exit 1
		174	+ fi
		175	+
		176	+ echo "ASCEND_MF_STORE_URL: $ASCEND_MF_STORE_URL"
		177	+ fi
		178	+}

Mexamples/deployer/startup/roles/engine.sh+2-0

	@@ -21,6 +21,8 @@ gen_kv_pool_config
21		21
22	set_cann_env	22	set_cann_env
23		23
		24	+set_mf_store_env
		25	+
24	# CRD scenario: refresh JOB_NAME with INFER_SERVICE_INDEX and INSTANCE_INDEX injected by CRD	26	# CRD scenario: refresh JOB_NAME with INFER_SERVICE_INDEX and INSTANCE_INDEX injected by CRD
25	# Final format: {namespace}-{InferServiceSet_name}-{INFER_SERVICE_INDEX}-p/d{INSTANCE_INDEX}	27	# Final format: {namespace}-{InferServiceSet_name}-{INFER_SERVICE_INDEX}-p/d{INSTANCE_INDEX}
26	if [ -n "$INFER_SERVICE_INDEX" ] && [ -n "$INSTANCE_INDEX" ]; then	28	if [ -n "$INFER_SERVICE_INDEX" ] && [ -n "$INSTANCE_INDEX" ]; then

Aexamples/deployer/startup/roles/mf_store.sh+20-0

@@ -0,0 +1,20 @@
	1	+#!/bin/bash
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
	3	+# MindIE is licensed under Mulan PSL v2.
	4	+# You can use this software according to the terms and conditions of the Mulan PSL v2.
	5	+# You may obtain a copy of Mulan PSL v2 at:
	6	+# http://license.coscl.org.cn/MulanPSL2
	7	+# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
	8	+# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
	9	+# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
	10	+# See the Mulan PSL v2 for more details.
	11	+
	12	+if [ "$ROLE" != "mf_store" ]; then
	13	+ echo "Error: This script is for mf_store role only. Current ROLE=$ROLE"
	14	+ exit 1
	15	+fi
	16	+
	17	+export ASCEND_MF_STORE_URL="tcp://$POD_IP:$ASCEND_MF_STORE_PORT"
	18	+export ASCEND_MF_LOG_LEVEL=0
	19	+
	20	+python3 -m memfabric_hybrid.launch_ascend_mf_store

Mexamples/deployer/yaml_template/engine_template.yaml+15-0

	@@ -85,6 +85,12 @@ spec:
85	mountPath: /mnt/weight	85	mountPath: /mnt/weight
86	- name: plog-path	86	- name: plog-path
87	mountPath: /root/ascend/log	87	mountPath: /root/ascend/log
		88	+ - name: ascend-driver
		89	+ mountPath: /usr/local/Ascend/driver
		90	+ - name: rdma
		91	+ mountPath: /dev/infiniband
		92	+ - name: hisi-hdc
		93	+ mountPath: /dev/hisi_hdc
88	volumes:	94	volumes:
89	- name: data	95	- name: data
90	hostPath:	96	hostPath:
	@@ -120,3 +126,12 @@ spec:
120	hostPath:	126	hostPath:
121	path: /root/ascend/log	127	path: /root/ascend/log
122	type: DirectoryOrCreate	128	type: DirectoryOrCreate
		129	+ - name: ascend-driver
		130	+ hostPath:
		131	+ path: /usr/local/Ascend/driver
		132	+ - name: rdma
		133	+ hostPath:
		134	+ path: /dev/infiniband
		135	+ - name: hisi-hdc
		136	+ hostPath:
		137	+ path: /dev/hisi_hdc

Aexamples/deployer/yaml_template/mf_store_template.yaml+91-0

@@ -0,0 +1,91 @@
	1	+apiVersion: apps/v1
	2	+kind: Deployment
	3	+metadata:
	4	+ name: mindie-motor-mf-store
	5	+ labels:
	6	+ app: mindie-motor-mf-store
	7	+ namespace: mindie
	8	+spec:
	9	+ replicas: 1
	10	+ selector:
	11	+ matchLabels:
	12	+ app: mindie-motor-mf-store
	13	+ template:
	14	+ metadata:
	15	+ labels:
	16	+ app: mindie-motor-mf-store
	17	+ deploy-name: mindie-motor-mf-store
	18	+ spec:
	19	+ terminationGracePeriodSeconds: 0
	20	+ automountServiceAccountToken: false
	21	+ securityContext:
	22	+ fsGroup: 1001
	23	+ containers:
	24	+ - image: mindie:1.0.0-aarch64-800I-A2
	25	+ imagePullPolicy: IfNotPresent
	26	+ name: mindie-motor-mf-store
	27	+ securityContext:
	28	+ allowPrivilegeEscalation: false
	29	+ capabilities:
	30	+ drop: ["ALL"]
	31	+ seccompProfile:
	32	+ type: "RuntimeDefault"
	33	+ env:
	34	+ - name: POD_IP
	35	+ valueFrom:
	36	+ fieldRef:
	37	+ fieldPath: status.podIP
	38	+ - name: CONFIGMAP_PATH
	39	+ value: /mnt/configmap
	40	+ - name: CONFIG_PATH
	41	+ value: /usr/local/Ascend/pyMotor/conf
	42	+ - name: ROLE
	43	+ value: mf_store
	44	+ command: ["/bin/bash", "-c", "
	45	+ source /mnt/configmap/boot.sh; \n
	46	+ "]
	47	+ resources:
	48	+ requests:
	49	+ memory: "2Gi"
	50	+ cpu: "4"
	51	+ limits:
	52	+ memory: "4Gi"
	53	+ cpu: "8"
	54	+ volumeMounts:
	55	+ - name: motor-config
	56	+ mountPath: /mnt/configmap
	57	+ - name: coredump
	58	+ mountPath: /var/coredump
	59	+ - name: ascend-driver
	60	+ mountPath: /usr/local/Ascend/driver
	61	+ volumes:
	62	+ - name: motor-config
	63	+ configMap:
	64	+ name: motor-config
	65	+ defaultMode: 0550
	66	+ - name: coredump
	67	+ hostPath:
	68	+ path: /var/coredump
	69	+ type: DirectoryOrCreate
	70	+ - name: ascend-driver
	71	+ hostPath:
	72	+ path: /usr/local/Ascend/driver
	73	+---
	74	+apiVersion: v1
	75	+kind: Service
	76	+metadata:
	77	+ labels:
	78	+ app: mindie-motor-mf-store
	79	+ name: mf-store
	80	+ namespace: mindie
	81	+spec:
	82	+ ports:
	83	+ - port: 50089
	84	+ protocol: TCP
	85	+ targetPort: 50089
	86	+ selector:
	87	+ app: mindie-motor-mf-store
	88	+ sessionAffinity: None
	89	+ clusterIP: None
	90	+status:
	91	+ loadBalancer: {}

Aexamples/infer_engines/sglang/env.json+35-0

@@ -0,0 +1,35 @@
	1	+{
	2	+ "version": "2.0.0",
	3	+ "motor_common_env": {
	4	+ "CANN_INSTALL_PATH": "/usr/local/Ascend",
	5	+ "MOTOR_LOG_ROOT_PATH": "/root/ascend/log"
	6	+ },
	7	+ "motor_controller_env": {
	8	+ },
	9	+ "motor_coordinator_env": {
	10	+ },
	11	+ "motor_engine_prefill_env": {
	12	+ "HCCL_BUFFSIZE": 200,
	13	+ "PYTORCH_NPU_ALLOC_CONF": "expandable_segments:True",
	14	+ "HCCL_OP_EXPANSION_MODE": "AIV",
	15	+ "OMP_PROC_BIND": "false",
	16	+ "OMP_NUM_THREADS": 100,
	17	+ "ASCEND_BUFFER_POOL": "4:8",
	18	+ "SGLANG_DEEPEP_NUM_MAX_DISPATCH_TOKENS_PER_RANK": 48,
	19	+ "STREAMS_PER_DEVICE": 32
	20	+ },
	21	+ "motor_engine_decode_env": {
	22	+ "HCCL_BUFFSIZE": 200,
	23	+ "PYTORCH_NPU_ALLOC_CONF": "expandable_segments:True",
	24	+ "HCCL_OP_EXPANSION_MODE": "AIV",
	25	+ "OMP_PROC_BIND": "false",
	26	+ "OMP_NUM_THREADS": 100,
	27	+ "ASCEND_BUFFER_POOL": "4:8",
	28	+ "SGLANG_DEEPEP_NUM_MAX_DISPATCH_TOKENS_PER_RANK": 48,
	29	+ "STREAMS_PER_DEVICE": 32
	30	+ },
	31	+ "motor_kv_cache_pool_env": {
	32	+ },
	33	+ "motor_kv_conductor_env": {
	34	+ }
	35	+}

Aexamples/infer_engines/sglang/user_config.json+114-0

@@ -0,0 +1,114 @@
	1	+{
	2	+ "version": "v2.0",
	3	+ "motor_deploy_config": {
	4	+ "p_instances_num": 1,
	5	+ "d_instances_num": 1,
	6	+ "single_p_instance_pod_num": 1,
	7	+ "single_d_instance_pod_num": 1,
	8	+ "p_pod_npu_num": 2,
	9	+ "d_pod_npu_num": 2,
	10	+ "image_name": "sglang:cann8.3.rc2-910b-release1225",
	11	+ "job_id": "mindie-lhb",
	12	+ "hardware_type": "800I_A2",
	13	+ "weight_mount_path": "/mnt/weight/",
	14	+ "deploy_mode": "multi_deployment"
	15	+ },
	16	+ "motor_controller_config": {
	17	+ "logging_config": {
	18	+ "log_level": "INFO"
	19	+ },
	20	+ "fault_tolerance_config": {
	21	+ "enable_fault_tolerance": true,
	22	+ "enable_scale_p2d": true,
	23	+ "enable_lingqu_network_recover": true
	24	+ },
	25	+ "standby_config": {
	26	+ "enable_master_standby": false
	27	+ }
	28	+ },
	29	+ "motor_coordinator_config": {
	30	+ "logging_config": {
	31	+ "log_level": "INFO"
	32	+ },
	33	+ "standby_config": {
	34	+ "enable_master_standby": false
	35	+ },
	36	+ "request_limit": {
	37	+ "single_node_max_requests": 4096,
	38	+ "max_requests": 10000
	39	+ },
	40	+ "scheduler_config": {
	41	+ "deploy_mode": "pd_dual_dispatch",
	42	+ "scheduler_type": "load_balance"
	43	+ }
	44	+ },
	45	+ "motor_nodemanger_config": {
	46	+ "logging_config": {
	47	+ "log_level": "INFO"
	48	+ }
	49	+ },
	50	+ "motor_engine_prefill_config": {
	51	+ "engine_type": "sglang",
	52	+ "enable_multi_endpoints": false,
	53	+ "model_config": {
	54	+ "model_name": "qwen3-8B",
	55	+ "model_path": "/mnt/share/weight/Qwen3-8B",
	56	+ "npu_mem_utils": 0.9,
	57	+ "prefill_parallel_config": {
	58	+ "dp_size": 1,
	59	+ "tp_size": 2,
	60	+ "pp_size": 1,
	61	+ "world_size": 2,
	62	+ "dp_rpc_port": 9000
	63	+ }
	64	+ },
	65	+ "engine_config": {
	66	+ "disaggregation_bootstrap_port": 9100,
	67	+ "enable_dp_attention": false,
	68	+ "moe_a2a_backend": "none",
	69	+ "disaggregation_transfer_backend": "ascend",
	70	+ "attention_backend": "ascend",
	71	+ "trust_remote_code": true,
	72	+ "log_level": "info",
	73	+ "load_balance_method": "round_robin",
	74	+ "base_gpu_id": 0,
	75	+ "nnodes": 1
	76	+ },
	77	+ "health_check_config": {
	78	+ "enable_virtual_inference": false,
	79	+ "npu_usage_threshold": 10
	80	+ }
	81	+ },
	82	+ "motor_engine_decode_config": {
	83	+ "engine_type": "sglang",
	84	+ "enable_multi_endpoints": false,
	85	+ "model_config": {
	86	+ "model_name": "qwen3-8B",
	87	+ "model_path": "/mnt/share/weight/Qwen3-8B",
	88	+ "npu_mem_utils": 0.9,
	89	+ "decode_parallel_config": {
	90	+ "dp_size": 1,
	91	+ "tp_size": 2,
	92	+ "pp_size": 1,
	93	+ "world_size": 2,
	94	+ "dp_rpc_port": 9000
	95	+ }
	96	+ },
	97	+ "engine_config": {
	98	+ "enable_dp_attention": false,
	99	+ "moe_a2a_backend": "none",
	100	+ "disaggregation_transfer_backend": "ascend",
	101	+ "attention_backend": "ascend",
	102	+ "trust_remote_code": true,
	103	+ "log_level": "info",
	104	+ "prefill_round_robin_balance": true,
	105	+ "base_gpu_id": 4,
	106	+ "nnodes": 1
	107	+ },
	108	+ "health_check_config": {
	109	+ "enable_virtual_inference": false,
	110	+ "npu_usage_threshold": 10
	111	+ }
	112	+ },
	113	+ "kv_cache_pool_config": {}
	114	+}

Mmotor/common/utils/env.py+4-0

	@@ -58,4 +58,8 @@ class Env:
58	def conductor_service(self):	58	def conductor_service(self):
59	return os.getenv("KV_CONDUCTOR_SERVICE", "")	59	return os.getenv("KV_CONDUCTOR_SERVICE", "")
60		60
		61	+ @property
		62	+ def disaggregation_bootstrap_port(self):
		63	+ return os.getenv("DISAGGREGATION_BOOTSTRAP_PORT", "")
		64	+
61	Env = Env()	65	Env = Env()

Mmotor/config/coordinator.py+1-0

	@@ -98,6 +98,7 @@ class DeployMode(Enum):
98	PD_SEPARATE = "pd_separate"	98	PD_SEPARATE = "pd_separate"
99	CDP_SEPARATE = "cdp_separate"	99	CDP_SEPARATE = "cdp_separate"
100	CPCD_SEPARATE = "cpcd_separate"	100	CPCD_SEPARATE = "cpcd_separate"
		101	+ PD_DUAL_DISPATCH = "pd_dual_dispatch"
101	PD_DISAGGREGATION_SINGLE_CONTAINER = "pd_disaggregation_single_container"	102	PD_DISAGGREGATION_SINGLE_CONTAINER = "pd_disaggregation_single_container"
102		103
103	@classmethod	104	@classmethod

Mmotor/coordinator/domain/instance_manager.py+1-0

	@@ -88,6 +88,7 @@ class InstanceManager:
88	DeployMode.CPCD_SEPARATE,	88	DeployMode.CPCD_SEPARATE,
89	DeployMode.PD_SEPARATE,	89	DeployMode.PD_SEPARATE,
90	DeployMode.PD_DISAGGREGATION_SINGLE_CONTAINER,	90	DeployMode.PD_DISAGGREGATION_SINGLE_CONTAINER,
		91	+ DeployMode.PD_DUAL_DISPATCH,
91	):	92	):
92	if has_p and has_d:	93	if has_p and has_d:
93	return InstanceReadiness.REQUIRED_MET	94	return InstanceReadiness.REQUIRED_MET

Mmotor/coordinator/router/base_router.py+13-0

	@@ -117,6 +117,19 @@ class BaseRouter(ABC):
117	async def handle_request(self) -> StreamingResponse \| JSONResponse:	117	async def handle_request(self) -> StreamingResponse \| JSONResponse:
118	pass	118	pass
119		119
		120	+ @contextlib.asynccontextmanager
		121	+ async def _manage_request_context(self):
		122	+ """
		123	+ Lifecycle management for request in the RequestManager.
		124	+ Ensures request info is added and cleaned up.
		125	+ """
		126	+ await self._request_manager.add_req_info(self.req_info)
		127	+ try:
		128	+ yield
		129	+ finally:
		130	+ await self._request_manager.del_req_info(self.req_info.req_id)
		131	+ self._log_request_details()
		132	+
120	@contextlib.asynccontextmanager	133	@contextlib.asynccontextmanager
121	async def _manage_client_context(self, resource: ScheduledResource):	134	async def _manage_client_context(self, resource: ScheduledResource):
122	endpoint = resource.endpoint	135	endpoint = resource.endpoint

Mmotor/coordinator/router/router.py+2-0

	@@ -37,6 +37,7 @@ from motor.common.resources.instance import PDRole
37	from motor.coordinator.router.pd_hybrid_router import PDHybridRouter	37	from motor.coordinator.router.pd_hybrid_router import PDHybridRouter
38	from motor.coordinator.router.separate_pd_router import SeparatePDRouter	38	from motor.coordinator.router.separate_pd_router import SeparatePDRouter
39	from motor.coordinator.router.separate_cdp_router import SeparateCDPRouter	39	from motor.coordinator.router.separate_cdp_router import SeparateCDPRouter
		40	+from motor.coordinator.router.separate_pd_dual_dispatch_router import SeparatePDDualDispatchRouter
40	from motor.common.utils.security_utils import (	41	from motor.common.utils.security_utils import (
41	sanitize_error_message,	42	sanitize_error_message,
42	filter_sensitive_headers,	43	filter_sensitive_headers,
	@@ -53,6 +54,7 @@ _ROUTER_MAP: dict[DeployMode, type['BaseRouter']] = {
53	DeployMode.CPCD_SEPARATE: SeparatePDRouter,	54	DeployMode.CPCD_SEPARATE: SeparatePDRouter,
54	DeployMode.SINGLE_NODE: PDHybridRouter,	55	DeployMode.SINGLE_NODE: PDHybridRouter,
55	DeployMode.PD_DISAGGREGATION_SINGLE_CONTAINER: SeparateCDPRouter,	56	DeployMode.PD_DISAGGREGATION_SINGLE_CONTAINER: SeparateCDPRouter,
		57	+ DeployMode.PD_DUAL_DISPATCH: SeparatePDDualDispatchRouter,
56	}	58	}
57		59
58		60

Mmotor/coordinator/router/separate_cdp_router.py+0-14

	@@ -10,7 +10,6 @@
10	# See the Mulan PSL v2 for more details.	10	# See the Mulan PSL v2 for more details.
11		11
12	import asyncio	12	import asyncio
13	-import contextlib
14	import time	13	import time
15	from typing import AsyncGenerator, Any	14	from typing import AsyncGenerator, Any
16		15
	@@ -28,19 +27,6 @@ from motor.coordinator.tracer.tracing import TracerManager
28		27
29	class SeparateCDPRouter(BaseRouter):	28	class SeparateCDPRouter(BaseRouter):
30		29
31	- @contextlib.asynccontextmanager
32	- async def _manage_request_context(self):
33	- """
34	- Lifecycle management for request in the RequestManager.
35	- Ensures request info is added and cleaned up.
36	- """
37	- await self._request_manager.add_req_info(self.req_info)
38	- try:
39	- yield
40	- finally:
41	- await self._request_manager.del_req_info(self.req_info.req_id)
42	- self._log_request_details()
43	-
44	async def handle_request(self) -> StreamingResponse \| JSONResponse:	30	async def handle_request(self) -> StreamingResponse \| JSONResponse:
45		31
46	req_data = self._gen_d_request()	32	req_data = self._gen_d_request()

Amotor/coordinator/router/separate_pd_dual_dispatch_router.py+204-0

@@ -0,0 +1,204 @@
	1	+# -- coding: utf-8 --
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
	3	+# MindIE is licensed under Mulan PSL v2.
	4	+# You can use this software according to the terms and conditions of the Mulan PSL v2.
	5	+# You may obtain a copy of Mulan PSL v2 at:
	6	+# http://license.coscl.org.cn/MulanPSL2
	7	+# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
	8	+# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
	9	+# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
	10	+# See the Mulan PSL v2 for more details.
	11	+
	12	+import asyncio
	13	+import random
	14	+from contextlib import asynccontextmanager
	15	+from typing import AsyncGenerator
	16	+
	17	+import anyio
	18	+from fastapi.responses import StreamingResponse, JSONResponse
	19	+
	20	+from motor.common.resources.instance import PDRole
	21	+from motor.common.utils.env import Env
	22	+from motor.coordinator.domain import ScheduledResource
	23	+from motor.coordinator.models.request import ReqState
	24	+from motor.coordinator.router.base_router import BaseRouter
	25	+
	26	+
	27	+class SeparatePDDualDispatchRouter(BaseRouter):
	28	+
	29	+ @staticmethod
	30	+ def _generate_bootstrap_room() -> int:
	31	+ """Generate a unique bootstrap room ID for disaggregated serving.
	32	+
	33	+ Returns:
	34	+ Random 63-bit integer.
	35	+ """
	36	+ return random.randint(0, 2**63 - 1)
	37	+
	38	+ @asynccontextmanager
	39	+ async def _dual_env_context(self):
	40	+ async with self._manage_request_context(), \
	41	+ self._manage_resource_context(PDRole.ROLE_P, self.release_all) as p_res, \
	42	+ self._manage_resource_context(PDRole.ROLE_D, self.release_tokens) as d_res, \
	43	+ self._manage_client_context(p_res) as p_client, \
	44	+ self._manage_client_context(d_res) as d_client:
	45	+ yield p_res, d_res, p_client, d_client
	46	+
	47	+ async def handle_request(self) -> StreamingResponse \| JSONResponse:
	48	+ """Entry point for handling dual dispatch requests."""
	49	+ is_stream = self.req_info.req_data.get("stream", False)
	50	+ if is_stream:
	51	+ return StreamingResponse(
	52	+ self._generate_stream_response(),
	53	+ media_type="text/event-stream"
	54	+ )
	55	+ return await self._generate_response()
	56	+
	57	+ async def _run_prefill(self, req_data, p_client, scope):
	58	+ with scope:
	59	+ try:
	60	+ response = await self.forward_request(
	61	+ req_data, p_client, self.config.exception_config.first_token_timeout
	62	+ )
	63	+ self.req_info.update_state(ReqState.PREFILL_END)
	64	+ return response
	65	+ except asyncio.CancelledError:
	66	+ pass
	67	+ except Exception as e:
	68	+ self.logger.error("Prefill error: %s", str(e))
	69	+ self.req_info.cancel_scope()
	70	+
	71	+ async def _generate_stream_response(self) -> AsyncGenerator[str, None]:
	72	+ """
	73	+ Handles streaming requests for Dual Dispatch with retry logic and scope management.
	74	+ """
	75	+ self.logger.debug("Handling streaming Dual Dispatch request")
	76	+ max_retry = self.config.exception_config.max_retry
	77	+
	78	+ for attempt in range(max_retry):
	79	+ try:
	80	+ # 1. Allocate resources & Initialize contexts
	81	+ async with self._dual_env_context() as (p_res, d_res, p_client, d_client):
	82	+ p_scope = anyio.CancelScope()
	83	+ d_scope = anyio.CancelScope()
	84	+ self.req_info.set_cancel_scope(p_scope, PDRole.ROLE_P)
	85	+ self.req_info.set_cancel_scope(d_scope, PDRole.ROLE_D)
	86	+
	87	+ req_data = await self._gen_dual_request(p_res)
	88	+ p_req_data = await self._gen_p_request(req_data)
	89	+
	90	+ # 2. Fire Prefill request in background
	91	+ p_task = asyncio.create_task(self._run_prefill(p_req_data, p_client, p_scope))
	92	+
	93	+ # 3. Fire Decode stream
	94	+ try:
	95	+ with d_scope:
	96	+ async for chunk in self.forward_stream_request(
	97	+ req_data, d_client, self.config.exception_config.first_token_timeout
	98	+ ):
	99	+ yield chunk
	100	+
	101	+ self.req_info.update_state(ReqState.DECODE_END)
	102	+ return
	103	+ finally:
	104	+ # Clean up Prefill task if Decode finishes early or encounters an error
	105	+ p_task.cancel()
	106	+ if self.req_info.is_cancelled:
	107	+ raise Exception("Exception occurred in dual dispatch")
	108	+
	109	+ except asyncio.CancelledError:
	110	+ self.logger.info("The streaming request was terminated because of timeout or client disconnect.")
	111	+ self.req_info.cancel_scope()
	112	+ raise
	113	+ except Exception as e:
	114	+ self.logger.error(
	115	+ "Error in dual dispatch streaming (attempt %d/%d): %s",
	116	+ attempt + 1, max_retry, str(e), exc_info=True
	117	+ )
	118	+ self.req_info.cancel_scope()
	119	+
	120	+ if self.first_chunk_sent or attempt == max_retry - 1:
	121	+ self.req_info.update_state(ReqState.EXCEPTION)
	122	+ yield self._generate_streaming_error_chunk(e)
	123	+ return
	124	+
	125	+ wait_time = self.config.exception_config.retry_delay * (2 ** attempt)
	126	+ self.logger.info("Retrying streaming request in %.2f seconds...", wait_time)
	127	+ await asyncio.sleep(wait_time)
	128	+
	129	+ async def _generate_response(self) -> JSONResponse:
	130	+ """
	131	+ Handles non-streaming requests for Dual Dispatch with retry logic.
	132	+ """
	133	+ self.logger.debug("Handling non-streaming Dual Dispatch request")
	134	+ max_retries = self.config.exception_config.max_retry
	135	+
	136	+ for attempt in range(max_retries):
	137	+ try:
	138	+ # 1. Allocate resources & Initialize contexts
	139	+ async with self._dual_env_context() as (p_res, d_res, p_client, d_client):
	140	+
	141	+ p_scope = anyio.CancelScope()
	142	+ d_scope = anyio.CancelScope()
	143	+ self.req_info.set_cancel_scope(p_scope, PDRole.ROLE_P)
	144	+ self.req_info.set_cancel_scope(d_scope, PDRole.ROLE_D)
	145	+
	146	+ req_data = await self._gen_dual_request(p_res)
	147	+ p_req_data = await self._gen_p_request(req_data)
	148	+
	149	+ # 2. Fire Prefill request in background
	150	+ p_task = asyncio.create_task(self._run_prefill(p_req_data, p_client, p_scope))
	151	+
	152	+ # 3. Fire Decode request
	153	+ try:
	154	+ with d_scope:
	155	+ response = await self.forward_request(
	156	+ req_data, d_client, self.config.exception_config.infer_timeout
	157	+ )
	158	+ self.req_info.update_state(ReqState.DECODE_END)
	159	+ return JSONResponse(content=response.json())
	160	+ finally:
	161	+ p_task.cancel()
	162	+ if self.req_info.is_cancelled:
	163	+ raise Exception("Exception occurred in dual dispatch")
	164	+
	165	+ except asyncio.CancelledError:
	166	+ self.logger.info("The non-streaming request was terminated because of timeout or client disconnect.")
	167	+ self.req_info.cancel_scope()
	168	+ raise
	169	+ except Exception as e:
	170	+ self.logger.error(
	171	+ "Error in dual dispatch decode (attempt %d/%d): %s",
	172	+ attempt + 1, max_retries, str(e)
	173	+ )
	174	+ self.req_info.cancel_scope()
	175	+
	176	+ if attempt < max_retries - 1:
	177	+ wait_time = self.config.exception_config.retry_delay * (2 ** attempt)
	178	+ self.logger.info("Retrying non-streaming request in %.2f seconds...", wait_time)
	179	+ await asyncio.sleep(wait_time)
	180	+ continue
	181	+
	182	+ self.logger.error("All retries failed for non-streaming dual dispatch request.")
	183	+ self.req_info.update_state(ReqState.EXCEPTION)
	184	+ raise e
	185	+
	186	+ async def _gen_dual_request(self, prefill: ScheduledResource) -> dict:
	187	+ """Inject bootstrap info"""
	188	+ req_data = self.req_info.req_data.copy()
	189	+ req_data.update({
	190	+ "bootstrap_host": prefill.endpoint.ip,
	191	+ "bootstrap_port": Env.disaggregation_bootstrap_port,
	192	+ "bootstrap_room": self._generate_bootstrap_room(),
	193	+ })
	194	+ return req_data
	195	+
	196	+ async def _gen_p_request(self, req_data) -> dict:
	197	+ p_req_data = req_data.copy()
	198	+ p_req_data["stream"] = False
	199	+ p_req_data["max_tokens"] = 1
	200	+
	201	+ if "stream_options" in p_req_data:
	202	+ del p_req_data["stream_options"]
	203	+
	204	+ return p_req_data

Mmotor/coordinator/scheduler/runtime/scheduler_client.py+1-1

	@@ -781,7 +781,7 @@ class AsyncSchedulerClient:
781		781
782	def _status(p_list: list, d_list: list, u_list: list) -> InstanceReadiness:	782	def _status(p_list: list, d_list: list, u_list: list) -> InstanceReadiness:
783	if mode in (DeployMode.CDP_SEPARATE, DeployMode.CPCD_SEPARATE, DeployMode.PD_SEPARATE, \	783	if mode in (DeployMode.CDP_SEPARATE, DeployMode.CPCD_SEPARATE, DeployMode.PD_SEPARATE, \
784	- DeployMode.PD_DISAGGREGATION_SINGLE_CONTAINER):	784	+ DeployMode.PD_DISAGGREGATION_SINGLE_CONTAINER, DeployMode.PD_DUAL_DISPATCH):
785	has_p, has_d = len(p_list) > 0, len(d_list) > 0	785	has_p, has_d = len(p_list) > 0, len(d_list) > 0
786	if has_p and has_d:	786	if has_p and has_d:
787	return InstanceReadiness.REQUIRED_MET	787	return InstanceReadiness.REQUIRED_MET

Atests/coordinator/router/test_separate_pd_dual_dispatch_router.py+453-0

@@ -0,0 +1,453 @@
	1	+#!/usr/bin/env python3
	2	+# -- coding: utf-8 --
	3	+# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved.
	4	+# MindIE is licensed under Mulan PSL v2.
	5	+# You can use this software according to the terms and conditions of the Mulan PSL v2.
	6	+# You may obtain a copy of Mulan PSL v2 at:
	7	+# http://license.coscl.org.cn/MulanPSL2
	8	+# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
	9	+# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
	10	+# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
	11	+# See the Mulan PSL v2 for more details.
	12	+
	13	+from pytest import MonkeyPatch
	14	+from fastapi import FastAPI, status, Request
	15	+from unittest.mock import patch, MagicMock, AsyncMock
	16	+from fastapi.testclient import TestClient
	17	+import asyncio
	18	+import httpx
	19	+import pytest
	20	+
	21	+from motor.config.coordinator import DeployMode, CoordinatorConfig, SchedulerType
	22	+from motor.coordinator.domain.instance_manager import InstanceManager
	23	+from motor.coordinator.domain import InstanceReadiness, ScheduledResource
	24	+from motor.coordinator.models.request import ReqState, RequestInfo
	25	+from motor.coordinator.router.base_router import BaseRouter
	26	+from motor.coordinator.router.separate_pd_dual_dispatch_router import SeparatePDDualDispatchRouter
	27	+from motor.coordinator.tracer.tracing import TracerManager
	28	+from motor.common.resources.endpoint import WorkloadAction
	29	+from motor.common.resources.instance import Endpoint, PDRole, Instance, InsStatus, ParallelConfig
	30	+from motor.coordinator.scheduler.scheduler import Scheduler
	31	+from motor.coordinator.domain.request_manager import RequestManager
	32	+from tests.coordinator.router.mock_openai_request import MockStreamResponse, create_mock_request_info
	33	+import motor.coordinator.router.router as router
	34	+
	35	+TracerManager()
	36	+
	37	+app = FastAPI()
	38	+_config = CoordinatorConfig()
	39	+_scheduler = Scheduler(instance_provider=InstanceManager(_config), config=_config)
	40	+_request_manager = RequestManager(_config)
	41	+
	42	+
	43	+@app.post("/v1/chat/completions")
	44	+async def handle_completions(request: Request):
	45	+ return await router.handle_request(
	46	+ request, _config, scheduler=_scheduler, request_manager=_request_manager
	47	+ )
	48	+
	49	+
	50	+class MockAsyncClient:
	51	+
	52	+ def __init__(self, post_exc: Exception = None, stream_exc: Exception = None,
	53	+ post_fail_times: int = 1, stream_fail_times: int = 1):
	54	+ self.post_exc = post_exc
	55	+ self.post_fail_times = post_fail_times
	56	+ self.post_count = 0
	57	+ self.post_fail_count = 0
	58	+
	59	+ self.stream_exc = stream_exc
	60	+ self.stream_fail_times = stream_fail_times
	61	+ self.stream_count = 0
	62	+ self.stream_fail_count = 0
	63	+
	64	+ self.base_url = "test-base-url"
	65	+ self.timeout = 1
	66	+ self.is_closed = True
	67	+
	68	+ async def __aenter__(self):
	69	+ return self
	70	+
	71	+ async def __aexit__(self, exc_type, exc_val, exc_tb):
	72	+ pass
	73	+
	74	+ async def aclose(self):
	75	+ pass
	76	+
	77	+ async def post(self, url, json=None, headers=None, **kwargs):
	78	+ self.post_count += 1
	79	+ if self.post_exc and self.post_fail_count < self.post_fail_times:
	80	+ self.post_fail_count += 1
	81	+ mock_response_fail = MagicMock()
	82	+ mock_response_fail.raise_for_status = MagicMock(side_effect=self.post_exc)
	83	+ return mock_response_fail
	84	+
	85	+ request = httpx.Request("POST", url, headers=headers or {}, json=json)
	86	+
	87	+ return httpx.Response(
	88	+ status_code=status.HTTP_200_OK,
	89	+ json={
	90	+ "choices": [{"delta": {"content": "chunk"}, "index": 0, "finish_reason": None}],
	91	+ "id": "chatcmpl-123"},
	92	+ request=request
	93	+ )
	94	+
	95	+ def stream(self, method, url, json=None, headers=None, **kwargs):
	96	+ self.stream_count += 1
	97	+
	98	+ if self.stream_exc and self.stream_fail_count < self.stream_fail_times:
	99	+ self.stream_fail_count += 1
	100	+ return MockStreamResponse(json or {}, recomputed=False, exc=self.stream_exc)
	101	+
	102	+ # Return an async context manager
	103	+ return MockStreamResponse(json or {}, recomputed=False, exc=None)
	104	+
	105	+
	106	+class TestPDDualDispatchRouter:
	107	+
	108	+ @pytest.fixture
	109	+ def client(self):
	110	+ return TestClient(app)
	111	+
	112	+ @classmethod
	113	+ def create_mock_instance(self, instance_id, role):
	114	+ """Create a proper mock Instance object"""
	115	+ mock_instance = Instance(
	116	+ job_name=f"test-job-{instance_id}",
	117	+ model_name=f"test-model-{instance_id}",
	118	+ id=instance_id,
	119	+ role=role,
	120	+ status=InsStatus.ACTIVE,
	121	+ parallel_config=ParallelConfig(dp_size=1, tp_size=1),
	122	+ endpoints={}
	123	+ )
	124	+ return mock_instance
	125	+
	126	+ @pytest.fixture
	127	+ def setup_dp_separation(self, monkeypatch: MonkeyPatch):
	128	+ host = "127.0.0.1"
	129	+ # Create proper instances for separate P/D flow
	130	+ mock_instance_p = self.create_mock_instance(0, PDRole.ROLE_P)
	131	+ mock_endpoint_p = Endpoint(id=0, ip=host, business_port="8000", mgmt_port="8000")
	132	+ mock_instance_p.endpoints = {host: {0: mock_endpoint_p}}
	133	+
	134	+ mock_instance_d = self.create_mock_instance(1, PDRole.ROLE_D)
	135	+ mock_endpoint_d = Endpoint(id=1, ip=host, business_port="8001", mgmt_port="8001")
	136	+ mock_instance_d.endpoints = {host: {1: mock_endpoint_d}}
	137	+
	138	+ # Mock functions (Scheduler uses get_required_instances_status for readiness)
	139	+ def mock_get_required_instances_status(self, deploy_mode=None):
	140	+ return InstanceReadiness.REQUIRED_MET
	141	+
	142	+ def mock_has_required_instances(self, deploy_mode=None):
	143	+ return True
	144	+
	145	+ def mock_get_available_instances(args, *kwargs):
	146	+ # Accept (self, role) when patched on InstanceManager; role is 2nd positional or in kwargs
	147	+ role = kwargs.get("role")
	148	+ if role is None and len(args) >= 2:
	149	+ role = args[1]
	150	+ elif role is None and len(args) == 1:
	151	+ role = args[0] # staticmethod-style call
	152	+ if role == PDRole.ROLE_U: # PD hybrid role
	153	+ return {} # No PD hybrid instances, will use separate P/D
	154	+ if role == PDRole.ROLE_P:
	155	+ return {mock_instance_p.id: mock_instance_p}
	156	+ if role == PDRole.ROLE_D:
	157	+ return {mock_instance_d.id: mock_instance_d}
	158	+ return {}
	159	+
	160	+ async def mock_select_instance_and_endpoint(self, role):
	161	+ if role == PDRole.ROLE_P:
	162	+ return mock_instance_p, mock_endpoint_p
	163	+ elif role == PDRole.ROLE_D:
	164	+ return mock_instance_d, mock_endpoint_d
	165	+ return None, None
	166	+
	167	+ async def mock_update_workload(self, params):
	168	+ return True
	169	+
	170	+ monkeypatch.setattr(InstanceManager, "get_required_instances_status", mock_get_required_instances_status)
	171	+ monkeypatch.setattr(InstanceManager, "has_required_instances", mock_has_required_instances)
	172	+ monkeypatch.setattr(InstanceManager, "get_available_instances", mock_get_available_instances)
	173	+ monkeypatch.setattr(Scheduler, "select_instance_and_endpoint", mock_select_instance_and_endpoint)
	174	+ monkeypatch.setattr(Scheduler, "update_workload", mock_update_workload)
	175	+
	176	+ # Mock CoordinatorConfig to return PD_DUAL_DISPATCH deploy mode
	177	+ mock_scheduler_config = MagicMock()
	178	+ mock_scheduler_config.deploy_mode = DeployMode.PD_DUAL_DISPATCH
	179	+ mock_scheduler_config.scheduler_type = SchedulerType.LOAD_BALANCE
	180	+ mock_exception_config = MagicMock()
	181	+ mock_exception_config.retry_delay = 0.0001
	182	+ mock_exception_config.max_retry = 5
	183	+ mock_http_config = MagicMock()
	184	+ mock_http_config.coordinator_api_host = "127.0.0.1"
	185	+ mock_http_config.coordinator_api_mgmt_port = 1025
	186	+ mock_tls_config = MagicMock()
	187	+ mock_tls_config.enable_tls = False
	188	+
	189	+ mock_config = MagicMock()
	190	+ mock_config.scheduler_config = mock_scheduler_config
	191	+ mock_config.exception_config = mock_exception_config
	192	+ mock_config.http_config = mock_http_config
	193	+ mock_config.infer_tls_config = mock_tls_config
	194	+ mock_config.mgmt_tls_config = mock_tls_config
	195	+ # So _gen_d_request uses coordinator_api_mgmt_port; avoid MagicMock as parsed_url.port
	196	+ mock_config.worker_metaserver_port = None
	197	+
	198	+ monkeypatch.setattr(CoordinatorConfig, "__new__", lambda cls: mock_config)
	199	+
	200	+ @pytest.fixture
	201	+ def mock_raw_request(self):
	202	+ # Mock Request
	203	+ mock_req = MagicMock(spec=Request)
	204	+ mock_req.body = AsyncMock(return_value=b'{"model": "test"}')
	205	+ mock_req.json = AsyncMock(return_value={"model": "test"})
	206	+ mock_req.headers = {}
	207	+ mock_req.url.path = "/v1/chat/completions"
	208	+ # Must be awaitable so listen_for_disconnect() does not raise; never completes so handler wins.
	209	+ never = asyncio.Future()
	210	+ mock_req.receive = AsyncMock(return_value=never)
	211	+ return mock_req
	212	+
	213	+ @pytest.mark.asyncio
	214	+ async def test_handle_request_stream_successful(self, client, monkeypatch: MonkeyPatch, setup_dp_separation):
	215	+ """Test case: PD_DUAL_DISPATCH mode stream request success
	216	+ Expected behavior:
	217	+ 1) Check request status is DecodeEnd
	218	+ 2) Return normal response
	219	+ """
	220	+
	221	+ mock_async_client = MockAsyncClient()
	222	+ req_info = await create_mock_request_info()
	223	+
	224	+ with patch('motor.coordinator.router.base_router.httpx.AsyncClient', return_value=mock_async_client):
	225	+ cdp_router = SeparatePDDualDispatchRouter(
	226	+ req_info, CoordinatorConfig(),
	227	+ scheduler=Scheduler(instance_provider=InstanceManager(CoordinatorConfig()), config=CoordinatorConfig()),
	228	+ request_manager=_request_manager
	229	+ )
	230	+ response = await cdp_router.handle_request()
	231	+ chunks = []
	232	+ async for chunk in response.body_iterator:
	233	+ chunks.append(chunk)
	234	+
	235	+ # Should get a 200 success status
	236	+ assert response.status_code == status.HTTP_200_OK
	237	+ # Should be a streaming response
	238	+ assert "text/event-stream" in response.headers.get("content-type")
	239	+
	240	+ # Check request state and metrics
	241	+ assert req_info.state == ReqState.DECODE_END
	242	+ assert req_info.status[ReqState.D_ALLOCATED] >= req_info.status[ReqState.ARRIVE]
	243	+ assert req_info.status[ReqState.P_ALLOCATED] >= req_info.status[ReqState.ARRIVE]
	244	+ assert req_info.status[ReqState.PREFILL_END] >= req_info.status[ReqState.P_ALLOCATED]
	245	+ assert req_info.status[ReqState.DECODE_END] >= req_info.status[ReqState.FIRST_TOKEN_FINISH]
	246	+
	247	+ @pytest.mark.asyncio
	248	+ async def test_handle_request_non_stream_successful(self, client, monkeypatch: MonkeyPatch, setup_dp_separation):
	249	+ """Test case: PD_DUAL_DISPATCH mode non_stream request success
	250	+ Expected behavior:
	251	+ 1) Check request status is DecodeEnd
	252	+ 2) Return normal response
	253	+ """
	254	+
	255	+ mock_async_client = MockAsyncClient()
	256	+ req_info = await create_mock_request_info(stream=False)
	257	+
	258	+ with patch('motor.coordinator.router.base_router.httpx.AsyncClient', return_value=mock_async_client):
	259	+ cdp_router = SeparatePDDualDispatchRouter(
	260	+ req_info, CoordinatorConfig(),
	261	+ scheduler=Scheduler(instance_provider=InstanceManager(CoordinatorConfig()), config=CoordinatorConfig()),
	262	+ request_manager=_request_manager
	263	+ )
	264	+ response = await cdp_router.handle_request()
	265	+
	266	+ # Should get a 200 success status
	267	+ assert response.status_code == status.HTTP_200_OK
	268	+ # Should be a streaming response
	269	+ assert "application/json" in response.headers.get("content-type")
	270	+
	271	+ # Check request state and metrics
	272	+ assert req_info.state == ReqState.DECODE_END
	273	+ assert req_info.status[ReqState.D_ALLOCATED] >= req_info.status[ReqState.ARRIVE]
	274	+ assert req_info.status[ReqState.P_ALLOCATED] >= req_info.status[ReqState.ARRIVE]
	275	+ assert req_info.status[ReqState.DECODE_END] >= req_info.status[ReqState.D_ALLOCATED]
	276	+
	277	+ @pytest.mark.asyncio
	278	+ async def test_handle_request_error_when_decode_4xx(self, client, monkeypatch: MonkeyPatch, setup_dp_separation):
	279	+ """Test case: Decode EngineServer returns 4XX status code
	280	+ Expected behavior:
	281	+ 1) No request retry triggered
	282	+ 2) Directly return error message
	283	+ """
	284	+ # Mock the HTTP forwarding function to return a 4XX error
	285	+ error_message = "Test Bad Request"
	286	+ mock_async_client = MockAsyncClient(stream_exc=httpx.HTTPStatusError(
	287	+ message=error_message,
	288	+ request=None,
	289	+ response=httpx.Response(status_code=status.HTTP_400_BAD_REQUEST, text=error_message)
	290	+ ), stream_fail_times=CoordinatorConfig().exception_config.max_retry)
	291	+ req_info = await create_mock_request_info()
	292	+
	293	+ release_p_tokens = 0
	294	+ release_p_kv = 0
	295	+ release_d_tokens = 0
	296	+
	297	+ async def mock_update_workload(self, resource: ScheduledResource, action: WorkloadAction):
	298	+ nonlocal release_p_tokens
	299	+ nonlocal release_p_kv
	300	+ nonlocal release_d_tokens
	301	+ if resource.instance.role == PDRole.ROLE_P:
	302	+ if action == WorkloadAction.RELEASE_TOKENS:
	303	+ release_p_tokens += 1
	304	+ elif action == WorkloadAction.RELEASE_KV:
	305	+ release_p_kv += 1
	306	+ elif resource.instance.role == PDRole.ROLE_D:
	307	+ if action == WorkloadAction.RELEASE_TOKENS:
	308	+ release_d_tokens += 1
	309	+ return True
	310	+
	311	+ monkeypatch.setattr(BaseRouter, "_update_workload", mock_update_workload)
	312	+
	313	+ with patch('motor.coordinator.router.base_router.httpx.AsyncClient', return_value=mock_async_client):
	314	+
	315	+ cdp_router = SeparatePDDualDispatchRouter(
	316	+ req_info, CoordinatorConfig(),
	317	+ scheduler=Scheduler(instance_provider=InstanceManager(CoordinatorConfig()), config=CoordinatorConfig()),
	318	+ request_manager=_request_manager
	319	+ )
	320	+ response = await cdp_router.handle_request()
	321	+ chunks = []
	322	+ async for chunk in response.body_iterator:
	323	+ chunks.append(chunk)
	324	+ chunk_str = "".join(chunks)
	325	+
	326	+ assert req_info.state == ReqState.EXCEPTION
	327	+ assert error_message in chunk_str
	328	+ # Should get a 4XX error
	329	+ assert str(status.HTTP_400_BAD_REQUEST) in chunk_str
	330	+ assert mock_async_client.stream_count == CoordinatorConfig().exception_config.max_retry
	331	+ assert release_d_tokens >= 1
	332	+ assert release_p_tokens >= 1
	333	+
	334	+ @pytest.mark.asyncio
	335	+ async def test_handle_request_error_when_decode_5xx(self, client, monkeypatch: MonkeyPatch, setup_dp_separation):
	336	+ """Test scenario: EngineServer Decode request continuously returns 5XX status code
	337	+ Expected behavior:
	338	+ 1) Check request status is Exception
	339	+ 2) Trigger request retry
	340	+ 3) Request retry fails: return error message
	341	+ """
	342	+ # Mock the HTTP forwarding function to return a 4XX error
	343	+ error_message = "Test Internal Server Error"
	344	+ mock_async_client = MockAsyncClient(stream_exc=httpx.HTTPStatusError(
	345	+ message=error_message,
	346	+ request=None,
	347	+ response=httpx.Response(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, text=error_message)
	348	+ ), stream_fail_times=CoordinatorConfig().exception_config.max_retry)
	349	+ req_info = await create_mock_request_info()
	350	+
	351	+ exec_release = 0
	352	+
	353	+ async def mock_update_workload(self, resource: ScheduledResource, action: WorkloadAction):
	354	+ nonlocal exec_release
	355	+ exec_release += 1
	356	+ return True
	357	+
	358	+ monkeypatch.setattr(BaseRouter, "_update_workload", mock_update_workload)
	359	+
	360	+ with patch('motor.coordinator.router.base_router.httpx.AsyncClient', return_value=mock_async_client):
	361	+ cdp_router = SeparatePDDualDispatchRouter(
	362	+ req_info, CoordinatorConfig(),
	363	+ scheduler=Scheduler(instance_provider=InstanceManager(CoordinatorConfig()), config=CoordinatorConfig()),
	364	+ request_manager=_request_manager
	365	+ )
	366	+ response = await cdp_router.handle_request()
	367	+ chunks = []
	368	+ async for chunk in response.body_iterator:
	369	+ chunks.append(chunk)
	370	+ chunk_str = "".join(chunks)
	371	+
	372	+ assert req_info.state == ReqState.EXCEPTION
	373	+ assert error_message in chunk_str
	374	+ # Should get a 500 error after max retries
	375	+ assert str(status.HTTP_500_INTERNAL_SERVER_ERROR) in chunk_str
	376	+ # Should retry exactly max_retry times
	377	+ assert mock_async_client.stream_count == CoordinatorConfig().exception_config.max_retry
	378	+ assert exec_release >= 1
	379	+
	380	+ @pytest.mark.asyncio
	381	+ async def test_handle_request_error_when_decode_once_5xx(
	382	+ self, client, monkeypatch: MonkeyPatch, setup_dp_separation
	383	+ ):
	384	+ """Test case: EngineServer Decode request first returns 5XX, then 200.
	385	+ Expected behavior:
	386	+ 1) Check request status is Exception
	387	+ 2) Trigger request retry
	388	+ 3) Request retry succeeds
	389	+ """
	390	+ # Mock the HTTP stream forwarding function to return a 5XX error once
	391	+ error_message = "Test Internal Server Error"
	392	+ mock_async_client = MockAsyncClient(stream_exc=httpx.HTTPStatusError(
	393	+ message=error_message,
	394	+ request=None,
	395	+ response=httpx.Response(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
	396	+ ), stream_fail_times=1)
	397	+ req_info = await create_mock_request_info()
	398	+
	399	+ with patch('motor.coordinator.router.base_router.httpx.AsyncClient', return_value=mock_async_client):
	400	+ cdp_router = SeparatePDDualDispatchRouter(
	401	+ req_info, CoordinatorConfig(),
	402	+ scheduler=Scheduler(instance_provider=InstanceManager(CoordinatorConfig()), config=CoordinatorConfig()),
	403	+ request_manager=_request_manager
	404	+ )
	405	+ response = await cdp_router.handle_request()
	406	+ chunks = []
	407	+ async for chunk in response.body_iterator:
	408	+ chunks.append(chunk)
	409	+
	410	+ # Should get a 200 after retry
	411	+ assert response.status_code == status.HTTP_200_OK
	412	+ # Decode: at least one fail then success; stream_count may be 2 or up to max_retry
	413	+ assert mock_async_client.stream_fail_count == 1
	414	+ assert mock_async_client.stream_count >= 2
	415	+ # Decode path may use stream only; post is used for metaserver/other branches
	416	+ assert mock_async_client.post_count >= 0
	417	+ assert req_info.state == ReqState.DECODE_END
	418	+
	419	+ @pytest.mark.asyncio
	420	+ async def test_handle_request_error_when_decode_network_exception(
	421	+ self, client, monkeypatch: MonkeyPatch, setup_dp_separation
	422	+ ):
	423	+ """Test case: EngineServer Decode network exception
	424	+ Expected behavior:
	425	+ 1) Check request status is Exception
	426	+ 2) No request retry triggered
	427	+ 3) Directly return error message
	428	+ """
	429	+ # Mock the HTTP forwarding function to always raise a network exception
	430	+ error_message = "Connection error"
	431	+ # mock AsyncClient in router
	432	+ mock_async_client = MockAsyncClient(stream_exc=httpx.ConnectError(
	433	+ error_message,
	434	+ request=MagicMock()
	435	+ ), stream_fail_times=CoordinatorConfig().exception_config.max_retry)
	436	+
	437	+ req_info = await create_mock_request_info()
	438	+
	439	+ with patch('motor.coordinator.router.base_router.httpx.AsyncClient', return_value=mock_async_client):
	440	+ cdp_router = SeparatePDDualDispatchRouter(
	441	+ req_info, CoordinatorConfig(),
	442	+ scheduler=Scheduler(instance_provider=InstanceManager(CoordinatorConfig()), config=CoordinatorConfig()),
	443	+ request_manager=_request_manager
	444	+ )
	445	+ response = await cdp_router.handle_request()
	446	+ chunks = []
	447	+ async for chunk in response.body_iterator:
	448	+ chunks.append(chunk)
	449	+ chunk_str = "".join(chunks)
	450	+ assert error_message in chunk_str
	451	+ assert mock_async_client.stream_count == CoordinatorConfig().exception_config.max_retry
	452	+ assert mock_async_client.stream_fail_count == CoordinatorConfig().exception_config.max_retry
	453	+ assert req_info.state == ReqState.EXCEPTION