diff options
Diffstat (limited to 'asyncio')
| -rw-r--r-- | asyncio/base_events.py | 54 | ||||
| -rw-r--r-- | asyncio/proactor_events.py | 31 | ||||
| -rw-r--r-- | asyncio/selector_events.py | 89 | ||||
| -rw-r--r-- | asyncio/unix_events.py | 36 | ||||
| -rw-r--r-- | asyncio/windows_events.py | 12 |
5 files changed, 206 insertions, 16 deletions
diff --git a/asyncio/base_events.py b/asyncio/base_events.py index 2f7c124..e5683fd 100644 --- a/asyncio/base_events.py +++ b/asyncio/base_events.py @@ -94,6 +94,9 @@ class Server(events.AbstractServer): 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 @@ -274,6 +277,8 @@ class BaseEventLoop(events.AbstractEventLoop): 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() @@ -400,10 +405,39 @@ class BaseEventLoop(events.AbstractEventLoop): 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 addresss info %s', msg) + + t0 = self.time() + addrinfo = socket.getaddrinfo(host, port, family, type, proto, flags) + dt = self.time() - t0 + + msg = ('Getting addresss 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): - return self.run_in_executor(None, socket.getaddrinfo, - host, port, family, type, proto, flags) + 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) @@ -490,6 +524,8 @@ class BaseEventLoop(events.AbstractEventLoop): 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: @@ -522,6 +558,9 @@ class BaseEventLoop(events.AbstractEventLoop): transport, protocol = yield from self._create_connection_transport( sock, protocol_factory, ssl, server_hostname) + if self._debug: + logger.debug("connected to %s:%r: (%r, %r)", + host, port, transport, protocol) return transport, protocol @coroutine @@ -612,6 +651,15 @@ class BaseEventLoop(events.AbstractEventLoop): 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 @@ -692,6 +740,8 @@ class BaseEventLoop(events.AbstractEventLoop): 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 diff --git a/asyncio/proactor_events.py b/asyncio/proactor_events.py index 5009b0d..d09e9fa 100644 --- a/asyncio/proactor_events.py +++ b/asyncio/proactor_events.py @@ -41,6 +41,23 @@ class _ProactorBasePipeTransport(transports._FlowControlMixin, # wait until protocol.connection_made() has been called self._loop.call_soon(waiter._set_result_unless_cancelled, None) + def __repr__(self): + info = [self.__class__.__name__, 'fd=%s' % self._sock.fileno()] + if self._read_fut is not None: + ov = "pending" if self._read_fut.ov.pending else "completed" + info.append('read=%s' % ov) + if self._write_fut is not None: + if self._write_fut.ov.pending: + info.append("write=pending=%s" % self._pending_write) + else: + info.append("write=completed") + if self._buffer: + bufsize = len(self._buffer) + info.append('write_bufsize=%s' % bufsize) + if self._eof_written: + info.append('EOF written') + return '<%s>' % ' '.join(info) + def _set_extra(self, sock): self._extra['pipe'] = sock @@ -55,7 +72,10 @@ class _ProactorBasePipeTransport(transports._FlowControlMixin, self._read_fut.cancel() def _fatal_error(self, exc, message='Fatal error on pipe transport'): - if not isinstance(exc, (BrokenPipeError, ConnectionResetError)): + if isinstance(exc, (BrokenPipeError, ConnectionResetError)): + if self._loop.get_debug(): + logger.debug("%r: %s", self, message, exc_info=True) + else: self._loop.call_exception_handler({ 'message': message, 'exception': exc, @@ -117,6 +137,8 @@ class _ProactorReadPipeTransport(_ProactorBasePipeTransport, if self._paused: raise RuntimeError('Already paused') self._paused = True + if self._loop.get_debug(): + logger.debug("%r pauses reading", self) def resume_reading(self): if not self._paused: @@ -125,6 +147,8 @@ class _ProactorReadPipeTransport(_ProactorBasePipeTransport, if self._closing: return self._loop.call_soon(self._loop_reading, self._read_fut) + if self._loop.get_debug(): + logger.debug("%r resumes reading", self) def _loop_reading(self, fut=None): if self._paused: @@ -165,6 +189,8 @@ class _ProactorReadPipeTransport(_ProactorBasePipeTransport, if data: self._protocol.data_received(data) elif data is not None: + if self._loop.get_debug(): + logger.debug("%r received EOF", self) keep_open = self._protocol.eof_received() if not keep_open: self.close() @@ -427,6 +453,9 @@ class BaseProactorEventLoop(base_events.BaseEventLoop): try: if f is not None: conn, addr = f.result() + if self._debug: + logger.debug("%r got a new connection from %r: %r", + server, addr, conn) protocol = protocol_factory() self._make_socket_transport( conn, protocol, diff --git a/asyncio/selector_events.py b/asyncio/selector_events.py index b965046..d79c080 100644 --- a/asyncio/selector_events.py +++ b/asyncio/selector_events.py @@ -23,6 +23,17 @@ from . import transports from .log import logger +def _test_selector_event(selector, fd, event): + # Test if the selector is monitoring 'event' events + # for the file descriptor 'fd'. + try: + key = selector.get_key(fd) + except KeyError: + return False + else: + return bool(key.events & event) + + class BaseSelectorEventLoop(base_events.BaseEventLoop): """Selector event loop. @@ -116,6 +127,9 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop): sslcontext=None, server=None): try: conn, addr = sock.accept() + if self._debug: + logger.debug("%r got a new connection from %r: %r", + server, addr, conn) conn.setblocking(False) except (BlockingIOError, InterruptedError, ConnectionAbortedError): pass # False alarm. @@ -419,6 +433,26 @@ class _SelectorTransport(transports._FlowControlMixin, if self._server is not None: self._server._attach() + def __repr__(self): + info = [self.__class__.__name__, 'fd=%s' % self._sock_fd] + polling = _test_selector_event(self._loop._selector, + self._sock_fd, selectors.EVENT_READ) + if polling: + info.append('read=polling') + else: + info.append('read=idle') + + polling = _test_selector_event(self._loop._selector, + self._sock_fd, selectors.EVENT_WRITE) + if polling: + state = 'polling' + else: + state = 'idle' + + bufsize = self.get_write_buffer_size() + info.append('write=<%s, bufsize=%s>' % (state, bufsize)) + return '<%s>' % ' '.join(info) + def abort(self): self._force_close(None) @@ -433,7 +467,10 @@ class _SelectorTransport(transports._FlowControlMixin, def _fatal_error(self, exc, message='Fatal error on transport'): # Should be called from exception handler only. - if not isinstance(exc, (BrokenPipeError, ConnectionResetError)): + if isinstance(exc, (BrokenPipeError, ConnectionResetError)): + if self._loop.get_debug(): + logger.debug("%r: %s", self, message, exc_info=True) + else: self._loop.call_exception_handler({ 'message': message, 'exception': exc, @@ -492,6 +529,8 @@ class _SelectorSocketTransport(_SelectorTransport): raise RuntimeError('Already paused') self._paused = True self._loop.remove_reader(self._sock_fd) + if self._loop.get_debug(): + logger.debug("%r pauses reading", self) def resume_reading(self): if not self._paused: @@ -500,6 +539,8 @@ class _SelectorSocketTransport(_SelectorTransport): if self._closing: return self._loop.add_reader(self._sock_fd, self._read_ready) + if self._loop.get_debug(): + logger.debug("%r resumes reading", self) def _read_ready(self): try: @@ -512,6 +553,8 @@ class _SelectorSocketTransport(_SelectorTransport): if data: self._protocol.data_received(data) else: + if self._loop.get_debug(): + logger.debug("%r received EOF", self) keep_open = self._protocol.eof_received() if keep_open: # We're keeping the connection open so the @@ -638,31 +681,37 @@ class _SelectorSslTransport(_SelectorTransport): # SSL-specific extra info. (peercert is set later) self._extra.update(sslcontext=sslcontext) - self._on_handshake() + if self._loop.get_debug(): + logger.debug("%r starts SSL handshake", self) + start_time = self._loop.time() + else: + start_time = None + self._on_handshake(start_time) - def _on_handshake(self): + def _on_handshake(self, start_time): try: self._sock.do_handshake() except ssl.SSLWantReadError: - self._loop.add_reader(self._sock_fd, self._on_handshake) + self._loop.add_reader(self._sock_fd, + self._on_handshake, start_time) return except ssl.SSLWantWriteError: - self._loop.add_writer(self._sock_fd, self._on_handshake) - return - except Exception as exc: - self._loop.remove_reader(self._sock_fd) - self._loop.remove_writer(self._sock_fd) - self._sock.close() - if self._waiter is not None: - self._waiter.set_exception(exc) + self._loop.add_writer(self._sock_fd, + self._on_handshake, start_time) return except BaseException as exc: + if self._loop.get_debug(): + logger.warning("%r: SSL handshake failed", + self, exc_info=True) self._loop.remove_reader(self._sock_fd) self._loop.remove_writer(self._sock_fd) self._sock.close() if self._waiter is not None: self._waiter.set_exception(exc) - raise + if isinstance(exc, Exception): + return + else: + raise self._loop.remove_reader(self._sock_fd) self._loop.remove_writer(self._sock_fd) @@ -676,6 +725,10 @@ class _SelectorSslTransport(_SelectorTransport): try: ssl.match_hostname(peercert, self._server_hostname) except Exception as exc: + if self._loop.get_debug(): + logger.warning("%r: SSL handshake failed " + "on matching the hostname", + self, exc_info=True) self._sock.close() if self._waiter is not None: self._waiter.set_exception(exc) @@ -696,6 +749,10 @@ class _SelectorSslTransport(_SelectorTransport): self._loop.call_soon(self._waiter._set_result_unless_cancelled, None) + if self._loop.get_debug(): + dt = self._loop.time() - start_time + logger.debug("%r: SSL handshake took %.1f ms", self, dt * 1e3) + def pause_reading(self): # XXX This is a bit icky, given the comment at the top of # _read_ready(). Is it possible to evoke a deadlock? I don't @@ -709,6 +766,8 @@ class _SelectorSslTransport(_SelectorTransport): raise RuntimeError('Already paused') self._paused = True self._loop.remove_reader(self._sock_fd) + if self._loop.get_debug(): + logger.debug("%r pauses reading", self) def resume_reading(self): if not self._paused: @@ -717,6 +776,8 @@ class _SelectorSslTransport(_SelectorTransport): if self._closing: return self._loop.add_reader(self._sock_fd, self._read_ready) + if self._loop.get_debug(): + logger.debug("%r resumes reading", self) def _read_ready(self): if self._write_wants_read: @@ -741,6 +802,8 @@ class _SelectorSslTransport(_SelectorTransport): self._protocol.data_received(data) else: try: + if self._loop.get_debug(): + logger.debug("%r received EOF", self) keep_open = self._protocol.eof_received() if keep_open: logger.warning('returning true from eof_received() ' diff --git a/asyncio/unix_events.py b/asyncio/unix_events.py index 764e719..09b875c 100644 --- a/asyncio/unix_events.py +++ b/asyncio/unix_events.py @@ -16,6 +16,7 @@ from . import base_subprocess from . import constants from . import events from . import selector_events +from . import selectors from . import transports from .coroutines import coroutine from .log import logger @@ -272,6 +273,20 @@ class _UnixReadPipeTransport(transports.ReadTransport): # wait until protocol.connection_made() has been called self._loop.call_soon(waiter._set_result_unless_cancelled, None) + def __repr__(self): + info = [self.__class__.__name__, 'fd=%s' % self._fileno] + if self._pipe is not None: + polling = selector_events._test_selector_event( + self._loop._selector, + self._fileno, selectors.EVENT_READ) + if polling: + info.append('polling') + else: + info.append('idle') + else: + info.append('closed') + return '<%s>' % ' '.join(info) + def _read_ready(self): try: data = os.read(self._fileno, self.max_size) @@ -283,6 +298,8 @@ class _UnixReadPipeTransport(transports.ReadTransport): if data: self._protocol.data_received(data) else: + if self._loop.get_debug(): + logger.info("%r was closed by peer", self) self._closing = True self._loop.remove_reader(self._fileno) self._loop.call_soon(self._protocol.eof_received) @@ -357,11 +374,30 @@ class _UnixWritePipeTransport(transports._FlowControlMixin, # wait until protocol.connection_made() has been called self._loop.call_soon(waiter._set_result_unless_cancelled, None) + def __repr__(self): + info = [self.__class__.__name__, 'fd=%s' % self._fileno] + if self._pipe is not None: + polling = selector_events._test_selector_event( + self._loop._selector, + self._fileno, selectors.EVENT_WRITE) + if polling: + info.append('polling') + else: + info.append('idle') + + bufsize = self.get_write_buffer_size() + info.append('bufsize=%s' % bufsize) + else: + info.append('closed') + return '<%s>' % ' '.join(info) + def get_write_buffer_size(self): return sum(len(data) for data in self._buffer) def _read_ready(self): # Pipe was closed by peer. + if self._loop.get_debug(): + logger.info("%r was closed by peer", self) if self._buffer: self._close(BrokenPipeError()) else: diff --git a/asyncio/windows_events.py b/asyncio/windows_events.py index 93b71b2..9d86c96 100644 --- a/asyncio/windows_events.py +++ b/asyncio/windows_events.py @@ -40,6 +40,18 @@ class _OverlappedFuture(futures.Future): super().__init__(loop=loop) self.ov = ov + def __repr__(self): + info = [self._state.lower()] + if self.ov.pending: + info.append('overlapped=pending') + else: + info.append('overlapped=completed') + if self._state == futures._FINISHED: + info.append(self._format_result()) + if self._callbacks: + info.append(self._format_callbacks()) + return '<%s %s>' % (self.__class__.__name__, ' '.join(info)) + def cancel(self): try: self.ov.cancel() |