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 baseevents from import basesubprocess from import constants from import coroutines from import events from import exceptions from import futures from import selectorevents from import tasks from import transports from log import logger all selectoreventloop abstractchildwatcher safechildwatcher fastchildwatcher pidfdchildwatcher multiloopchildwatcher threadedchildwatcher defaulteventlooppolicy eventloop if sys platform win32 pragma no cover raise importerror signals are not really supported on windows def sighandlernoopsignum frame 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 unix event loop adds signal handling and unix domain socket support to selectoreventloop ignore null bytes written by writetoself 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 setwakeupfd 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 a signal handler register a dummy signal handler to ask python to write the signal number in the wakeup file descriptor processselfdata will read signal numbers from this file descriptor to handle signals set sarestart to limit eintr occurrences internal helper that is the actual signal handler handle self signalhandlers getsig if handle is none return assume it s some race condition if handle cancelled self removesignalhandlersig remove it properly else self addcallbacksignalsafehandle def removesignalhandlerself sig self checksignalsig try del self signalhandlerssig except keyerror return false if sig signal sigint handler signal defaultinthandler else handler signal sigdfl try signal signalsig handler except oserror as exc if exc errno errno einval raise runtimeerrorf sig sig cannot be caught else raise if not self signalhandlers try signal setwakeupfd1 except valueerror oserror as exc logger info setwakeupfd1 failed s exc return true def checksignalself sig if not isinstancesig int raise typeerrorf sig must be an int not sig r if sig not in signal validsignals raise valueerrorf invalid signal number sig def makereadpipetransportself pipe protocol waiternone extranone return unixreadpipetransportself pipe protocol waiter extra def makewritepipetransportself pipe protocol waiternone extranone return unixwritepipetransportself pipe protocol waiter extra async def makesubprocesstransportself protocol args shell stdin stdout stderr bufsize extranone kwargs with warnings catchwarnings warnings simplefilter ignore deprecationwarning watcher events getchildwatcher with watcher if not watcher isactive check early raising exception before process creation prevents subprocess execution if the watcher is not ready to handle it raise runtimeerrorasyncio getchildwatcher is not activated subprocess support is not installed waiter self createfuture transp unixsubprocesstransportself protocol args shell stdin stdout stderr bufsize waiterwaiter extraextra kwargs watcher addchildhandlertransp getpid self childwatchercallback transp try await waiter except systemexit keyboardinterrupt raise except baseexception transp close await transp wait raise return transp def childwatchercallbackself pid returncode transp self callsoonthreadsafetransp processexited returncode async def createunixconnection self protocolfactory pathnone sslnone socknone serverhostnamenone sslhandshaketimeoutnone sslshutdowntimeoutnone assert serverhostname is none or isinstanceserverhostname str if ssl if serverhostname is none raise valueerror you have to pass serverhostname when using ssl else if serverhostname is not none raise valueerror serverhostname is only meaningful with ssl if sslhandshaketimeout is not none raise valueerror sslhandshaketimeout is only meaningful with ssl if sslshutdowntimeout is not none raise valueerror sslshutdowntimeout is only meaningful with ssl if path is not none if sock is not none raise valueerror path and sock can not be specified at the same time path os fspathpath sock socket socketsocket afunix socket sockstream 0 try sock setblockingfalse await self sockconnectsock path except sock close raise else if sock is none raise valueerror no path and sock were specified if sock family socket afunix or sock type socket sockstream raise valueerror f a unix domain stream socket was expected got sock r sock setblockingfalse transport protocol await self createconnectiontransport sock protocolfactory ssl serverhostname sslhandshaketimeoutsslhandshaketimeout sslshutdowntimeoutsslshutdowntimeout return transport protocol async def createunixserver self protocolfactory pathnone socknone backlog100 sslnone sslhandshaketimeoutnone sslshutdowntimeoutnone startservingtrue cleanupsockettrue if isinstancessl bool raise typeerror ssl argument must be an sslcontext or none if sslhandshaketimeout is not none and not ssl raise valueerror sslhandshaketimeout is only meaningful with ssl if sslshutdowntimeout is not none and not ssl raise valueerror sslshutdowntimeout is only meaningful with ssl if path is not none if sock is not none raise valueerror path and sock can not be specified at the same time path os fspathpath sock socket socketsocket afunix socket sockstream check for abstract socket str and bytes paths are supported if path0 not in 0 x00 try if stat sissockos statpath stmode os removepath except filenotfounderror pass except oserror as err directory may have permissions only to create socket logger error unable to check or remove stale unix socket r r path err try sock bindpath except oserror as exc sock close if exc errno errno eaddrinuse let s improve the error message by adding with what exact address it occurs msg f address path r is already in use raise oserrorerrno eaddrinuse msg from none else raise except sock close raise else if sock is none raise valueerror path was not specified and no sock specified if sock family socket afunix or sock type socket sockstream raise valueerror f a unix domain stream socket was expected got sock r if cleanupsocket path sock getsockname check for abstract socket str and bytes paths are supported if path0 not in 0 x00 try self unixserversocketssock os statpath stino except filenotfounderror pass sock setblockingfalse server baseevents serverself sock protocolfactory ssl backlog sslhandshaketimeout sslshutdowntimeout if startserving server startserving skip one loop iteration so that all loop addreader go through await tasks sleep0 return server async def socksendfilenativeself sock file offset count try os sendfile except attributeerror raise exceptions sendfilenotavailableerror os sendfile is not available try fileno file fileno except attributeerror io unsupportedoperation as err raise exceptions sendfilenotavailableerrornot a regular file try fsize os fstatfileno stsize except oserror raise exceptions sendfilenotavailableerrornot a regular file blocksize count if count else fsize if not blocksize return 0 empty file fut self createfuture self socksendfilenativeimplfut none sock fileno offset count blocksize 0 return await fut def socksendfilenativeimplself fut registeredfd sock fileno offset count blocksize totalsent fd sock fileno if registeredfd is not none remove the callback early it should be rare that the selector says the fd is ready but the call still returns eagain and i am willing to take a hit in that case in order to simplify the common case self removewriterregisteredfd if fut cancelled self socksendfileupdatefileposfileno offset totalsent return if count blocksize count totalsent if blocksize 0 self socksendfileupdatefileposfileno offset totalsent fut setresulttotalsent return on 32bit architectures truncate to 1gib to avoid overflowerror blocksize minblocksize sys maxsize2 1 try sent os sendfilefd fileno offset blocksize except blockingioerror interruptederror if registeredfd is none self sockaddcancellationcallbackfut sock self addwriterfd self socksendfilenativeimpl fut fd sock fileno offset count blocksize totalsent except oserror as exc if registeredfd is not none and exc errno errno enotconn and typeexc is not connectionerror if we have an enotconn and this isn t a first call to sendfile i e the connection was closed in the middle of the operation normalize the error to connectionerror to make it consistent across all posix systems newexc connectionerror socket is not connected errno enotconn newexc cause exc exc newexc if totalsent 0 we can get here for different reasons the main one being file is not a regular mmap2like file in which case we ll fall back on using plain send err exceptions sendfilenotavailableerror os sendfile call failed self socksendfileupdatefileposfileno offset totalsent fut setexceptionerr else self socksendfileupdatefileposfileno offset totalsent fut setexceptionexc except systemexit keyboardinterrupt raise except baseexception as exc self socksendfileupdatefileposfileno offset totalsent fut setexceptionexc else if sent 0 eof self socksendfileupdatefileposfileno offset totalsent fut setresulttotalsent else offset sent totalsent sent if registeredfd is none self sockaddcancellationcallbackfut sock self addwriterfd self socksendfilenativeimpl fut fd sock fileno offset count blocksize totalsent def socksendfileupdatefileposself fileno offset totalsent if totalsent 0 os lseekfileno offset os seekset def sockaddcancellationcallbackself fut sock def cbfut if fut cancelled fd sock fileno if fd 1 self removewriterfd fut adddonecallbackcb def stopservingself sock is this a unix socket that needs cleanup if sock in self unixserversockets path sock getsockname else path none super stopservingsock if path is not none previno self unixserversocketssock del self unixserversocketssock try if os statpath stino previno os unlinkpath except filenotfounderror pass except oserror as err logger error unable to clean up listening unix socket r r path err class unixreadpipetransporttransports readtransport maxsize 256 1024 max bytes we read in one event loop iteration def initself loop pipe protocol waiternone extranone super initextra self extra pipe pipe self loop loop self pipe pipe self fileno pipe fileno self protocol protocol self closing false self paused false mode os fstatself fileno stmode if not stat sisfifomode or stat sissockmode or stat sischrmode self pipe none self fileno none self protocol none raise valueerrorpipe transport is for pipessockets only os setblockingself fileno false self loop callsoonself protocol connectionmade self only start reading when connectionmade has been called self loop callsoonself addreader self fileno self readready if waiter is not none only wake up the waiter when connectionmade has been called self loop callsoonfutures setresultunlesscancelled waiter none def addreaderself fd callback if not self isreading return self loop addreaderfd callback def isreadingself return not self paused and not self closing def reprself info self class name if self pipe is none info append closed elif self closing info append closing info appendf fdself fileno selector getattrself loop selector none if self pipe is not none and selector is not none polling selectorevents testselectorevent selector self fileno selectors eventread if polling info append polling else info append idle elif self pipe is not none info append open else info append closed return format joininfo def readreadyself try data os readself fileno self maxsize except blockingioerror interruptederror pass except oserror as exc self fatalerrorexc fatal read error on pipe transport else if data self protocol datareceiveddata else if self loop getdebug logger infor was closed by peer self self closing true self loop removereaderself fileno self loop callsoonself protocol eofreceived self loop callsoonself callconnectionlost none def pausereadingself if not self isreading return self paused true self loop removereaderself fileno if self loop getdebug logger debugr pauses reading self def resumereadingself if self closing or not self paused return self paused false self loop addreaderself fileno self readready if self loop getdebug logger debugr resumes reading self def setprotocolself protocol self protocol protocol def getprotocolself return self protocol def isclosingself return self closing def closeself if not self closing self closenone def delself warnwarnings warn if self pipe is not none warnfunclosed transport self r resourcewarning sourceself self pipe close def fatalerrorself exc message fatal error on pipe transport should be called by exception handler only if isinstanceexc oserror and exc errno errno eio if self loop getdebug logger debugr s self message excinfotrue else self loop callexceptionhandler message message exception exc transport self protocol self protocol self closeexc def closeself exc self closing true self loop removereaderself fileno self loop callsoonself callconnectionlost exc def callconnectionlostself exc try self protocol connectionlostexc finally self pipe close self pipe none self protocol none self loop none class unixwritepipetransporttransports flowcontrolmixin transports writetransport def initself loop pipe protocol waiternone extranone super initextra loop self extra pipe pipe self pipe pipe self fileno pipe fileno self protocol protocol self buffer bytearray self connlost 0 self closing false set when close or writeeof called mode os fstatself fileno stmode ischar stat sischrmode isfifo stat sisfifomode issocket stat sissockmode if not ischar or isfifo or issocket self pipe none self fileno none self protocol none raise valueerrorpipe transport is only for pipes sockets and character devices os setblockingself fileno false self loop callsoonself protocol connectionmade self on aix the reader trick to be notified when the read end of the socket is closed only works for sockets on other platforms it works for pipes and sockets exception os x 10 4 issue 19294 if issocket or isfifo and not sys platform startswithaix only start reading when connectionmade has been called self loop callsoonself loop addreader self fileno self readready if waiter is not none only wake up the waiter when connectionmade has been called self loop callsoonfutures setresultunlesscancelled waiter none def reprself info self class name if self pipe is none info append closed elif self closing info append closing info appendf fdself fileno selector getattrself loop selector none if self pipe is not none and selector is not none polling selectorevents testselectorevent selector self fileno selectors eventwrite if polling info append polling else info append idle bufsize self getwritebuffersize info appendf bufsizebufsize elif self pipe is not none info append open else info append closed return format joininfo def getwritebuffersizeself return lenself buffer def readreadyself pipe was closed by peer if self loop getdebug logger infor was closed by peer self if self buffer self closebrokenpipeerror else self close def writeself data assert isinstancedata bytes bytearray memoryview reprdata if isinstancedata bytearray data memoryviewdata if not data return if self connlost or self closing if self connlost constants logthresholdforconnlostwrites logger warning pipe closed by peer or os writepipe data raised exception self connlost 1 return if not self buffer attempt to send it right away first try n os writeself fileno data except blockingioerror interruptederror n 0 except systemexit keyboardinterrupt raise except baseexception as exc self connlost 1 self fatalerrorexc fatal write error on pipe transport return if n lendata return elif n 0 data memoryviewdatan self loop addwriterself fileno self writeready self buffer data self maybepauseprotocol def writereadyself assert self buffer data should not be empty try n os writeself fileno self buffer except blockingioerror interruptederror pass except systemexit keyboardinterrupt raise except baseexception as exc self buffer clear self connlost 1 remove writer here fatalerror doesn t it because buffer is empty self loop removewriterself fileno self fatalerrorexc fatal write error on pipe transport else if n lenself buffer self buffer clear self loop removewriterself fileno self mayberesumeprotocol may append to buffer if self closing self loop removereaderself fileno self callconnectionlostnone return elif n 0 del self buffer n def canwriteeofself return true def writeeofself if self closing return assert self pipe self closing true if not self buffer self loop removereaderself fileno self loop callsoonself callconnectionlost none def setprotocolself protocol self protocol protocol def getprotocolself return self protocol def isclosingself return self closing def closeself if self pipe is not none and not self closing writeeof is all what we needed to close the write pipe self writeeof def delself warnwarnings warn if self pipe is not none warnfunclosed transport self r resourcewarning sourceself self pipe close def abortself self closenone def fatalerrorself exc message fatal error on pipe transport should be called by exception handler only if isinstanceexc oserror if self loop getdebug logger debugr s self message excinfotrue else self loop callexceptionhandler message message exception exc transport self protocol self protocol self closeexc def closeself excnone self closing true if self buffer self loop removewriterself fileno self buffer clear self loop removereaderself fileno self loop callsoonself callconnectionlost exc def callconnectionlostself exc try self protocol connectionlostexc finally self pipe close self pipe none self protocol none self loop none class unixsubprocesstransportbasesubprocess basesubprocesstransport def startself args shell stdin stdout stderr bufsize kwargs stdinw none if stdin subprocess pipe and sys platform startswith aix use a socket pair for stdin on aix since it does not support selecting read events on the write end of a socket which we use in order to detect closing of the other end stdin stdinw socket socketpair try self proc subprocess popen args shellshell stdinstdin stdoutstdout stderrstderr universalnewlinesfalse bufsizebufsize kwargs if stdinw is not none stdin close self proc stdin openstdinw detach wb bufferingbufsize stdinw none finally if stdinw is not none stdin close stdinw close class abstractchildwatcher def initsubclasscls none if cls module name warnings deprecatedabstractchildwatcher name r is deprecated as of python 3 12 and will be removed in python remove remove3 14 def addchildhandlerself pid callback args raise notimplementederror def removechildhandlerself pid attach the watcher to an event loop if the watcher was previously attached to an event loop then it is first detached before attaching to the new loop note loop may be none close the watcher this must be called to make sure that any underlying resource is freed return true if the watcher is active and is used by the event loop return true if the watcher is installed and ready to handle process exit notifications enter the watcher s context and allow starting new processes this function must return self raise notimplementederror def exitself a b c child watcher implementation using linux s pid file descriptors this child watcher polls process file descriptors pidfds to await child process termination in some respects pidfdchildwatcher is a goldilocks child watcher implementation it doesn t require signals or threads doesn t interfere with any processes launched outside the event loop and scales linearly with the number of subprocesses launched by the event loop the main disadvantage is that pidfds are specific to linux and only work on recent 5 3 kernels the child process is already reaped may happen if waitpid is called elsewhere asyncio never calls removechildhandler the method is noop but is implemented because abstract base classes require it prevent a race condition in case a child terminated during the switch self loop should always be available here as sigchld is added as a signal handler in attachloop safe child watcher implementation this implementation avoids disrupting other code spawning processes by polling explicitly each process in the sigchld handler instead of calling os waitpid1 this is a safe solution but it has a significant overhead when handling a big number of children on each time sigchld is raised prevent a race condition in case the child is already terminated the child process is already reaped may happen if waitpid is called elsewhere the child process is still alive may happen if removechildhandler is called after os waitpid returns fast child watcher implementation this implementation reaps every terminated processes by calling os waitpid1 directly possibly breaking other code spawning processes and waiting for their termination there is no noticeable overhead when handling a big number of children o1 each time a child terminates the child is running the child is dead already we can fire the callback because of signal coalescing we must keep calling waitpid as long as we re able to reap a child no more child processes exist a child process is still alive unknown child it may not be registered yet a watcher that doesn t require running loop in the main thread this implementation registers a sigchld signal handler on instantiation which may conflict with other code that install own handler for this signal the solution is safe but it has a significant overhead when handling a big number of processes on each time a sigchld is received implementation note the class keeps compatibility with abstractchildwatcher abc to achieve this it has empty attachloop method and doesn t accept explicit loop argument for addchildhandlerremovechildhandler but retrieves the current loop by getrunningloop prevent a race condition in case the child is already terminated don t save the loop but initialize itself if called first time the reason to do it here is that attachloop is called from unix policy only for the main thread main thread is required for subscription on sigchld signal set sarestart to limit eintr occurrences the child process is already reaped may happen if waitpid is called elsewhere the child process is still alive may happen if removechildhandler is called after os waitpid returns threaded child watcher implementation the watcher uses a thread per process for waiting for the process finish it doesn t require subscription on posix signal but a thread creation is not free the watcher has o1 complexity its performance doesn t depend on amount of spawn processes asyncio never calls removechildhandler the method is noop but is implemented because abstract base classes require it the child process is already reaped may happen if waitpid is called elsewhere blocked by security policy like seccomp unix event loop policy with a watcher for child processes loopfactory unixselectoreventloop def initself super init self watcher none def initwatcherself with events lock if self watcher is none pragma no branch if canusepidfd self watcher pidfdchildwatcher else self watcher threadedchildwatcher def seteventloopself loop super seteventlooploop if self watcher is not none and threading currentthread is threading mainthread self watcher attachlooploop def getchildwatcherself if self watcher is none self initwatcher warnings deprecatedgetchildwatcher name r is deprecated as of python 3 12 and will be removed in python remove remove3 14 return self watcher def setchildwatcherself watcher pragma no cover dummy signal handler 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 unix event loop adds signal handling and unix domain socket support to selectoreventloop ignore null bytes written by _write_to_self 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 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 a signal handler register a dummy signal handler to ask python to write the signal number in the wakeup file descriptor _process_self_data will read signal numbers from this file descriptor to handle signals set sa_restart to limit eintr occurrences internal helper that is the actual signal handler assume it s some race condition remove it properly remove a handler for a signal unix only return true if a signal handler was removed false if not internal helper to validate a signal raise valueerror if the signal number is invalid or uncatchable raise runtimeerror if there is a problem setting up the handler check early raising exception before process creation prevents subprocess execution if the watcher is not ready to handle it check for abstract socket str and bytes paths are supported directory may have permissions only to create socket let s improve the error message by adding with what exact address it occurs check for abstract socket str and bytes paths are supported skip one loop iteration so that all loop add_reader go through empty file remove the callback early it should be rare that the selector says the fd is ready but the call still returns eagain and i am willing to take a hit in that case in order to simplify the common case on 32 bit architectures truncate to 1gib to avoid overflowerror if we have an enotconn and this isn t a first call to sendfile i e the connection was closed in the middle of the operation normalize the error to connectionerror to make it consistent across all posix systems we can get here for different reasons the main one being file is not a regular mmap 2 like file in which case we ll fall back on using plain send eof is this a unix socket that needs cleanup max bytes we read in one event loop iteration only start reading when connection_made has been called only wake up the waiter when connection_made has been called should be called by exception handler only set when close or write_eof called on aix the reader trick to be notified when the read end of the socket is closed only works for sockets on other platforms it works for pipes and sockets exception os x 10 4 issue 19294 only start reading when connection_made has been called only wake up the waiter when connection_made has been called pipe was closed by peer attempt to send it right away first remove writer here _fatal_error doesn t it because _buffer is empty may append to buffer write_eof is all what we needed to close the write pipe should be called by exception handler only use a socket pair for stdin on aix since it does not support selecting read events on the write end of a socket which we use in order to detect closing of the other end abstract base class for monitoring child processes objects derived from this class monitor a collection of subprocesses and report their termination or interruption by a signal new callbacks are registered with add_child_handler starting a new process must be done within a with block to allow the watcher to suspend its activity until the new process if fully registered this is needed to prevent a race condition in some implementations example with watcher proc subprocess popen sleep 1 watcher add_child_handler proc pid callback notes implementations of this class must be thread safe since child watcher objects may catch the sigchld signal and call waitpid 1 there should be only one active object per process register a new child handler arrange for callback pid returncode args to be called when process pid terminates specifying another callback for the same process replaces the previous handler note callback must be thread safe removes the handler for process pid the function returns true if the handler was successfully removed false if there was nothing to remove attach the watcher to an event loop if the watcher was previously attached to an event loop then it is first detached before attaching to the new loop note loop may be none close the watcher this must be called to make sure that any underlying resource is freed return true if the watcher is active and is used by the event loop return true if the watcher is installed and ready to handle process exit notifications enter the watcher s context and allow starting new processes this function must return self exit the watcher s context child watcher implementation using linux s pid file descriptors this child watcher polls process file descriptors pidfds to await child process termination in some respects pidfdchildwatcher is a goldilocks child watcher implementation it doesn t require signals or threads doesn t interfere with any processes launched outside the event loop and scales linearly with the number of subprocesses launched by the event loop the main disadvantage is that pidfds are specific to linux and only work on recent 5 3 kernels the child process is already reaped may happen if waitpid is called elsewhere asyncio never calls remove_child_handler the method is no op but is implemented because abstract base classes require it prevent a race condition in case a child terminated during the switch self _loop should always be available here as _sig_chld is added as a signal handler in attach_loop safe child watcher implementation this implementation avoids disrupting other code spawning processes by polling explicitly each process in the sigchld handler instead of calling os waitpid 1 this is a safe solution but it has a significant overhead when handling a big number of children o n each time sigchld is raised prevent a race condition in case the child is already terminated the child process is already reaped may happen if waitpid is called elsewhere the child process is still alive pragma no cover may happen if remove_child_handler is called after os waitpid returns fast child watcher implementation this implementation reaps every terminated processes by calling os waitpid 1 directly possibly breaking other code spawning processes and waiting for their termination there is no noticeable overhead when handling a big number of children o 1 each time a child terminates the child is running the child is dead already we can fire the callback because of signal coalescing we must keep calling waitpid as long as we re able to reap a child no more child processes exist a child process is still alive unknown child it may not be registered yet a watcher that doesn t require running loop in the main thread this implementation registers a sigchld signal handler on instantiation which may conflict with other code that install own handler for this signal the solution is safe but it has a significant overhead when handling a big number of processes o n each time a sigchld is received implementation note the class keeps compatibility with abstractchildwatcher abc to achieve this it has empty attach_loop method and doesn t accept explicit loop argument for add_child_handler remove_child_handler but retrieves the current loop by get_running_loop prevent a race condition in case the child is already terminated don t save the loop but initialize itself if called first time the reason to do it here is that attach_loop is called from unix policy only for the main thread main thread is required for subscription on sigchld signal set sa_restart to limit eintr occurrences the child process is already reaped may happen if waitpid is called elsewhere the child process is still alive pragma no cover may happen if remove_child_handler is called after os waitpid returns threaded child watcher implementation the watcher uses a thread per process for waiting for the process finish it doesn t require subscription on posix signal but a thread creation is not free the watcher has o 1 complexity its performance doesn t depend on amount of spawn processes asyncio never calls remove_child_handler the method is no op but is implemented because abstract base classes require it the child process is already reaped may happen if waitpid is called elsewhere blocked by security policy like seccomp unix event loop policy with a watcher for child processes pragma no branch set the event loop as a side effect if a child watcher was set before then calling set_event_loop from the main thread will call attach_loop loop on the child watcher get the watcher for child processes if not yet set a threadedchildwatcher object is automatically created set the watcher for child processes
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',
    'EventLoop',
)


