base implementation of event loop the event loop can be broken up into a multiplexer the part responsible for notifying us of io events and the event loop proper which wraps a multiplexer with functionality for scheduling callbacks immediately or at a given time in the future whenever a public api takes a callback subsequent positional arguments will be passed to the callback ifwhen it is called this avoids the proliferation of trivial lambdas implementing closures keyword arguments for the callback are not supported this is a conscious design decision leaving the door open for keyword arguments to modify the meaning of the api call itself minimum number of scheduled timer handles before cleanup of cancelled handles is performed minimum fraction of scheduled timer handles that are cancelled before cleanup of cancelled handles is performed maximum timeout passed to select to avoid os limitations format the task try to skip getaddrinfo if host is already an ip users might have handled name resolution in their own code and pass in resolved ips if port s a service name like http don t skip getaddrinfo linux s inetpton doesn t accept an ipv6 zone index after host like 1lo0 the host has already been resolved host is not an ip address interleave list of addrinfo tuples by family group addresses by family addrinfosbyfamily collections ordereddict for addr in addrinfos family addr0 if family not in addrinfosbyfamily addrinfosbyfamilyfamily addrinfosbyfamilyfamily appendaddr addrinfoslists listaddrinfosbyfamily values reordered if firstaddressfamilycount 1 reordered extendaddrinfoslists0 firstaddressfamilycount 1 del addrinfoslists0 firstaddressfamilycount 1 reordered extend a for a in itertools chain fromiterable itertools ziplongestaddrinfoslists if a is not none return reordered def rununtilcompletecbfut if not fut cancelled exc fut exception if isinstanceexc systemexit keyboardinterrupt issue 22429 runforever already finished no need to stop it return futures getloopfut stop if hasattrsocket tcpnodelay def setnodelaysock if sock family in socket afinet socket afinet6 and sock type socket sockstream and sock proto socket ipprototcp sock setsockoptsocket ipprototcp socket tcpnodelay 1 else def setnodelaysock pass def checksslsocketsock if ssl is not none and isinstancesock ssl sslsocket raise typeerrorsocket cannot be of type sslsocket class sendfilefallbackprotocolprotocols protocol def initself transp if not isinstancetransp transports flowcontrolmixin raise typeerrortransport should be flowcontrolmixin instance self transport transp self proto transp getprotocol self shouldresumereading transp isreading self shouldresumewriting transp protocolpaused transp pausereading transp setprotocolself if self shouldresumewriting self writereadyfut self transport loop createfuture else self writereadyfut none async def drainself if self transport isclosing raise connectionerrorconnection closed by peer fut self writereadyfut if fut is none return await fut def connectionmadeself transport raise runtimeerrorinvalid state connection should have been established already def connectionlostself exc if self writereadyfut is not none never happens if peer disconnects after sending the whole content thus disconnection is always an exception from user perspective if exc is none self writereadyfut setexception connectionerrorconnection is closed by peer else self writereadyfut setexceptionexc self proto connectionlostexc def pausewritingself if self writereadyfut is not none return self writereadyfut self transport loop createfuture def resumewritingself if self writereadyfut is none return self writereadyfut setresultfalse self writereadyfut none def datareceivedself data raise runtimeerrorinvalid state reading should be paused def eofreceivedself raise runtimeerrorinvalid state reading should be paused async def restoreself self transport setprotocolself proto if self shouldresumereading self transport resumereading if self writereadyfut is not none cancel the future basically it has no effect because protocol is switched back no code should wait for it anymore self writereadyfut cancel if self shouldresumewriting self proto resumewriting class serverevents abstractserver def initself loop sockets protocolfactory sslcontext backlog sslhandshaketimeout sslshutdowntimeoutnone self loop loop self sockets sockets self activecount 0 self waiters self protocolfactory protocolfactory self backlog backlog self sslcontext sslcontext self sslhandshaketimeout sslhandshaketimeout self sslshutdowntimeout sslshutdowntimeout self serving false self servingforeverfut none def reprself return f self class name socketsself sockets r def attachself assert self sockets is not none self activecount 1 def detachself assert self activecount 0 self activecount 1 if self activecount 0 and self sockets is none self wakeup def wakeupself waiters self waiters self waiters none for waiter in waiters if not waiter done waiter setresultnone def startservingself if self serving return self serving true for sock in self sockets sock listenself backlog self loop startserving self protocolfactory sock self sslcontext self self backlog self sslhandshaketimeout self sslshutdowntimeout def getloopself return self loop def isservingself return self serving property def socketsself if self sockets is none return return tupletrsock transportsockets for s in self sockets def closeself sockets self sockets if sockets is none return self sockets none for sock in sockets self loop stopservingsock self serving false if self servingforeverfut is not none and not self servingforeverfut done self servingforeverfut cancel self servingforeverfut none if self activecount 0 self wakeup async def startservingself self startserving skip one loop iteration so that all loop addreader go through await tasks sleep0 async def serveforeverself if self servingforeverfut is not none raise runtimeerror f server self r is already being awaited on serveforever if self sockets is none raise runtimeerrorf server self r is closed self startserving self servingforeverfut self loop createfuture try await self servingforeverfut except exceptions cancellederror try self close await self waitclosed finally raise finally self servingforeverfut none async def waitclosedself waiters are unblocked by self wakeup which is called from two places self close and self detach but only when both conditions have become true to signal that this has happened self wakeup sets self waiters to none if self waiters is none return waiter self loop createfuture self waiters appendwaiter await waiter class baseeventloopevents abstracteventloop def initself self timercancelledcount 0 self closed false self stopping false self ready collections deque self scheduled self defaultexecutor none self internalfds 0 identifier of the thread running the event loop or none if the event loop is not running self threadid none self clockresolution time getclockinfo monotonic resolution self exceptionhandler none self setdebugcoroutines isdebugmode the preserved state of async generator hooks self oldagenhooks none in debug mode if the execution of a callback or a step of a task exceed this duration in seconds the slow callbacktask is logged self slowcallbackduration 0 1 self currenthandle none self taskfactory none self coroutineorigintrackingenabled false self coroutineorigintrackingsaveddepth none a weak set of all asynchronous generators that are being iterated by the loop self asyncgens weakref weakset set to true when loop shutdownasyncgens is called self asyncgensshutdowncalled false set to true when loop shutdowndefaultexecutor is called self executorshutdowncalled false def reprself return f self class name runningself isrunning f closedself isclosed debugself getdebug def createfutureself schedule a coroutine object return a task object use legacy api if context is not needed set a task factory that will be used by loop createtask if factory is none the default task factory will be set if factory is a callable it should have a signature matching loop coro where loop will be a reference to the active event loop coro will be a coroutine object the callable must return a future return a task factory or none if the default one is in use return self taskfactory def makesockettransportself sock protocol waiternone extranone servernone create ssl transport raise notimplementederror def makedatagramtransportself sock protocol addressnone waiternone extranone create read pipe transport raise notimplementederror def makewritepipetransportself pipe protocol waiternone extranone create subprocess transport raise notimplementederror def writetoselfself raise notimplementederror def processeventsself eventlist shutdown all active asynchronous generators self asyncgensshutdowncalled true if not lenself asyncgens if python version is 3 6 or we don t have any asynchronous generators alive return closingagens listself asyncgens self asyncgens clear results await tasks gather ag aclose for ag in closingagens returnexceptionstrue for result agen in zipresults closingagens if isinstanceresult exception self callexceptionhandler message f an error occurred during closing of f asynchronous generator agen r exception result asyncgen agen async def shutdowndefaultexecutorself timeoutnone self executorshutdowncalled true if self defaultexecutor is none return future self createfuture thread threading threadtargetself doshutdown argsfuture thread start try await future finally thread jointimeout if thread isalive warnings warnthe executor did not finishing joining fits threads within timeout seconds runtimewarning stacklevel2 self defaultexecutor shutdownwaitfalse def doshutdownself future try self defaultexecutor shutdownwaittrue if not self isclosed self callsoonthreadsafefuture setresult none except exception as ex if not self isclosed self callsoonthreadsafefuture setexception ex def checkrunningself if self isrunning raise runtimeerror this event loop is already running if events getrunningloop is not none raise runtimeerror cannot run the event loop while another loop is running def runforeversetupself self checkclosed self checkrunning self setcoroutineorigintrackingself debug self oldagenhooks sys getasyncgenhooks self threadid threading getident sys setasyncgenhooks firstiterself asyncgenfirstiterhook finalizerself asyncgenfinalizerhook events setrunningloopself def runforevercleanupself self stopping false self threadid none events setrunningloopnone self setcoroutineorigintrackingfalse restore any preexisting async generator hooks if self oldagenhooks is not none sys setasyncgenhooksself oldagenhooks self oldagenhooks none def runforeverself run until the future is done if the argument is a coroutine it is wrapped in a task warning it would be disastrous to call rununtilcomplete with the same coroutine twice it would wrap it in two different tasks and that can t be good return the future s result or raise its exception an exception is raised if the future didn t complete so there is no need to log the destroy pending task message the coroutine raised a baseexception consume the exception to not log a warning the caller doesn t have access to the local task stop running the event loop every callback already scheduled will still run this simply informs runforever to stop looping after a complete iteration close the event loop this clears the queues and shuts down the executor but does not wait for the executor to finish the event loop must not be running returns true if the event loop was closed return self closed def delself warnwarnings warn if not self isclosed warnfunclosed event loop self r resourcewarning sourceself if not self isrunning self close def isrunningself return the time according to the event loop s clock this is a float expressed in seconds since an epoch but the epoch precision accuracy and drift are unspecified and may differ per event loop arrange for a callback to be called at a given time return a handle an opaque object with a cancel method that can be used to cancel the call the delay can be an int or float expressed in seconds it is always relative to the current time each callback will be called exactly once if two callbacks are scheduled for exactly the same time it is undefined which will be called first any positional arguments after the callback will be passed to the callback when it is called like calllater but uses an absolute time absolute time corresponds to the event loop s time method arrange for a callback to be called as soon as possible this operates as a fifo queue callbacks are called in the order in which they are registered each callback will be called exactly once any positional arguments after the callback will be passed to the callback when it is called check that the current thread is the thread running the event loop nonthreadsafe methods of this class make this assumption and will likely behave incorrectly when the assumption is violated should only be called when self debug true the caller is responsible for checking this condition for performance reasons like callsoon but threadsafe self checkclosed if self debug self checkcallbackcallback callsoonthreadsafe handle self callsooncallback args context if handle sourcetraceback del handle sourcetraceback1 self writetoself return handle def runinexecutorself executor func args self checkclosed if self debug self checkcallbackfunc runinexecutor if executor is none executor self defaultexecutor only check when the default executor is being used self checkdefaultexecutor if executor is none executor concurrent futures threadpoolexecutor threadnameprefix asyncio self defaultexecutor executor return futures wrapfuture executor submitfunc args loopself def setdefaultexecutorself executor if not isinstanceexecutor concurrent futures threadpoolexecutor raise typeerror executor must be threadpoolexecutor instance self defaultexecutor executor def getaddrinfodebugself host port family type proto flags msg fhost port r if family msg appendf familyfamily r if type msg appendf typetype r if proto msg appendf protoproto r if flags msg appendf flagsflags r msg joinmsg logger debug get address info s msg t0 self time addrinfo socket getaddrinfohost port family type proto flags dt self time t0 msg f getting address info msg took dt 1e3 3fms addrinfo r if dt self slowcallbackduration logger infomsg else logger debugmsg return addrinfo async def getaddrinfoself host port family0 type0 proto0 flags0 if self debug getaddrfunc self getaddrinfodebug else getaddrfunc socket getaddrinfo return await self runinexecutor none getaddrfunc host port family type proto flags async def getnameinfoself sockaddr flags0 return await self runinexecutor none socket getnameinfo sockaddr flags async def socksendfileself sock file offset0 countnone fallbacktrue if self debug and sock gettimeout 0 raise valueerrorthe socket must be nonblocking checksslsocketsock self checksendfileparamssock file offset count try return await self socksendfilenativesock file offset count except exceptions sendfilenotavailableerror as exc if not fallback raise return await self socksendfilefallbacksock file offset count async def socksendfilenativeself sock file offset count nb sendfile syscall is not supported for ssl sockets and nonmmap files even if sendfile is supported by os raise exceptions sendfilenotavailableerror fsyscall sendfile is not available for socket sock r fand file file r combination async def socksendfilefallbackself sock file offset count if offset file seekoffset blocksize mincount constants sendfilefallbackreadbuffersize if count else constants sendfilefallbackreadbuffersize buf bytearrayblocksize totalsent 0 try while true if count blocksize mincount totalsent blocksize if blocksize 0 break view memoryviewbuf blocksize read await self runinexecutornone file readinto view if not read break eof await self socksendallsock view read totalsent read return totalsent finally if totalsent 0 and hasattrfile seek file seekoffset totalsent def checksendfileparamsself sock file offset count if b not in getattrfile mode b raise valueerrorfile should be opened in binary mode if not sock type socket sockstream raise valueerroronly sockstream type sockets are supported if count is not none if not isinstancecount int raise typeerror count must be a positive integer got r formatcount if count 0 raise valueerror count must be a positive integer got r formatcount if not isinstanceoffset int raise typeerror offset must be a nonnegative integer got r format offset if offset 0 raise valueerror offset must be a nonnegative integer got r format offset async def connectsockself exceptions addrinfo localaddrinfosnone skip local addresses of different family connect to a tcp server create a streaming transport connection to a given internet host and port socket family afinet or socket afinet6 depending on host or family if specified socket type sockstream protocolfactory must be a callable returning a protocol instance this method is a coroutine which will try to establish the connection in the background when successful the coroutine returns a transport protocol pair use host as default for serverhostname it is an error if host is empty or not set e g when an alreadyconnected socket was passed or when only a port is given to avoid this error you can pass serverhostname this will bypass the hostname check this also means that if host is a numeric ipipv6 address we will attempt to verify that exact address this will probably fail but it is possible to create a certificate for a specific ip address so we don t judge it here if using happy eyeballs default to interleave addresses by family not using happy eyeballs if they all have the same str raise one raise a combined exception so the user can see all the various error messages we allow afinet afinet6 afunix as long as they are sockstream we support passing afunix sockets even though we have a dedicated api for that createunixconnection disallowing afunix in this method breaks backwards compatibility get the socket from the transport because ssl transport closes the old socket and creates a new ssl socket send a file to transport return the total number of bytes which were sent the method uses highperformance os sendfile if available file must be a regular file object opened in binary mode offset tells from where to start reading the file if specified count is the total number of bytes to transmit as opposed to sending the file until eof is reached file position is updated on return or also in case of error in which case file tell can be used to figure out the number of bytes which were sent fallback set to true makes asyncio to manually read and send the file when the platform does not support the sendfile syscall e g windows or ssl socket on unix raise sendfilenotavailableerror if the system does not support sendfile syscall and fallback is false upgrade transport to tls return a new transport that protocol should start using immediately pause early so that sslprotocol datareceived doesn t have a chance to get called before sslprotocol connectionmade create datagram connection if sock is not none if sock type socket sockdgram raise valueerror f a udp socket was expected got sock r if localaddr or remoteaddr or family or proto or flags or reuseport or allowbroadcast show the problematic kwargs in exception msg opts dictlocaladdrlocaladdr remoteaddrremoteaddr familyfamily protoproto flagsflags reuseportreuseport allowbroadcastallowbroadcast problems joinf kv for k v in opts items if v raise valueerror f socket modifier keyword arguments can not be used f when sock is specified problems sock setblockingfalse raddr none else if not localaddr or remoteaddr if family 0 raise valueerror unexpected address family addrpairsinfo family proto none none elif hasattrsocket afunix and family socket afunix for addr in localaddr remoteaddr if addr is not none and not isinstanceaddr str raise typeerror string is expected if localaddr and localaddr0 not in 0 x00 try if stat sissockos statlocaladdr stmode os removelocaladdr 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 localaddr err addrpairsinfo family proto localaddr remoteaddr else join address by family protocol addrinfos using order preserving dict for idx addr in 0 localaddr 1 remoteaddr if addr is not none if not isinstanceaddr tuple and lenaddr 2 raise typeerror 2tuple is expected infos await self ensureresolved addr familyfamily typesocket sockdgram protoproto flagsflags loopself if not infos raise oserror getaddrinfo returned empty list for fam pro address in infos key fam pro if key not in addrinfos addrinfoskey none none addrinfoskeyidx address each addr has to have info for each family proto pair addrpairsinfo key addrpair for key addrpair in addrinfos items if not localaddr and addrpair0 is none or remoteaddr and addrpair1 is none if not addrpairsinfo raise valueerror can not get address information exceptions for family proto localaddress remoteaddress in addrpairsinfo sock none raddr none try sock socket socket familyfamily typesocket sockdgram protoproto if reuseport setreuseportsock if allowbroadcast sock setsockopt socket solsocket socket sobroadcast 1 sock setblockingfalse if localaddr sock bindlocaladdress if remoteaddr if not allowbroadcast await self sockconnectsock remoteaddress raddr remoteaddress except oserror as exc if sock is not none sock close exceptions appendexc except if sock is not none sock close raise else break else raise exceptions0 protocol protocolfactory waiter self createfuture transport self makedatagramtransport sock protocol raddr waiter if self debug if localaddr logger infodatagram endpoint localaddrr remoteaddrr created r r localaddr remoteaddr transport protocol else logger debugdatagram endpoint remoteaddrr created r r remoteaddr transport protocol try await waiter except transport close raise return transport protocol async def ensureresolvedself address family0 typesocket sockstream proto0 flags0 loop host port address 2 info ipaddrinfohost port family type proto address2 if info is not none host is already a resolved ip return info else return await loop getaddrinfohost port familyfamily typetype protoproto flagsflags async def createservergetaddrinfoself host port family flags infos await self ensureresolvedhost port familyfamily typesocket sockstream flagsflags loopself if not infos raise oserrorf getaddrinfohost r returned empty list return infos async def createserver self protocolfactory hostnone portnone familysocket afunspec flagssocket aipassive socknone backlog100 sslnone reuseaddressnone reuseportnone sslhandshaketimeoutnone sslshutdowntimeoutnone startservingtrue if isinstancessl bool raise typeerror ssl argument must be an sslcontext or none if sslhandshaketimeout is not none and ssl is none raise valueerror sslhandshaketimeout is only meaningful with ssl if sslshutdowntimeout is not none and ssl is none raise valueerror sslshutdowntimeout is only meaningful with ssl if sock is not none checksslsocketsock if host is not none or port is not none if sock is not none raise valueerror hostport and sock can not be specified at the same time if reuseaddress is none reuseaddress os name posix and sys platform cygwin sockets if host hosts none elif isinstancehost str or not isinstancehost collections abc iterable hosts host else hosts host fs self createservergetaddrinfohost port familyfamily flagsflags for host in hosts infos await tasks gatherfs infos setitertools chain fromiterableinfos completed false try for res in infos af socktype proto canonname sa res try sock socket socketaf socktype proto except socket error assume it s a bad familytypeprotocol combination if self debug logger warning createserver failed to create socket socketr r r af socktype proto excinfotrue continue sockets appendsock if reuseaddress sock setsockopt socket solsocket socket soreuseaddr true if reuseport setreuseportsock disable ipv4ipv6 dual stack support enabled by default on linux which makes a single socket listen on both address families if hasipv6 and af socket afinet6 and hasattrsocket ipprotoipv6 sock setsockoptsocket ipprotoipv6 socket ipv6v6only true try sock bindsa except oserror as err raise oserrorerr errno error while attempting to bind on address r s sa err strerror lower from none completed true finally if not completed for sock in sockets sock close else if sock is none raise valueerror neither hostport nor sock were specified if sock type socket sockstream raise valueerrorf a stream socket was expected got sock r sockets sock for sock in sockets sock setblockingfalse server serverself sockets protocolfactory ssl backlog sslhandshaketimeout sslshutdowntimeout if startserving server startserving skip one loop iteration so that all loop addreader go through await tasks sleep0 if self debug logger infor is serving server return server async def connectacceptedsocket self protocolfactory sock sslnone sslhandshaketimeoutnone sslshutdowntimeoutnone if sock type socket sockstream raise valueerrorf a stream socket was expected got sock r 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 sock is not none checksslsocketsock transport protocol await self createconnectiontransport sock protocolfactory ssl serversidetrue sslhandshaketimeoutsslhandshaketimeout sslshutdowntimeoutsslshutdowntimeout if self debug get the socket from the transport because ssl transport closes the old socket and creates a new ssl socket sock transport getextrainfo socket logger debugr handled r r sock transport protocol return transport protocol async def connectreadpipeself protocolfactory pipe protocol protocolfactory waiter self createfuture transport self makereadpipetransportpipe protocol waiter try await waiter except transport close raise if self debug logger debug read pipe r connected r r pipe fileno transport protocol return transport protocol async def connectwritepipeself protocolfactory pipe protocol protocolfactory waiter self createfuture transport self makewritepipetransportpipe protocol waiter try await waiter except transport close raise if self debug logger debug write pipe r connected r r pipe fileno transport protocol return transport protocol def logsubprocessself msg stdin stdout stderr info msg if stdin is not none info appendf stdinformatpipestdin if stdout is not none and stderr subprocess stdout info appendf stdoutstderrformatpipestdout else if stdout is not none info appendf stdoutformatpipestdout if stderr is not none info appendf stderrformatpipestderr logger debug joininfo async def subprocessshellself protocolfactory cmd stdinsubprocess pipe stdoutsubprocess pipe stderrsubprocess pipe universalnewlinesfalse shelltrue bufsize0 encodingnone errorsnone textnone kwargs if not isinstancecmd bytes str raise valueerrorcmd must be a string if universalnewlines raise valueerroruniversalnewlines must be false if not shell raise valueerrorshell must be true if bufsize 0 raise valueerrorbufsize must be 0 if text raise valueerrortext must be false if encoding is not none raise valueerrorencoding must be none if errors is not none raise valueerrorerrors must be none protocol protocolfactory debuglog none if self debug don t log parameters they may contain sensitive information password and may be too long debuglog run shell command r cmd self logsubprocessdebuglog stdin stdout stderr transport await self makesubprocesstransport protocol cmd true stdin stdout stderr bufsize kwargs if self debug and debuglog is not none logger info s r debuglog transport return transport protocol async def subprocessexecself protocolfactory program args stdinsubprocess pipe stdoutsubprocess pipe stderrsubprocess pipe universalnewlinesfalse shellfalse bufsize0 encodingnone errorsnone textnone kwargs if universalnewlines raise valueerroruniversalnewlines must be false if shell raise valueerrorshell must be false if bufsize 0 raise valueerrorbufsize must be 0 if text raise valueerrortext must be false if encoding is not none raise valueerrorencoding must be none if errors is not none raise valueerrorerrors must be none popenargs program args protocol protocolfactory debuglog none if self debug don t log parameters they may contain sensitive information password and may be too long debuglog f execute program program r self logsubprocessdebuglog stdin stdout stderr transport await self makesubprocesstransport protocol popenargs false stdin stdout stderr bufsize kwargs if self debug and debuglog is not none logger info s r debuglog transport return transport protocol def getexceptionhandlerself return self exceptionhandler def setexceptionhandlerself handler if handler is not none and not callablehandler raise typeerrorf a callable object or none is expected f got handler r self exceptionhandler handler def defaultexceptionhandlerself context message context get message if not message message unhandled exception in event loop exception context get exception if exception is not none excinfo typeexception exception exception traceback else excinfo false if sourcetraceback not in context and self currenthandle is not none and self currenthandle sourcetraceback context handletraceback self currenthandle sourcetraceback loglines message for key in sortedcontext if key in message exception continue value contextkey if key sourcetraceback tb jointraceback formatlistvalue value object created at most recent call last n value tb rstrip elif key handletraceback tb jointraceback formatlistvalue value handle created at most recent call last n value tb rstrip else value reprvalue loglines appendf key value logger error n joinloglines excinfoexcinfo def callexceptionhandlerself context if self exceptionhandler is none try self defaultexceptionhandlercontext except systemexit keyboardinterrupt raise except baseexception second protection layer for unexpected errors in the default implementation as well as for subclassed event loops with overloaded defaultexceptionhandler logger error exception in default exception handler excinfotrue else try ctx none thing context gettask if thing is none even though futures don t have a context task is a subclass of future and sometimes the future key holds a task thing context getfuture if thing is none handles also have a context thing context gethandle if thing is not none and hasattrthing getcontext ctx thing getcontext if ctx is not none and hasattrctx run ctx runself exceptionhandler self context else self exceptionhandlerself context except systemexit keyboardinterrupt raise except baseexception as exc exception in the user set custom exception handler try let s try default handler self defaultexceptionhandler message unhandled error in exception handler exception exc context context except systemexit keyboardinterrupt raise except baseexception guard defaultexceptionhandler in case it is overloaded logger error exception in default exception handler while handling an unexpected error in custom exception handler excinfotrue def addcallbackself handle like addcallback but called from a signal handler self addcallbackhandle self writetoself def timerhandlecancelledself handle run one full iteration of the event loop this calls all currently ready callbacks polls for io schedules the resulting callbacks and finally schedules calllater callbacks remove delayed calls that were cancelled if their number is too high remove delayed calls that were cancelled from head of queue compute the desired timeout needed to break cycles when an exception occurs handle later callbacks that are ready this is the only place where callbacks are actually called all other places just add them to ready note we run all currently scheduled callbacks but not any callbacks scheduled by callbacks run this time around they will be run the next time after another io poll use an idiom that is threadsafe without using locks pragma no cover minimum number of _scheduled timer handles before cleanup of cancelled handles is performed minimum fraction of _scheduled timer handles that are cancelled before cleanup of cancelled handles is performed maximum timeout passed to select to avoid os limitations format the task try to skip getaddrinfo if host is already an ip users might have handled name resolution in their own code and pass in resolved ips if port s a service name like http don t skip getaddrinfo linux s inet_pton doesn t accept an ipv6 zone index after host like 1 lo0 the host has already been resolved host is not an ip address interleave list of addrinfo tuples by family group addresses by family issue 22429 run_forever already finished no need to stop it never happens if peer disconnects after sending the whole content thus disconnection is always an exception from user perspective cancel the future basically it has no effect because protocol is switched back no code should wait for it anymore skip one loop iteration so that all loop add_reader go through wait until server is closed and all connections are dropped if the server is not closed wait if it is closed but there are still active connections wait anyone waiting here will be unblocked once both conditions server is closed and all connections have been dropped have become true in either order historical note in 3 11 and before this was broken returning immediately if the server was already closed even if there were still active connections an attempted fix in 3 12 0 was still broken returning immediately if the server was still open and there were no active connections hopefully in 3 12 1 we have it right waiters are unblocked by self _wakeup which is called from two places self close and self _detach but only when both conditions have become true to signal that this has happened self _wakeup sets self _waiters to none identifier of the thread running the event loop or none if the event loop is not running the preserved state of async generator hooks in debug mode if the execution of a callback or a step of a task exceed this duration in seconds the slow callback task is logged a weak set of all asynchronous generators that are being iterated by the loop set to true when loop shutdown_asyncgens is called set to true when loop shutdown_default_executor is called create a future object attached to the loop schedule a coroutine object return a task object use legacy api if context is not needed set a task factory that will be used by loop create_task if factory is none the default task factory will be set if factory is a callable it should have a signature matching loop coro where loop will be a reference to the active event loop coro will be a coroutine object the callable must return a future return a task factory or none if the default one is in use create socket transport create ssl transport create datagram transport create read pipe transport create write pipe transport create subprocess transport write a byte to self pipe to wake up the event loop this may be called from a different thread the subclass is responsible for implementing the self pipe process selector events shutdown all active asynchronous generators if python version is 3 6 or we don t have any asynchronous generators alive schedule the shutdown of the default executor the timeout parameter specifies the amount of time the executor will be given to finish joining the default value is none which means that the executor will be given an unlimited amount of time prepare the run loop to process events this method exists so that custom custom event loop subclasses e g event loops that integrate a gui event loop with python s event loop have access to all the loop setup logic clean up after an event loop finishes the looping over events this method exists so that custom custom event loop subclasses e g event loops that integrate a gui event loop with python s event loop have access to all the loop cleanup logic restore any pre existing async generator hooks run until stop is called run until the future is done if the argument is a coroutine it is wrapped in a task warning it would be disastrous to call run_until_complete with the same coroutine twice it would wrap it in two different tasks and that can t be good return the future s result or raise its exception an exception is raised if the future didn t complete so there is no need to log the destroy pending task message the coroutine raised a baseexception consume the exception to not log a warning the caller doesn t have access to the local task stop running the event loop every callback already scheduled will still run this simply informs run_forever to stop looping after a complete iteration close the event loop this clears the queues and shuts down the executor but does not wait for the executor to finish the event loop must not be running returns true if the event loop was closed returns true if the event loop is running return the time according to the event loop s clock this is a float expressed in seconds since an epoch but the epoch precision accuracy and drift are unspecified and may differ per event loop arrange for a callback to be called at a given time return a handle an opaque object with a cancel method that can be used to cancel the call the delay can be an int or float expressed in seconds it is always relative to the current time each callback will be called exactly once if two callbacks are scheduled for exactly the same time it is undefined which will be called first any positional arguments after the callback will be passed to the callback when it is called like call_later but uses an absolute time absolute time corresponds to the event loop s time method arrange for a callback to be called as soon as possible this operates as a fifo queue callbacks are called in the order in which they are registered each callback will be called exactly once any positional arguments after the callback will be passed to the callback when it is called check that the current thread is the thread running the event loop non thread safe methods of this class make this assumption and will likely behave incorrectly when the assumption is violated should only be called when self _debug true the caller is responsible for checking this condition for performance reasons like call_soon but thread safe only check when the default executor is being used nb sendfile syscall is not supported for ssl sockets and non mmap files even if sendfile is supported by os eof create bind and connect one socket skip local addresses of different family all bind attempts failed connect to a tcp server create a streaming transport connection to a given internet host and port socket family af_inet or socket af_inet6 depending on host or family if specified socket type sock_stream protocol_factory must be a callable returning a protocol instance this method is a coroutine which will try to establish the connection in the background when successful the coroutine returns a transport protocol pair use host as default for server_hostname it is an error if host is empty or not set e g when an already connected socket was passed or when only a port is given to avoid this error you can pass server_hostname this will bypass the hostname check this also means that if host is a numeric ip ipv6 address we will attempt to verify that exact address this will probably fail but it is possible to create a certificate for a specific ip address so we don t judge it here if using happy eyeballs default to interleave addresses by family not using happy eyeballs using happy eyeballs if they all have the same str raise one raise a combined exception so the user can see all the various error messages we allow af_inet af_inet6 af_unix as long as they are sock_stream we support passing af_unix sockets even though we have a dedicated api for that create_unix_connection disallowing af_unix in this method breaks backwards compatibility get the socket from the transport because ssl transport closes the old socket and creates a new ssl socket send a file to transport return the total number of bytes which were sent the method uses high performance os sendfile if available file must be a regular file object opened in binary mode offset tells from where to start reading the file if specified count is the total number of bytes to transmit as opposed to sending the file until eof is reached file position is updated on return or also in case of error in which case file tell can be used to figure out the number of bytes which were sent fallback set to true makes asyncio to manually read and send the file when the platform does not support the sendfile syscall e g windows or ssl socket on unix raise sendfilenotavailableerror if the system does not support sendfile syscall and fallback is false eof upgrade transport to tls return a new transport that protocol should start using immediately pause early so that ssl_protocol data_received doesn t have a chance to get called before ssl_protocol connection_made create datagram connection show the problematic kwargs in exception msg directory may have permissions only to create socket join address by family protocol using order preserving dict each addr has to have info for each family proto pair host is already a resolved ip create a tcp server the host parameter can be a string in that case the tcp server is bound to host and port the host parameter can also be a sequence of strings and in that case the tcp server is bound to all hosts of the sequence if a host appears multiple times possibly indirectly e g when hostnames resolve to the same ip address the server is only bound once to that host return a server object which can be used to stop the service this method is a coroutine assume it s a bad family type protocol combination disable ipv4 ipv6 dual stack support enabled by default on linux which makes a single socket listen on both address families skip one loop iteration so that all loop add_reader go through get the socket from the transport because ssl transport closes the old socket and creates a new ssl socket don t log parameters they may contain sensitive information password and may be too long don t log parameters they may contain sensitive information password and may be too long return an exception handler or none if the default one is in use set handler as the new event loop exception handler if handler is none the default exception handler will be set if handler is a callable object it should have a signature matching loop context where loop will be a reference to the active event loop context will be a dict object see call_exception_handler documentation for details about context default exception handler this is called when an exception occurs and no exception handler is set and can be called by a custom exception handler that wants to defer to the default behavior this default handler logs the error message and other context dependent information in debug mode a truncated stack trace is also appended showing where the given object e g a handle or future or task was created if any the context parameter has the same meaning as in call_exception_handler call the current event loop s exception handler the context argument is a dict containing the following keys message error message exception optional exception object future optional future instance task optional task instance handle optional handle instance protocol optional protocol instance transport optional transport instance socket optional socket instance asyncgen optional asynchronous generator that caused the exception new keys maybe introduced in the future note do not overload this method in an event loop subclass for custom exception handling use the set_exception_handler method second protection layer for unexpected errors in the default implementation as well as for subclassed event loops with overloaded default_exception_handler even though futures don t have a context task is a subclass of future and sometimes the future key holds a task handles also have a context exception in the user set custom exception handler let s try default handler guard default_exception_handler in case it is overloaded add a handle to _ready like _add_callback but called from a signal handler notification that a timerhandle has been cancelled run one full iteration of the event loop this calls all currently ready callbacks polls for i o schedules the resulting callbacks and finally schedules call_later callbacks remove delayed calls that were cancelled if their number is too high remove delayed calls that were cancelled from head of queue compute the desired timeout needed to break cycles when an exception occurs handle later callbacks that are ready this is the only place where callbacks are actually called all other places just add them to ready note we run all currently scheduled callbacks but not any callbacks scheduled by callbacks run this time around they will be run the next time after another i o poll use an idiom that is thread safe without using locks needed to break cycles when an exception occurs
import collections
import collections.abc
import concurrent.futures
import functools
import heapq
import itertools
import os
import socket
import stat
import subprocess
import threading
import time
import traceback
import sys
import warnings
import weakref

