(Merge 3.4) asyncio: sync with Tulip

Improve stability of the proactor event loop, especially operations on
overlapped objects:

* Tulip issue 195: Don't call UnregisterWait() twice if a _WaitHandleFuture is
  cancelled twice to fix a crash.
* IocpProactor.close(): cancel futures to cancel overlapped operations, instead
  of cancelling directly overlapped operations. Future objects may not call
  ov.cancel() if the future was cancelled or if the overlapped was already
  cancelled. The cancel() method of the future may also catch exceptions. Log
  also errors on cancellation.
* tests: rename "f" to "fut"
* Add a __repr__() method to IocpProactor
* Add a destructor to IocpProactor which closes it
* _OverlappedFuture.cancel() doesn't cancel the overlapped anymore if it is
  done: if it is already cancelled or completed. Log also an error if the
  cancellation failed.
* Add the address of the overlapped object in repr(_OverlappedFuture)
* _OverlappedFuture truncates the source traceback to hide the call to the
  parent constructor (useless in debug).
This commit is contained in:
Victor Stinner 2014-07-25 00:55:23 +02:00
commit 3953398363
2 changed files with 66 additions and 30 deletions

View File

@ -38,14 +38,14 @@ class _OverlappedFuture(futures.Future):
def __init__(self, ov, *, loop=None): def __init__(self, ov, *, loop=None):
super().__init__(loop=loop) super().__init__(loop=loop)
if self._source_traceback:
del self._source_traceback[-1]
self.ov = ov self.ov = ov
def __repr__(self): def __repr__(self):
info = [self._state.lower()] info = [self._state.lower()]
if self.ov.pending: state = 'pending' if self.ov.pending else 'completed'
info.append('overlapped=pending') info.append('overlapped=<%s, %#x>' % (state, self.ov.address))
else:
info.append('overlapped=completed')
if self._state == futures._FINISHED: if self._state == futures._FINISHED:
info.append(self._format_result()) info.append(self._format_result())
if self._callbacks: if self._callbacks:
@ -53,10 +53,18 @@ class _OverlappedFuture(futures.Future):
return '<%s %s>' % (self.__class__.__name__, ' '.join(info)) return '<%s %s>' % (self.__class__.__name__, ' '.join(info))
def cancel(self): def cancel(self):
if not self.done():
try: try:
self.ov.cancel() self.ov.cancel()
except OSError: except OSError as exc:
pass context = {
'message': 'Cancelling an overlapped future failed',
'exception': exc,
'future': self,
}
if self._source_traceback:
context['source_traceback'] = self._source_traceback
self._loop.call_exception_handler(context)
return super().cancel() return super().cancel()
@ -67,13 +75,20 @@ class _WaitHandleFuture(futures.Future):
super().__init__(loop=loop) super().__init__(loop=loop)
self._wait_handle = wait_handle self._wait_handle = wait_handle
def cancel(self): def _unregister(self):
super().cancel() if self._wait_handle is None:
return
try: try:
_overlapped.UnregisterWait(self._wait_handle) _overlapped.UnregisterWait(self._wait_handle)
except OSError as e: except OSError as e:
if e.winerror != _overlapped.ERROR_IO_PENDING: if e.winerror != _overlapped.ERROR_IO_PENDING:
raise raise
# ERROR_IO_PENDING is not an error, the wait was unregistered
self._wait_handle = None
def cancel(self):
self._unregister()
super().cancel()
class PipeServer(object): class PipeServer(object):
@ -208,6 +223,11 @@ class IocpProactor:
self._registered = weakref.WeakSet() self._registered = weakref.WeakSet()
self._stopped_serving = weakref.WeakSet() self._stopped_serving = weakref.WeakSet()
def __repr__(self):
return ('<%s overlapped#=%s result#=%s>'
% (self.__class__.__name__, len(self._cache),
len(self._results)))
def set_loop(self, loop): def set_loop(self, loop):
self._loop = loop self._loop = loop
@ -353,12 +373,7 @@ class IocpProactor:
f = _WaitHandleFuture(wh, loop=self._loop) f = _WaitHandleFuture(wh, loop=self._loop)
def finish_wait_for_handle(trans, key, ov): def finish_wait_for_handle(trans, key, ov):
if not f.cancelled(): f._unregister()
try:
_overlapped.UnregisterWait(wh)
except OSError as e:
if e.winerror != _overlapped.ERROR_IO_PENDING:
raise
# Note that this second wait means that we should only use # Note that this second wait means that we should only use
# this with handles types where a successful wait has no # this with handles types where a successful wait has no
# effect. So events or processes are all right, but locks # effect. So events or processes are all right, but locks
@ -455,7 +470,7 @@ class IocpProactor:
def close(self): def close(self):
# Cancel remaining registered operations. # Cancel remaining registered operations.
for address, (f, ov, obj, callback) in list(self._cache.items()): for address, (fut, ov, obj, callback) in list(self._cache.items()):
if obj is None: if obj is None:
# The operation was started with connect_pipe() which # The operation was started with connect_pipe() which
# queues a task to Windows' thread pool. This cannot # queues a task to Windows' thread pool. This cannot
@ -463,9 +478,17 @@ class IocpProactor:
del self._cache[address] del self._cache[address]
else: else:
try: try:
ov.cancel() fut.cancel()
except OSError: except OSError as exc:
pass if self._loop is not None:
context = {
'message': 'Cancelling a future failed',
'exception': exc,
'future': fut,
}
if fut._source_traceback:
context['source_traceback'] = fut._source_traceback
self._loop.call_exception_handler(context)
while self._cache: while self._cache:
if not self._poll(1): if not self._poll(1):
@ -476,6 +499,9 @@ class IocpProactor:
_winapi.CloseHandle(self._iocp) _winapi.CloseHandle(self._iocp)
self._iocp = None self._iocp = None
def __del__(self):
self.close()
class _WindowsSubprocessTransport(base_subprocess.BaseSubprocessTransport): class _WindowsSubprocessTransport(base_subprocess.BaseSubprocessTransport):

