DRIVER=processor MODALIAS=cpu:type:x86,ven0002fam001Amod0002:feature:,0000,0001,0002,0003,0004,0005,0006,0007,0008,0009,000B,000C,000D,000E,000F,0010,0011,0013,0017,0018,0019,001A,0020,0021,0022,0023,0024,0025,0026,0027,0028,0029,002B,002C,002D,002E,002F,0030,0031,0034,0036,0037,0038,0039,003A,003B,003D,0064,0065,006E,0070,0074,0075,0079,007A,007F,0080,0081,0089,008C,008D,0091,0093,0094,0095,0096,0097,0098,0099,009A,009B,009C,009D,009E,009F,00C0,00C1,00C4,00C5,00C6,00C7,00C8,00C9,00D7,00EA,00F0,00F1,00F4,00F9,00FA,00FB,00FC,00FE,010F,0114,0115,0120,0121,0123,0125,0127,0128,0129,012A,0130,0131,0132,0133,0134,0135,0137,0138,013C,013D,013E,013F,0140,0141,0142,0143,0165,017A,0184,0185,018B,01A0,01A2,01A9,01AC,01AE,01AF,01B1,01B8,01BC,01C2,0201,0202,0203,0204,0206,0208,0209,020A,020B,020C,020E,0210,0216,021B,021C,0220,0221,0244,0248,025A,025B,025C,025D,025F,0280,0281,0282,0286,0288,0294,029B,029C,029E #ˆl–ß=¿JJj"Tâ‚ ÆÃq§”Å£Ë ƒh¹ÊÀÈ1 $ˆl–ßi~”*e¶¥6 ^¸Ë×Ú˜´oÌ „l/…¿ËÁÉ?÷"""Selector event loop for Unix with signal handling.""" import errno import io import itertools import os import selectors import signal import socket import stat import subprocess import sys import threading import warnings from . import base_events from . import base_subprocess from . import constants from . import coroutines from . import events from . import exceptions from . import futures from . import selector_events from . import tasks from . import transports from .log import logger __all__ = ( 'SelectorEventLoop', 'AbstractChildWatcher', 'SafeChildWatcher', 'FastChildWatcher', 'PidfdChildWatcher', 'MultiLoopChildWatcher', 'ThreadedChildWatcher', 'DefaultEventLoopPolicy', ) if sys.platform == 'win32': # pragma: no cover raise ImportError('Signals are not really supported on Windows') def _sighandler_noop(signum, frame): """Dummy signal handler.""" pass def waitstatus_to_exitcode(status): try: return os.waitstatus_to_exitcode(status) except ValueError: # The child exited, but we don't understand its status. # This shouldn't happen, but if it does, let's just # return that status; perhaps that helps debug it. return status class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop): """Unix event loop. Adds signal handling and UNIX Domain Socket support to SelectorEventLoop. """ def __init__(self, selector=None): super().__init__(selector) self._signal_handlers = {} def close(self): super().close() if not sys.is_finalizing(): for sig in list(self._signal_handlers): self.remove_signal_handler(sig) else: if self._signal_handlers: warnings.warn(f"Closing the loop {self!r} " f"on interpreter shutdown " f"stage, skipping signal handlers removal", ResourceWarning, source=self) self._signal_handlers.clear() def _process_self_data(self, data): for signum in data: if not signum: # ignore null bytes written by _write_to_self() continue self._handle_signal(signum) def add_signal_handler(self, sig, callback, *args): """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. """ if (coroutines.iscoroutine(callback) or coroutines.iscoroutinefunction(callback)): raise TypeError("coroutines cannot be used " "with add_signal_handler()") self._check_signal(sig) self._check_closed() try: # set_wakeup_fd() raises ValueError if this is not the # main thread. By calling it early we ensure that an # event loop running in another thread cannot add