event loop and event loop policy all abstracteventlooppolicy abstracteventloop abstractserver handle timerhandle geteventlooppolicy seteventlooppolicy geteventloop seteventloop neweventloop getchildwatcher setchildwatcher setrunningloop getrunningloop getrunningloop import contextvars import os import signal import socket import subprocess import sys import threading from import formathelpers class handle keep a representation in debug mode to keep callback and parameters for example to log the warning executing handle took 2 5 second object returned by timed callback registration methods slots scheduled when def initself when callback args loop contextnone super initcallback args loop context if self sourcetraceback del self sourcetraceback1 self when when self scheduled false def reprinfoself info super reprinfo pos 2 if self cancelled else 1 info insertpos f whenself when return info def hashself return hashself when def ltself other if isinstanceother timerhandle return self when other when return notimplemented def leself other if isinstanceother timerhandle return self when other when or self eqother return notimplemented def gtself other if isinstanceother timerhandle return self when other when return notimplemented def geself other if isinstanceother timerhandle return self when other when or self eqother return notimplemented def eqself other if isinstanceother timerhandle return self when other when and self callback other callback and self args other args and self cancelled other cancelled return notimplemented def cancelself if not self cancelled self loop timerhandlecancelledself super cancel def whenself return self when class abstractserver stop serving this leaves existing connections open raise notimplementederror def getloopself return true if the server is accepting connections raise notimplementederror async def startservingself raise notimplementederror async def serveforeverself raise notimplementederror async def waitclosedself abstract event loop running and stopping the event loop def runforeverself run the event loop until a future is done return the future s result or raise its exception stop the event loop as soon as reasonable exactly how soon that is may depend on the implementation but no more io callbacks should be scheduled return whether the event loop is currently running raise notimplementederror def isclosedself close the loop the loop should not be running this is idempotent and irreversible no other methods should be called after this one shutdown all active asynchronous generators raise notimplementederror async def shutdowndefaultexecutorself methods scheduling callbacks all these return handles notification that a timerhandle has been cancelled raise notimplementederror def callsoonself callback args contextnone return self calllater0 callback args contextcontext def calllaterself delay callback args contextnone raise notimplementederror def callatself when callback args contextnone raise notimplementederror def timeself raise notimplementederror def createfutureself raise notimplementederror method scheduling a coroutine object create a task def createtaskself coro namenone contextnone raise notimplementederror methods for interacting with threads def callsoonthreadsafeself callback args contextnone raise notimplementederror def runinexecutorself executor func args raise notimplementederror def setdefaultexecutorself executor raise notimplementederror network io methods returning futures async def getaddrinfoself host port family0 type0 proto0 flags0 raise notimplementederror async def getnameinfoself sockaddr flags0 raise notimplementederror async def createconnection self protocolfactory hostnone portnone sslnone family0 proto0 flags0 socknone localaddrnone serverhostnamenone sslhandshaketimeoutnone sslshutdowntimeoutnone happyeyeballsdelaynone interleavenone raise notimplementederror async def createserver self protocolfactory hostnone portnone familysocket afunspec flagssocket aipassive socknone backlog100 sslnone reuseaddressnone reuseportnone sslhandshaketimeoutnone sslshutdowntimeoutnone startservingtrue raise notimplementederror async def sendfileself transport file offset0 countnone fallbacktrue raise notimplementederror async def starttlsself transport protocol sslcontext serversidefalse serverhostnamenone sslhandshaketimeoutnone sslshutdowntimeoutnone raise notimplementederror async def createunixconnection self protocolfactory pathnone sslnone socknone serverhostnamenone sslhandshaketimeoutnone sslshutdowntimeoutnone raise notimplementederror async def createunixserver self protocolfactory pathnone socknone backlog100 sslnone sslhandshaketimeoutnone sslshutdowntimeoutnone startservingtrue raise notimplementederror async def connectacceptedsocket self protocolfactory sock sslnone sslhandshaketimeoutnone sslshutdowntimeoutnone raise notimplementederror async def createdatagramendpointself protocolfactory localaddrnone remoteaddrnone family0 proto0 flags0 reuseaddressnone reuseportnone allowbroadcastnone socknone raise notimplementederror pipes and subprocesses async def connectreadpipeself protocolfactory pipe the reason to accept filelike object instead of just file descriptor is we need to own pipe and close it at transport finishing can got complicated errors if pass f fileno close fd in pipe transport then close f and vice versa register write pipe in event loop protocolfactory should instantiate object with baseprotocol interface pipe is filelike object already switched to nonblocking return pair transport protocol where transport support writetransport interface the reason to accept filelike object instead of just file descriptor is we need to own pipe and close it at transport finishing can got complicated errors if pass f fileno close fd in pipe transport then close f and vice versa raise notimplementederror async def subprocessshellself protocolfactory cmd stdinsubprocess pipe stdoutsubprocess pipe stderrsubprocess pipe kwargs raise notimplementederror async def subprocessexecself protocolfactory args stdinsubprocess pipe stdoutsubprocess pipe stderrsubprocess pipe kwargs raise notimplementederror readybased callback registration methods the add methods return none the remove methods return true if something was removed false if there was nothing to delete def addreaderself fd callback args raise notimplementederror def removereaderself fd raise notimplementederror def addwriterself fd callback args raise notimplementederror def removewriterself fd raise notimplementederror completion based io methods returning futures async def sockrecvself sock nbytes raise notimplementederror async def sockrecvintoself sock buf raise notimplementederror async def sockrecvfromself sock bufsize raise notimplementederror async def sockrecvfromintoself sock buf nbytes0 raise notimplementederror async def socksendallself sock data raise notimplementederror async def socksendtoself sock data address raise notimplementederror async def sockconnectself sock address raise notimplementederror async def sockacceptself sock raise notimplementederror async def socksendfileself sock file offset0 countnone fallbacknone raise notimplementederror signal handling def addsignalhandlerself sig callback args raise notimplementederror def removesignalhandlerself sig raise notimplementederror task factory def settaskfactoryself factory raise notimplementederror def gettaskfactoryself raise notimplementederror error handlers def getexceptionhandlerself raise notimplementederror def setexceptionhandlerself handler raise notimplementederror def defaultexceptionhandlerself context raise notimplementederror def callexceptionhandlerself context raise notimplementederror debug flag management def getdebugself raise notimplementederror def setdebugself enabled raise notimplementederror class abstracteventlooppolicy get the event loop for the current context returns an event loop object implementing the abstracteventloop interface or raises an exception in case no event loop has been set for the current context and the current policy does not specify to create one it should never return none raise notimplementederror def seteventloopself loop create and return a new event loop object according to this policy s rules if there s need to set this loop as the event loop for the current context seteventloop must be called explicitly raise notimplementederror child processes handling unix only def getchildwatcherself get the watcher for child processes raise notimplementederror def setchildwatcherself watcher default policy implementation for accessing the event loop in this policy each thread has its own event loop however we only automatically create an event loop by default for the main thread other threads by default have no event loop other policies may have different rules e g a single global event loop or automatically creating an event loop per thread or using some other notion of context to which an event loop is associated get the event loop for the current context returns an instance of eventloop or raises an exception move up the call stack so that the warning is attached to the line outside asyncio itself set the event loop self local setcalled true if loop is not none and not isinstanceloop abstracteventloop raise typeerrorfloop must be an instance of abstracteventloop or none not typeloop name self local loop loop def neweventloopself return self loopfactory event loop policy the policy itself is always global even if the policy s rules say that there is an event loop per thread or other notion of context the default policy is installed by the first call to geteventlooppolicy eventlooppolicy none lock for protecting the onthefly creation of the event loop policy lock threading lock a tls for the running event loop used by getrunningloop class runningloopthreading local looppid none none runningloop runningloop def getrunningloop note this function is implemented in c see asynciomodule c loop getrunningloop if loop is none raise runtimeerror no running event loop return loop def getrunningloop note this function is implemented in c see asynciomodule c runningloop pid runningloop looppid if runningloop is not none and pid os getpid return runningloop def setrunninglooploop note this function is implemented in c see asynciomodule c runningloop looppid loop os getpid def initeventlooppolicy global eventlooppolicy with lock if eventlooppolicy is none pragma no branch from import defaulteventlooppolicy eventlooppolicy defaulteventlooppolicy def geteventlooppolicy set the current event loop policy if policy is none the default policy is restored global eventlooppolicy if policy is not none and not isinstancepolicy abstracteventlooppolicy raise typeerrorfpolicy must be an instance of abstracteventlooppolicy or none not typepolicy name eventlooppolicy policy def geteventloop note this function is implemented in c see asynciomodule c currentloop getrunningloop if currentloop is not none return currentloop return geteventlooppolicy geteventloop def seteventlooploop equivalent to calling geteventlooppolicy neweventloop return geteventlooppolicy neweventloop def getchildwatcher equivalent to calling geteventlooppolicy setchildwatcherwatcher return geteventlooppolicy setchildwatcherwatcher alias purepython implementations for testing purposes pygetrunningloop getrunningloop pysetrunningloop setrunningloop pygetrunningloop getrunningloop pygeteventloop geteventloop try geteventloop is one of the most frequently called functions in asyncio pure python implementation is about 4 times slower than caccelerated from asyncio import getrunningloop setrunningloop getrunningloop geteventloop except importerror pass else alias c implementations for testing purposes cgetrunningloop getrunningloop csetrunningloop setrunningloop cgetrunningloop getrunningloop cgeteventloop geteventloop if hasattros fork def onfork reset the loop and wakeupfd in the forked child process if eventlooppolicy is not none eventlooppolicy local basedefaulteventlooppolicy local setrunningloopnone signal setwakeupfd1 os registeratforkafterinchildonfork object returned by callback registration methods keep a representation in debug mode to keep callback and parameters for example to log the warning executing handle took 2 5 second needed to break cycles when an exception occurs object returned by timed callback registration methods return a scheduled callback time the time is an absolute timestamp using the same time reference as loop time abstract server returned by create_server stop serving this leaves existing connections open get the event loop the server object is attached to return true if the server is accepting connections start accepting connections this method is idempotent so it can be called when the server is already being serving start accepting connections until the coroutine is cancelled the server is closed when the coroutine is cancelled coroutine to wait until service is closed abstract event loop running and stopping the event loop run the event loop until stop is called run the event loop until a future is done return the future s result or raise its exception stop the event loop as soon as reasonable exactly how soon that is may depend on the implementation but no more i o callbacks should be scheduled return whether the event loop is currently running returns true if the event loop was closed close the loop the loop should not be running this is idempotent and irreversible no other methods should be called after this one shutdown all active asynchronous generators schedule the shutdown of the default executor methods scheduling callbacks all these return handles notification that a timerhandle has been cancelled method scheduling a coroutine object create a task methods for interacting with threads network i o methods returning futures a coroutine which creates a tcp server bound to host and port the return value is a server object which can be used to stop the service if host is an empty string or none all interfaces are assumed and a list of multiple sockets will be returned most likely one for ipv4 and another one for ipv6 the host parameter can also be a sequence e g list of hosts to bind to family can be set to either af_inet or af_inet6 to force the socket to use ipv4 or ipv6 if not set it will be determined from host defaults to af_unspec flags is a bitmask for getaddrinfo sock can optionally be specified in order to use a preexisting socket object backlog is the maximum number of queued connections passed to listen defaults to 100 ssl can be set to an sslcontext to enable ssl over the accepted connections reuse_address tells the kernel to reuse a local socket in time_wait state without waiting for its natural timeout to expire if not specified will automatically be set to true on unix reuse_port tells the kernel to allow this endpoint to be bound to the same port as other existing endpoints are bound to so long as they all set this flag when being created this option is not supported on windows ssl_handshake_timeout is the time in seconds that an ssl server will wait for completion of the ssl handshake before aborting the connection default is 60s ssl_shutdown_timeout is the time in seconds that an ssl server will wait for completion of the ssl shutdown procedure before aborting the connection default is 30s start_serving set to true default causes the created server to start accepting connections immediately when set to false the user should await server start_serving or server serve_forever to make the server to start accepting connections send a file through a transport return an amount of sent bytes upgrade a transport to tls return a new transport that protocol should start using immediately a coroutine which creates a unix domain socket server the return value is a server object which can be used to stop the service path is a str representing a file system path to bind the server socket to sock can optionally be specified in order to use a preexisting socket object backlog is the maximum number of queued connections passed to listen defaults to 100 ssl can be set to an sslcontext to enable ssl over the accepted connections ssl_handshake_timeout is the time in seconds that an ssl server will wait for the ssl handshake to complete defaults to 60s ssl_shutdown_timeout is the time in seconds that an ssl server will wait for the ssl shutdown to finish defaults to 30s start_serving set to true default causes the created server to start accepting connections immediately when set to false the user should await server start_serving or server serve_forever to make the server to start accepting connections handle an accepted connection this is used by servers that accept connections outside of asyncio but use asyncio to handle connections this method is a coroutine when completed the coroutine returns a transport protocol pair a coroutine which creates a datagram endpoint this method will try to establish the endpoint in the background when successful the coroutine returns a transport protocol pair protocol_factory must be a callable returning a protocol instance socket family af_inet socket af_inet6 or socket af_unix depending on host or family if specified socket type sock_dgram reuse_address tells the kernel to reuse a local socket in time_wait state without waiting for its natural timeout to expire if not specified it will automatically be set to true on unix reuse_port tells the kernel to allow this endpoint to be bound to the same port as other existing endpoints are bound to so long as they all set this flag when being created this option is not supported on windows and some unix s if the py data socket so_reuseport constant is not defined then this capability is unsupported allow_broadcast tells the kernel to allow this endpoint to send messages to the broadcast address sock can optionally be specified in order to use a preexisting socket object pipes and subprocesses register read pipe in event loop set the pipe to non blocking mode protocol_factory should instantiate object with protocol interface pipe is a file like object return pair transport protocol where transport supports the readtransport interface the reason to accept file like object instead of just file descriptor is we need to own pipe and close it at transport finishing can got complicated errors if pass f fileno close fd in pipe transport then close f and vice versa register write pipe in event loop protocol_factory should instantiate object with baseprotocol interface pipe is file like object already switched to nonblocking return pair transport protocol where transport support writetransport interface the reason to accept file like object instead of just file descriptor is we need to own pipe and close it at transport finishing can got complicated errors if pass f fileno close fd in pipe transport then close f and vice versa ready based callback registration methods the add_ methods return none the remove_ methods return true if something was removed false if there was nothing to delete completion based i o methods returning futures signal handling task factory error handlers debug flag management abstract policy for accessing the event loop get the event loop for the current context returns an event loop object implementing the abstracteventloop interface or raises an exception in case no event loop has been set for the current context and the current policy does not specify to create one it should never return none set the event loop for the current context to loop create and return a new event loop object according to this policy s rules if there s need to set this loop as the event loop for the current context set_event_loop must be called explicitly child processes handling unix only set the watcher for child processes default policy implementation for accessing the event loop in this policy each thread has its own event loop however we only automatically create an event loop by default for the main thread other threads by default have no event loop other policies may have different rules e g a single global event loop or automatically creating an event loop per thread or using some other notion of context to which an event loop is associated get the event loop for the current context returns an instance of eventloop or raises an exception move up the call stack so that the warning is attached to the line outside asyncio itself set the event loop create a new event loop you must call set_event_loop to make this the current event loop event loop policy the policy itself is always global even if the policy s rules say that there is an event loop per thread or other notion of context the default policy is installed by the first call to get_event_loop_policy lock for protecting the on the fly creation of the event loop policy a tls for the running event loop used by _get_running_loop return the running event loop raise a runtimeerror if there is none this function is thread specific note this function is implemented in c see _asynciomodule c return the running event loop or none this is a low level function intended to be used by event loops this function is thread specific note this function is implemented in c see _asynciomodule c set the running event loop this is a low level function intended to be used by event loops this function is thread specific note this function is implemented in c see _asynciomodule c pragma no branch get the current event loop policy set the current event loop policy if policy is none the default policy is restored return an asyncio event loop when called from a coroutine or a callback e g scheduled with call_soon or similar api this function will always return the running event loop if there is no running event loop set the function will return the result of get_event_loop_policy get_event_loop call note this function is implemented in c see _asynciomodule c equivalent to calling get_event_loop_policy set_event_loop loop equivalent to calling get_event_loop_policy new_event_loop equivalent to calling get_event_loop_policy get_child_watcher equivalent to calling get_event_loop_policy set_child_watcher watcher alias pure python implementations for testing purposes get_event_loop is one of the most frequently called functions in asyncio pure python implementation is about 4 times slower than c accelerated alias c implementations for testing purposes reset the loop and wakeupfd in the forked child process
__all__ = (
    'AbstractEventLoopPolicy',
    'AbstractEventLoop', 'AbstractServer',
    'Handle', 'TimerHandle',
    'get_event_loop_policy', 'set_event_loop_policy',
    'get_event_loop', 'set_event_loop', 'new_event_loop',
    'get_child_watcher', 'set_child_watcher',
    '_set_running_loop', 'get_running_loop',
    '_get_running_loop',
)

import contextvars
import os
import signal
import socket
import subprocess
import sys
import threading

from . import format_helpers


class Handle:
    

    __slots__ = ('_callback', '_args', '_cancelled', '_loop',
                 '_source_traceback', '_repr', '__weakref__',
                 '_context')

    def __init__(self, callback, args, loop, context=None):
        if context is None:
            context = contextvars.copy_context()
        self._context = context
        self._loop = loop
        self._callback = callback
        self._args = args
        self._cancelled = False
        self._repr = None
        if self._loop.get_debug():
            self._source_traceback = format_helpers.extract_stack(
                sys._getframe(1))
        else:
            self._source_traceback = None

    def _repr_info(self):
        info = [self.__class__.__name__]
        if self._cancelled:
            info.append('cancelled')
        if self._callback is not None:
            info.append(format_helpers._format_callback_source(
                self._callback, self._args))
        if self._source_traceback:
            frame = self._source_traceback[-1]
            info.append(f'created at {frame[0]}:{frame[1]}')
        return info

    def __repr__(self):
        if self._repr is not None:
            return self._repr
        info = self._repr_info()
        return '<{}>'.format(' '.join(info))

    def get_context(self):
        return self._context

    def cancel(self):
        if not self._cancelled:
            self._cancelled = True
            if self._loop.get_debug():
                
                
                
                self._repr = repr(self)
            self._callback = None
            self._args = None

    def cancelled(self):
        return self._cancelled

    def _run(self):
        try:
            self._context.run(self._callback, *self._args)
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException as exc:
            cb = format_helpers._format_callback_source(
                self._callback, self._args)
            msg = f'Exception in callback {cb}'
            context = {
                'message': msg,
                'exception': exc,
                'handle': self,
            }
            if self._source_traceback:
                context['source_traceback'] = self._source_traceback
            self._loop.call_exception_handler(context)
        self = None  


