Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/common/ascend910_93/ascend91093_handler.py: 29%

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 

19from threading import Thread 

20from common.device import DeviceInfo 

21import common.interface as interface 

22 

23 

24class Ascend91093Handler(DeviceInfo): 

25 def __init__(self): 

26 super().__init__() 

27 

28 @classmethod 

29 def need_lp_param(cls): 

30 return False 

31 

32 def run_diagnose(self, device_obj, diagnose_devices, run_mode): 

33 """Multi-thread parallel execution""" 

34 threads = [] 

35 ret = {} 

36 master_ids = {} 

37 

38 logic_master_info = interface.get_devices_master_id(device_obj, diagnose_devices) 

39 if run_mode == "hbm_detect": 

40 _target_func = interface.run_hbm 

41 elif run_mode == "cpu_detect": 

42 _target_func = interface.run_cpu 

43 else: 

44 return interface.run_diagnose(device_obj, diagnose_devices, run_mode) 

45 

46 for device_id in diagnose_devices: 

47 if logic_master_info[device_id] not in master_ids: 

48 # new thread 

49 t = Thread(target=_target_func, args=(device_id, device_obj, ret), daemon=True) 

50 t.start() 

51 threads.append(t) 

52 master_ids[logic_master_info[device_id]] = device_id 

53 

54 # wait for all threads to end. 

55 for t in threads: 

56 t.join() 

57 

58 ret_logic_id = {} 

59 for device_id in diagnose_devices: 

60 ret_logic_id[device_id] = ret[master_ids[logic_master_info[device_id]]] 

61 

62 return ret_logic_id