Skip to content

mode.debug

Debugging utilities.

Blocking

Bases: RuntimeError

Exception raised when event loop is blocked.

Source code in mode/debug.py 
36
37
class Blocking(RuntimeError):
    """Exception raised when event loop is blocked."""

BlockingDetector

Bases: Service

Service that detects blocking code using alarm/itimer.

Examples:

blockdetect = BlockingDetector(timeout=10.0)
await blockdetect.start()

Other Parameters:

Name Type Description
timeout Seconds

number of seconds that the event loop can be blocked.
raises Type[BaseException]

Exception to raise when the blocking timeout is exceeded. Defaults to :exc:Blocking.
Source code in mode/debug.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
class BlockingDetector(Service):
    """Service that detects blocking code using alarm/itimer.

    Examples:

    ```python
    blockdetect = BlockingDetector(timeout=10.0)
    await blockdetect.start()
    ```

    Keyword Arguments:
        timeout (Seconds): number of seconds that the event loop can
            be blocked.
        raises (Type[BaseException]): Exception to raise when the blocking
            timeout is exceeded.  Defaults to :exc:`Blocking`.
    """

    logger = logger

    def __init__(
        self,
        timeout: Seconds,
        raises: Type[BaseException] = Blocking,
        **kwargs: Any,
    ) -> None:
        self.timeout: float = want_seconds(timeout)
        self.raises: Type[BaseException] = raises
        super().__init__(**kwargs)

    @Service.task
    async def _deadman_switch(self) -> None:
        try:
            while not self.should_stop:
                self._reset_signal()
                await self.sleep(self.timeout * 0.96)
        finally:
            self._clear_signal()

    def _reset_signal(self) -> None:
        signal.signal(signal.SIGALRM, self._on_alarm)
        self._arm(self.timeout)

    def _clear_signal(self) -> None:
        self._arm(0)

    def _arm(self, timeout: float) -> None:
        arm_alarm(timeout)

    def _on_alarm(self, signum: int, frame: FrameType) -> None:
        msg = f"Blocking detected (timeout={self.timeout})"
        stack = "".join(traceback.format_stack(frame))
        self.log.warning(
            "Blocking detected (timeout=%r) %s", self.timeout, stack
        )
        self._reset_signal()
        raise self.raises(msg)