Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / auto_parallel / sapp_ppb / simulator / causal_error.py: 100%

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"""Exception types raised by the pipeline simulator when dependency loops are detected.""" 

16from __future__ import annotations 

17 

18import matplotlib.pyplot as plt 

19 

20from hyper_parallel.auto_parallel.sapp_ppb.simulator.plot_manager import PlotMgr 

21from hyper_parallel.auto_parallel.sapp_ppb.simulator.sim_block import BlockSim, MicroBlockSim 

22from hyper_parallel.auto_parallel.sapp_ppb.utils.logger import logger 

23 

24 

25class CausalError(Exception): 

26 """Raised when the block pipeline (without comm) contains a dependency loop.""" 

27 

28 def __init__(self, msg: str, blocks: list[list[MicroBlockSim]], loop: list[BlockSim]) -> None: 

29 """Create the error, draw the offending loop and log a diagnostic message. 

30 

31 Args: 

32 msg: Human-readable description of the loop. 

33 blocks: Full 2-D grid of simulator blocks, ``[pp_rank][block_idx]``. 

34 loop: Sequence of blocks participating in the dependency loop. 

35 """ 

36 super().__init__() 

37 self.msg = msg 

38 self.canvas = PlotMgr(num_plots=1, figsize=(12, 6)) 

39 self.canvas.draw_loop(blocks, loop, 0, False, False, True) 

40 self.canvas.ax[0].set_title("Block pipeline dependency") 

41 logger.error("%s", self.canvas.msg) 

42 

43 def __str__(self) -> str: 

44 """Show the diagnostic plot and return the error message.""" 

45 plt.show() 

46 return f"{self.msg}" 

47 

48 

49class CausalCommError(Exception): 

50 """Raised when the block pipeline with communication contains a dependency loop.""" 

51 

52 def __init__(self, msg: str, blocks: list[list[MicroBlockSim]], loop: list[BlockSim]) -> None: 

53 """Create the error, draw the offending loop and log a diagnostic message. 

54 

55 Args: 

56 msg: Human-readable description of the loop. 

57 blocks: Full 2-D grid of simulator blocks, ``[pp_rank][block_idx]``. 

58 loop: Sequence of blocks (compute + comm) participating in the dependency loop. 

59 """ 

60 super().__init__() 

61 self.msg = msg 

62 self.canvas = PlotMgr(num_plots=1, figsize=(12, 6)) 

63 self.canvas.draw_comm_loop(blocks, loop, 0) 

64 self.canvas.ax[0].set_title("Block comm pipeline dependency") 

65 logger.error("%s", self.canvas.msg) 

66 

67 def __str__(self) -> str: 

68 """Show the diagnostic plot and return the error message.""" 

69 plt.show() 

70 return f"{self.msg}"