-"""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 heapq
-import inspect
-import logging
-import os
-import socket
-import subprocess
-import sys
-import traceback
-import warnings
-try:
- from collections import OrderedDict
-except ImportError:
- # Python 2.6: use ordereddict backport
- from ordereddict import OrderedDict
-try:
- from threading import get_ident as _get_thread_ident
-except ImportError:
- # Python 2
- from threading import _get_ident as _get_thread_ident
-
-from . import compat
-from . import coroutines
-from . import events
-from . import futures
-from . import tasks
-from .coroutines import coroutine, From, Return
-from .executor import get_default_executor
-from .log import logger
-from .time_monotonic import time_monotonic, time_monotonic_resolution
-
-
-__all__ = ['BaseEventLoop']
-
-
-# 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.
- #
- # getaddrinfo() is slow (around 10 us per call): this function should only
- # be called in debug mode
- family = sock.family
-
- if family == socket.AF_INET:
- host, port = address
- elif family == socket.AF_INET6:
- host, port = address[:2]
- else:
- return
-
- # On Windows, socket.inet_pton() is only available since Python 3.4
- if hasattr(socket, 'inet_pton'):
- # getaddrinfo() is slow and has known issue: prefer inet_pton()
- # if available
- try:
- socket.inet_pton(family, host)
- except socket.error as exc:
- raise ValueError("address must be resolved (IP address), "
- "got host %r: %s"
- % (host, exc))
- else:
- # Use getaddrinfo(flags=AI_NUMERICHOST) to ensure that the address is
- # already resolved.
- type_mask = 0
- if hasattr(socket, 'SOCK_NONBLOCK'):
- type_mask |= socket.SOCK_NONBLOCK
- if hasattr(socket, 'SOCK_CLOEXEC'):
- type_mask |= socket.SOCK_CLOEXEC
- try:
- socket.getaddrinfo(host, port,
- family,
- (sock.type & ~type_mask),
- sock.proto,
- socket.AI_NUMERICHOST)
- except socket.gaierror as err:
- raise ValueError("address must be resolved (IP address), "
- "got host %r: %s"
- % (host, err))
-
-def _raise_stop_error(*args):
- raise _StopError
-
-
-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
- _raise_stop_error()
-
-
-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:
- raise 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
- # 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_monotonic_resolution
- self._exception_handler = None
- self.set_debug(bool(os.environ.get('TROLLIUSDEBUG')))
- # 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
-
- 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.
- """
- 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 run_forever(self):
- """Run until stop() is called."""
- self._check_closed()
- if self.is_running():
- raise RuntimeError('Event loop is running.')
- self._set_coroutine_wrapper(self._debug)
- self._thread_id = _get_thread_ident()
- try:
- while True:
- try:
- self._run_once()
- except _StopError:
- break
- finally:
- self._thread_id = None
- self._set_coroutine_wrapper(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_CLASSES)
- 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 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.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()
- del self._scheduled[:]
- 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.
-
- 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.iscoroutine(callback)
- or coroutines.iscoroutinefunction(callback)):
- raise TypeError("coroutines cannot be used with call_at()")
- self._check_closed()
- if self._debug:
- self._check_thread()
- 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.
- """
- if self._debug:
- self._check_thread()
- handle = self._call_soon(callback, args)
- if handle._source_traceback:
- del handle._source_traceback[-1]
- return handle
-
- def _call_soon(self, callback, args):
- if (coroutines.iscoroutine(callback)
- or coroutines.iscoroutinefunction(callback)):
- raise TypeError("coroutines cannot be used with call_soon()")
- self._check_closed()
- 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 = _get_thread_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."""
- 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):
- if (coroutines.iscoroutine(func)
- or coroutines.iscoroutinefunction(func)):
- raise TypeError("coroutines cannot be used with run_in_executor()")
- self._check_closed()
- if isinstance(func, events.Handle):
- assert not args
- assert not isinstance(func, events.TimerHandle)
- if func._cancelled:
- f = futures.Future(loop=self)
- f.set_result(None)
- return f
- func, args = func._callback, func._args
- if executor is None:
- executor = self._default_executor
- if executor is None:
- executor = get_default_executor()
- 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 = 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 socket.error('getaddrinfo() returned empty list')
- if f2 is not None:
- laddr_infos = f2.result()
- if not laddr_infos:
- raise socket.error('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 socket.error as exc:
- exc = socket.error(
- exc.errno, 'error while '
- 'attempting to bind on address '
- '{0!r}: {1}'.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 socket.error 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 socket.error('Multiple exceptions: {0}'.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)
- raise 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)
-
- try:
- yield From(waiter)
- except:
- transport.close()
- raise
-
- raise 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 = 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 socket.error('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 socket.error 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)
-
- try:
- yield From(waiter)
- except:
- transport.close()
- raise
-
- raise 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 socket.error('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 socket.error as err:
- raise socket.error(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)
- raise 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)
-
- try:
- yield From(waiter)
- except:
- transport.close()
- raise
-
- if self._debug:
- logger.debug('Read pipe %r connected: (%r, %r)',
- pipe.fileno(), transport, protocol)
- raise 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)
-
- try:
- yield From(waiter)
- except:
- transport.close()
- raise
-
- if self._debug:
- logger.debug('Write pipe %r connected: (%r, %r)',
- pipe.fileno(), transport, protocol)
- raise 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, compat.string_types):
- 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))
- raise Return(transport, protocol)
-
- @coroutine
- def subprocess_exec(self, protocol_factory, program, *args, **kwargs):
- stdin = kwargs.pop('stdin', subprocess.PIPE)
- stdout = kwargs.pop('stdout', subprocess.PIPE)
- stderr = kwargs.pop('stderr', subprocess.PIPE)
- universal_newlines = kwargs.pop('universal_newlines', False)
- shell = kwargs.pop('shell', False)
- bufsize = kwargs.pop('bufsize', 0)
- 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, compat.string_types ):
- 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))
- raise 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 {0!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:
- if hasattr(exception, '__traceback__'):
- # Python 3
- tb = exception.__traceback__
- else:
- # call_exception_handler() is usually called indirectly
- # from an except block. If it's not the case, the traceback
- # is undefined...
- tb = sys.exc_info()[2]
- exc_info = (type(exception), exception, tb)
- 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('{0}: {1}'.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
- float(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:
- 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)