mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-10-14 02:43:49 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

asyncio: sync with Tulip

Browse source 
* Tulip issue #183: log socket events in debug mode

  - Log most important socket events: socket connected, new client, connection
    reset or closed by peer (EOF), etc.
  - Log time elapsed in DNS resolution (getaddrinfo)
  - Log pause/resume reading
  - Log time of SSL handshake
  - Log SSL handshake errors
  - Add a __repr__() method to many classes

* Fix ProactorEventLoop() in debug mode. ProactorEventLoop._make_self_pipe()
  doesn't call call_soon() directly because it checks for the current loop
  which fails, because the method is called to build the event loop.

* Cleanup _ProactorReadPipeTransport constructor. Not need to set again
  _read_fut attribute to None, it is already done in the base class.
This commit is contained in:
Victor Stinner 2014-07-12 03:11:53 +02:00
parent 8ebeb03740
commit e912e652f8
7 changed files with 219 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

89
Lib/asyncio/selector_events.py
View file

@ -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() '