fix asyncio subprocess

Lib/asyncio/base_subprocess.py

@@ -222,6 +222,7 @@ def _process_exited(self, returncode):
             if not waiter.cancelled():
                 waiter.set_result(returncode)
         self._exit_waiters = None
+        self.close()
 
     async def _wait(self):
         """Wait until the process exit and return the process return code.
@@ -234,6 +235,7 @@ async def _wait(self):
         self._exit_waiters.append(waiter)
         return await waiter
 
+
     def _try_finish(self):
         assert not self._finished
         if self._returncode is None:

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(

Misc/NEWS.d/next/Library/2022-02-28-09-06-26.bpo-43884.XcNo89.rst

@@ -0,0 +1 @@
+Fix :mod:`asyncio` subprocess to close it's transport when the process exits.