class TimerHandle(Handle):
    

    __slots__ = ['_scheduled', '_when']

    def __init__(self, when, callback, args, loop, context=None):
        super().__init__(callback, args, loop, context)
        if self._source_traceback:
            del self._source_traceback[-1]
        self._when = when
        self._scheduled = False

    def _repr_info(self):
        info = super()._repr_info()
        pos = 2 if self._cancelled else 1
        info.insert(pos, f'when={self._when}')
        return info

    def __hash__(self):
        return hash(self._when)

    def __lt__(self, other):
        if isinstance(other, TimerHandle):
            return self._when < other._when
        return NotImplemented

    def __le__(self, other):
        if isinstance(other, TimerHandle):
            return self._when < other._when or self.__eq__(other)
        return NotImplemented

    def __gt__(self, other):
        if isinstance(other, TimerHandle):
            return self._when > other._when
        return NotImplemented

    def __ge__(self, other):
        if isinstance(other, TimerHandle):
            return self._when > other._when or self.__eq__(other)
        return NotImplemented

    def __eq__(self, other):
        if isinstance(other, TimerHandle):
            return (self._when == other._when and
                    self._callback == other._callback and
                    self._args == other._args and
                    self._cancelled == other._cancelled)
        return NotImplemented

    def cancel(self):
        if not self._cancelled:
            self._loop._timer_handle_cancelled(self)
        super().cancel()

    def when(self):
        
        return self._when


class AbstractServer:
    

    def close(self):
        
        raise NotImplementedError

    def get_loop(self):
        
        raise NotImplementedError

    def is_serving(self):
        
        raise NotImplementedError

    async def start_serving(self):
        
        raise NotImplementedError

    async def serve_forever(self):
        
        raise NotImplementedError

    async def wait_closed(self):
        
        raise NotImplementedError

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        self.close()
        await self.wait_closed()


class AbstractEventLoop:
    

    

    def run_forever(self):
        
        raise NotImplementedError

    def run_until_complete(self, future):
        
        raise NotImplementedError

    def stop(self):
        
        raise NotImplementedError

    def is_running(self):
        
        raise NotImplementedError

    def is_closed(self):
        
        raise NotImplementedError

    def close(self):
        
        raise NotImplementedError

    async def shutdown_asyncgens(self):
        
        raise NotImplementedError

    async def shutdown_default_executor(self):
        
        raise NotImplementedError

    

    def _timer_handle_cancelled(self, handle):
        
        raise NotImplementedError

    def call_soon(self, callback, *args, context=None):
        return self.call_later(0, callback, *args, context=context)

    def call_later(self, delay, callback, *args, context=None):
        raise NotImplementedError

    def call_at(self, when, callback, *args, context=None):
        raise NotImplementedError

    def time(self):
        raise NotImplementedError

    def create_future(self):
        raise NotImplementedError

    

    def create_task(self, coro, *, name=None, context=None):
        raise NotImplementedError

    

    def call_soon_threadsafe(self, callback, *args, context=None):
        raise NotImplementedError

    def run_in_executor(self, executor, func, *args):
        raise NotImplementedError

    def set_default_executor(self, executor):
        raise NotImplementedError

    

    async def getaddrinfo(self, host, port, *,
                          family=0, type=0, proto=0, flags=0):
        raise NotImplementedError

    async def getnameinfo(self, sockaddr, flags=0):
        raise NotImplementedError

    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):
        raise NotImplementedError

    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):
        
        raise NotImplementedError

    async def sendfile(self, transport, file, offset=0, count=None,
                       *, fallback=True):
        
        raise NotImplementedError

    async def start_tls(self, transport, protocol, sslcontext, *,
                        server_side=False,
                        server_hostname=None,
                        ssl_handshake_timeout=None,
                        ssl_shutdown_timeout=None):
        
        raise NotImplementedError

    async def create_unix_connection(
            self, protocol_factory, path=None, *,
            ssl=None, sock=None,
            server_hostname=None,
            ssl_handshake_timeout=None,
            ssl_shutdown_timeout=None):
        raise NotImplementedError

    async def create_unix_server(
            self, protocol_factory, path=None, *,
            sock=None, backlog=100, ssl=None,
            ssl_handshake_timeout=None,
            ssl_shutdown_timeout=None,
            start_serving=True):
        
        raise NotImplementedError

    async def connect_accepted_socket(
            self, protocol_factory, sock,
            *, ssl=None,
            ssl_handshake_timeout=None,
            ssl_shutdown_timeout=None):
        
        raise NotImplementedError

    async def create_datagram_endpoint(self, protocol_factory,
                                       local_addr=None, remote_addr=None, *,
                                       family=0, proto=0, flags=0,
                                       reuse_address=None, reuse_port=None,
                                       allow_broadcast=None, sock=None):
        
        raise NotImplementedError

    

    async def connect_read_pipe(self, protocol_factory, pipe):
        
        
        
        
        
        raise NotImplementedError

    async def connect_write_pipe(self, protocol_factory, pipe):
        
        
        
        
        
        raise NotImplementedError

    async def subprocess_shell(self, protocol_factory, cmd, *,
                               stdin=subprocess.PIPE,
                               stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE,
                               **kwargs):
        raise NotImplementedError

    async def subprocess_exec(self, protocol_factory, *args,
                              stdin=subprocess.PIPE,
                              stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE,
                              **kwargs):
        raise NotImplementedError


    def add_reader(self, fd, callback, *args):
        raise NotImplementedError

    def remove_reader(self, fd):
        raise NotImplementedError

    def add_writer(self, fd, callback, *args):
        raise NotImplementedError

    def remove_writer(self, fd):
        raise NotImplementedError

    
    async def sock_recv(self, sock, nbytes):
        raise NotImplementedError

    async def sock_recv_into(self, sock, buf):
        raise NotImplementedError

    async def sock_recvfrom(self, sock, bufsize):
        raise NotImplementedError

    async def sock_recvfrom_into(self, sock, buf, nbytes=0):
        raise NotImplementedError

    async def sock_sendall(self, sock, data):
        raise NotImplementedError

    async def sock_sendto(self, sock, data, address):
        raise NotImplementedError

    async def sock_connect(self, sock, address):
        raise NotImplementedError

    async def sock_accept(self, sock):
        raise NotImplementedError

    async def sock_sendfile(self, sock, file, offset=0, count=None,
                            *, fallback=None):
        raise NotImplementedError

    

    def add_signal_handler(self, sig, callback, *args):
        raise NotImplementedError

    def remove_signal_handler(self, sig):
        raise NotImplementedError

    

    def set_task_factory(self, factory):
        raise NotImplementedError

    def get_task_factory(self):
        raise NotImplementedError

    

    def get_exception_handler(self):
        raise NotImplementedError

    def set_exception_handler(self, handler):
        raise NotImplementedError

    def default_exception_handler(self, context):
        raise NotImplementedError

    def call_exception_handler(self, context):
        raise NotImplementedError

    

    def get_debug(self):
        raise NotImplementedError

    def set_debug(self, enabled):
        raise NotImplementedError


