Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/dttest/api/python/llm_datadist_v1/config.py: 54%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# ------------------------------------------------------------------- 

4# ----------------------------------------------------------------------------------------------------------- 

5# Copyright (c) 2025 Huawei Technologies Co., Ltd. 

6# This program is free software, you can redistribute it and/or modify it under the terms and conditions of 

7# CANN Open Software License Agreement Version 2.0 (the "License"). 

8# Please refer to the License for details. You may not use this file except in compliance with the License. 

9# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, 

10# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. 

11# See LICENSE in the root of the software repository for the full text of the License. 

12# ----------------------------------------------------------------------------------------------------------- 

13 

14import json 

15import os 

16from typing import Dict 

17 

18from .configs import LlmConfig, LLMRole, trans_str_ip 

19from .status import raise_if_false 

20from .utils import log 

21from .utils.utils import check_isinstance 

22 

23 

24class ClusterConfig: 

25 def __init__(self, rank_id_to_device_id: Dict[int, int]): 

26 self.rank_id_to_device_id = rank_id_to_device_id 

27 

28 @classmethod 

29 def from_engine_options(cls, engine_options: Dict[str, str]) -> "ClusterConfig": 

30 logical_device_id_to_rank_id = cls._parse_rank_mapping(engine_options) 

31 logical_device_id_to_device_id = cls._parse_numa_config(engine_options) 

32 local_rank_ids = [] 

33 if "ge.exec.rankId" in engine_options: 

34 rank_id = int(engine_options["ge.exec.rankId"]) 

35 log.info(f"get rank_id by option ge.exec.rankId={rank_id}") 

36 local_rank_ids.append(rank_id) 

37 else: 

38 for logical_device_id in logical_device_id_to_device_id.keys(): 

39 if logical_device_id in logical_device_id_to_rank_id: 

40 rank_id = logical_device_id_to_rank_id[logical_device_id] 

41 log.info(f"append rank_id = {rank_id}, logical_device_id = {logical_device_id}") 

42 local_rank_ids.append(rank_id) 

43 rank_id_to_device_id: Dict[int, int] = {} 

44 for logical_device_id, rank_id in logical_device_id_to_rank_id.items(): 

45 if rank_id in local_rank_ids: 

46 device_id = logical_device_id_to_device_id[logical_device_id] 

47 rank_id_to_device_id[rank_id] = device_id 

48 raise_if_false( 

49 len(rank_id_to_device_id) > 0, 

50 f"rank_id_to_device_id is empty, " 

51 f"logical_device_id_to_rank_id = {logical_device_id_to_rank_id}, " 

52 f"logical_device_id_to_device_id = {logical_device_id_to_device_id}", 

53 ) 

54 return cls(rank_id_to_device_id) 

55 

56 @staticmethod 

57 def _parse_rank_mapping(engine_options: Dict[str, str]) -> Dict[str, int]: 

58 raise_if_false( 

59 "llm.ClusterInfo" in engine_options, 

60 "option 'llm.ClusterInfo' is not defined", 

61 ) 

62 cluster_info = json.loads(engine_options["llm.ClusterInfo"]) 

63 logical_device_id_to_rank_id = {} 

64 if "ge.exec.deviceId" in engine_options and "ge.exec.rankId" in engine_options: 

65 rank_ids = [int(engine_options["ge.exec.rankId"])] 

66 log.info(f"both ge.exec.rankId and ge.exec.deviceId are defined, rank_id = {rank_ids[0]}") 

67 else: 

68 rank_ids = [i for i in range(len(cluster_info["logic_device_id"]))] 

69 for rank_id, logical_device_id in zip(rank_ids, cluster_info["logic_device_id"]): 

70 logical_device_id_to_rank_id[logical_device_id] = rank_id 

71 return logical_device_id_to_rank_id 

72 

73 @staticmethod 

74 def _parse_numa_config(engine_options) -> Dict[str, int]: 

75 raise_if_false( 

76 "ge.resourceConfigPath" in engine_options, 

77 "option 'ge.resourceConfigPath' is not defined", 

78 ) 

79 numa_config_path = engine_options["ge.resourceConfigPath"] 

80 with open(numa_config_path) as f: 

81 numa_config = json.load(f) 

82 logical_device_id_to_device_id: Dict[str, int] = {} 

83 for custer_idx, cluster in enumerate(numa_config["cluster"]): 

84 for node_idx, cluster_node in enumerate(cluster["cluster_nodes"]): 

85 is_local = cluster_node.get("is_local", False) 

86 if not is_local: 

87 continue 

88 for item_idx, item in enumerate(cluster_node["item_list"]): 

89 device_id = int(item["item_id"]) 

90 logical_device_id = ":".join([str(custer_idx), str(node_idx), str(item_idx), "0"]) 

91 logical_device_id_to_device_id[logical_device_id] = device_id 

92 return logical_device_id_to_device_id 

93 

94 

95class EngineConfig: 

96 def __init__(self, is_prompt: bool, cluster_config: ClusterConfig) -> None: 

97 self.is_prompt = is_prompt 

98 self.cluster_config = cluster_config 

99 

100 @classmethod 

101 def from_engine_options(cls, is_prompt: bool, engine_options: Dict[str, str]) -> "EngineConfig": 

102 cluster_config = ClusterConfig.from_engine_options(engine_options) 

103 return cls(is_prompt, cluster_config) 

104 

105 @staticmethod 

106 def gen_numa_config(device_id, deploy_res_path: str) -> str: 

