diff options
author | Victor Stinner <victor.stinner@gmail.com> | 2014-08-25 21:20:52 (GMT) |
---|---|---|
committer | Victor Stinner <victor.stinner@gmail.com> | 2014-08-25 21:20:52 (GMT) |
commit | b261475a48d905f160bc1f499e90b995b0d0b6c0 (patch) | |
tree | 1151160908141088d3fa02db4028006f67603d07 /Lib/asyncio | |
parent | d71dcbb043578c0abe770a2f37fac36e1a402821 (diff) | |
download | cpython-b261475a48d905f160bc1f499e90b995b0d0b6c0.zip cpython-b261475a48d905f160bc1f499e90b995b0d0b6c0.tar.gz cpython-b261475a48d905f160bc1f499e90b995b0d0b6c0.tar.bz2 |
asyncio: sync with Tulip
* PipeServer.close() now cancels the "accept pipe" future which cancels the
overlapped operation.
* Fix _SelectorTransport.__repr__() if the transport was closed
* Fix debug log in BaseEventLoop.create_connection(): get the socket object
from the transport because SSL transport closes the old socket and creates a
new SSL socket object. Remove also the _SelectorSslTransport._rawsock
attribute: it contained the closed socket (not very useful) and it was not
used.
* Issue #22063: socket operations (sock_recv, sock_sendall, sock_connect,
sock_accept) of the proactor event loop don't raise an exception in debug
mode if the socket are in blocking mode. Overlapped operations also work on
blocking sockets.
* Fix unit tests in debug mode: mock a non-blocking socket for socket
operations which now raise an exception if the socket is blocking.
* _fatal_error() method of _UnixReadPipeTransport and _UnixWritePipeTransport
now log all exceptions in debug mode
* Don't log expected errors in unit tests
* Tulip issue 200: _WaitHandleFuture._unregister_wait() now catchs and logs
exceptions.
* Tulip issue 200: Log errors in debug mode instead of simply ignoring them.
Diffstat (limited to 'Lib/asyncio')
-rw-r--r-- | Lib/asyncio/base_events.py | 7 | ||||
-rw-r--r-- | Lib/asyncio/proactor_events.py | 24 | ||||
-rw-r--r-- | Lib/asyncio/selector_events.py | 31 | ||||
-rw-r--r-- | Lib/asyncio/test_utils.py | 6 | ||||
-rw-r--r-- | Lib/asyncio/unix_events.py | 14 | ||||
-rw-r--r-- | Lib/asyncio/windows_events.py | 31 |
6 files changed, 78 insertions, 35 deletions
diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py index d0a337b..db13250 100644 --- a/Lib/asyncio/base_events.py +++ b/Lib/asyncio/base_events.py @@ -578,6 +578,9 @@ class BaseEventLoop(events.AbstractEventLoop): 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 @@ -725,6 +728,10 @@ class BaseEventLoop(events.AbstractEventLoop): 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: diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py index 751155b..0ad0656 100644 --- a/Lib/asyncio/proactor_events.py +++ b/Lib/asyncio/proactor_events.py @@ -172,6 +172,9 @@ class _ProactorReadPipeTransport(_ProactorBasePipeTransport, except ConnectionAbortedError as exc: if not self._closing: self._fatal_error(exc, 'Fatal read error on pipe transport') + elif self._loop.get_debug(): + logger.debug("Read error on pipe transport while closing", + exc_info=True) except ConnectionResetError as exc: self._force_close(exc) except OSError as exc: @@ -324,12 +327,16 @@ class _ProactorSocketTransport(_ProactorReadPipeTransport, try: self._extra['sockname'] = sock.getsockname() except (socket.error, AttributeError): - pass + if self._loop.get_debug(): + logger.warning("getsockname() failed on %r", + sock, exc_info=True) if 'peername' not in self._extra: try: self._extra['peername'] = sock.getpeername() except (socket.error, AttributeError): - pass + if self._loop.get_debug(): + logger.warning("getpeername() failed on %r", + sock, exc_info=True) def can_write_eof(self): return True @@ -385,18 +392,12 @@ class BaseProactorEventLoop(base_events.BaseEventLoop): self._selector = None def sock_recv(self, sock, n): - if self.get_debug() and sock.gettimeout() != 0: - raise ValueError("the socket must be non-blocking") return self._proactor.recv(sock, n) def sock_sendall(self, sock, data): - if self.get_debug() and sock.gettimeout() != 0: - raise ValueError("the socket must be non-blocking") return self._proactor.send(sock, data) def sock_connect(self, sock, address): - if self.get_debug() and sock.gettimeout() != 0: - raise ValueError("the socket must be non-blocking") try: base_events._check_resolved_address(sock, address) except ValueError as err: @@ -407,8 +408,6 @@ class BaseProactorEventLoop(base_events.BaseEventLoop): return self._proactor.connect(sock, address) def sock_accept(self, sock): - if self.get_debug() and sock.gettimeout() != 0: - raise ValueError("the socket must be non-blocking") return self._proactor.accept(sock) def _socketpair(self): @@ -470,11 +469,14 @@ class BaseProactorEventLoop(base_events.BaseEventLoop): except OSError as exc: if sock.fileno() != -1: self.call_exception_handler({ - 'message': 'Accept failed', + 'message': 'Accept failed on a socket', 'exception': exc, 'socket': sock, }) sock.close() + elif self._debug: + logger.debug("Accept failed on socket %r", + sock, exc_info=True) except futures.CancelledError: sock.close() else: diff --git a/Lib/asyncio/selector_events.py b/Lib/asyncio/selector_events.py index 6b7bdf0..0434a70 100644 --- a/Lib/asyncio/selector_events.py +++ b/Lib/asyncio/selector_events.py @@ -450,22 +450,24 @@ class _SelectorTransport(transports._FlowControlMixin, 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') + # test if the transport was closed + if self._loop is not None: + 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' + 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)) + bufsize = self.get_write_buffer_size() + info.append('write=<%s, bufsize=%s>' % (state, bufsize)) return '<%s>' % ' '.join(info) def abort(self): @@ -689,7 +691,6 @@ class _SelectorSslTransport(_SelectorTransport): self._server_hostname = server_hostname self._waiter = waiter - self._rawsock = rawsock self._sslcontext = sslcontext self._paused = False diff --git a/Lib/asyncio/test_utils.py b/Lib/asyncio/test_utils.py index 840bbf9..ac7680d 100644 --- a/Lib/asyncio/test_utils.py +++ b/Lib/asyncio/test_utils.py @@ -417,3 +417,9 @@ def disable_logger(): yield finally: logger.setLevel(old_level) + +def mock_nonblocking_socket(): + """Create a mock of a non-blocking socket.""" + sock = mock.Mock(socket.socket) + sock.gettimeout.return_value = 0.0 + return sock diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py index 8d3e25e..37310cf 100644 --- a/Lib/asyncio/unix_events.py +++ b/Lib/asyncio/unix_events.py @@ -337,7 +337,10 @@ class _UnixReadPipeTransport(transports.ReadTransport): def _fatal_error(self, exc, message='Fatal error on pipe transport'): # should be called by exception handler only - if not (isinstance(exc, OSError) and exc.errno == errno.EIO): + if (isinstance(exc, OSError) and exc.errno == errno.EIO): + if self._loop.get_debug(): + logger.debug("%r: %s", self, message, exc_info=True) + else: self._loop.call_exception_handler({ 'message': message, 'exception': exc, @@ -509,7 +512,10 @@ class _UnixWritePipeTransport(transports._FlowControlMixin, def _fatal_error(self, exc, message='Fatal error on pipe transport'): # should be called by 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, @@ -750,7 +756,9 @@ class SafeChildWatcher(BaseChildWatcher): except KeyError: # pragma: no cover # May happen if .remove_child_handler() is called # after os.waitpid() returns. - pass + if self._loop.get_debug(): + logger.warning("Child watcher got an unexpected pid: %r", + pid, exc_info=True) else: callback(pid, returncode, *args) diff --git a/Lib/asyncio/windows_events.py b/Lib/asyncio/windows_events.py index ec427d5..6763f0b 100644 --- a/Lib/asyncio/windows_events.py +++ b/Lib/asyncio/windows_events.py @@ -111,10 +111,17 @@ class _WaitHandleFuture(futures.Future): return try: _overlapped.UnregisterWait(self._wait_handle) - except OSError as e: - if e.winerror != _overlapped.ERROR_IO_PENDING: - raise + except OSError as exc: # ERROR_IO_PENDING is not an error, the wait was unregistered + if exc.winerror != _overlapped.ERROR_IO_PENDING: + context = { + 'message': 'Failed to unregister the wait handle', + 'exception': exc, + 'future': self, + } + if self._source_traceback: + context['source_traceback'] = self._source_traceback + self._loop.call_exception_handler(context) self._wait_handle = None self._iocp = None self._ov = None @@ -145,6 +152,11 @@ class PipeServer(object): def __init__(self, address): self._address = address self._free_instances = weakref.WeakSet() + # initialize the pipe attribute before calling _server_pipe_handle() + # because this function can raise an exception and the destructor calls + # the close() method + self._pipe = None + self._accept_pipe_future = None self._pipe = self._server_pipe_handle(True) def _get_unconnected_pipe(self): @@ -174,6 +186,9 @@ class PipeServer(object): return pipe def close(self): + if self._accept_pipe_future is not None: + self._accept_pipe_future.cancel() + self._accept_pipe_future = None # Close all instances which have not been connected to by a client. if self._address is not None: for pipe in self._free_instances: @@ -216,7 +231,7 @@ class ProactorEventLoop(proactor_events.BaseProactorEventLoop): def start_serving_pipe(self, protocol_factory, address): server = PipeServer(address) - def loop(f=None): + def loop_accept_pipe(f=None): pipe = None try: if f: @@ -237,13 +252,17 @@ class ProactorEventLoop(proactor_events.BaseProactorEventLoop): 'pipe': pipe, }) pipe.close() + elif self._debug: + logger.warning("Accept pipe failed on pipe %r", + pipe, exc_info=True) except futures.CancelledError: if pipe: pipe.close() else: - f.add_done_callback(loop) + server._accept_pipe_future = f + f.add_done_callback(loop_accept_pipe) - self.call_soon(loop) + self.call_soon(loop_accept_pipe) return [server] @coroutine |