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

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"""Ordinary runtime storage and lazy, tool-neutral diagnostic metadata.""" 

16 

17from __future__ import annotations 

18 

19from collections.abc import Callable, Mapping 

20from enum import IntEnum 

21from threading import Lock 

22from typing import Any 

23 

24from hyper_multicore.diagnostics.common.graph import ( 

25 DiagnosticSpec, GraphDiagnosticMetadata, build_graph_metadata, validate_mega_kernel_diagnostic_contract, 

26) 

27from hyper_multicore.scheduler.config import RuntimeConfigC 

28from hyper_multicore.scheduler.graph import ComputeGraph 

29from hyper_multicore.scheduler.runtime import serialize_runtime_config 

30 

31 

32class DiagnosticMode(IntEnum): 

33 """Device execution modes sharing one diagnostic_buffer ABI argument.""" 

34 

35 NORMAL = 0 

36 PROFILE = 1 

37 TENSOR_DUMP_L1 = 2 

38 

39 

40class PreparedDiagnosticRuntime: 

41 """Own the ordinary runtime; prepare diagnostic metadata only on demand.""" 

42 

43 def __init__( 

44 self, graph: ComputeGraph, runtime_config: RuntimeConfigC, spec: DiagnosticSpec, *, 

45 parameter_count: int, tensor_factory: Callable[[bytes], Any], rank: int, device_id: int, 

46 ) -> None: 

47 """Save the graph contract and create only the normal execution image.""" 

48 self._graph = graph 

49 self.spec = spec 

50 self.parameter_count = parameter_count 

51 self.rank = rank 

52 self.device_id = device_id 

53 self.kernel_name = spec.kernel_name 

54 self.owner_label = spec.owner_label 

55 self._tensor_factory = tensor_factory 

56 self._host_config = type(runtime_config).from_buffer_copy(bytes(runtime_config)) 

57 self._host_config.diagnostic_mode = DiagnosticMode.NORMAL 

58 self.normal_tensor = tensor_factory(serialize_runtime_config(self._host_config)) 

59 self._metadata: GraphDiagnosticMetadata | None = None 

60 self._metadata_lock = Lock() 

61 self._diagnostic_config: RuntimeConfigC | None = None 

62 

63 @property 

64 def graph(self) -> ComputeGraph: 

65 """Return the diagnostic graph after lazy business-owned ABI binding.""" 

66 if self.spec.graph_adapter is not None and self._metadata is None: 

67 _ = self.metadata 

68 return self._graph 

69 

70 @property 

71 def host_config(self) -> RuntimeConfigC: 

72 """Return the preserved Host configuration used to describe task operands.""" 

73 if self.spec.graph_adapter is not None: 

74 _ = self.metadata 

75 return self._diagnostic_config if self._diagnostic_config is not None else self._host_config 

76 

77 @property 

78 def metadata(self) -> GraphDiagnosticMetadata: 

79 """Validate and resolve graph identity only when a tool actually needs it.""" 

80 if self._metadata is None: 

81 with self._metadata_lock: 

82 if self._metadata is None: 

83 config = type(self._host_config).from_buffer_copy(bytes(self._host_config)) 

84 graph = self._graph 

85 if self.spec.graph_adapter is not None: 

86 graph = self.spec.graph_adapter(graph, config) 

87 validate_mega_kernel_diagnostic_contract(graph, config, self.parameter_count) 

88 metadata = build_graph_metadata(graph, config, self.spec) 

89 self._graph = graph 

90 self._diagnostic_config = config 

91 self._metadata = metadata 

92 return self._metadata 

93 

94 @property 

95 def task_stage_names(self) -> Mapping[int, str]: 

96 """Return the lazily resolved task-to-stage names.""" 

97 return self.metadata.task_stage_names 

98 

99 def copy_host_config(self) -> RuntimeConfigC: 

100 """Create an independent Host configuration for one tool's device variant.""" 

101 config = self.host_config 

102 return type(config).from_buffer_copy(bytes(config)) 

103 

104 def tensor_from_config(self, runtime_config: RuntimeConfigC) -> Any: 

105 """Serialize a tool-owned configuration without changing the normal image. 

106 

107 Args: 

108 runtime_config: Complete Host configuration describing native task queues. 

109 """ 

110 return self._tensor_factory(serialize_runtime_config(runtime_config))