bpo-43884: Fix asyncio subprocess kill process cleanly when process i…

…s blocked

Lib/asyncio/base_subprocess.py

@@ -214,14 +214,22 @@ def _process_exited(self, returncode):
             # asyncio uses a child watcher: copy the status into the Popen
             # object. On Python 3.6, it is required to avoid a ResourceWarning.
             self._proc.returncode = returncode
-        self._call(self._protocol.process_exited)
-        self._try_finish()
 
-        # wake up futures waiting for wait()
-        for waiter in self._exit_waiters:
-            if not waiter.cancelled():
-                waiter.set_result(returncode)
-        self._exit_waiters = None
+        def __process_exited():
+            self._protocol.process_exited()
+            # Since process exited, clear all pipes
+            for proto in self._pipes.values():
+                if proto is None:
+                    continue
+                proto.pipe.close()
+            # Call _wait waiters
+            for waiter in self._exit_waiters:
+                if not waiter.cancelled():
+                    waiter.set_result(self._returncode)
+            self._exit_waiters = None
+
+        self._call(__process_exited)
+        self._try_finish()
 
     async def _wait(self):
         """Wait until the process exit and return the process return code.
@@ -252,6 +260,7 @@ def _call_connection_lost(self, exc):
             self._protocol = None
 
 
+
 class WriteSubprocessPipeProto(protocols.BaseProtocol):
 
     def __init__(self, proc, fd):

Lib/test/test_asyncio/test_subprocess.py

@@ -181,6 +181,20 @@ def test_kill(self):
         else:
             self.assertEqual(-signal.SIGKILL, returncode)
 
+    def test_kill_issue43884(self):
+        blocking_shell_command = f'{sys.executable} -c "import time; time.sleep(10000)"'
+        proc = self.loop.run_until_complete(
+            asyncio.create_subprocess_shell(blocking_shell_command, stdout=asyncio.subprocess.PIPE)
+        )
+        self.loop.run_until_complete(asyncio.sleep(1))
+        proc.kill()
+        returncode = self.loop.run_until_complete(proc.wait())
+        if sys.platform == 'win32':
+            self.assertIsInstance(returncode, int)
+            # expect 1 but sometimes get 0
+        else:
+            self.assertEqual(-signal.SIGKILL, returncode)
+
     def test_terminate(self):
         args = PROGRAM_BLOCKED
         proc = self.loop.run_until_complete(