107 node_type = "FakeNodeType" 

108 item_type = "FakeItemType" 

109 item_list = [] 

110 for dev_id in device_id.split(";"): 

111 item_list.append( 

112 { 

113 "item_id": int(dev_id), 

114 "device_id": int(dev_id), 

115 "ipaddr": "192.168.0.1", 

116 } 

117 ) 

118 numa_config = { 

119 "cluster": [ 

120 { 

121 "cluster_nodes": [ 

122 { 

123 "node_id": 0, 

124 "node_type": node_type, 

125 "ipaddr": "127.0.0.1", 

126 "port": -1, 

127 "is_local": True, 

128 "data_panel": {"avail_ports": "65000~65535"}, 

129 "item_list": item_list, 

130 } 

131 ], 

132 } 

133 ], 

134 "node_def": [ 

135 { 

136 "node_type": node_type, 

137 "resource_type": "Aarch", 

138 "support_links": "[HCCS,PCIE,ROCE]", 

139 "item_type": item_type, 

140 } 

141 ], 

142 "item_def": [ 

143 { 

144 "item_type": item_type, 

145 "resource_type": "Ascend", 

146 "memory": "[DDR:64GB]", 

147 "aic_type": "[FakeAicType]", 

148 } 

149 ], 

150 } 

151 if deploy_res_path is not None: 

152 check_isinstance("llm.deployResPath", deploy_res_path, str) 

153 numa_config["cluster"][0]["cluster_nodes"][0]["deploy_res_path"] = deploy_res_path 

154 return json.dumps(numa_config) 

155 

156 @staticmethod 

157 def gen_cluster_info_if_not_exist(cluster_id: int, role: LLMRole, engine_options: Dict[str, str]) -> None: 

158 if "llm.ClusterInfo" in engine_options: 

159 return 

160 device_num = len(engine_options["ge.exec.deviceId"].split(";")) 

161 cluster_info = { 

162 "cluster_id": cluster_id, 

163 "logic_device_id": [f"0:0:{i}:0" for i in range(device_num)], 

164 } 

165 if role == LLMRole.PROMPT: 

166 raise_if_false( 

167 "llm.listenIpInfo" in engine_options, 

168 "neither llm.ClusterInfo nor llm.listenIp was specified", 

169 ) 

170 listen_ip_info = engine_options["llm.listenIpInfo"] 

171 check_isinstance("listen_ip_info", listen_ip_info, [str]) 

172 sub_ip_infos = listen_ip_info.split(";") 

173 raise_if_false( 

174 len(sub_ip_infos) == device_num, 

175 f"listen ip info num:{len(sub_ip_infos)} in llm.listenIpInfo is not equal to device num:{device_num}.", 

176 ) 

177 cluster_info["listen_ip_info"] = [] 

178 for sub_ip_info in sub_ip_infos: 

179 ip_and_port = sub_ip_info.split(":") 

180 raise_if_false( 

181 len(ip_and_port) == 2, 

182 f'llm.listenIpInfo "{ip_and_port}" is invalid, format should be "ip:port"', 

183 ) 

184 cluster_info["listen_ip_info"].append({"ip": ip_and_port[0], "port": int(ip_and_port[1])}) 

185 llm_config = LlmConfig() 

186 llm_config.cluster_info = json.dumps(cluster_info) 

187 converted_options = llm_config.gen_options() 

188 engine_options["llm.ClusterInfo"] = converted_options["llm.ClusterInfo"] 

189 # create numa config if needed 

190 if "RESOURCE_CONFIG_PATH" in os.environ: 

191 numa_config_path = os.getenv("RESOURCE_CONFIG_PATH", "") 

192 else: 

193 raise_if_false( 

194 "ge.exec.deviceId" in engine_options, 

195 "neither llm.ClusterInfo nor ge.exec.deviceId was specified", 

196 ) 

197 device_id = engine_options["ge.exec.deviceId"] 

198 check_isinstance("ge.exec.deviceId", device_id, [str]) 

199 for dev_id in device_id.split(";"): 

200 raise_if_false( 

201 dev_id.isdigit(), 

202 f'ge.exec.deviceId is invalid, value="{device_id}",' 

203 f" it should be composed of numbers, separated by semicolons.", 

204 ) 

205 raise_if_false( 

206 len(device_id.split(";")) > 0, 

207 f'ge.exec.deviceId is invalid, value="{device_id}", At least one device id is required.', 

208 ) 

209 deploy_res_path = engine_options.get("llm.deployResPath", None) 

210 numa_config_str = EngineConfig.gen_numa_config(device_id, deploy_res_path) 

211 numa_config_path = f"/tmp/stub_numa_config_{role.name.lower()}_{'_'.join(device_id.split(';'))}.json" 

212 with open(numa_config_path, "w") as f: 

213 f.write(numa_config_str) 

214 engine_options["ge.resourceConfigPath"] = numa_config_path 

215 log.info("using numa config: %s", numa_config_path) 

216 

217 @staticmethod 

218 def parse_listen_ip_info(listen_ip_info: str) -> (int, int): 

219 check_isinstance("listen_ip_info", listen_ip_info, [str]) 

220 ip_and_port = listen_ip_info.split(":") 

221 raise_if_false( 

222 len(ip_and_port) == 2, 

223 f'llm.listenIpInfo "{ip_and_port}" is invalid, format should be "ip:port"', 

224 ) 

225 ip_int = trans_str_ip(ip_and_port[0]) 

226 port = int(ip_and_port[1]) 

227 return ip_int, port