| """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 itertools |
| import logging |
| import os |
| import socket |
| import subprocess |
| import threading |
| import time |
| import traceback |
| import sys |
| import warnings |
| import weakref |
| |
| from . import compat |
| from . import coroutines |
| from . import events |
| from . import futures |
| from . import tasks |
| from .coroutines import coroutine |
| from .log import logger |
| |
| |
| __all__ = ['BaseEventLoop'] |
| |
| |
| # 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 |
| |
| # Exceptions which must not call the exception handler in fatal error |
| # methods (_fatal_error()) |
| _FATAL_ERROR_IGNORE = (BrokenPipeError, |
| ConnectionResetError, ConnectionAbortedError) |
| |
| |
| def _format_handle(handle): |
| cb = handle._callback |
| if isinstance(getattr(cb, '__self__', None), 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) |
| |
| |
| def _set_reuseport(sock): |
| if not hasattr(socket, 'SO_REUSEPORT'): |
| raise ValueError('reuse_port not supported by socket module') |
| else: |
| try: |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) |
| except OSError: |
| raise ValueError('reuse_port not supported by socket module, ' |
| 'SO_REUSEPORT defined but not implemented.') |
| |
| |
| def _is_stream_socket(sock): |
| # 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`. |
| return (sock.type & socket.SOCK_STREAM) == socket.SOCK_STREAM |
| |
| |
| def _is_dgram_socket(sock): |
| # 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`. |
| return (sock.type & socket.SOCK_DGRAM) == socket.SOCK_DGRAM |
| |
| |
| def _ipaddr_info(host, port, family, type, proto): |
| # 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. |
| return None |
| |
| |
| def _ensure_resolved(address, *, family=0, type=socket.SOCK_STREAM, proto=0, |
| 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, |
| proto=proto, flags=flags) |
| |
| |
| def _run_until_complete_cb(fut): |
| exc = fut._exception |
| if (isinstance(exc, BaseException) |
| and not isinstance(exc, Exception)): |
| # Issue #22429: run_forever() already finished, no need to |
| # stop it. |
| return |
| fut._loop.stop() |
| |
| |
| 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 = self._loop.create_future() |
| self._waiters.append(waiter) |
| yield from waiter |
| |
| |
| class BaseEventLoop(events.AbstractEventLoop): |
| |
| def __init__(self): |
| self._timer_cancelled_count = 0 |
| self._closed = False |
| self._stopping = False |
| self._ready = collections.deque() |
| self._scheduled = [] |
| self._default_executor = None |
| self._internal_fds = 0 |
| # 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. |
| |
| Return a task 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 None the default task factory will be set. |
| |
| If factory is a callable, it should have a signature matching |
| '(loop, coro)', where 'loop' will be a reference to the active |
| event loop, 'coro' will be a coroutine object. The callable |
| must return a Future. |
| """ |
| 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. |
| |
| 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 _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. |
| |
| 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 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. |
| |
| Every callback already scheduled will still run. This simply informs |
| run_forever to stop looping after a complete iteration. |
| """ |
| self._stopping = True |
| |
| 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.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, |
| 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. |
| |
| 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. |
| """ |
| 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. |
| |
| 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. |
| """ |
| 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. |
| |
| 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. |
| """ |
| 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. |
| |
| 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 = _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. |
| |
| The host parameter can be a string, in that case the TCP server is bound |
| to host and port. |
| |
| The host parameter can also be a sequence of strings and in that case |
| the TCP server is bound to all hosts of the sequence. If a host |
| appears multiple times (possibly indirectly e.g. when hostnames |
| resolve to the same IP address), the server is only bound once to that |
| host. |
| |
| Return a Server object which can be used to stop the service. |
| |
| This method is a coroutine. |
| """ |
| 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 == '': |
| hosts = [None] |
| elif (isinstance(host, str) or |
| not isinstance(host, collections.Iterable)): |
| hosts = [host] |
| else: |
| hosts = host |
| |
| fs = [self._create_server_getaddrinfo(host, port, family=family, |
| flags=flags) |
| for host in hosts] |
| infos = yield from tasks.gather(*fs, loop=self) |
| infos = set(itertools.chain.from_iterable(infos)) |
| |
| completed = False |
| try: |
| for res in infos: |
| af, socktype, proto, canonname, sa = res |
| try: |
| sock = socket.socket(af, socktype, proto) |
| except socket.error: |
| # 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) |
| if reuse_port: |
| _set_reuseport(sock) |
| # 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') |
| if not _is_stream_socket(sock): |
| raise ValueError( |
| 'A Stream Socket was expected, got {!r}'.format(sock)) |
| sockets = [sock] |
| |
| server = Server(self, sockets) |
| for sock in sockets: |
| sock.listen(backlog) |
| sock.setblocking(False) |
| self._start_serving(protocol_factory, sock, ssl, server, backlog) |
| if self._debug: |
| logger.info("%r is serving", server) |
| return server |
| |
| @coroutine |
| def connect_accepted_socket(self, protocol_factory, sock, *, ssl=None): |
| """Handle an accepted connection. |
| |
| This is used by servers that accept connections outside of |
| asyncio but that use asyncio to handle connections. |
| |
| This method is a coroutine. When completed, the coroutine |
| returns a (transport, protocol) pair. |
| """ |
| if not _is_stream_socket(sock): |
| raise ValueError( |
| 'A Stream Socket was expected, got {!r}'.format(sock)) |
| |
| transport, protocol = yield from self._create_connection_transport( |
| sock, protocol_factory, ssl, '', server_side=True) |
| 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 handled: (%r, %r)", sock, transport, protocol) |
| return transport, protocol |
| |
| @coroutine |
| def connect_read_pipe(self, protocol_factory, pipe): |
| protocol = protocol_factory() |
| waiter = self.create_future() |
| transport = self._make_read_pipe_transport(pipe, protocol, waiter) |
| |
| try: |
| yield from waiter |
| except: |
| transport.close() |
| raise |
| |
| 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 = self.create_future() |
| transport = self._make_write_pipe_transport(pipe, protocol, waiter) |
| |
| try: |
| yield from waiter |
| except: |
| transport.close() |
| raise |
| |
| if self._debug: |
| logger.debug('Write pipe %r connected: (%r, %r)', |
| pipe.fileno(), transport, protocol) |
| return transport, protocol |
| |
| def _log_subprocess(self, msg, stdin, stdout, stderr): |
| info = [msg] |
| if stdin is not None: |
| info.append('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 get_exception_handler(self): |
| """Return an exception handler, or None if the default one is in use. |
| """ |
| return self._exception_handler |
| |
| 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 |
| |
| if ('source_traceback' not in context |
| and self._current_handle is not None |
| and self._current_handle._source_traceback): |
| context['handle_traceback'] = self._current_handle._source_traceback |
| |
| log_lines = [message] |
| for key in sorted(context): |
| if key in {'message', 'exception'}: |
| continue |
| value = context[key] |
| if key == 'source_traceback': |
| tb = ''.join(traceback.format_list(value)) |
| value = 'Object created at (most recent call last):\n' |
| value += tb.rstrip() |
| elif key == 'handle_traceback': |
| tb = ''.join(traceback.format_list(value)) |
| value = 'Handle created at (most recent call last):\n' |
| value += tb.rstrip() |
| else: |
| value = repr(value) |
| log_lines.append('{}: {}'.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; |
| - 'asyncgen' (optional): Asynchronous generator that caused |
| the exception. |
| |
| New keys maybe introduced in the future. |
| |
| Note: do not overload this method in an event loop subclass. |
| For custom exception handling, use the |
| `set_exception_handler()` method. |
| """ |
| 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 or self._stopping: |
| 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: |
| try: |
| self._current_handle = handle |
| t0 = self.time() |
| handle._run() |
| dt = self.time() - t0 |
| if dt >= self.slow_callback_duration: |
| logger.warning('Executing %s took %.3f seconds', |
| _format_handle(handle), dt) |
| finally: |
| self._current_handle = None |
| else: |
| handle._run() |
| handle = None # Needed to break cycles when an exception occurs. |
| |
| def _set_coroutine_wrapper(self, enabled): |
| try: |
| set_wrapper = sys.set_coroutine_wrapper |
| get_wrapper = sys.get_coroutine_wrapper |
| except AttributeError: |
| return |
| |
| enabled = bool(enabled) |
| if self._coroutine_wrapper_set == enabled: |
| return |
| |
| wrapper = coroutines.debug_wrapper |
| current_wrapper = get_wrapper() |
| |
| if enabled: |
| if current_wrapper not in (None, wrapper): |
| warnings.warn( |
| "loop.set_debug(True): cannot set debug coroutine " |
| "wrapper; another wrapper is already set %r" % |
| current_wrapper, RuntimeWarning) |
| else: |
| set_wrapper(wrapper) |
| self._coroutine_wrapper_set = True |
| else: |
| if current_wrapper not in (None, wrapper): |
| warnings.warn( |
| "loop.set_debug(False): cannot unset debug coroutine " |
| "wrapper; another wrapper was set %r" % |
| current_wrapper, RuntimeWarning) |
| else: |
| set_wrapper(None) |
| self._coroutine_wrapper_set = False |
| |
| def get_debug(self): |
| return self._debug |
| |
| def set_debug(self, enabled): |
| self._debug = enabled |
| |
| if self.is_running(): |
| self._set_coroutine_wrapper(enabled) |