Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_multicore / diagnostics / common / graph.py: 90%

1# Copyright 2026 Huawei Technologies Co., Ltd 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14# ============================================================================ 

15"""Business-independent graph contracts and diagnostic task identity.""" 

16 

17from __future__ import annotations 

18 

19from collections.abc import Callable, Mapping 

20from copy import deepcopy 

21from dataclasses import dataclass 

22from types import MappingProxyType 

23from typing import Any 

24 

25from hyper_multicore.scheduler.config import RuntimeConfigC, TaskDescC, TaskType 

26from hyper_multicore.scheduler.graph import ComputeGraph, OperatorNode 

27 

28 

29@dataclass(frozen=True) 

30class DiagnosticSpec: 

31 """Stable kernel identity and optional business-owned task-owner resolver.""" 

32 

33 kernel_name: str 

34 owner_label: str = "Owner" 

35 owner_resolver: Callable[[OperatorNode, TaskDescC, Any], int | None] | None = None 

36 owner_context: Any = None 

37 execution_mode: str = "event_driven" 

38 vector_worker_stride: int = 2 

39 graph_adapter: Callable[[Any, RuntimeConfigC], ComputeGraph] | None = None 

40 

41 

42@dataclass(frozen=True) 

43class GraphDiagnosticMetadata: 

44 """Host-only task identity, independent of trace or Tensor-dump formats.""" 

45 

46 kernel_name: str 

47 owner_label: str 

48 stage_names: Mapping[int, str] 

49 task_stage_ids: Mapping[int, int] 

50 task_stage_names: Mapping[int, str] 

51 task_owner_ids: Mapping[int, int | None] 

52 

53 

54def bind_graph_tensor_ports( 

55 graph: ComputeGraph, 

56 ports: Mapping[str, tuple[list[Any], list[Any]]], resolver: Callable[..., Any], 

57) -> ComputeGraph: 

58 """Bind real GM operands on a diagnostic-only copy of a scheduling graph. 

59 

60 Native task builders may describe dependencies without encoding tensor 

61 operands. The business adapter supplies the actual input-list positions; 

62 neither the scheduling graph nor the normal Device runtime is modified. 

63 

64 Args: 

65 graph: Business compute graph whose scheduling storage remains unchanged. 

66 ports: Mapping from operator names to actual native input and output operands. 

67 resolver: Business callback that resolves a selected operand's contiguous range. 

68 """ 

69 graph = deepcopy(graph) 

70 for operator in graph.topological_sort(): 

71 operator.inputs, operator.outputs = ports[operator.name] 

72 operator.param_positions = [tensor.param_position for tensor in operator.inputs + operator.outputs] 

73 operator.diagnostic_view_resolver = resolver 

74 operator.diagnostic_host_ports = True 

75 return graph 

76 

77 

78def operator_diagnostic_name(operator: OperatorNode) -> str: 

79 """Return an explicit diagnostic name or derive one from the graph node name. 

80 

81 Args: 

82 operator: Graph operator associated with the selected task. 

83 """ 

84 diagnostic_name = operator.diagnostic_name 

85 if diagnostic_name is not None: 

86 if not isinstance(diagnostic_name, str) or not diagnostic_name.strip(): 

87 raise ValueError( 

88 f"OperatorNode.diagnostic_name must be a non-empty string when set: {diagnostic_name!r}" 

89 ) 

90 return diagnostic_name.strip() 

91 if not isinstance(operator.name, str) or not operator.name.strip(): 

92 raise ValueError(f"OperatorNode.name must be a non-empty string, got {operator.name!r}") 

93 stage_name = "".join(part[:1].upper() + part[1:] for part in operator.name.strip().split("_") if part) 

94 if not stage_name: 

95 raise ValueError(f"OperatorNode.name must contain a diagnostic display character: {operator.name!r}") 

96 return stage_name 

97 

98 

99def _scheduled_task_ids(runtime_config: RuntimeConfigC) -> set[int]: 

100 """Collect queue task IDs, including tasks outside RuntimeConfig.task_num.""" 

101 task_ids = set() 

102 task_schedules = ( 

103 (runtime_config.cube_task_indices, int(runtime_config.task_index_num[0])), 

104 (runtime_config.vector_task_indices, int(runtime_config.task_index_num[1])), 

105 (runtime_config.mix_task_indices, int(runtime_config.task_index_num[2])), 

106 ) 

107 for task_indices, scheduled_task_count in task_schedules: 

108 if not 0 <= scheduled_task_count <= len(task_indices): 

