cb = handle._callback

if inspect.ismethod(cb) and isinstance(cb.__self__, 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, '

# Linux's socket.type is a bitmask that can include extra info

# about socket, therefore we can't do simple

# `sock_type == socket.SOCK_STREAM`.

# Linux's socket.type is a bitmask that can include extra info

# about socket, therefore we can't do simple

# `sock_type == socket.SOCK_DGRAM`.

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

# Linux only:

# getaddrinfo() can raise when socket.type is a bit mask.

# So if socket.type is a bit mask of SOCK_STREAM, and say

# SOCK_NONBLOCK, we simply return None, which will trigger

# a call to getaddrinfo() letting it process this request.

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 hasattr(socket, 'AF_INET6'):

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.

return af, type, proto, '', (host, port)

except OSError:

pass

# "host" is not an IP address.

flags=0, loop):

host, port = address[:2]

info = _ipaddr_info(host, port, family, type, proto)

if info is not None:

# "host" is already a resolved IP.

fut = loop.create_future()

fut.set_result([info])

return fut

else:

return loop.getaddrinfo(host, port, family=family, type=type,

exc = fut._exception

if (isinstance(exc, BaseException)

and not isinstance(exc, Exception)):

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

# stop it.

return

def __init__(self, loop, sockets):

self._loop = loop

self.sockets = sockets

self._active_count = 0

self._waiters = []

def __repr__(self):

return '<%s sockets=%r>' % (self.__class__.__name__, self.sockets)

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

sockets = self.sockets

if sockets is None:

return

self.sockets = None

for sock in sockets:

self._loop._stop_serving(sock)

if self._active_count == 0:

self._wakeup()

def _wakeup(self):

waiters = self._waiters

self._waiters = None

for waiter in waiters:

if not waiter.done():

waiter.set_result(waiter)

@coroutine

def wait_closed(self):

if self.sockets is None or 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((not sys.flags.ignore_environment

and bool(os.environ.get('PYTHONASYNCIODEBUG'))))

# 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_wrapper_set = False

if hasattr(sys, 'get_asyncgen_hooks'):

# Python >= 3.6

# A weak set of all asynchronous generators that are

# being iterated by the loop.

self._asyncgens = weakref.WeakSet()

else:

self._asyncgens = None

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

self._asyncgens_shutdown_called = False

def __repr__(self):

return ('<%s running=%s closed=%s debug=%s>'

% (self.__class__.__name__, self.is_running(),

self.is_closed(), 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):

"""Schedule a coroutine object.

self._check_closed()

if self._task_factory is None:

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

if task._source_traceback:

del task._source_traceback[-1]

else:

task = self._task_factory(self, coro)

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

"""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

@coroutine

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 _asyncgen_finalizer_hook(self, agen):

self._asyncgens.discard(agen)

if not self.is_closed():

self.create_task(agen.aclose())

# Wake up the loop if the finalizer was called from

# a different thread.

self._write_to_self()

def _asyncgen_firstiter_hook(self, agen):

if self._asyncgens_shutdown_called:

warnings.warn(

"asynchronous generator {!r} was scheduled after "

"loop.shutdown_asyncgens() call".format(agen),

ResourceWarning, source=self)

self._asyncgens.add(agen)

@coroutine

def shutdown_asyncgens(self):

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

self._asyncgens_shutdown_called = True

if self._asyncgens is None or 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()

shutdown_coro = tasks.gather(

*[ag.aclose() for ag in closing_agens],

return_exceptions=True,

loop=self)

results = yield from shutdown_coro

for result, agen in zip(results, closing_agens):

if isinstance(result, Exception):

self.call_exception_handler({

'message': 'an error occurred during closing of '

'asynchronous generator {!r}'.format(agen),

'exception': result,

'asyncgen': agen

})

def run_forever(self):

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

self._check_closed()

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

self._set_coroutine_wrapper(self._debug)

self._thread_id = threading.get_ident()

if self._asyncgens is not None:

old_agen_hooks = sys.get_asyncgen_hooks()

sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter_hook,

finalizer=self._asyncgen_finalizer_hook)

try:

events._set_running_loop(self)

while True:

self._run_once()

if self._stopping:

break

finally:

self._stopping = False

self._thread_id = None

events._set_running_loop(None)

self._set_coroutine_wrapper(False)

if self._asyncgens is not None:

sys.set_asyncgen_hooks(*old_agen_hooks)

def run_until_complete(self, future):

"""Run until the Future is done.

self._check_closed()

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

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

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

# On Python 3.3 and older, objects with a destructor part of a reference

# cycle are never destroyed. It's not more the case on Python 3.4 thanks

# to the PEP 442.

if compat.PY34:

def __del__(self):

if not self.is_closed():

warnings.warn("unclosed event loop %r" % self, ResourceWarning)

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

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

timer = self.call_at(self.time() + delay, callback, *args)

if timer._source_traceback:

del timer._source_traceback[-1]

return timer

def call_at(self, when, callback, *args):

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

self._check_closed()

if self._debug:

self._check_thread()

self._check_callback(callback, 'call_at')

timer = events.TimerHandle(when, callback, args, self)

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

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

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(

"coroutines cannot be used with {}()".format(method))

if not callable(callback):

raise TypeError(

'a callable object was expected by {}(), got {!r}'.format(

method, callback))

def _call_soon(self, callback, args):

handle = events.Handle(callback, args, self)

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

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

self._check_closed()

if self._debug:

self._check_callback(callback, 'call_soon_threadsafe')

handle = self._call_soon(callback, args)

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

if executor is None:

executor = concurrent.futures.ThreadPoolExecutor()

self._default_executor = executor

return futures.wrap_future(executor.submit(func, *args), loop=self)

def set_default_executor(self, executor):

self._default_executor = executor

def _getaddrinfo_debug(self, host, port, family, type, proto, flags):

msg = ["%s:%r" % (host, port)]

if family:

msg.append('family=%r' % family)

if type:

msg.append('type=%r' % type)

if proto:

msg.append('proto=%r' % proto)

if flags:

msg.append('flags=%r' % flags)

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 = ('Getting address info %s took %.3f ms: %r'

% (msg, dt * 1e3, addrinfo))

if dt >= self.slow_callback_duration:

logger.info(msg)

else:

logger.debug(msg)

return addrinfo

def getaddrinfo(self, host, port, *,

family=0, type=0, proto=0, flags=0):

if self._debug:

return self.run_in_executor(None, self._getaddrinfo_debug,

host, port, family, type, proto, flags)

else:

return self.run_in_executor(None, socket.getaddrinfo,

host, port, family, type, proto, flags)

def getnameinfo(self, sockaddr, flags=0):

return self.run_in_executor(None, socket.getnameinfo, sockaddr, flags)

@coroutine

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

"""Connect to a TCP server.

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:

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

raise ValueError('You must set server_hostname '

'when using ssl without a host')

server_hostname = host

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

f1 = _ensure_resolved((host, port), family=family,

type=socket.SOCK_STREAM, proto=proto,

flags=flags, loop=self)

fs = [f1]

if local_addr is not None:

f2 = _ensure_resolved(local_addr, family=family,

type=socket.SOCK_STREAM, proto=proto,

flags=flags, loop=self)

fs.append(f2)

else:

f2 = None

yield from tasks.wait(fs, loop=self)

infos = f1.result()

if not infos:

raise OSError('getaddrinfo() returned empty list')

if f2 is not None:

laddr_infos = f2.result()

if not laddr_infos:

raise OSError('getaddrinfo() returned empty list')

exceptions = []

for family, type, proto, cname, address in infos:

try:

sock = socket.socket(family=family, type=type, proto=proto)

sock.setblocking(False)

if f2 is not None:

for _, _, _, _, laddr in laddr_infos:

try:

sock.bind(laddr)

break

except OSError as exc:

exc = OSError(

exc.errno, 'error while '

'attempting to bind on address '

'{!r}: {}'.format(

laddr, exc.strerror.lower()))

exceptions.append(exc)

else:

sock.close()

sock = None

continue

if self._debug:

logger.debug("connect %r to %r", sock, address)

yield from self.sock_connect(sock, 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:

if len(exceptions) == 1:

raise exceptions[0]

else:

# If they all have the same str(), raise one.

model = str(exceptions[0])

if all(str(exc) == model for exc in exceptions):

raise exceptions[0]

# Raise a combined exception so the user can see all

# the various error messages.

raise OSError('Multiple exceptions: {}'.format(

', '.join(str(exc) for exc in exceptions)))

else:

if sock is None:

raise ValueError(

'host and port was not specified and no sock specified')

if not _is_stream_socket(sock):

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

raise ValueError(

'A Stream Socket was expected, got {!r}'.format(sock))

transport, protocol = yield from self._create_connection_transport(

sock, protocol_factory, ssl, server_hostname)

if self._debug:

# Get the socket from the transport because SSL transport closes

# the old socket and creates a new SSL socket

sock = transport.get_extra_info('socket')

logger.debug("%r connected to %s:%r: (%r, %r)",

sock, host, port, transport, protocol)

return transport, protocol

@coroutine

def _create_connection_transport(self, sock, protocol_factory, ssl,

server_hostname, server_side=False):

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)

else:

transport = self._make_socket_transport(sock, protocol, waiter)

try:

yield from waiter

except:

transport.close()

raise

return transport, protocol

@coroutine

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

"""Create datagram connection."""

if sock is not None:

if not _is_dgram_socket(sock):

raise ValueError(

'A UDP Socket was expected, got {!r}'.format(sock))

if (local_addr or remote_addr or

family or proto or flags or

reuse_address or reuse_port or allow_broadcast):

# show the problematic kwargs in exception msg

opts = dict(local_addr=local_addr, remote_addr=remote_addr,

family=family, proto=proto, flags=flags,

reuse_address=reuse_address, reuse_port=reuse_port,

allow_broadcast=allow_broadcast)

problems = ', '.join(

'{}={}'.format(k, v) for k, v in opts.items() if v)

raise ValueError(

'socket modifier keyword arguments can not be used '

'when sock is specified. ({})'.format(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)),)

else:

# join address by (family, protocol)

addr_infos = collections.OrderedDict()

for idx, addr in ((0, local_addr), (1, remote_addr)):

if addr is not None:

assert isinstance(addr, tuple) and len(addr) == 2, (

'2-tuple is expected')

infos = yield from _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

# each addr has to have info for each (family, proto) pair

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 = []

if reuse_address is None:

reuse_address = os.name == 'posix' and sys.platform != 'cygwin'

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

sock.setsockopt(

socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

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:

yield from 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:

yield from waiter

except:

transport.close()

raise

return transport, protocol

@coroutine

def _create_server_getaddrinfo(self, host, port, family, flags):

infos = yield from _ensure_resolved((host, port), family=family,

type=socket.SOCK_STREAM,

flags=flags, loop=self)

if not infos:

raise OSError('getaddrinfo({!r}) returned empty list'.format(host))

return infos

@coroutine

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

"""Create a TCP server.

if isinstance(ssl, bool):

raise TypeError('ssl argument must be an SSLContext or None')

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

AF_INET6 = getattr(socket, 'AF_INET6', 0)

if reuse_address is None:

reuse_address = os.name == 'posix' and sys.platform != 'cygwin'

sockets = []