try:
    import ssl
except ImportError:  
    ssl = None

from . import constants
from . import coroutines
from . import events
from . import exceptions
from . import futures
from . import protocols
from . import sslproto
from . import staggered
from . import tasks
from . import transports
from . import trsock
from .log import logger


__all__ = 'BaseEventLoop','Server',




_MIN_SCHEDULED_TIMER_HANDLES = 100



_MIN_CANCELLED_TIMER_HANDLES_FRACTION = 0.5


_HAS_IPv6 = hasattr(socket, 'AF_INET6')


MAXIMUM_SELECT_TIMEOUT = 24 * 3600


def _format_handle(handle):
    cb = handle._callback
    if isinstance(getattr(cb, '__self__', None), tasks.Task):
        
        return repr(cb.__self__)
    else:
        return str(handle)


def _format_pipe(fd):
    if fd == subprocess.PIPE:
        return '<pipe>'
    elif fd == subprocess.STDOUT:
        return '<stdout>'
    else:
        return repr(fd)


def _set_reuseport(sock):
    if not hasattr(socket, 'SO_REUSEPORT'):
        raise ValueError('reuse_port not supported by socket module')
    else:
        try:
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
        except OSError:
            raise ValueError('reuse_port not supported by socket module, '
                             'SO_REUSEPORT defined but not implemented.')


def _ipaddr_info(host, port, family, type, proto, flowinfo=0, scopeid=0):
    
    
    if not hasattr(socket, 'inet_pton'):
        return

    if proto not in {0, socket.IPPROTO_TCP, socket.IPPROTO_UDP} or \
            host is None:
        return None

    if type == socket.SOCK_STREAM:
        proto = socket.IPPROTO_TCP
    elif type == socket.SOCK_DGRAM:
        proto = socket.IPPROTO_UDP
    else:
        return None

    if port is None:
        port = 0
    elif isinstance(port, bytes) and port == b'':
        port = 0
    elif isinstance(port, str) and port == '':
        port = 0
    else:
        
        try:
            port = int(port)
        except (TypeError, ValueError):
            return None

    if family == socket.AF_UNSPEC:
        afs = [socket.AF_INET]
        if _HAS_IPv6:
            afs.append(socket.AF_INET6)
    else:
        afs = [family]

    if isinstance(host, bytes):
        host = host.decode('idna')
    if '%' in host:
        
        
        return None

    for af in afs:
        try:
            socket.inet_pton(af, host)
            
            if _HAS_IPv6 and af == socket.AF_INET6:
                return af, type, proto, '', (host, port, flowinfo, scopeid)
            else:
                return af, type, proto, '', (host, port)
        except OSError:
            pass

    
    return None