109 raise ValueError(f"scheduled task count is outside RuntimeConfig capacity: {scheduled_task_count}") 

110 for schedule_index in range(scheduled_task_count): 

111 task_id = int(task_indices[schedule_index]) 

112 if not 0 <= task_id < len(runtime_config.all_tasks): 

113 raise ValueError(f"scheduled task ID is outside RuntimeConfig capacity: {task_id}") 

114 task_ids.add(task_id) 

115 return task_ids 

116 

117 

118def _validate_parameter_position(position: int, parameter_count: int, label: str) -> None: 

119 if isinstance(position, bool) or not isinstance(position, int) or not 0 <= position < parameter_count: 

120 raise ValueError(f"{label} parameter position {position!r} is outside ABI parameter_count={parameter_count}") 

121 

122 

123def _validate_operator(operator: OperatorNode, parameter_count: int) -> None: 

124 """Validate graph operand names, shapes and native ABI bindings.""" 

125 if isinstance(operator.task_num, bool) or not isinstance(operator.task_num, int) or operator.task_num < 0: 

126 raise ValueError(f"operator {operator.name!r} has invalid task_num={operator.task_num!r}") 

127 ports = operator.inputs + operator.outputs 

128 if len(operator.param_positions) != len(ports): 

129 raise ValueError(f"operator {operator.name!r} param_positions must match its input/output ports") 

130 for position, tensor_spec in zip(operator.param_positions, ports): 

131 _validate_parameter_position(tensor_spec.param_position, parameter_count, tensor_spec.name) 

132 if position != tensor_spec.param_position: 

133 raise ValueError(f"operator {operator.name!r} param_positions disagrees with Tensor {tensor_spec.name!r}") 

134 if not isinstance(tensor_spec.name, str) or not tensor_spec.name.strip(): 

135 raise ValueError("TensorSpec.name must be a non-empty string") 

136 if any(isinstance(dim, bool) or not isinstance(dim, int) or dim < 0 for dim in tensor_spec.shape): 

137 raise ValueError(f"Tensor {tensor_spec.name!r} shape must contain non-negative integers") 

138 if (isinstance(tensor_spec.dtype_size, bool) or not isinstance(tensor_spec.dtype_size, int) 

139 or tensor_spec.dtype_size <= 0): 

140 raise ValueError(f"Tensor {tensor_spec.name!r} dtype_size must be a positive integer") 

141 

142 

143def _validate_task_ports( 

144 task: TaskDescC, operator: OperatorNode, task_id: int, 

145) -> None: 

146 """Check a scheduled task's operand descriptors against the graph contract.""" 

147 if operator.diagnostic_host_ports: 

148 if not callable(operator.diagnostic_view_resolver): 

149 raise ValueError(f"native ABI stage {operator.name} requires a view resolver") 

150 return 

151 for io, descriptors, specs, count in ( 

152 ("input", task.inputs, operator.inputs, int(task.num_inputs)), 

153 ("output", task.outputs, operator.outputs, int(task.num_outputs)), 

154 ): 

155 if count != len(specs) or count > len(descriptors): 

156 raise ValueError(f"task {task_id} stage {operator.name!r} {io} count does not match graph/ABI capacity") 

157 for index, spec in enumerate(specs): 

158 descriptor = descriptors[index] 

159 if int(descriptor.input_position) != spec.param_position: 

160 raise ValueError(f"task {task_id} {io}{index} parameter position disagrees with Tensor {spec.name!r}") 

161 if int(descriptor.data_type) != spec.dtype_size: 

162 raise ValueError(f"task {task_id} {io}{index} element byte size disagrees with Tensor {spec.name!r}") 

163 

164 

165def _validate_task_schedule(runtime_config: RuntimeConfigC, graph_task_count: int) -> None: 

166 """Reject missing, duplicated or unmapped compute tasks in Device queues.""" 

167 queues = (runtime_config.cube_task_indices, runtime_config.vector_task_indices, runtime_config.mix_task_indices) 

168 scheduled_ids = _scheduled_task_ids(runtime_config) 

169 seen = set() 

170 for indices, count in zip(queues, runtime_config.task_index_num[:3]): 

171 for task_id in indices[:count]: 

172 if task_id in seen: 

173 raise ValueError(f"task {task_id} appears more than once in Device schedules") 

174 seen.add(task_id) 

175 missing = set(range(graph_task_count)) - scheduled_ids 

176 if missing: 

177 raise ValueError(f"graph tasks are absent from Device schedules: {sorted(missing)}") 

178 for task_id in scheduled_ids | set(range(int(runtime_config.task_num))): 