if sys.platform == 'win32':  
    raise ImportError('Signals are not really supported on Windows')


def _sighandler_noop(signum, frame):
    
    pass


def waitstatus_to_exitcode(status):
    try:
        return os.waitstatus_to_exitcode(status)
    except ValueError:
        
        
        
        return status


class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
    

    def __init__(self, selector=None):
        super().__init__(selector)
        self._signal_handlers = {}
        self._unix_server_sockets = {}

    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:
                
                continue
            self._handle_signal(signum)

    def add_signal_handler(self, sig, callback, *args):
        
        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:
            
            
            
            
            signal.set_wakeup_fd(self._csock.fileno())
        except (ValueError, OSError) as exc:
            raise RuntimeError(str(exc))

        handle = events.Handle(callback, args, self, None)
        self._signal_handlers[sig] = handle

        try:
            
            
            
            signal.signal(sig, _sighandler_noop)

            
            signal.siginterrupt(sig, False)
        except OSError as exc:
            del self._signal_handlers[sig]
            if not self._signal_handlers:
                try:
                    signal.set_wakeup_fd(-1)
                except (ValueError, OSError) as nexc:
                    logger.info('set_wakeup_fd(-1) failed: %s', nexc)

            if exc.errno == errno.EINVAL:
                raise RuntimeError(f'sig {sig} cannot be caught')
            else:
                raise

    def _handle_signal(self, sig):
        
        handle = self._signal_handlers.get(sig)
        if handle is None:
            return  
        if handle._cancelled:
            self.remove_signal_handler(sig)  
        else:
            self._add_callback_signalsafe(handle)

    def remove_signal_handler(self, sig):
        
        self._check_signal(sig)
        try:
            del self._signal_handlers[sig]
        except KeyError:
            return False

        if sig == signal.SIGINT:
            handler = signal.default_int_handler
        else:
            handler = signal.SIG_DFL

        try:
            signal.signal(sig, handler)
        except OSError as exc:
            if exc.errno == errno.EINVAL:
                raise RuntimeError(f'sig {sig} cannot be caught')
            else:
                raise

        if not self._signal_handlers:
            try:
                signal.set_wakeup_fd(-1)
            except (ValueError, OSError) as exc:
                logger.info('set_wakeup_fd(-1) failed: %s', exc)

        return True

    def _check_signal(self, sig):
        
        if not isinstance(sig, int):
            raise TypeError(f'sig must be an int, not {sig!r}')

        if sig not in signal.valid_signals():
            raise ValueError(f'invalid signal number {sig}')

    def _make_read_pipe_transport(self, pipe, protocol, waiter=None,
                                  extra=None):
        return _UnixReadPipeTransport(self, pipe, protocol, waiter, extra)

    def _make_write_pipe_transport(self, pipe, protocol, waiter=None,
                                   extra=None):
        return _UnixWritePipeTransport(self, pipe, protocol, waiter, extra)

    async def _make_subprocess_transport(self, protocol, args, shell,
                                         stdin, stdout, stderr, bufsize,
                                         extra=None, **kwargs):
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', DeprecationWarning)
            watcher = events.get_child_watcher()

        with watcher:
            if not watcher.is_active():
                
                
                
                
                raise RuntimeError("asyncio.get_child_watcher() is not activated, "
                                "subprocess support is not installed.")
            waiter = self.create_future()
            transp = _UnixSubprocessTransport(self, protocol, args, shell,
                                            stdin, stdout, stderr, bufsize,
                                            waiter=waiter, extra=extra,
                                            **kwargs)
            watcher.add_child_handler(transp.get_pid(),
                                    self._child_watcher_callback, transp)
            try:
                await waiter
            except (SystemExit, KeyboardInterrupt):
                raise
            except BaseException:
                transp.close()
                await transp._wait()
                raise

        return transp

    def _child_watcher_callback(self, pid, returncode, transp):
        self.call_soon_threadsafe(transp._process_exited, returncode)

    async def create_unix_connection(
            self, protocol_factory, path=None, *,
            ssl=None, sock=None,
            server_hostname=None,
            ssl_handshake_timeout=None,
            ssl_shutdown_timeout=None):
        assert server_hostname is None or isinstance(server_hostname, str)
        if ssl:
            if server_hostname is None:
                raise ValueError(
                    'you have to pass server_hostname when using ssl')
        else:
            if server_hostname is not None:
                raise ValueError('server_hostname is only meaningful with ssl')
            if ssl_handshake_timeout is not None:
                raise ValueError(
                    'ssl_handshake_timeout is only meaningful with ssl')
            if ssl_shutdown_timeout is not None:
                raise ValueError(
                    'ssl_shutdown_timeout is only meaningful with ssl')

        if path is not None:
            if sock is not None:
                raise ValueError(
                    'path and sock can not be specified at the same time')

            path = os.fspath(path)
            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM, 0)
            try:
                sock.setblocking(False)
                await self.sock_connect(sock, path)
            except:
                sock.close()
                raise

        else:
            if sock is None:
                raise ValueError('no path and sock were specified')
            if (sock.family != socket.AF_UNIX or
                    sock.type != socket.SOCK_STREAM):
                raise ValueError(
                    f'A UNIX Domain Stream Socket was expected, got {sock!r}')
            sock.setblocking(False)

        transport, protocol = await self._create_connection_transport(
            sock, protocol_factory, ssl, server_hostname,
            ssl_handshake_timeout=ssl_handshake_timeout,
            ssl_shutdown_timeout=ssl_shutdown_timeout)
        return transport, protocol

    async def create_unix_server(
            self, protocol_factory, path=None, *,
            sock=None, backlog=100, ssl=None,
            ssl_handshake_timeout=None,
            ssl_shutdown_timeout=None,
            start_serving=True, cleanup_socket=True):
        if isinstance(ssl, bool):
            raise TypeError('ssl argument must be an SSLContext or None')

        if ssl_handshake_timeout is not None and not ssl:
            raise ValueError(
                'ssl_handshake_timeout is only meaningful with ssl')

        if ssl_shutdown_timeout is not None and not ssl:
            raise ValueError(
                'ssl_shutdown_timeout is only meaningful with ssl')

        if path is not None:
            if sock is not None:
                raise ValueError(
                    'path and sock can not be specified at the same time')

            path = os.fspath(path)
            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)

            
            if path[0] not in (0, '\x00'):
                try:
                    if stat.S_ISSOCK(os.stat(path).st_mode):
                        os.remove(path)
                except FileNotFoundError:
                    pass
                except OSError as err:
                    
                    logger.error('Unable to check or remove stale UNIX socket '
                                 '%r: %r', path, err)

            try:
                sock.bind(path)
            except OSError as exc:
                sock.close()
                if exc.errno == errno.EADDRINUSE:
                    
                    
                    msg = f'Address {path!r} is already in use'
                    raise OSError(errno.EADDRINUSE, msg) from None
                else:
                    raise
            except:
                sock.close()
                raise
        else:
            if sock is None:
                raise ValueError(
                    'path was not specified, and no sock specified')

            if (sock.family != socket.AF_UNIX or
                    sock.type != socket.SOCK_STREAM):
                raise ValueError(
                    f'A UNIX Domain Stream Socket was expected, got {sock!r}')

        if cleanup_socket:
            path = sock.getsockname()
            
            if path[0] not in (0, '\x00'):
                try:
                    self._unix_server_sockets[sock] = os.stat(path).st_ino
                except FileNotFoundError:
                    pass

        sock.setblocking(False)
        server = base_events.Server(self, [sock], protocol_factory,
                                    ssl, backlog, ssl_handshake_timeout,
                                    ssl_shutdown_timeout)
        if start_serving:
            server._start_serving()
            
            
            await tasks.sleep(0)

        return server

    async def _sock_sendfile_native(self, sock, file, offset, count):
        try:
            os.sendfile
        except AttributeError:
            raise exceptions.SendfileNotAvailableError(
                "os.sendfile() is not available")
        try:
            fileno = file.fileno()
        except (AttributeError, io.UnsupportedOperation) as err:
            raise exceptions.SendfileNotAvailableError("not a regular file")
        try:
            fsize = os.fstat(fileno).st_size
        except OSError:
            raise exceptions.SendfileNotAvailableError("not a regular file")
        blocksize = count if count else fsize
        if not blocksize:
            return 0  

        fut = self.create_future()
        self._sock_sendfile_native_impl(fut, None, sock, fileno,
                                        offset, count, blocksize, 0)
        return await fut

    def _sock_sendfile_native_impl(self, fut, registered_fd, sock, fileno,
                                   offset, count, blocksize, total_sent):
        fd = sock.fileno()
        if registered_fd is not None:
            
            
            
            
            self.remove_writer(registered_fd)
        if fut.cancelled():
            self._sock_sendfile_update_filepos(fileno, offset, total_sent)
            return
        if count:
            blocksize = count - total_sent
            if blocksize <= 0:
                self._sock_sendfile_update_filepos(fileno, offset, total_sent)
                fut.set_result(total_sent)
                return

        
        blocksize = min(blocksize, sys.maxsize//2 + 1)

        try:
            sent = os.sendfile(fd, fileno, offset, blocksize)
        except (BlockingIOError, InterruptedError):
            if registered_fd is None:
                self._sock_add_cancellation_callback(fut, sock)
            self.add_writer(fd, self._sock_sendfile_native_impl, fut,
                            fd, sock, fileno,
                            offset, count, blocksize, total_sent)
        except OSError as exc:
            if (registered_fd is not None and
                    exc.errno == errno.ENOTCONN and
                    type(exc) is not ConnectionError):
                
                
                
                
                new_exc = ConnectionError(
                    "socket is not connected", errno.ENOTCONN)
                new_exc.__cause__ = exc
                exc = new_exc
            if total_sent == 0:
                
                
                
                
                err = exceptions.SendfileNotAvailableError(
                    "os.sendfile call failed")
                self._sock_sendfile_update_filepos(fileno, offset, total_sent)
                fut.set_exception(err)
            else:
                self._sock_sendfile_update_filepos(fileno, offset, total_sent)
                fut.set_exception(exc)
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException as exc:
            self._sock_sendfile_update_filepos(fileno, offset, total_sent)
            fut.set_exception(exc)
        else:
            if sent == 0:
                
                self._sock_sendfile_update_filepos(fileno, offset, total_sent)
                fut.set_result(total_sent)
            else:
                offset += sent
                total_sent += sent
                if registered_fd is None:
                    self._sock_add_cancellation_callback(fut, sock)
                self.add_writer(fd, self._sock_sendfile_native_impl, fut,
                                fd, sock, fileno,
                                offset, count, blocksize, total_sent)

    def _sock_sendfile_update_filepos(self, fileno, offset, total_sent):
        if total_sent > 0:
            os.lseek(fileno, offset, os.SEEK_SET)

    def _sock_add_cancellation_callback(self, fut, sock):
        def cb(fut):
            if fut.cancelled():
                fd = sock.fileno()
                if fd != -1:
                    self.remove_writer(fd)
        fut.add_done_callback(cb)

    def _stop_serving(self, sock):
        
        if sock in self._unix_server_sockets:
            path = sock.getsockname()
        else:
            path = None

        super()._stop_serving(sock)

        if path is not None:
            prev_ino = self._unix_server_sockets[sock]
            del self._unix_server_sockets[sock]
            try:
                if os.stat(path).st_ino == prev_ino:
                    os.unlink(path)
            except FileNotFoundError:
                pass
            except OSError as err:
                logger.error('Unable to clean up listening UNIX socket '
                             '%r: %r', path, err)


class _UnixReadPipeTransport(transports.ReadTransport):

    max_size = 256 * 1024  

    def __init__(self, loop, pipe, protocol, waiter=None, extra=None):
        super().__init__(extra)
        self._extra['pipe'] = pipe
        self._loop = loop
        self._pipe = pipe
        self._fileno = pipe.fileno()
        self._protocol = protocol
        self._closing = False
        self._paused = False

        mode = os.fstat(self._fileno).st_mode
        if not (stat.S_ISFIFO(mode) or
                stat.S_ISSOCK(mode) or
                stat.S_ISCHR(mode)):
            self._pipe = None
            self._fileno = None
            self._protocol = None
            raise ValueError("Pipe transport is for pipes/sockets only.")

        os.set_blocking(self._fileno, False)

        self._loop.call_soon(self._protocol.connection_made, self)
        
        self._loop.call_soon(self._add_reader,
                             self._fileno, self._read_ready)
        if waiter is not None:
            
            self._loop.call_soon(futures._set_result_unless_cancelled,
                                 waiter, None)

    def _add_reader(self, fd, callback):
        if not self.is_reading():
            return
        self._loop._add_reader(fd, callback)

    def is_reading(self):
        return not self._paused and not self._closing

    def __repr__(self):
        info = [self.__class__.__name__]
        if self._pipe is None:
            info.append('closed')
        elif self._closing:
            info.append('closing')
        info.append(f'fd={self._fileno}')
        selector = getattr(self._loop, '_selector', None)
        if self._pipe is not None and selector is not None:
            polling = selector_events._test_selector_event(
                selector, self._fileno, selectors.EVENT_READ)
            if polling:
                info.append('polling')
            else:
                info.append('idle')
        elif self._pipe is not None:
            info.append('open')
        else:
            info.append('closed')
        return '<{}>'.format(' '.join(info))

    def _read_ready(self):
        try:
            data = os.read(self._fileno, self.max_size)
        except (BlockingIOError, InterruptedError):
            pass
        except OSError as exc:
            self._fatal_error(exc, 'Fatal read error on pipe transport')
        else:
            if data:
                self._protocol.data_received(data)
            else:
                if self._loop.get_debug():
                    logger.info("%r was closed by peer", self)
                self._closing = True
                self._loop._remove_reader(self._fileno)
                self._loop.call_soon(self._protocol.eof_received)
                self._loop.call_soon(self._call_connection_lost, None)

    def pause_reading(self):
        if not self.is_reading():
            return
        self._paused = True
        self._loop._remove_reader(self._fileno)
        if self._loop.get_debug():
            logger.debug("%r pauses reading", self)

    def resume_reading(self):
        if self._closing or not self._paused:
            return
        self._paused = False
        self._loop._add_reader(self._fileno, self._read_ready)
        if self._loop.get_debug():
            logger.debug("%r resumes reading", self)

    def set_protocol(self, protocol):
        self._protocol = protocol

    def get_protocol(self):
        return self._protocol

    def is_closing(self):
        return self._closing

    def close(self):
        if not self._closing:
            self._close(None)

    def __del__(self, _warn=warnings.warn):
        if self._pipe is not None:
            _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
            self._pipe.close()

    def _fatal_error(self, exc, message='Fatal error on pipe transport'):
        
        if (isinstance(exc, OSError) and exc.errno == errno.EIO):
            if self._loop.get_debug():
                logger.debug("%r: %s", self, message, exc_info=True)
        else:
            self._loop.call_exception_handler({
                'message': message,
                'exception': exc,
                'transport': self,
                'protocol': self._protocol,
            })
        self._close(exc)

    def _close(self, exc):
        self._closing = True
        self._loop._remove_reader(self._fileno)
        self._loop.call_soon(self._call_connection_lost, exc)

    def _call_connection_lost(self, exc):
        try:
            self._protocol.connection_lost(exc)
        finally:
            self._pipe.close()
            self._pipe = None
            self._protocol = None
            self._loop = None


class _UnixWritePipeTransport(transports._FlowControlMixin,
                              transports.WriteTransport):

    def __init__(self, loop, pipe, protocol, waiter=None, extra=None):
        super().__init__(extra, loop)
        self._extra['pipe'] = pipe
        self._pipe = pipe
        self._fileno = pipe.fileno()
        self._protocol = protocol
        self._buffer = bytearray()
        self._conn_lost = 0
        self._closing = False  

        mode = os.fstat(self._fileno).st_mode
        is_char = stat.S_ISCHR(mode)
        is_fifo = stat.S_ISFIFO(mode)
        is_socket = stat.S_ISSOCK(mode)
        if not (is_char or is_fifo or is_socket):
            self._pipe = None
            self._fileno = None
            self._protocol = None
            raise ValueError("Pipe transport is only for "
                             "pipes, sockets and character devices")

        os.set_blocking(self._fileno, False)
        self._loop.call_soon(self._protocol.connection_made, self)

        
        
        
        if is_socket or (is_fifo and not sys.platform.startswith("aix")):
            
            self._loop.call_soon(self._loop._add_reader,
                                 self._fileno, self._read_ready)

        if waiter is not None:
            
            self._loop.call_soon(futures._set_result_unless_cancelled,
                                 waiter, None)

    def __repr__(self):
        info = [self.__class__.__name__]
        if self._pipe is None:
            info.append('closed')
        elif self._closing:
            info.append('closing')
        info.append(f'fd={self._fileno}')
        selector = getattr(self._loop, '_selector', None)
        if self._pipe is not None and selector is not None:
            polling = selector_events._test_selector_event(
                selector, self._fileno, selectors.EVENT_WRITE)
            if polling:
                info.append('polling')
            else:
                info.append('idle')

            bufsize = self.get_write_buffer_size()
            info.append(f'bufsize={bufsize}')
        elif self._pipe is not None:
            info.append('open')
        else:
            info.append('closed')
        return '<{}>'.format(' '.join(info))

    def get_write_buffer_size(self):
        return len(self._buffer)

    def _read_ready(self):
        
        if self._loop.get_debug():
            logger.info("%r was closed by peer", self)
        if self._buffer:
            self._close(BrokenPipeError())
        else:
            self._close()

    def write(self, data):
        assert isinstance(data, (bytes, bytearray, memoryview)), repr(data)
        if isinstance(data, bytearray):
            data = memoryview(data)
        if not data:
            return

        if self._conn_lost or self._closing:
            if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
                logger.warning('pipe closed by peer or '
                               'os.write(pipe, data) raised exception.')
            self._conn_lost += 1
            return

        if not self._buffer:
            
            try:
                n = os.write(self._fileno, data)
            except (BlockingIOError, InterruptedError):
                n = 0
            except (SystemExit, KeyboardInterrupt):
                raise
            except BaseException as exc:
                self._conn_lost += 1
                self._fatal_error(exc, 'Fatal write error on pipe transport')
                return
            if n == len(data):
                return
            elif n > 0:
                data = memoryview(data)[n:]
            self._loop._add_writer(self._fileno, self._write_ready)

        self._buffer += data
        self._maybe_pause_protocol()

    def _write_ready(self):
        assert self._buffer, 'Data should not be empty'

        try:
            n = os.write(self._fileno, self._buffer)
        except (BlockingIOError, InterruptedError):
            pass
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException as exc:
            self._buffer.clear()
            self._conn_lost += 1
            
            
            self._loop._remove_writer(self._fileno)
            self._fatal_error(exc, 'Fatal write error on pipe transport')
        else:
            if n == len(self._buffer):
                self._buffer.clear()
                self._loop._remove_writer(self._fileno)
                self._maybe_resume_protocol()  
                if self._closing:
                    self._loop._remove_reader(self._fileno)
                    self._call_connection_lost(None)
                return
            elif n > 0:
                del self._buffer[:n]

    def can_write_eof(self):
        return True

    def write_eof(self):
        if self._closing:
            return
        assert self._pipe
        self._closing = True
        if not self._buffer:
            self._loop._remove_reader(self._fileno)
            self._loop.call_soon(self._call_connection_lost, None)

    def set_protocol(self, protocol):
        self._protocol = protocol

    def get_protocol(self):
        return self._protocol

    def is_closing(self):
        return self._closing

    def close(self):
        if self._pipe is not None and not self._closing:
            
            self.write_eof()

    def __del__(self, _warn=warnings.warn):
        if self._pipe is not None:
            _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
            self._pipe.close()

    def abort(self):
        self._close(None)

    def _fatal_error(self, exc, message='Fatal error on pipe transport'):
        
        if isinstance(exc, OSError):
            if self._loop.get_debug():
                logger.debug("%r: %s", self, message, exc_info=True)
        else:
            self._loop.call_exception_handler({
                'message': message,
                'exception': exc,
                'transport': self,
                'protocol': self._protocol,
            })
        self._close(exc)

    def _close(self, exc=None):
        self._closing = True
        if self._buffer:
            self._loop._remove_writer(self._fileno)
        self._buffer.clear()
        self._loop._remove_reader(self._fileno)
        self._loop.call_soon(self._call_connection_lost, exc)

    def _call_connection_lost(self, exc):
        try:
            self._protocol.connection_lost(exc)
        finally:
            self._pipe.close()
            self._pipe = None
            self._protocol = None
            self._loop = None


class _UnixSubprocessTransport(base_subprocess.BaseSubprocessTransport):

    def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
        stdin_w = None
        if stdin == subprocess.PIPE and sys.platform.startswith('aix'):
            
            
            
            
            stdin, stdin_w = socket.socketpair()
        try:
            self._proc = subprocess.Popen(
                args, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr,
                universal_newlines=False, bufsize=bufsize, **kwargs)
            if stdin_w is not None:
                stdin.close()
                self._proc.stdin = open(stdin_w.detach(), 'wb', buffering=bufsize)
                stdin_w = None
        finally:
            if stdin_w is not None:
                stdin.close()
                stdin_w.close()


class AbstractChildWatcher:
    

    def __init_subclass__(cls) -> None:
        if cls.__module__ != __name__:
            warnings._deprecated("AbstractChildWatcher",
                             "{name!r} is deprecated as of Python 3.12 and will be "
                             "removed in Python {remove}.",
                              remove=(3, 14))

    def add_child_handler(self, pid, callback, *args):
        
        raise NotImplementedError()

    def remove_child_handler(self, pid):
        

        raise NotImplementedError()

    def attach_loop(self, loop):
        
        raise NotImplementedError()

    def close(self):
        
        raise NotImplementedError()

    def is_active(self):
        
        raise NotImplementedError()

    def __enter__(self):
        
        raise NotImplementedError()

    def __exit__(self, a, b, c):
        
        raise NotImplementedError()


class PidfdChildWatcher(AbstractChildWatcher):
    

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, exc_traceback):
        pass

    def is_active(self):
        return True

    def close(self):
        pass

    def attach_loop(self, loop):
        pass

    def add_child_handler(self, pid, callback, *args):
        loop = events.get_running_loop()
        pidfd = os.pidfd_open(pid)
        loop._add_reader(pidfd, self._do_wait, pid, pidfd, callback, args)

    def _do_wait(self, pid, pidfd, callback, args):
        loop = events.get_running_loop()
        loop._remove_reader(pidfd)
        try:
            _, status = os.waitpid(pid, 0)
        except ChildProcessError:
            
            
            returncode = 255
            logger.warning(
                "child process pid %d exit status already read: "
                " will report returncode 255",
                pid)
        else:
            returncode = waitstatus_to_exitcode(status)

        os.close(pidfd)
        callback(pid, returncode, *args)

    def remove_child_handler(self, pid):
        
        
        
        return True