class AbstractEventLoopPolicy:
    

    def get_event_loop(self):
        
        raise NotImplementedError

    def set_event_loop(self, loop):
        
        raise NotImplementedError

    def new_event_loop(self):
        
        raise NotImplementedError

    

    def get_child_watcher(self):
        "Get the watcher for child processes."
        raise NotImplementedError

    def set_child_watcher(self, watcher):
        
        raise NotImplementedError


class BaseDefaultEventLoopPolicy(AbstractEventLoopPolicy):
    

    _loop_factory = None

    class _Local(threading.local):
        _loop = None
        _set_called = False

    def __init__(self):
        self._local = self._Local()

    def get_event_loop(self):
        
        if (self._local._loop is None and
                not self._local._set_called and
                threading.current_thread() is threading.main_thread()):
            stacklevel = 2
            try:
                f = sys._getframe(1)
            except AttributeError:
                pass
            else:
                
                
                while f:
                    module = f.f_globals.get('__name__')
                    if not (module == 'asyncio' or module.startswith('asyncio.')):
                        break
                    f = f.f_back
                    stacklevel += 1
            import warnings
            warnings.warn('There is no current event loop',
                          DeprecationWarning, stacklevel=stacklevel)
            self.set_event_loop(self.new_event_loop())

        if self._local._loop is None:
            raise RuntimeError('There is no current event loop in thread %r.'
                               % threading.current_thread().name)

        return self._local._loop

    def set_event_loop(self, loop):
        
        self._local._set_called = True
        if loop is not None and not isinstance(loop, AbstractEventLoop):
            raise TypeError(f"loop must be an instance of AbstractEventLoop or None, not '{type(loop).__name__}'")
        self._local._loop = loop

    def new_event_loop(self):
        
        return self._loop_factory()






