bpo-26819: Prevent proactor double read on resume (#6921)
The proactor event loop has a race condition when reading with
pausing/resuming. `resume_reading()` unconditionally schedules the read
function to read from the current future. If `resume_reading()` was
called before the previously scheduled done callback fires, this results
in two attempts to get the data from the most recent read and an
assertion failure. This commit tracks whether or not `resume_reading`
needs to reschedule the callback to restart the loop, preventing a
second attempt to read the data.
diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py
index b675c82..877dfb0 100644
--- a/Lib/asyncio/proactor_events.py
+++ b/Lib/asyncio/proactor_events.py
@@ -161,6 +161,7 @@
extra=None, server=None):
super().__init__(loop, sock, protocol, waiter, extra, server)
self._paused = False
+ self._reschedule_on_resume = False
if protocols._is_buffered_protocol(protocol):
self._loop_reading = self._loop_reading__get_buffer
@@ -180,6 +181,7 @@
if self._read_fut is not None and not self._read_fut.done():
self._read_fut.cancel()
self._read_fut = None
+ self._reschedule_on_resume = True
if self._loop.get_debug():
logger.debug("%r pauses reading", self)
@@ -188,7 +190,9 @@
if self._closing or not self._paused:
return
self._paused = False
- self._loop.call_soon(self._loop_reading, self._read_fut)
+ if self._reschedule_on_resume:
+ self._loop.call_soon(self._loop_reading, self._read_fut)
+ self._reschedule_on_resume = False
if self._loop.get_debug():
logger.debug("%r resumes reading", self)
@@ -208,6 +212,7 @@
def _loop_reading__data_received(self, fut=None):
if self._paused:
+ self._reschedule_on_resume = True
return
data = None
@@ -257,6 +262,7 @@
def _loop_reading__get_buffer(self, fut=None):
if self._paused:
+ self._reschedule_on_resume = True
return
nbytes = None
diff --git a/Lib/test/test_asyncio/test_proactor_events.py b/Lib/test/test_asyncio/test_proactor_events.py
index 98e6989..6313d59 100644
--- a/Lib/test/test_asyncio/test_proactor_events.py
+++ b/Lib/test/test_asyncio/test_proactor_events.py
@@ -334,7 +334,7 @@
def test_pause_resume_reading(self):
tr = self.socket_transport()
futures = []
- for msg in [b'data1', b'data2', b'data3', b'data4', b'']:
+ for msg in [b'data1', b'data2', b'data3', b'data4', b'data5', b'']:
f = asyncio.Future(loop=self.loop)
f.set_result(msg)
futures.append(f)
@@ -364,6 +364,13 @@
self.protocol.data_received.assert_called_with(b'data3')
self.loop._run_once()
self.protocol.data_received.assert_called_with(b'data4')
+
+ tr.pause_reading()
+ tr.resume_reading()
+ self.loop.call_exception_handler = mock.Mock()
+ self.loop._run_once()
+ self.loop.call_exception_handler.assert_not_called()
+ self.protocol.data_received.assert_called_with(b'data5')
tr.close()
self.assertFalse(tr.is_reading())
diff --git a/Misc/NEWS.d/next/Library/2018-05-16-05-24-43.bpo-26819.taxbVT.rst b/Misc/NEWS.d/next/Library/2018-05-16-05-24-43.bpo-26819.taxbVT.rst
new file mode 100644
index 0000000..d407a58
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2018-05-16-05-24-43.bpo-26819.taxbVT.rst
@@ -0,0 +1,2 @@
+Fix race condition with `ReadTransport.resume_reading` in Windows proactor
+event loop.