1097 lines
41 KiB
Python
1097 lines
41 KiB
Python
"""Base implementation of event loop.
|
|
|
|
The event loop can be broken up into a multiplexer (the part
|
|
responsible for notifying us of I/O 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 if/when 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.
|
|
"""
|
|
|
|
|
|
import collections
|
|
import concurrent.futures
|
|
import heapq
|
|
import inspect
|
|
import logging
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import time
|
|
import traceback
|
|
import sys
|
|
|
|
from . import coroutines
|
|
from . import events
|
|
from . import futures
|
|
from . import tasks
|
|
from .coroutines import coroutine
|
|
from .log import logger
|
|
|
|
|
|
__all__ = ['BaseEventLoop', 'Server']
|
|
|
|
|
|
# Argument for default thread pool executor creation.
|
|
_MAX_WORKERS = 5
|
|
|
|
# Minimum number of _scheduled timer handles before cleanup of
|
|
# cancelled handles is performed.
|
|
_MIN_SCHEDULED_TIMER_HANDLES = 100
|
|
|
|
# Minimum fraction of _scheduled timer handles that are cancelled
|
|
# before cleanup of cancelled handles is performed.
|
|
_MIN_CANCELLED_TIMER_HANDLES_FRACTION = 0.5
|
|
|
|
def _format_handle(handle):
|
|
cb = handle._callback
|
|
if inspect.ismethod(cb) and isinstance(cb.__self__, tasks.Task):
|
|
# format the 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)
|
|
|
|
|
|
class _StopError(BaseException):
|
|
"""Raised to stop the event loop."""
|
|
|
|
|
|
def _check_resolved_address(sock, address):
|
|
# Ensure that the address is already resolved to avoid the trap of hanging
|
|
# the entire event loop when the address requires doing a DNS lookup.
|
|
family = sock.family
|
|
if family == socket.AF_INET:
|
|
host, port = address
|
|
elif family == socket.AF_INET6:
|
|
host, port = address[:2]
|
|
else:
|
|
return
|
|
|
|
type_mask = 0
|
|
if hasattr(socket, 'SOCK_NONBLOCK'):
|
|
type_mask |= socket.SOCK_NONBLOCK
|
|
if hasattr(socket, 'SOCK_CLOEXEC'):
|
|
type_mask |= socket.SOCK_CLOEXEC
|
|
# Use getaddrinfo(flags=AI_NUMERICHOST) to ensure that the address is
|
|
# already resolved.
|
|
try:
|
|
socket.getaddrinfo(host, port,
|
|
family=family,
|
|
type=(sock.type & ~type_mask),
|
|
proto=sock.proto,
|
|
flags=socket.AI_NUMERICHOST)
|
|
except socket.gaierror as err:
|
|
raise ValueError("address must be resolved (IP address), got %r: %s"
|
|
% (address, err))
|
|
|
|
def _raise_stop_error(*args):
|
|
raise _StopError
|
|
|
|
|
|
class Server(events.AbstractServer):
|
|
|
|
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 = futures.Future(loop=self._loop)
|
|
self._waiters.append(waiter)
|
|
yield from waiter
|
|
|
|
|
|
class BaseEventLoop(events.AbstractEventLoop):
|
|
|
|
def __init__(self):
|
|
self._timer_cancelled_count = 0
|
|
self._closed = False
|
|
self._ready = collections.deque()
|
|
self._scheduled = []
|
|
self._default_executor = None
|
|
self._internal_fds = 0
|
|
self._running = False
|
|
self._clock_resolution = time.get_clock_info('monotonic').resolution
|
|
self._exception_handler = None
|
|
self._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
|
|
|
|
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_task(self, coro):
|
|
"""Schedule a coroutine object.
|
|
|
|
Return a task object.
|
|
"""
|
|
task = tasks.Task(coro, loop=self)
|
|
if task._source_traceback:
|
|
del task._source_traceback[-1]
|
|
return task
|
|
|
|
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, *,
|
|
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.
|
|
|
|
This may be called from a different thread.
|
|
|
|
The subclass is responsible for implementing the self-pipe.
|
|
"""
|
|
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 run_forever(self):
|
|
"""Run until stop() is called."""
|
|
self._check_closed()
|
|
if self._running:
|
|
raise RuntimeError('Event loop is running.')
|
|
self._running = True
|
|
try:
|
|
while True:
|
|
try:
|
|
self._run_once()
|
|
except _StopError:
|
|
break
|
|
finally:
|
|
self._running = False
|
|
|
|
def run_until_complete(self, future):
|
|
"""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.
|
|
"""
|
|
self._check_closed()
|
|
|
|
new_task = not isinstance(future, futures.Future)
|
|
future = tasks.async(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(_raise_stop_error)
|
|
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(_raise_stop_error)
|
|
if not future.done():
|
|
raise RuntimeError('Event loop stopped before Future completed.')
|
|
|
|
return future.result()
|
|
|
|
def stop(self):
|
|
"""Stop running the event loop.
|
|
|
|
Every callback scheduled before stop() is called will run. Callbacks
|
|
scheduled after stop() is called will not run. However, those callbacks
|
|
will run if run_forever is called again later.
|
|
"""
|
|
self.call_soon(_raise_stop_error)
|
|
|
|
def close(self):
|
|
"""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.
|
|
"""
|
|
if self._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
|
|
|
|
def is_running(self):
|
|
"""Returns True if the event loop is running."""
|
|
return self._running
|
|
|
|
def time(self):
|
|
"""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.
|
|
"""
|
|
return time.monotonic()
|
|
|
|
def call_later(self, delay, callback, *args):
|
|
"""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 undefined which
|
|
will be called first.
|
|
|
|
Any positional arguments after the callback will be passed to
|
|
the callback when it is called.
|
|
"""
|
|
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.
|
|
|
|
Absolute time corresponds to the event loop's time() method.
|
|
"""
|
|
if coroutines.iscoroutinefunction(callback):
|
|
raise TypeError("coroutines cannot be used with call_at()")
|
|
if self._debug:
|
|
self._assert_is_current_event_loop()
|
|
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.
|
|
|
|
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.
|
|
"""
|
|
handle = self._call_soon(callback, args, check_loop=True)
|
|
if handle._source_traceback:
|
|
del handle._source_traceback[-1]
|
|
return handle
|
|
|
|
def _call_soon(self, callback, args, check_loop):
|
|
if coroutines.iscoroutinefunction(callback):
|
|
raise TypeError("coroutines cannot be used with call_soon()")
|
|
if self._debug and check_loop:
|
|
self._assert_is_current_event_loop()
|
|
handle = events.Handle(callback, args, self)
|
|
if handle._source_traceback:
|
|
del handle._source_traceback[-1]
|
|
self._ready.append(handle)
|
|
return handle
|
|
|
|
def _assert_is_current_event_loop(self):
|
|
"""Asserts that this event loop is the current 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.
|
|
"""
|
|
try:
|
|
current = events.get_event_loop()
|
|
except AssertionError:
|
|
return
|
|
if current is not self:
|
|
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."""
|
|
handle = self._call_soon(callback, args, check_loop=False)
|
|
if handle._source_traceback:
|
|
del handle._source_traceback[-1]
|
|
self._write_to_self()
|
|
return handle
|
|
|
|
def run_in_executor(self, executor, callback, *args):
|
|
if coroutines.iscoroutinefunction(callback):
|
|
raise TypeError("Coroutines cannot be used with run_in_executor()")
|
|
if isinstance(callback, events.Handle):
|
|
assert not args
|
|
assert not isinstance(callback, events.TimerHandle)
|
|
if callback._cancelled:
|
|
f = futures.Future(loop=self)
|
|
f.set_result(None)
|
|
return f
|
|
callback, args = callback._callback, callback._args
|
|
if executor is None:
|
|
executor = self._default_executor
|
|
if executor is None:
|
|
executor = concurrent.futures.ThreadPoolExecutor(_MAX_WORKERS)
|
|
self._default_executor = executor
|
|
return futures.wrap_future(executor.submit(callback, *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.
|
|
|
|
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.
|
|
"""
|
|
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 = self.getaddrinfo(
|
|
host, port, family=family,
|
|
type=socket.SOCK_STREAM, proto=proto, flags=flags)
|
|
fs = [f1]
|
|
if local_addr is not None:
|
|
f2 = self.getaddrinfo(
|
|
*local_addr, family=family,
|
|
type=socket.SOCK_STREAM, proto=proto, flags=flags)
|
|
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)))
|
|
|
|
elif sock is None:
|
|
raise ValueError(
|
|
'host and port was not specified and no sock specified')
|
|
|
|
sock.setblocking(False)
|
|
|
|
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):
|
|
protocol = protocol_factory()
|
|
waiter = futures.Future(loop=self)
|
|
if ssl:
|
|
sslcontext = None if isinstance(ssl, bool) else ssl
|
|
transport = self._make_ssl_transport(
|
|
sock, protocol, sslcontext, waiter,
|
|
server_side=False, server_hostname=server_hostname)
|
|
else:
|
|
transport = self._make_socket_transport(sock, protocol, waiter)
|
|
|
|
yield from waiter
|
|
return transport, protocol
|
|
|
|
@coroutine
|
|
def create_datagram_endpoint(self, protocol_factory,
|
|
local_addr=None, remote_addr=None, *,
|
|
family=0, proto=0, flags=0):
|
|
"""Create datagram connection."""
|
|
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 self.getaddrinfo(
|
|
*addr, family=family, type=socket.SOCK_DGRAM,
|
|
proto=proto, flags=flags)
|
|
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 = []
|
|
|
|
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)
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 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 = futures.Future(loop=self)
|
|
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)
|
|
yield from waiter
|
|
return transport, protocol
|
|
|
|
@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):
|
|
"""Create a TCP server bound to host and port.
|
|
|
|
Return a Server object which can be used to stop the service.
|
|
|
|
This method is a coroutine.
|
|
"""
|
|
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 = []
|
|
if host == '':
|
|
host = None
|
|
|
|
infos = yield from self.getaddrinfo(
|
|
host, port, family=family,
|
|
type=socket.SOCK_STREAM, proto=0, flags=flags)
|
|
if not infos:
|
|
raise OSError('getaddrinfo() returned empty list')
|
|
|
|
completed = False
|
|
try:
|
|
for res in infos:
|
|
af, socktype, proto, canonname, sa = res
|
|
try:
|
|
sock = socket.socket(af, socktype, proto)
|
|
except socket.error:
|
|
# Assume it's a bad family/type/protocol combination.
|
|
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)
|
|
# Disable IPv4/IPv6 dual stack support (enabled by
|
|
# default on Linux) which makes a single socket
|
|
# listen on both address families.
|
|
if af == 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()))
|
|
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')
|
|
sockets = [sock]
|
|
|
|
server = Server(self, sockets)
|
|
for sock in sockets:
|
|
sock.listen(backlog)
|
|
sock.setblocking(False)
|
|
self._start_serving(protocol_factory, sock, ssl, server)
|
|
if self._debug:
|
|
logger.info("%r is serving", server)
|
|
return server
|
|
|
|
@coroutine
|
|
def connect_read_pipe(self, protocol_factory, pipe):
|
|
protocol = protocol_factory()
|
|
waiter = futures.Future(loop=self)
|
|
transport = self._make_read_pipe_transport(pipe, protocol, waiter)
|
|
yield from waiter
|
|
if self._debug:
|
|
logger.debug('Read pipe %r connected: (%r, %r)',
|
|
pipe.fileno(), transport, protocol)
|
|
return transport, protocol
|
|
|
|
@coroutine
|
|
def connect_write_pipe(self, protocol_factory, pipe):
|
|
protocol = protocol_factory()
|
|
waiter = futures.Future(loop=self)
|
|
transport = self._make_write_pipe_transport(pipe, protocol, waiter)
|
|
yield from waiter
|
|
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('stdin=%s' % _format_pipe(stdin))
|
|
if stdout is not None and stderr == subprocess.STDOUT:
|
|
info.append('stdout=stderr=%s' % _format_pipe(stdout))
|
|
else:
|
|
if stdout is not None:
|
|
info.append('stdout=%s' % _format_pipe(stdout))
|
|
if stderr is not None:
|
|
info.append('stderr=%s' % _format_pipe(stderr))
|
|
logger.debug(' '.join(info))
|
|
|
|
@coroutine
|
|
def subprocess_shell(self, protocol_factory, cmd, *, stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
universal_newlines=False, shell=True, bufsize=0,
|
|
**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")
|
|
protocol = protocol_factory()
|
|
if self._debug:
|
|
# don't log parameters: they may contain sensitive information
|
|
# (password) and may be too long
|
|
debug_log = 'run shell command %r' % cmd
|
|
self._log_subprocess(debug_log, stdin, stdout, stderr)
|
|
transport = yield from self._make_subprocess_transport(
|
|
protocol, cmd, True, stdin, stdout, stderr, bufsize, **kwargs)
|
|
if self._debug:
|
|
logger.info('%s: %r' % (debug_log, transport))
|
|
return transport, protocol
|
|
|
|
@coroutine
|
|
def subprocess_exec(self, protocol_factory, program, *args,
|
|
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE, universal_newlines=False,
|
|
shell=False, bufsize=0, **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")
|
|
popen_args = (program,) + args
|
|
for arg in popen_args:
|
|
if not isinstance(arg, (str, bytes)):
|
|
raise TypeError("program arguments must be "
|
|
"a bytes or text string, not %s"
|
|
% type(arg).__name__)
|
|
protocol = protocol_factory()
|
|
if self._debug:
|
|
# don't log parameters: they may contain sensitive information
|
|
# (password) and may be too long
|
|
debug_log = 'execute program %r' % program
|
|
self._log_subprocess(debug_log, stdin, stdout, stderr)
|
|
transport = yield from self._make_subprocess_transport(
|
|
protocol, popen_args, False, stdin, stdout, stderr,
|
|
bufsize, **kwargs)
|
|
if self._debug:
|
|
logger.info('%s: %r' % (debug_log, transport))
|
|
return transport, protocol
|
|
|
|
def set_exception_handler(self, handler):
|
|
"""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).
|
|
"""
|
|
if handler is not None and not callable(handler):
|
|
raise TypeError('A callable object or None is expected, '
|
|
'got {!r}'.format(handler))
|
|
self._exception_handler = handler
|
|
|
|
def default_exception_handler(self, 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.
|
|
|
|
The context parameter has the same meaning as in
|
|
`call_exception_handler()`.
|
|
"""
|
|
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
|
|
|
|
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()
|
|
else:
|
|
value = repr(value)
|
|
log_lines.append('{}: {}'.format(key, value))
|
|
|
|
logger.error('\n'.join(log_lines), exc_info=exc_info)
|
|
|
|
def call_exception_handler(self, context):
|
|
"""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;
|
|
- 'handle' (optional): Handle instance;
|
|
- 'protocol' (optional): Protocol instance;
|
|
- 'transport' (optional): Transport instance;
|
|
- 'socket' (optional): Socket instance.
|
|
|
|
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.
|
|
"""
|
|
if self._exception_handler is None:
|
|
try:
|
|
self.default_exception_handler(context)
|
|
except Exception:
|
|
# Second protection layer for unexpected errors
|
|
# in the default implementation, as well as for subclassed
|
|
# event loops with overloaded "default_exception_handler".
|
|
logger.error('Exception in default exception handler',
|
|
exc_info=True)
|
|
else:
|
|
try:
|
|
self._exception_handler(self, context)
|
|
except Exception as exc:
|
|
# Exception in the user set custom exception handler.
|
|
try:
|
|
# Let's try default handler.
|
|
self.default_exception_handler({
|
|
'message': 'Unhandled error in exception handler',
|
|
'exception': exc,
|
|
'context': context,
|
|
})
|
|
except Exception:
|
|
# Guard 'default_exception_handler' in case it is
|
|
# overloaded.
|
|
logger.error('Exception in default exception handler '
|
|
'while handling an unexpected error '
|
|
'in custom exception handler',
|
|
exc_info=True)
|
|
|
|
def _add_callback(self, handle):
|
|
"""Add a Handle to _scheduled (TimerHandle) or _ready."""
|
|
assert isinstance(handle, events.Handle), 'A Handle is required here'
|
|
if handle._cancelled:
|
|
return
|
|
assert not isinstance(handle, events.TimerHandle)
|
|
self._ready.append(handle)
|
|
|
|
def _add_callback_signalsafe(self, handle):
|
|
"""Like _add_callback() but called from a signal handler."""
|
|
self._add_callback(handle)
|
|
self._write_to_self()
|
|
|
|
def _timer_handle_cancelled(self, handle):
|
|
"""Notification that a TimerHandle has been cancelled."""
|
|
if handle._scheduled:
|
|
self._timer_cancelled_count += 1
|
|
|
|
def _run_once(self):
|
|
"""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.
|
|
"""
|
|
|
|
sched_count = len(self._scheduled)
|
|
if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and
|
|
self._timer_cancelled_count / sched_count >
|
|
_MIN_CANCELLED_TIMER_HANDLES_FRACTION):
|
|
# Remove delayed calls that were cancelled if their number
|
|
# is too high
|
|
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:
|
|
# Remove delayed calls that were cancelled from head of queue.
|
|
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:
|
|
timeout = 0
|
|
elif self._scheduled:
|
|
# Compute the desired timeout.
|
|
when = self._scheduled[0]._when
|
|
timeout = max(0, when - self.time())
|
|
|
|
if self._debug and timeout != 0:
|
|
t0 = self.time()
|
|
event_list = self._selector.select(timeout)
|
|
dt = self.time() - t0
|
|
if dt >= 1.0:
|
|
level = logging.INFO
|
|
else:
|
|
level = logging.DEBUG
|
|
nevent = len(event_list)
|
|
if timeout is None:
|
|
logger.log(level, 'poll took %.3f ms: %s events',
|
|
dt * 1e3, nevent)
|
|
elif nevent:
|
|
logger.log(level,
|
|
'poll %.3f ms took %.3f ms: %s events',
|
|
timeout * 1e3, dt * 1e3, nevent)
|
|
elif dt >= 1.0:
|
|
logger.log(level,
|
|
'poll %.3f ms took %.3f ms: timeout',
|
|
timeout * 1e3, dt * 1e3)
|
|
else:
|
|
event_list = self._selector.select(timeout)
|
|
self._process_events(event_list)
|
|
|
|
# Handle 'later' callbacks that are ready.
|
|
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)
|
|
|
|
# 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.
|
|
ntodo = len(self._ready)
|
|
for i in range(ntodo):
|
|
handle = self._ready.popleft()
|
|
if handle._cancelled:
|
|
continue
|
|
if self._debug:
|
|
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)
|
|
else:
|
|
handle._run()
|
|
handle = None # Needed to break cycles when an exception occurs.
|
|
|
|
def get_debug(self):
|
|
return self._debug
|
|
|
|
def set_debug(self, enabled):
|
|
self._debug = enabled
|