_event_loop_policy = None


_lock = threading.Lock()



class _RunningLoop(threading.local):
    loop_pid = (None, None)


_running_loop = _RunningLoop()


def get_running_loop():
    
    
    loop = _get_running_loop()
    if loop is None:
        raise RuntimeError('no running event loop')
    return loop


def _get_running_loop():
    
    
    running_loop, pid = _running_loop.loop_pid
    if running_loop is not None and pid == os.getpid():
        return running_loop


def _set_running_loop(loop):
    
    
    _running_loop.loop_pid = (loop, os.getpid())


def _init_event_loop_policy():
    global _event_loop_policy
    with _lock:
        if _event_loop_policy is None:  
            from . import DefaultEventLoopPolicy
            _event_loop_policy = DefaultEventLoopPolicy()


def get_event_loop_policy():
    
    if _event_loop_policy is None:
        _init_event_loop_policy()
    return _event_loop_policy


def set_event_loop_policy(policy):
    
    global _event_loop_policy
    if policy is not None and not isinstance(policy, AbstractEventLoopPolicy):
        raise TypeError(f"policy must be an instance of AbstractEventLoopPolicy or None, not '{type(policy).__name__}'")
    _event_loop_policy = policy


def get_event_loop():
    
    
    current_loop = _get_running_loop()
    if current_loop is not None:
        return current_loop
    return get_event_loop_policy().get_event_loop()


def set_event_loop(loop):
    
    get_event_loop_policy().set_event_loop(loop)


def new_event_loop():
    
    return get_event_loop_policy().new_event_loop()


def get_child_watcher():
    
    return get_event_loop_policy().get_child_watcher()


def set_child_watcher(watcher):
    
    return get_event_loop_policy().set_child_watcher(watcher)



_py__get_running_loop = _get_running_loop
_py__set_running_loop = _set_running_loop
_py_get_running_loop = get_running_loop
_py_get_event_loop = get_event_loop


try:
    
    
    
    from _asyncio import (_get_running_loop, _set_running_loop,
                          get_running_loop, get_event_loop)
except ImportError:
    pass
else:
    
    _c__get_running_loop = _get_running_loop
    _c__set_running_loop = _set_running_loop
    _c_get_running_loop = get_running_loop
    _c_get_event_loop = get_event_loop


if hasattr(os, 'fork'):
    def on_fork():
        
        if _event_loop_policy is not None:
            _event_loop_policy._local = BaseDefaultEventLoopPolicy._Local()
        _set_running_loop(None)
        signal.set_wakeup_fd(-1)

    os.register_at_fork(after_in_child=on_fork)