From 58498bc7178608b1ab031994ca09c43889ce3e76 Mon Sep 17 00:00:00 2001
From: Andrew Svetlov <andrew.svetlov@gmail.com>
Date: Sun, 29 Sep 2019 15:00:35 +0300
Subject: [PATCH] bpo-38019: correctly handle pause/resume reading of closed
 asyncio unix pipe (GH-16472)

---
 Lib/asyncio/unix_events.py                    | 11 ++++++++
 Lib/test/test_asyncio/test_unix_events.py     | 27 +++++++++++++++++++
 .../2019-09-29-13-50-24.bpo-38019.6MoOE3.rst  |  1 +
 3 files changed, 39 insertions(+)
 create mode 100644 Misc/NEWS.d/next/Library/2019-09-29-13-50-24.bpo-38019.6MoOE3.rst

diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py
index cbbb1065aed..d8f653045ae 100644
--- a/Lib/asyncio/unix_events.py
+++ b/Lib/asyncio/unix_events.py
@@ -445,6 +445,7 @@ class _UnixReadPipeTransport(transports.ReadTransport):
         self._fileno = pipe.fileno()
         self._protocol = protocol
         self._closing = False
+        self._paused = False
 
         mode = os.fstat(self._fileno).st_mode
         if not (stat.S_ISFIFO(mode) or
@@ -506,10 +507,20 @@ class _UnixReadPipeTransport(transports.ReadTransport):
                 self._loop.call_soon(self._call_connection_lost, None)
 
     def pause_reading(self):
+        if self._closing or self._paused:
+            return
+        self._paused = True
         self._loop._remove_reader(self._fileno)
+        if self._loop.get_debug():
+            logger.debug("%r pauses reading", self)
 
     def resume_reading(self):
+        if self._closing or not self._paused:
+            return
+        self._paused = False
         self._loop._add_reader(self._fileno, self._read_ready)
+        if self._loop.get_debug():
+            logger.debug("%r resumes reading", self)
 
     def set_protocol(self, protocol):
         self._protocol = protocol
diff --git a/Lib/test/test_asyncio/test_unix_events.py b/Lib/test/test_asyncio/test_unix_events.py
index 02ce7e89da0..5487b7afef8 100644
--- a/Lib/test/test_asyncio/test_unix_events.py
+++ b/Lib/test/test_asyncio/test_unix_events.py
@@ -736,6 +736,7 @@ class UnixReadPipeTransportTests(test_utils.TestCase):
     @mock.patch('os.read')
     def test_resume_reading(self, m_read):
         tr = self.read_pipe_transport()
+        tr.pause_reading()
         tr.resume_reading()
         self.loop.assert_reader(5, tr._read_ready)
 
@@ -790,6 +791,32 @@ class UnixReadPipeTransportTests(test_utils.TestCase):
         self.assertIsNone(tr._protocol)
         self.assertIsNone(tr._loop)
 
+    def test_pause_reading_on_closed_pipe(self):
+        tr = self.read_pipe_transport()
+        tr.close()
+        test_utils.run_briefly(self.loop)
+        self.assertIsNone(tr._loop)
+        tr.pause_reading()
+
+    def test_pause_reading_on_paused_pipe(self):
+        tr = self.read_pipe_transport()
+        tr.pause_reading()
+        # the second call should do nothing
+        tr.pause_reading()
+
+    def test_resume_reading_on_closed_pipe(self):
+        tr = self.read_pipe_transport()
+        tr.close()
+        test_utils.run_briefly(self.loop)
+        self.assertIsNone(tr._loop)
+        tr.resume_reading()
+
+    def test_resume_reading_on_paused_pipe(self):
+        tr = self.read_pipe_transport()
+        # the pipe is not paused
+        # resuming should do nothing
+        tr.resume_reading()
+
 
 class UnixWritePipeTransportTests(test_utils.TestCase):
 
diff --git a/Misc/NEWS.d/next/Library/2019-09-29-13-50-24.bpo-38019.6MoOE3.rst b/Misc/NEWS.d/next/Library/2019-09-29-13-50-24.bpo-38019.6MoOE3.rst
new file mode 100644
index 00000000000..aa5a23de246
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2019-09-29-13-50-24.bpo-38019.6MoOE3.rst
@@ -0,0 +1 @@
+Correctly handle pause/resume reading of closed asyncio unix pipe.