[feature]支持EPD分离部署能力-MindIE-Motor-AtomGit

M.gitignore+1-1

	@@ -109,4 +109,4 @@ venv.bak/
109	*/_pb2_grpc.py	109	*/_pb2_grpc.py
110		110
111	# deployer	111	# deployer
112	-deployer/output/*	112	+examples/deployer/output_yamls/*

Mexamples/deployer/delete.sh+3-0

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

	@@ -12,11 +12,14 @@ import os
12	GREEN = '\033[32m'	12	GREEN = '\033[32m'
13	RESET = '\033[0m'	13	RESET = '\033[0m'
14		14
		15	+E_INSTANCES_NUM = "e_instances_num"
15	P_INSTANCES_NUM = "p_instances_num"	16	P_INSTANCES_NUM = "p_instances_num"
16	D_INSTANCES_NUM = "d_instances_num"	17	D_INSTANCES_NUM = "d_instances_num"
17	CONFIG_JOB_ID = "job_id"	18	CONFIG_JOB_ID = "job_id"
		19	+SINGER_E_INSTANCES_NUM = "single_e_instance_pod_num"
18	SINGER_P_INSTANCES_NUM = "single_p_instance_pod_num"	20	SINGER_P_INSTANCES_NUM = "single_p_instance_pod_num"
19	SINGER_D_INSTANCES_NUM = "single_d_instance_pod_num"	21	SINGER_D_INSTANCES_NUM = "single_d_instance_pod_num"
		22	+E_POD_NPU_NUM = "e_pod_npu_num"
20	P_POD_NPU_NUM = "p_pod_npu_num"	23	P_POD_NPU_NUM = "p_pod_npu_num"
21	D_POD_NPU_NUM = "d_pod_npu_num"	24	D_POD_NPU_NUM = "d_pod_npu_num"
22	ASCEND_910_NPU_NUM = "huawei.com/Ascend910"	25	ASCEND_910_NPU_NUM = "huawei.com/Ascend910"
	@@ -107,10 +110,12 @@ ACCELERATOR_TYPE_A3 = "module-a3-16"
107	CONTAINERS = "containers"	110	CONTAINERS = "containers"
108	IMAGE = "image"	111	IMAGE = "image"
109	IMAGE_NAME = "image_name"	112	IMAGE_NAME = "image_name"
		113	+ROLE_ENCODE = "encode"
110	ROLE_PREFILL = "prefill"	114	ROLE_PREFILL = "prefill"
111	ROLE_DECODE = "decode"	115	ROLE_DECODE = "decode"
112	ROLE_KV_POOL = "kv-pool"	116	ROLE_KV_POOL = "kv-pool"
113	ROLE_KV_CONDUCTOR = "kv-conductor"	117	ROLE_KV_CONDUCTOR = "kv-conductor"
		118	+NODE_TYPE_E = "e"
114	NODE_TYPE_P = "p"	119	NODE_TYPE_P = "p"
115	NODE_TYPE_D = "d"	120	NODE_TYPE_D = "d"
116	ROLE_SINGLE_CONTAINER = "SINGLE_CONTAINER"	121	ROLE_SINGLE_CONTAINER = "SINGLE_CONTAINER"

Mexamples/deployer/lib/generator/engine.py+20-3

	@@ -13,7 +13,7 @@ import os
13	import lib.constant as C	13	import lib.constant as C
14	from lib.utils import (	14	from lib.utils import (
15	generate_unique_id, load_yaml, write_yaml, safe_exec_cmd, logger,	15	generate_unique_id, load_yaml, write_yaml, safe_exec_cmd, logger,
16	- modify_log_mount, obtain_engine_instance_total	16	+ modify_log_mount, obtain_engine_instance_total, obtain_engine_e_instance_total
17	)	17	)
18	from lib.generator import k8s_utils	18	from lib.generator import k8s_utils
19	from lib.generator.k8s_utils import set_engine_base_name, modify_sp_block_num	19	from lib.generator.k8s_utils import set_engine_base_name, modify_sp_block_num
	@@ -67,7 +67,12 @@ def set_engine_metadata(deployment_data, deploy_config, index, node_type, job_na
67		67
68		68
69	def set_engine_env(container, deploy_config, node_type, job_name):	69	def set_engine_env(container, deploy_config, node_type, job_name):
70	- role = C.ROLE_PREFILL if node_type == C.NODE_TYPE_P else C.ROLE_DECODE	70	+ role_map = {
		71	+ C.NODE_TYPE_E: C.ROLE_ENCODE,
		72	+ C.NODE_TYPE_P: C.ROLE_PREFILL,
		73	+ C.NODE_TYPE_D: C.ROLE_DECODE
		74	+ }
		75	+ role = role_map.get(node_type)
71	if C.ENV not in container:	76	if C.ENV not in container:
72	container[C.ENV] = []	77	container[C.ENV] = []
73	container[C.ENV].extend(build_engine_env_items(role, deploy_config, job_name, include_kv_pool=True))	78	container[C.ENV].extend(build_engine_env_items(role, deploy_config, job_name, include_kv_pool=True))
	@@ -87,7 +92,9 @@ def set_container_npu(container, npu_num):
87		92
88		93
89	def set_engine_npu(container, deploy_config, node_type):	94	def set_engine_npu(container, deploy_config, node_type):
90	- if node_type == C.NODE_TYPE_P and C.P_POD_NPU_NUM in deploy_config:	95	+ if node_type == C.NODE_TYPE_E and C.E_POD_NPU_NUM in deploy_config:
		96	+ npu_num = int(deploy_config[C.E_POD_NPU_NUM])
		97	+ elif node_type == C.NODE_TYPE_P and C.P_POD_NPU_NUM in deploy_config:
91	npu_num = int(deploy_config[C.P_POD_NPU_NUM])	98	npu_num = int(deploy_config[C.P_POD_NPU_NUM])
92	elif node_type == C.NODE_TYPE_D and C.D_POD_NPU_NUM in deploy_config:	99	elif node_type == C.NODE_TYPE_D and C.D_POD_NPU_NUM in deploy_config:
93	npu_num = int(deploy_config[C.D_POD_NPU_NUM])	100	npu_num = int(deploy_config[C.D_POD_NPU_NUM])
	@@ -180,6 +187,16 @@ def validate_instance_nums(user_config):
180	def generate_yaml_engine(input_yaml, output_file, user_config):	187	def generate_yaml_engine(input_yaml, output_file, user_config):
181	logger.info(f"Generating YAML from {input_yaml} to {output_file}")	188	logger.info(f"Generating YAML from {input_yaml} to {output_file}")
182	deploy_config = user_config[C.MOTOR_DEPLOY_CONFIG]	189	deploy_config = user_config[C.MOTOR_DEPLOY_CONFIG]
		190	+ # generate yaml engine E
		191	+ e_total = obtain_engine_e_instance_total(deploy_config)
		192	+ for e_index in range(e_total):
		193	+ data = load_yaml(input_yaml, True)
		194	+ modify_engine_yaml(data, user_config, e_index, C.NODE_TYPE_E)
		195	+ output_file_e = output_file + f"_{C.NODE_TYPE_E}{e_index}.yaml"
		196	+ write_yaml(data, output_file_e, True)
		197	+ k8s_utils.g_generate_yaml_list.append(output_file_e)
		198	+
		199	+ # generate yaml engine P/D
183	p_total, d_total = obtain_engine_instance_total(deploy_config)	200	p_total, d_total = obtain_engine_instance_total(deploy_config)
184	for p_index in range(p_total):	201	for p_index in range(p_total):
185	data = load_yaml(input_yaml, True)	202	data = load_yaml(input_yaml, True)

Mexamples/deployer/lib/generator/k8s_utils.py+3-1

	@@ -241,7 +241,9 @@ def modify_sp_block_num(data, pd_flag, config):
241	if C.ANNOTATIONS in data[C.METADATA]:	241	if C.ANNOTATIONS in data[C.METADATA]:
242	del data[C.METADATA][C.ANNOTATIONS]	242	del data[C.METADATA][C.ANNOTATIONS]
243	return	243	return
244	- if pd_flag == C.NODE_TYPE_D:	244	+ if pd_flag == C.NODE_TYPE_E:
		245	+ sp_block_num = int(config[C.SINGER_E_INSTANCES_NUM]) * int(config[C.E_POD_NPU_NUM])
		246	+ elif pd_flag == C.NODE_TYPE_D:
245	sp_block_num = int(config[C.SINGER_D_INSTANCES_NUM]) * int(config[C.D_POD_NPU_NUM])	247	sp_block_num = int(config[C.SINGER_D_INSTANCES_NUM]) * int(config[C.D_POD_NPU_NUM])
246	elif pd_flag == C.NODE_TYPE_P:	248	elif pd_flag == C.NODE_TYPE_P:
247	sp_block_num = int(config[C.SINGER_P_INSTANCES_NUM]) * int(config[C.P_POD_NPU_NUM])	249	sp_block_num = int(config[C.SINGER_P_INSTANCES_NUM]) * int(config[C.P_POD_NPU_NUM])

Mexamples/deployer/lib/utils.py+12-0

	@@ -152,6 +152,16 @@ def obtain_engine_instance_total(deploy_config):
152	return p_instances, d_instances	152	return p_instances, d_instances
153		153
154		154
		155	+def obtain_engine_e_instance_total(deploy_config):
		156	+ if C.E_INSTANCES_NUM not in deploy_config:
		157	+ return 0
		158	+ try:
		159	+ e_instances = int(deploy_config[C.E_INSTANCES_NUM])
		160	+ except (TypeError, ValueError) as e:
		161	+ raise ValueError(f"{C.E_INSTANCES_NUM} must be integers") from e
		162	+ return e_instances
		163	+
		164	+
155	def modify_log_mount(deployment_data, user_config, app_type):	165	def modify_log_mount(deployment_data, user_config, app_type):
156	host_log_dir = "/root/ascend/log"	166	host_log_dir = "/root/ascend/log"
157	temp_app_config = None	167	temp_app_config = None
	@@ -206,6 +216,7 @@ def set_env_to_shell(user_config, env_config_path, deploy_mode):
206	if deploy_mode == C.DEPLOY_MODE_SINGLE_CONTAINER:	216	if deploy_mode == C.DEPLOY_MODE_SINGLE_CONTAINER:
207	update_shell_safely(C.SINGLE_CONTAINER_SHELL_PATH, env_config, "motor_controller_env", "set_controller_env")	217	update_shell_safely(C.SINGLE_CONTAINER_SHELL_PATH, env_config, "motor_controller_env", "set_controller_env")
208	update_shell_safely(C.SINGLE_CONTAINER_SHELL_PATH, env_config, "motor_coordinator_env", "set_coordinator_env")	218	update_shell_safely(C.SINGLE_CONTAINER_SHELL_PATH, env_config, "motor_coordinator_env", "set_coordinator_env")
		219	+ update_shell_safely(C.SINGLE_CONTAINER_SHELL_PATH, env_config, "motor_engine_encode_env", "set_encode_env")
209	update_shell_safely(C.SINGLE_CONTAINER_SHELL_PATH, env_config, "motor_engine_prefill_env", "set_prefill_env")	220	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_decode_env", "set_decode_env")	221	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_kv_cache_pool_env", "set_kv_pool_env")	222	update_shell_safely(C.SINGLE_CONTAINER_SHELL_PATH, env_config, "motor_kv_cache_pool_env", "set_kv_pool_env")
	@@ -216,6 +227,7 @@ def set_env_to_shell(user_config, env_config_path, deploy_mode):
216	else:	227	else:
217	update_shell_safely(C.CONTROLLER_SHELL_PATH, env_config, "motor_controller_env", "set_controller_env")	228	update_shell_safely(C.CONTROLLER_SHELL_PATH, env_config, "motor_controller_env", "set_controller_env")
218	update_shell_safely(C.COORDINATOR_SHELL_PATH, env_config, "motor_coordinator_env", "set_coordinator_env")	229	update_shell_safely(C.COORDINATOR_SHELL_PATH, env_config, "motor_coordinator_env", "set_coordinator_env")
		230	+ update_shell_safely(C.ENGINE_SHELL_PATH, env_config, "motor_engine_encode_env", "set_encode_env")
219	update_shell_safely(C.ENGINE_SHELL_PATH, env_config, "motor_engine_prefill_env", "set_prefill_env")	231	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")	232	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")	233	update_shell_safely(C.KV_POOL_SHELL_PATH, env_config, "motor_kv_cache_pool_env", "set_kv_pool_env")

Mexamples/deployer/startup/boot.sh+1-1

	@@ -16,7 +16,7 @@ case "$ROLE" in
16	"SINGLE_CONTAINER")	16	"SINGLE_CONTAINER")
17	source "$SCRIPT_DIR/all_combine_in_single_container.sh"	17	source "$SCRIPT_DIR/all_combine_in_single_container.sh"
18	;;	18	;;
19	- "prefill"\|"decode")	19	+ "encode"\|"prefill"\|"decode")
20	source "$SCRIPT_DIR/engine.sh"	20	source "$SCRIPT_DIR/engine.sh"
21	;;	21	;;
22	"controller")	22	"controller")

Mexamples/deployer/startup/roles/engine.sh+8-3

	@@ -9,7 +9,7 @@
9	# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.	9	# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
10	# See the Mulan PSL v2 for more details.	10	# See the Mulan PSL v2 for more details.
11		11
12	-if [ "$ROLE" != "prefill" ] && [ "$ROLE" != "decode" ]; then	12	+if [ "$ROLE" != "encode" ] && [ "$ROLE" != "prefill" ] && [ "$ROLE" != "decode" ]; then
13	echo "Error: This script is for prefill or decode role only. Current ROLE=$ROLE"	13	echo "Error: This script is for prefill or decode role only. Current ROLE=$ROLE"
14	exit 1	14	exit 1
15	fi	15	fi
	@@ -26,7 +26,9 @@ set_mf_store_env
26	# 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
27	# 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}
28	if [ -n "$INFER_SERVICE_INDEX" ] && [ -n "$INSTANCE_INDEX" ]; then	28	if [ -n "$INFER_SERVICE_INDEX" ] && [ -n "$INSTANCE_INDEX" ]; then
29	- if [ "$ROLE" = "prefill" ]; then	29	+ if [ "$ROLE" = "encode" ]; then
		30	+ export JOB_NAME="${JOB_NAME}-${INFER_SERVICE_INDEX}-e${INSTANCE_INDEX}"
		31	+ elif [ "$ROLE" = "prefill" ]; then
30	export JOB_NAME="${JOB_NAME}-${INFER_SERVICE_INDEX}-p${INSTANCE_INDEX}"	32	export JOB_NAME="${JOB_NAME}-${INFER_SERVICE_INDEX}-p${INSTANCE_INDEX}"
31	elif [ "$ROLE" = "decode" ]; then	33	elif [ "$ROLE" = "decode" ]; then
32	export JOB_NAME="${JOB_NAME}-${INFER_SERVICE_INDEX}-d${INSTANCE_INDEX}"	34	export JOB_NAME="${JOB_NAME}-${INFER_SERVICE_INDEX}-d${INSTANCE_INDEX}"
	@@ -38,7 +40,9 @@ setup_motor_log_path
38	setup_ascend_work_path	40	setup_ascend_work_path
39	setup_ascend_cache_path	41	setup_ascend_cache_path
40		42
41	-if [ "$ROLE" = "decode" ]; then	43	+if [ "$ROLE" = "encode" ]; then
		44	+ set_encode_env
		45	+elif [ "$ROLE" = "decode" ]; then
42	set_decode_env	46	set_decode_env
43	elif [ "$ROLE" = "prefill" ]; then	47	elif [ "$ROLE" = "prefill" ]; then
44	set_prefill_env	48	set_prefill_env
	@@ -49,6 +53,7 @@ pid=$!
49	echo "pull up $ROLE instance"	53	echo "pull up $ROLE instance"
50	wait $pid	54	wait $pid
51	exit_code=$?	55	exit_code=$?
		56	+sleep 60000
		zhoujing5月26日调试用的，后续PR已删除
52	if [ $exit_code -ne 0 ]; then	57	if [ $exit_code -ne 0 ]; then
53	echo "Error: mindie daemon exited with code $exit_code"	58	echo "Error: mindie daemon exited with code $exit_code"
54	exit 1	59	exit 1

Mexamples/deployer/startup/set_env_docker.py+2-0

Mmotor/common/resources/endpoint.py+2-0

	@@ -51,6 +51,8 @@ class Workload(BaseModel):
51	return self.active_tokens + self.active_kv_cache * 0.3	51	return self.active_tokens + self.active_kv_cache * 0.3
52	elif role_value == "decode":	52	elif role_value == "decode":
53	return self.active_tokens	53	return self.active_tokens
		54	+ elif role_value == "encode":
		55	+ return self.active_tokens
54	elif role_value == "both":	56	elif role_value == "both":
55	return self.active_tokens + self.active_kv_cache * 0.15	57	return self.active_tokens + self.active_kv_cache * 0.15
56	else:	58	else:

Mmotor/common/resources/instance.py+1-0

	@@ -36,6 +36,7 @@ class InsStatus(str, Enum):
36		36
37		37
38	class PDRole(str, Enum):	38	class PDRole(str, Enum):
		39	+ ROLE_E = "encode"
39	ROLE_P = "prefill"	40	ROLE_P = "prefill"
40	ROLE_D = "decode"	41	ROLE_D = "decode"
41	ROLE_U = "both"	42	ROLE_U = "both"

Amotor/common/utils/image_utils.py+192-0

@@ -0,0 +1,192 @@
	1	+# -- coding: utf-8 --
	2	+# Copyright (c) Huawei Technologies Co., Ltd. 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 base64
	13	+import io
	14	+import math
	15	+from PIL import Image
	16	+
	17	+
	18	+# Set of JPEG Start Of Frame (SOF) markers.
	19	+# These markers indicate the beginning of the frame header,
	20	+# which contains image dimensions (width and height).
	21	+SOF_MARKERS = {
	22	+ 0xC0, # SOF0 (Baseline DCT)
	23	+ 0xC1, # SOF1 (Extended Sequential DCT)
	24	+ 0xC2, # SOF2 (Progressive DCT)
	25	+ 0xC3, # SOF3 (Lossless Sequential)
	26	+ 0xC5, # SOF5 (Differential Sequential)
	27	+ 0xC6, # SOF6 (Differential Progressive)
	28	+ 0xC7, # SOF7 (Differential Lossless)
	29	+ 0xC9, # SOF9 (Extended Sequential DCT, Arithmetic Coding)
	30	+ 0xCA, # SOF10 (Progressive DCT, Arithmetic Coding)
	31	+ 0xCB, # SOF11 (Lossless Sequential, Arithmetic Coding)
	32	+ 0xCD, # SOF13 (Differential Sequential, Arithmetic Coding)
	33	+ 0xCE, # SOF14 (Differential Progressive, Arithmetic Coding)
	34	+ 0xCF, # SOF15 (Differential Lossless, Arithmetic Coding)
	35	+}
	36	+
	37	+
	38	+def parse_jpeg_size(data: bytes):
	39	+ """
	40	+ Parses JPEG binary data to extract width and height by reading the SOF marker.
	41	+ This is faster than decoding the entire image as it only reads the header.
	42	+
	43	+ Args:
	44	+ data: Raw JPEG byte data.
	45	+
	46	+ Returns:
	47	+ A tuple (width, height).
	48	+
	49	+ Raises:
	50	+ ValueError: If the data is not a valid JPEG or SOF marker is not found.
	51	+ """
	52	+ idx = 0
	53	+ length = len(data)
	54	+
	55	+ # Check for JPEG magic number (SOI - Start Of Image)
	56	+ if length < 2 or data[0:2] != b"\xff\xd8":
	57	+ raise ValueError("Not a JPEG")
	58	+
	59	+ # Start parsing after the SOI marker
	60	+ idx = 2
	61	+ while idx + 9 < length:
	62	+ # Look for the next marker prefix (0xFF)
	63	+ if data[idx] != 0xFF:
	64	+ idx += 1
	65	+ continue
	66	+
	67	+ marker = data[idx + 1]
	68	+
	69	+ # Handle padding bytes (0xFF followed by 0xFF)
	70	+ if marker == 0xFF:
	71	+ idx += 1
	72	+ continue
	73	+
	74	+ # Check if this is a Start Of Frame marker containing dimensions
	75	+ if marker in SOF_MARKERS:
	76	+ # Structure of SOF segment:
	77	+ # [Marker(2)] [Length(2)] [Precision(1)] [Height(2)] [Width(2)] ...
	78	+ # Height is at offset +5, +6; Width is at offset +7, +8
	79	+ h = (data[idx + 5] << 8) \| data[idx + 6]
	80	+ w = (data[idx + 7] << 8) \| data[idx + 8]
	81	+ return w, h
	82	+
	83	+ # Stop if we reach End Of Image (EOI) or Start Of Scan (SOS)
	84	+ # SOS indicates the start of compressed image data, so headers are done.
	85	+ if marker in (0xD9, 0xDA):
	86	+ break
	87	+
	88	+ # Ensure there are enough bytes to read the segment length
	89	+ if idx + 3 >= length:
	90	+ break
	91	+
	92	+ # Read the length of the current segment (includes the 2 bytes for length itself)
	93	+ seg_len = (data[idx + 2] << 8) \| data[idx + 3]
	94	+ if seg_len < 2:
	95	+ break
	96	+
	97	+ # Skip to the next segment
	98	+ idx += 2 + seg_len
	99	+
	100	+ raise ValueError("JPEG SOF marker not found")
	101	+
	102	+
	103	+def parse_png_size(data: bytes):
	104	+ """
	105	+ Parses PNG binary data to extract width and height from the IHDR chunk.
	106	+
	107	+ Args:
	108	+ data: Raw PNG byte data (must be at least 24 bytes).
	109	+
	110	+ Returns:
	111	+ A tuple (width, height).
	112	+ """
	113	+ # PNG IHDR chunk structure:
	114	+ # Bytes 16-19: Width (Big Endian)
	115	+ # Bytes 20-23: Height (Big Endian)
	116	+ w = int.from_bytes(data[16:20], "big")
	117	+ h = int.from_bytes(data[20:24], "big")
	118	+ return w, h
	119	+
	120	+
	121	+def fast_get_hw(b64_str: str):
	122	+ """
	123	+ Quickly extracts image dimensions from a Base64 encoded data URI using PIL.
	124	+ Note: While 'fast' compared to full processing, it still decodes the header via PIL.
	125	+
	126	+ Args:
	127	+ b64_str: A data URI string (e.g., "data:image/jpeg;base64,...").
	128	+
	129	+ Returns:
	130	+ A tuple (width, height).
	131	+ """
	132	+ # Split the data URI to get only the base64 encoded part
	133	+ # Assumes format: "data:<mime>;base64,<encoded_data>"
	134	+ img_bytes = base64.b64decode(b64_str.split(",")[1])
	135	+
	136	+ # Open image using PIL to get dimensions
	137	+ with io.BytesIO(img_bytes) as f:
	138	+ with Image.open(f) as img:
	139	+ return img.width, img.height
	140	+
	141	+
	142	+def get_hw_from_local(path: str):
	143	+ """
	144	+ Reads the first 64KB of a local image file to determine dimensions.
	145	+ Supports PNG and JPEG formats.
	146	+
	147	+ Args:
	148	+ path: File path or file URI (starting with "file://").
	149	+
	150	+ Returns:
	151	+ A tuple (width, height).
	152	+ """
	153	+ # Remove file:// protocol prefix if present
	154	+ if path.startswith("file://"):
	155	+ path = path[7:]
	156	+
	157	+ # Read only the first 64KB, which is sufficient for headers of most images
	158	+ with open(path, "rb") as f:
	159	+ data = f.read(65536)
	160	+
	161	+ # Check PNG signature
	162	+ if data.startswith(b"\x89PNG"):
	163	+ return parse_png_size(data)
	164	+
	165	+ # Assume JPEG if not PNG
	166	+ return parse_jpeg_size(data)
	167	+
	168	+
	169	+def get_mul_token(img_url: str) -> float:
	170	+ """
	171	+ Calculates a token multiplier based on image dimensions.
	172	+ The formula divides the image into 32x32 patches and counts them.
	173	+
	174	+ Args:
	175	+ img_url: A local file path, file URI, or base64 data URI.
	176	+
	177	+ Returns:
	178	+ The calculated multiplier (float).
	179	+ """
	180	+ if img_url.startswith("data:image"):
	181	+ # Handle base64 encoded images
	182	+ h, w = fast_get_hw(img_url)
	183	+ else:
	184	+ # Handle local file paths
	185	+ h, w = get_hw_from_local(img_url)
	186	+
	187	+ # Calculate number of 32x32 patches needed to cover the image
	188	+ # Note: The original code had a bug: 'mul_token' was used before assignment.
	189	+ # It should likely be 'mul_token = ...' or returned directly.
	190	+ mul_token = math.ceil(h / 32) * math.ceil(w / 32)
	191	+
	192	+ return mul_token

Mmotor/config/endpoint.py+19-8

	@@ -22,12 +22,14 @@ from motor.engine_server.utils.ip import ip_valid_check, port_valid_check
22	from motor.engine_server.utils.validators import FileValidator	22	from motor.engine_server.utils.validators import FileValidator
23		23
24	supported_engine = ["vllm", "sglang"]	24	supported_engine = ["vllm", "sglang"]
25	-supported_role = ["prefill", "decode", "union"]	25	+supported_role = ["encode", "prefill", "decode", "union"]
26		26
		27	+MOTOR_ENGINE_ENCODE_CONFIG_KEY = "motor_engine_encode_config"
27	MOTOR_ENGINE_PREFILL_CONFIG_KEY = "motor_engine_prefill_config"	28	MOTOR_ENGINE_PREFILL_CONFIG_KEY = "motor_engine_prefill_config"
28	MOTOR_ENGINE_DECODE_CONFIG_KEY = "motor_engine_decode_config"	29	MOTOR_ENGINE_DECODE_CONFIG_KEY = "motor_engine_decode_config"
29	MODEL_CONFIG_KEY = "model_config"	30	MODEL_CONFIG_KEY = "model_config"
30	PARALLEL_CONFIG_KEY = "parallel_config"	31	PARALLEL_CONFIG_KEY = "parallel_config"
		32	+ENCODE_PARALLEL_CONFIG_KEY = "encode_parallel_config"
31	PREFILL_PARALLEL_CONFIG_KEY = "prefill_parallel_config"	33	PREFILL_PARALLEL_CONFIG_KEY = "prefill_parallel_config"
32	DECODE_PARALLEL_CONFIG_KEY = "decode_parallel_config"	34	DECODE_PARALLEL_CONFIG_KEY = "decode_parallel_config"
33		35
	@@ -57,6 +59,7 @@ class ModelConfig:
57	model_name: str	59	model_name: str
58	model_path: str	60	model_path: str
59	npu_mem_utils: float	61	npu_mem_utils: float
		62	+ encode_parallel_config: ParallelConfig
60	prefill_parallel_config: ParallelConfig	63	prefill_parallel_config: ParallelConfig
61	decode_parallel_config: ParallelConfig	64	decode_parallel_config: ParallelConfig
62		65
	@@ -66,6 +69,7 @@ class ModelConfig:
66	model_name=data["model_name"],	69	model_name=data["model_name"],
67	model_path=data["model_path"],	70	model_path=data["model_path"],
68	npu_mem_utils=data["npu_mem_utils"],	71	npu_mem_utils=data["npu_mem_utils"],
		72	+ encode_parallel_config=ParallelConfig.from_dict(data.get(ENCODE_PARALLEL_CONFIG_KEY, {})),
69	prefill_parallel_config=ParallelConfig.from_dict(data[PREFILL_PARALLEL_CONFIG_KEY]),	73	prefill_parallel_config=ParallelConfig.from_dict(data[PREFILL_PARALLEL_CONFIG_KEY]),
70	decode_parallel_config=ParallelConfig.from_dict(data[DECODE_PARALLEL_CONFIG_KEY])	74	decode_parallel_config=ParallelConfig.from_dict(data[DECODE_PARALLEL_CONFIG_KEY])
71	)	75	)
	@@ -126,18 +130,21 @@ class DeployConfig:
126	MOTOR_ENGINE_PREFILL_CONFIG_KEY in raw_data	130	MOTOR_ENGINE_PREFILL_CONFIG_KEY in raw_data
127	or MOTOR_ENGINE_DECODE_CONFIG_KEY in raw_data	131	or MOTOR_ENGINE_DECODE_CONFIG_KEY in raw_data
128	):	132	):
129	- key = (	133	+ key_map = {
130	- MOTOR_ENGINE_DECODE_CONFIG_KEY	134	+ "encode": MOTOR_ENGINE_ENCODE_CONFIG_KEY,
		zhoujing5月7日 encode改成encoder zhoujing 5月9日评论： 5月9日评论：
131	- if role == "decode"	135	+ "prefill": MOTOR_ENGINE_PREFILL_CONFIG_KEY,
132	- else MOTOR_ENGINE_PREFILL_CONFIG_KEY	136	+ "decode": MOTOR_ENGINE_DECODE_CONFIG_KEY
133	- )	137	+ }
134	- data = raw_data.get(key, {})	138	+ data = raw_data.get(key_map.get(role, ""), {})
135	_update_engine_server_tls_config(data, raw_data)	139	_update_engine_server_tls_config(data, raw_data)
136		140
137	model_cfg = data.get(MODEL_CONFIG_KEY, {})	141	model_cfg = data.get(MODEL_CONFIG_KEY, {})
		142	+ encode_cfg = raw_data.get(MOTOR_ENGINE_ENCODE_CONFIG_KEY, {}).get(MODEL_CONFIG_KEY, {})
138	prefill_cfg = raw_data.get(MOTOR_ENGINE_PREFILL_CONFIG_KEY, {}).get(MODEL_CONFIG_KEY, {})	143	prefill_cfg = raw_data.get(MOTOR_ENGINE_PREFILL_CONFIG_KEY, {}).get(MODEL_CONFIG_KEY, {})
139	decode_cfg = raw_data.get(MOTOR_ENGINE_DECODE_CONFIG_KEY, {}).get(MODEL_CONFIG_KEY, {})	144	decode_cfg = raw_data.get(MOTOR_ENGINE_DECODE_CONFIG_KEY, {}).get(MODEL_CONFIG_KEY, {})
140		145
		146	+ if ENCODE_PARALLEL_CONFIG_KEY not in model_cfg and PARALLEL_CONFIG_KEY in encode_cfg:
		147	+ model_cfg[ENCODE_PARALLEL_CONFIG_KEY] = encode_cfg[PARALLEL_CONFIG_KEY]
141	if PREFILL_PARALLEL_CONFIG_KEY not in model_cfg and PARALLEL_CONFIG_KEY in prefill_cfg:	148	if PREFILL_PARALLEL_CONFIG_KEY not in model_cfg and PARALLEL_CONFIG_KEY in prefill_cfg:
142	model_cfg[PREFILL_PARALLEL_CONFIG_KEY] = prefill_cfg[PARALLEL_CONFIG_KEY]	149	model_cfg[PREFILL_PARALLEL_CONFIG_KEY] = prefill_cfg[PARALLEL_CONFIG_KEY]
143	if DECODE_PARALLEL_CONFIG_KEY not in model_cfg and PARALLEL_CONFIG_KEY in decode_cfg:	150	if DECODE_PARALLEL_CONFIG_KEY not in model_cfg and PARALLEL_CONFIG_KEY in decode_cfg:
	@@ -168,8 +175,10 @@ class DeployConfig:
168	return self.model_config.prefill_parallel_config	175	return self.model_config.prefill_parallel_config
169	elif role == "decode":	176	elif role == "decode":
170	return self.model_config.decode_parallel_config	177	return self.model_config.decode_parallel_config
		178	+ elif role == "encode":
		179	+ return self.model_config.encode_parallel_config
171	else:	180	else:
172	- raise ValueError(f"Unsupported role: {role}. Allowed values: 'union', 'prefill', 'decode'")	181	+ raise ValueError(f"Unsupported role: {role}. Allowed values: 'union', 'prefill', 'decode', 'encode'")
173		182
174		183
175	@dataclass	184	@dataclass
	@@ -265,6 +274,8 @@ class EndpointConfig:
265	else:	274	else:
266	if self.kv_port is not None:	275	if self.kv_port is not None:
267	kv_config[constants.KV_PORT] = str(self.kv_port)	276	kv_config[constants.KV_PORT] = str(self.kv_port)
		277	+ if self.role == "encode" and self.dp_rpc_port is not None:
		278	+ self.deploy_config.model_config.encode_parallel_config.dp_rpc_port = self.dp_rpc_port
268	if self.role == "prefill" and self.dp_rpc_port is not None:	279	if self.role == "prefill" and self.dp_rpc_port is not None:
269	self.deploy_config.model_config.prefill_parallel_config.dp_rpc_port = self.dp_rpc_port	280	self.deploy_config.model_config.prefill_parallel_config.dp_rpc_port = self.dp_rpc_port
270	if self.role == "decode" and self.dp_rpc_port is not None:	281	if self.role == "decode" and self.dp_rpc_port is not None:

Mmotor/config/node_manager.py+40-8

	@@ -34,6 +34,7 @@ BASIC_CONFIG_KEY = "basic_config"
34	MODEL_CONFIG_KEY = "model_config"	34	MODEL_CONFIG_KEY = "model_config"
35	PARALLEL_CONFIG_KEY = "parallel_config"	35	PARALLEL_CONFIG_KEY = "parallel_config"
36	MOTOR_NODE_MANAGER_CONFIG_KEY = "motor_nodemanger_config"	36	MOTOR_NODE_MANAGER_CONFIG_KEY = "motor_nodemanger_config"
		37	+MOTOR_ENGINE_ENCODE_CONFIG_KEY = "motor_engine_encode_config"
37	MOTOR_ENGINE_PREFILL_CONFIG_KEY = "motor_engine_prefill_config"	38	MOTOR_ENGINE_PREFILL_CONFIG_KEY = "motor_engine_prefill_config"
38	MOTOR_ENGINE_DECODE_CONFIG_KEY = "motor_engine_decode_config"	39	MOTOR_ENGINE_DECODE_CONFIG_KEY = "motor_engine_decode_config"
39	ENGINE_CONFIG_KEY = "engine_config"	40	ENGINE_CONFIG_KEY = "engine_config"
	@@ -125,18 +126,38 @@ class SingleContainerNodemanagerConfig:
125	config.single_container_flag = True	126	config.single_container_flag = True
126	p_instances_num = user_config_data['motor_deploy_config']['p_instances_num']	127	p_instances_num = user_config_data['motor_deploy_config']['p_instances_num']
127	d_instances_num = user_config_data['motor_deploy_config']['d_instances_num']	128	d_instances_num = user_config_data['motor_deploy_config']['d_instances_num']
		129	+
		130	+ encode_model_cfg = user_config_data[MOTOR_ENGINE_ENCODE_CONFIG_KEY][MODEL_CONFIG_KEY]
128	prefill_model_cfg = user_config_data[MOTOR_ENGINE_PREFILL_CONFIG_KEY][MODEL_CONFIG_KEY]	131	prefill_model_cfg = user_config_data[MOTOR_ENGINE_PREFILL_CONFIG_KEY][MODEL_CONFIG_KEY]
129	decode_model_cfg = user_config_data[MOTOR_ENGINE_DECODE_CONFIG_KEY][MODEL_CONFIG_KEY]	132	decode_model_cfg = user_config_data[MOTOR_ENGINE_DECODE_CONFIG_KEY][MODEL_CONFIG_KEY]
		133	+
		134	+ encode_parallel_config = encode_model_cfg[PARALLEL_CONFIG_KEY]
130	prefill_parallel_config = prefill_model_cfg[PARALLEL_CONFIG_KEY]	135	prefill_parallel_config = prefill_model_cfg[PARALLEL_CONFIG_KEY]
131	decode_parallel_config = decode_model_cfg[PARALLEL_CONFIG_KEY]	136	decode_parallel_config = decode_model_cfg[PARALLEL_CONFIG_KEY]
		137	+
		138	+ e_dp_size = encode_parallel_config[DP]
		139	+ e_tp_size = encode_parallel_config[TP]
		140	+ e_pp_size = encode_parallel_config[PP]
		141	+
132	p_dp_size = prefill_parallel_config[DP]	142	p_dp_size = prefill_parallel_config[DP]
133	p_tp_size = prefill_parallel_config[TP]	143	p_tp_size = prefill_parallel_config[TP]
134	p_pp_size = prefill_parallel_config[PP]	144	p_pp_size = prefill_parallel_config[PP]
		145	+
135	d_dp_size = decode_parallel_config[DP]	146	d_dp_size = decode_parallel_config[DP]
136	d_tp_size = decode_parallel_config[TP]	147	d_tp_size = decode_parallel_config[TP]
137	d_pp_size = decode_parallel_config[PP]	148	d_pp_size = decode_parallel_config[PP]
138		149
139	index = int(Env.index)	150	index = int(Env.index)
		151	+
		152	+ d_node_manager_port_offset = p_instances_num * p_dp_size + index
		153	+ d_base_port_offset = (p_instances_num * p_dp_size + index * d_dp_size) * 2
		154	+ d_device_offset = (p_instances_num * p_dp_size * p_tp_size * p_pp_size +
		155	+ index * d_dp_size * d_tp_size * d_pp_size)
		156	+
		157	+ e_node_manager_port_offset = d_instances_num * d_dp_size + d_node_manager_port_offset
		158	+ e_base_port_offset = d_base_port_offset + e_dp_size
		159	+ e_device_offset = d_device_offset + e_dp_size * e_tp_size * e_pp_size
		160	+
140	if Env.role == 'prefill':	161	if Env.role == 'prefill':
141	config.node_manager_port_offset = index	162	config.node_manager_port_offset = index
142	config.base_port_offset = index * d_dp_size * 2	163	config.base_port_offset = index * d_dp_size * 2
	@@ -145,15 +166,22 @@ class SingleContainerNodemanagerConfig:
145	kv_port_offset = config.device_offset	166	kv_port_offset = config.device_offset
146	lookup_rpc_port_offset = index	167	lookup_rpc_port_offset = index
147	dp_rpc_port_offset = index	168	dp_rpc_port_offset = index
148	- else:	169	+ elif Env.role == 'decode':
149	- config.node_manager_port_offset = p_instances_num * p_dp_size + index	170	+ config.node_manager_port_offset = d_node_manager_port_offset
150	- config.base_port_offset = (p_instances_num * p_dp_size + index * d_dp_size) * 2	171	+ config.base_port_offset = d_base_port_offset
151	- config.device_offset = p_instances_num * p_dp_size * p_tp_size * p_pp_size + \	172	+ config.device_offset = d_device_offset
152	- index * d_dp_size * d_tp_size * d_pp_size
153	config.device_num = d_dp_size * d_tp_size * d_pp_size	173	config.device_num = d_dp_size * d_tp_size * d_pp_size
154	kv_port_offset = config.device_offset	174	kv_port_offset = config.device_offset
155	lookup_rpc_port_offset = p_instances_num + index	175	lookup_rpc_port_offset = p_instances_num + index
156	dp_rpc_port_offset = p_instances_num + index	176	dp_rpc_port_offset = p_instances_num + index
		177	+ elif Env.role == 'encode':
		178	+ config.node_manager_port_offset = e_node_manager_port_offset
		179	+ config.base_port_offset = e_base_port_offset
		180	+ config.device_offset = e_device_offset
		181	+ config.device_num = e_dp_size * e_tp_size * e_pp_size
		182	+ kv_port_offset = config.device_offset
		183	+ lookup_rpc_port_offset = index
		184	+ dp_rpc_port_offset = index
157		185
158	kv_config = user_config_data[MOTOR_ENGINE_PREFILL_CONFIG_KEY][ENGINE_CONFIG_KEY].get(KV_TRANSFER_CONFIG_KEY, {})	186	kv_config = user_config_data[MOTOR_ENGINE_PREFILL_CONFIG_KEY][ENGINE_CONFIG_KEY].get(KV_TRANSFER_CONFIG_KEY, {})
159	if kv_config:	187	if kv_config:
	@@ -245,7 +273,9 @@ class NodeManagerConfig:
245	def _load_node_manager_config_data(cls, user_cfg: dict[str, Any]) -> dict[str, Any]:	273	def _load_node_manager_config_data(cls, user_cfg: dict[str, Any]) -> dict[str, Any]:
246	"""Load node_manager_config from engine config based on role"""	274	"""Load node_manager_config from engine config based on role"""
247	engine_config_key = None	275	engine_config_key = None
248	- if Env.role == "prefill":	276	+ if Env.role == "encode":
		277	+ engine_config_key = MOTOR_ENGINE_ENCODE_CONFIG_KEY
		278	+ elif Env.role == "prefill":
249	engine_config_key = MOTOR_ENGINE_PREFILL_CONFIG_KEY	279	engine_config_key = MOTOR_ENGINE_PREFILL_CONFIG_KEY
250	elif Env.role == "decode":	280	elif Env.role == "decode":
251	engine_config_key = MOTOR_ENGINE_DECODE_CONFIG_KEY	281	engine_config_key = MOTOR_ENGINE_DECODE_CONFIG_KEY
	@@ -266,7 +296,7 @@ class NodeManagerConfig:
266	engine_config[MODEL_CONFIG_KEY][MODEL_NAME_KEY]	296	engine_config[MODEL_CONFIG_KEY][MODEL_NAME_KEY]
267	config_data[BASIC_CONFIG_KEY][HARDWARE_TYPE_KEY] = user_cfg["motor_deploy_config"][HARDWARE_TYPE_KEY]	297	config_data[BASIC_CONFIG_KEY][HARDWARE_TYPE_KEY] = user_cfg["motor_deploy_config"][HARDWARE_TYPE_KEY]
268		298
269	- if Env.role in ("prefill", "decode"):	299	+ if Env.role in ("encode", "prefill", "decode"):
270	config_data[BASIC_CONFIG_KEY]["parallel_config"] = \	300	config_data[BASIC_CONFIG_KEY]["parallel_config"] = \
271	engine_config[MODEL_CONFIG_KEY][PARALLEL_CONFIG_KEY]	301	engine_config[MODEL_CONFIG_KEY][PARALLEL_CONFIG_KEY]
272	enable_multi_endpoints = engine_config.get(ENABLE_MULTI_ENDPOINTS_KEY, True)	302	enable_multi_endpoints = engine_config.get(ENABLE_MULTI_ENDPOINTS_KEY, True)
	@@ -367,7 +397,9 @@ class NodeManagerConfig:
367	def _set_device_count_for_normal_mode(cls, config: 'NodeManagerConfig', raw: dict):	397	def _set_device_count_for_normal_mode(cls, config: 'NodeManagerConfig', raw: dict):
368	"""Set device count for normal mode using motor_deploy_config"""	398	"""Set device count for normal mode using motor_deploy_config"""
369	deploy_config = raw["motor_deploy_config"]	399	deploy_config = raw["motor_deploy_config"]
370	- if Env.role == "prefill":	400	+ if Env.role == "encode":
		401	+ device_count = deploy_config.get("e_pod_npu_num", 0)
		402	+ elif Env.role == "prefill":
371	device_count = deploy_config.get("p_pod_npu_num", 0)	403	device_count = deploy_config.get("p_pod_npu_num", 0)
372	elif Env.role == "decode":	404	elif Env.role == "decode":
373	device_count = deploy_config.get("d_pod_npu_num", 0)	405	device_count = deploy_config.get("d_pod_npu_num", 0)

Mmotor/coordinator/api_server/inference_server.py+1-1

	@@ -285,7 +285,7 @@ class InferenceServer(BaseCoordinatorServer):
285	if client is None:	285	if client is None:
286	return False	286	return False
287	readiness = await client.has_required_instances()	287	readiness = await client.has_required_instances()
288	- return readiness.is_ready() or readiness == InstanceReadiness.ONLY_PREFILL	288	+ return readiness.is_run()
289		289
290	def _register_routes(self) -> None:	290	def _register_routes(self) -> None:
291	@self._inference_app.post("/v1/completions")	291	@self._inference_app.post("/v1/completions")

Mmotor/coordinator/domain/instance_manager.py+14-2

	@@ -54,11 +54,13 @@ class InstanceManager:
54	self._available_pool: dict[int, Instance] = {}	54	self._available_pool: dict[int, Instance] = {}
55	self._unavailable_pool: dict[int, Instance] = {}	55	self._unavailable_pool: dict[int, Instance] = {}
56		56
		57	+ self._encode_pool: dict[int, Instance] = {}
57	self._prefill_pool: dict[int, Instance] = {}	58	self._prefill_pool: dict[int, Instance] = {}
58	self._decode_pool: dict[int, Instance] = {}	59	self._decode_pool: dict[int, Instance] = {}
59	self._hybrid_pool: dict[int, Instance] = {}	60	self._hybrid_pool: dict[int, Instance] = {}
60		61
61	self._available_role_pools = {	62	self._available_role_pools = {
		63	+ PDRole.ROLE_E: self._encode_pool,
62	PDRole.ROLE_P: self._prefill_pool,	64	PDRole.ROLE_P: self._prefill_pool,
63	PDRole.ROLE_D: self._decode_pool,	65	PDRole.ROLE_D: self._decode_pool,
64	PDRole.ROLE_U: self._hybrid_pool	66	PDRole.ROLE_U: self._hybrid_pool
	@@ -80,6 +82,7 @@ class InstanceManager:
80	if deploy_mode is None:	82	if deploy_mode is None:
81	logger.error("deploy_mode is required for get_required_instances_status()")	83	logger.error("deploy_mode is required for get_required_instances_status()")
82	return InstanceReadiness.UNKNOWN	84	return InstanceReadiness.UNKNOWN
		85	+ has_e = len(self._encode_pool) > 0
83	has_p = len(self._prefill_pool) > 0	86	has_p = len(self._prefill_pool) > 0
84	has_d = len(self._decode_pool) > 0	87	has_d = len(self._decode_pool) > 0
85	has_u = len(self._hybrid_pool) > 0	88	has_u = len(self._hybrid_pool) > 0
	@@ -90,12 +93,18 @@ class InstanceManager:
90	DeployMode.PD_DISAGGREGATION_SINGLE_CONTAINER,	93	DeployMode.PD_DISAGGREGATION_SINGLE_CONTAINER,
91	DeployMode.PD_DUAL_DISPATCH,	94	DeployMode.PD_DUAL_DISPATCH,
92	):	95	):
		96	+ if has_e and has_p and has_d:
		97	+ return InstanceReadiness.REQUIRED_MET_EPD
93	if has_p and has_d:	98	if has_p and has_d:
94	return InstanceReadiness.REQUIRED_MET	99	return InstanceReadiness.REQUIRED_MET
		100	+ if has_p and has_e:
		101	+ return InstanceReadiness.ENCODE_PREFILL
95	if has_p:	102	if has_p:
96	return InstanceReadiness.ONLY_PREFILL	103	return InstanceReadiness.ONLY_PREFILL
97	if has_d:	104	if has_d:
98	return InstanceReadiness.ONLY_DECODE	105	return InstanceReadiness.ONLY_DECODE
		106	+ if has_e:
		107	+ return InstanceReadiness.ONLY_ENCODE
99	return InstanceReadiness.NONE	108	return InstanceReadiness.NONE
100	if deploy_mode == DeployMode.SINGLE_NODE:	109	if deploy_mode == DeployMode.SINGLE_NODE:
101	return InstanceReadiness.REQUIRED_MET if has_u else InstanceReadiness.NONE	110	return InstanceReadiness.REQUIRED_MET if has_u else InstanceReadiness.NONE
	@@ -115,10 +124,12 @@ class InstanceManager:
115	async with self._lock:	124	async with self._lock:
116	self._available_pool = {}	125	self._available_pool = {}
117	self._unavailable_pool = {}	126	self._unavailable_pool = {}
		127	+ self._encode_pool = {}
118	self._prefill_pool = {}	128	self._prefill_pool = {}
119	self._decode_pool = {}	129	self._decode_pool = {}
120	self._hybrid_pool = {}	130	self._hybrid_pool = {}
121	self._available_role_pools = {	131	self._available_role_pools = {
		132	+ PDRole.ROLE_E: self._encode_pool,
122	PDRole.ROLE_P: self._prefill_pool,	133	PDRole.ROLE_P: self._prefill_pool,
123	PDRole.ROLE_D: self._decode_pool,	134	PDRole.ROLE_D: self._decode_pool,
124	PDRole.ROLE_U: self._hybrid_pool	135	PDRole.ROLE_U: self._hybrid_pool
	@@ -135,6 +146,7 @@ class InstanceManager:
135	# no need to lock here, asynchrony is acceptable	146	# no need to lock here, asynchrony is acceptable
136	if role is None:	147	if role is None:
137	merged = {	148	merged = {
		149	+ **self._encode_pool,
138	**self._prefill_pool,	150	**self._prefill_pool,
139	**self._decode_pool,	151	**self._decode_pool,
140	**self._hybrid_pool,	152	**self._hybrid_pool,
	@@ -294,8 +306,8 @@ class InstanceManager:
294	logger.error("Unknown event type: %s, cannot refresh instances", event_type)	306	logger.error("Unknown event type: %s, cannot refresh instances", event_type)
295	result = False	307	result = False
296	logger.info(	308	logger.info(
297	- "Refresh instances done: P=%d, D=%d, U=%d",	309	+ "Refresh instances done: E=%d, P=%d, D=%d, U=%d",
298	- len(self._prefill_pool), len(self._decode_pool), len(self._hybrid_pool),	310	+ len(self._encode_pool), len(self._prefill_pool), len(self._decode_pool), len(self._hybrid_pool),
299	)	311	)
300	return result	312	return result
301		313

Mmotor/coordinator/domain/probe.py+2-2

	@@ -219,7 +219,7 @@ class ReadinessProbe:
219	self._instance_manager.get_required_instances_status,	219	self._instance_manager.get_required_instances_status,
220	self._deploy_mode,	220	self._deploy_mode,
221	)	221	)
222	- is_ready = readiness.is_ready() or readiness == InstanceReadiness.ONLY_PREFILL	222	+ is_run= readiness.is_run()
223		223
224	r = self._daemon.read_role_and_heartbeat()	224	r = self._daemon.read_role_and_heartbeat()
225	if r.orphaned:	225	if r.orphaned:
	@@ -231,5 +231,5 @@ class ReadinessProbe:
231	else:	231	else:
232	result = ReadinessResult.OK_STANDBY	232	result = ReadinessResult.OK_STANDBY
233	# Only report ready when result is OK_*; otherwise force False (orphaned/heartbeat_stale/not_master).	233	# Only report ready when result is OK_*; otherwise force False (orphaned/heartbeat_stale/not_master).
234	- out_ready = (result in (ReadinessResult.OK_MASTER, ReadinessResult.OK_STANDBY)) and is_ready	234	+ out_ready = (result in (ReadinessResult.OK_MASTER, ReadinessResult.OK_STANDBY)) and is_run
235	return ReadinessProbeOutput(result, out_ready, readiness)	235	return ReadinessProbeOutput(result, out_ready, readiness)

Mmotor/coordinator/domain/request_manager.py+1-1

	@@ -107,7 +107,7 @@ class RequestManager:
107	async with self._lock:	107	async with self._lock:
108	key = (req_id, role)	108	key = (req_id, role)
109	if key in self._req_workload_dict:	109	if key in self._req_workload_dict:
110	- logger.debug("Workload for request %s, role %s already exists", req_id, role)	110	+ logger.warning("Workload for request %s, role %s already exists", req_id, role)
111	return False	111	return False
112	self._req_workload_dict[key] = workload	112	self._req_workload_dict[key] = workload
113	logger.debug("Added workload for request %s, role %s", req_id, role)	113	logger.debug("Added workload for request %s, role %s", req_id, role)

Mmotor/coordinator/domain/scheduling.py+15-7

	@@ -30,16 +30,24 @@ class InstanceReadiness(str, Enum):
30	Instance readiness state for deploy mode (e.g. PD separate).	30	Instance readiness state for deploy mode (e.g. PD separate).
31	Callers can distinguish "both P and D", "only P", "only D", "none" for routing/readiness.	31	Callers can distinguish "both P and D", "only P", "only D", "none" for routing/readiness.
32	"""	32	"""
33	- REQUIRED_MET = "required_met" # PD: both P and D; SINGLE_NODE: has hybrid	33	+ REQUIRED_MET_EPD = "required_met_epd" # PD: both E, P and D
34	- ONLY_PREFILL = "only_prefill" # PD mode: only prefill instances	34	+ ENCODE_PREFILL = "encode_prefill" # PD: only encode and prefill instances
35	- ONLY_DECODE = "only_decode" # PD mode: only decode instances	35	+ REQUIRED_MET = "required_met" # PD: both P and D; SINGLE_NODE: has hybrid
36	- NONE = "none" # No required instances	36	+ ONLY_PREFILL = "only_prefill" # PD mode: only prefill instances
37	- UNKNOWN = "unknown" # Unknown deploy mode	37	+ ONLY_DECODE = "only_decode" # PD mode: only decode instances
		38	+ ONLY_ENCODE = "only_encode" # EPD mode: only encode instances
		39	+ NONE = "none" # No required instances
		40	+ UNKNOWN = "unknown" # Unknown deploy mode
38		41
39	def is_ready(self) -> bool:	42	def is_ready(self) -> bool:
40	"""True if required instances are present for the deploy mode."""	43	"""True if required instances are present for the deploy mode."""
41	- return self == InstanceReadiness.REQUIRED_MET	44	+ return self == InstanceReadiness.REQUIRED_MET or self == InstanceReadiness.REQUIRED_MET_EPD
42	-	45	+
		46	+ def is_run(self) -> bool:
		47	+ """True indicates that it can run normally."""
		48	+ return (self.is_ready()
		49	+ or self == InstanceReadiness.ONLY_PREFILL
		50	+ or self == InstanceReadiness.ENCODE_PREFILL)
43		51
44	class ScheduledResource(BaseModel):	52	class ScheduledResource(BaseModel):
45	"""	53	"""

Mmotor/coordinator/domain/workload_calculator.py+36-5

	@@ -16,35 +16,66 @@ from __future__ import annotations
16	from motor.common.resources.endpoint import Workload	16	from motor.common.resources.endpoint import Workload
17	from motor.common.resources.instance import PDRole	17	from motor.common.resources.instance import PDRole
18	from motor.common.logger import get_logger	18	from motor.common.logger import get_logger
		19	+from motor.coordinator.models.request import RequestInfo
		20	+from motor.common.utils.image_utils import get_mul_token
		21	+
19		22
20	logger = get_logger(__name__)	23	logger = get_logger(__name__)
21		24
22		25
23	-def calculate_demand_workload(role: PDRole, request_length: int) -> Workload:	26	+def calculate_demand_workload(role: PDRole, req_info: RequestInfo) -> Workload:
24	"""	27	"""
25	Compute demand workload for this allocation from role and request length.	28	Compute demand workload for this allocation from role and request length.
26	Shared by BaseRouter.prepare_resource and WorkloadActionHandler ALLOCATION.	29	Shared by BaseRouter.prepare_resource and WorkloadActionHandler ALLOCATION.
27		30
28	Args:	31	Args:
29	- role: PDRole enum (prefill/decode/both)	32	+ role: PDRole enum (encode/prefill/decode/both)
30	request_length: Request length	33	request_length: Request length
31		34
32	Returns:	35	Returns:
33	Workload: Load for ALLOCATION (used by select_and_allocate / add_req_workload)	36	Workload: Load for ALLOCATION (used by select_and_allocate / add_req_workload)
34	"""	37	"""
		38	+
		39	+ if role == PDRole.ROLE_E:
		40	+ score = _calculate_encode_scores(req_info)
		41	+ return Workload(active_kv_cache=score, active_tokens=score)
35	if role == PDRole.ROLE_P:	42	if role == PDRole.ROLE_P:
36	- score = _calculate_prefill_scores(request_length)	43	+ score = _calculate_prefill_scores(req_info.req_len)
37	return Workload(active_kv_cache=score, active_tokens=score)	44	return Workload(active_kv_cache=score, active_tokens=score)
38	if role == PDRole.ROLE_D:	45	if role == PDRole.ROLE_D:
39	- score = _calculate_decode_scores(request_length)	46	+ score = _calculate_decode_scores(req_info.req_len)
40	return Workload(active_tokens=score)	47	return Workload(active_tokens=score)
41	if role == PDRole.ROLE_U:	48	if role == PDRole.ROLE_U:
42	- score = _calculate_both_scores(request_length)	49	+ score = _calculate_both_scores(req_info.req_len)
43	return Workload(active_kv_cache=score, active_tokens=score)	50	return Workload(active_kv_cache=score, active_tokens=score)
44	logger.warning("Unknown role %s for workload calculation", role)	51	logger.warning("Unknown role %s for workload calculation", role)
45	return Workload()	52	return Workload()
46		53
47		54
		55	+def _calculate_encode_scores(req_info: RequestInfo) -> float:
		56	+ """Encode role workload score."""
		57	+ messages = req_info.req_data.get("messages")
		58	+ mul_token = 0
		59	+ if not messages:
		60	+ return mul_token
		61	+
		62	+ for msg in messages:
		63	+ if not isinstance(msg.get("content"), list):
		64	+ continue
		65	+
		66	+ for content_item in msg["content"]:
		67	+ content_type = content_item.get("type")
		68	+ if not content_type:
		69	+ continue
		70	+
		71	+ if content_type == "image_url":
		72	+ img_url = content_item.get("image_url", {}).get("url", "")
		73	+ mul_token += get_mul_token(img_url)
		74	+ elif content_type == "video_url":
		75	+ mul_token += len(req_info.req_len) * 32
		76	+ return mul_token
		77	+
		78	+
48	def _calculate_prefill_scores(request_length: int) -> float:	79	def _calculate_prefill_scores(request_length: int) -> float:
49	"""Prefill role workload score."""	80	"""Prefill role workload score."""
50	length_score = request_length / 4.0	81	length_score = request_length / 4.0

Mmotor/coordinator/models/request.py+8-2

	@@ -65,6 +65,7 @@ class RequestInfo(BaseModel):
65	trace_obj: TraceObj = Field(default_factory=TraceObj, description="Tracing object")	65	trace_obj: TraceObj = Field(default_factory=TraceObj, description="Tracing object")
66	_p_cancel_scope: anyio.CancelScope \| None = PrivateAttr(default=None)	66	_p_cancel_scope: anyio.CancelScope \| None = PrivateAttr(default=None)
67	_d_cancel_scope: anyio.CancelScope \| None = PrivateAttr(default=None)	67	_d_cancel_scope: anyio.CancelScope \| None = PrivateAttr(default=None)
		68	+ _e_cancel_scope: anyio.CancelScope \| None = PrivateAttr(default=None)
68	prompt_tokens_details: dict = Field(default={}, description="prefill prompt_tokens_details")	69	prompt_tokens_details: dict = Field(default={}, description="prefill prompt_tokens_details")
69		70
70	def __init__(self, **data):	71	def __init__(self, **data):
	@@ -74,7 +75,8 @@ class RequestInfo(BaseModel):
74	@property	75	@property
75	def is_cancelled(self) -> bool:	76	def is_cancelled(self) -> bool:
76	return (self._p_cancel_scope and self._p_cancel_scope.cancel_called) \	77	return (self._p_cancel_scope and self._p_cancel_scope.cancel_called) \
77	- or (self._d_cancel_scope and self._d_cancel_scope.cancel_called)	78	+ or (self._d_cancel_scope and self._d_cancel_scope.cancel_called) \
		79	+ or (self._e_cancel_scope and self._e_cancel_scope.cancel_called)
78		80
79	def effective_entry_api(self) -> str:	81	def effective_entry_api(self) -> str:
80	"""Path used for client-contract checks (Chat vs Completion); falls back to ``api``."""	82	"""Path used for client-contract checks (Chat vs Completion); falls back to ``api``."""
	@@ -92,9 +94,13 @@ class RequestInfo(BaseModel):
92	self._p_cancel_scope = cancel_scope	94	self._p_cancel_scope = cancel_scope
93	elif role == PDRole.ROLE_D:	95	elif role == PDRole.ROLE_D:
94	self._d_cancel_scope = cancel_scope	96	self._d_cancel_scope = cancel_scope
		97	+ elif role == PDRole.ROLE_E:
		98	+ self._e_cancel_scope = cancel_scope
95		99
96	def cancel_scope(self):	100	def cancel_scope(self):
97	if self._p_cancel_scope and not self._p_cancel_scope.cancel_called:	101	if self._p_cancel_scope and not self._p_cancel_scope.cancel_called:
98	self._p_cancel_scope.cancel()	102	self._p_cancel_scope.cancel()
99	if self._d_cancel_scope and not self._d_cancel_scope.cancel_called:	103	if self._d_cancel_scope and not self._d_cancel_scope.cancel_called:
100	- self._d_cancel_scope.cancel()	104	+ self._d_cancel_scope.cancel()
		105	+ if self._e_cancel_scope and not self._e_cancel_scope.cancel_called:
		106	+ self._e_cancel_scope.cancel()

Mmotor/coordinator/router/dispatch.py+1-1

	@@ -150,7 +150,7 @@ async def handle_request(
150	DeployMode.CPCD_SEPARATE,	150	DeployMode.CPCD_SEPARATE,
151	DeployMode.PD_DISAGGREGATION_SINGLE_CONTAINER,	151	DeployMode.PD_DISAGGREGATION_SINGLE_CONTAINER,
152	)	152	)
153	- and readiness == InstanceReadiness.ONLY_PREFILL	153	+ and (readiness == InstanceReadiness.ONLY_PREFILL or readiness == InstanceReadiness.ENCODE_PREFILL)
154	):	154	):
155	deploy_mode = DeployMode.SINGLE_NODE # fallback only when has P but no D	155	deploy_mode = DeployMode.SINGLE_NODE # fallback only when has P but no D
156	else:	156	else:

Mmotor/coordinator/router/strategies/base.py+99-9

	@@ -43,6 +43,7 @@ from motor.coordinator.domain.request_manager import RequestManager
43	import motor.coordinator.router.recompute as recompute_common	43	import motor.coordinator.router.recompute as recompute_common
44	from motor.coordinator.router.workload import WorkloadActionHandler	44	from motor.coordinator.router.workload import WorkloadActionHandler
45	from motor.coordinator.tracer.tracing import TracerManager	45	from motor.coordinator.tracer.tracing import TracerManager
		46	+from motor.coordinator.domain.scheduling import InstanceReadiness
46		47
47	logger = get_logger(__name__)	48	logger = get_logger(__name__)
48		49
	@@ -243,16 +244,19 @@ class BaseRouter(ABC):
243	)	244	)
244		245
245	if result is None:	246	if result is None:
246	- raise ValueError(f"No instance available for role {role} or allocate failed")	247	+ msg = f"No instance available for role {role} or allocate failed"
		248	+ raise ValueError(msg)
247		249
248	ins, endpoint, allocate_workload = result	250	ins, endpoint, allocate_workload = result
249	if not ins or not endpoint:	251	if not ins or not endpoint:
250	- raise ValueError(f"Invalid scheduler result: {result}")	252	+ msg = f"Invalid scheduler result: {result}"
		253	+ raise ValueError(msg)
251		254
252	if not await self._request_manager.add_req_workload(	255	if not await self._request_manager.add_req_workload(
253	self.req_info.req_id, role, allocate_workload	256	self.req_info.req_id, role, allocate_workload
254	):	257	):
255	- raise RuntimeError(f"Request {self.req_info.req_id} already allocated for role {role}")	258	+ msg = f"Request {self.req_info.req_id} already allocated for role {role}"
		259	+ raise RuntimeError(msg)
256		260
257	self.req_info.update_state(	261	self.req_info.update_state(
258	ReqState.P_ALLOCATED if role == PDRole.ROLE_P else ReqState.D_ALLOCATED	262	ReqState.P_ALLOCATED if role == PDRole.ROLE_P else ReqState.D_ALLOCATED
	@@ -274,9 +278,10 @@ class BaseRouter(ABC):
274		278
275	except Exception as e:	279	except Exception as e:
276	last_exception = e	280	last_exception = e
		281	+ exc_info_flag = (attempt == 0)
277	self.logger.warning(	282	self.logger.warning(
278	"Scheduling attempt %d/%d failed for role %s: %s",	283	"Scheduling attempt %d/%d failed for role %s: %s",
279	- attempt + 1, self.config.exception_config.max_retry, role, e	284	+ attempt + 1, self.config.exception_config.max_retry, role, e, exc_info=exc_info_flag
280	)	285	)
281		286
282	if attempt < self.config.exception_config.max_retry - 1:	287	if attempt < self.config.exception_config.max_retry - 1:
	@@ -410,10 +415,11 @@ class BaseRouter(ABC):
410	is_meta=self.is_meta	415	is_meta=self.is_meta
411	)	416	)
412	t0_forward = time.perf_counter()	417	t0_forward = time.perf_counter()
413	- response = await client.post(f"/{self.req_info.api}",	418	+ url = f"/{self.req_info.api}"
414	- json=engine_req,	419	+ response = await client.post(url,
415	- headers=headers,	420	+ json=engine_req,
416	- timeout=timeout)	421	+ headers=headers,
		422	+ timeout=timeout)
417	trace_obj.add_trace_event(f"Post ok: {response.status_code}", is_meta=self.is_meta)	423	trace_obj.add_trace_event(f"Post ok: {response.status_code}", is_meta=self.is_meta)
418	elapsed_forward_ms = (time.perf_counter() - t0_forward) * 1000	424	elapsed_forward_ms = (time.perf_counter() - t0_forward) * 1000
419	if _should_log_scheduling_sample(self.req_info.req_id):	425	if _should_log_scheduling_sample(self.req_info.req_id):
	@@ -444,6 +450,90 @@ class BaseRouter(ABC):
444	async def release_kv(self, resource: ScheduledResource):	450	async def release_kv(self, resource: ScheduledResource):
445	return await self._update_workload(resource, WorkloadAction.RELEASE_KV)	451	return await self._update_workload(resource, WorkloadAction.RELEASE_KV)
446		452
		453	+ async def do_encode(self):
		454	+ if not await self._check_can_encode():
		455	+ return
		456	+ trace_obj = self.req_info.trace_obj
		457	+ headers = trace_obj.get_trace_headers_dict(self.is_meta)
		458	+ trace_context = TracerManager().extract_trace_context(headers)
		459	+ with TracerManager().tracer.start_as_current_span("CDP_Encode", context=trace_context) as span:
		460	+ self.is_meta = True
		461	+ trace_obj.meta_span = span
		462	+ trace_obj.meta_trace_headers = TracerManager().inject_trace_context()
		463	+ trace_obj.set_trace_attribute("requestId", self.req_info.req_id, is_meta=True)
		464	+
		465	+ req_data = self.req_info.req_data.copy()
		466	+ max_retry = self.config.exception_config.transport_retry_limit
		467	+ for attempt in range(max_retry):
		468	+ req_data["stream"] = False
		469	+ req_data["max_tokens"] = 1
		470	+ req_data["min_tokens"] = 1
		471	+ if "stream_options" in req_data:
		472	+ del req_data["stream_options"]
		473	+
		474	+ try:
		475	+ async with self._manage_resource_context(PDRole.ROLE_E, self.release_tokens) as resource, \
		476	+ self._manage_client_context(resource) as client:
		477	+
		478	+ cancel_scope = CancelScope()
		479	+ self.req_info.set_cancel_scope(cancel_scope, PDRole.ROLE_E)
		480	+ with cancel_scope:
		481	+ await self.forward_request(
		482	+ req_data, client, self.config.exception_config.infer_timeout
		483	+ )
		484	+ break
		485	+ except asyncio.CancelledError:
		486	+ self.logger.info("The non streaming request was terminated because of "
		487	+ "infer timeout or client disconnect.")
		488	+ self.req_info.cancel_scope()
		489	+ raise
		490	+ except HTTPException:
		491	+ self.req_info.cancel_scope()
		492	+ raise
		493	+ except Exception as e:
		494	+ last_error_str = self._log_cdp_decode_retry_error(
		495	+ "post Decode", attempt, max_retry, e, last_error_str
		496	+ )
		497	+ self.req_info.cancel_scope()
		498	+ trace_obj.set_trace_exception(e)
		499	+
		500	+ if attempt < max_retry - 1:
		501	+ wait_time = self.config.exception_config.retry_delay * (2 ** attempt)
		502	+ self.logger.info("Retrying non-streaming request in %.2f seconds...", wait_time)
		503	+ await asyncio.sleep(wait_time)
		504	+ continue
		505	+
		506	+ self.req_info.update_state(ReqState.EXCEPTION)
		507	+ raise e
		508	+
		509	+ async def _check_can_encode(self) -> bool:
		510	+ messages = self.req_info.req_data.get("messages")
		511	+ if not messages:
		512	+ return False
		513	+ is_multimodal = False
		514	+ for msg in messages:
		515	+ if not isinstance(msg.get("content"), list):
		516	+ continue
		517	+
		518	+ for content_item in msg["content"]:
		519	+ content_type = content_item.get("type")
		520	+ if not content_type:
		521	+ continue
		522	+
		523	+ if content_type == "image_url" or content_type == "video_url":
		524	+ is_multimodal = True
		525	+ break
		526	+
		527	+ if not is_multimodal:
		528	+ return False
		529	+
		530	+ instance_readiness = await self._scheduler.has_required_instances()
		531	+ if instance_readiness != InstanceReadiness.REQUIRED_MET_EPD and \
		532	+ instance_readiness != InstanceReadiness.ENCODE_PREFILL:
		533	+ return False
		534	+
		535	+ return True
		536	+
447	def _check_recompute_limit(self, retry_count: int, rmax: int) -> None:	537	def _check_recompute_limit(self, retry_count: int, rmax: int) -> None:
448	if recompute_common.recompute_limit_reached(retry_count, rmax):	538	if recompute_common.recompute_limit_reached(retry_count, rmax):
449	raise HTTPException(	539	raise HTTPException(
	@@ -495,7 +585,7 @@ class BaseRouter(ABC):
495	resource,	585	resource,
496	self.req_info.req_id,	586	self.req_info.req_id,
497	action,	587	action,
498	- self.req_info.req_len,	588	+ self.req_info,
499	)	589	)
500	if workload_change is None or role is None:	590	if workload_change is None or role is None:
501	return False	591	return False

Mmotor/coordinator/router/strategies/cdp_separate.py+6-2

	@@ -127,7 +127,9 @@ class SeparateCDPRouter(BaseRouter):
127	Handles streaming Decode requests	127	Handles streaming Decode requests
128	"""	128	"""
129	trace_obj = self.req_info.trace_obj	129	trace_obj = self.req_info.trace_obj
130	- with self._trace_span("CDP_Decode_stream", True):	130	+ with self._trace_span("CDP_Router_Stream", True):
		131	+ await self.do_encode()
		132	+ self.is_meta = False
131	self.logger.debug("Handling streaming Decode request")	133	self.logger.debug("Handling streaming Decode request")
132	max_retry = self.config.exception_config.transport_retry_limit	134	max_retry = self.config.exception_config.transport_retry_limit
133	rmax = self.config.exception_config.recompute_retry_limit	135	rmax = self.config.exception_config.recompute_retry_limit
	@@ -240,7 +242,9 @@ class SeparateCDPRouter(BaseRouter):
240	Handles non-streaming Decode requests	242	Handles non-streaming Decode requests
241	"""	243	"""
242	trace_obj = self.req_info.trace_obj	244	trace_obj = self.req_info.trace_obj
243	- with self._trace_span("CDP_Decode", False):	245	+ with self._trace_span("CDP_Router", False):
		246	+ await self.do_encode()
		247	+ self.is_meta = False
244	self.logger.debug("Handling non-streaming Decode request")	248	self.logger.debug("Handling non-streaming Decode request")
245	max_retries = self.config.exception_config.transport_retry_limit	249	max_retries = self.config.exception_config.transport_retry_limit
246	rmax = self.config.exception_config.recompute_retry_limit	250	rmax = self.config.exception_config.recompute_retry_limit

Mmotor/coordinator/router/strategies/pd_hybrid.py+5-1

	@@ -38,7 +38,9 @@ class PDHybridRouter(BaseRouter):
38	Handling hybrid streaming requests	38	Handling hybrid streaming requests
39	"""	39	"""
40	trace_obj = self.req_info.trace_obj	40	trace_obj = self.req_info.trace_obj
41	- with self._trace_span("PDHybrid_stream", True):	41	+ with self._trace_span("PDHybrid_Stream", True):
		42	+ await self.do_encode()
		43	+ self.is_meta = False
42	self.logger.debug("Handling hybrid streaming request")	44	self.logger.debug("Handling hybrid streaming request")
43	max_retry = self.config.exception_config.transport_retry_limit	45	max_retry = self.config.exception_config.transport_retry_limit
44		46
	@@ -87,6 +89,8 @@ class PDHybridRouter(BaseRouter):
87	"""	89	"""
88	trace_obj = self.req_info.trace_obj	90	trace_obj = self.req_info.trace_obj
89	with self._trace_span("PDHybrid", False):	91	with self._trace_span("PDHybrid", False):
		92	+ await self.do_encode()
		93	+ self.is_meta = False
90	self.logger.debug("Handling hybrid non-streaming request")	94	self.logger.debug("Handling hybrid non-streaming request")
91	max_retries = self.config.exception_config.transport_retry_limit	95	max_retries = self.config.exception_config.transport_retry_limit
92		96

Mmotor/coordinator/router/strategies/pd_separate.py+167-146

	@@ -19,6 +19,7 @@ import motor.coordinator.router.recompute as recompute_common
19	from motor.coordinator.router.adapters.completion_to_chat import adapt_completion_nonstream_to_chat	19	from motor.coordinator.router.adapters.completion_to_chat import adapt_completion_nonstream_to_chat
20	from motor.config.coordinator import CoordinatorConfig	20	from motor.config.coordinator import CoordinatorConfig
21	from motor.common.resources.instance import PDRole	21	from motor.common.resources.instance import PDRole
		22	+from motor.coordinator.tracer.tracing import TracerManager
22		23
23		24
24	class _DecodeTransportRetry(Exception):	25	class _DecodeTransportRetry(Exception):
	@@ -48,30 +49,34 @@ class SeparatePDRouter(BaseRouter):
48	self._stream_chunk_sent_to_client = False	49	self._stream_chunk_sent_to_client = False
49		50
50	async def generate_stream(self):	51	async def generate_stream(self):
51	- tmax = self.config.exception_config.transport_retry_limit	52	+ with self._trace_span("CPCP_Router_Stream", True):
52	- for attempt in range(tmax):	53	+ await self.do_encode()
53	- self._recompute.wants_retry = True	54	+
54	- while self._recompute.wants_retry:	55	+ tmax = self.config.exception_config.transport_retry_limit
55	- self.first_chunk_sent = False	56	+ for attempt in range(tmax):
56	- self._stream_chunk_sent_to_client = False	57	+ self._recompute.wants_retry = True
57	- self._recompute.wants_retry = False	58	+ while self._recompute.wants_retry:
58	- decode_wants_transport_retry = False	59	+ self.first_chunk_sent = False
59	- try:	60	+ self._stream_chunk_sent_to_client = False
60	- async for chunk in self.process_single_attempt(attempt):	61	+ self._recompute.wants_retry = False
61	- yield chunk	62	+ decode_wants_transport_retry = False
62	- except _DecodeTransportRetry:	63	+ try:
63	- decode_wants_transport_retry = True	64	+ async for chunk in self.process_single_attempt(attempt):
64	- if self._recompute.wants_retry:	65	+ yield chunk
65	- self._bump_req_id_after_recompute_workloads_released(	66	+ except _DecodeTransportRetry:
66	- self._recompute.retry_count	67	+ decode_wants_transport_retry = True
67	- )	68	+ if self._recompute.wants_retry:
68	- if self.is_finished:	69	+ self._bump_req_id_after_recompute_workloads_released(
69	- return	70	+ self._recompute.retry_count
70	- if decode_wants_transport_retry:	71	+ )
71	- break	72	+ if self.is_finished:
		73	+ return
		74	+ if decode_wants_transport_retry:
		75	+ break
72		76
73	async def handle_request(self) -> StreamingResponse \| JSONResponse:	77	async def handle_request(self) -> StreamingResponse \| JSONResponse:
74	"""Handle request with separate P and D instances"""	78	"""Handle request with separate P and D instances"""
		79	+
75	if self.req_info.req_data.get("stream", False):	80	if self.req_info.req_data.get("stream", False):
76	return StreamingResponse(	81	return StreamingResponse(
77	self.generate_stream(),	82	self.generate_stream(),
	@@ -80,146 +85,162 @@ class SeparatePDRouter(BaseRouter):
80	return await self._generate_nonstream_json()	85	return await self._generate_nonstream_json()
81		86
82	async def process_single_attempt(self, attempt):	87	async def process_single_attempt(self, attempt):
83	- tmax = self.config.exception_config.transport_retry_limit	88	+ trace_obj = self.req_info.trace_obj
84	- prefill_resource: ScheduledResource = None	89	+ headers = trace_obj.get_trace_headers_dict(is_meta=False)
85	- try:	90	+ trace_context = TracerManager().extract_trace_context(headers)
86	- # Schedule P instance	91	+ with TracerManager().tracer.start_as_current_span("CPCP_Prefill", context=trace_context) as span:
87	- prefill_resource = await self.prepare_resource(PDRole.ROLE_P)	92	+ trace_obj.meta_span = span
88	- # Forward P request	93	+ trace_obj.meta_trace_headers = TracerManager().inject_trace_context()
89	- p_resp_json = await self._forward_p_request(prefill_resource)	94	+ trace_obj.set_trace_attribute("requestId", self.req_info.req_id, is_meta=True)
90	- self.logger.debug("Prefill response received: %s", p_resp_json)
91	- except Exception as e:
92	- self.logger.error("Error occurred while forwarding P request: %s", e)
93	- if attempt != tmax - 1:
94	- self.is_finished = False
95	- return
96	- yield self._generate_streaming_error_chunk(e)
97	- self.is_finished = True
98	- return
99	- finally:
100	- if prefill_resource and self.req_info.state != ReqState.PREFILL_END:
101	- if not await self.release_all(prefill_resource):
102	- self.logger.debug(
103	- "release_all(prefill) returned False instance_id=%s endpoint_id=%s state=%s",
104	- prefill_resource.instance.id, prefill_resource.endpoint.id, self.req_info.state)
105		95
106	- decode_resource: ScheduledResource = None	96	+ tmax = self.config.exception_config.transport_retry_limit
107	- try:	97	+ try:
108	- # Schedule D instance	98	+ # Schedule P instance
109	- decode_resource = await self.prepare_resource(PDRole.ROLE_D)	99	+ async with self._manage_request_context(), \
110	- # Forward D request	100	+ self._manage_resource_context(PDRole.ROLE_P, self.release_tokens) as prefill_resource:
111	- async for chunk in self._forward_d_request(p_resp_json, prefill_resource, decode_resource):	101	+ # Forward P request
112	- if chunk:	102	+ p_resp_json = await self._forward_p_request(prefill_resource)
113	- self._stream_chunk_sent_to_client = True	103	+ self.logger.debug("Prefill response received: %s", p_resp_json)
114	- yield chunk	104	+ except Exception as e:
115	- if not self._recompute.wants_retry:	105	+ self.logger.error("Error occurred while forwarding P request: %s", e)
116	- self.is_finished = True	106	+ if attempt != tmax - 1:
117	- return	107	+ self.is_finished = False
118	- except HTTPException as e:	108	+ return
119	- self.logger.error("Error occurred while forwarding Decode request: %s", e)
120	- await self._handle_stream_error(prefill_resource, e)
121	- yield self._generate_streaming_error_chunk(e)
122	- self.is_finished = True
123	- return
124	- except Exception as e:
125	- self.logger.error("Error occurred while forwarding Decode request: %s", e)
126	- await self._handle_stream_error(prefill_resource, e)
127	- if self._stream_chunk_sent_to_client or attempt == tmax - 1:
128	yield self._generate_streaming_error_chunk(e)	109	yield self._generate_streaming_error_chunk(e)
129	self.is_finished = True	110	self.is_finished = True
130	- else:	111	+ trace_obj.set_trace_exception(e)
131	- raise _DecodeTransportRetry() from e	112	+ return
132	- finally:	113	+
133	- if decode_resource:	114	+ with TracerManager().tracer.start_as_current_span("CPCP_Decode", context=trace_context) as span:
134	- released = await self.release_tokens(decode_resource)	115	+ trace_obj.meta_span = span
135	- if not released:	116	+ trace_obj.meta_trace_headers = TracerManager().inject_trace_context()
136	- self.logger.debug(	117	+ trace_obj.set_trace_attribute("requestId", self.req_info.req_id, is_meta=True)
137	- "release_tokens(decode) returned False instance_id=%s endpoint_id=%s state=%s",	118	+
138	- decode_resource.instance.id, decode_resource.endpoint.id, self.req_info.state)	119	+ try:
		120	+ # Schedule D instance
		121	+ async with self._manage_request_context(), \
		122	+ self._manage_resource_context(PDRole.ROLE_D, self.release_tokens) as decode_resource:
		123	+ # Forward D request
		124	+ async for chunk in self._forward_d_request(p_resp_json, prefill_resource, decode_resource):
		125	+ if chunk:
		126	+ self._stream_chunk_sent_to_client = True
		127	+ yield chunk
		128	+ if not self._recompute.wants_retry:
		129	+ self.is_finished = True
		130	+ self.logger.info(trace_obj.set_end_and_ttft_tpot())
		131	+ return
		132	+ except HTTPException as e:
		133	+ trace_obj.set_trace_exception(e)
		134	+ self.logger.error("Error occurred while forwarding Decode request: %s", e)
		135	+ await self._handle_stream_error(prefill_resource, e)
		136	+ yield self._generate_streaming_error_chunk(e)
		137	+ self.is_finished = True
		138	+ return
		139	+ except Exception as e:
		140	+ self.logger.error("Error occurred while forwarding Decode request: %s", e)
		141	+ await self._handle_stream_error(prefill_resource, e)
		142	+ if self._stream_chunk_sent_to_client or attempt == tmax - 1:
		143	+ yield self._generate_streaming_error_chunk(e)
		144	+ self.is_finished = True
		145	+ trace_obj.set_trace_exception(e)
		146	+ else:
		147	+ raise _DecodeTransportRetry() from e
139		148
140	async def _generate_nonstream_json(self) -> JSONResponse:	149	async def _generate_nonstream_json(self) -> JSONResponse:
141	- tmax = self.config.exception_config.transport_retry_limit	150	+ with self._trace_span("CPCP_Router", False):
142	- for attempt in range(tmax):	151	+ await self.do_encode()
143	- self._recompute.wants_retry = True	152	+
144	- while self._recompute.wants_retry:	153	+ tmax = self.config.exception_config.transport_retry_limit
145	- self.first_chunk_sent = False	154	+ for attempt in range(tmax):
146	- self._recompute.wants_retry = False	155	+ self._recompute.wants_retry = True
147	- resp = await self._nonstream_single_attempt(attempt)	156	+ while self._recompute.wants_retry:
148	- if resp is not None:	157	+ self.first_chunk_sent = False
149	- return resp	158	+ self._recompute.wants_retry = False
150	- if self._recompute.wants_retry:	159	+ resp = await self._nonstream_single_attempt(attempt)
151	- self._bump_req_id_after_recompute_workloads_released(	160	+ if resp is not None:
152	- self._recompute.retry_count	161	+ return resp
153	- )	162	+ if self._recompute.wants_retry:
154	- if self.is_finished:	163	+ self._bump_req_id_after_recompute_workloads_released(
155	- return JSONResponse(	164	+ self._recompute.retry_count
156	- status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,	165	+ )
157	- content=self.build_error_response(	166	+ if self.is_finished:
158	- RuntimeError("Non-stream PD request ended without response")	167	+ return JSONResponse(
159	- ).model_dump(),	168	+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
160	- )	169	+ content=self.build_error_response(
161	- return JSONResponse(	170	+ RuntimeError("Non-stream PD request ended without response")
162	- status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,	171	+ ).model_dump(),
163	- content=self.build_error_response(	172	+ )
164	- RuntimeError("All retries exhausted for non-stream PD request")	173	+ return JSONResponse(
165	- ).model_dump(),	174	+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
166	- )	175	+ content=self.build_error_response(
		176	+ RuntimeError("All retries exhausted for non-stream PD request")
		177	+ ).model_dump(),
		178	+ )
167		179
168	def _nonstream_error_json_response(self, exc: Exception) -> JSONResponse:	180	def _nonstream_error_json_response(self, exc: Exception) -> JSONResponse:
169	er = self.build_error_response(exc)	181	er = self.build_error_response(exc)
170	return JSONResponse(status_code=er.code, content=er.model_dump())	182	return JSONResponse(status_code=er.code, content=er.model_dump())
171		183
172	async def _nonstream_single_attempt(self, attempt: int) -> JSONResponse \| None:	184	async def _nonstream_single_attempt(self, attempt: int) -> JSONResponse \| None:
173	- tmax = self.config.exception_config.transport_retry_limit	185	+ trace_obj = self.req_info.trace_obj
174	- prefill_resource: ScheduledResource = None	186	+ headers = trace_obj.get_trace_headers_dict(is_meta=False)
175	- try:	187	+ trace_context = TracerManager().extract_trace_context(headers)
176	- prefill_resource = await self.prepare_resource(PDRole.ROLE_P)	188	+ with TracerManager().tracer.start_as_current_span("CPCP_Prefill", context=trace_context) as span:
177	- p_resp_json = await self._forward_p_request(prefill_resource)	189	+ trace_obj.meta_span = span
178	- self.logger.debug("Prefill response received: %s", p_resp_json)	190	+ trace_obj.meta_trace_headers = TracerManager().inject_trace_context()
179	- except Exception as e:	191	+ trace_obj.set_trace_attribute("requestId", self.req_info.req_id, is_meta=True)
180	- self.logger.error("Error occurred while forwarding P request: %s", e)	192	+
181	- if attempt != tmax - 1:	193	+ tmax = self.config.exception_config.transport_retry_limit
182	- self.is_finished = False
183	- return None
184	- self.is_finished = True
185	- return self._nonstream_error_json_response(e)
186	- finally:
187	- if prefill_resource and self.req_info.state != ReqState.PREFILL_END:
188	- if not await self.release_all(prefill_resource):
189	- self.logger.debug(
190	- "release_all(prefill) returned False instance_id=%s endpoint_id=%s state=%s",
191	- prefill_resource.instance.id, prefill_resource.endpoint.id, self.req_info.state,
192	- )
193	-
194	- decode_resource: ScheduledResource = None
195	- try:
196	- decode_resource = await self.prepare_resource(PDRole.ROLE_D)
197	try:	194	try:
198	- req_data = self._gen_d_request(p_resp_json)	195	+ async with self._manage_request_context(), \
199	- except BaseException:	196	+ self._manage_resource_context(PDRole.ROLE_P, self.release_tokens) as prefill_resource:
200	- await self.release_tokens(decode_resource)	197	+ p_resp_json = await self._forward_p_request(prefill_resource)
201	- raise	198	+ self.logger.debug("Prefill response received: %s", p_resp_json)
202	- body = await self._fetch_nonstream_decode_body(	199	+ except Exception as e:
203	- req_data, prefill_resource, decode_resource	200	+ self.logger.error("Error occurred while forwarding P request: %s", e)
204	- )	201	+ if attempt != tmax - 1:
205	- if body is not None:	202	+ self.is_finished = False
206	- self.req_info.update_state(ReqState.DECODE_END)	203	+ return None
207	- self.logger.debug("Completed non-stream decode for request %s", self.req_info)
208	self.is_finished = True	204	self.is_finished = True
209	- return JSONResponse(content=body)	205	+ trace_obj.set_trace_exception(e)
210	- return None	206	+ return self._nonstream_error_json_response(e)
211	- except HTTPException as e:	207	+
212	- self.logger.error("Error occurred while forwarding Decode request: %s", e)	208	+ with TracerManager().tracer.start_as_current_span("CPCP_Decode", context=trace_context) as span:
213	- await self._handle_stream_error(prefill_resource, e)	209	+ trace_obj.meta_span = span
214	- self.is_finished = True	210	+ trace_obj.meta_trace_headers = TracerManager().inject_trace_context()
215	- return self._nonstream_error_json_response(e)	211	+ trace_obj.set_trace_attribute("requestId", self.req_info.req_id, is_meta=True)
216	- except Exception as e:	212	+
217	- self.logger.error("Error occurred while forwarding Decode request: %s", e)	213	+ try:
218	- await self._handle_stream_error(prefill_resource, e)	214	+ async with self._manage_request_context(), \
219	- if self.first_chunk_sent or attempt == tmax - 1:	215	+ self._manage_resource_context(PDRole.ROLE_D, self.release_tokens) as decode_resource:
		216	+ try:
		217	+ req_data = self._gen_d_request(p_resp_json)
		218	+ except BaseException:
		219	+ await self.release_tokens(decode_resource)
		220	+ raise
		221	+ body = await self._fetch_nonstream_decode_body(
		222	+ req_data, prefill_resource, decode_resource
		223	+ )
		224	+ if body is not None:
		225	+ self.req_info.update_state(ReqState.DECODE_END)
		226	+ self.logger.debug("Completed non-stream decode for request %s", self.req_info)
		227	+ self.is_finished = True
		228	+ return JSONResponse(content=body)
		229	+ return None
		230	+ except HTTPException as e:
		231	+ trace_obj.set_trace_exception(e)
		232	+ self.logger.error("Error occurred while forwarding Decode request: %s", e)
		233	+ await self._handle_stream_error(prefill_resource, e)
220	self.is_finished = True	234	self.is_finished = True
221	return self._nonstream_error_json_response(e)	235	return self._nonstream_error_json_response(e)
222	- return None	236	+ except Exception as e:
		237	+ self.logger.error("Error occurred while forwarding Decode request: %s", e)
		238	+ await self._handle_stream_error(prefill_resource, e)
		239	+ if self.first_chunk_sent or attempt == tmax - 1:
		240	+ self.is_finished = True
		241	+ trace_obj.set_trace_exception(e)
		242	+ return self._nonstream_error_json_response(e)
		243	+ return None
223		244
224	def _gen_p_request(self) -> dict:	245	def _gen_p_request(self) -> dict:
225	"""Generate P request parameters"""	246	"""Generate P request parameters"""

Mmotor/coordinator/router/workload.py+3-2

	@@ -21,6 +21,7 @@ from motor.common.logger import get_logger
21	from motor.coordinator.domain.request_manager import RequestManager	21	from motor.coordinator.domain.request_manager import RequestManager
22	from motor.coordinator.domain import ScheduledResource	22	from motor.coordinator.domain import ScheduledResource
23	from motor.coordinator.domain.workload_calculator import calculate_demand_workload	23	from motor.coordinator.domain.workload_calculator import calculate_demand_workload
		24	+from motor.coordinator.models.request import RequestInfo
24		25
25	logger = get_logger(__name__)	26	logger = get_logger(__name__)
26		27
	@@ -61,7 +62,7 @@ class WorkloadActionHandler:
61	resource: ScheduledResource,	62	resource: ScheduledResource,
62	req_id: str,	63	req_id: str,
63	action: WorkloadAction,	64	action: WorkloadAction,
64	- req_len: int,	65	+ req_info: RequestInfo,
65	) -> Tuple[Workload \| None, PDRole \| None]:	66	) -> Tuple[Workload \| None, PDRole \| None]:
66	"""	67	"""
67	Get/compute workload_change from RequestManager by action, update RequestManager, return (change, role).	68	Get/compute workload_change from RequestManager by action, update RequestManager, return (change, role).
	@@ -87,7 +88,7 @@ class WorkloadActionHandler:
87	workload_change: Workload \| None = None	88	workload_change: Workload \| None = None
88		89
89	if action == WorkloadAction.ALLOCATION:	90	if action == WorkloadAction.ALLOCATION:
90	- allocate_workload = calculate_demand_workload(role, req_len)	91	+ allocate_workload = calculate_demand_workload(role, req_info)
91	if not await request_mgr.add_req_workload(req_id, role, allocate_workload):	92	if not await request_mgr.add_req_workload(req_id, role, allocate_workload):
92	logger.debug(	93	logger.debug(
93	"Request %s already allocated for role %s, allocation ignored",	94	"Request %s already allocated for role %s, allocation ignored",

Mmotor/coordinator/scheduler/runtime/scheduler_client.py+17-8

	@@ -98,11 +98,13 @@ class _SchedulerInstanceCache:
98		98
99	def __init__(self):	99	def __init__(self):
100	self._instance_cache: dict[PDRole, list[Instance]] = {	100	self._instance_cache: dict[PDRole, list[Instance]] = {
		101	+ PDRole.ROLE_E: [],
101	PDRole.ROLE_P: [],	102	PDRole.ROLE_P: [],
102	PDRole.ROLE_D: [],	103	PDRole.ROLE_D: [],
103	PDRole.ROLE_U: [],	104	PDRole.ROLE_U: [],
104	}	105	}
105	self._instance_map: dict[PDRole, dict[int, Instance]] = {	106	self._instance_map: dict[PDRole, dict[int, Instance]] = {
		107	+ PDRole.ROLE_E: {},
106	PDRole.ROLE_P: {},	108	PDRole.ROLE_P: {},
107	PDRole.ROLE_D: {},	109	PDRole.ROLE_D: {},
108	PDRole.ROLE_U: {},	110	PDRole.ROLE_U: {},
	@@ -614,7 +616,7 @@ class AsyncSchedulerClient:
614	workload = (	616	workload = (
615	Workload()	617	Workload()
616	if (self._scheduler_type or "round_robin") == "round_robin"	618	if (self._scheduler_type or "round_robin") == "round_robin"
617	- else calculate_demand_workload(role, req_info.req_len)	619	+ else calculate_demand_workload(role, req_info)
618	)	620	)
619		621
620	request_id = str(uuid.uuid4())	622	request_id = str(uuid.uuid4())
	@@ -653,6 +655,7 @@ class AsyncSchedulerClient:
653	role_str, out_instance.id, out_endpoint.id	655	role_str, out_instance.id, out_endpoint.id
654	)	656	)
655	return (out_instance, out_endpoint, workload)	657	return (out_instance, out_endpoint, workload)
		658	+
656	return None	659	return None
657		660
658	async def update_workload(self, params: UpdateWorkloadParams) -> bool:	661	async def update_workload(self, params: UpdateWorkloadParams) -> bool:
	@@ -743,11 +746,13 @@ class AsyncSchedulerClient:
743	)	746	)
744	else:	747	else:
745	role_to_list: dict[PDRole, list] = {	748	role_to_list: dict[PDRole, list] = {
		749	+ PDRole.ROLE_E: [],
746	PDRole.ROLE_P: [],	750	PDRole.ROLE_P: [],
747	PDRole.ROLE_D: [],	751	PDRole.ROLE_D: [],
748	PDRole.ROLE_U: [],	752	PDRole.ROLE_U: [],
749	}	753	}
750	_role_map = {	754	_role_map = {
		755	+ "encode": PDRole.ROLE_E,
751	"prefill": PDRole.ROLE_P,	756	"prefill": PDRole.ROLE_P,
752	"decode": PDRole.ROLE_D,	757	"decode": PDRole.ROLE_D,
753	"both": PDRole.ROLE_U,	758	"both": PDRole.ROLE_U,
	@@ -779,35 +784,39 @@ class AsyncSchedulerClient:
779	elif isinstance(mode, str):	784	elif isinstance(mode, str):
780	mode = DeployMode.from_string(mode) or DeployMode.PD_SEPARATE	785	mode = DeployMode.from_string(mode) or DeployMode.PD_SEPARATE
781		786
782	- def _status(p_list: list, d_list: list, u_list: list) -> InstanceReadiness:	787	+ def _status(e_list: list, p_list: list, d_list: list, u_list: list) -> InstanceReadiness:
783	if mode in (DeployMode.CDP_SEPARATE, DeployMode.CPCD_SEPARATE, DeployMode.PD_SEPARATE, \	788	if mode in (DeployMode.CDP_SEPARATE, DeployMode.CPCD_SEPARATE, DeployMode.PD_SEPARATE, \
784	DeployMode.PD_DISAGGREGATION_SINGLE_CONTAINER, DeployMode.PD_DUAL_DISPATCH):	789	DeployMode.PD_DISAGGREGATION_SINGLE_CONTAINER, DeployMode.PD_DUAL_DISPATCH):
785	- has_p, has_d = len(p_list) > 0, len(d_list) > 0	790	+ has_e, has_p, has_d = len(e_list) > 0, len(p_list) > 0, len(d_list) > 0
		791	+ if has_e and has_p and has_d:
		792	+ return InstanceReadiness.REQUIRED_MET_EPD
786	if has_p and has_d:	793	if has_p and has_d:
787	return InstanceReadiness.REQUIRED_MET	794	return InstanceReadiness.REQUIRED_MET
		795	+ if has_p and has_e:
		796	+ return InstanceReadiness.ENCODE_PREFILL
788	if has_p:	797	if has_p:
789	return InstanceReadiness.ONLY_PREFILL	798	return InstanceReadiness.ONLY_PREFILL
790	if has_d:	799	if has_d:
791	return InstanceReadiness.ONLY_DECODE	800	return InstanceReadiness.ONLY_DECODE
		801	+ if has_e:
		802	+ return InstanceReadiness.ONLY_ENCODE
792	return InstanceReadiness.NONE	803	return InstanceReadiness.NONE
793	if mode == DeployMode.SINGLE_NODE:	804	if mode == DeployMode.SINGLE_NODE:
794	return InstanceReadiness.REQUIRED_MET if len(u_list) > 0 else InstanceReadiness.NONE	805	return InstanceReadiness.REQUIRED_MET if len(u_list) > 0 else InstanceReadiness.NONE
795	return InstanceReadiness.UNKNOWN	806	return InstanceReadiness.UNKNOWN
796		807
		808	+ e_list = self._cache.get_instances(PDRole.ROLE_E)
797	p_list = self._cache.get_instances(PDRole.ROLE_P)	809	p_list = self._cache.get_instances(PDRole.ROLE_P)
798	d_list = self._cache.get_instances(PDRole.ROLE_D)	810	d_list = self._cache.get_instances(PDRole.ROLE_D)
799	u_list = self._cache.get_instances(PDRole.ROLE_U)	811	u_list = self._cache.get_instances(PDRole.ROLE_U)
800	- status = _status(p_list, d_list, u_list)	812	+ status = _status(e_list, p_list, d_list, u_list)
801	if status.is_ready():	813	if status.is_ready():
802	return status	814	return status
803	try:	815	try:
804	await self.get_available_instances(None)	816	await self.get_available_instances(None)
805	except Exception as e:	817	except Exception as e:
806	logger.debug("has_required_instances: warm-up get_available_instances failed: %s", e)	818	logger.debug("has_required_instances: warm-up get_available_instances failed: %s", e)
807	- p_list = self._cache.get_instances(PDRole.ROLE_P)	819	+ return _status(e_list, p_list, d_list, u_list)
808	- d_list = self._cache.get_instances(PDRole.ROLE_D)
809	- u_list = self._cache.get_instances(PDRole.ROLE_U)
810	- return _status(p_list, d_list, u_list)
811		820
812	async def get_all_instances(self) -> tuple[dict[int, Instance], dict[int, Instance]]:	821	async def get_all_instances(self) -> tuple[dict[int, Instance], dict[int, Instance]]:
813	"""Interface compat; returns empty (Mgmt process uses local InstanceManager)."""	822	"""Interface compat; returns empty (Mgmt process uses local InstanceManager)."""

Mmotor/coordinator/scheduler/runtime/scheduler_server.py+2-2

	@@ -305,7 +305,7 @@ class _SchedulerRequestDispatcher:
305	request_id=request.request_id,	305	request_id=request.request_id,
306	data={_KEY_INSTANCE: None, _KEY_ENDPOINT: None},	306	data={_KEY_INSTANCE: None, _KEY_ENDPOINT: None},
307	)	307	)
308	- role = PDRole(role_str) if role_str in ("prefill", "decode", "both") else PDRole.ROLE_U	308	+ role = PDRole(role_str) if role_str in ("encode", "prefill", "decode", "both") else PDRole.ROLE_U
309	params = UpdateWorkloadParams(	309	params = UpdateWorkloadParams(
310	instance_id=iid,	310	instance_id=iid,
311	endpoint_id=eid,	311	endpoint_id=eid,
	@@ -325,7 +325,7 @@ class _SchedulerRequestDispatcher:
325	await self._workload_writer.write_single_entry(iid, eid)	325	await self._workload_writer.write_single_entry(iid, eid)
326	instance = None	326	instance = None
327	endpoint = None	327	endpoint = None
328	- for r in (PDRole.ROLE_P, PDRole.ROLE_D, PDRole.ROLE_U):	328	+ for r in (PDRole.ROLE_E, PDRole.ROLE_P, PDRole.ROLE_D, PDRole.ROLE_U):
329	inst = self._instance_manager.get_available_instances(r).get(iid)	329	inst = self._instance_manager.get_available_instances(r).get(iid)
330	if inst:	330	if inst:
331	instance = inst	331	instance = inst

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

	@@ -104,7 +104,7 @@ class Scheduler:
104	workload = (	104	workload = (
105	Workload()	105	Workload()
106	if not hasattr(self._scheduling_policy, "update_workload")	106	if not hasattr(self._scheduling_policy, "update_workload")
107	- else calculate_demand_workload(role, req_info.req_len)	107	+ else calculate_demand_workload(role, req_info)
108	)	108	)
109	params = UpdateWorkloadParams(	109	params = UpdateWorkloadParams(
110	instance_id=instance.id,	110	instance_id=instance.id,

Mmotor/engine_server/core/health_collector.py+1-1

	@@ -35,7 +35,7 @@ class HealthCollector:
35	response = await client.get("/health")	35	response = await client.get("/health")
36	response.raise_for_status()	36	response.raise_for_status()
37	response_text = await response.aread()	37	response_text = await response.aread()
38	- health_status = response_text.decode('utf-8').lower() == 'true'	38	+ health_status = response_text.decode('utf-8').lower() != 'false'
39	self._has_connected = True	39	self._has_connected = True
40	return health_status	40	return health_status
41	except Exception as e:	41	except Exception as e:

Mrequirements.txt+1-0

	@@ -21,3 +21,4 @@ anyio>=4.0.0
21	setproctitle>=1.3.0	21	setproctitle>=1.3.0
22	confluent_kafka>=2.13.2	22	confluent_kafka>=2.13.2
23	transformers>=4.57.6	23	transformers>=4.57.6
		24	+Pillow>=12.2.0

Mtests/coordinator/core/test_scheduler.py+14-6

	@@ -9,7 +9,7 @@
9	# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,	9	# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
10	# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.	10	# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
11	# See the Mulan PSL v2 for more details.	11	# See the Mulan PSL v2 for more details.
12	-from unittest.mock import patch, AsyncMock	12	+from unittest.mock import patch, AsyncMock, MagicMock
13		13
14	import pytest	14	import pytest
15	import httpx	15	import httpx
	@@ -152,7 +152,9 @@ async def test_request_processing_pd_separation_scenario(scheduler_setup):
152	assert selected_prefill_instance.role == PDRole.ROLE_P	152	assert selected_prefill_instance.role == PDRole.ROLE_P
153		153
154	# 2. allocate prefill workload	154	# 2. allocate prefill workload
155	- workload_p = calculate_demand_workload(PDRole.ROLE_P, request_length)	155	+ req_info = MagicMock()
		156	+ req_info.req_len = request_length
		157	+ workload_p = calculate_demand_workload(PDRole.ROLE_P, req_info)
156	result = await load_balance_scheduler.update_workload(	158	result = await load_balance_scheduler.update_workload(
157	selected_prefill_instance.id, selected_prefill_endpoint.id, req_id,	159	selected_prefill_instance.id, selected_prefill_endpoint.id, req_id,
158	WorkloadAction.ALLOCATION, workload_p	160	WorkloadAction.ALLOCATION, workload_p
	@@ -181,7 +183,7 @@ async def test_request_processing_pd_separation_scenario(scheduler_setup):
181	assert selected_decode_instance.role == PDRole.ROLE_D	183	assert selected_decode_instance.role == PDRole.ROLE_D
182		184
183	# 5. allocate decode workload	185	# 5. allocate decode workload
184	- workload_d = calculate_demand_workload(PDRole.ROLE_D, request_length)	186	+ workload_d = calculate_demand_workload(PDRole.ROLE_D, req_info)
185	result = await load_balance_scheduler.update_workload(	187	result = await load_balance_scheduler.update_workload(
186	selected_decode_instance.id, selected_decode_endpoint.id, req_id,	188	selected_decode_instance.id, selected_decode_endpoint.id, req_id,
187	WorkloadAction.ALLOCATION, workload_d	189	WorkloadAction.ALLOCATION, workload_d
	@@ -228,7 +230,9 @@ async def test_request_processing_mix_scenario(scheduler_setup):
228		230
229	# 2. allocate mix workload	231	# 2. allocate mix workload
230	load_balance_scheduler = scheduler.get_scheduling_policy()	232	load_balance_scheduler = scheduler.get_scheduling_policy()
231	- workload_u = calculate_demand_workload(PDRole.ROLE_U, request_length)	233	+ req_info = MagicMock()
		234	+ req_info.req_len = request_length
		235	+ workload_u = calculate_demand_workload(PDRole.ROLE_U, req_info)
232	result = await load_balance_scheduler.update_workload(	236	result = await load_balance_scheduler.update_workload(
233	selected_instance.id, selected_endpoint.id, req_id,	237	selected_instance.id, selected_endpoint.id, req_id,
234	WorkloadAction.ALLOCATION, workload_u	238	WorkloadAction.ALLOCATION, workload_u
	@@ -276,7 +280,9 @@ async def test_multiple_requests_load_balancing(scheduler_setup, request_length)
276		280
277	# allocate workload	281	# allocate workload
278	load_balance_scheduler = scheduler.get_scheduling_policy()	282	load_balance_scheduler = scheduler.get_scheduling_policy()
279	- workload = calculate_demand_workload(PDRole.ROLE_P, request_length)	283	+ req_info = MagicMock()
		284	+ req_info.req_len = request_length
		285	+ workload = calculate_demand_workload(PDRole.ROLE_P, req_info)
280	result = await load_balance_scheduler.update_workload(	286	result = await load_balance_scheduler.update_workload(
281	selected_instance.id, selected_endpoint.id, req_id,	287	selected_instance.id, selected_endpoint.id, req_id,
282	WorkloadAction.ALLOCATION, workload	288	WorkloadAction.ALLOCATION, workload
	@@ -316,7 +322,9 @@ async def test_workload_calculation_accuracy(scheduler_setup):
316	selected_instance, selected_endpoint = result	322	selected_instance, selected_endpoint = result
317		323
318	# allocate prefill workload	324	# allocate prefill workload
319	- workload = calculate_demand_workload(PDRole.ROLE_P, request_length)	325	+ req_info = MagicMock()
		326	+ req_info.req_len = request_length
		327	+ workload = calculate_demand_workload(PDRole.ROLE_P, req_info)
320	result = await load_balance_scheduler.update_workload(	328	result = await load_balance_scheduler.update_workload(
321	selected_instance.id, selected_endpoint.id, req_id,	329	selected_instance.id, selected_endpoint.id, req_id,
322	WorkloadAction.ALLOCATION, workload	330	WorkloadAction.ALLOCATION, workload

Mtests/coordinator/router/test_router_pd_separation.py+3-1

	@@ -380,7 +380,7 @@ class TestRouterPDSeparation:
380	return InstanceReadiness.REQUIRED_MET	380	return InstanceReadiness.REQUIRED_MET
381	monkeypatch.setattr(InstanceManager, "get_required_instances_status", mock_get_required_instances_status)	381	monkeypatch.setattr(InstanceManager, "get_required_instances_status", mock_get_required_instances_status)
382		382
383	- async def mock_select_and_allocate(self, role, req_id, req_len):	383	+ async def mock_select_and_allocate(self, role, req_info):
384	return None	384	return None
385	monkeypatch.setattr(Scheduler, "select_and_allocate", mock_select_and_allocate)	385	monkeypatch.setattr(Scheduler, "select_and_allocate", mock_select_and_allocate)
386		386
	@@ -667,6 +667,8 @@ class TestRouterPDSeparation:
667	"messages": [{"role": "user", "content": "Hello"}]	667	"messages": [{"role": "user", "content": "Hello"}]
668	})	668	})
669		669
		670	+ print("Response text:::::: ", response.text)
		671	+
670	assert error_message in response.text	672	assert error_message in response.text
671	# May be 400 or 500 if upstream wraps 4XX	673	# May be 400 or 500 if upstream wraps 4XX
672	assert str(status.HTTP_400_BAD_REQUEST) in response.text or "Bad Request" in response.text	674	assert str(status.HTTP_400_BAD_REQUEST) in response.text or "Bad Request" in response.text

Mtests/coordinator/router/test_workload_action_handler.py+27-9

	@@ -26,7 +26,9 @@ class TestCalculateDemandWorkload:
26		26
27	def test_prefill_role(self):	27	def test_prefill_role(self):
28	"""ROLE_P: active_kv_cache and active_tokens both set from prefill formula."""	28	"""ROLE_P: active_kv_cache and active_tokens both set from prefill formula."""
29	- w = calculate_demand_workload(PDRole.ROLE_P, 4)	29	+ req_info = MagicMock()
		30	+ req_info.req_len = 4
		31	+ w = calculate_demand_workload(PDRole.ROLE_P, req_info)
30	assert isinstance(w, Workload)	32	assert isinstance(w, Workload)
31	# request_length=4 -> length_score=1.0 -> score = 1.0*0.0345+120.0745 = 120.109	33	# request_length=4 -> length_score=1.0 -> score = 1.0*0.0345+120.0745 = 120.109
32	assert w.active_kv_cache > 0	34	assert w.active_kv_cache > 0
	@@ -35,13 +37,17 @@ class TestCalculateDemandWorkload:
35		37
36	def test_decode_role(self):	38	def test_decode_role(self):
37	"""ROLE_D: only active_tokens set (request_length)."""	39	"""ROLE_D: only active_tokens set (request_length)."""
38	- w = calculate_demand_workload(PDRole.ROLE_D, 10)	40	+ req_info = MagicMock()
		41	+ req_info.req_len = 10
		42	+ w = calculate_demand_workload(PDRole.ROLE_D, req_info)
39	assert w.active_tokens == 10.0	43	assert w.active_tokens == 10.0
40	assert w.active_kv_cache == 0	44	assert w.active_kv_cache == 0
41		45
42	def test_hybrid_role(self):	46	def test_hybrid_role(self):
43	"""ROLE_U: both set, average of prefill and decode scores."""	47	"""ROLE_U: both set, average of prefill and decode scores."""
44	- w = calculate_demand_workload(PDRole.ROLE_U, 4)	48	+ req_info = MagicMock()
		49	+ req_info.req_len = 4
		50	+ w = calculate_demand_workload(PDRole.ROLE_U, req_info)
45	assert w.active_kv_cache > 0	51	assert w.active_kv_cache > 0
46	assert w.active_tokens > 0	52	assert w.active_tokens > 0
47		53
	@@ -55,7 +61,9 @@ class TestCalculateDemandWorkload:
55	# the type checker. The implementation does "else: return Workload()" for	61	# the type checker. The implementation does "else: return Workload()" for
56	# any other role. So we test that for a valid role we get non-empty, and	62	# any other role. So we test that for a valid role we get non-empty, and
57	# we skip testing invalid enum value from outside.	63	# we skip testing invalid enum value from outside.
58	- w = calculate_demand_workload(PDRole.ROLE_P, 0)	64	+ req_info = MagicMock()
		65	+ req_info.req_len = 0
		66	+ w = calculate_demand_workload(PDRole.ROLE_P, req_info)
59	assert w.active_kv_cache >= 0	67	assert w.active_kv_cache >= 0
60	assert w.active_tokens >= 0	68	assert w.active_tokens >= 0
61		69
	@@ -96,8 +104,10 @@ class TestWorkloadActionHandler:
96	async def test_compute_and_update_allocation_success(self, mock_request_manager, valid_resource):	104	async def test_compute_and_update_allocation_success(self, mock_request_manager, valid_resource):
97	"""ALLOCATION: add_req_workload called, returns (workload, role)."""	105	"""ALLOCATION: add_req_workload called, returns (workload, role)."""
98	handler = WorkloadActionHandler(mock_request_manager)	106	handler = WorkloadActionHandler(mock_request_manager)
		107	+ req_info = MagicMock()
		108	+ req_info.req_len = 4
99	workload_change, role = await handler.compute_and_update(	109	workload_change, role = await handler.compute_and_update(
100	- valid_resource, "req-1", WorkloadAction.ALLOCATION, req_len=4	110	+ valid_resource, "req-1", WorkloadAction.ALLOCATION, req_info=req_info
101	)	111	)
102	assert role == PDRole.ROLE_P	112	assert role == PDRole.ROLE_P
103	assert workload_change is not None	113	assert workload_change is not None
	@@ -112,8 +122,10 @@ class TestWorkloadActionHandler:
112	"""ALLOCATION when add_req_workload returns False -> (None, None)."""	122	"""ALLOCATION when add_req_workload returns False -> (None, None)."""
113	mock_request_manager.add_req_workload = AsyncMock(return_value=False)	123	mock_request_manager.add_req_workload = AsyncMock(return_value=False)
114	handler = WorkloadActionHandler(mock_request_manager)	124	handler = WorkloadActionHandler(mock_request_manager)
		125	+ req_info = MagicMock()
		126	+ req_info.req_len = 4
115	workload_change, role = await handler.compute_and_update(	127	workload_change, role = await handler.compute_and_update(
116	- valid_resource, "req-1", WorkloadAction.ALLOCATION, req_len=4	128	+ valid_resource, "req-1", WorkloadAction.ALLOCATION, req_info=req_info
117	)	129	)
118	assert workload_change is None	130	assert workload_change is None
119	assert role is None	131	assert role is None
	@@ -124,8 +136,10 @@ class TestWorkloadActionHandler:
124	current = Workload(active_kv_cache=100.0, active_tokens=50.0)	136	current = Workload(active_kv_cache=100.0, active_tokens=50.0)
125	mock_request_manager.get_req_workload = AsyncMock(return_value=current)	137	mock_request_manager.get_req_workload = AsyncMock(return_value=current)
126	handler = WorkloadActionHandler(mock_request_manager)	138	handler = WorkloadActionHandler(mock_request_manager)
		139	+ req_info = MagicMock()
		140	+ req_info.req_len = 4
127	workload_change, role = await handler.compute_and_update(	141	workload_change, role = await handler.compute_and_update(
128	- valid_resource, "req-1", WorkloadAction.RELEASE_KV, req_len=4	142	+ valid_resource, "req-1", WorkloadAction.RELEASE_KV, req_info=req_info
129	)	143	)
130	assert role == PDRole.ROLE_P	144	assert role == PDRole.ROLE_P
131	assert workload_change is not None	145	assert workload_change is not None
	@@ -139,8 +153,10 @@ class TestWorkloadActionHandler:
139	"""RELEASE_TOKENS when no workload record -> (None, None)."""	153	"""RELEASE_TOKENS when no workload record -> (None, None)."""
140	mock_request_manager.get_req_workload = AsyncMock(return_value=None)	154	mock_request_manager.get_req_workload = AsyncMock(return_value=None)
141	handler = WorkloadActionHandler(mock_request_manager)	155	handler = WorkloadActionHandler(mock_request_manager)
		156	+ req_info = MagicMock()
		157	+ req_info.req_len = 4
142	workload_change, role = await handler.compute_and_update(	158	workload_change, role = await handler.compute_and_update(
143	- valid_resource, "req-1", WorkloadAction.RELEASE_TOKENS, req_len=4	159	+ valid_resource, "req-1", WorkloadAction.RELEASE_TOKENS, req_info=req_info
144	)	160	)
145	assert workload_change is None	161	assert workload_change is None
146	assert role is None	162	assert role is None
	@@ -149,8 +165,10 @@ class TestWorkloadActionHandler:
149	async def test_compute_and_update_invalid_resource_returns_none(self, mock_request_manager):	165	async def test_compute_and_update_invalid_resource_returns_none(self, mock_request_manager):
150	"""Empty or invalid resource -> (None, None)."""	166	"""Empty or invalid resource -> (None, None)."""
151	handler = WorkloadActionHandler(mock_request_manager)	167	handler = WorkloadActionHandler(mock_request_manager)
		168	+ req_info = MagicMock()
		169	+ req_info.req_len = 4
152	workload_change, role = await handler.compute_and_update(	170	workload_change, role = await handler.compute_and_update(
153	- None, "req-1", WorkloadAction.ALLOCATION, req_len=4	171	+ None, "req-1", WorkloadAction.ALLOCATION, req_info=req_info
154	)	172	)
155	assert workload_change is None	173	assert workload_change is None
156	assert role is None	174	assert role is None

Mtests/engine_server/config/test_endpoint.py+8-1

	@@ -28,6 +28,7 @@ from motor.config.endpoint import (
28	PARALLEL_CONFIG_KEY,	28	PARALLEL_CONFIG_KEY,
29	PREFILL_PARALLEL_CONFIG_KEY,	29	PREFILL_PARALLEL_CONFIG_KEY,
30	DECODE_PARALLEL_CONFIG_KEY,	30	DECODE_PARALLEL_CONFIG_KEY,
		31	+ ENCODE_PARALLEL_CONFIG_KEY
31	)	32	)
32	from motor.config.tls_config import TLSConfig	33	from motor.config.tls_config import TLSConfig
33	from motor.engine_server.constants import constants	34	from motor.engine_server.constants import constants
	@@ -81,6 +82,7 @@ def test_model_config_from_dict():
81	"npu_mem_utils": 0.9,	82	"npu_mem_utils": 0.9,
82	PREFILL_PARALLEL_CONFIG_KEY: {"dp_size": 2, "tp_size": 1},	83	PREFILL_PARALLEL_CONFIG_KEY: {"dp_size": 2, "tp_size": 1},
83	DECODE_PARALLEL_CONFIG_KEY: {"dp_size": 4, "tp_size": 2},	84	DECODE_PARALLEL_CONFIG_KEY: {"dp_size": 4, "tp_size": 2},
		85	+ ENCODE_PARALLEL_CONFIG_KEY: {"dp_size": 2, "tp_size": 1},
84	}	86	}
85	config = ModelConfig.from_dict(data)	87	config = ModelConfig.from_dict(data)
86	assert config.model_name == "test-model"	88	assert config.model_name == "test-model"
	@@ -88,7 +90,7 @@ def test_model_config_from_dict():
88	assert config.npu_mem_utils == 0.9	90	assert config.npu_mem_utils == 0.9
89	assert config.prefill_parallel_config.dp_size == 2	91	assert config.prefill_parallel_config.dp_size == 2
90	assert config.decode_parallel_config.dp_size == 4	92	assert config.decode_parallel_config.dp_size == 4
91	-	93	+ assert config.encode_parallel_config.dp_size == 2
92		94
93	# --- EngineConfig tests ---	95	# --- EngineConfig tests ---
94		96
	@@ -154,6 +156,7 @@ def simple_engine_config_file():
154	"npu_mem_utils": 0.9,	156	"npu_mem_utils": 0.9,
155	PREFILL_PARALLEL_CONFIG_KEY: {"dp_size": 2, "tp_size": 1},	157	PREFILL_PARALLEL_CONFIG_KEY: {"dp_size": 2, "tp_size": 1},
156	DECODE_PARALLEL_CONFIG_KEY: {"dp_size": 2, "tp_size": 1},	158	DECODE_PARALLEL_CONFIG_KEY: {"dp_size": 2, "tp_size": 1},
		159	+ ENCODE_PARALLEL_CONFIG_KEY: {"dp_size": 2, "tp_size": 1},
157	},	160	},
158	"engine_config": {"max_model_len": 2048},	161	"engine_config": {"max_model_len": 2048},
159	}	162	}
	@@ -556,10 +559,12 @@ def test_endpoint_config_update_engine_config():
556	"""Test update_engine_config modifies kv-events-config endpoint and replay_endpoint"""	559	"""Test update_engine_config modifies kv-events-config endpoint and replay_endpoint"""
557	prefill = ParallelConfig(dp_size=1, tp_size=1)	560	prefill = ParallelConfig(dp_size=1, tp_size=1)
558	decode = ParallelConfig(dp_size=1, tp_size=1)	561	decode = ParallelConfig(dp_size=1, tp_size=1)
		562	+ encode = ParallelConfig(dp_size=2, tp_size=1)
559	model_config = ModelConfig(	563	model_config = ModelConfig(
560	model_name="m",	564	model_name="m",
561	model_path="/p",	565	model_path="/p",
562	npu_mem_utils=0.9,	566	npu_mem_utils=0.9,
		567	+ encode_parallel_config=encode,
563	prefill_parallel_config=prefill,	568	prefill_parallel_config=prefill,
564	decode_parallel_config=decode,	569	decode_parallel_config=decode,
565	)	570	)
	@@ -595,6 +600,7 @@ def test_endpoint_config_update_engine_config_no_kv_events():
595	model_name="m",	600	model_name="m",
596	model_path="/p",	601	model_path="/p",
597	npu_mem_utils=0.9,	602	npu_mem_utils=0.9,
		603	+ encode_parallel_config=ParallelConfig(),
598	prefill_parallel_config=ParallelConfig(),	604	prefill_parallel_config=ParallelConfig(),
599	decode_parallel_config=ParallelConfig(),	605	decode_parallel_config=ParallelConfig(),
600	),	606	),
	@@ -621,6 +627,7 @@ def test_endpoint_config_update_engine_config_invalid_endpoint_format():
621	model_name="m",	627	model_name="m",
622	model_path="/p",	628	model_path="/p",
623	npu_mem_utils=0.9,	629	npu_mem_utils=0.9,
		630	+ encode_parallel_config=ParallelConfig(),
624	prefill_parallel_config=ParallelConfig(),	631	prefill_parallel_config=ParallelConfig(),
625	decode_parallel_config=ParallelConfig(),	632	decode_parallel_config=ParallelConfig(),
626	),	633	),

	@@ -45,6 +45,7 @@ done
45	sed -i '/^# patch_begin/,/^# patch_end/d' ./startup/boot.sh	45	sed -i '/^# patch_begin/,/^# patch_end/d' ./startup/boot.sh
46	sed -i '/^function set_controller_env()/,/^}/d' ./startup/roles/controller.sh	46	sed -i '/^function set_controller_env()/,/^}/d' ./startup/roles/controller.sh
47	sed -i '/^function set_coordinator_env()/,/^}/d' ./startup/roles/coordinator.sh	47	sed -i '/^function set_coordinator_env()/,/^}/d' ./startup/roles/coordinator.sh
		48	+sed -i '/^function set_encode_env()/,/^}/d' ./startup/roles/engine.sh
48	sed -i '/^function set_prefill_env()/,/^}/d' ./startup/roles/engine.sh	49	sed -i '/^function set_prefill_env()/,/^}/d' ./startup/roles/engine.sh
49	sed -i '/^function set_decode_env()/,/^}/d' ./startup/roles/engine.sh	50	sed -i '/^function set_decode_env()/,/^}/d' ./startup/roles/engine.sh
50	sed -i '/^function set_common_env()/,/^}/d' ./startup/common.sh	51	sed -i '/^function set_common_env()/,/^}/d' ./startup/common.sh
	@@ -52,10 +53,12 @@ sed -i '/^function set_kv_pool_env()/,/^}/d' ./startup/roles/kv_pool.sh
52	sed -i '/^function set_kv_conductor_env()/,/^}/d' ./startup/roles/kv_conductor.sh	53	sed -i '/^function set_kv_conductor_env()/,/^}/d' ./startup/roles/kv_conductor.sh
53	sed -i '/^function set_controller_env()/,/^}/d' ./startup/roles/all_combine_in_single_container.sh	54	sed -i '/^function set_controller_env()/,/^}/d' ./startup/roles/all_combine_in_single_container.sh
54	sed -i '/^function set_coordinator_env()/,/^}/d' ./startup/roles/all_combine_in_single_container.sh	55	sed -i '/^function set_coordinator_env()/,/^}/d' ./startup/roles/all_combine_in_single_container.sh
		56	+sed -i '/^function set_encode_env()/,/^}/d' ./startup/roles/all_combine_in_single_container.sh
55	sed -i '/^function set_prefill_env()/,/^}/d' ./startup/roles/all_combine_in_single_container.sh	57	sed -i '/^function set_prefill_env()/,/^}/d' ./startup/roles/all_combine_in_single_container.sh
56	sed -i '/^function set_decode_env()/,/^}/d' ./startup/roles/all_combine_in_single_container.sh	58	sed -i '/^function set_decode_env()/,/^}/d' ./startup/roles/all_combine_in_single_container.sh
57	sed -i '/^function set_kv_pool_env()/,/^}/d' ./startup/roles/all_combine_in_single_container.sh	59	sed -i '/^function set_kv_pool_env()/,/^}/d' ./startup/roles/all_combine_in_single_container.sh
58	sed -i '/^function set_kv_conductor_env()/,/^}/d' ./startup/roles/all_combine_in_single_container.sh	60	sed -i '/^function set_kv_conductor_env()/,/^}/d' ./startup/roles/all_combine_in_single_container.sh
		61	+sed -i '/^function set_mf_store_env()/,/^}/d' ./startup/roles/mf_store.sh
59	sed -i '/./,$!d' ./startup/common.sh	62	sed -i '/./,$!d' ./startup/common.sh
60		63
61	echo "Delete completed."	64	echo "Delete completed."

	@@ -136,6 +136,7 @@ def set_env_docker(configmap_path):
136	if deploy_mode == "single_container":	136	if deploy_mode == "single_container":
137	update_shell_safely(single_container_shell_path, env_config, "motor_controller_env", "set_controller_env")	137	update_shell_safely(single_container_shell_path, env_config, "motor_controller_env", "set_controller_env")
138	update_shell_safely(single_container_shell_path, env_config, "motor_coordinator_env", "set_coordinator_env")	138	update_shell_safely(single_container_shell_path, env_config, "motor_coordinator_env", "set_coordinator_env")
		139	+ update_shell_safely(single_container_shell_path, env_config, "motor_engine_encode_env", "set_encode_env")
139	update_shell_safely(single_container_shell_path, env_config, "motor_engine_prefill_env", "set_prefill_env")	140	update_shell_safely(single_container_shell_path, env_config, "motor_engine_prefill_env", "set_prefill_env")
140	update_shell_safely(single_container_shell_path, env_config, "motor_engine_decode_env", "set_decode_env")	141	update_shell_safely(single_container_shell_path, env_config, "motor_engine_decode_env", "set_decode_env")
141	update_shell_safely(single_container_shell_path, env_config, "motor_kv_cache_pool_env", "set_kv_pool_env")	142	update_shell_safely(single_container_shell_path, env_config, "motor_kv_cache_pool_env", "set_kv_pool_env")
	@@ -145,6 +146,7 @@ def set_env_docker(configmap_path):
145	else:	146	else:
146	update_shell_safely(controller_shell_path, env_config, "motor_controller_env", "set_controller_env")	147	update_shell_safely(controller_shell_path, env_config, "motor_controller_env", "set_controller_env")
147	update_shell_safely(coordinator_shell_path, env_config, "motor_coordinator_env", "set_coordinator_env")	148	update_shell_safely(coordinator_shell_path, env_config, "motor_coordinator_env", "set_coordinator_env")
		149	+ update_shell_safely(engine_shell_path, env_config, "motor_engine_encode_env", "set_encode_env")
148	update_shell_safely(engine_shell_path, env_config, "motor_engine_prefill_env", "set_prefill_env")	150	update_shell_safely(engine_shell_path, env_config, "motor_engine_prefill_env", "set_prefill_env")
149	update_shell_safely(engine_shell_path, env_config, "motor_engine_decode_env", "set_decode_env")	151	update_shell_safely(engine_shell_path, env_config, "motor_engine_decode_env", "set_decode_env")
150	update_shell_safely(kv_pool_shell_path, env_config, "motor_kv_cache_pool_env", "set_kv_pool_env")	152	update_shell_safely(kv_pool_shell_path, env_config, "motor_kv_cache_pool_env", "set_kv_pool_env")