179 if task_id >= graph_task_count and runtime_config.all_tasks[task_id].task_type != TaskType.TASK_TERMINATE: 

180 raise ValueError(f"task {task_id} has no graph stage and is not a termination task") 

181 

182 

183def validate_mega_kernel_diagnostic_contract( 

184 graph: ComputeGraph, 

185 runtime_config: RuntimeConfigC, 

186 parameter_count: int, 

187) -> None: 

188 """Validate graph/task/ABI agreement before allocating Device diagnostic resources. 

189 

190 Args: 

191 graph: Propagated graph; task IDs must follow contiguous topological ranges. 

192 runtime_config: Filled graph-sized Host configuration including Device queues. 

193 parameter_count: Number of launch binding entries. Device task ports 

194 must use real native input_list positions. Zero-task nodes may 

195 describe Host-visible results produced by an enclosing native call. 

196 

197 Raises: 

198 TypeError: Invalid graph or runtime configuration. 

199 ValueError: Inconsistent graph/task/ABI mapping. 

200 

201 Note: 

202 This read-only check does not call view resolvers or inspect Device Tensors. 

203 Actual offset/shape memory bounds are validated when L1 prepares a launch. 

204 """ 

205 if not isinstance(graph, ComputeGraph): 

206 raise TypeError("graph must be ComputeGraph") 

207 if not isinstance(runtime_config, RuntimeConfigC) or not hasattr(runtime_config, "all_tasks"): 

208 raise TypeError("runtime_config must be an allocated graph-sized RuntimeConfigC") 

209 if isinstance(parameter_count, bool) or not isinstance(parameter_count, int) or parameter_count < 0: 

210 raise ValueError("parameter_count must be a non-negative integer") 

211 if runtime_config.task_num > len(runtime_config.all_tasks): 

212 raise ValueError("RuntimeConfig.task_num exceeds allocated task capacity") 

213 task_id = 0 

214 for operator in graph.topological_sort(): 

215 _validate_operator(operator, parameter_count) 

216 if task_id + operator.task_num > runtime_config.task_num: 

217 raise ValueError(f"graph tasks for stage {operator.name!r} exceed RuntimeConfig.task_num") 

218 for current_task_id in range(task_id, task_id + operator.task_num): 

219 _validate_task_ports(runtime_config.all_tasks[current_task_id], operator, current_task_id) 

220 task_id += operator.task_num 

221 _validate_task_schedule(runtime_config, task_id) 

222 

223 

224def build_graph_metadata( 

225 graph: ComputeGraph, runtime_config: RuntimeConfigC, spec: DiagnosticSpec, 

226) -> GraphDiagnosticMetadata: 

227 """Resolve stable graph stage IDs and optional owners without modifying tasks. 

228 

229 Args: 

230 graph: Business compute graph whose scheduling storage remains unchanged. 

231 runtime_config: Complete Host configuration describing native task queues. 

232 spec: Tool-neutral kernel identity and optional business owner callbacks. 

233 """ 

234 for label, value in (("kernel_name", spec.kernel_name), ("owner_label", spec.owner_label)): 

235 if not isinstance(value, str) or not value.strip(): 

236 raise ValueError(f"{label} must be a non-empty string, got {value!r}") 

237 if spec.owner_resolver is not None and not callable(spec.owner_resolver): 

238 raise TypeError("owner_resolver must be callable or None") 

239 stages = {} 

240 task_stages = {} 

241 task_names = {} 

242 owners = {} 

243 task_id = 0 

244 for stage_id, operator in enumerate(graph.topological_sort()): 

245 name = operator_diagnostic_name(operator) 

246 stages[stage_id] = name 

247 for current_id in range(task_id, task_id + operator.task_num): 

248 owner = None 

249 if spec.owner_resolver is not None: 

250 owner = spec.owner_resolver(operator, runtime_config.all_tasks[current_id], spec.owner_context) 

251 if owner is not None and (isinstance(owner, bool) or not isinstance(owner, int) 

252 or not 0 <= owner < 0xFFFFFFFF): 

253 raise ValueError(f"diagnostic owner_id must be a uint32 integer or None, got {owner!r}") 

254 task_stages[current_id] = stage_id 

255 task_names[current_id] = name 

256 owners[current_id] = owner 

257 task_id += operator.task_num 

258 return GraphDiagnosticMetadata( 

259 spec.kernel_name.strip(), spec.owner_label.strip(), 

260 MappingProxyType(stages), MappingProxyType(task_stages), 

261 MappingProxyType(task_names), MappingProxyType(owners), 

262 )