[Python-checkins] bpo-26819: Prevent proactor double read on resume (#6921)

[Andrew Svetlov]

https://github.com/python/cpython/commit/4151061855b571bf8a7579daa7875b8e243057b9
commit: 4151061855b571bf8a7579daa7875b8e243057b9
branch: master
author: CtrlZvi <viz+github at flippedperspective.com>
committer: Andrew Svetlov <andrew.svetlov at gmail.com>
date: 2018-05-20T13:21:10+03:00
summary:

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.

files:
A Misc/NEWS.d/next/Library/2018-05-16-05-24-43.bpo-26819.taxbVT.rst
M Lib/asyncio/proactor_events.py
M Lib/test/test_asyncio/test_proactor_events.py

diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py
index b675c8200ce3..877dfb074670 100644
--- a/Lib/asyncio/proactor_events.py
+++ b/Lib/asyncio/proactor_events.py
@@ -161,6 +161,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
 
         if protocols._is_buffered_protocol(protocol):
             self._loop_reading = self._loop_reading__get_buffer
@@ -180,6 +181,7 @@ def pause_reading(self):
         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 @@ def resume_reading(self):
         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__on_eof(self):
 
     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__data_received(self, fut=None):
 
     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 98e698983eab..6313d594477a 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_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)
@@ -364,6 +364,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()
 
         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 000000000000..d407a5800318
--- /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.