View File

@ -96,36 +96,46 @@ class ProactorTests(test_utils.TestCase):
# Wait for unset event with 0.5s timeout; # Wait for unset event with 0.5s timeout;
# result should be False at timeout # result should be False at timeout
f = self.loop._proactor.wait_for_handle(event, 0.5) fut = self.loop._proactor.wait_for_handle(event, 0.5)
start = self.loop.time() start = self.loop.time()
self.loop.run_until_complete(f) self.loop.run_until_complete(fut)
elapsed = self.loop.time() - start elapsed = self.loop.time() - start
self.assertFalse(f.result()) self.assertFalse(fut.result())
self.assertTrue(0.48 < elapsed < 0.9, elapsed) self.assertTrue(0.48 < elapsed < 0.9, elapsed)
_overlapped.SetEvent(event) _overlapped.SetEvent(event)
# Wait for for set event; # Wait for for set event;
# result should be True immediately # result should be True immediately
f = self.loop._proactor.wait_for_handle(event, 10) fut = self.loop._proactor.wait_for_handle(event, 10)
start = self.loop.time() start = self.loop.time()
self.loop.run_until_complete(f) self.loop.run_until_complete(fut)
elapsed = self.loop.time() - start elapsed = self.loop.time() - start
self.assertTrue(f.result()) self.assertTrue(fut.result())
self.assertTrue(0 <= elapsed < 0.3, elapsed) self.assertTrue(0 <= elapsed < 0.3, elapsed)
_overlapped.ResetEvent(event) # Tulip issue #195: cancelling a done _WaitHandleFuture must not crash
fut.cancel()
def test_wait_for_handle_cancel(self):
event = _overlapped.CreateEvent(None, True, False, None)
self.addCleanup(_winapi.CloseHandle, event)
# Wait for unset event with a cancelled future; # Wait for unset event with a cancelled future;
# CancelledError should be raised immediately # CancelledError should be raised immediately
f = self.loop._proactor.wait_for_handle(event, 10) fut = self.loop._proactor.wait_for_handle(event, 10)
f.cancel() fut.cancel()
start = self.loop.time() start = self.loop.time()
with self.assertRaises(asyncio.CancelledError): with self.assertRaises(asyncio.CancelledError):
self.loop.run_until_complete(f) self.loop.run_until_complete(fut)
elapsed = self.loop.time() - start elapsed = self.loop.time() - start
self.assertTrue(0 <= elapsed < 0.1, elapsed) self.assertTrue(0 <= elapsed < 0.1, elapsed)
# Tulip issue #195: cancelling a _WaitHandleFuture twice must not crash
fut = self.loop._proactor.wait_for_handle(event)
fut.cancel()
fut.cancel()
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()