diff options
author | Antoine Pitrou <solipsis@pitrou.net> | 2012-04-18 18:51:15 (GMT) |
---|---|---|
committer | Antoine Pitrou <solipsis@pitrou.net> | 2012-04-18 18:51:15 (GMT) |
commit | 23bba4ca398ff2e096413e3bbaf9a6cdaa1d1846 (patch) | |
tree | a9818d43b30d0d94d124ec15d45f01b4db36f5b7 /Lib/multiprocessing | |
parent | c51b7fd65b8c7476180c965d48390431b2d558e6 (diff) | |
download | cpython-23bba4ca398ff2e096413e3bbaf9a6cdaa1d1846.zip cpython-23bba4ca398ff2e096413e3bbaf9a6cdaa1d1846.tar.gz cpython-23bba4ca398ff2e096413e3bbaf9a6cdaa1d1846.tar.bz2 |
Issue #11750: The Windows API functions scattered in the _subprocess and
_multiprocessing.win32 modules now live in a single module "_winapi".
Patch by sbt.
Diffstat (limited to 'Lib/multiprocessing')
-rw-r--r-- | Lib/multiprocessing/connection.py | 124 | ||||
-rw-r--r-- | Lib/multiprocessing/forking.py | 31 | ||||
-rw-r--r-- | Lib/multiprocessing/heap.py | 6 | ||||
-rw-r--r-- | Lib/multiprocessing/reduction.py | 6 |
4 files changed, 83 insertions, 84 deletions
diff --git a/Lib/multiprocessing/connection.py b/Lib/multiprocessing/connection.py index 90c1ea7..3a61e5e 100644 --- a/Lib/multiprocessing/connection.py +++ b/Lib/multiprocessing/connection.py @@ -51,12 +51,12 @@ from multiprocessing import current_process, AuthenticationError, BufferTooShort 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 @@ class _ConnectionBase: return self._poll(timeout) -if win32: +if _winapi: class PipeConnection(_ConnectionBase): """ @@ -292,14 +292,14 @@ if win32: """ _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 @@ if win32: 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,10 +333,10 @@ if win32: 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 @@ -344,7 +344,7 @@ if win32: 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 @@ if win32: 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 @@ else: ''' 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 @@ if sys.platform == 'win32': ) 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 @@ if sys.platform == 'win32': 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 @@ if sys.platform == 'win32': 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 @@ if sys.platform == 'win32': 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 @@ if sys.platform == 'win32': 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 @@ if sys.platform == 'win32': 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 @@ if sys.platform == 'win32': 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: diff --git a/Lib/multiprocessing/forking.py b/Lib/multiprocessing/forking.py index 020508a..0cbb741 100644 --- a/Lib/multiprocessing/forking.py +++ b/Lib/multiprocessing/forking.py @@ -181,10 +181,9 @@ if sys.platform != 'win32': 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 @@ else: 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 @@ else: 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 @@ else: # 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 @@ else: 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 @@ else: 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 diff --git a/Lib/multiprocessing/heap.py b/Lib/multiprocessing/heap.py index 7366bd2..7e19434 100644 --- a/Lib/multiprocessing/heap.py +++ b/Lib/multiprocessing/heap.py @@ -51,7 +51,7 @@ __all__ = ['BufferWrapper'] if sys.platform == 'win32': - from _multiprocessing import win32 + import _winapi class Arena(object): @@ -61,7 +61,7 @@ if sys.platform == 'win32': 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 @@ if sys.platform == 'win32': 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: diff --git a/Lib/multiprocessing/reduction.py b/Lib/multiprocessing/reduction.py index dda4a41..c80de59 100644 --- a/Lib/multiprocessing/reduction.py +++ b/Lib/multiprocessing/reduction.py @@ -60,11 +60,11 @@ if not(sys.platform == 'win32' or (hasattr(socket, 'CMSG_LEN') and # if sys.platform == 'win32': - from _multiprocessing import win32 + import _winapi def send_handle(conn, handle, destination_pid): - process_handle = win32.OpenProcess( - win32.PROCESS_ALL_ACCESS, False, destination_pid + process_handle = _winapi.OpenProcess( + _winapi.PROCESS_ALL_ACCESS, False, destination_pid ) try: new_handle = duplicate(handle, process_handle) |