Issue #11750: The Windows API functions scattered in the _subprocess and

_multiprocessing.win32 modules now live in a single module "_winapi".
Patch by sbt.

Author: pitrou

Lib/multiprocessing/connection.py

@@ -51,12 +51,12 @@
 from multiprocessing.util import (
     get_temp_dir, Finalize, sub_debug, debug, _eintr_retry)
 try:
-    from _multiprocessing import win32
-    from _subprocess import WAIT_OBJECT_0, WAIT_TIMEOUT, INFINITE
+    import _winapi
+    from _winapi import WAIT_OBJECT_0, WAIT_TIMEOUT, INFINITE
 except ImportError:
     if sys.platform == 'win32':
         raise
-    win32 = None
+    _winapi = None
 
 #
 #
@@ -282,7 +282,7 @@ def poll(self, timeout=0.0):
         return self._poll(timeout)
 
 
-if win32:
+if _winapi:
 
     class PipeConnection(_ConnectionBase):
         """
@@ -292,14 +292,14 @@ class PipeConnection(_ConnectionBase):
         """
         _got_empty_message = False
 
-        def _close(self, _CloseHandle=win32.CloseHandle):
+        def _close(self, _CloseHandle=_winapi.CloseHandle):
             _CloseHandle(self._handle)
 
         def _send_bytes(self, buf):
-            ov, err = win32.WriteFile(self._handle, buf, overlapped=True)
+            ov, err = _winapi.WriteFile(self._handle, buf, overlapped=True)
             try:
-                if err == win32.ERROR_IO_PENDING:
-                    waitres = win32.WaitForMultipleObjects(
+                if err == _winapi.ERROR_IO_PENDING:
+                    waitres = _winapi.WaitForMultipleObjects(
                         [ov.event], False, INFINITE)
                     assert waitres == WAIT_OBJECT_0
             except:
@@ -317,11 +317,11 @@ def _recv_bytes(self, maxsize=None):
             else:
                 bsize = 128 if maxsize is None else min(maxsize, 128)
                 try:
-                    ov, err = win32.ReadFile(self._handle, bsize,
-                                             overlapped=True)
+                    ov, err = _winapi.ReadFile(self._handle, bsize,
+                                                overlapped=True)
                     try:
-                        if err == win32.ERROR_IO_PENDING:
-                            waitres = win32.WaitForMultipleObjects(
+                        if err == _winapi.ERROR_IO_PENDING:
+                            waitres = _winapi.WaitForMultipleObjects(
                                 [ov.event], False, INFINITE)
                             assert waitres == WAIT_OBJECT_0
                     except:
@@ -333,18 +333,18 @@ def _recv_bytes(self, maxsize=None):
                             f = io.BytesIO()
                             f.write(ov.getbuffer())
                             return f
-                        elif err == win32.ERROR_MORE_DATA:
+                        elif err == _winapi.ERROR_MORE_DATA:
                             return self._get_more_data(ov, maxsize)
                 except IOError as e:
-                    if e.winerror == win32.ERROR_BROKEN_PIPE:
+                    if e.winerror == _winapi.ERROR_BROKEN_PIPE:
                         raise EOFError
                     else:
                         raise
             raise RuntimeError("shouldn't get here; expected KeyboardInterrupt")
 
         def _poll(self, timeout):
             if (self._got_empty_message or
-                        win32.PeekNamedPipe(self._handle)[0] != 0):
+                        _winapi.PeekNamedPipe(self._handle)[0] != 0):
                 return True
             if timeout < 0:
                 timeout = None
@@ -354,11 +354,11 @@ def _get_more_data(self, ov, maxsize):
             buf = ov.getbuffer()
             f = io.BytesIO()
             f.write(buf)
-            left = win32.PeekNamedPipe(self._handle)[1]
+            left = _winapi.PeekNamedPipe(self._handle)[1]
             assert left > 0
             if maxsize is not None and len(buf) + left > maxsize:
                 self._bad_message_length()
-            ov, err = win32.ReadFile(self._handle, left, overlapped=True)
+            ov, err = _winapi.ReadFile(self._handle, left, overlapped=True)
             rbytes, err = ov.GetOverlappedResult(True)
             assert err == 0
             assert rbytes == left
@@ -372,11 +372,11 @@ class Connection(_ConnectionBase):
     a socket handle (Windows).
     """
 
-    if win32:
-        def _close(self, _close=win32.closesocket):
+    if _winapi:
+        def _close(self, _close=_multiprocessing.closesocket):
             _close(self._handle)
-        _write = win32.send
-        _read = win32.recv
+        _write = _multiprocessing.send
+        _read = _multiprocessing.recv
     else:
         def _close(self, _close=os.close):
             _close(self._handle)
@@ -526,30 +526,30 @@ def Pipe(duplex=True):
         '''
         address = arbitrary_address('AF_PIPE')
         if duplex:
-            openmode = win32.PIPE_ACCESS_DUPLEX
-            access = win32.GENERIC_READ | win32.GENERIC_WRITE
+            openmode = _winapi.PIPE_ACCESS_DUPLEX
+            access = _winapi.GENERIC_READ | _winapi.GENERIC_WRITE
             obsize, ibsize = BUFSIZE, BUFSIZE
         else:
-            openmode = win32.PIPE_ACCESS_INBOUND
-            access = win32.GENERIC_WRITE
+            openmode = _winapi.PIPE_ACCESS_INBOUND
+            access = _winapi.GENERIC_WRITE
             obsize, ibsize = 0, BUFSIZE
 
-        h1 = win32.CreateNamedPipe(
-            address, openmode | win32.FILE_FLAG_OVERLAPPED |
-            win32.FILE_FLAG_FIRST_PIPE_INSTANCE,
-            win32.PIPE_TYPE_MESSAGE | win32.PIPE_READMODE_MESSAGE |
-            win32.PIPE_WAIT,
-            1, obsize, ibsize, win32.NMPWAIT_WAIT_FOREVER, win32.NULL
+        h1 = _winapi.CreateNamedPipe(
+            address, openmode | _winapi.FILE_FLAG_OVERLAPPED |
+            _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE,
+            _winapi.PIPE_TYPE_MESSAGE | _winapi.PIPE_READMODE_MESSAGE |
+            _winapi.PIPE_WAIT,
+            1, obsize, ibsize, _winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL
             )
-        h2 = win32.CreateFile(
-            address, access, 0, win32.NULL, win32.OPEN_EXISTING,
-            win32.FILE_FLAG_OVERLAPPED, win32.NULL
+        h2 = _winapi.CreateFile(
+            address, access, 0, _winapi.NULL, _winapi.OPEN_EXISTING,
+            _winapi.FILE_FLAG_OVERLAPPED, _winapi.NULL
             )
-        win32.SetNamedPipeHandleState(
-            h2, win32.PIPE_READMODE_MESSAGE, None, None
+        _winapi.SetNamedPipeHandleState(
+            h2, _winapi.PIPE_READMODE_MESSAGE, None, None
             )
 
-        overlapped = win32.ConnectNamedPipe(h1, overlapped=True)
+        overlapped = _winapi.ConnectNamedPipe(h1, overlapped=True)
         _, err = overlapped.GetOverlappedResult(True)
         assert err == 0
 
@@ -630,26 +630,26 @@ def __init__(self, address, backlog=None):
                 )
 
         def _new_handle(self, first=False):
-            flags = win32.PIPE_ACCESS_DUPLEX | win32.FILE_FLAG_OVERLAPPED
+            flags = _winapi.PIPE_ACCESS_DUPLEX | _winapi.FILE_FLAG_OVERLAPPED
             if first:
-                flags |= win32.FILE_FLAG_FIRST_PIPE_INSTANCE
-            return win32.CreateNamedPipe(
+                flags |= _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE
+            return _winapi.CreateNamedPipe(
                 self._address, flags,
-                win32.PIPE_TYPE_MESSAGE | win32.PIPE_READMODE_MESSAGE |
-                win32.PIPE_WAIT,
-                win32.PIPE_UNLIMITED_INSTANCES, BUFSIZE, BUFSIZE,
-                win32.NMPWAIT_WAIT_FOREVER, win32.NULL
+                _winapi.PIPE_TYPE_MESSAGE | _winapi.PIPE_READMODE_MESSAGE |
+                _winapi.PIPE_WAIT,
+                _winapi.PIPE_UNLIMITED_INSTANCES, BUFSIZE, BUFSIZE,
+                _winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL
                 )
 
         def accept(self):
             self._handle_queue.append(self._new_handle())
             handle = self._handle_queue.pop(0)
-            ov = win32.ConnectNamedPipe(handle, overlapped=True)
+            ov = _winapi.ConnectNamedPipe(handle, overlapped=True)
             try:
-                res = win32.WaitForMultipleObjects([ov.event], False, INFINITE)
+                res = _winapi.WaitForMultipleObjects([ov.event], False, INFINITE)
             except:
                 ov.cancel()
-                win32.CloseHandle(handle)
+                _winapi.CloseHandle(handle)
                 raise
             finally:
                 _, err = ov.GetOverlappedResult(True)
@@ -660,7 +660,7 @@ def accept(self):
         def _finalize_pipe_listener(queue, address):
             sub_debug('closing listener with address=%r', address)
             for handle in queue:
-                win32.CloseHandle(handle)
+                _winapi.CloseHandle(handle)
 
     def PipeClient(address):
         '''
@@ -669,23 +669,23 @@ def PipeClient(address):
         t = _init_timeout()
         while 1:
             try:
-                win32.WaitNamedPipe(address, 1000)
-                h = win32.CreateFile(
-                    address, win32.GENERIC_READ | win32.GENERIC_WRITE,
-                    0, win32.NULL, win32.OPEN_EXISTING,
-                    win32.FILE_FLAG_OVERLAPPED, win32.NULL
+                _winapi.WaitNamedPipe(address, 1000)
+                h = _winapi.CreateFile(
+                    address, _winapi.GENERIC_READ | _winapi.GENERIC_WRITE,
+                    0, _winapi.NULL, _winapi.OPEN_EXISTING,
+                    _winapi.FILE_FLAG_OVERLAPPED, _winapi.NULL
                     )
             except WindowsError as e:
-                if e.winerror not in (win32.ERROR_SEM_TIMEOUT,
-                                      win32.ERROR_PIPE_BUSY) or _check_timeout(t):
+                if e.winerror not in (_winapi.ERROR_SEM_TIMEOUT,
+                                      _winapi.ERROR_PIPE_BUSY) or _check_timeout(t):
                     raise
             else:
                 break
         else:
             raise
 
-        win32.SetNamedPipeHandleState(
-            h, win32.PIPE_READMODE_MESSAGE, None, None
+        _winapi.SetNamedPipeHandleState(
+            h, _winapi.PIPE_READMODE_MESSAGE, None, None
             )
         return PipeConnection(h)
 
@@ -774,7 +774,7 @@ def _exhaustive_wait(handles, timeout):
         L = list(handles)
         ready = []
         while L:
-            res = win32.WaitForMultipleObjects(L, False, timeout)
+            res = _winapi.WaitForMultipleObjects(L, False, timeout)
             if res == WAIT_TIMEOUT:
                 break
             elif WAIT_OBJECT_0 <= res < WAIT_OBJECT_0 + len(L):
@@ -788,7 +788,7 @@ def _exhaustive_wait(handles, timeout):
             timeout = 0
         return ready
 
-    _ready_errors = {win32.ERROR_BROKEN_PIPE, win32.ERROR_NETNAME_DELETED}
+    _ready_errors = {_winapi.ERROR_BROKEN_PIPE, _winapi.ERROR_NETNAME_DELETED}
 
     def wait(object_list, timeout=None):
         '''
@@ -818,12 +818,12 @@ def wait(object_list, timeout=None):
                 else:
                     # start an overlapped read of length zero
                     try:
-                        ov, err = win32.ReadFile(fileno(), 0, True)
+                        ov, err = _winapi.ReadFile(fileno(), 0, True)
                     except OSError as e:
                         err = e.winerror
                         if err not in _ready_errors:
                             raise
-                    if err == win32.ERROR_IO_PENDING:
+                    if err == _winapi.ERROR_IO_PENDING:
                         ov_list.append(ov)
                         waithandle_to_obj[ov.event] = o
                     else:
@@ -847,7 +847,7 @@ def wait(object_list, timeout=None):
                     err = e.winerror
                     if err not in _ready_errors:
                         raise
-                if err != win32.ERROR_OPERATION_ABORTED:
+                if err != _winapi.ERROR_OPERATION_ABORTED:
                     o = waithandle_to_obj[ov.event]
                     ready_objects.add(o)
                     if err == 0:

Lib/multiprocessing/forking.py

@@ -181,10 +181,9 @@ def thread_is_spawning():
 else:
     import _thread
     import msvcrt
-    import _subprocess
+    import _winapi
 
     from pickle import load, HIGHEST_PROTOCOL
-    from _multiprocessing import win32
 
     def dump(obj, file, protocol=None):
         ForkingPickler(file, protocol).dump(obj)
@@ -197,8 +196,8 @@ def dump(obj, file, protocol=None):
     WINEXE = (sys.platform == 'win32' and getattr(sys, 'frozen', False))
     WINSERVICE = sys.executable.lower().endswith("pythonservice.exe")
 
-    exit = win32.ExitProcess
-    close = win32.CloseHandle
+    exit = _winapi.ExitProcess
+    close = _winapi.CloseHandle
 
     #
     # _python_exe is the assumed path to the python executable.
@@ -220,11 +219,11 @@ def set_executable(exe):
 
     def duplicate(handle, target_process=None, inheritable=False):
         if target_process is None:
-            target_process = _subprocess.GetCurrentProcess()
-        return _subprocess.DuplicateHandle(
-            _subprocess.GetCurrentProcess(), handle, target_process,
-            0, inheritable, _subprocess.DUPLICATE_SAME_ACCESS
-            ).Detach()
+            target_process = _winapi.GetCurrentProcess()
+        return _winapi.DuplicateHandle(
+            _winapi.GetCurrentProcess(), handle, target_process,
+            0, inheritable, _winapi.DUPLICATE_SAME_ACCESS
+            )
 
     #
     # We define a Popen class similar to the one from subprocess, but
@@ -248,10 +247,10 @@ def __init__(self, process_obj):
             # start process
             cmd = get_command_line() + [rhandle]
             cmd = ' '.join('"%s"' % x for x in cmd)
-            hp, ht, pid, tid = _subprocess.CreateProcess(
+            hp, ht, pid, tid = _winapi.CreateProcess(
                 _python_exe, cmd, None, None, 1, 0, None, None, None
                 )
-            ht.Close()
+            _winapi.CloseHandle(ht)
             close(rhandle)
 
             # set attributes of self
@@ -282,13 +281,13 @@ def duplicate_for_child(handle):
         def wait(self, timeout=None):
             if self.returncode is None:
                 if timeout is None:
-                    msecs = _subprocess.INFINITE
+                    msecs = _winapi.INFINITE
                 else:
                     msecs = max(0, int(timeout * 1000 + 0.5))
 
-                res = _subprocess.WaitForSingleObject(int(self._handle), msecs)
-                if res == _subprocess.WAIT_OBJECT_0:
-                    code = _subprocess.GetExitCodeProcess(self._handle)
+                res = _winapi.WaitForSingleObject(int(self._handle), msecs)
+                if res == _winapi.WAIT_OBJECT_0:
+                    code = _winapi.GetExitCodeProcess(self._handle)
                     if code == TERMINATE:
                         code = -signal.SIGTERM
                     self.returncode = code
@@ -301,7 +300,7 @@ def poll(self):
         def terminate(self):
             if self.returncode is None:
                 try:
-                    _subprocess.TerminateProcess(int(self._handle), TERMINATE)
+                    _winapi.TerminateProcess(int(self._handle), TERMINATE)
                 except WindowsError:
                     if self.wait(timeout=0.1) is None:
                         raise

Lib/multiprocessing/heap.py

@@ -51,7 +51,7 @@
 
 if sys.platform == 'win32':
 
-    from _multiprocessing import win32
+    import _winapi
 
     class Arena(object):
 
@@ -61,7 +61,7 @@ def __init__(self, size):
             self.size = size
             self.name = 'pym-%d-%d' % (os.getpid(), next(Arena._counter))
             self.buffer = mmap.mmap(-1, self.size, tagname=self.name)
-            assert win32.GetLastError() == 0, 'tagname already in use'
+            assert _winapi.GetLastError() == 0, 'tagname already in use'
             self._state = (self.size, self.name)
 
         def __getstate__(self):
@@ -71,7 +71,7 @@ def __getstate__(self):
         def __setstate__(self, state):
             self.size, self.name = self._state = state
             self.buffer = mmap.mmap(-1, self.size, tagname=self.name)
-            assert win32.GetLastError() == win32.ERROR_ALREADY_EXISTS
+            assert _winapi.GetLastError() == _winapi.ERROR_ALREADY_EXISTS
 
 else: