From: CtrlZvi Date: Sun, 20 May 2018 10:21:10 +0000 (-0700) Subject: bpo-26819: Prevent proactor double read on resume (#6921) X-Git-Tag: v3.8.0a1~1801 X-Git-Url: https://granicus.if.org/sourcecode?a=commitdiff_plain;h=4151061855b571bf8a7579daa7875b8e243057b9;p=python 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 b675c8200c..877dfb0746 100644 --- 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 98e698983e..6313d59447 100644 --- 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 index 0000000000..d407a58003 --- /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.