Add a handler for a signal. UNIX only. Raise ValueError if the signal number is invalid or uncatchable. Raise RuntimeError if there is a problem setting up the handler.
(self, sig, callback, *args)
| 88 | self._handle_signal(signum) |
| 89 | |
| 90 | def add_signal_handler(self, sig, callback, *args): |
| 91 | """Add a handler for a signal. UNIX only. |
| 92 | |
| 93 | Raise ValueError if the signal number is invalid or uncatchable. |
| 94 | Raise RuntimeError if there is a problem setting up the handler. |
| 95 | """ |
| 96 | if (coroutines.iscoroutine(callback) or |
| 97 | coroutines._iscoroutinefunction(callback)): |
| 98 | raise TypeError("coroutines cannot be used " |
| 99 | "with add_signal_handler()") |
| 100 | self._check_signal(sig) |
| 101 | self._check_closed() |
| 102 | try: |
| 103 | # set_wakeup_fd() raises ValueError if this is not the |
| 104 | # main thread. By calling it early we ensure that an |
| 105 | # event loop running in another thread cannot add a signal |
| 106 | # handler. |
| 107 | signal.set_wakeup_fd(self._csock.fileno()) |
| 108 | except (ValueError, OSError) as exc: |
| 109 | raise RuntimeError(str(exc)) |
| 110 | |
| 111 | handle = events.Handle(callback, args, self, None) |
| 112 | self._signal_handlers[sig] = handle |
| 113 | |
| 114 | try: |
| 115 | # Register a dummy signal handler to ask Python to write the signal |
| 116 | # number in the wakeup file descriptor. _process_self_data() will |
| 117 | # read signal numbers from this file descriptor to handle signals. |
| 118 | signal.signal(sig, _sighandler_noop) |
| 119 | |
| 120 | # Set SA_RESTART to limit EINTR occurrences. |
| 121 | signal.siginterrupt(sig, False) |
| 122 | except OSError as exc: |
| 123 | del self._signal_handlers[sig] |
| 124 | if not self._signal_handlers: |
| 125 | try: |
| 126 | signal.set_wakeup_fd(-1) |
| 127 | except (ValueError, OSError) as nexc: |
| 128 | logger.info('set_wakeup_fd(-1) failed: %s', nexc) |
| 129 | |
| 130 | if exc.errno == errno.EINVAL: |
| 131 | raise RuntimeError(f'sig {sig} cannot be caught') |
| 132 | else: |
| 133 | raise |
| 134 | |
| 135 | def _handle_signal(self, sig): |
| 136 | """Internal helper that is the actual signal handler.""" |
nothing calls this directly
no test coverage detected