已关闭
[Usage]: atb-models多机推理过程中使用dist.broadcast_object_list报错！ #167
gbdj创建于 1月26日关闭于 5月25日
gbdj
1月26日 评论：
1月26日 评论：
只要在推理过程中加入dist.broadcast相关的操作就会报错，要么报错要么卡住，最小化复现过程如下，也可参考关联issue，内容一样的。
准备文件：run_pa.py
相比源文件，添加了dist.init_process_group过程与test_infer、test_broadcast_data两个函数以及在main中调用。
# Copyright Huawei Technologies Co., Ltd. 2023-2024. All rights reserved.
import copy
import json
import math
import os
import time
import torch.distributed as dist
import torch
import torch_npu
from atb_llm.runner.model_runner import ModelRunner
from atb_llm.utils.cpu_binding import NpuHbmInfo
from atb_llm.utils.env import ENV
from atb_llm.utils.log import logger, print_log
from atb_llm.utils import file_utils
from atb_llm.models.deepseekv2.eplb.eplb_planner.eplb_worker import EplbWorker
from atb_llm.utils.moe_utils import EPLBType
from atb_llm.utils.argument_utils import ArgumentParser, StringArgumentValidator, \
NumberArgumentValidator, MAX_JSON_LENGTH
from examples.server.cache import CacheConfig, ModelConfig, CacheManager, HiddenStateCache
from examples.server.generate import decode_token, generate_req
from examples.server.request import request_from_token
OPERATOR_BOUND = 576
MAX_KW_ARGS_LENGTH = 4096
PERF_FILE = "./examples/models/performance.csv"
PERF_COLUMNS = "batch, input_len, output_len, embedding_len, first_token_time(ms), \
non_first_token_time(ms), ResponseTime(ms),E2E Throughput Average(Tokens/s)\n"
PRED_FILE = "./examples/models/result.json"
MICROBATCH_SIZE = "microbatch_size"
class PARunner:
def __init__(self, **kwargs):
from datetime import timedelta
dist.init_process_group(
backend="hccl",
world_size=kwargs.get('world_size', '1'),
rank=kwargs.get('rank', '0'),
timeout=timedelta(seconds=1800)
)
logger.info(f"dgi: kwargs:{kwargs}")
self.rank = kwargs.get('rank', '0')
logger.info(f"Initialized HCCL process group for rank {self.rank}")
self.local_rank = kwargs.get('local_rank', self.rank)
self.world_size = kwargs.get('world_size', '1')
self.model_path = kwargs.get('model_path', None)
self.input_text = kwargs.get('input_text', None)
self.max_prefill_batch_size = kwargs.get('max_prefill_batch_size', None)
self.max_batch_size = kwargs.get('max_batch_size', None)
self.max_input_length = kwargs.get('max_input_length', None)
self.max_output_length = kwargs.get('max_output_length', None)
self.max_position_embeddings = kwargs.get('max_position_embeddings', None)
self.max_prefill_tokens = kwargs.get('max_prefill_tokens', None)
self.enable_atb_torch = kwargs.get('enable_atb_torch', False)
self.trust_remote_code = kwargs.get('trust_remote_code', False)
self.block_size = kwargs.get('block_size', None)
self.chat_template = kwargs.get('chat_template', None)
self.load_tokenizer = kwargs.get('load_tokenizer', True)
self.prefill_length = kwargs.get('prefill_length', 8192)
self.distributed_enable = kwargs.get('distributed_enable', False)
self.llm_config_path = kwargs.get('llm_config_path', None)
self.check_limits()
kw_args = kwargs.get('kw_args', '')
if len(kw_args) > MAX_KW_ARGS_LENGTH:
raise ValueError(f'The length of kw_args is too long and should be within (0, {MAX_KW_ARGS_LENGTH}]')
if kw_args:
try:
kw_args = json.loads(kw_args)
except json.decoder.JSONDecodeError:
print_log(self.local_rank, logger.warning, 'Please checkout input of kw_args!')
kw_args = {}
else:
kw_args = {}
if kwargs.get(MICROBATCH_SIZE, -1) == -1:
microbatch_size = self.max_batch_size
elif kwargs.get(MICROBATCH_SIZE, -1) <= self.max_batch_size and kwargs.get(MICROBATCH_SIZE, -1) > 0:
microbatch_size = kwargs.get(MICROBATCH_SIZE, -1)
else:
raise ValueError(f'Micro batch size should be within [1, {self.max_batch_size}] or -1!')
if 'num_speculative_tokens' not in kw_args:
kw_args['num_speculative_tokens'] = int(ENV.deepseek_mtp)
self.model = ModelRunner(
self.model_path, rank=self.rank, world_size=self.world_size,
local_rank=self.local_rank,
max_position_embeddings=self.max_position_embeddings,
enable_atb_torch=self.enable_atb_torch,
load_tokenizer=self.load_tokenizer,
lora_modules=kwargs.get('lora_modules', None),
max_loras=kwargs.get('max_loras', 0),
max_lora_rank=kwargs.get('max_lora_rank', 0),
trust_remote_code=self.trust_remote_code,
llm_config_path=self.llm_config_path,
dp=kwargs.get("dp", -1),
tp=kwargs.get("tp", -1),
moe_tp=kwargs.get("moe_tp", -1),
pp=kwargs.get("pp", -1),
microbatch_size=microbatch_size,
moe_ep=kwargs.get("moe_ep", -1),
sp=kwargs.get("sp", -1),
cp=kwargs.get("cp", -1),
distributed_enable=self.distributed_enable,
max_batch_size=self.max_batch_size,
**kw_args
)
self.tokenizer = self.model.tokenizer
if self.chat_template:
self.tokenizer.chat_template = self._load_chat_template(self.chat_template)
self.dtype = self.model.dtype
self.quantize = self.model.quantize
self.kv_quant_type = self.model.kv_quant_type
self.fa_quant_type = self.model.fa_quant_type
self.model.load_weights(**kw_args)
self.device = self.model.device
self.model_config = ModelConfig(self.model.num_heads,
self.model.num_kv_heads,
self.model.config.num_key_value_heads \
if hasattr(self.model.config, 'num_key_value_heads') \
else self.model.num_kv_heads,
self.model.k_head_size,
self.model.v_head_size,
self.model.num_layers,
self.model.device,
self.model.dtype,
self.model.soc_info,
self.kv_quant_type,
self.fa_quant_type,
self.model.mapping,
self.model.kvcache_quant_layers,
getattr(self.model.config, 'cla_share_factor', 1),
getattr(self.model.config, 'model_type', None),
self.model.enable_nz)
self.max_memory = NpuHbmInfo.get_hbm_capacity(self.local_rank, self.world_size, self.model.soc_info.need_nz)
self.init_memory = int(
self.max_memory * NpuHbmInfo.get_hbm_usage(self.local_rank, self.world_size, self.model.soc_info.need_nz))
print_log(self.rank, logger.info, f'hbm_capacity(GB): {self.max_memory / (1024 ** 3)}, '
f'init_memory(GB): {self.init_memory / (1024 ** 3)}')
self.warm_up_memory = 0
self.warm_up_num_blocks = 0
self.cache_manager = None
self.hidden_state_cache = None
self.compress_head_enable = ENV.compress_head_enable
self.omni_attention_enable = ENV.omni_attention_enable
self.eplb_forwarder = None
eplb_level = getattr(self.model.model, "eplb_level", 0)
if eplb_level == EPLBType.DYNAMIC_EPLB:
self.eplb_worker = EplbWorker(self.model, self.rank, self.model_path, self.device)
self.eplb_forwarder = self.eplb_worker.eplb_forwarder
def test_broadcast_data(self):
if self.rank != 0:
data = [None]
else:
data = [1]
logger.info("before broadcast")
dist.broadcast_object_list(data, src=0)
logger.info("after broadcast")
def __repr__(self):
return (
"PARunner("
+ f"model_path={self.model_path}, "
+ f"input_text={self.input_text}, "
+ f"max_position_embeddings={self.max_position_embeddings}, "
+ f"max_input_length={self.max_input_length}, "
+ f"max_output_length={self.max_output_length}, "
+ f"max_prefill_tokens={self.max_prefill_tokens}, "
+ f"load_tokenizer={self.load_tokenizer}, "
+ f"enable_atb_torch={self.enable_atb_torch}, "
+ f"max_prefill_batch_size={self.max_prefill_batch_size}, "
+ f"max_batch_size={self.max_batch_size}, "
+ f"dtype={self.dtype}, "
+ f"block_size={self.block_size}, "
+ f"model_config={self.model_config}, "
+ f"max_memory={self.max_memory}, "
)
@staticmethod
def save_input_output_ids(req_list):
for idx, req in enumerate(req_list):
input_ids_save_filename = f"input_ids_{idx}.pth"
input_ids_save_path = os.path.join(ENV.token_ids_save_folder, input_ids_save_filename)
input_ids_save_path = file_utils.standardize_path(input_ids_save_path)
file_utils.check_file_safety(input_ids_save_path, 'w', is_check_file_size=False)
torch.save(req.input_ids.cpu(), input_ids_save_path)
os.chmod(input_ids_save_path, 0o640)
output_ids_save_filename = f"output_ids_{idx}.txt"
output_ids_save_path = os.path.join(ENV.token_ids_save_folder, output_ids_save_filename)
with file_utils.safe_open(output_ids_save_path, 'w', encoding='utf-8') as f:
f.write(' '.join(map(str, req.out_token_list)))
@staticmethod
def _update_inputs_items(inputs):
input_texts, input_ids, input_conversations, input_adapters = [], [], [], []
if isinstance(inputs, list) and inputs:
if isinstance(inputs[0], str):
input_texts = inputs
elif isinstance(inputs[0], torch.Tensor):
input_ids = inputs
elif isinstance(inputs[0], list) and inputs[0]:
if isinstance(inputs[0][0], int):
input_ids = inputs
elif isinstance(inputs[0][0], dict):
input_conversations = inputs
elif isinstance(inputs[0], dict):
prompt_key = "prompt"
if isinstance(inputs[0].get(prompt_key), torch.Tensor):
input_ids = [input.get(prompt_key) for input in inputs]
else:
input_texts = [input.get(prompt_key) for input in inputs]
input_adapters = [input.get("adapter") for input in inputs]
return input_texts, input_ids, input_conversations, input_adapters
@staticmethod
def _load_chat_template(chat_template: str):
if os.path.exists(chat_template):
with file_utils.safe_open(chat_template, 'r', encoding='utf-8') as f:
chat_template_content = f.read()
else:
chat_template_content = chat_template
return chat_template_content
def check_limits(self):
if self.max_input_length <= 0:
print_log(self.rank, logger.warning,
f"'max_input_length' must be greater than 0, but is {self.max_input_length}. "
"Please increase the value of 'max_input_length'. ")
if self.max_output_length <= 0:
print_log(self.rank, logger.warning,
f"'max_output_length' must be greater than 0, but is {self.max_output_length}. "
"Please increase the value of 'max_output_length'. ")
if self.max_position_embeddings and \
self.max_position_embeddings < self.max_input_length + self.max_output_length:
print_log(self.rank, logger.warning,
f"'max_position_embeddings' must be no less than "
f"{self.max_input_length + self.max_output_length}, but is {self.max_position_embeddings}. "
"Please increase the value of 'max_position_embeddings'. ")
if self.max_prefill_batch_size and self.max_prefill_batch_size <= 0:
print_log(self.rank, logger.warning,
f"'max_prefill_batch_size' must be greater than 0, but is {self.max_prefill_batch_size}. "
"Please increase the value of 'max_prefill_batch_size'. ")
if self.max_batch_size <= 0:
print_log(self.rank, logger.warning,
f"'max_batch_size' must be greater than 0, but is {self.max_batch_size}. "
"Please increase the value of 'max_batch_size'. ")
if self.block_size > OPERATOR_BOUND:
print_log(self.rank, logger.warning,
f"'block_size' must be lower than or equal to 576, but is {self.block_size}. "
"Please decrease the value of 'block_size'. ")
if self.block_size <= 0:
print_log(self.rank, logger.warning,
f"'block_size' is too small: {self.block_size}. "
"Please increase the value of 'block_size'. ")
if self.max_prefill_tokens != -1 and \
self.max_prefill_tokens < self.max_input_length:
print_log(self.rank, logger.warning,
f"'max_prefill_tokens' must be greater than or equal to {self.max_input_length}, "
f"but is {self.max_prefill_tokens}. Please increse the value of 'max_prefill_tokens'. ")
def calc_max_length(self):
if self.model.mapping.has_attn_cp() and self.model.mapping.has_attn_inner_sp():
max_length = math.ceil((math.ceil(self.max_input_length / self.model.mapping.attn_cp.group_size) +
self.max_output_length) / self.model.mapping.attn_inner_sp.group_size)
elif self.model.mapping.has_attn_inner_sp():
max_length = math.ceil((self.max_input_length + self.max_output_length) /
self.model.mapping.attn_inner_sp.group_size)
elif self.model.mapping.has_attn_cp():
# CP will perform padding on the input.
max_length = (math.ceil(self.max_input_length / self.model.mapping.attn_cp.group_size) +
math.ceil(self.max_output_length / self.model.mapping.attn_cp.group_size))
else:
max_length = self.max_input_length + self.max_output_length
return max_length
def warm_up(self):
self.model.model.warmup_is_end = False
if self.max_prefill_batch_size is None:
self.max_prefill_batch_size = self.max_batch_size
if ENV.modeltest_pd_split_enable:
if self.max_input_length > self.prefill_length:
self.max_prefill_batch_size = 1
else:
self.max_prefill_batch_size = math.floor(self.prefill_length / self.max_input_length)
if self.max_prefill_batch_size > self.max_batch_size:
self.max_prefill_batch_size = self.max_batch_size
effective_max_prefill_batch_size = math.ceil(
self.max_prefill_batch_size / self.model.mapping.attn_dp.group_size)
effective_max_batch_size = math.ceil(
self.max_batch_size / self.model.mapping.attn_dp.group_size)
if self.max_prefill_tokens == -1:
self.max_prefill_tokens = effective_max_prefill_batch_size * (self.max_input_length +
self.max_output_length)
print_log(self.rank, logger.info, "---------------begin warm_up---------------")
try:
max_length = self.calc_max_length()
self.warm_up_num_blocks = math.ceil(max_length / self.block_size) * effective_max_batch_size
except ZeroDivisionError as e:
raise ZeroDivisionError from e
cache_config = CacheConfig(self.warm_up_num_blocks, self.block_size)
if self.compress_head_enable or self.omni_attention_enable:
cache_config = CacheConfig(self.warm_up_num_blocks, self.block_size, \
self.max_input_length, self.max_output_length, effective_max_batch_size, self.rank, self.world_size)
self.cache_manager = CacheManager(cache_config, self.model_config)
if ENV.deepseek_mtp:
batch_size_for_cache = max(self.max_batch_size, self.max_prefill_batch_size)
self.cache_manager.hidden_state_cache = HiddenStateCache(self.model.config.hidden_size,
ENV.deepseek_mtp, batch_size_for_cache, self.model.dtype, self.cache_manager.device)
else:
self.hidden_state_cache = None
self.model.postprocessor.max_new_tokens = 1
input_ids = torch.ones(self.max_input_length, dtype=torch.int64)
req_list = [
request_from_token(input_ids, 1, self.block_size, req_idx=idx, mapping=self.model.mapping)
for idx in range(self.max_prefill_batch_size)
]
generate_req(req_list, self.model, self.max_batch_size, self.max_prefill_tokens, self.cache_manager,
self.eplb_forwarder)
self.warm_up_memory = int(
self.max_memory * NpuHbmInfo.get_hbm_usage(self.local_rank, self.world_size, self.model.soc_info.need_nz))
print_log(self.rank, logger.info, f'warmup_memory(GB): {self.warm_up_memory / (1024 ** 3): .2f}')
print_log(self.rank, logger.info, "---------------end warm_up---------------")
self.model.model.warmup_is_end = True
def infer(self, inputs, batch_size, max_output_length, ignore_eos, is_chat_model=False, **kwargs):
if torch.distributed.is_initialized():
logger.info(">>>> torch.distributed is already initialized.")
else:
logger.error(">>>> torch.distributed has not been initialized!")
print_log(self.rank, logger.info, "---------------begin inference---------------")
if ignore_eos:
self.model.postprocessor.eos_token_id = []
is_truncation = kwargs.get("truncation", False)
input_ids, adapter_ids = self._build_model_inputs(inputs, is_chat_model, is_truncation)
if len(input_ids) == 1:
req_list = [request_from_token(input_ids[0], max_output_length, self.block_size, req_idx=idx,
adapter_id=adapter_ids[0], mapping=self.model.mapping,
postprocessor=self.model.postprocessor)
for idx in range(batch_size)]
else:
req_list = [request_from_token(input_ids_ins, max_output_length, self.block_size, req_idx=idx,
adapter_id=adapter_id, mapping=self.model.mapping, postprocessor=self.model.postprocessor)
for idx, (input_ids_ins, adapter_id) in enumerate(zip(input_ids, adapter_ids))]
print_log(self.rank, logger.debug, f'req_list[0].input_ids: {req_list[0].input_ids}')
if not self.cache_manager:
if self.max_prefill_tokens == -1:
self.max_prefill_tokens = self.max_batch_size * (self.max_input_length + self.max_output_length)
cache_block_size = self.block_size * self.model.num_kv_heads * self.model.head_size
dtype_size = CacheManager.get_dtype_size(self.dtype)
total_cache_size = self.model.num_layers * cache_block_size * 2 * dtype_size
max_memory = ENV.memory_fraction * self.max_memory
free_memory = max_memory - ENV.reserved_memory_gb * (1 << 30) - (
self.warm_up_memory if self.warm_up_memory != 0 else self.init_memory)
print_log(self.rank, logger.info,
f"infer max_memory(GB): {max_memory / (1024 ** 3): .2f}, "
f"warm_up_memory(GB): {self.warm_up_memory / (1024 ** 3): .2f}, "
f"free_memory(GB): {free_memory / (1024 ** 3): .2f}")
try:
num_blocks = int(free_memory // total_cache_size)
except ZeroDivisionError as e:
raise ZeroDivisionError("total_cache_size should not be 0") from e
print_log(self.rank, logger.info, f"num_blocks: {num_blocks}, free_memory: {free_memory}")
cache_config = CacheConfig(num_blocks, self.block_size)
if self.compress_head_enable:
cache_config = CacheConfig(self.warm_up_num_blocks, self.block_size, \
self.max_input_length, self.max_output_length, self.max_batch_size, self.rank, self.world_size)
self.cache_manager = CacheManager(cache_config, self.model_config)
if ENV.benchmark_enable:
req_list_dummy = copy.deepcopy(req_list)
self.model.postprocessor.max_new_tokens = 2
generate_req(req_list_dummy, self.model, self.max_batch_size, self.max_prefill_tokens, self.cache_manager,
self.eplb_forwarder)
self.model.postprocessor.max_new_tokens = max_output_length
skip_special_tokens = kwargs.get("skip_special_tokens", False)
if not ENV.profiling_enable:
print_log(self.rank, logger.debug, "No profiling")
torch.npu.synchronize()
e2e_start = time.time()
generate_req(req_list, self.model, self.max_batch_size, self.max_prefill_tokens, self.cache_manager,
self.eplb_forwarder)
_, _ = decode_token(req_list, self.tokenizer, skip_special_tokens)
torch.npu.synchronize()
e2e_end = time.time()
e2e_time = e2e_end - e2e_start
else:
print_log(self.rank, logger.debug, "Enter profiling")
profiling_path = ENV.profiling_filepath
profiler_level = torch_npu.profiler.ProfilerLevel
actual_profiler_level = getattr(profiler_level, ENV.profiling_level)
torch.npu.synchronize()
e2e_start = time.time()
experimental_config = torch_npu.profiler._ExperimentalConfig(
aic_metrics=torch_npu.profiler.AiCMetrics.PipeUtilization,
profiler_level=actual_profiler_level,
l2_cache=False,
data_simplification=False
)
with torch_npu.profiler.profile(
activities=[
torch_npu.profiler.ProfilerActivity.CPU,
torch_npu.profiler.ProfilerActivity.NPU
],
on_trace_ready=torch_npu.profiler.tensorboard_trace_handler(profiling_path),
record_shapes=True,
profile_memory=True,
with_stack=False,
with_flops=False,
with_modules=False,
experimental_config=experimental_config):
generate_req(req_list, self.model, self.max_batch_size, self.max_prefill_tokens, self.cache_manager,
self.eplb_forwarder)
torch.npu.synchronize()
e2e_end = time.time()
e2e_time = e2e_end - e2e_start
generate_text_list, token_num_list = decode_token(req_list, self.tokenizer, skip_special_tokens)
if ENV.modeltest_dataset_specified:
ENV.update()
if ENV.token_ids_save_enable:
if self.model.mapping.has_pp():
if self.rank == self.world_size - 1:
self.save_input_output_ids(req_list)
else:
if self.local_rank == 0:
self.save_input_output_ids(req_list)
print_log(self.rank, logger.info, "---------------end inference---------------")
return generate_text_list, token_num_list, e2e_time
def shutdown(self):
eplb_level = getattr(self.model.config, "eplb_level", 0)
if eplb_level == EPLBType.DYNAMIC_EPLB:
self.eplb_worker.eplb_loader.shutdown()
self.eplb_worker.eplb_planner.shutdown()
def _build_model_inputs(self, inputs, is_chat_model, is_truncation=False):
input_texts, input_ids, input_conversations, input_adapters = self._update_inputs_items(inputs)
if not (input_texts or input_ids or input_conversations):
raise ValueError(f"The inputs of `PARunner.infer` must be as List[str], List[torch.Tensor], List[List[int]]"
f" or List[List[Dict]]. Now the inputs ({inputs}) is not acceptable or is empty.")
if not input_ids and not self.tokenizer:
raise ValueError("Can‘t encode input_texts, because load_tokenizer=False, "
"please enter input_ids or enable load_tokenizer")
if is_chat_model:
if input_conversations:
input_ids = self.model.build_inputs(input_conversations)
elif input_texts:
input_conversations = [[{"role": "user", "content": t}] for t in input_texts]
input_ids = self.model.build_inputs(input_conversations)
else:
print_log(self.rank, logger.warning, "Neither conversations nor input_texts exist, "
"'chat' parameter is not effective.")
elif input_texts:
input_ids = [self.tokenizer([text], return_tensors="pt", truncation=is_truncation)["input_ids"].flatten()
for text in input_texts]
# token长度校验
for item in input_ids:
if len(item) > self.max_input_length:
print_log(self.rank, logger.warning,
"Num of tokens in input_text is larger than max_input_length. "
"Please shorten input_text to avoid out of memory.")
if not input_adapters:
input_adapters = [None] * len(input_ids)
return input_ids, input_adapters
def cmd_bool(cmd_arg):
if cmd_arg == "True":
return True
elif cmd_arg == "False":
return False
raise ValueError(f"{cmd_arg} should be a boolean")
def parse_list_of_json(list_json):
return json.loads(list_json)
def parse_ids(list_str):
return [int(item) for item in list_str.split(',')]
def input_texts_parser(value):
if os.path.isfile(value):
with file_utils.safe_open(value, 'r') as opened_file:
return opened_file.read()
else:
return value
#define Argument Parser
def parse_arguments():
store_true = 'store_true'
parser = ArgumentParser()
parser.add_argument('--model_path', type=str, help="model and tokenizer path")
parser.add_argument(
'--input_texts',
type=input_texts_parser,
nargs='+',
default=["What's deep learning?"])
parser.add_argument(
'--input_ids',
type=parse_ids,
nargs='+',
default=None)
parser.add_argument(
'--input_file',
type=str,
help='This parameter is used to input multi-turn dialogue information in the form '
'of a jsonl file, with each line in the format of a List[Dict]. Each dictionary '
'(Dict) must contain at least two fields: "role" and "content".',
default=None)
parser.add_argument(
'--input_dict',
help="Lora input, accepted format: "
"'[{\"prompt\": \"prompt in text\", \"adapter\": \"adapter id defined in lora_adapter param\"}]'",
type=parse_list_of_json,
default=None)
parser.add_argument("--max_prefill_batch_size", type=int, default=None)
parser.add_argument("--max_batch_size", type=int, default=1)
parser.add_argument('--max_input_length', type=int, default=1024)
parser.add_argument('--max_output_length', type=int, default=20)
parser.add_argument('--max_position_embeddings', type=int, default=None)
parser.add_argument('--max_prefill_tokens', type=int, default=-1)
parser.add_argument("--block_size", type=int, default=128)
parser.add_argument('--chat_template', type=str, default=None)
parser.add_argument('--ignore_eos', action=store_true)
parser.add_argument('--is_chat_model', action=store_true)
parser.add_argument('--is_embedding_model', action=store_true)
parser.add_argument('--load_tokenizer', type=cmd_bool, default=True)
parser.add_argument('--enable_atb_torch', action=store_true)
parser.add_argument('--trust_remote_code', action=store_true)
parser.add_argument('--dp', type=int, default=-1)
parser.add_argument('--tp', type=int, default=-1)
parser.add_argument('--moe_tp', type=int, default=-1)
parser.add_argument('--pp', type=int, default=-1)
parser.add_argument('--microbatch_size', type=int, default=-1)
parser.add_argument('--moe_ep', type=int, default=-1)
parser.add_argument('--sp', type=int, default=-1)
parser.add_argument('--cp', type=int, default=-1)
parser.add_argument('--kw_args', type=str, default='', help='json input')
parser.add_argument('--lora_modules', type=str, default=None, help="Define lora modules, accepted format:"
"'{\"adapter1\": \"/path/to/lora1\", \"adapter2\": \"/path/to/lora2\"}'",
validator=StringArgumentValidator(min_length=2, max_length=MAX_JSON_LENGTH, allow_none=True))
parser.add_argument('--max_loras', type=int, default=0, help="Define the MAX number of loRAs to store in NPU memory",
validator=NumberArgumentValidator(min_value=0))
parser.add_argument('--max_lora_rank', type=int, default=0, help="Define the MAX loRA rank",
validator=NumberArgumentValidator(min_value=0))
parser.add_argument('--distributed_enable', action=store_true)
return parser.parse_args()
def test_infer(pa_runner):
infer_params = {
"inputs": ["What's deep learning?"],
"batch_size": 1,
"max_output_length": 10,
"ignore_eos": True,
"is_chat_model": True
}
generate_texts, token_nums, _ = pa_runner.infer(**infer_params)
length = len(infer_inputs)
for i, generate_text in enumerate(generate_texts):
if input_dict['is_embedding_model']:
embedding_tensor_path = f"{os.getcwd()}/examples/embedding_tensor"
print_log(rank, logger.info, f"Context[{i}]: \nembedding tensor path is: {embedding_tensor_path}")
else:
print_log(rank, logger.info, f'Answer[{i}]: {generate_text}')
print_log(rank, logger.info, f'Generate[{i}] token num: {token_nums[i]}')
if __name__ == '__main__':
args = parse_arguments()
rank = ENV.rank
local_rank = ENV.local_rank
world_size = ENV.world_size
input_dict = {
'rank': rank,
'world_size': world_size,
'local_rank': local_rank,
**vars(args)
}
# 输入优先级input_dict > input_ids > input_texts
if args.input_dict:
infer_inputs = args.input_dict
elif args.input_ids:
infer_inputs = args.input_ids
else:
infer_inputs = args.input_texts
if args.is_chat_model and args.input_file:
conversations = []
with file_utils.safe_open(args.input_file, 'r', encoding='utf-8') as file:
for line in file_utils.safe_readlines(file):
data_line = json.loads(line)
conversations.append(data_line)
infer_inputs = conversations
pa_runner = PARunner(**input_dict)
print_log(rank, logger.info, f'pa_runner: {pa_runner}')
# pa_runner.test_broadcast_data()
print_log(rank, logger.info, f'before_test_infer')
test_infer(pa_runner)
test_infer(pa_runner)
print_log(rank, logger.info, f'after_test_infer')
print_log(rank, logger.info, f'before_test_infer1')
test_infer(pa_runner)
print_log(rank, logger.info, f'after_test_infer1')
# exit(0)
pa_runner.warm_up()
infer_params = {
"inputs": infer_inputs,
"batch_size": args.max_batch_size,
"max_output_length": args.max_output_length,
"ignore_eos": args.ignore_eos,
"is_chat_model": args.is_chat_model
}
generate_texts, token_nums, _ = pa_runner.infer(**infer_params)
length = len(infer_inputs)
for i, generate_text in enumerate(generate_texts):
if input_dict['is_embedding_model']:
embedding_tensor_path = f"{os.getcwd()}/examples/embedding_tensor"
print_log(rank, logger.info, f"Context[{i}]: \nembedding tensor path is: {embedding_tensor_path}")
else:
print_log(rank, logger.info, f'Answer[{i}]: {generate_text}')
print_log(rank, logger.info, f'Generate[{i}] token num: {token_nums[i]}')
# 下面的命令两台主机都要执行,注意：1、run_pa的挂载（容器内修改文件也可）2.rank_table_file的挂载
docker run -it \
--name dgi-dev-mindie \
--network=host \
--privileged=true \
--device=/dev/davinci_manager \
--device=/dev/hisi_hdc \
--device=/dev/devmm_svm \
-v /usr/local/lib/node_modules/:/usr/local/lib/node_modules/:ro \
-v /usr/local/bin/node:/usr/local/bin/node:ro \
-v /usr/local/dcmi:/usr/local/dcmi \
-v /usr/local/bin/npu-smi:/usr/local/bin/npu-smi \
-v /usr/local/Ascend/driver/lib64/common:/usr/local/Ascend/driver/lib64/common \
-v /usr/local/Ascend/driver/lib64/driver:/usr/local/Ascend/driver/lib64/driver \
-v /usr/local/Ascend/driver/version.info:/usr/local/Ascend/driver/version.info \
-v /etc/ascend_install.info:/etc/ascend_install.info \
-v /etc/vnpu.cfg:/etc/vnpu.cfg \
-v /usr/local/sbin:/usr/local/sbin \
-v /usr/share/zoneinfo/Asia/Shanghai:/etc/localtime \
-v /data1/:/data1/ \
-v /data2/:/data2/ \
-v /data1/MindIE_Auto_Deploy_Tool/rank_table_file.json:/data1/MindIE_Auto_Deploy_Tool/rank_table_file.json \
-v /data1/atb-models/examples/run_pa.py:/usr/local/Ascend/atb-models/examples/run_pa.py \
-w /root \
-u root \
--shm-size=64GB \
--ulimit msgqueue=-1 \
-e ASCEND_RT_VISIBLE_DEVICES="0,1,2,3,4,5,6,7" \
swr.cn-south-1.myhuaweicloud.com/ascendhub/mindie:2.2.RC1-800I-A2-py311-openeuler24.03-lts
# 进容器后执行下面的命令
source /usr/local/Ascend/ascend-toolkit/set_env.sh
source /usr/local/Ascend/nnal/atb/set_env.sh
source /usr/local/Ascend/atb-models/set_env.sh
export ATB_LLM_HCCL_ENABLE=1
export ATB_LLM_COMM_BACKEND="hccl"
export HCCL_CONNECT_TIMEOUT=7200
export HCCL_EXEC_TIMEOUT=0
export ASCEND_RT_VISIBLE_DEVICES=0,1,2,3,4,5,6,7
export ATB_LLM_BENCHMARK_ENABLE=1
export ATB_LLM_ENABLE_AUTO_TRANSPOSE=0
# 打开 ATB 日志
# export ASCEND_GLOBAL_LOG_LEVEL=3
# export ASCEND_SLOG_PRINT_TO_STDOUT=1
# MINDIE日志
export MINDIE_LOG_TO_STDOUT=1
export HCCL_DETERMINISTIC=false
export LCCL_DETERMINISTIC=0
export HCCL_BUFFSIZE=120
export ATB_WORKSPACE_MEM_ALLOC_GLOBAL=1
# 使用ranktable！
export RANK_TABLE_FILE=/data1/MindIE_Auto_Deploy_Tool/rank_table_file.json
首先是使用rank_table推理，不用broadcast
主机执行：
# 主机执行
torchrun \
--nproc_per_node 8 \
--master_port 12345 \
--nnodes 2 \
--node_rank 0 \
--master_addr 10.17.44.64 \
/usr/local/Ascend/atb-models/examples/run_pa.py \
--model_path /data1/models/Qwen3-30B-A3B-Instruct-2507 2>&1 | tee ~/master.log
从机执行：
# 从机执行
torchrun \
--nproc_per_node 8 \
--master_port 12345 \
--nnodes 2 \
--node_rank 1 \
--master_addr 10.17.44.64 \
/usr/local/Ascend/atb-models/examples/run_pa.py \
--model_path /data1/models/Qwen3-30B-A3B-Instruct-2507 2>&1 | tee ~/slave.log
2026-01-25 19:55启动
日志中可能报警告：
[2026-01-25 19:55:53.460882] [833] [833] [llmmodels] [WARN] [context_factory.cpp:37] Failed to aclInit: 100002
但结果正常推理，输出
[2026-01-25 19:56:42,324] [829] [281473611860544] [llmmodels] [INFO] [run_pa.py-652] : Answer[0]: What's the difference between deep learning and machine learning? What's the difference between deep learning and neural
[2026-01-25 19:56:42,324] [829] [281473611860544] [llmmodels] [INFO] [run_pa.py-653] : Generate[0] token num: (0, 20)
使用rank_table推理，推理前加入broadcast
两台容器同时进vim /usr/local/Ascend/atb-models/examples/run_pa.py +626
取消注释pa_runner.test_broadcast_data()
主机执行：
# 主机执行
torchrun \
--nproc_per_node 8 \
--master_port 12345 \
--nnodes 2 \
--node_rank 0 \
--master_addr 10.17.44.64 \
/usr/local/Ascend/atb-models/examples/run_pa.py \
--model_path /data1/models/Qwen3-30B-A3B-Instruct-2507 2>&1 | tee ~/master1.log
从机执行：
# 从机执行
torchrun \
--nproc_per_node 8 \
--master_port 12345 \
--nnodes 2 \
--node_rank 1 \
--master_addr 10.17.44.64 \
/usr/local/Ascend/atb-models/examples/run_pa.py \
--model_path /data1/models/Qwen3-30B-A3B-Instruct-2507 2>&1 | tee ~/slave1.log
2026-01-25 20:01启动，报错：
[2026-01-25 20:02:05,455] [3754] [281473078336064] [llmmodels] [INFO] [cache.py-154] : kv cache will allocate 53.3173828125GB memory
/usr/local/lib64/python3.11/site-packages/torch_npu/utils/storage.py:41: UserWarning: TypedStorage is deprecated. It will be removed in the future and UntypedStorage will be the only storage class. This should only matter to you if you are using storages directly. To access UntypedStorage directly, use tensor.untyped_storage() instead of tensor.storage()
if self.device.type != 'cpu':
[2026-01-25 20:02:05,521] [3754] [281473078336064] [llmmodels] [INFO] [generate.py-1130] : ------total req num: 1, infer start--------
[2026-01-25 20:02:05,521] [3758] [281473476311616] [llmmodels] [INFO] [run_pa.py-163] : before broadcast
[2026-01-25 20:02:06,782] [3755] [281473268972096] [llmmodels] [INFO] [run_pa.py-163] : before broadcast
[2026-01-25 20:02:07,495] [3760] [281473129114176] [llmmodels] [INFO] [run_pa.py-163] : before broadcast
[2026-01-25 20:02:08,468] [3761] [281472942824000] [llmmodels] [INFO] [run_pa.py-163] : before broadcast
[2026-01-25 20:02:08,474] [3757] [281473277147712] [llmmodels] [INFO] [run_pa.py-163] : before broadcast
[2026-01-25 20:02:10,166] [3759] [281473160243776] [llmmodels] [INFO] [run_pa.py-163] : before broadcast
[2026-01-25 20:02:13,227] [3756] [281473215760960] [llmmodels] [INFO] [run_pa.py-163] : before broadcast
[2026-01-25 20:02:15,449] [3759] [281473160243776] [llmmodels] [INFO] [run_pa.py-165] : after broadcast
[2026-01-25 20:02:15,449] [3755] [281473268972096] [llmmodels] [INFO] [run_pa.py-165] : after broadcast
[2026-01-25 20:02:15,449] [3758] [281473476311616] [llmmodels] [INFO] [run_pa.py-165] : after broadcast
[2026-01-25 20:02:15,450] [3759] [281473160243776] [llmmodels] [INFO] [run_pa.py-342] : >>>> torch.distributed is already initialized.
[2026-01-25 20:02:15,450] [3760] [281473129114176] [llmmodels] [INFO] [run_pa.py-165] : after broadcast
[2026-01-25 20:02:15,450] [3755] [281473268972096] [llmmodels] [INFO] [run_pa.py-342] : >>>> torch.distributed is already initialized.
[2026-01-25 20:02:15,450] [3758] [281473476311616] [llmmodels] [INFO] [run_pa.py-342] : >>>> torch.distributed is already initialized.
[2026-01-25 20:02:15,450] [3760] [281473129114176] [llmmodels] [INFO] [run_pa.py-342] : >>>> torch.distributed is already initialized.
[2026-01-25 20:02:15,450] [3757] [281473277147712] [llmmodels] [INFO] [run_pa.py-165] : after broadcast
[2026-01-25 20:02:15,450] [3757] [281473277147712] [llmmodels] [INFO] [run_pa.py-342] : >>>> torch.distributed is already initialized.
[2026-01-25 20:02:15,454] [3756] [281473215760960] [llmmodels] [INFO] [run_pa.py-165] : after broadcast
[2026-01-25 20:02:15,454] [3756] [281473215760960] [llmmodels] [INFO] [run_pa.py-342] : >>>> torch.distributed is already initialized.
[2026-01-25 20:02:15,455] [3761] [281472942824000] [llmmodels] [INFO] [run_pa.py-165] : after broadcast
[2026-01-25 20:02:15,455] [3761] [281472942824000] [llmmodels] [INFO] [run_pa.py-342] : >>>> torch.distributed is already initialized.
[2026-01-25 20:02:15,553] [3759] [281473160243776] [llmmodels] [INFO] [cache.py-154] : kv cache will allocate 53.3173828125GB memory
[2026-01-25 20:02:15,555] [3755] [281473268972096] [llmmodels] [INFO] [cache.py-154] : kv cache will allocate 53.3173828125GB memory
[2026-01-25 20:02:15,556] [3761] [281472942824000] [llmmodels] [INFO] [cache.py-154] : kv cache will allocate 53.3173828125GB memory
[2026-01-25 20:02:15,562] [3758] [281473476311616] [llmmodels] [INFO] [cache.py-154] : kv cache will allocate 53.3173828125GB memory
[2026-01-25 20:02:15,564] [3756] [281473215760960] [llmmodels] [INFO] [cache.py-154] : kv cache will allocate 53.3173828125GB memory
[2026-01-25 20:02:15,565] [3757] [281473277147712] [llmmodels] [INFO] [cache.py-154] : kv cache will allocate 53.3173828125GB memory
[2026-01-25 20:02:15,567] [3760] [281473129114176] [llmmodels] [INFO] [cache.py-154] : kv cache will allocate 53.3173828125GB memory
mki_log mkdir /root/ascend/log/atb
Traceback (most recent call last):
File "/usr/local/Ascend/atb-models/examples/run_pa.py", line 628, in <module>
test_infer(pa_runner)
File "/usr/local/Ascend/atb-models/examples/run_pa.py", line 584, in test_infer
generate_texts, token_nums, _ = pa_runner.infer(**infer_params)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/Ascend/atb-models/examples/run_pa.py", line 390, in infer
generate_req(req_list_dummy, self.model, self.max_batch_size, self.max_prefill_tokens, self.cache_manager,
File "/usr/local/Ascend/atb-models/examples/server/generate.py", line 1218, in generate_req
generate_token_with_clocking(model, cache_manager, batch, eplb_forwarder)
File "/usr/local/Ascend/atb-models/examples/server/generate.py", line 892, in generate_token_with_clocking
res = generate_token(model, cache_manager, input_batch_in, eplb_forwarder)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/Ascend/atb-models/examples/server/generate.py", line 616, in generate_token
logits = model.forward(
^^^^^^^^^^^^^^
File "/usr/local/Ascend/atb-models/atb_llm/runner/model_runner.py", line 324, in forward
res = self.model.forward(**kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/Ascend/atb-models/atb_llm/models/qwen2_moe/flash_causal_qwen2_moe.py", line 438, in forward
logits = super().forward(**kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/Ascend/atb-models/atb_llm/models/base/flash_causal_lm.py", line 519, in forward
self.init_ascend_weight()
File "/usr/local/Ascend/atb-models/atb_llm/models/qwen2_moe/flash_causal_qwen2_moe.py", line 311, in init_ascend_weight
self.acl_encoder_operation.set_param(self.acl_param_encoder)
RuntimeError: External Comm Manager: Create the hccl communication group failed. export ASCEND_GLOBAL_LOG_LEVEL=3, export ASCEND_SLOG_PRINT_TO_STDOUT=1 to see more details. Default log path is $HOME/atb/log.
Traceback (most recent call last):
File "/usr/local/Ascend/atb-models/examples/run_pa.py", line 628, in <module>
test_infer(pa_runner)
File "/usr/local/Ascend/atb-models/examples/run_pa.py", line 584, in test_infer
generate_texts, token_nums, _ = pa_runner.infer(**infer_params)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/Ascend/atb-models/examples/run_pa.py", line 390, in infer
generate_req(req_list_dummy, self.model, self.max_batch_size, self.max_prefill_tokens, self.cache_manager,
File "/usr/local/Ascend/atb-models/examples/server/generate.py", line 1218, in generate_req
generate_token_with_clocking(model, cache_manager, batch, eplb_forwarder)
File "/usr/local/Ascend/atb-models/examples/server/generate.py", line 892, in generate_token_with_clocking
res = generate_token(model, cache_manager, input_batch_in, eplb_forwarder)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/Ascend/atb-models/examples/server/generate.py", line 616, in generate_token
logits = model.forward(
^^^^^^^^^^^^^^
File "/usr/local/Ascend/atb-models/atb_llm/runner/model_runner.py", line 324, in forward
res = self.model.forward(**kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/Ascend/atb-models/atb_llm/models/qwen2_moe/flash_causal_qwen2_moe.py", line 438, in forward
logits = super().forward(**kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/Ascend/atb-models/atb_llm/models/base/flash_causal_lm.py", line 519, in forward
self.init_ascend_weight()
File "/usr/local/Ascend/atb-models/atb_llm/models/qwen2_moe/flash_causal_qwen2_moe.py", line 311, in init_ascend_weight
self.acl_encoder_operation.set_param(self.acl_param_encoder)
RuntimeError: External Comm Manager: Create the hccl communication group failed. export ASCEND_GLOBAL_LOG_LEVEL=3, export ASCEND_SLOG_PRINT_TO_STDOUT=1 to see more details. Default log path is $HOME/atb/log.
使用rank_table推理，推理后尝试broadcast：
两台容器同时进vim /usr/local/Ascend/atb-models/examples/run_pa.py +626
将626行的pa_runner.test_broadcast_data()移动628行到两个test_infer之间，像是这样:
624 pa_runner = PARunner(**input_dict)
625 print_log(rank, logger.info, f'pa_runner: {pa_runner}')
626 print_log(rank, logger.info, f'before_test_infer')
627 test_infer(pa_runner)
628 pa_runner.test_broadcast_data()
629 test_infer(pa_runner)
630 print_log(rank, logger.info, f'after_test_infer')
主机执行：
# 主机执行
torchrun \
--nproc_per_node 8 \
--master_port 12345 \
--nnodes 2 \
--node_rank 0 \
--master_addr 10.17.44.64 \
/usr/local/Ascend/atb-models/examples/run_pa.py \
--model_path /data1/models/Qwen3-30B-A3B-Instruct-2507 2>&1 | tee ~/master2.log
从机执行：
# 从机执行
torchrun \
--nproc_per_node 8 \
--master_port 12345 \
--nnodes 2 \
--node_rank 1 \
--master_addr 10.17.44.64 \
/usr/local/Ascend/atb-models/examples/run_pa.py \
--model_path /data1/models/Qwen3-30B-A3B-Instruct-2507 2>&1 | tee ~/slave2.log
2026-01-25 20:08 启动，第一个test_infer正常，程序卡死在before_broadcast与after_broadcast之间，日志例如（从机，主机同样）：等了十分钟还是没响应，ctrl+c停掉了。
[2026-01-25 20:09:56,791] [5677] [281473705536064] [llmmodels] [INFO] [run_pa.py-163] : before broadcast
[2026-01-25 20:09:56,792] [5676] [281472899271232] [llmmodels] [INFO] [run_pa.py-163] : before broadcast
[2026-01-25 20:09:56,792] [5678] [281473714686528] [llmmodels] [INFO] [run_pa.py-163] : before broadcast
[2026-01-25 20:09:56,792] [5672] [281473198320192] [llmmodels] [INFO] [run_pa.py-163] : before broadcast
[2026-01-25 20:09:56,792] [5674] [281472934701632] [llmmodels] [INFO] [run_pa.py-163] : before broadcast
[2026-01-25 20:09:56,792] [5675] [281473338534464] [llmmodels] [INFO] [run_pa.py-163] : before broadcast
[2026-01-25 20:09:56,792] [5673] [281473130969664] [llmmodels] [INFO] [run_pa.py-163] : before broadcast
[2026-01-25 20:09:56,792] [5679] [281473051490880] [llmmodels] [INFO] [run_pa.py-163] : before broadcast
在上一个的基础上不使用rank_table推理
紧接上一个例子，两个test_infer之间使用broadcast，并且unset RANK_TABLE_FILE变量
unset RANK_TABLE_FILE
主机执行：
# 主机执行
torchrun \
--nproc_per_node 8 \
--master_port 12345 \
--nnodes 2 \
--node_rank 0 \
--master_addr 10.17.44.64 \
/usr/local/Ascend/atb-models/examples/run_pa.py \
--model_path /data1/models/Qwen3-30B-A3B-Instruct-2507 2>&1 | tee ~/master3.log
从机执行：
# 从机执行
torchrun \
--nproc_per_node 8 \
--master_port 12345 \
--nnodes 2 \
--node_rank 1 \
--master_addr 10.17.44.64 \
/usr/local/Ascend/atb-models/examples/run_pa.py \
--model_path /data1/models/Qwen3-30B-A3B-Instruct-2507 2>&1 | tee ~/slave3.log
2026-01-25 20:20启动，报错变成了：
[2026-01-25 20:21:21,470] [7981] [281473698683456] [llmmodels] [INFO] [generate.py-1130] : ------total req num: 1, infer start--------
[2026-01-25 20:21:21,969] [7985] [281473459296832] [llmmodels] [INFO] [run_pa.py-342] : >>>> torch.distributed is already initialized.
[2026-01-25 20:21:22,102] [7985] [281473459296832] [llmmodels] [INFO] [cache.py-154] : kv cache will allocate 53.3173828125GB memory
[2026-01-25 20:21:22.214388] [7981] [7981] [llmmodels] [INFO] [moe_decoder_model.cpp:211] model enableSplitFuse: 0
Traceback (most recent call last):
File "/usr/local/Ascend/atb-models/examples/run_pa.py", line 627, in <module>
test_infer(pa_runner)
File "/usr/local/Ascend/atb-models/examples/run_pa.py", line 584, in test_infer
generate_texts, token_nums, _ = pa_runner.infer(**infer_params)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/Ascend/atb-models/examples/run_pa.py", line 390, in infer
generate_req(req_list_dummy, self.model, self.max_batch_size, self.max_prefill_tokens, self.cache_manager,
File "/usr/local/Ascend/atb-models/examples/server/generate.py", line 1218, in generate_req
generate_token_with_clocking(model, cache_manager, batch, eplb_forwarder)
File "/usr/local/Ascend/atb-models/examples/server/generate.py", line 892, in generate_token_with_clocking
res = generate_token(model, cache_manager, input_batch_in, eplb_forwarder)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/Ascend/atb-models/examples/server/generate.py", line 616, in generate_token
logits = model.forward(
^^^^^^^^^^^^^^
File "/usr/local/Ascend/atb-models/atb_llm/runner/model_runner.py", line 324, in forward
res = self.model.forward(**kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/Ascend/atb-models/atb_llm/models/qwen2_moe/flash_causal_qwen2_moe.py", line 438, in forward
logits = super().forward(**kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/Ascend/atb-models/atb_llm/models/base/flash_causal_lm.py", line 519, in forward
self.init_ascend_weight()
File "/usr/local/Ascend/atb-models/atb_llm/models/qwen2_moe/flash_causal_qwen2_moe.py", line 311, in init_ascend_weight
self.acl_encoder_operation.set_param(self.acl_param_encoder)
RuntimeError: Build model graph failed. enable log: export ASCEND_SLOG_PRINT_TO_STDOUT=1, export ASCEND_MODULE_LOG_LEVEL=ATB=3 to find the first error. For more details, see the MindIE official document.
总结：rank_table设置且不用dist.broadcast_object_list的时候一切正常，只要使用了broadcast_object_list，要么forward报异常，要么直接给卡死。不使用ranktable的时候也是直接报错
可能的原因：
1、双机 Asend HDK驱动版本不一致，主机 是 24.1.rc3; 从机 是 25.3.rc1;查看mindie官方的文档（2.2.RCX文档显示这两个驱动都是支持的）
2、torch_npu 的生态问题。torch 集合通信接口与底层的 HCCL 接口没有适配好。从而导致 torch 的 broadcast 与算子通信互相冲突；
容器中~下的所有日志文件我已经打包好，参考下面的附件。如果可以的话希望能够帮忙分析一下。
master_log.tar.gz
slave_log.tar.gz
1月26日 添加了label：usage
1月29日 关联了看板：@zeshengzong的看板 20260129
吴铭泾
3月9日 评论：
3月9日 评论：
收到反馈，感谢！我会尽快查看并回复进展。
gbdj
3月11日 评论：
3月11日 评论：
收到反馈，感谢！我会尽快查看并回复进展。
感谢您的回复，想确认下这个是否属于bug还是我的调用方式有问题？我只是想使用broadcast_objectlist做多机数据同步同时不影响模型推理，目前的情况下，有什么好的解决方式？期待您的回复！
gbdj
4月1日 评论：
4月1日 评论：
@wumingjing 您好，这个问题有什么进展吗？？？
吴铭泾
5月25日 评论：
5月25日 评论：
1、根据当前plog的报错，显示是有“hccl multiple initialization”，对应表示同一个hccl通信域初始化了2次。怀疑有残留进程没杀干净。
不同的HDK版本在底层通信能力的支持上存在区别，无法完全保证能兼容的。 推荐升级至同样的驱动版本进行测试验证，并保证残留进程杀干净。
5月25日 关闭了 issue
5月25日 重新打开了 issue
5月25日 issue状态由 TODO 改变为 DONE
5月25日 关闭了 issue
在提交新问题之前，请确保您已经在社区中搜索过相关问题，并使用了社区中提供的资源/工具后，仍未找到满意的解决方式。
环境信息
使用场景及问题
关联issue:https://gitcode.com/Ascend/pytorch/issues/1520
欢迎加入社区，感谢您对社区的贡献 🎉!