(self, signum, frame, main_task)
| 157 | self._state = _State.INITIALIZED |
| 158 | |
| 159 | def _on_sigint(self, signum, frame, main_task): |
| 160 | self._interrupt_count += 1 |
| 161 | if self._interrupt_count == 1 and not main_task.done(): |
| 162 | main_task.cancel() |
| 163 | # wakeup loop if it is blocked by select() with long timeout |
| 164 | self._loop.call_soon_threadsafe(lambda: None) |
| 165 | return |
| 166 | raise KeyboardInterrupt() |
| 167 | |
| 168 | |
| 169 | def run(main, *, debug=None, loop_factory=None): |
nothing calls this directly
no test coverage detected