class BaseChildWatcher(AbstractChildWatcher):

    def __init__(self):
        self._loop = None
        self._callbacks = {}

    def close(self):
        self.attach_loop(None)

    def is_active(self):
        return self._loop is not None and self._loop.is_running()

    def _do_waitpid(self, expected_pid):
        raise NotImplementedError()

    def _do_waitpid_all(self):
        raise NotImplementedError()

    def attach_loop(self, loop):
        assert loop is None or isinstance(loop, events.AbstractEventLoop)

        if self._loop is not None and loop is None and self._callbacks:
            warnings.warn(
                'A loop is being detached '
                'from a child watcher with pending handlers',
                RuntimeWarning)

        if self._loop is not None:
            self._loop.remove_signal_handler(signal.SIGCHLD)

        self._loop = loop
        if loop is not None:
            loop.add_signal_handler(signal.SIGCHLD, self._sig_chld)

            
            
            self._do_waitpid_all()

    def _sig_chld(self):
        try:
            self._do_waitpid_all()
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException as exc:
            
            
            
            self._loop.call_exception_handler({
                'message': 'Unknown exception in SIGCHLD handler',
                'exception': exc,
            })


class SafeChildWatcher(BaseChildWatcher):
    

    def __init__(self):
        super().__init__()
        warnings._deprecated("SafeChildWatcher",
                             "{name!r} is deprecated as of Python 3.12 and will be "
                             "removed in Python {remove}.",
                              remove=(3, 14))

    def close(self):
        self._callbacks.clear()
        super().close()

    def __enter__(self):
        return self

    def __exit__(self, a, b, c):
        pass

    def add_child_handler(self, pid, callback, *args):
        self._callbacks[pid] = (callback, args)

        
        self._do_waitpid(pid)

    def remove_child_handler(self, pid):
        try:
            del self._callbacks[pid]
            return True
        except KeyError:
            return False

    def _do_waitpid_all(self):

        for pid in list(self._callbacks):
            self._do_waitpid(pid)

    def _do_waitpid(self, expected_pid):
        assert expected_pid > 0

        try:
            pid, status = os.waitpid(expected_pid, os.WNOHANG)
        except ChildProcessError:
            
            
            pid = expected_pid
            returncode = 255
            logger.warning(
                "Unknown child process pid %d, will report returncode 255",
                pid)
        else:
            if pid == 0:
                
                return

            returncode = waitstatus_to_exitcode(status)
            if self._loop.get_debug():
                logger.debug('process %s exited with returncode %s',
                             expected_pid, returncode)

        try:
            callback, args = self._callbacks.pop(pid)
        except KeyError:  
            
            
            if self._loop.get_debug():
                logger.warning("Child watcher got an unexpected pid: %r",
                               pid, exc_info=True)
        else:
            callback(pid, returncode, *args)


