Coverage for /usr/local/lib/python3.12/site-packages/prefect/_internal/concurrency/inspection.py: 0%

1""" 

2Utilities for inspection of stack frames and threads. 

3""" 

4 

5import dis 

6import linecache 

7import sys 

8import threading 

9from types import FrameType 

10 

11""" 

12The following functions are derived from dask/distributed which is licensed under the 

13BSD 3-Clause License. 

14 

15Copyright (c) 2015, Anaconda, Inc. and contributors 

16All rights reserved. 

17 

18Redistribution and use in source and binary forms, with or without 

19modification, are permitted provided that the following conditions are met: 

20 

21* Redistributions of source code must retain the above copyright notice, this 

22 list of conditions and the following disclaimer. 

23 

24* Redistributions in binary form must reproduce the above copyright notice, 

25 this list of conditions and the following disclaimer in the documentation 

26 and/or other materials provided with the distribution. 

27 

28* Neither the name of the copyright holder nor the names of its 

29 contributors may be used to endorse or promote products derived from 

30 this software without specific prior written permission. 

31 

32THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 

33AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 

34IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE 

35DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE 

36FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL 

37DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR 

38SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER 

39CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, 

40OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE 

41OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 

42""" 

43 

44 

45def _f_lineno(frame: FrameType) -> int: 

46 """Work around some frames lacking an f_lineno 

47 See: https://bugs.python.org/issue47085 

48 """ 

49 f_lineno = frame.f_lineno 

50 if f_lineno is not None: 

51 return f_lineno 

52 

53 f_lasti = frame.f_lasti 

54 code = frame.f_code 

55 prev_line = code.co_firstlineno 

56 

57 for start, next_line in dis.findlinestarts(code): 

58 if f_lasti < start: 

59 return prev_line 

60 prev_line = next_line 

61 

62 return prev_line 

63 

64 

65def repr_frame(frame: FrameType) -> str: 

66 """Render a frame as a line for inclusion into a text traceback""" 

67 co = frame.f_code 

68 f_lineno = _f_lineno(frame) 

69 text = f' File "{co.co_filename}", line {f_lineno}, in {co.co_name}' 

70 line = linecache.getline(co.co_filename, f_lineno, frame.f_globals).lstrip() 

71 return text + "\n\t" + line 

72 

73 

74def call_stack(frame: FrameType) -> list[str]: 

75 """Create a call text stack from a frame""" 

76 frames: list[str] = [] 

77 cur_frame = frame 

78 while cur_frame: 

79 frames.append(repr_frame(cur_frame)) 

80 cur_frame = cur_frame.f_back 

81 return frames[::-1] 

82 

83 

84def stack_for_threads(*threads: threading.Thread) -> list[str]: 

85 frames = sys._current_frames() # pyright: ignore[reportPrivateUsage] 

86 try: 

87 lines: list[str] = [] 

88 for thread in threads: 

89 ident = thread.ident 

90 hex_ident = hex(ident) if ident is not None else "<unknown>" 

91 lines.append(f"------ Call stack of {thread.name} ({hex_ident}) -----") 

92 if ident is not None and (thread_frames := frames.get(ident)): 

93 lines.append("".join(call_stack(thread_frames))) 

94 else: 

95 lines.append("No stack frames found") 

96 finally: 

97 del frames 

98 

99 return lines