From 749afe81ec0a4b92ad6b89a67c82f2c04f79c5ac Mon Sep 17 00:00:00 2001
From: CtrlZvi <viz+github@flippedperspective.com>
Date: Fri, 25 May 2018 01:03:25 -0700
Subject: [PATCH] [3.6] bpo-26819: Prevent proactor double read on resume
 (GH-6921) (#7110)

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 4151061855b571bf8a7579daa7875b8e243057b9)

Co-authored-by: CtrlZvi <viz+github@flippedperspective.com>
---
 Lib/asyncio/proactor_events.py                           | 6 +++++-
 Lib/test/test_asyncio/test_proactor_events.py            | 9 ++++++++-
 .../Library/2018-05-16-05-24-43.bpo-26819.taxbVT.rst     | 2 ++
 3 files changed, 15 insertions(+), 2 deletions(-)
 create mode 100644 Misc/NEWS.d/next/Library/2018-05-16-05-24-43.bpo-26819.taxbVT.rst

diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py
index a81645d9c5d..967a696961a 100644
--- a/Lib/asyncio/proactor_events.py
+++ b/Lib/asyncio/proactor_events.py
@@ -156,6 +156,7 @@ class _ProactorReadPipeTransport(_ProactorBasePipeTransport,
                  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 @@ class _ProactorReadPipeTransport(_ProactorBasePipeTransport,
         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
index d76da661ab9..edf0461957f 100644
--- a/Lib/test/test_asyncio/test_proactor_events.py
+++ b/Lib/test/test_asyncio/test_proactor_events.py
@@ -330,7 +330,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)
@@ -352,6 +352,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()
 
 
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 00000000000..d407a580031
--- /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.