[3.6] bpo-26819: Prevent proactor double read on resume (GH-6921) (#7110)

CtrlZvi · asvetlov · commit 749afe81ec0a · 2018-05-25T11:03:25.000+03:00
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.. (cherry picked from commit 4151061) Co-authored-by: CtrlZvi <viz+github@flippedperspective.com>
diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py
@@ -156,6 +156,7 @@ def __init__(self, loop, sock, protocol, waiter=None,
                  extra=None, server=None):
         super().__init__(loop, sock, protocol, waiter, extra, server)
         self._paused = False
+        self._reschedule_on_resume = False
         self._loop.call_soon(self._loop_reading)
 
     def pause_reading(self):
@@ -173,12 +174,15 @@ def resume_reading(self):
         self._paused = False
         if self._closing:
             return
-        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)
 
     def _loop_reading(self, fut=None):
         if self._paused:
+            self._reschedule_on_resume = True
             return
         data = None
 
diff --git a/Lib/test/test_asyncio/test_proactor_events.py b/Lib/test/test_asyncio/test_proactor_events.py
@@ -330,7 +330,7 @@ def test_write_eof_duplex_pipe(self):
     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)
@@ -352,6 +352,13 @@ def test_pause_resume_reading(self):
         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()
 
 
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
@@ -0,0 +1,2 @@
+Fix race condition with `ReadTransport.resume_reading` in Windows proactor
+event loop.

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	+Fix race condition with `ReadTransport.resume_reading` in Windows proactor
	`2`	`+event loop.`