Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/launch/asys_launch.py: 97%

1#!/usr/bin/env python3 

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

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

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

5# 

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

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

8# You may obtain a copy of the License at 

9# 

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

11# 

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

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

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

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

16# limitations under the License. 

17# ---------------------------------------------------------------------------- 

18 

19import os 

20import subprocess 

21import signal 

22import sys 

23import threading 

24 

25from params import ParamDict 

26from common import log_info, log_error, log_debug, log_warning 

27from common import FileOperate as f 

28from common.const import RetCode 

29from view.progress_display import waiting 

30from collect import AsysCollect 

31 

32__all__ = ["AsysLaunch"] 

33 

34 

35class AsysLaunch: 

36 def __init__(self): 

37 self.output_root_path = ParamDict().asys_output_timestamp_dir 

38 self.finish_flag = False 

39 self.user_cmd = ParamDict().get_arg("task") 

40 self.console_output = "" 

41 npu_collect_path = os.path.join(ParamDict().asys_output_timestamp_dir, "npu_collect_intermediates") 

42 self.env_prepare = { 

43 "DUMP_GE_GRAPH": ParamDict().get_ini("DUMP_GE_GRAPH"), 

44 "DUMP_GRAPH_LEVEL": ParamDict().get_ini("DUMP_GRAPH_LEVEL"), 

45 "ASCEND_GLOBAL_LOG_LEVEL": ParamDict().get_ini("ASCEND_GLOBAL_LOG_LEVEL"), 

46 "ASCEND_GLOBAL_EVENT_ENABLE": ParamDict().get_ini("ASCEND_GLOBAL_EVENT_ENABLE"), 

47 "ASCEND_SLOG_PRINT_TO_STDOUT": ParamDict().get_ini("ASCEND_SLOG_PRINT_TO_STDOUT"), 

48 "ASCEND_HOST_LOG_FILE_NUM": "1000", 

49 "ASCEND_PROCESS_LOG_PATH": os.path.join(npu_collect_path, "task_launch_host_log"), 

50 "ASCEND_WORK_PATH": os.path.join(npu_collect_path, "task_launch_host_log"), 

51 "NPU_COLLECT_PATH": npu_collect_path, 

52 } 

53 

54 def prepare_for_launch(self): 

55 # prepare environment variable 

56 for env_name, env_val in self.env_prepare.items(): 

57 log_debug("env_name: {}, env_val: {}".format(env_name, env_val)) 

58 os.environ[env_name] = env_val 

59 # prepare npu collect path dir 

60 if not f.create_dir(os.environ["NPU_COLLECT_PATH"]): 

61 log_error(f"Create npu collect path failed, NPU_COLLECT_PATH={self.env_prepare['NPU_COLLECT_PATH']}") 

62 return RetCode.FAILED 

63 # collect atrace log 

64 if not f.create_dir(os.environ["ASCEND_WORK_PATH"]): 

65 log_error(f"Create ascend work path failed, ASCEND_WORK_PATH={self.env_prepare['ASCEND_WORK_PATH']}") 

66 return RetCode.FAILED 

67 

68 log_debug("Prepare for launch finished.") 

69 return RetCode.SUCCESS 

70 

71 def execute_task(self): 

72 def interrupt_handler(signum, frame): 

73 os.killpg(os.getpgid(pro.pid), signal.SIGTERM) 

74 signal.signal(signal.SIGINT, signal.SIG_DFL) 

75 # exit the current main process group. 

76 os.killpg(os.getpgid(0), signal.SIGINT) 

77 

78 log_info('launch task start, running:') 

79 signal.signal(signal.SIGINT, interrupt_handler) 

80 pro = subprocess.Popen(self.user_cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, 

81 encoding='utf-8', preexec_fn=os.setsid, env=os.environ, errors='ignore') 

82 t = threading.Thread(target=self.wait_view, daemon=True) 

83 t.start() 

84 

85 ParamDict().set_task_pid(pid=pro.pid) 

86 

87 self.console_output, _ = pro.communicate() 

88 # subprocess end, restore the default SIGINT signal 

89 signal.signal(signal.SIGINT, signal.SIG_DFL) 

90 

91 self.finish_flag = True 

92 t.join() # wait print process_display end 

93 

94 if pro.returncode == 0: 

95 log_info('Task execute finished, output:\n{0}'.format(self.console_output)) 

96 else: 

97 log_warning("Task occurred error, output:\n{0}".format(self.console_output)) 

98 

99 def task_out_collect(self, output_root_path): 

100 def collect_task_output_info(): 

101 if ParamDict().get_env_type() == "EP": 

102 dir_path = os.path.join(output_root_path, "dfx", "log", "host") 

103 else: 

104 dir_path = os.path.join(output_root_path, "dfx", "log") 

105 if not f.check_dir(dir_path): 

106 f.create_dir(dir_path) 

107 user_cmd_path = os.path.join(dir_path, "user_cmd") 

108 screen_print_path = os.path.join(dir_path, "screen.txt") 

109 f.write_file(user_cmd_path, self.user_cmd) 

110 f.write_file(screen_print_path, self.console_output) 

111 log_debug("Collect user cmd and task print successfully.") 

112 

113 collect_task_output_info() 

114 task_collector = AsysCollect() 

115 if not task_collector.collect(): 

116 log_error("Collect information after task failed.") 

117 else: 

118 log_info("Collect information after task successfully.") 

119 

120 def launch(self): 

121 ret = self.prepare_for_launch() 

122 if ret != RetCode.SUCCESS: 

123 log_error("Prepare for launch failed.") 

124 return False 

125 

126 self.execute_task() 

127 self.task_out_collect(self.output_root_path) 

128 

129 return True 

130 

131 def wait_view(self): 

132 while not self.finish_flag: 

133 waiting() 

134 continue 

135 

136 def clean_work(self): 

137 npu_collect_path = os.path.join(self.output_root_path, "npu_collect_intermediates") 

138 msnpureport_export_path = os.path.join(self.output_root_path, "export_tmp") 

139 dir_list = [npu_collect_path, msnpureport_export_path] 

140 f.delete_dirs(dir_list) 

141 

142 def run(self): 

143 task_res = self.launch() 

144 self.clean_work() 

145 return task_res