cb = handle._callback

if isinstance(getattr(cb, '__self__', None), tasks.Task):

# format the task

return repr(cb.__self__)

else:

if fd == subprocess.PIPE:

return '<pipe>'

elif fd == subprocess.STDOUT:

return '<stdout>'

else:

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, '

# 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 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:

# If port's a service name like "http", don't skip getaddrinfo.

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:

# Linux's inet_pton doesn't accept an IPv6 zone index after host,

# like '::1%lo0'.

return None

for af in afs:

try:

socket.inet_pton(af, host)

# The host has already been resolved.

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

# "host" is not an IP address.

"""Interleave list of addrinfo tuples by family."""

# Group addresses by family

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)

if not fut.cancelled():

exc = fut.exception()

if isinstance(exc, (SystemExit, KeyboardInterrupt)):

# Issue #22429: run_forever() already finished, no need to

# stop it.

return

if (sock.family in {socket.AF_INET, socket.AF_INET6} and

sock.type == socket.SOCK_STREAM and

sock.proto == socket.IPPROTO_TCP):

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:

# Never happens if peer disconnects after sending the whole content

# Thus disconnection is always an exception from user perspective

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:

# Cancel the future.

# Basically it has no effect because protocol is switched back,

# no code should wait for it anymore.

self._write_ready_fut.cancel()

if self._should_resume_writing:

def __init__(self, loop, sockets, protocol_factory, ssl_context, backlog,

ssl_handshake_timeout, ssl_shutdown_timeout=None):

self._loop = loop

self._sockets = sockets

# Weak references so we don't break Transport's ability to

# detect abandoned transports

self._clients = weakref.WeakSet()

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, transport):

assert self._sockets is not None

self._clients.add(transport)

def _detach(self, transport):

self._clients.discard(transport)

if len(self._clients) == 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 len(self._clients) == 0:

self._wakeup()

def close_clients(self):

for transport in self._clients.copy():

transport.close()

def abort_clients(self):

for transport in self._clients.copy():

transport.abort()

async def start_serving(self):

self._start_serving()

# Skip one loop iteration so that all 'loop.add_reader'

# go through.

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

"""Wait until server is closed and all connections are dropped.

# 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.create_future()

self._waiters.append(waiter)

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

# Identifier of the thread running the event loop, or None if the

# event loop is not running

self._thread_id = None

self._clock_resolution = time.get_clock_info('monotonic').resolution

self._exception_handler = None

self.set_debug(coroutines._is_debug_mode())

# The preserved state of async generator hooks.

self._old_agen_hooks = None

# 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.

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

# A weak set of all asynchronous generators that are

# being iterated by the loop.

self._asyncgens = weakref.WeakSet()

# Set to True when `loop.shutdown_asyncgens` is called.

self._asyncgens_shutdown_called = False

# Set to True when `loop.shutdown_default_executor` is called.

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

"""Create a Future object attached to the loop."""

return futures.Future(loop=self)

def create_task(self, coro, **kwargs):

"""Schedule or begin executing a coroutine object.

self._check_closed()

if self._task_factory is not None:

return self._task_factory(self, coro, **kwargs)

task = tasks.Task(coro, loop=self, **kwargs)

if task._source_traceback:

del task._source_traceback[-1]

try:

return task

finally:

# gh-128552: prevent a refcycle of

# task.exception().__traceback__->BaseEventLoop.create_task->task

del task

def set_task_factory(self, factory):

"""Set a task factory that will be used by loop.create_task().

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 a task factory, or None if the default one is in use."""

return self._task_factory

def _make_socket_transport(self, sock, protocol, waiter=None, *,

extra=None, server=None):

"""Create socket transport."""

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

"""Create SSL transport."""

raise NotImplementedError

def _make_datagram_transport(self, sock, protocol,

address=None, waiter=None, extra=None):

"""Create datagram transport."""

raise NotImplementedError

def _make_read_pipe_transport(self, pipe, protocol, waiter=None,

extra=None):

"""Create read pipe transport."""

raise NotImplementedError

def _make_write_pipe_transport(self, pipe, protocol, waiter=None,

extra=None):

"""Create write pipe transport."""

raise NotImplementedError

async def _make_subprocess_transport(self, protocol, args, shell,

stdin, stdout, stderr, bufsize,

extra=None, **kwargs):

"""Create subprocess transport."""

raise NotImplementedError

def _write_to_self(self):

"""Write a byte to self-pipe, to wake up the event loop.

raise NotImplementedError

def _process_events(self, event_list):

"""Process selector events."""

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

"""Shutdown all active asynchronous generators."""

self._asyncgens_shutdown_called = True

if not len(self._asyncgens):

# If Python version is <3.6 or we don't have any asynchronous

# generators alive.

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

"""Schedule the shutdown of the default executor.

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:

async with timeouts.timeout(timeout):

await future

except TimeoutError:

warnings.warn("The executor did not finishing joining "

f"its threads within {timeout} seconds.",

RuntimeWarning, stacklevel=2)

self._default_executor.shutdown(wait=False)

else:

thread.join()

def _do_shutdown(self, future):

try:

self._default_executor.shutdown(wait=True)

if not self.is_closed():

self.call_soon_threadsafe(futures._set_result_unless_cancelled,

future, None)

except Exception as ex:

if not self.is_closed() and not future.cancelled():

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

"""Prepare the run loop to process events.

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

"""Clean up after an event loop finishes the looping over events.

self._stopping = False

self._thread_id = None

events._set_running_loop(None)

self._set_coroutine_origin_tracking(False)

# Restore any pre-existing async generator hooks.

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

"""Run until stop() is called."""

self._run_forever_setup()

try:

while True:

self._run_once()

if self._stopping:

break

finally:

self._run_forever_cleanup()

def run_until_complete(self, future):

"""Run until the Future is done.

self._check_closed()

self._check_running()

new_task = not futures.isfuture(future)

future = tasks.ensure_future(future, loop=self)

if new_task:

# An exception is raised if the future didn't complete, so there

# is no need to log the "destroy pending task" message

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():

# The coroutine raised a BaseException. Consume the exception

# to not log a warning, the caller doesn't have access to the

# local task.

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

"""Stop running the event loop.

self._stopping = True

def close(self):

"""Close the event loop.

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

"""Returns True if the event loop was closed."""

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

"""Returns True if the event loop is running."""

return (self._thread_id is not None)

def time(self):

"""Return the time according to the event loop's clock.

return time.monotonic()

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

"""Arrange for a callback to be called at a given time.

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

"""Like call_later(), but uses an absolute time.

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

"""Arrange for a callback to be called as soon as possible.

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

"""Check that the current thread is the thread running the event loop.

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

"""Like call_soon(), but thread-safe."""

self._check_closed()

if self._debug:

self._check_callback(callback, 'call_soon_threadsafe')

handle = events._ThreadSafeHandle(callback, args, self, context)

self._ready.append(handle)

if handle._source_traceback:

del handle._source_traceback[-1]

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

# Only check when the default executor is being used

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

# NB: sendfile syscall is not supported for SSL sockets and

# non-mmap files even if sendfile is supported by OS

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 # EOF

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: