gh-110036: multiprocessing Popen.terminate() catches PermissionError (#110037)

On Windows, multiprocessing Popen.terminate() now catchs
PermissionError and get the process exit code. If the process is
still running, raise again the PermissionError. Otherwise, the
process terminated as expected: store its exit code.

2 files changed, 12 insertions, 4 deletions

diff --git a/Lib/multiprocessing/popen_spawn_win32.py b/Lib/multiprocessing/popen_spawn_win32.py
index 4d60ffc..af04430 100644
--- a/Lib/multiprocessing/popen_spawn_win32.py
+++ b/Lib/multiprocessing/popen_spawn_win32.py
@@ -14,6 +14,7 @@ __all__ = ['Popen']
 #
 #
 
+# Exit code used by Popen.terminate()
 TERMINATE = 0x10000
 WINEXE = (sys.platform == 'win32' and getattr(sys, 'frozen', False))
 WINSERVICE = sys.executable.lower().endswith("pythonservice.exe")
@@ -122,9 +123,15 @@ class Popen(object):
         if self.returncode is None:
             try:
                 _winapi.TerminateProcess(int(self._handle), TERMINATE)
-            except OSError:
-                if self.wait(timeout=1.0) is None:
+            except PermissionError:
+                # ERROR_ACCESS_DENIED (winerror 5) is received when the
+                # process already died.
+                code = _winapi.GetExitCodeProcess(int(self._handle))
+                if code == _winapi.STILL_ACTIVE:
                     raise
+                self.returncode = code
+            else:
+                self.returncode = -signal.SIGTERM
 
     kill = terminate
 
diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py
index 756d680..39666dd 100644
--- a/Lib/test/_test_multiprocessing.py
+++ b/Lib/test/_test_multiprocessing.py
@@ -557,13 +557,14 @@ class _TestProcess(BaseTestCase):
 
     def test_terminate(self):
         exitcode = self._kill_process(multiprocessing.Process.terminate)
-        if os.name != 'nt':
-            self.assertEqual(exitcode, -signal.SIGTERM)
+        self.assertEqual(exitcode, -signal.SIGTERM)
 
     def test_kill(self):
         exitcode = self._kill_process(multiprocessing.Process.kill)
         if os.name != 'nt':
             self.assertEqual(exitcode, -signal.SIGKILL)
+        else:
+            self.assertEqual(exitcode, -signal.SIGTERM)
 
     def test_cpu_count(self):
         try: