[feature] EPD特性补充DT&适配静态扩缩容-MindIE-Motor-AtomGit

Mexamples/deployer/lib/config_validator.py+3-1

	@@ -79,6 +79,7 @@ def resolve_config_paths(config_dir, user_config_path, env_config_path):
79		79
80	def strip_instance_nums(config_dict):	80	def strip_instance_nums(config_dict):
81	cleaned = json.loads(json.dumps(config_dict))	81	cleaned = json.loads(json.dumps(config_dict))
		82	+ cleaned["motor_deploy_config"].pop(C.E_INSTANCES_NUM, None)
82	cleaned["motor_deploy_config"].pop(C.P_INSTANCES_NUM, None)	83	cleaned["motor_deploy_config"].pop(C.P_INSTANCES_NUM, None)
83	cleaned["motor_deploy_config"].pop(C.D_INSTANCES_NUM, None)	84	cleaned["motor_deploy_config"].pop(C.D_INSTANCES_NUM, None)
84	cleaned["motor_deploy_config"].pop(C.HYBRID_INSTANCES_NUM, None)	85	cleaned["motor_deploy_config"].pop(C.HYBRID_INSTANCES_NUM, None)
	@@ -88,7 +89,8 @@ def strip_instance_nums(config_dict):
88	def validate_only_instance_changed(current_config, baseline_config):	89	def validate_only_instance_changed(current_config, baseline_config):
89	if strip_instance_nums(current_config) != strip_instance_nums(baseline_config):	90	if strip_instance_nums(current_config) != strip_instance_nums(baseline_config):
90	raise ValueError("user_config changes detected beyond instance numbers. "	91	raise ValueError("user_config changes detected beyond instance numbers. "
91	- "Only p_instances_num/d_instances_num/hybrid_instances_num can be modified for scaling.")	92	+ "Only e_instances_num/p_instances_num/d_instances_num/hybrid_instances_num "
		93	+ "can be modified for scaling.")
92		94
93		95
94	def validate_deploy_mode_consistency(deploy_config, baseline_config):	96	def validate_deploy_mode_consistency(deploy_config, baseline_config):

Mexamples/deployer/lib/generator/k8s_utils.py+21-0

	@@ -335,6 +335,25 @@ def scale_engine_by_type(deploy_config, baseline_deploy_config, out_deploy_yaml_
335	yaml_path = os.path.join(out_deploy_yaml_path, f"{g_engine_base_name}_{node_type}{index}.yaml")	335	yaml_path = os.path.join(out_deploy_yaml_path, f"{g_engine_base_name}_{node_type}{index}.yaml")
336	safe_exec_cmd(f"kubectl apply -f {yaml_path} -n {job_id}")	336	safe_exec_cmd(f"kubectl apply -f {yaml_path} -n {job_id}")
337		337
		338	+def scale_engine_e_by_type(deploy_config, baseline_deploy_config, out_deploy_yaml_path):
		339	+ """Scale engine instances by type (p, d or u)."""
		340	+ from lib.generator.engine import obtain_engine_e_instance_total
		341	+
		342	+ job_id = deploy_config[C.CONFIG_JOB_ID]
		343	+ total = obtain_engine_e_instance_total(deploy_config)
		344	+ base = obtain_engine_e_instance_total(baseline_deploy_config)
		345	+ if total < base:
		346	+ logger.info(f"Scale-in {C.NODE_TYPE_E} instance, {base} -> {total}")
		347	+ for index in reversed(range(total, base)):
		348	+ yaml_path = os.path.join(out_deploy_yaml_path, f"{g_engine_base_name}_{C.NODE_TYPE_E}{index}.yaml")
		349	+ safe_exec_cmd(f"kubectl delete -f {yaml_path} -n {job_id}")
		350	+ if os.path.exists(yaml_path):
		351	+ os.remove(yaml_path)
		352	+ if total > base:
		353	+ logger.info(f"Scale-out {C.NODE_TYPE_E} instance, {base} -> {total}")
		354	+ for index in range(base, total):
		355	+ yaml_path = os.path.join(out_deploy_yaml_path, f"{g_engine_base_name}_{C.NODE_TYPE_E}{index}.yaml")
		356	+ safe_exec_cmd(f"kubectl apply -f {yaml_path} -n {job_id}")
338		357
339	def elastic_distributed_engine_deploy(deploy_config, baseline_deploy_config, out_deploy_yaml_path):	358	def elastic_distributed_engine_deploy(deploy_config, baseline_deploy_config, out_deploy_yaml_path):
340	"""Elastic distributed engine deployment - scale in/out engine instances."""	359	"""Elastic distributed engine deployment - scale in/out engine instances."""
	@@ -345,6 +364,8 @@ def elastic_distributed_engine_deploy(deploy_config, baseline_deploy_config, out
345		364
346	scale_engine_by_type(deploy_config, baseline_deploy_config, out_deploy_yaml_path, C.NODE_TYPE_P)	365	scale_engine_by_type(deploy_config, baseline_deploy_config, out_deploy_yaml_path, C.NODE_TYPE_P)
347	scale_engine_by_type(deploy_config, baseline_deploy_config, out_deploy_yaml_path, C.NODE_TYPE_D)	366	scale_engine_by_type(deploy_config, baseline_deploy_config, out_deploy_yaml_path, C.NODE_TYPE_D)
		367	+ if C.E_INSTANCES_NUM in deploy_config:
		368	+ scale_engine_e_by_type(deploy_config, baseline_deploy_config, out_deploy_yaml_path)
348	logger.info("Engine scale done.")	369	logger.info("Engine scale done.")
349		370
350		371

Atests/common/utils/test_image_utils.py+103-0

@@ -0,0 +1,103 @@
	1	+# Copyright (c) Huawei Technologies Co., Ltd. 2026. All rights reserved.
	2	+# MindIE is licensed under Mulan PSL v2.
	3	+# You can use this software according to the terms and conditions of the Mulan PSL v2.
	4	+# You may obtain a copy of Mulan PSL v2 at:
	5	+# http://license.coscl.org.cn/MulanPSL2
	6	+# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
	7	+# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
	8	+# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
	9	+# See the Mulan PSL v2 for more details.
	10	+
	11	+import base64
	12	+import io
	13	+import os
	14	+import tempfile
	15	+
	16	+import pytest
	17	+from PIL import Image
	18	+
	19	+from motor.common.utils import image_utils
	20	+
	21	+
	22	+def _create_image_bytes(fmt: str, size=(100, 80)) -> bytes:
	23	+ """Create in-memory image bytes for a given format and size."""
	24	+ img = Image.new("RGB", size, color=(10, 20, 30))
	25	+ with io.BytesIO() as buf:
	26	+ img.save(buf, format=fmt)
	27	+ return buf.getvalue()
	28	+
	29	+
	30	+def test_parse_png_size():
	31	+ """Test that PNG size can be parsed from the image header."""
	32	+ w, h = 123, 45
	33	+ data = _create_image_bytes("PNG", size=(w, h))
	34	+ got_w, got_h = image_utils.parse_png_size(data)
	35	+ assert (got_w, got_h) == (w, h)
	36	+
	37	+
	38	+def test_parse_jpeg_size():
	39	+ """Test that JPEG size can be parsed from the image header."""
	40	+ w, h = 77, 99
	41	+ data = _create_image_bytes("JPEG", size=(w, h))
	42	+ got_w, got_h = image_utils.parse_jpeg_size(data)
	43	+ assert (got_w, got_h) == (w, h)
	44	+
	45	+
	46	+def test_parse_jpeg_size_invalid_raises():
	47	+ """Test that invalid JPEG data raises a ValueError."""
	48	+ with pytest.raises(ValueError):
	49	+ image_utils.parse_jpeg_size(b"not a jpeg")
	50	+
	51	+
	52	+def test_fast_get_hw_with_base64():
	53	+ """Test fast_get_hw by decoding a base64 data URI and reading dimensions via PIL."""
	54	+ w, h = 64, 48
	55	+ data = _create_image_bytes("JPEG", size=(w, h))
	56	+ b64 = base64.b64encode(data).decode("ascii")
	57	+ uri = f"data:image/jpeg;base64,{b64}"
	58	+ got_w, got_h = image_utils.fast_get_hw(uri)
	59	+ assert (got_w, got_h) == (w, h)
	60	+
	61	+
	62	+def test_get_hw_from_local_png_and_file_uri():
	63	+ """Test get_hw_from_local for PNG files and file:// URI paths."""
	64	+ w, h = 31, 65
	65	+ data = _create_image_bytes("PNG", size=(w, h))
	66	+ # write to a temp file
	67	+ fd, path = tempfile.mkstemp(suffix=".png")
	68	+ os.close(fd)
	69	+ try:
	70	+ with open(path, "wb") as f:
	71	+ f.write(data)
	72	+
	73	+ got_w, got_h = image_utils.get_hw_from_local(path)
	74	+ assert (got_w, got_h) == (w, h)
	75	+
	76	+ # file:// scheme
	77	+ got_w2, got_h2 = image_utils.get_hw_from_local("file://" + path)
	78	+ assert (got_w2, got_h2) == (w, h)
	79	+ finally:
	80	+ os.remove(path)
	81	+
	82	+
	83	+def test_get_mul_token_for_base64_and_local():
	84	+ """Test get_mul_token for both base64 and local image inputs."""
	85	+ w, h = 100, 100
	86	+ data = _create_image_bytes("JPEG", size=(w, h))
	87	+ # base64
	88	+ b64 = base64.b64encode(data).decode("ascii")
	89	+ uri = f"data:image/jpeg;base64,{b64}"
	90	+ mul = image_utils.get_mul_token(uri)
	91	+ expected = ( (h + 31) // 32 ) * ( (w + 31) // 32 )
	92	+ assert mul == expected
	93	+
	94	+ # local file
	95	+ fd, path = tempfile.mkstemp(suffix=".jpg")
	96	+ os.close(fd)
	97	+ try:
	98	+ with open(path, "wb") as f:
	99	+ f.write(data)
	100	+ mul2 = image_utils.get_mul_token(path)
	101	+ assert mul2 == expected
	102	+ finally:
	103	+ os.remove(path)

Mtests/coordinator/core/test_scheduler.py+76-3

	@@ -82,6 +82,23 @@ def mix_instances():
82	return instances	82	return instances
83		83
84		84
		85	+@pytest.fixture
		86	+def encode_instances():
		87	+ """Create encode (E) instances for testing."""
		88	+ instances = []
		89	+ for i in range(2):
		90	+ instance = Instance(
		91	+ job_name=f"encode_instance_{i+1}",
		92	+ model_name="test_model",
		93	+ id=i+8,
		94	+ role=PDRole.ROLE_E,
		95	+ status=InsStatus.ACTIVE,
		96	+ parallel_config=ParallelConfig(dp_size=2)
		97	+ )
		98	+ instances.append(instance)
		99	+ return instances
		100	+
		101	+
85	def mock_create_client(address, tls_config=None, **kwargs):	102	def mock_create_client(address, tls_config=None, **kwargs):
86	client = AsyncMock()	103	client = AsyncMock()
87	client.base_url = f"http://{address}"	104	client.base_url = f"http://{address}"
	@@ -92,7 +109,7 @@ def mock_create_client(address, tls_config=None, **kwargs):
92		109
93		110
94	@pytest.fixture	111	@pytest.fixture
95	-async def scheduler_setup(prefill_instances, decode_instances, mix_instances):	112	+async def scheduler_setup(prefill_instances, decode_instances, mix_instances, encode_instances):
96	"""Setup scheduler with instances and endpoints."""	113	"""Setup scheduler with instances and endpoints."""
97	config = CoordinatorConfig()	114	config = CoordinatorConfig()
98	instance_manager = InstanceManager(config)	115	instance_manager = InstanceManager(config)
	@@ -102,7 +119,7 @@ async def scheduler_setup(prefill_instances, decode_instances, mix_instances):
102	if all_existing_instances:	119	if all_existing_instances:
103	await instance_manager.refresh_instances(EventType.DEL, all_existing_instances)	120	await instance_manager.refresh_instances(EventType.DEL, all_existing_instances)
104		121
105	- all_instances = prefill_instances + decode_instances + mix_instances	122	+ all_instances = prefill_instances + decode_instances + mix_instances + encode_instances
106	await instance_manager.refresh_instances(EventType.DEL, all_instances)	123	await instance_manager.refresh_instances(EventType.DEL, all_instances)
107	for instance in all_instances:	124	for instance in all_instances:
108	endpoints = {}	125	endpoints = {}
	@@ -122,7 +139,7 @@ async def scheduler_setup(prefill_instances, decode_instances, mix_instances):
122	await instance_manager.refresh_instances(EventType.ADD, all_instances)	139	await instance_manager.refresh_instances(EventType.ADD, all_instances)
123		140
124	# Fail fast if pool was not populated (e.g. CI missing asyncio_mode or different impl)	141	# Fail fast if pool was not populated (e.g. CI missing asyncio_mode or different impl)
125	- for role in (PDRole.ROLE_P, PDRole.ROLE_D):	142	+ for role in (PDRole.ROLE_P, PDRole.ROLE_D, PDRole.ROLE_E):
126	pool = instance_manager.get_available_instances(role)	143	pool = instance_manager.get_available_instances(role)
127	assert len(pool) > 0, (	144	assert len(pool) > 0, (
128	f"scheduler_setup: get_available_instances({role}) is empty after ADD. "	145	f"scheduler_setup: get_available_instances({role}) is empty after ADD. "
	@@ -213,6 +230,42 @@ async def test_request_processing_pd_separation_scenario(scheduler_setup):
213	assert selected_prefill_endpoint.workload.active_kv_cache == 0	230	assert selected_prefill_endpoint.workload.active_kv_cache == 0
214		231
215		232
		233	+@pytest.mark.asyncio
		234	+async def test_request_processing_e_scenario(scheduler_setup):
		235	+ """Test E (encode) role processing similar to P/D scenarios."""
		236	+ all_instances, instance_manager = scheduler_setup
		237	+ scheduler = Scheduler(instance_provider=instance_manager, config=SchedulerType.LOAD_BALANCE)
		238	+ load_balance_scheduler = scheduler.get_scheduling_policy()
		239	+ request_length = 2
		240	+ req_id = "test_request_e_1"
		241	+
		242	+ # select encode instance and endpoint
		243	+ res = await scheduler.select_instance_and_endpoint(role=PDRole.ROLE_E)
		244	+ assert res is not None, "select_instance_and_endpoint(ROLE_E) returned None."
		245	+ selected_instance, selected_endpoint = res
		246	+ assert selected_instance.role == PDRole.ROLE_E
		247	+
		248	+ # allocate encode workload
		249	+ req_info = MagicMock()
		250	+ req_info.req_len = request_length
		251	+ workload_e = calculate_demand_workload(PDRole.ROLE_E, req_info)
		252	+ result = await load_balance_scheduler.update_workload(
		253	+ selected_instance.id, selected_endpoint.id, req_id,
		254	+ WorkloadAction.ALLOCATION, workload_e
		255	+ )
		256	+ assert result
		257	+ assert selected_endpoint.workload.active_tokens > 0 or selected_endpoint.workload.active_kv_cache >= 0
		258	+
		259	+ # release tokens if any allocated
		260	+ release_tokens = Workload(active_tokens=-selected_endpoint.workload.active_tokens)
		261	+ result = await load_balance_scheduler.update_workload(
		262	+ selected_instance.id, selected_endpoint.id, req_id,
		263	+ WorkloadAction.RELEASE_TOKENS, release_tokens
		264	+ )
		265	+ assert result
		266	+ assert selected_endpoint.workload.active_tokens == 0
		267	+
		268	+
216	@pytest.mark.asyncio	269	@pytest.mark.asyncio
217	async def test_request_processing_mix_scenario(scheduler_setup):	270	async def test_request_processing_mix_scenario(scheduler_setup):
218	"""Test mixed role scenario with load balance policy."""	271	"""Test mixed role scenario with load balance policy."""
	@@ -378,6 +431,13 @@ async def test_load_balance_policy_selection_logic(scheduler_setup):
378	assert mix_instance is not None	431	assert mix_instance is not None
379	assert mix_instance.role == PDRole.ROLE_U	432	assert mix_instance.role == PDRole.ROLE_U
380		433
		434	+ # also verify encode (E) selection
		435	+ res_e = await scheduler.select_instance_and_endpoint(role=PDRole.ROLE_E)
		436	+ assert res_e is not None, "select_instance_and_endpoint(ROLE_E) returned None."
		437	+ encode_instance, _ = res_e
		438	+ assert encode_instance is not None
		439	+ assert encode_instance.role == PDRole.ROLE_E
		440	+
381	res_p2 = await scheduler.select_instance_and_endpoint(role=PDRole.ROLE_P)	441	res_p2 = await scheduler.select_instance_and_endpoint(role=PDRole.ROLE_P)
382	assert res_p2 is not None, "select_instance_and_endpoint(ROLE_P) returned None."	442	assert res_p2 is not None, "select_instance_and_endpoint(ROLE_P) returned None."
383	_, endpoint = res_p2	443	_, endpoint = res_p2
	@@ -415,6 +475,17 @@ async def test_round_robin_instance_selection(scheduler_setup):
415	expected_decode_order = [4, 5, 4, 5]	475	expected_decode_order = [4, 5, 4, 5]
416	assert selected_decode_instances == expected_decode_order	476	assert selected_decode_instances == expected_decode_order
417		477
		478	+ # select 4 times for encode instances (ids 8,9)
		479	+ selected_encode_instances = []
		480	+ for _ in range(4):
		481	+ instance, _ = await scheduler.select_instance_and_endpoint(role=PDRole.ROLE_E)
		482	+ assert instance is not None
		483	+ assert instance.role == PDRole.ROLE_E
		484	+ selected_encode_instances.append(instance.id)
		485	+
		486	+ expected_encode_order = [8, 9, 8, 9]
		487	+ assert selected_encode_instances == expected_encode_order
		488	+
418		489
419	@pytest.mark.asyncio	490	@pytest.mark.asyncio
420	async def test_round_robin_endpoint_selection(scheduler_setup):	491	async def test_round_robin_endpoint_selection(scheduler_setup):
	@@ -489,3 +560,5 @@ async def test_round_robin_edge_cases():
489		560
490	result = await empty_scheduler.select_instance_and_endpoint(role=PDRole.ROLE_D)	561	result = await empty_scheduler.select_instance_and_endpoint(role=PDRole.ROLE_D)
491	assert result is None	562	assert result is None
		563	+ result = await empty_scheduler.select_instance_and_endpoint(role=PDRole.ROLE_E)
		564	+ assert result is None

Mtests/engine_server/config/test_endpoint.py+76-0

	@@ -284,6 +284,40 @@ def test_deploy_config_load_with_role_union(pd_hybrid_engine_config_file):
284	assert config.model_config.decode_parallel_config == config.model_config.prefill_parallel_config	284	assert config.model_config.decode_parallel_config == config.model_config.prefill_parallel_config
285		285
286		286
		287	+@pytest.fixture
		288	+def encode_engine_config_file():
		289	+ """Create a temporary JSON file with encode role config section."""
		290	+ config = {
		291	+ "motor_deploy_config": {},
		292	+ "motor_engine_encode_config": {
		293	+ "engine_type": "vllm",
		294	+ "model_config": {
		295	+ "model_name": "test-model-encode",
		296	+ "model_path": "/path/to/encode_model",
		297	+ "npu_mem_utils": 0.9,
		298	+ "parallel_config": {"dp_size": 2, "tp_size": 1},
		299	+ },
		300	+ "engine_config": {"max_model_len": 2048},
		301	+ }
		302	+ }
		303	+ with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
		304	+ json.dump(config, f)
		305	+ temp_path = os.path.realpath(f.name)
		306	+ yield temp_path
		307	+ try:
		308	+ os.unlink(temp_path)
		309	+ except FileNotFoundError:
		310	+ pass
		311	+
		312	+
		313	+def test_deploy_config_load_with_role_encode(encode_engine_config_file):
		314	+ """Test DeployConfig.load with role=encode."""
		315	+ config = DeployConfig.load(encode_engine_config_file, role="encode")
		316	+ assert config.engine_type == "vllm"
		317	+ assert config.model_config.model_name == "test-model-encode"
		318	+ assert config.model_config.encode_parallel_config.dp_size == 2
		319	+
		320	+
287	def test_deploy_config_load_with_health_check(simple_engine_config_file):	321	def test_deploy_config_load_with_health_check(simple_engine_config_file):
288	"""Test DeployConfig.load includes health_check_config"""	322	"""Test DeployConfig.load includes health_check_config"""
289	with open(simple_engine_config_file) as f:	323	with open(simple_engine_config_file) as f:
	@@ -315,6 +349,14 @@ def test_deploy_config_get_parallel_config_decode(pd_engine_config_file):
315	assert parallel == config.model_config.decode_parallel_config	349	assert parallel == config.model_config.decode_parallel_config
316		350
317		351
		352	+def test_deploy_config_get_parallel_config_encode(simple_engine_config_file):
		353	+ """Test DeployConfig.get_parallel_config for encode role"""
		354	+ config = DeployConfig.load(simple_engine_config_file, role="encode")
		355	+ parallel = config.get_parallel_config("encode")
		356	+ assert parallel == config.model_config.encode_parallel_config
		357	+ assert parallel.dp_size == 2
		358	+
		359	+
318	def test_deploy_config_get_parallel_config_invalid_role(simple_engine_config_file):	360	def test_deploy_config_get_parallel_config_invalid_role(simple_engine_config_file):
319	"""Test DeployConfig.get_parallel_config raises for invalid role"""	361	"""Test DeployConfig.get_parallel_config raises for invalid role"""
320	config = DeployConfig.load(simple_engine_config_file)	362	config = DeployConfig.load(simple_engine_config_file)
	@@ -606,6 +648,40 @@ def test_endpoint_config_load_deploy_config_updates_dp_rpc_port_decode(valid_con
606	os.unlink(path)	648	os.unlink(path)
607		649
608		650
		651	+def test_endpoint_config_load_deploy_config_updates_dp_rpc_port_encode():
		652	+ """Test load_deploy_config updates dp_rpc_port for encode role"""
		653	+ encode_config = {
		654	+ "motor_deploy_config": {},
		655	+ "motor_engine_encode_config": {
		656	+ "engine_type": "vllm",
		657	+ "model_config": {
		658	+ "model_name": "m",
		659	+ "model_path": "/p",
		660	+ "npu_mem_utils": 0.9,
		661	+ "parallel_config": {"dp_size": 1, "dp_rpc_port": 9000},
		662	+ },
		663	+ "engine_config": {},
		664	+ }
		665	+ }
		666	+ with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
		667	+ json.dump(encode_config, f)
		668	+ path = os.path.realpath(f.name)
		669	+ try:
		670	+ config = EndpointConfig(
		671	+ host="127.0.0.1",
		672	+ role="encode",
		673	+ port=8000,
		674	+ mgmt_port=9001,
		675	+ config_path=path,
		676	+ dp_rpc_port=9020,
		677	+ )
		678	+ config.deploy_config = DeployConfig.load(path, role="encode")
		679	+ config.load_deploy_config()
		680	+ assert config.deploy_config.model_config.encode_parallel_config.dp_rpc_port == 9020
		681	+ finally:
		682	+ os.unlink(path)
		683	+
		684	+
609	def test_endpoint_config_update_engine_config():	685	def test_endpoint_config_update_engine_config():
610	"""Test update_engine_config modifies kv-events-config endpoint and replay_endpoint"""	686	"""Test update_engine_config modifies kv-events-config endpoint and replay_endpoint"""
611	prefill = ParallelConfig(dp_size=1, tp_size=1)	687	prefill = ParallelConfig(dp_size=1, tp_size=1)