class FastChildWatcher(BaseChildWatcher):
    
    def __init__(self):
        super().__init__()
        self._lock = threading.Lock()
        self._zombies = {}
        self._forks = 0
        warnings._deprecated("FastChildWatcher",
                             "{name!r} is deprecated as of Python 3.12 and will be "
                             "removed in Python {remove}.",
                              remove=(3, 14))

    def close(self):
        self._callbacks.clear()
        self._zombies.clear()
        super().close()

    def __enter__(self):
        with self._lock:
            self._forks += 1

            return self

    def __exit__(self, a, b, c):
        with self._lock:
            self._forks -= 1

            if self._forks or not self._zombies:
                return

            collateral_victims = str(self._zombies)
            self._zombies.clear()

        logger.warning(
            "Caught subprocesses termination from unknown pids: %s",
            collateral_victims)

    def add_child_handler(self, pid, callback, *args):
        assert self._forks, "Must use the context manager"

        with self._lock:
            try:
                returncode = self._zombies.pop(pid)
            except KeyError:
                
                self._callbacks[pid] = callback, args
                return

        
        callback(pid, returncode, *args)

    def remove_child_handler(self, pid):
        try:
            del self._callbacks[pid]
            return True
        except KeyError:
            return False

    def _do_waitpid_all(self):
        
        
        while True:
            try:
                pid, status = os.waitpid(-1, os.WNOHANG)
            except ChildProcessError:
                
                return
            else:
                if pid == 0:
                    
                    return

                returncode = waitstatus_to_exitcode(status)

            with self._lock:
                try:
                    callback, args = self._callbacks.pop(pid)
                except KeyError:
                    
                    if self._forks:
                        
                        self._zombies[pid] = returncode
                        if self._loop.get_debug():
                            logger.debug('unknown process %s exited '
                                         'with returncode %s',
                                         pid, returncode)
                        continue
                    callback = None
                else:
                    if self._loop.get_debug():
                        logger.debug('process %s exited with returncode %s',
                                     pid, returncode)

            if callback is None:
                logger.warning(
                    "Caught subprocess termination from unknown pid: "
                    "%d -> %d", pid, returncode)
            else:
                callback(pid, returncode, *args)