def _interleave_addrinfos(addrinfos, first_address_family_count=1):
    
    
    addrinfos_by_family = collections.OrderedDict()
    for addr in addrinfos:
        family = addr[0]
        if family not in addrinfos_by_family:
            addrinfos_by_family[family] = []
        addrinfos_by_family[family].append(addr)
    addrinfos_lists = list(addrinfos_by_family.values())

    reordered = []
    if first_address_family_count > 1:
        reordered.extend(addrinfos_lists[0][:first_address_family_count - 1])
        del addrinfos_lists[0][:first_address_family_count - 1]
    reordered.extend(
        a for a in itertools.chain.from_iterable(
            itertools.zip_longest(*addrinfos_lists)
        ) if a is not None)
    return reordered


def _run_until_complete_cb(fut):
    if not fut.cancelled():
        exc = fut.exception()
        if isinstance(exc, (SystemExit, KeyboardInterrupt)):
            
            
            return
    futures._get_loop(fut).stop()


if hasattr(socket, 'TCP_NODELAY'):
    def _set_nodelay(sock):
        if (sock.family in {socket.AF_INET, socket.AF_INET6} and
                sock.type == socket.SOCK_STREAM and
                sock.proto == socket.IPPROTO_TCP):
            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
else:
    def _set_nodelay(sock):
        pass


def _check_ssl_socket(sock):
    if ssl is not None and isinstance(sock, ssl.SSLSocket):
        raise TypeError("Socket cannot be of type SSLSocket")


class _SendfileFallbackProtocol(protocols.Protocol):
    def __init__(self, transp):
        if not isinstance(transp, transports._FlowControlMixin):
            raise TypeError("transport should be _FlowControlMixin instance")
        self._transport = transp
        self._proto = transp.get_protocol()
        self._should_resume_reading = transp.is_reading()
        self._should_resume_writing = transp._protocol_paused
        transp.pause_reading()
        transp.set_protocol(self)
        if self._should_resume_writing:
            self._write_ready_fut = self._transport._loop.create_future()
        else:
            self._write_ready_fut = None

    async def drain(self):
        if self._transport.is_closing():
            raise ConnectionError("Connection closed by peer")
        fut = self._write_ready_fut
        if fut is None:
            return
        await fut

    def connection_made(self, transport):
        raise RuntimeError("Invalid state: "
                           "connection should have been established already.")

    def connection_lost(self, exc):
        if self._write_ready_fut is not None:
            
            
            if exc is None:
                self._write_ready_fut.set_exception(
                    ConnectionError("Connection is closed by peer"))
            else:
                self._write_ready_fut.set_exception(exc)
        self._proto.connection_lost(exc)

    def pause_writing(self):
        if self._write_ready_fut is not None:
            return
        self._write_ready_fut = self._transport._loop.create_future()

    def resume_writing(self):
        if self._write_ready_fut is None:
            return
        self._write_ready_fut.set_result(False)
        self._write_ready_fut = None

    def data_received(self, data):
        raise RuntimeError("Invalid state: reading should be paused")

    def eof_received(self):
        raise RuntimeError("Invalid state: reading should be paused")

    async def restore(self):
        self._transport.set_protocol(self._proto)
        if self._should_resume_reading:
            self._transport.resume_reading()
        if self._write_ready_fut is not None:
            
            
            
            self._write_ready_fut.cancel()
        if self._should_resume_writing:
            self._proto.resume_writing()


