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 b675c8200ce30d8c0e62811b4ef0f79baca695df..877dfb0746708ecb9ab51adb76ec0eead3810f80 100644 (file)
--- a/Lib/asyncio/proactor_events.py
+++ b/Lib/asyncio/proactor_events.py
@@ -161,6 +161,7 @@ class _ProactorReadPipeTransport(_ProactorBasePipeTransport,
                  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 @@ class _ProactorReadPipeTransport(_ProactorBasePipeTransport,
         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 @@ class _ProactorReadPipeTransport(_ProactorBasePipeTransport,
         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 @@ class _ProactorReadPipeTransport(_ProactorBasePipeTransport,
 
     def _loop_reading__data_received(self, fut=None):
         if self._paused:
+            self._reschedule_on_resume = True
             return
 
         data = None
@@ -257,6 +262,7 @@ class _ProactorReadPipeTransport(_ProactorBasePipeTransport,
 
     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 98e698983eab5c1f7cb65c4b379595b78a1e2136..6313d594477a74d4984302f268f94965d12b1fc7 100644 (file)
--- a/Lib/test/test_asyncio/test_proactor_events.py
+++ b/Lib/test/test_asyncio/test_proactor_events.py
@@ -334,7 +334,7 @@ class ProactorSocketTransportTests(test_utils.TestCase):
     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 @@ class ProactorSocketTransportTests(test_utils.TestCase):
         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 (file)
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.