class MultiLoopChildWatcher(AbstractChildWatcher):
    

    
    
    
    
    
    

    def __init__(self):
        self._callbacks = {}
        self._saved_sighandler = None
        warnings._deprecated("MultiLoopChildWatcher",
                             "{name!r} is deprecated as of Python 3.12 and will be "
                             "removed in Python {remove}.",
                              remove=(3, 14))

    def is_active(self):
        return self._saved_sighandler is not None

    def close(self):
        self._callbacks.clear()
        if self._saved_sighandler is None:
            return

        handler = signal.getsignal(signal.SIGCHLD)
        if handler != self._sig_chld:
            logger.warning("SIGCHLD handler was changed by outside code")
        else:
            signal.signal(signal.SIGCHLD, self._saved_sighandler)
        self._saved_sighandler = None

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

    def add_child_handler(self, pid, callback, *args):
        loop = events.get_running_loop()
        self._callbacks[pid] = (loop, callback, args)

        
        self._do_waitpid(pid)

    def remove_child_handler(self, pid):
        try:
            del self._callbacks[pid]
            return True
        except KeyError:
            return False

    def attach_loop(self, loop):
        
        
        
        
        if self._saved_sighandler is not None:
            return

        self._saved_sighandler = signal.signal(signal.SIGCHLD, self._sig_chld)
        if self._saved_sighandler is None:
            logger.warning("Previous SIGCHLD handler was set by non-Python code, "
                           "restore to default handler on watcher close.")
            self._saved_sighandler = signal.SIG_DFL

        
        signal.siginterrupt(signal.SIGCHLD, False)

    def _do_waitpid_all(self):
        for pid in list(self._callbacks):
            self._do_waitpid(pid)

    def _do_waitpid(self, expected_pid):
        assert expected_pid > 0

        try:
            pid, status = os.waitpid(expected_pid, os.WNOHANG)
        except ChildProcessError:
            
            
            pid = expected_pid
            returncode = 255
            logger.warning(
                "Unknown child process pid %d, will report returncode 255",
                pid)
            debug_log = False
        else:
            if pid == 0:
                
                return

            returncode = waitstatus_to_exitcode(status)
            debug_log = True
        try:
            loop, callback, args = self._callbacks.pop(pid)
        except KeyError:  
            
            
            logger.warning("Child watcher got an unexpected pid: %r",
                           pid, exc_info=True)
        else:
            if loop.is_closed():
                logger.warning("Loop %r that handles pid %r is closed", loop, pid)
            else:
                if debug_log and loop.get_debug():
                    logger.debug('process %s exited with returncode %s',
                                 expected_pid, returncode)
                loop.call_soon_threadsafe(callback, pid, returncode, *args)

    def _sig_chld(self, signum, frame):
        try:
            self._do_waitpid_all()
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException:
            logger.warning('Unknown exception in SIGCHLD handler', exc_info=True)


