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