class Server(events.AbstractServer):

    def __init__(self, loop, sockets, protocol_factory, ssl_context, backlog,
                 ssl_handshake_timeout, ssl_shutdown_timeout=None):
        self._loop = loop
        self._sockets = sockets
        self._active_count = 0
        self._waiters = []
        self._protocol_factory = protocol_factory
        self._backlog = backlog
        self._ssl_context = ssl_context
        self._ssl_handshake_timeout = ssl_handshake_timeout
        self._ssl_shutdown_timeout = ssl_shutdown_timeout
        self._serving = False
        self._serving_forever_fut = None

    def __repr__(self):
        return f'<{self.__class__.__name__} sockets={self.sockets!r}>'

    def _attach(self):
        assert self._sockets is not None
        self._active_count += 1

    def _detach(self):
        assert self._active_count > 0
        self._active_count -= 1
        if self._active_count == 0 and self._sockets is None:
            self._wakeup()

    def _wakeup(self):
        waiters = self._waiters
        self._waiters = None
        for waiter in waiters:
            if not waiter.done():
                waiter.set_result(None)

    def _start_serving(self):
        if self._serving:
            return
        self._serving = True
        for sock in self._sockets:
            sock.listen(self._backlog)
            self._loop._start_serving(
                self._protocol_factory, sock, self._ssl_context,
                self, self._backlog, self._ssl_handshake_timeout,
                self._ssl_shutdown_timeout)

    def get_loop(self):
        return self._loop

    def is_serving(self):
        return self._serving

    @property
    def sockets(self):
        if self._sockets is None:
            return ()
        return tuple(trsock.TransportSocket(s) for s in self._sockets)

    def close(self):
        sockets = self._sockets
        if sockets is None:
            return
        self._sockets = None

        for sock in sockets:
            self._loop._stop_serving(sock)

        self._serving = False

        if (self._serving_forever_fut is not None and
                not self._serving_forever_fut.done()):
            self._serving_forever_fut.cancel()
            self._serving_forever_fut = None

        if self._active_count == 0:
            self._wakeup()

    async def start_serving(self):
        self._start_serving()
        
        
        await tasks.sleep(0)

    async def serve_forever(self):
        if self._serving_forever_fut is not None:
            raise RuntimeError(
                f'server {self!r} is already being awaited on serve_forever()')
        if self._sockets is None:
            raise RuntimeError(f'server {self!r} is closed')

        self._start_serving()
        self._serving_forever_fut = self._loop.create_future()

        try:
            await self._serving_forever_fut
        except exceptions.CancelledError:
            try:
                self.close()
                await self.wait_closed()
            finally:
                raise
        finally:
            self._serving_forever_fut = None

    async def wait_closed(self):
        
        
        
        
        
        if self._waiters is None:
            return
        waiter = self._loop.create_future()
        self._waiters.append(waiter)
        await waiter


class BaseEventLoop(events.AbstractEventLoop):

    def __init__(self):
        self._timer_cancelled_count = 0
        self._closed = False
        self._stopping = False
        self._ready = collections.deque()
        self._scheduled = []
        self._default_executor = None
        self._internal_fds = 0
        
        
        self._thread_id = None
        self._clock_resolution = time.get_clock_info('monotonic').resolution
        self._exception_handler = None
        self.set_debug(coroutines._is_debug_mode())
        
        self._old_agen_hooks = None
        
        
        self.slow_callback_duration = 0.1
        self._current_handle = None
        self._task_factory = None
        self._coroutine_origin_tracking_enabled = False
        self._coroutine_origin_tracking_saved_depth = None

        
        
        self._asyncgens = weakref.WeakSet()
        
        self._asyncgens_shutdown_called = False
        
        self._executor_shutdown_called = False

    def __repr__(self):
        return (
            f'<{self.__class__.__name__} running={self.is_running()} '
            f'closed={self.is_closed()} debug={self.get_debug()}>'
        )

    def create_future(self):
        
        return futures.Future(loop=self)

    def create_task(self, coro, *, name=None, context=None):
        
        self._check_closed()
        if self._task_factory is None:
            task = tasks.Task(coro, loop=self, name=name, context=context)
            if task._source_traceback:
                del task._source_traceback[-1]
        else:
            if context is None:
                
                task = self._task_factory(self, coro)
            else:
                task = self._task_factory(self, coro, context=context)

            task.set_name(name)

        return task

    def set_task_factory(self, factory):
        
        if factory is not None and not callable(factory):
            raise TypeError('task factory must be a callable or None')
        self._task_factory = factory

    def get_task_factory(self):
        
        return self._task_factory

    def _make_socket_transport(self, sock, protocol, waiter=None, *,
                               extra=None, server=None):
        
        raise NotImplementedError

    def _make_ssl_transport(
            self, rawsock, protocol, sslcontext, waiter=None,
            *, server_side=False, server_hostname=None,
            extra=None, server=None,
            ssl_handshake_timeout=None,
            ssl_shutdown_timeout=None,
            call_connection_made=True):
        
        raise NotImplementedError

    def _make_datagram_transport(self, sock, protocol,
                                 address=None, waiter=None, extra=None):
        
        raise NotImplementedError

    def _make_read_pipe_transport(self, pipe, protocol, waiter=None,
                                  extra=None):
        
        raise NotImplementedError

    def _make_write_pipe_transport(self, pipe, protocol, waiter=None,
                                   extra=None):
        
        raise NotImplementedError

    async def _make_subprocess_transport(self, protocol, args, shell,
                                         stdin, stdout, stderr, bufsize,
                                         extra=None, **kwargs):
        
        raise NotImplementedError

    def _write_to_self(self):
        
        raise NotImplementedError

    def _process_events(self, event_list):
        
        raise NotImplementedError

    def _check_closed(self):
        if self._closed:
            raise RuntimeError('Event loop is closed')

    def _check_default_executor(self):
        if self._executor_shutdown_called:
            raise RuntimeError('Executor shutdown has been called')

    def _asyncgen_finalizer_hook(self, agen):
        self._asyncgens.discard(agen)
        if not self.is_closed():
            self.call_soon_threadsafe(self.create_task, agen.aclose())

    def _asyncgen_firstiter_hook(self, agen):
        if self._asyncgens_shutdown_called:
            warnings.warn(
                f"asynchronous generator {agen!r} was scheduled after "
                f"loop.shutdown_asyncgens() call",
                ResourceWarning, source=self)

        self._asyncgens.add(agen)

    async def shutdown_asyncgens(self):
        
        self._asyncgens_shutdown_called = True

        if not len(self._asyncgens):
            
            
            return

        closing_agens = list(self._asyncgens)
        self._asyncgens.clear()

        results = await tasks.gather(
            *[ag.aclose() for ag in closing_agens],
            return_exceptions=True)

        for result, agen in zip(results, closing_agens):
            if isinstance(result, Exception):
                self.call_exception_handler({
                    'message': f'an error occurred during closing of '
                               f'asynchronous generator {agen!r}',
                    'exception': result,
                    'asyncgen': agen
                })

    async def shutdown_default_executor(self, timeout=None):
        
        self._executor_shutdown_called = True
        if self._default_executor is None:
            return
        future = self.create_future()
        thread = threading.Thread(target=self._do_shutdown, args=(future,))
        thread.start()
        try:
            await future
        finally:
            thread.join(timeout)

        if thread.is_alive():
            warnings.warn("The executor did not finishing joining "
                             f"its threads within {timeout} seconds.",
                             RuntimeWarning, stacklevel=2)
            self._default_executor.shutdown(wait=False)

    def _do_shutdown(self, future):
        try:
            self._default_executor.shutdown(wait=True)
            if not self.is_closed():
                self.call_soon_threadsafe(future.set_result, None)
        except Exception as ex:
            if not self.is_closed():
                self.call_soon_threadsafe(future.set_exception, ex)

    def _check_running(self):
        if self.is_running():
            raise RuntimeError('This event loop is already running')
        if events._get_running_loop() is not None:
            raise RuntimeError(
                'Cannot run the event loop while another loop is running')

    def _run_forever_setup(self):
        
        self._check_closed()
        self._check_running()
        self._set_coroutine_origin_tracking(self._debug)

        self._old_agen_hooks = sys.get_asyncgen_hooks()
        self._thread_id = threading.get_ident()
        sys.set_asyncgen_hooks(
            firstiter=self._asyncgen_firstiter_hook,
            finalizer=self._asyncgen_finalizer_hook
        )

        events._set_running_loop(self)

    def _run_forever_cleanup(self):
        
        self._stopping = False
        self._thread_id = None
        events._set_running_loop(None)
        self._set_coroutine_origin_tracking(False)
        
        if self._old_agen_hooks is not None:
            sys.set_asyncgen_hooks(*self._old_agen_hooks)
            self._old_agen_hooks = None

    def run_forever(self):
        
        try:
            self._run_forever_setup()
            while True:
                self._run_once()
                if self._stopping:
                    break
        finally:
            self._run_forever_cleanup()

    def run_until_complete(self, future):
        
        self._check_closed()
        self._check_running()

        new_task = not futures.isfuture(future)
        future = tasks.ensure_future(future, loop=self)
        if new_task:
            
            
            future._log_destroy_pending = False

        future.add_done_callback(_run_until_complete_cb)
        try:
            self.run_forever()
        except:
            if new_task and future.done() and not future.cancelled():
                
                
                
                future.exception()
            raise
        finally:
            future.remove_done_callback(_run_until_complete_cb)
        if not future.done():
            raise RuntimeError('Event loop stopped before Future completed.')

        return future.result()

    def stop(self):
        
        self._stopping = True

    def close(self):
        
        if self.is_running():
            raise RuntimeError("Cannot close a running event loop")
        if self._closed:
            return
        if self._debug:
            logger.debug("Close %r", self)
        self._closed = True
        self._ready.clear()
        self._scheduled.clear()
        self._executor_shutdown_called = True
        executor = self._default_executor
        if executor is not None:
            self._default_executor = None
            executor.shutdown(wait=False)

    def is_closed(self):
        
        return self._closed

    def __del__(self, _warn=warnings.warn):
        if not self.is_closed():
            _warn(f"unclosed event loop {self!r}", ResourceWarning, source=self)
            if not self.is_running():
                self.close()

    def is_running(self):
        
        return (self._thread_id is not None)

    def time(self):
        
        return time.monotonic()

    def call_later(self, delay, callback, *args, context=None):
        
        if delay is None:
            raise TypeError('delay must not be None')
        timer = self.call_at(self.time() + delay, callback, *args,
                             context=context)
        if timer._source_traceback:
            del timer._source_traceback[-1]
        return timer

    def call_at(self, when, callback, *args, context=None):
        
        if when is None:
            raise TypeError("when cannot be None")
        self._check_closed()
        if self._debug:
            self._check_thread()
            self._check_callback(callback, 'call_at')
        timer = events.TimerHandle(when, callback, args, self, context)
        if timer._source_traceback:
            del timer._source_traceback[-1]
        heapq.heappush(self._scheduled, timer)
        timer._scheduled = True
        return timer

    def call_soon(self, callback, *args, context=None):
        
        self._check_closed()
        if self._debug:
            self._check_thread()
            self._check_callback(callback, 'call_soon')
        handle = self._call_soon(callback, args, context)
        if handle._source_traceback:
            del handle._source_traceback[-1]
        return handle

    def _check_callback(self, callback, method):
        if (coroutines.iscoroutine(callback) or
                coroutines.iscoroutinefunction(callback)):
            raise TypeError(
                f"coroutines cannot be used with {method}()")
        if not callable(callback):
            raise TypeError(
                f'a callable object was expected by {method}(), '
                f'got {callback!r}')

    def _call_soon(self, callback, args, context):
        handle = events.Handle(callback, args, self, context)
        if handle._source_traceback:
            del handle._source_traceback[-1]
        self._ready.append(handle)
        return handle

    def _check_thread(self):
        
        if self._thread_id is None:
            return
        thread_id = threading.get_ident()
        if thread_id != self._thread_id:
            raise RuntimeError(
                "Non-thread-safe operation invoked on an event loop other "
                "than the current one")

    def call_soon_threadsafe(self, callback, *args, context=None):
        
        self._check_closed()
        if self._debug:
            self._check_callback(callback, 'call_soon_threadsafe')
        handle = self._call_soon(callback, args, context)
        if handle._source_traceback:
            del handle._source_traceback[-1]
        self._write_to_self()
        return handle

    def run_in_executor(self, executor, func, *args):
        self._check_closed()
        if self._debug:
            self._check_callback(func, 'run_in_executor')
        if executor is None:
            executor = self._default_executor
            
            self._check_default_executor()
            if executor is None:
                executor = concurrent.futures.ThreadPoolExecutor(
                    thread_name_prefix='asyncio'
                )
                self._default_executor = executor
        return futures.wrap_future(
            executor.submit(func, *args), loop=self)

    def set_default_executor(self, executor):
        if not isinstance(executor, concurrent.futures.ThreadPoolExecutor):
            raise TypeError('executor must be ThreadPoolExecutor instance')
        self._default_executor = executor

    def _getaddrinfo_debug(self, host, port, family, type, proto, flags):
        msg = [f"{host}:{port!r}"]
        if family:
            msg.append(f'family={family!r}')
        if type:
            msg.append(f'type={type!r}')
        if proto:
            msg.append(f'proto={proto!r}')
        if flags:
            msg.append(f'flags={flags!r}')
        msg = ', '.join(msg)
        logger.debug('Get address info %s', msg)

        t0 = self.time()
        addrinfo = socket.getaddrinfo(host, port, family, type, proto, flags)
        dt = self.time() - t0

        msg = f'Getting address info {msg} took {dt * 1e3:.3f}ms: {addrinfo!r}'
        if dt >= self.slow_callback_duration:
            logger.info(msg)
        else:
            logger.debug(msg)
        return addrinfo

    async def getaddrinfo(self, host, port, *,
                          family=0, type=0, proto=0, flags=0):
        if self._debug:
            getaddr_func = self._getaddrinfo_debug
        else:
            getaddr_func = socket.getaddrinfo

        return await self.run_in_executor(
            None, getaddr_func, host, port, family, type, proto, flags)

    async def getnameinfo(self, sockaddr, flags=0):
        return await self.run_in_executor(
            None, socket.getnameinfo, sockaddr, flags)

    async def sock_sendfile(self, sock, file, offset=0, count=None,
                            *, fallback=True):
        if self._debug and sock.gettimeout() != 0:
            raise ValueError("the socket must be non-blocking")
        _check_ssl_socket(sock)
        self._check_sendfile_params(sock, file, offset, count)
        try:
            return await self._sock_sendfile_native(sock, file,
                                                    offset, count)
        except exceptions.SendfileNotAvailableError as exc:
            if not fallback:
                raise
        return await self._sock_sendfile_fallback(sock, file,
                                                  offset, count)

    async def _sock_sendfile_native(self, sock, file, offset, count):
        
        
        raise exceptions.SendfileNotAvailableError(
            f"syscall sendfile is not available for socket {sock!r} "
            f"and file {file!r} combination")

    async def _sock_sendfile_fallback(self, sock, file, offset, count):
        if offset:
            file.seek(offset)
        blocksize = (
            min(count, constants.SENDFILE_FALLBACK_READBUFFER_SIZE)
            if count else constants.SENDFILE_FALLBACK_READBUFFER_SIZE
        )
        buf = bytearray(blocksize)
        total_sent = 0
        try:
            while True:
                if count:
                    blocksize = min(count - total_sent, blocksize)
                    if blocksize <= 0:
                        break
                view = memoryview(buf)[:blocksize]
                read = await self.run_in_executor(None, file.readinto, view)
                if not read:
                    break  
                await self.sock_sendall(sock, view[:read])
                total_sent += read
            return total_sent
        finally:
            if total_sent > 0 and hasattr(file, 'seek'):
                file.seek(offset + total_sent)

    def _check_sendfile_params(self, sock, file, offset, count):
        if 'b' not in getattr(file, 'mode', 'b'):
            raise ValueError("file should be opened in binary mode")
        if not sock.type == socket.SOCK_STREAM:
            raise ValueError("only SOCK_STREAM type sockets are supported")
        if count is not None:
            if not isinstance(count, int):
                raise TypeError(
                    "count must be a positive integer (got {!r})".format(count))
            if count <= 0:
                raise ValueError(
                    "count must be a positive integer (got {!r})".format(count))
        if not isinstance(offset, int):
            raise TypeError(
                "offset must be a non-negative integer (got {!r})".format(
                    offset))
        if offset < 0:
            raise ValueError(
                "offset must be a non-negative integer (got {!r})".format(
                    offset))

    async def _connect_sock(self, exceptions, addr_info, local_addr_infos=None):
        
        my_exceptions = []
        exceptions.append(my_exceptions)
        family, type_, proto, _, address = addr_info
        sock = None
        try:
            sock = socket.socket(family=family, type=type_, proto=proto)
            sock.setblocking(False)
            if local_addr_infos is not None:
                for lfamily, _, _, _, laddr in local_addr_infos:
                    
                    if lfamily != family:
                        continue
                    try:
                        sock.bind(laddr)
                        break
                    except OSError as exc:
                        msg = (
                            f'error while attempting to bind on '
                            f'address {laddr!r}: '
                            f'{exc.strerror.lower()}'
                        )
                        exc = OSError(exc.errno, msg)
                        my_exceptions.append(exc)
                else:  
                    if my_exceptions:
                        raise my_exceptions.pop()
                    else:
                        raise OSError(f"no matching local address with {family=} found")
            await self.sock_connect(sock, address)
            return sock
        except OSError as exc:
            my_exceptions.append(exc)
            if sock is not None:
                sock.close()
            raise
        except:
            if sock is not None:
                sock.close()
            raise
        finally:
            exceptions = my_exceptions = None

    async def create_connection(
            self, protocol_factory, host=None, port=None,
            *, ssl=None, family=0,
            proto=0, flags=0, sock=None,
            local_addr=None, server_hostname=None,
            ssl_handshake_timeout=None,
            ssl_shutdown_timeout=None,
            happy_eyeballs_delay=None, interleave=None,
            all_errors=False):
        
        if server_hostname is not None and not ssl:
            raise ValueError('server_hostname is only meaningful with ssl')

        if server_hostname is None and ssl:
            
            
            
            
            
            
            
            
            
            
            if not host:
                raise ValueError('You must set server_hostname '
                                 'when using ssl without a host')
            server_hostname = host

        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 sock is not None:
            _check_ssl_socket(sock)

        if happy_eyeballs_delay is not None and interleave is None:
            
            interleave = 1

        if host is not None or port is not None:
            if sock is not None:
                raise ValueError(
                    'host/port and sock can not be specified at the same time')

            infos = await self._ensure_resolved(
                (host, port), family=family,
                type=socket.SOCK_STREAM, proto=proto, flags=flags, loop=self)
            if not infos:
                raise OSError('getaddrinfo() returned empty list')

            if local_addr is not None:
                laddr_infos = await self._ensure_resolved(
                    local_addr, family=family,
                    type=socket.SOCK_STREAM, proto=proto,
                    flags=flags, loop=self)
                if not laddr_infos:
                    raise OSError('getaddrinfo() returned empty list')
            else:
                laddr_infos = None

            if interleave:
                infos = _interleave_addrinfos(infos, interleave)

            exceptions = []
            if happy_eyeballs_delay is None:
                
                for addrinfo in infos:
                    try:
                        sock = await self._connect_sock(
                            exceptions, addrinfo, laddr_infos)
                        break
                    except OSError:
                        continue
            else:  
                sock, _, _ = await staggered.staggered_race(
                    (functools.partial(self._connect_sock,
                                       exceptions, addrinfo, laddr_infos)
                     for addrinfo in infos),
                    happy_eyeballs_delay, loop=self)

            if sock is None:
                exceptions = [exc for sub in exceptions for exc in sub]
                try:
                    if all_errors:
                        raise ExceptionGroup("create_connection failed", exceptions)
                    if len(exceptions) == 1:
                        raise exceptions[0]
                    else:
                        
                        model = str(exceptions[0])
                        if all(str(exc) == model for exc in exceptions):
                            raise exceptions[0]
                        
                        
                        raise OSError('Multiple exceptions: {}'.format(
                            ', '.join(str(exc) for exc in exceptions)))
                finally:
                    exceptions = None

        else:
            if sock is None:
                raise ValueError(
                    'host and port was not specified and no sock specified')
            if sock.type != socket.SOCK_STREAM:
                
                raise ValueError(
                    f'A Stream Socket was expected, got {sock!r}')

        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)
        if self._debug:
            sock = transport.get_extra_info('socket')
            logger.debug("%r connected to %s:%r: (%r, %r)",
                         sock, host, port, transport, protocol)
        return transport, protocol

    async def _create_connection_transport(
            self, sock, protocol_factory, ssl,
            server_hostname, server_side=False,
            ssl_handshake_timeout=None,
            ssl_shutdown_timeout=None):

        sock.setblocking(False)

        protocol = protocol_factory()
        waiter = self.create_future()
        if ssl:
            sslcontext = None if isinstance(ssl, bool) else ssl
            transport = self._make_ssl_transport(
                sock, protocol, sslcontext, waiter,
                server_side=server_side, server_hostname=server_hostname,
                ssl_handshake_timeout=ssl_handshake_timeout,
                ssl_shutdown_timeout=ssl_shutdown_timeout)
        else:
            transport = self._make_socket_transport(sock, protocol, waiter)

        try:
            await waiter
        except:
            transport.close()
            raise

        return transport, protocol

    async def sendfile(self, transport, file, offset=0, count=None,
                       *, fallback=True):
        
        if transport.is_closing():
            raise RuntimeError("Transport is closing")
        mode = getattr(transport, '_sendfile_compatible',
                       constants._SendfileMode.UNSUPPORTED)
        if mode is constants._SendfileMode.UNSUPPORTED:
            raise RuntimeError(
                f"sendfile is not supported for transport {transport!r}")
        if mode is constants._SendfileMode.TRY_NATIVE:
            try:
                return await self._sendfile_native(transport, file,
                                                   offset, count)
            except exceptions.SendfileNotAvailableError as exc:
                if not fallback:
                    raise

        if not fallback:
            raise RuntimeError(
                f"fallback is disabled and native sendfile is not "
                f"supported for transport {transport!r}")

        return await self._sendfile_fallback(transport, file,
                                             offset, count)

    async def _sendfile_native(self, transp, file, offset, count):
        raise exceptions.SendfileNotAvailableError(
            "sendfile syscall is not supported")

    async def _sendfile_fallback(self, transp, file, offset, count):
        if offset:
            file.seek(offset)
        blocksize = min(count, 16384) if count else 16384
        buf = bytearray(blocksize)
        total_sent = 0
        proto = _SendfileFallbackProtocol(transp)
        try:
            while True:
                if count:
                    blocksize = min(count - total_sent, blocksize)
                    if blocksize <= 0:
                        return total_sent
                view = memoryview(buf)[:blocksize]
                read = await self.run_in_executor(None, file.readinto, view)
                if not read:
                    return total_sent  
                await proto.drain()
                transp.write(view[:read])
                total_sent += read
        finally:
            if total_sent > 0 and hasattr(file, 'seek'):
                file.seek(offset + total_sent)
            await proto.restore()

    async def start_tls(self, transport, protocol, sslcontext, *,
                        server_side=False,
                        server_hostname=None,
                        ssl_handshake_timeout=None,
                        ssl_shutdown_timeout=None):
        
        if ssl is None:
            raise RuntimeError('Python ssl module is not available')

        if not isinstance(sslcontext, ssl.SSLContext):
            raise TypeError(
                f'sslcontext is expected to be an instance of ssl.SSLContext, '
                f'got {sslcontext!r}')

        if not getattr(transport, '_start_tls_compatible', False):
            raise TypeError(
                f'transport {transport!r} is not supported by start_tls()')

        waiter = self.create_future()
        ssl_protocol = sslproto.SSLProtocol(
            self, protocol, sslcontext, waiter,
            server_side, server_hostname,
            ssl_handshake_timeout=ssl_handshake_timeout,
            ssl_shutdown_timeout=ssl_shutdown_timeout,
            call_connection_made=False)

        
        
        transport.pause_reading()
        transport.set_protocol(ssl_protocol)
        conmade_cb = self.call_soon(ssl_protocol.connection_made, transport)
        resume_cb = self.call_soon(transport.resume_reading)

        try:
            await waiter
        except BaseException:
            transport.close()
            conmade_cb.cancel()
            resume_cb.cancel()
            raise

        return ssl_protocol._app_transport

    async def create_datagram_endpoint(self, protocol_factory,
                                       local_addr=None, remote_addr=None, *,
                                       family=0, proto=0, flags=0,
                                       reuse_port=None,
                                       allow_broadcast=None, sock=None):
        
        if sock is not None:
            if sock.type != socket.SOCK_DGRAM:
                raise ValueError(
                    f'A UDP Socket was expected, got {sock!r}')
            if (local_addr or remote_addr or
                    family or proto or flags or
                    reuse_port or allow_broadcast):
                
                opts = dict(local_addr=local_addr, remote_addr=remote_addr,
                            family=family, proto=proto, flags=flags,
                            reuse_port=reuse_port,
                            allow_broadcast=allow_broadcast)
                problems = ', '.join(f'{k}={v}' for k, v in opts.items() if v)
                raise ValueError(
                    f'socket modifier keyword arguments can not be used '
                    f'when sock is specified. ({problems})')
            sock.setblocking(False)
            r_addr = None
        else:
            if not (local_addr or remote_addr):
                if family == 0:
                    raise ValueError('unexpected address family')
                addr_pairs_info = (((family, proto), (None, None)),)
            elif hasattr(socket, 'AF_UNIX') and family == socket.AF_UNIX:
                for addr in (local_addr, remote_addr):
                    if addr is not None and not isinstance(addr, str):
                        raise TypeError('string is expected')

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

                addr_pairs_info = (((family, proto),
                                    (local_addr, remote_addr)), )
            else:
                
                addr_infos = {}  
                for idx, addr in ((0, local_addr), (1, remote_addr)):
                    if addr is not None:
                        if not (isinstance(addr, tuple) and len(addr) == 2):
                            raise TypeError('2-tuple is expected')

                        infos = await self._ensure_resolved(
                            addr, family=family, type=socket.SOCK_DGRAM,
                            proto=proto, flags=flags, loop=self)
                        if not infos:
                            raise OSError('getaddrinfo() returned empty list')

                        for fam, _, pro, _, address in infos:
                            key = (fam, pro)
                            if key not in addr_infos:
                                addr_infos[key] = [None, None]
                            addr_infos[key][idx] = address

                
                addr_pairs_info = [
                    (key, addr_pair) for key, addr_pair in addr_infos.items()
                    if not ((local_addr and addr_pair[0] is None) or
                            (remote_addr and addr_pair[1] is None))]

                if not addr_pairs_info:
                    raise ValueError('can not get address information')

            exceptions = []

            for ((family, proto),
                 (local_address, remote_address)) in addr_pairs_info:
                sock = None
                r_addr = None
                try:
                    sock = socket.socket(
                        family=family, type=socket.SOCK_DGRAM, proto=proto)
                    if reuse_port:
                        _set_reuseport(sock)
                    if allow_broadcast:
                        sock.setsockopt(
                            socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
                    sock.setblocking(False)

                    if local_addr:
                        sock.bind(local_address)
                    if remote_addr:
                        if not allow_broadcast:
                            await self.sock_connect(sock, remote_address)
                        r_addr = remote_address
                except OSError as exc:
                    if sock is not None:
                        sock.close()
                    exceptions.append(exc)
                except:
                    if sock is not None:
                        sock.close()
                    raise
                else:
                    break
            else:
                raise exceptions[0]

        protocol = protocol_factory()
        waiter = self.create_future()
        transport = self._make_datagram_transport(
            sock, protocol, r_addr, waiter)
        if self._debug:
            if local_addr:
                logger.info("Datagram endpoint local_addr=%r remote_addr=%r "
                            "created: (%r, %r)",
                            local_addr, remote_addr, transport, protocol)
            else:
                logger.debug("Datagram endpoint remote_addr=%r created: "
                             "(%r, %r)",
                             remote_addr, transport, protocol)

        try:
            await waiter
        except:
            transport.close()
            raise

        return transport, protocol

    async def _ensure_resolved(self, address, *,
                               family=0, type=socket.SOCK_STREAM,
                               proto=0, flags=0, loop):
        host, port = address[:2]
        info = _ipaddr_info(host, port, family, type, proto, *address[2:])
        if info is not None:
            
            return [info]
        else:
            return await loop.getaddrinfo(host, port, family=family, type=type,
                                          proto=proto, flags=flags)

    async def _create_server_getaddrinfo(self, host, port, family, flags):
        infos = await self._ensure_resolved((host, port), family=family,
                                            type=socket.SOCK_STREAM,
                                            flags=flags, loop=self)
        if not infos:
            raise OSError(f'getaddrinfo({host!r}) returned empty list')
        return infos

    async def create_server(
            self, protocol_factory, host=None, port=None,
            *,
            family=socket.AF_UNSPEC,
            flags=socket.AI_PASSIVE,
            sock=None,
            backlog=100,
            ssl=None,
            reuse_address=None,
            reuse_port=None,
            ssl_handshake_timeout=None,
            ssl_shutdown_timeout=None,
            start_serving=True):
        
        if isinstance(ssl, bool):
            raise TypeError('ssl argument must be an SSLContext or None')

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

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

        if sock is not None:
            _check_ssl_socket(sock)

        if host is not None or port is not None:
            if sock is not None:
                raise ValueError(
                    'host/port and sock can not be specified at the same time')

            if reuse_address is None:
                reuse_address = os.name == "posix" and sys.platform != "cygwin"
            sockets = []
            if host == '':
                hosts = [None]
            elif (isinstance(host, str) or
                  not isinstance(host, collections.abc.Iterable)):
                hosts = [host]
            else:
                hosts = host

            fs = [self._create_server_getaddrinfo(host, port, family=family,
                                                  flags=flags)
                  for host in hosts]
            infos = await tasks.gather(*fs)
            infos = set(itertools.chain.from_iterable(infos))

            completed = False
            try:
                for res in infos:
                    af, socktype, proto, canonname, sa = res
                    try:
                        sock = socket.socket(af, socktype, proto)
                    except socket.error:
                        
                        if self._debug:
                            logger.warning('create_server() failed to create '
                                           'socket.socket(%r, %r, %r)',
                                           af, socktype, proto, exc_info=True)
                        continue
                    sockets.append(sock)
                    if reuse_address:
                        sock.setsockopt(
                            socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
                    if reuse_port:
                        _set_reuseport(sock)
                    
                    
                    
                    if (_HAS_IPv6 and
                            af == socket.AF_INET6 and
                            hasattr(socket, 'IPPROTO_IPV6')):
                        sock.setsockopt(socket.IPPROTO_IPV6,
                                        socket.IPV6_V6ONLY,
                                        True)
                    try:
                        sock.bind(sa)
                    except OSError as err:
                        raise OSError(err.errno, 'error while attempting '
                                      'to bind on address %r: %s'
                                      % (sa, err.strerror.lower())) from None
                completed = True
            finally:
                if not completed:
                    for sock in sockets:
                        sock.close()
        else:
            if sock is None:
                raise ValueError('Neither host/port nor sock were specified')
            if sock.type != socket.SOCK_STREAM:
                raise ValueError(f'A Stream Socket was expected, got {sock!r}')
            sockets = [sock]

        for sock in sockets:
            sock.setblocking(False)

        server = Server(self, sockets, protocol_factory,
                        ssl, backlog, ssl_handshake_timeout,
                        ssl_shutdown_timeout)
        if start_serving:
            server._start_serving()
            
            
            await tasks.sleep(0)

        if self._debug:
            logger.info("%r is serving", server)
        return server

    async def connect_accepted_socket(
            self, protocol_factory, sock,
            *, ssl=None,
            ssl_handshake_timeout=None,
            ssl_shutdown_timeout=None):
        if sock.type != socket.SOCK_STREAM:
            raise ValueError(f'A Stream Socket was expected, got {sock!r}')

        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 sock is not None:
            _check_ssl_socket(sock)

        transport, protocol = await self._create_connection_transport(
            sock, protocol_factory, ssl, '', server_side=True,
            ssl_handshake_timeout=ssl_handshake_timeout,
            ssl_shutdown_timeout=ssl_shutdown_timeout)
        if self._debug:
            
            
            sock = transport.get_extra_info('socket')
            logger.debug("%r handled: (%r, %r)", sock, transport, protocol)
        return transport, protocol

    async def connect_read_pipe(self, protocol_factory, pipe):
        protocol = protocol_factory()
        waiter = self.create_future()
        transport = self._make_read_pipe_transport(pipe, protocol, waiter)

        try:
            await waiter
        except:
            transport.close()
            raise

        if self._debug:
            logger.debug('Read pipe %r connected: (%r, %r)',
                         pipe.fileno(), transport, protocol)
        return transport, protocol

    async def connect_write_pipe(self, protocol_factory, pipe):
        protocol = protocol_factory()
        waiter = self.create_future()
        transport = self._make_write_pipe_transport(pipe, protocol, waiter)

        try:
            await waiter
        except:
            transport.close()
            raise

        if self._debug:
            logger.debug('Write pipe %r connected: (%r, %r)',
                         pipe.fileno(), transport, protocol)
        return transport, protocol

    def _log_subprocess(self, msg, stdin, stdout, stderr):
        info = [msg]
        if stdin is not None:
            info.append(f'stdin={_format_pipe(stdin)}')
        if stdout is not None and stderr == subprocess.STDOUT:
            info.append(f'stdout=stderr={_format_pipe(stdout)}')
        else:
            if stdout is not None:
                info.append(f'stdout={_format_pipe(stdout)}')
            if stderr is not None:
                info.append(f'stderr={_format_pipe(stderr)}')
        logger.debug(' '.join(info))

    async def subprocess_shell(self, protocol_factory, cmd, *,
                               stdin=subprocess.PIPE,
                               stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE,
                               universal_newlines=False,
                               shell=True, bufsize=0,
                               encoding=None, errors=None, text=None,
                               **kwargs):
        if not isinstance(cmd, (bytes, str)):
            raise ValueError("cmd must be a string")
        if universal_newlines:
            raise ValueError("universal_newlines must be False")
        if not shell:
            raise ValueError("shell must be True")
        if bufsize != 0:
            raise ValueError("bufsize must be 0")
        if text:
            raise ValueError("text must be False")
        if encoding is not None:
            raise ValueError("encoding must be None")
        if errors is not None:
            raise ValueError("errors must be None")

        protocol = protocol_factory()
        debug_log = None
        if self._debug:
            
            
            debug_log = 'run shell command %r' % cmd
            self._log_subprocess(debug_log, stdin, stdout, stderr)
        transport = await self._make_subprocess_transport(
            protocol, cmd, True, stdin, stdout, stderr, bufsize, **kwargs)
        if self._debug and debug_log is not None:
            logger.info('%s: %r', debug_log, transport)
        return transport, protocol

    async def subprocess_exec(self, protocol_factory, program, *args,
                              stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE, universal_newlines=False,
                              shell=False, bufsize=0,
                              encoding=None, errors=None, text=None,
                              **kwargs):
        if universal_newlines:
            raise ValueError("universal_newlines must be False")
        if shell:
            raise ValueError("shell must be False")
        if bufsize != 0:
            raise ValueError("bufsize must be 0")
        if text:
            raise ValueError("text must be False")
        if encoding is not None:
            raise ValueError("encoding must be None")
        if errors is not None:
            raise ValueError("errors must be None")

        popen_args = (program,) + args
        protocol = protocol_factory()
        debug_log = None
        if self._debug:
            
            
            debug_log = f'execute program {program!r}'
            self._log_subprocess(debug_log, stdin, stdout, stderr)
        transport = await self._make_subprocess_transport(
            protocol, popen_args, False, stdin, stdout, stderr,
            bufsize, **kwargs)
        if self._debug and debug_log is not None:
            logger.info('%s: %r', debug_log, transport)
        return transport, protocol

    def get_exception_handler(self):
        
        return self._exception_handler

    def set_exception_handler(self, handler):
        
        if handler is not None and not callable(handler):
            raise TypeError(f'A callable object or None is expected, '
                            f'got {handler!r}')
        self._exception_handler = handler

    def default_exception_handler(self, context):
        
        message = context.get('message')
        if not message:
            message = 'Unhandled exception in event loop'

        exception = context.get('exception')
        if exception is not None:
            exc_info = (type(exception), exception, exception.__traceback__)
        else:
            exc_info = False

        if ('source_traceback' not in context and
                self._current_handle is not None and
                self._current_handle._source_traceback):
            context['handle_traceback'] = \
                self._current_handle._source_traceback

        log_lines = [message]
        for key in sorted(context):
            if key in {'message', 'exception'}:
                continue
            value = context[key]
            if key == 'source_traceback':
                tb = ''.join(traceback.format_list(value))
                value = 'Object created at (most recent call last):\n'
                value += tb.rstrip()
            elif key == 'handle_traceback':
                tb = ''.join(traceback.format_list(value))
                value = 'Handle created at (most recent call last):\n'
                value += tb.rstrip()
            else:
                value = repr(value)
            log_lines.append(f'{key}: {value}')

        logger.error('\n'.join(log_lines), exc_info=exc_info)

    def call_exception_handler(self, context):
        
        if self._exception_handler is None:
            try:
                self.default_exception_handler(context)
            except (SystemExit, KeyboardInterrupt):
                raise
            except BaseException:
                
                
                
                logger.error('Exception in default exception handler',
                             exc_info=True)
        else:
            try:
                ctx = None
                thing = context.get("task")
                if thing is None:
                    
                    
                    
                    thing = context.get("future")
                if thing is None:
                    
                    thing = context.get("handle")
                if thing is not None and hasattr(thing, "get_context"):
                    ctx = thing.get_context()
                if ctx is not None and hasattr(ctx, "run"):
                    ctx.run(self._exception_handler, self, context)
                else:
                    self._exception_handler(self, context)
            except (SystemExit, KeyboardInterrupt):
                raise
            except BaseException as exc:
                
                try:
                    
                    self.default_exception_handler({
                        'message': 'Unhandled error in exception handler',
                        'exception': exc,
                        'context': context,
                    })
                except (SystemExit, KeyboardInterrupt):
                    raise
                except BaseException:
                    
                    
                    logger.error('Exception in default exception handler '
                                 'while handling an unexpected error '
                                 'in custom exception handler',
                                 exc_info=True)

    def _add_callback(self, handle):
        
        if not handle._cancelled:
            self._ready.append(handle)

    def _add_callback_signalsafe(self, handle):
        
        self._add_callback(handle)
        self._write_to_self()

    def _timer_handle_cancelled(self, handle):
        
        if handle._scheduled:
            self._timer_cancelled_count += 1

    def _run_once(self):

        sched_count = len(self._scheduled)
        if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and
            self._timer_cancelled_count / sched_count >
                _MIN_CANCELLED_TIMER_HANDLES_FRACTION):
            
            
            new_scheduled = []
            for handle in self._scheduled:
                if handle._cancelled:
                    handle._scheduled = False
                else:
                    new_scheduled.append(handle)

            heapq.heapify(new_scheduled)
            self._scheduled = new_scheduled
            self._timer_cancelled_count = 0
        else:
            
            while self._scheduled and self._scheduled[0]._cancelled:
                self._timer_cancelled_count -= 1
                handle = heapq.heappop(self._scheduled)
                handle._scheduled = False

        timeout = None
        if self._ready or self._stopping:
            timeout = 0
        elif self._scheduled:
            
            timeout = self._scheduled[0]._when - self.time()
            if timeout > MAXIMUM_SELECT_TIMEOUT:
                timeout = MAXIMUM_SELECT_TIMEOUT
            elif timeout < 0:
                timeout = 0

        event_list = self._selector.select(timeout)
        self._process_events(event_list)
        
        event_list = None

        
        end_time = self.time() + self._clock_resolution
        while self._scheduled:
            handle = self._scheduled[0]
            if handle._when >= end_time:
                break
            handle = heapq.heappop(self._scheduled)
            handle._scheduled = False
            self._ready.append(handle)

        ntodo = len(self._ready)
        for i in range(ntodo):
            handle = self._ready.popleft()
            if handle._cancelled:
                continue
            if self._debug:
                try:
                    self._current_handle = handle
                    t0 = self.time()
                    handle._run()
                    dt = self.time() - t0
                    if dt >= self.slow_callback_duration:
                        logger.warning('Executing %s took %.3f seconds',
                                       _format_handle(handle), dt)
                finally:
                    self._current_handle = None
            else:
                handle._run()
        handle = None  

    def _set_coroutine_origin_tracking(self, enabled):
        if bool(enabled) == bool(self._coroutine_origin_tracking_enabled):
            return

        if enabled:
            self._coroutine_origin_tracking_saved_depth = (
                sys.get_coroutine_origin_tracking_depth())
            sys.set_coroutine_origin_tracking_depth(
                constants.DEBUG_STACK_DEPTH)
        else:
            sys.set_coroutine_origin_tracking_depth(
                self._coroutine_origin_tracking_saved_depth)

        self._coroutine_origin_tracking_enabled = enabled

    def get_debug(self):
        return self._debug

    def set_debug(self, enabled):
        self._debug = enabled

        if self.is_running():
            self.call_soon_threadsafe(self._set_coroutine_origin_tracking, enabled)