class ThreadedChildWatcher(AbstractChildWatcher):
    

    def __init__(self):
        self._pid_counter = itertools.count(0)
        self._threads = {}

    def is_active(self):
        return True

    def close(self):
        pass

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

    def __del__(self, _warn=warnings.warn):
        threads = [thread for thread in list(self._threads.values())
                   if thread.is_alive()]
        if threads:
            _warn(f"{self.__class__} has registered but not finished child processes",
                  ResourceWarning,
                  source=self)

    def add_child_handler(self, pid, callback, *args):
        loop = events.get_running_loop()
        thread = threading.Thread(target=self._do_waitpid,
                                  name=f"asyncio-waitpid-{next(self._pid_counter)}",
                                  args=(loop, pid, callback, args),
                                  daemon=True)
        self._threads[pid] = thread
        thread.start()

    def remove_child_handler(self, pid):
        
        
        
        return True

    def attach_loop(self, loop):
        pass

    def _do_waitpid(self, loop, expected_pid, callback, args):
        assert expected_pid > 0

        try:
            pid, status = os.waitpid(expected_pid, 0)
        except ChildProcessError:
            
            
            pid = expected_pid
            returncode = 255
            logger.warning(
                "Unknown child process pid %d, will report returncode 255",
                pid)
        else:
            returncode = waitstatus_to_exitcode(status)
            if loop.get_debug():
                logger.debug('process %s exited with returncode %s',
                             expected_pid, returncode)

        if loop.is_closed():
            logger.warning("Loop %r that handles pid %r is closed", loop, pid)
        else:
            loop.call_soon_threadsafe(callback, pid, returncode, *args)

        self._threads.pop(expected_pid)

