Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_multicore / diagnostics / common / call.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"""Complete launch arguments and a tool-neutral per-invocation lifecycle.""" 

16 

17from __future__ import annotations 

18 

19from collections.abc import Sequence 

20from dataclasses import dataclass 

21from typing import Any 

22 

23 

24@dataclass 

25class LaunchArguments: 

26 """Complete native arguments selected by exactly one diagnostic tool.""" 

27 

28 runtime_config: Any 

29 event_counters: Any 

30 diagnostic_buffer: Any 

31 clear_event_counters: Any 

32 

33 

34@dataclass(frozen=True) 

35class LaunchContext: 

36 """Business-independent inputs available to a selected diagnostic session.""" 

37 

38 runtime: Any 

39 parameters: Sequence[Any] 

40 direction: str 

41 fallback_event_counters: Any 

42 

43 def ordinary_arguments(self) -> LaunchArguments: 

44 """Reuse the original runtime and counters without allocating device storage.""" 

45 return LaunchArguments( 

46 self.runtime.normal_tensor, self.fallback_event_counters, 

47 self.runtime.normal_tensor if self.fallback_event_counters is None else self.fallback_event_counters, 

48 self.fallback_event_counters, 

49 ) 

50 

51 

52class DiagnosticCall: 

53 """Context boundary shared by ordinary, profiling and Tensor-dump calls.""" 

54 

55 def __init__(self, arguments: LaunchArguments) -> None: 

56 """Store the full argument selection and idempotent completion state.""" 

57 self.arguments = arguments 

58 self.finished = False 

59 

60 @property 

61 def runtime_config(self) -> Any: 

62 """Return this call's complete runtime configuration.""" 

63 return self.arguments.runtime_config 

64 

65 @property 

66 def event_counters(self) -> Any: 

67 """Return the existing execution event counters.""" 

68 return self.arguments.event_counters 

69 

70 @property 

71 def diagnostic_buffer(self) -> Any: 

72 """Return this call's shared diagnostic-buffer argument.""" 

73 return self.arguments.diagnostic_buffer 

74 

75 @property 

76 def clear_event_counters(self) -> Any: 

77 """Return the event-counter region cleared by the execution owner.""" 

78 return self.arguments.clear_event_counters 

79 

80 def __enter__(self) -> DiagnosticCall: 

81 """Enter the native launch boundary.""" 

82 return self 

83 

84 def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None: 

85 """Complete successful calls and cancel failed preparation or capture.""" 

86 if exc_type is not None: 

87 self.cancel() 

88 return 

89 try: 

90 self.complete() 

91 except BaseException: 

92 self.cancel() 

93 raise 

94 

95 def complete(self) -> None: 

96 """Finish an ordinary call without any diagnostic work.""" 

97 self.finished = True 

98 

99 def cancel(self) -> None: 

100 """Discard an ordinary call without touching execution-owned resources.""" 

101 self.finished = True