def can_use_pidfd():
    if not hasattr(os, 'pidfd_open'):
        return False
    try:
        pid = os.getpid()
        os.close(os.pidfd_open(pid, 0))
    except OSError:
        
        return False
    return True


class _UnixDefaultEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
    
    _loop_factory = _UnixSelectorEventLoop

    def __init__(self):
        super().__init__()
        self._watcher = None

    def _init_watcher(self):
        with events._lock:
            if self._watcher is None:  
                if can_use_pidfd():
                    self._watcher = PidfdChildWatcher()
                else:
                    self._watcher = ThreadedChildWatcher()

    def set_event_loop(self, loop):
        

        super().set_event_loop(loop)

        if (self._watcher is not None and
                threading.current_thread() is threading.main_thread()):
            self._watcher.attach_loop(loop)

    def get_child_watcher(self):
        
        if self._watcher is None:
            self._init_watcher()

        warnings._deprecated("get_child_watcher",
                            "{name!r} is deprecated as of Python 3.12 and will be "
                            "removed in Python {remove}.", remove=(3, 14))
        return self._watcher

    def set_child_watcher(self, watcher):
        

        assert watcher is None or isinstance(watcher, AbstractChildWatcher)

        if self._watcher is not None:
            self._watcher.close()

        self._watcher = watcher
        warnings._deprecated("set_child_watcher",
                            "{name!r} is deprecated as of Python 3.12 and will be "
                            "removed in Python {remove}.", remove=(3, 14))


SelectorEventLoop = _UnixSelectorEventLoop
DefaultEventLoopPolicy = _UnixDefaultEventLoopPolicy
EventLoop = SelectorEventLoop