diff --git a/Lib/concurrent/futures/__init__.py b/Lib/concurrent/futures/__init__.py index d6ac4b3e0..5d53a5ac5 100644 --- a/Lib/concurrent/futures/__init__.py +++ b/Lib/concurrent/futures/__init__.py @@ -17,6 +17,9 @@ from concurrent.futures._base import (FIRST_COMPLETED, wait, as_completed) +lazy from .process import ProcessPoolExecutor +lazy from .thread import ThreadPoolExecutor + __all__ = [ 'FIRST_COMPLETED', 'FIRST_EXCEPTION', @@ -40,26 +43,9 @@ except ImportError: _interpreters = None if _interpreters: + lazy from .interpreter import InterpreterPoolExecutor # noqa: F401 __all__.append('InterpreterPoolExecutor') def __dir__(): return __all__ + ['__author__', '__doc__'] - - -def __getattr__(name): - global ProcessPoolExecutor, ThreadPoolExecutor, InterpreterPoolExecutor - - if name == 'ProcessPoolExecutor': - from .process import ProcessPoolExecutor - return ProcessPoolExecutor - - if name == 'ThreadPoolExecutor': - from .thread import ThreadPoolExecutor - return ThreadPoolExecutor - - if _interpreters and name == 'InterpreterPoolExecutor': - from .interpreter import InterpreterPoolExecutor - return InterpreterPoolExecutor - - raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/Lib/concurrent/futures/process.py b/Lib/concurrent/futures/process.py index 122be5827..51df68540 100644 --- a/Lib/concurrent/futures/process.py +++ b/Lib/concurrent/futures/process.py @@ -119,10 +119,11 @@ EXTRA_QUEUED_CALLS = 1 # On Windows, WaitForMultipleObjects is used to wait for processes to finish. -# It can wait on, at most, 63 objects. There is an overhead of two objects: +# It can wait on, at most, 64 objects. There is an overhead of three objects: # - the result queue reader # - the thread wakeup reader -_MAX_WINDOWS_WORKERS = 63 - 2 +# - the SIGINT handler +_MAX_WINDOWS_WORKERS = 64 - 3 # Hack to embed stringification of remote traceback in local traceback @@ -269,6 +270,20 @@ def _process_worker(call_queue, result_queue, initializer, initargs, max_tasks=N return +def _spawn_worker(mp_context, call_queue, result_queue, initializer, + initargs, max_tasks_per_child, processes): + """Start one worker process and record it in *processes* by pid.""" + p = mp_context.Process( + target=_process_worker, + args=(call_queue, + result_queue, + initializer, + initargs, + max_tasks_per_child)) + p.start() + processes[p.pid] = p + + class _ExecutorManagerThread(threading.Thread): """Manages the communication between this process and the worker processes. @@ -321,6 +336,15 @@ class _ExecutorManagerThread(threading.Thread): # exiting safely self.max_tasks_per_child = executor._max_tasks_per_child + # gh-119592: Needed to size worker replacement, and immutable, so + # keep a copy rather than reading it back through the executor + # weakref. The rest of the spawn configuration is deliberately NOT + # copied here: holding user-provided objects (initializer, + # initargs, mp_context) in this always-reachable running thread + # could keep the executor itself reachable through them, breaking + # garbage-collection-triggered shutdown. + self.max_workers = executor._max_workers + # A dict mapping work ids to _WorkItems e.g. # {5: <_WorkItem...>, 6: <_WorkItem...>, ...} self.pending_work_items = executor._pending_work_items @@ -357,12 +381,14 @@ class _ExecutorManagerThread(threading.Thread): # while waiting on new results. del result_item - if executor := self.executor_reference(): - if process_exited: - with self.shutdown_lock: - executor._adjust_process_count() - else: - executor._idle_worker_semaphore.release() + if process_exited: + with self.shutdown_lock: + broken = self._replace_dead_worker() + if broken is not None: + self.terminate_broken(*broken) + return + elif executor := self.executor_reference(): + executor._idle_worker_semaphore.release() del executor if self.is_shutting_down(): @@ -379,6 +405,71 @@ class _ExecutorManagerThread(threading.Thread): self.join_executor_internals() return + def _replace_dead_worker(self): + """Spawn a replacement for a worker that exited at its + max_tasks_per_child limit. Called under self.shutdown_lock. + + Returns None while the pool can still make progress, otherwise a + (cause, message) tuple describing why the remaining work items can + never run, so that run() can fail their futures. + """ + assert self.shutdown_lock.locked() + cause = None + message = None + executor = self.executor_reference() + if executor is None: + # gh-152967: The executor was garbage collected; nothing can + # spawn a replacement worker for it anymore. + message = ("The ProcessPoolExecutor was garbage collected with " + "work pending after its last worker process exited " + "upon reaching max_tasks_per_child; the pending work " + "can never be run.") + elif executor._force_shutting_down: + # terminate_workers()/kill_workers() is tearing the pool down; + # a replacement worker would escape the kill and run work + # items that were enqueued before it. + message = ("A worker process exited while the pool was being " + "forcefully shut down; work that was still enqueued " + "will not be run.") + elif self.pending_work_items or not self.is_shutting_down(): + # gh-115634: Do not consult the executor's + # _idle_worker_semaphore here: it counts task completions, not + # idle workers, so it can hold a stale token released by the + # now-dead worker. Trusting such a token would leave the pool + # a worker short, deadlocking once all workers reach their + # task limit. Spawning from this (manager) thread is safe + # despite gh-90622 because max_tasks_per_child is rejected for + # the "fork" start method. + if len(self.processes) < self.max_workers: + # gh-119592: Spawn using state owned by this thread and + # configuration read through the live weakref (which + # shutdown() never clears), not the executor state that + # shutdown(wait=False) clears concurrently. + try: + _spawn_worker(executor._mp_context, self.call_queue, + self.result_queue, executor._initializer, + executor._initargs, + self.max_tasks_per_child, self.processes) + except Exception as exc: + # While other workers remain the pool has merely lost + # capacity and they keep draining the queue; with none + # left the failure is reported below. + cause = format_exception(exc) + message = ("A replacement worker process could not be " + "started, leaving the pool without workers " + "to run the remaining work.") + del executor + + if not self.processes and (self.pending_work_items + or cause is not None): + # No worker processes remain and no replacement can be + # spawned: any remaining work items can never run. A spawn + # failure breaks the pool even with nothing pending; leaving + # a zero-worker pool alive would hang a later submit() on a + # stale _idle_worker_semaphore token instead of raising. + return (cause, message) + return None + def add_call_item_to_queue(self): # Fills call_queue with _WorkItems from pending_work_items. # This function never blocks. @@ -455,10 +546,11 @@ class _ExecutorManagerThread(threading.Thread): return (_global_shutdown or executor is None or executor._shutdown_thread) - def _terminate_broken(self, cause): + def _terminate_broken(self, cause, bpe_message=None): # Terminate the executor because it is in a broken state. The cause # argument can be used to display more information on the error that - # lead the executor into becoming broken. + # lead the executor into becoming broken. bpe_message overrides the + # default message on the BrokenProcessPool set on pending futures. # Mark the process pool broken so that submits fail right now. executor = self.executor_reference() @@ -469,17 +561,34 @@ class _ExecutorManagerThread(threading.Thread): executor._shutdown_thread = True executor = None - # All pending tasks are to be marked failed with the following - # BrokenProcessPool error - bpe = BrokenProcessPool("A process in the process pool was " - "terminated abruptly while the future was " - "running or pending.") + # All pending tasks are to be marked failed with a + # BrokenProcessPool error, as separate instances to avoid sharing + # a traceback (gh-101267). + cause_str = None if cause is not None: - bpe.__cause__ = _RemoteTraceback( - f"\n'''\n{''.join(cause)}'''") + cause_str = ''.join(cause) + else: + # No cause known, so report any processes that have + # terminated with nonzero exit codes, e.g. from a + # segfault. Multiple may terminate simultaneously, + # so include all of them in the traceback. + errors = [] + for p in self.processes.values(): + if p.exitcode is not None and p.exitcode != 0: + errors.append(f"Process {p.pid} terminated abruptly " + f"with exit code {p.exitcode}") + if errors: + cause_str = "\n".join(errors) + cause_tb = f"\n'''\n{cause_str}'''" if cause_str else None + if bpe_message is None: + bpe_message = ("A process in the process pool was terminated " + "abruptly while the future was running or pending.") # Mark pending tasks as failed. for work_id, work_item in self.pending_work_items.items(): + bpe = BrokenProcessPool(bpe_message) + if cause_tb is not None: + bpe.__cause__ = _RemoteTraceback(cause_tb) try: work_item.future.set_exception(bpe) except _base.InvalidStateError: @@ -502,9 +611,9 @@ class _ExecutorManagerThread(threading.Thread): # clean up resources self._join_executor_internals(broken=True) - def terminate_broken(self, cause): + def terminate_broken(self, cause, bpe_message=None): with self.shutdown_lock: - self._terminate_broken(cause) + self._terminate_broken(cause, bpe_message) def flag_executor_shutting_down(self): # Flag the executor as shutting down and cancel remaining tasks if @@ -717,6 +826,7 @@ class ProcessPoolExecutor(_base.Executor): self._queue_count = 0 self._pending_work_items = {} self._cancel_pending_futures = False + self._force_shutting_down = False # _ThreadWakeup is a communication channel used to interrupt the wait # of the main loop of executor_manager_thread from another thread (e.g. @@ -762,18 +872,23 @@ class ProcessPoolExecutor(_base.Executor): if self._processes is None: return + # gh-152967: A forceful shutdown is in progress; a worker spawned + # here could escape its process snapshot and keep running work. + if self._force_shutting_down: + return + # if there's an idle process, we don't need to spawn a new one. if self._idle_worker_semaphore.acquire(blocking=False): return process_count = len(self._processes) if process_count < self._max_workers: - # Assertion disabled as this codepath is also used to replace a - # worker that unexpectedly dies, even when using the 'fork' start - # method. That means there is still a potential deadlock bug. If a - # 'fork' mp_context worker dies, we'll be forking a new one when - # we know a thread is running (self._executor_manager_thread). - #assert self._safe_to_dynamically_spawn_children or not self._executor_manager_thread, 'https://github.com/python/cpython/issues/90622' + # gh-90622: spawning a child via fork while another thread is + # running can deadlock in the child. submit() only calls this + # method when using a non-fork start method. + assert (self._safe_to_dynamically_spawn_children + or not self._executor_manager_thread), ( + 'https://github.com/python/cpython/issues/90622') self._spawn_process() def _launch_processes(self): @@ -785,15 +900,10 @@ class ProcessPoolExecutor(_base.Executor): self._spawn_process() def _spawn_process(self): - p = self._mp_context.Process( - target=_process_worker, - args=(self._call_queue, - self._result_queue, - self._initializer, - self._initargs, - self._max_tasks_per_child)) - p.start() - self._processes[p.pid] = p + _spawn_worker(self._mp_context, self._call_queue, + self._result_queue, self._initializer, + self._initargs, self._max_tasks_per_child, + self._processes) def submit(self, fn, /, *args, **kwargs): with self._shutdown_lock: @@ -890,6 +1000,14 @@ class ProcessPoolExecutor(_base.Executor): if operation not in _SHUTDOWN_CALLBACK_OPERATION: raise ValueError(f"Unsupported operation: {operation!r}") + # gh-152967: Stop the manager thread from spawning replacement + # workers before we copy the processes to signal: a worker spawned + # after the copy would survive the loop below and run enqueued + # work items. Taking the lock orders this against the manager's + # worker replacement, which runs under the same lock. + with self._shutdown_lock: + self._force_shutting_down = True + processes = {} if self._processes: processes = self._processes.copy() diff --git a/Lib/multiprocessing/connection.py b/Lib/multiprocessing/connection.py index f577186e9..e37ec07d7 100644 --- a/Lib/multiprocessing/connection.py +++ b/Lib/multiprocessing/connection.py @@ -734,8 +734,7 @@ if sys.platform == 'win32': # written data and then disconnected -- see Issue 14725. else: try: - res = _winapi.WaitForMultipleObjects( - [ov.event], False, INFINITE) + _winapi.WaitForMultipleObjects([ov.event], False, INFINITE) except: ov.cancel() _winapi.CloseHandle(handle) @@ -1089,14 +1088,22 @@ if sys.platform == 'win32': Returns list of those objects in object_list which are ready/readable. ''' + object_list = list(object_list) + + if not object_list: + if timeout is None: + while True: + time.sleep(1e6) + elif timeout > 0: + time.sleep(timeout) + return [] + if timeout is None: timeout = INFINITE elif timeout < 0: timeout = 0 else: timeout = int(timeout * 1000 + 0.5) - - object_list = list(object_list) waithandle_to_obj = {} ov_list = [] ready_objects = set() diff --git a/Lib/multiprocessing/context.py b/Lib/multiprocessing/context.py index 5fa6d7e48..e1d251456 100644 --- a/Lib/multiprocessing/context.py +++ b/Lib/multiprocessing/context.py @@ -183,12 +183,15 @@ class BaseContext(object): from .spawn import set_executable set_executable(executable) - def set_forkserver_preload(self, module_names): + def set_forkserver_preload(self, module_names, *, on_error='ignore'): '''Set list of module names to try to load in forkserver process. - This is really just a hint. + + The on_error parameter controls how import failures are handled: + "ignore" (default) silently ignores failures, "warn" emits warnings, + and "fail" raises exceptions breaking the forkserver context. ''' from .forkserver import set_forkserver_preload - set_forkserver_preload(module_names) + set_forkserver_preload(module_names, on_error=on_error) def get_context(self, method=None): if method is None: diff --git a/Lib/multiprocessing/forkserver.py b/Lib/multiprocessing/forkserver.py index e431b3f12..b204a3ad0 100644 --- a/Lib/multiprocessing/forkserver.py +++ b/Lib/multiprocessing/forkserver.py @@ -42,6 +42,7 @@ class ForkServer(object): self._inherited_fds = None self._lock = threading.Lock() self._preload_modules = ['__main__'] + self._preload_on_error = 'ignore' def _stop(self): # Method used by unit tests to stop the server @@ -64,11 +65,22 @@ class ForkServer(object): self._forkserver_address = None self._forkserver_authkey = None - def set_forkserver_preload(self, modules_names): - '''Set list of module names to try to load in forkserver process.''' + def set_forkserver_preload(self, modules_names, *, on_error='ignore'): + '''Set list of module names to try to load in forkserver process. + + The on_error parameter controls how import failures are handled: + "ignore" (default) silently ignores failures, "warn" emits warnings, + and "fail" raises exceptions breaking the forkserver context. + ''' if not all(type(mod) is str for mod in modules_names): raise TypeError('module_names must be a list of strings') + if on_error not in ('ignore', 'warn', 'fail'): + raise ValueError( + f"on_error must be 'ignore', 'warn', or 'fail', " + f"not {on_error!r}" + ) self._preload_modules = modules_names + self._preload_on_error = on_error def get_inherited_fds(self): '''Return list of fds inherited from parent process. @@ -107,6 +119,14 @@ class ForkServer(object): wrapped_client, self._forkserver_authkey) connection.deliver_challenge( wrapped_client, self._forkserver_authkey) + except (EOFError, ConnectionError, BrokenPipeError) as exc: + if (self._preload_modules and + self._preload_on_error == 'fail'): + exc.add_note( + "Forkserver process may have crashed during module " + "preloading. Check stderr." + ) + raise finally: wrapped_client._detach() del wrapped_client @@ -161,6 +181,8 @@ class ForkServer(object): main_kws['main_path'] = data['init_main_from_path'] if 'sys_argv' in data: sys_argv = data['sys_argv'] + if self._preload_on_error != 'ignore': + main_kws['on_error'] = self._preload_on_error with socket.socket(socket.AF_UNIX) as listener: address = connection.arbitrary_address('AF_UNIX') @@ -207,8 +229,69 @@ class ForkServer(object): # # +def _handle_import_error(on_error, modinfo, exc, *, warn_stacklevel): + """Handle an import error according to the on_error policy.""" + match on_error: + case 'fail': + raise + case 'warn': + warnings.warn( + f"Failed to preload {modinfo}: {exc}", + ImportWarning, + stacklevel=warn_stacklevel + 1 + ) + case 'ignore': + pass + + +def _handle_preload(preload, main_path=None, sys_path=None, sys_argv=None, + on_error='ignore'): + """Handle module preloading with configurable error handling. + + Args: + preload: List of module names to preload. + main_path: Path to __main__ module if '__main__' is in preload. + sys_path: sys.path to use for imports (None means use current). + sys_argv: sys.argv to use (None means use current). + on_error: How to handle import errors ("ignore", "warn", or "fail"). + """ + if not preload: + return + + if sys_argv is not None: + sys.argv[:] = sys_argv + if sys_path is not None: + sys.path[:] = sys_path + + if '__main__' in preload and main_path is not None: + process.current_process()._inheriting = True + try: + spawn.import_main_path(main_path) + except Exception as e: + # Catch broad Exception because import_main_path() uses + # runpy.run_path() which executes the script and can raise + # any exception, not just ImportError + _handle_import_error( + on_error, f"__main__ from {main_path!r}", e, warn_stacklevel=2 + ) + finally: + del process.current_process()._inheriting + + for modname in preload: + try: + __import__(modname) + except ImportError as e: + _handle_import_error( + on_error, f"module {modname!r}", e, warn_stacklevel=2 + ) + + # gh-135335: flush stdout/stderr in case any of the preloaded modules + # wrote to them, otherwise children might inherit buffered data + util._flush_std_streams() + + def main(listener_fd, alive_r, preload, main_path=None, sys_path=None, - *, sys_argv=None, authkey_r=None): + *, sys_argv=None, authkey_r=None, on_error='ignore'): """Run forkserver.""" if authkey_r is not None: try: @@ -219,26 +302,7 @@ def main(listener_fd, alive_r, preload, main_path=None, sys_path=None, else: authkey = b'' - if preload: - if sys_argv is not None: - sys.argv[:] = sys_argv - if sys_path is not None: - sys.path[:] = sys_path - if '__main__' in preload and main_path is not None: - process.current_process()._inheriting = True - try: - spawn.import_main_path(main_path) - finally: - del process.current_process()._inheriting - for modname in preload: - try: - __import__(modname) - except ImportError: - pass - - # gh-135335: flush stdout/stderr in case any of the preloaded modules - # wrote to them, otherwise children might inherit buffered data - util._flush_std_streams() + _handle_preload(preload, main_path, sys_path, sys_argv, on_error) util._close_stdin() diff --git a/Lib/multiprocessing/heap.py b/Lib/multiprocessing/heap.py index 6217dfe12..5c8356483 100644 --- a/Lib/multiprocessing/heap.py +++ b/Lib/multiprocessing/heap.py @@ -324,10 +324,6 @@ class BufferWrapper(object): _heap = Heap() def __init__(self, size): - if size < 0: - raise ValueError("Size {0:n} out of range".format(size)) - if sys.maxsize <= size: - raise OverflowError("Size {0:n} too large".format(size)) block = BufferWrapper._heap.malloc(size) self._state = (block, size) util.Finalize(self, BufferWrapper._heap.free, args=(block,)) diff --git a/Lib/multiprocessing/managers.py b/Lib/multiprocessing/managers.py index 91bcf243e..11c6e21de 100644 --- a/Lib/multiprocessing/managers.py +++ b/Lib/multiprocessing/managers.py @@ -1294,9 +1294,9 @@ if HAS_SHMEM: class _SharedMemoryTracker: "Manages one or more shared memory segments." - def __init__(self, name, segment_names=[]): + def __init__(self, name, segment_names=None): self.shared_memory_context_name = name - self.segment_names = segment_names + self.segment_names = [] if segment_names is None else segment_names def register_segment(self, segment_name): "Adds the supplied shared memory block name to tracker." diff --git a/Lib/multiprocessing/resource_tracker.py b/Lib/multiprocessing/resource_tracker.py index dfb5a3dfe..d3328a8c6 100644 --- a/Lib/multiprocessing/resource_tracker.py +++ b/Lib/multiprocessing/resource_tracker.py @@ -74,7 +74,7 @@ class ResourceTracker(object): # Filenames not supported by the simple format will always be sent # using JSON. # The reader should understand all formats. - self._use_simple_format = True + self._use_simple_format = False # Set to True by _stop_locked() if the waitpid polling loop ran to # its timeout without reaping the tracker. Exposed for tests. diff --git a/Lib/multiprocessing/sharedctypes.py b/Lib/multiprocessing/sharedctypes.py index 607170702..eee1172e6 100644 --- a/Lib/multiprocessing/sharedctypes.py +++ b/Lib/multiprocessing/sharedctypes.py @@ -37,7 +37,12 @@ typecode_to_type = { # def _new_value(type_): - size = ctypes.sizeof(type_) + try: + size = ctypes.sizeof(type_) + except TypeError as e: + raise TypeError("bad typecode (must be a ctypes type or one of " + "c, b, B, u, h, H, i, I, l, L, q, Q, f or d)") from e + wrapper = heap.BufferWrapper(size) return rebuild_ctype(type_, wrapper, None) diff --git a/Lib/subprocess.py b/Lib/subprocess.py index 52b7b7117..5948cd841 100644 --- a/Lib/subprocess.py +++ b/Lib/subprocess.py @@ -250,6 +250,82 @@ else: else: _PopenSelector = selectors.SelectSelector + def _communicate_io_posix(selector, stdin, input_view, input_offset, + output_buffers, endtime, *, close_on_eof=False): + """ + Low-level POSIX I/O multiplexing loop used by Popen._communicate. + + Handles the select loop for reading/writing but does not manage + stream lifecycle or raise timeout exceptions. + + Args: + selector: A _PopenSelector with streams already registered + stdin: Writable file object for input, or None + input_view: memoryview of input bytes, or None + input_offset: Starting offset into input_view (for resume support) + output_buffers: Dict {file_object: list} to append read chunks to + endtime: Deadline timestamp, or None for no timeout + close_on_eof: If True, close output streams immediately when they + EOF rather than leaving them open for the caller to close. + Used by Popen._communicate() to match its historical behavior + of releasing fds as soon as the child closes the corresponding + pipe. + + Returns: + (new_input_offset, completed) + - new_input_offset: How many bytes of input were written + - completed: True if all I/O finished, False if timed out + + Note: + - Closes output streams on EOF only if close_on_eof=True + - Does NOT raise TimeoutExpired (caller handles) + - Appends to output_buffers lists in place + """ + stdin_fd = stdin.fileno() if stdin else None + + while selector.get_map(): + remaining = _deadline_remaining(endtime) + if remaining is not None and remaining <= 0: + return (input_offset, False) # Timed out + + ready = selector.select(remaining) + + # Check timeout after select (may have woken spuriously) + if endtime is not None and _time() > endtime: + return (input_offset, False) # Timed out + + for key, events in ready: + if key.fd == stdin_fd: + chunk = input_view[input_offset:input_offset + _PIPE_BUF] + try: + input_offset += os.write(key.fd, chunk) + except BrokenPipeError: + selector.unregister(key.fd) + try: + stdin.close() + except BrokenPipeError: + pass + else: + if input_offset >= len(input_view): + selector.unregister(key.fd) + try: + stdin.close() + except BrokenPipeError: + pass + elif key.fileobj in output_buffers: + data = os.read(key.fd, 32768) + if not data: + selector.unregister(key.fileobj) + if close_on_eof: + try: + key.fileobj.close() + except OSError: + pass + else: + output_buffers[key.fileobj].append(data) + + return (input_offset, True) # Completed + if _mswindows: # On Windows we just need to close `Popen._handle` when we no longer need @@ -289,6 +365,45 @@ STDOUT = -2 DEVNULL = -3 +def _deadline_remaining(endtime): + """Calculate remaining time until deadline.""" + if endtime is None: + return None + return endtime - _time() + + +def _flush_stdin(stdin): + """Flush stdin, ignoring BrokenPipeError and closed file ValueError.""" + try: + stdin.flush() + except BrokenPipeError: + pass # communicate() must ignore BrokenPipeError. + except ValueError: + # Ignore ValueError: I/O operation on closed file. + if not stdin.closed: + raise + + +def _make_input_view(input_data): + """Convert input data to a byte memoryview for writing. + + Handles the case where input_data is already a memoryview with + non-byte elements (e.g., int32 array) by casting to a byte view. + This ensures len(view) returns the byte count, not element count. + """ + if not input_data: + return None + if isinstance(input_data, memoryview): + return input_data.cast("b") # ensure byte view for correct len() + return memoryview(input_data) + + +def _translate_newlines(data, encoding, errors): + """Decode bytes to str and translate newlines to \n.""" + data = data.decode(encoding, errors) + return data.replace("\r\n", "\n").replace("\r", "\n") + + # XXX This function is only used by multiprocessing and the test suite, # but it's here so that it can be imported when Python is compiled without # threads. @@ -381,8 +496,7 @@ def _text_encoding(): if sys.flags.utf8_mode: return "utf-8" - else: - return locale.getencoding() + return locale.getencoding() def call(*popenargs, timeout=None, **kwargs): @@ -750,6 +864,66 @@ def _use_posix_spawn(): return False +def _can_use_pidfd_open(): + # Availability: Linux >= 5.3 + if not hasattr(os, "pidfd_open"): + return False + try: + pidfd = os.pidfd_open(os.getpid(), 0) + except OSError as err: + if err.errno in {errno.EMFILE, errno.ENFILE}: + # transitory 'too many open files' + return True + # likely blocked by security policy like SECCOMP (EPERM, + # EACCES, ENOSYS) + return False + else: + os.close(pidfd) + return True + + +def _can_use_kqueue(): + # Availability: macOS, BSD + names = ( + "kqueue", + "KQ_EV_ADD", + "KQ_EV_ONESHOT", + "KQ_FILTER_PROC", + "KQ_NOTE_EXIT", + ) + if not all(hasattr(select, x) for x in names): + return False + kq = None + try: + kq = select.kqueue() + kev = select.kevent( + os.getpid(), + filter=select.KQ_FILTER_PROC, + flags=select.KQ_EV_ADD | select.KQ_EV_ONESHOT, + fflags=select.KQ_NOTE_EXIT, + ) + kq.control([kev], 1, 0) + return True + except OSError as err: + if err.errno in {errno.EMFILE, errno.ENFILE}: + # transitory 'too many open files' + return True + return False + finally: + if kq is not None: + kq.close() + + +_CAN_USE_PIDFD_OPEN = not _mswindows and _can_use_pidfd_open() +_CAN_USE_KQUEUE = not _mswindows and _can_use_kqueue() + +# Maximum timeout passed to poll() / kqueue.control() by Popen._wait(): +# very large values (e.g. timeout=float('inf')) overflow the C timestamp +# conversion (gh-154836). Longer waits are performed in bounded slices +# until the deadline, like asyncio's MAXIMUM_SELECT_TIMEOUT. +_MAXIMUM_WAIT_TIMEOUT = 24 * 3600 + + # These are primarily fail-safe knobs for negatives. A True value does not # guarantee the given libc/syscall API will be used. _USE_POSIX_SPAWN = _use_posix_spawn() @@ -1096,8 +1270,8 @@ class Popen: self.text_mode = bool(universal_newlines) def _translate_newlines(self, data, encoding, errors): - data = data.decode(encoding, errors) - return data.replace("\r\n", "\n").replace("\r", "\n") + # Subclass-overridable hook; defers to the module-level helper. + return _translate_newlines(data, encoding, errors) def __enter__(self): return self @@ -1224,7 +1398,7 @@ class Popen: # See the detailed comment in .wait(). if timeout is not None: sigint_timeout = min(self._sigint_wait_secs, - self._remaining_time(endtime)) + _deadline_remaining(endtime)) else: sigint_timeout = self._sigint_wait_secs self._sigint_wait_secs = 0 # nothing else should wait. @@ -1237,7 +1411,7 @@ class Popen: finally: self._communication_started = True try: - sts = self.wait(timeout=self._remaining_time(endtime)) + self.wait(timeout=_deadline_remaining(endtime)) except TimeoutExpired as exc: exc.timeout = timeout raise @@ -1251,14 +1425,6 @@ class Popen: return self._internal_poll() - def _remaining_time(self, endtime): - """Convenience for _communicate when computing timeouts.""" - if endtime is None: - return None - else: - return endtime - _time() - - def _check_timeout(self, endtime, orig_timeout, stdout_seq, stderr_seq, skip_check_and_raise=False): """Convenience for checking if a timeout has expired.""" @@ -1284,7 +1450,7 @@ class Popen: # generated SIGINT and will exit rapidly. if timeout is not None: sigint_timeout = min(self._sigint_wait_secs, - self._remaining_time(endtime)) + _deadline_remaining(endtime)) else: sigint_timeout = self._sigint_wait_secs self._sigint_wait_secs = 0 # nothing else should wait. @@ -1651,7 +1817,7 @@ class Popen: # thread remains writing and the fd left open in case the user # calls communicate again. if hasattr(self, "_stdin_thread"): - self._stdin_thread.join(self._remaining_time(endtime)) + self._stdin_thread.join(_deadline_remaining(endtime)) if self._stdin_thread.is_alive(): raise TimeoutExpired(self.args, orig_timeout) @@ -1659,11 +1825,11 @@ class Popen: # threads remain reading and the fds left open in case the user # calls communicate again. if self.stdout is not None: - self.stdout_thread.join(self._remaining_time(endtime)) + self.stdout_thread.join(_deadline_remaining(endtime)) if self.stdout_thread.is_alive(): raise TimeoutExpired(self.args, orig_timeout) if self.stderr is not None: - self.stderr_thread.join(self._remaining_time(endtime)) + self.stderr_thread.join(_deadline_remaining(endtime)) if self.stderr_thread.is_alive(): raise TimeoutExpired(self.args, orig_timeout) @@ -2048,14 +2214,121 @@ class Popen: sts = 0 return (pid, sts) + def _wait_pidfd(self, timeout): + """Wait for PID to terminate using pidfd_open() + poll(). + Linux >= 5.3 only. + """ + if not _CAN_USE_PIDFD_OPEN: + return False + try: + pidfd = os.pidfd_open(self.pid, 0) + except OSError: + # May be: + # - ESRCH: no such process + # - EMFILE, ENFILE: too many open files (usually 1024) + # - ENODEV: anonymous inode filesystem not supported + # - EPERM, EACCES, ENOSYS: undocumented; may happen if + # blocked by security policy like SECCOMP + return False + + try: + poller = select.poll() + poller.register(pidfd, select.POLLIN) + endtime = _time() + timeout + while True: + # Clamp very large timeouts (e.g. float('inf')): + # they overflow the C timestamp conversion in + # poll(). Wait in bounded slices until the real + # deadline (gh-154836). + delay = min(_deadline_remaining(endtime), + _MAXIMUM_WAIT_TIMEOUT) + events = poller.poll(max(delay, 0) * 1000) + if events: + return True + if _deadline_remaining(endtime) <= 0: + raise TimeoutExpired(self.args, timeout) + finally: + os.close(pidfd) + + def _wait_kqueue(self, timeout): + """Wait for PID to terminate using kqueue(). macOS and BSD only.""" + if not _CAN_USE_KQUEUE: + return False + try: + kq = select.kqueue() + except OSError: + # likely EMFILE / ENFILE (too many open files) + return False + + try: + kev = select.kevent( + self.pid, + filter=select.KQ_FILTER_PROC, + flags=select.KQ_EV_ADD | select.KQ_EV_ONESHOT, + fflags=select.KQ_NOTE_EXIT, + ) + changelist = [kev] + endtime = _time() + timeout + while True: + # Clamp very large timeouts (e.g. float('inf')): + # they overflow the C timestamp conversion in + # kqueue.control(). Wait in bounded slices until + # the real deadline (gh-154836). + delay = min(_deadline_remaining(endtime), + _MAXIMUM_WAIT_TIMEOUT) + try: + events = kq.control(changelist, 1, max(delay, 0)) + except OSError: + return False + if events: + return True + if _deadline_remaining(endtime) <= 0: + raise TimeoutExpired(self.args, timeout) + # The kevent was registered by the first control() + # call; don't re-add it on later slices. + changelist = None + finally: + kq.close() def _wait(self, timeout): - """Internal implementation of wait() on POSIX.""" + """Internal implementation of wait() on POSIX. + + Uses efficient pidfd_open() + poll() on Linux or kqueue() + on macOS/BSD when available. Falls back to polling + waitpid(WNOHANG) otherwise. + """ if self.returncode is not None: return self.returncode if timeout is not None: - endtime = _time() + timeout + if timeout < 0: + raise TimeoutExpired(self.args, timeout) + started = _time() + endtime = started + timeout + + # Try efficient wait first. + if self._wait_pidfd(timeout) or self._wait_kqueue(timeout): + # Process is gone. At this point os.waitpid(pid, 0) + # will return immediately, but in very rare races + # the PID may have been reused. + # os.waitpid(pid, WNOHANG) ensures we attempt a + # non-blocking reap without blocking indefinitely. + with self._waitpid_lock: + if self.returncode is not None: + return self.returncode # Another thread waited. + (pid, sts) = self._try_wait(os.WNOHANG) + assert pid == self.pid or pid == 0 + if pid == self.pid: + self._handle_exitstatus(sts) + return self.returncode + # os.waitpid(pid, WNOHANG) returned 0 instead + # of our PID, meaning PID has not yet exited, + # even though poll() / kqueue() said so. Very + # rare and mostly theoretical. Fallback to busy + # polling. + elapsed = _time() - started + endtime -= elapsed + # Enter a busy loop if we have a timeout. This busy loop was # cribbed from Lib/threading.py in Thread.wait() at r71065. delay = 0.0005 # 500 us -> initial delay of 1 ms @@ -2071,7 +2344,7 @@ class Popen: break finally: self._waitpid_lock.release() - remaining = self._remaining_time(endtime) + remaining = _deadline_remaining(endtime) if remaining <= 0: raise TimeoutExpired(self.args, timeout) delay = min(delay * 2, remaining, .05) @@ -2087,6 +2360,7 @@ class Popen: # http://bugs.python.org/issue14396. if pid == self.pid: self._handle_exitstatus(sts) + return self.returncode @@ -2094,14 +2368,7 @@ class Popen: if self.stdin and not self._communication_started: # Flush stdio buffer. This might block, if the user has # been writing to .stdin in an uncontrolled fashion. - try: - self.stdin.flush() - except BrokenPipeError: - pass # communicate() must ignore BrokenPipeError. - except ValueError: - # ignore ValueError: I/O operation on closed file. - if not self.stdin.closed: - raise + _flush_stdin(self.stdin) if not input: try: self.stdin.close() @@ -2126,11 +2393,8 @@ class Popen: self._save_input(input) - if self._input: - if not isinstance(self._input, memoryview): - input_view = memoryview(self._input) - else: - input_view = self._input.cast("b") # byte input required + input_view = _make_input_view(self._input) + input_offset = self._input_offset if self._input else 0 with _PopenSelector() as selector: if self.stdin and not self.stdin.closed and self._input: @@ -2140,43 +2404,31 @@ class Popen: if self.stderr and not self.stderr.closed: selector.register(self.stderr, selectors.EVENT_READ) - while selector.get_map(): - timeout = self._remaining_time(endtime) - if timeout is not None and timeout <= 0: - self._check_timeout(endtime, orig_timeout, - stdout, stderr, - skip_check_and_raise=True) - raise RuntimeError( # Impossible :) - '_check_timeout(..., skip_check_and_raise=True) ' - 'failed to raise TimeoutExpired.') + stdin_to_write = (self.stdin if self.stdin and self._input + and not self.stdin.closed else None) + # Persist the returned offset on self so a subsequent + # communicate() after a TimeoutExpired resumes mid-input + # rather than re-sending bytes the child already consumed. + new_offset, completed = _communicate_io_posix( + selector, + stdin_to_write, + input_view, + input_offset, + self._fileobj2output, + endtime, + close_on_eof=True) + if self._input: + self._input_offset = new_offset - ready = selector.select(timeout) - self._check_timeout(endtime, orig_timeout, stdout, stderr) + if not completed: + self._check_timeout(endtime, orig_timeout, stdout, stderr, + skip_check_and_raise=True) + raise RuntimeError( # Impossible :) + '_check_timeout(..., skip_check_and_raise=True) ' + 'failed to raise TimeoutExpired.') - # XXX Rewrite these to use non-blocking I/O on the file - # objects; they are no longer using C stdio! - - for key, events in ready: - if key.fileobj is self.stdin: - chunk = input_view[self._input_offset : - self._input_offset + _PIPE_BUF] - try: - self._input_offset += os.write(key.fd, chunk) - except BrokenPipeError: - selector.unregister(key.fileobj) - key.fileobj.close() - else: - if self._input_offset >= len(input_view): - selector.unregister(key.fileobj) - key.fileobj.close() - elif key.fileobj in (self.stdout, self.stderr): - data = os.read(key.fd, 32768) - if not data: - selector.unregister(key.fileobj) - key.fileobj.close() - self._fileobj2output[key.fileobj].append(data) try: - self.wait(timeout=self._remaining_time(endtime)) + self.wait(timeout=_deadline_remaining(endtime)) except TimeoutExpired as exc: exc.timeout = orig_timeout raise diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index f9e48c7c7..90e07c0d3 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -40,6 +40,7 @@ from test.support import socket_helper from test.support import threading_helper from test.support import warnings_helper from test.support import subTests +from test.support.script_helper import assert_python_failure, assert_python_ok # Skip tests if _multiprocessing wasn't built. _multiprocessing = import_helper.import_module('_multiprocessing') @@ -377,6 +378,7 @@ class _TestProcess(BaseTestCase): self.assertEqual(current.ident, os.getpid()) self.assertEqual(current.exitcode, None) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_set_executable(self): if self.TYPE == 'threads': self.skipTest(f'test not appropriate for {self.TYPE}') @@ -393,6 +395,7 @@ class _TestProcess(BaseTestCase): p.join() self.assertEqual(p.exitcode, 0) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @support.requires_resource('cpu') def test_args_argument(self): # bpo-45735: Using list or tuple as *args* in constructor could @@ -440,6 +443,7 @@ class _TestProcess(BaseTestCase): q.put(bytes(current.authkey)) q.put(current.pid) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_parent_process_attributes(self): if self.TYPE == "threads": self.skipTest('test not appropriate for {}'.format(self.TYPE)) @@ -460,6 +464,7 @@ class _TestProcess(BaseTestCase): from multiprocessing.process import parent_process wconn.send([parent_process().pid, parent_process().name]) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_parent_process(self): if self.TYPE == "threads": self.skipTest('test not appropriate for {}'.format(self.TYPE)) @@ -498,6 +503,7 @@ class _TestProcess(BaseTestCase): parent_process().join(timeout=support.SHORT_TIMEOUT) wconn.send("alive" if parent_process().is_alive() else "not alive") + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_process(self): q = self.Queue(1) e = self.Event() @@ -538,6 +544,7 @@ class _TestProcess(BaseTestCase): self.assertNotIn(p, self.active_children()) close_queue(q) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipUnless(threading._HAVE_THREAD_NATIVE_ID, "needs native_id") def test_process_mainthread_native_id(self): if self.TYPE == 'threads': @@ -578,6 +585,7 @@ class _TestProcess(BaseTestCase): def _test_sleep(cls, delay): time.sleep(delay) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def _kill_process(self, meth, target=None): if self.TYPE == 'threads': self.skipTest('test not appropriate for {}'.format(self.TYPE)) @@ -635,6 +643,7 @@ class _TestProcess(BaseTestCase): return p.exitcode + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipIf(os.name == 'nt', "POSIX only") def test_interrupt(self): exitcode = self._kill_process(multiprocessing.Process.interrupt) @@ -643,15 +652,18 @@ class _TestProcess(BaseTestCase): # (KeyboardInterrupt in this case) # in multiprocessing.BaseProcess._bootstrap + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipIf(os.name == 'nt', "POSIX only") def test_interrupt_no_handler(self): exitcode = self._kill_process(multiprocessing.Process.interrupt, target=self._sleep_no_int_handler) self.assertEqual(exitcode, -signal.SIGINT) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_terminate(self): exitcode = self._kill_process(multiprocessing.Process.terminate) self.assertEqual(exitcode, -signal.SIGTERM) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_kill(self): exitcode = self._kill_process(multiprocessing.Process.kill) if os.name != 'nt': @@ -667,6 +679,7 @@ class _TestProcess(BaseTestCase): self.assertIsInstance(cpus, int) self.assertGreaterEqual(cpus, 1) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_active_children(self): self.assertEqual(type(self.active_children()), list) @@ -695,6 +708,7 @@ class _TestProcess(BaseTestCase): p.start() p.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_recursion(self): rconn, wconn = self.Pipe(duplex=False) self._test_recursion(wconn, []) @@ -719,6 +733,7 @@ class _TestProcess(BaseTestCase): def _test_sentinel(cls, event): event.wait(10.0) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_sentinel(self): if self.TYPE == "threads": self.skipTest('test not appropriate for {}'.format(self.TYPE)) @@ -741,6 +756,7 @@ class _TestProcess(BaseTestCase): q.get() sys.exit(rc) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_close(self): if self.TYPE == "threads": self.skipTest('test not appropriate for {}'.format(self.TYPE)) @@ -773,6 +789,7 @@ class _TestProcess(BaseTestCase): close_queue(q) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @support.requires_resource('walltime') def test_many_processes(self): if self.TYPE == 'threads': @@ -809,6 +826,7 @@ class _TestProcess(BaseTestCase): for p in procs: self.assertIn(p.exitcode, exitcodes) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_lose_target_ref(self): c = DummyCallable() wr = weakref.ref(c) @@ -871,6 +889,7 @@ class _TestProcess(BaseTestCase): threading.Thread(target=func1).start() threading.Thread(target=func2, daemon=True).start() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_wait_for_threads(self): # A child process should wait for non-daemonic threads to end # before exiting @@ -895,6 +914,7 @@ class _TestProcess(BaseTestCase): setattr(sys, stream_name, None) evt.set() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_error_on_stdio_flush_1(self): # Check that Process works with broken standard streams streams = [io.StringIO(), None] @@ -914,6 +934,7 @@ class _TestProcess(BaseTestCase): finally: setattr(sys, stream_name, old_stream) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_error_on_stdio_flush_2(self): # Same as test_error_on_stdio_flush_1(), but standard streams are # broken by the child process @@ -1064,6 +1085,7 @@ class _TestSubclassingProcess(BaseTestCase): ALLOWED_TYPES = ('processes',) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_subclassing(self): uppercaser = _UpperCaser() uppercaser.daemon = True @@ -1073,6 +1095,7 @@ class _TestSubclassingProcess(BaseTestCase): uppercaser.stop() uppercaser.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_stderr_flush(self): # sys.stderr is flushed at process shutdown (issue #13812) if self.TYPE == "threads": @@ -1103,6 +1126,7 @@ class _TestSubclassingProcess(BaseTestCase): sys.stderr = open(fd, 'w', encoding="utf-8", closefd=False) sys.exit(reason) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_sys_exit(self): # See Issue 13854 if self.TYPE == 'threads': @@ -1170,6 +1194,7 @@ class _TestQueue(BaseTestCase): queue.get() parent_can_continue.set() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_put(self): MAXSIZE = 6 queue = self.Queue(maxsize=MAXSIZE) @@ -1239,6 +1264,7 @@ class _TestQueue(BaseTestCase): queue.put(5) parent_can_continue.set() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_get(self): queue = self.Queue() child_can_start = self.Event() @@ -1301,6 +1327,7 @@ class _TestQueue(BaseTestCase): # process cannot shutdown until the feeder thread has finished # pushing items onto the pipe. + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_fork(self): # Old versions of Queue would fail to create a new feeder # thread for a forked process if the original process had its @@ -1351,6 +1378,7 @@ class _TestQueue(BaseTestCase): time.sleep(DELTA) q.task_done() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_task_done(self): queue = self.JoinableQueue() @@ -1394,6 +1422,7 @@ class _TestQueue(BaseTestCase): self.fail("Probable regression on import lock contention;" " see Issue #22853") + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_timeout(self): q = multiprocessing.Queue() start = time.monotonic() @@ -1517,6 +1546,7 @@ class _TestLock(BaseTestCase): event.set() time.sleep(1.0) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_repr_lock(self): if self.TYPE != 'processes': self.skipTest('test not appropriate for {}'.format(self.TYPE)) @@ -1564,6 +1594,7 @@ class _TestLock(BaseTestCase): event.wait() self.assertEqual(f'', repr(lock)) p.terminate() + p.join() def test_lock(self): lock = self.Lock() @@ -1580,6 +1611,7 @@ class _TestLock(BaseTestCase): res.value = lock.locked() event.set() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipUnless(HAS_SHAREDCTYPES, 'needs sharedctypes') def test_lock_locked_2processes(self): if self.TYPE != 'processes': @@ -1606,6 +1638,7 @@ class _TestLock(BaseTestCase): for _ in range(n): lock.release() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_repr_rlock(self): if self.TYPE != 'processes': self.skipTest('test not appropriate for {}'.format(self.TYPE)) @@ -1665,6 +1698,7 @@ class _TestLock(BaseTestCase): self.assertFalse(lock.locked()) self.assertRaises((AssertionError, RuntimeError), lock.release) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipUnless(HAS_SHAREDCTYPES, 'needs sharedctypes') def test_rlock_locked_2processes(self): if self.TYPE != 'processes': @@ -1776,6 +1810,7 @@ class _TestCondition(BaseTestCase): except NotImplementedError: pass + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_notify(self): cond = self.Condition() sleeping = self.Semaphore(0) @@ -1818,6 +1853,7 @@ class _TestCondition(BaseTestCase): threading_helper.join_thread(t) join_process(p) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_notify_all(self): cond = self.Condition() sleeping = self.Semaphore(0) @@ -1887,6 +1923,7 @@ class _TestCondition(BaseTestCase): # NOTE: join_process and join_thread are the same threading_helper.join_thread(w) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_notify_n(self): cond = self.Condition() sleeping = self.Semaphore(0) @@ -1960,6 +1997,7 @@ class _TestCondition(BaseTestCase): if not result or state.value != 4: sys.exit(1) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipUnless(HAS_SHAREDCTYPES, 'needs sharedctypes') def test_waitfor(self): # based on test in test/lock_tests.py @@ -1995,6 +2033,7 @@ class _TestCondition(BaseTestCase): if not result and (expected - CLOCK_RES) <= dt: success.value = True + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipUnless(HAS_SHAREDCTYPES, 'needs sharedctypes') def test_waitfor_timeout(self): # based on test in test/lock_tests.py @@ -2027,6 +2066,7 @@ class _TestCondition(BaseTestCase): if pid is not None: os.kill(pid, signal.SIGINT) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_wait_result(self): if isinstance(self, ProcessesMixin) and sys.platform != 'win32': pid = os.getpid() @@ -2055,6 +2095,7 @@ class _TestEvent(BaseTestCase): time.sleep(TIMEOUT2) event.set() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_event(self): event = self.Event() wait = TimingWrapper(event.wait) @@ -2251,6 +2292,7 @@ class _TestBarrier(BaseTestCase): pass assert not barrier.broken + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_barrier(self, passes=1): """ Test that a barrier is passed in lockstep @@ -2258,6 +2300,7 @@ class _TestBarrier(BaseTestCase): results = [self.DummyList(), self.DummyList()] self.run_threads(self.multipass, (self.barrier, results, passes)) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_barrier_10(self): """ Test that a barrier works for 10 consecutive runs @@ -2269,6 +2312,7 @@ class _TestBarrier(BaseTestCase): res = barrier.wait() queue.put(res) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_wait_return(self): """ test the return value from barrier.wait @@ -2285,6 +2329,7 @@ class _TestBarrier(BaseTestCase): if len(results) != 1: raise RuntimeError + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_action(self): """ Test the 'action' callback @@ -2307,6 +2352,7 @@ class _TestBarrier(BaseTestCase): except RuntimeError: barrier.abort() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_abort(self): """ Test that an abort will put the barrier in a broken state @@ -2337,6 +2383,7 @@ class _TestBarrier(BaseTestCase): barrier.wait() results3.append(True) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_reset(self): """ Test that a 'reset' on a barrier frees the waiting threads @@ -2372,6 +2419,7 @@ class _TestBarrier(BaseTestCase): barrier.wait() results3.append(True) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_abort_and_reset(self): """ Test that a barrier can be reset after being broken. @@ -2398,6 +2446,7 @@ class _TestBarrier(BaseTestCase): except threading.BrokenBarrierError: results.append(True) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_timeout(self): """ Test wait(timeout) @@ -2417,6 +2466,7 @@ class _TestBarrier(BaseTestCase): except threading.BrokenBarrierError: results.append(True) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_default_timeout(self): """ Test the barrier's default timeout @@ -2438,6 +2488,7 @@ class _TestBarrier(BaseTestCase): with lock: conn.send(i) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_thousand(self): if self.TYPE == 'manager': self.skipTest('test not appropriate for {}'.format(self.TYPE)) @@ -2479,7 +2530,7 @@ class _TestValue(BaseTestCase): for sv, cv in zip(values, cls.codes_values): sv.value = cv[2] - + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_value(self, raw=False): if raw: values = [self.RawValue(code, value) @@ -2527,6 +2578,12 @@ class _TestValue(BaseTestCase): self.assertNotHasAttr(arr5, 'get_lock') self.assertNotHasAttr(arr5, 'get_obj') + @unittest.skipIf(c_int is None, "requires _ctypes") + def test_invalid_typecode(self): + with self.assertRaisesRegex(TypeError, 'bad typecode'): + self.Value('x', None) + with self.assertRaisesRegex(TypeError, 'bad typecode'): + self.RawValue('x', None) class _TestArray(BaseTestCase): @@ -2537,6 +2594,7 @@ class _TestArray(BaseTestCase): for i in range(1, len(seq)): seq[i] += seq[i-1] + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipIf(c_int is None, "requires _ctypes") def test_array(self, raw=False): seq = [680, 626, 934, 821, 150, 233, 548, 982, 714, 831] @@ -2607,6 +2665,12 @@ class _TestArray(BaseTestCase): self.assertNotHasAttr(arr5, 'get_lock') self.assertNotHasAttr(arr5, 'get_obj') + @unittest.skipIf(c_int is None, "requires _ctypes") + def test_invalid_typecode(self): + with self.assertRaisesRegex(TypeError, 'bad typecode'): + self.Array('x', []) + with self.assertRaisesRegex(TypeError, 'bad typecode'): + self.RawArray('x', []) # # # @@ -2842,8 +2906,9 @@ class _TestPool(BaseTestCase): @classmethod def setUpClass(cls): - super().setUpClass() - cls.pool = cls.Pool(4) + with warnings_helper.ignore_fork_in_thread_deprecation_warnings(): + super().setUpClass() + cls.pool = cls.Pool(4) @classmethod def tearDownClass(cls): @@ -2942,6 +3007,7 @@ class _TestPool(BaseTestCase): self.assertEqual(get(), 49) self.assertTimingAlmostEqual(get.elapsed, TIMEOUT1) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_async_timeout(self): p = self.Pool(3) try: @@ -3039,6 +3105,7 @@ class _TestPool(BaseTestCase): self.assertIn(value, expected_values) expected_values.remove(value) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_make_pool(self): expected_error = (RemoteError if self.TYPE == 'manager' else ValueError) @@ -3054,6 +3121,7 @@ class _TestPool(BaseTestCase): p.close() p.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_terminate(self): # Simulate slow tasks which take "forever" to complete sleep_time = support.LONG_TIMEOUT @@ -3071,6 +3139,7 @@ class _TestPool(BaseTestCase): p.terminate() p.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_empty_iterable(self): # See Issue 12157 p = self.Pool(1) @@ -3083,6 +3152,7 @@ class _TestPool(BaseTestCase): p.close() p.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_context(self): if self.TYPE == 'processes': L = list(range(10)) @@ -3097,6 +3167,7 @@ class _TestPool(BaseTestCase): def _test_traceback(cls): raise RuntimeError(123) # some comment + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_traceback(self): # We want ensure that the traceback from the child process is # contained in the traceback raised in the main process. @@ -3136,9 +3207,11 @@ class _TestPool(BaseTestCase): p.join() @classmethod + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def _test_wrapped_exception(cls): raise RuntimeError('foo') + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_wrapped_exception(self): # Issue #20980: Should not wrap exception when using thread pool with self.Pool(1) as p: @@ -3146,6 +3219,7 @@ class _TestPool(BaseTestCase): p.apply(self._test_wrapped_exception) p.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_map_no_failfast(self): # Issue #23992: the fail-fast behaviour when an exception is raised # during map() would make Pool.join() deadlock, because a worker @@ -3181,6 +3255,7 @@ class _TestPool(BaseTestCase): # they were released too. self.assertEqual(CountedObject.n_instances, 0) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_enter(self): if self.TYPE == 'manager': self.skipTest("test not applicable to manager") @@ -3197,6 +3272,7 @@ class _TestPool(BaseTestCase): pass pool.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_resource_warning(self): if self.TYPE == 'manager': self.skipTest("test not applicable to manager") @@ -3222,6 +3298,7 @@ def unpickleable_result(): class _TestPoolWorkerErrors(BaseTestCase): ALLOWED_TYPES = ('processes', ) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_async_error_callback(self): p = multiprocessing.Pool(2) @@ -3237,6 +3314,7 @@ class _TestPoolWorkerErrors(BaseTestCase): p.close() p.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_unpickleable_result(self): from multiprocessing.pool import MaybeEncodingError p = multiprocessing.Pool(2) @@ -3262,6 +3340,7 @@ class _TestPoolWorkerErrors(BaseTestCase): class _TestPoolWorkerLifetime(BaseTestCase): ALLOWED_TYPES = ('processes', ) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_pool_worker_lifetime(self): p = multiprocessing.Pool(3, maxtasksperchild=10) self.assertEqual(3, len(p._pool)) @@ -3291,6 +3370,7 @@ class _TestPoolWorkerLifetime(BaseTestCase): p.close() p.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_pool_worker_lifetime_early_close(self): # Issue #10332: closing a pool whose workers have limited lifetimes # before all the tasks completed would make join() hang. @@ -3364,6 +3444,7 @@ class _TestMyManager(BaseTestCase): ALLOWED_TYPES = ('manager',) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @support.skip_if_sanitizer('TSan: leaks threads', thread=True) def test_mymanager(self): manager = MyManager(shutdown_timeout=SHUTDOWN_TIMEOUT) @@ -3376,6 +3457,7 @@ class _TestMyManager(BaseTestCase): # which happens on slow buildbots. self.assertIn(manager._process.exitcode, (0, -signal.SIGTERM)) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @support.skip_if_sanitizer('TSan: leaks threads', thread=True) def test_mymanager_context(self): manager = MyManager(shutdown_timeout=SHUTDOWN_TIMEOUT) @@ -3386,6 +3468,7 @@ class _TestMyManager(BaseTestCase): # which happens on slow buildbots. self.assertIn(manager._process.exitcode, (0, -signal.SIGTERM)) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @support.skip_if_sanitizer('TSan: leaks threads', thread=True) def test_mymanager_context_prestarted(self): manager = MyManager(shutdown_timeout=SHUTDOWN_TIMEOUT) @@ -3457,6 +3540,7 @@ class _TestRemoteManager(BaseTestCase): # Note that xmlrpclib will deserialize object as a list not a tuple queue.put(tuple(cls.values)) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @support.skip_if_sanitizer('TSan: leaks threads', thread=True) def test_remote(self): authkey = os.urandom(32) @@ -3499,6 +3583,7 @@ class _TestManagerRestart(BaseTestCase): queue = manager.get_queue() queue.put('hello world') + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @support.skip_if_sanitizer("TSan: leaks threads", thread=True) def test_rapid_restart(self): authkey = os.urandom(32) @@ -3552,12 +3637,14 @@ class FakeConnection: class TestManagerExceptions(unittest.TestCase): # Issue 106558: Manager exceptions avoids creating cyclic references. def setUp(self): - self.mgr = multiprocessing.Manager() + with warnings_helper.ignore_fork_in_thread_deprecation_warnings(): + self.mgr = multiprocessing.Manager() def tearDown(self): self.mgr.shutdown() self.mgr.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_queue_get(self): queue = self.mgr.Queue() if gc.isenabled(): @@ -3569,6 +3656,7 @@ class TestManagerExceptions(unittest.TestCase): wr = weakref.ref(e) self.assertEqual(wr(), None) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_dispatch(self): if gc.isenabled(): gc.disable() @@ -3595,6 +3683,7 @@ class _TestConnection(BaseTestCase): conn.send_bytes(msg) conn.close() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_connection(self): conn, child_conn = self.Pipe() @@ -3687,6 +3776,7 @@ class _TestConnection(BaseTestCase): self.assertRaises(OSError, writer.recv) self.assertRaises(OSError, writer.poll) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_spawn_close(self): # We test that a pipe connection can be closed by parent # process immediately after child is spawned. On Windows this @@ -3763,6 +3853,7 @@ class _TestConnection(BaseTestCase): os.write(fd, data) os.close(fd) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") def test_fd_transfer(self): if self.TYPE != 'processes': @@ -3782,6 +3873,7 @@ class _TestConnection(BaseTestCase): with open(os_helper.TESTFN, "rb") as f: self.assertEqual(f.read(), b"foo") + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") @unittest.skipIf(sys.platform == "win32", "test semantics don't make sense on Windows") @@ -3819,6 +3911,7 @@ class _TestConnection(BaseTestCase): def _send_data_without_fd(self, conn): os.write(conn.fileno(), b"\0") + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") @unittest.skipIf(sys.platform == "win32", "doesn't make sense on Windows") def test_missing_fd_transfer(self): @@ -3850,6 +3943,19 @@ class _TestConnection(BaseTestCase): self.assertRaises(OSError, a.recv) self.assertRaises(OSError, b.recv) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_wait_empty(self): + if self.TYPE != 'processes': + self.skipTest('test not appropriate for {}'.format(self.TYPE)) + # gh-145587: wait() with empty list should respect timeout + timeout = 0.5 + start = time.monotonic() + res = self.connection.wait([], timeout=timeout) + duration = time.monotonic() - start + + self.assertEqual(res, []) + self.assertGreaterEqual(duration, timeout - 0.1) + class _TestListener(BaseTestCase): ALLOWED_TYPES = ('processes',) @@ -3918,6 +4024,7 @@ class _TestListenerClient(BaseTestCase): conn.send('hello') conn.close() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_listener_client(self): for family in self.connection.families: l = self.connection.Listener(family=family) @@ -3929,6 +4036,7 @@ class _TestListenerClient(BaseTestCase): p.join() l.close() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_issue14725(self): l = self.connection.Listener() p = self.Process(target=self._test, args=(l.address,)) @@ -3974,6 +4082,7 @@ class _TestPoll(BaseTestCase): conn.send_bytes(s) conn.close() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_strings(self): strings = (b'hello', b'', b'a', b'b', b'', b'bye', b'', b'lop') a, b = self.Pipe() @@ -3997,6 +4106,7 @@ class _TestPoll(BaseTestCase): # read from it. r.poll(5) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_boundaries(self): r, w = self.Pipe(False) p = self.Process(target=self._child_boundaries, args=(r,)) @@ -4015,6 +4125,7 @@ class _TestPoll(BaseTestCase): b.send_bytes(b'b') b.send_bytes(b'cd') + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_dont_merge(self): a, b = self.Pipe() self.assertEqual(a.poll(0.0), False) @@ -4083,6 +4194,7 @@ class _TestPicklingConnections(BaseTestCase): conn.close() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_pickling(self): families = self.connection.families @@ -4141,6 +4253,7 @@ class _TestPicklingConnections(BaseTestCase): conn.close() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_access(self): # On Windows, if we do not specify a destination pid when # using DupHandle then we need to be careful to use the @@ -4304,6 +4417,7 @@ class _TestSharedCTypes(BaseTestCase): for i in range(len(arr)): arr[i] *= 2 + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_sharedctypes(self, lock=False): x = Value('i', 7, lock=lock) y = Value(c_double, 1.0/3.0, lock=lock) @@ -4578,6 +4692,7 @@ class _TestSharedMemory(BaseTestCase): with self.assertRaises(FileNotFoundError): pickle.loads(pickled_sms) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_shared_memory_across_processes(self): # bpo-40135: don't define shared memory block's name in case of # the failure when we run multiprocessing tests in parallel. @@ -4606,6 +4721,7 @@ class _TestSharedMemory(BaseTestCase): sms.close() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipIf(os.name != "posix", "not feasible in non-posix platforms") def test_shared_memory_SharedMemoryServer_ignores_sigint(self): # bpo-36368: protect SharedMemoryManager server process from @@ -4652,6 +4768,7 @@ class _TestSharedMemory(BaseTestCase): # properly released sl. self.assertFalse(err) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_shared_memory_SharedMemoryManager_basics(self): smm1 = multiprocessing.managers.SharedMemoryManager() with self.assertRaises(ValueError): @@ -4993,6 +5110,7 @@ class _TestFinalize(BaseTestCase): conn.close() os._exit(0) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_finalize(self): conn, child_conn = self.Pipe() @@ -5120,6 +5238,7 @@ class _TestLogging(BaseTestCase): logger = multiprocessing.get_logger() conn.send(logger.getEffectiveLevel()) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_level(self): LEVEL1 = 32 LEVEL2 = 37 @@ -5204,6 +5323,7 @@ class _TestPollEintr(BaseTestCase): time.sleep(0.1) os.kill(pid, signal.SIGUSR1) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipUnless(hasattr(signal, 'SIGUSR1'), 'requires SIGUSR1') def test_poll_eintr(self): got_signal = [False] @@ -5343,14 +5463,16 @@ def initializer(ns): @hashlib_helper.requires_hashdigest('sha256') class TestInitializers(unittest.TestCase): def setUp(self): - self.mgr = multiprocessing.Manager() - self.ns = self.mgr.Namespace() - self.ns.test = 0 + with warnings_helper.ignore_fork_in_thread_deprecation_warnings(): + self.mgr = multiprocessing.Manager() + self.ns = self.mgr.Namespace() + self.ns.test = 0 def tearDown(self): self.mgr.shutdown() self.mgr.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_manager_initializer(self): m = multiprocessing.managers.SyncManager() self.assertRaises(TypeError, m.start, 1) @@ -5359,6 +5481,7 @@ class TestInitializers(unittest.TestCase): m.shutdown() m.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_pool_initializer(self): self.assertRaises(TypeError, multiprocessing.Pool, initializer=1) p = multiprocessing.Pool(1, initializer, (self.ns,)) @@ -5416,16 +5539,19 @@ class _file_like(object): class TestStdinBadfiledescriptor(unittest.TestCase): + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_queue_in_process(self): proc = multiprocessing.Process(target=_test_process) proc.start() proc.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_pool_in_process(self): p = multiprocessing.Process(target=pool_in_process) p.start() p.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_flushing(self): sio = io.StringIO() flike = _file_like(sio) @@ -5445,6 +5571,7 @@ class TestWait(unittest.TestCase): w.send((i, os.getpid())) w.close() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_wait(self, slow=False): from multiprocessing.connection import wait readers = [] @@ -5485,6 +5612,7 @@ class TestWait(unittest.TestCase): s.sendall(('%s\n' % i).encode('ascii')) s.close() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_wait_socket(self, slow=False): from multiprocessing.connection import wait l = socket.create_server((socket_helper.HOST, 0)) @@ -5549,6 +5677,7 @@ class TestWait(unittest.TestCase): sem.release() time.sleep(period) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @support.requires_resource('walltime') def test_wait_integer(self): from multiprocessing.connection import wait @@ -5593,6 +5722,7 @@ class TestWait(unittest.TestCase): p.terminate() p.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_neg_timeout(self): from multiprocessing.connection import wait a, b = multiprocessing.Pipe() @@ -5670,6 +5800,7 @@ class TestTimeouts(unittest.TestCase): conn.send(456) conn.close() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_timeout(self): old_timeout = socket.getdefaulttimeout() try: @@ -5727,6 +5858,7 @@ class TestForkAwareThreadLock(unittest.TestCase): conn.send(len(util._afterfork_registry)) conn.close() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_lock(self): r, w = multiprocessing.Pipe(False) l = util.ForkAwareThreadLock() @@ -5778,6 +5910,7 @@ class TestCloseFds(unittest.TestCase): s.close() conn.send(None) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_closefd(self): if not HAS_REDUCTION: raise unittest.SkipTest('requires fd pickling') @@ -5823,6 +5956,7 @@ class TestIgnoreEINTR(unittest.TestCase): conn.send(x) conn.send_bytes(b'x' * cls.CONN_MAX_SIZE) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipUnless(hasattr(signal, 'SIGUSR1'), 'requires SIGUSR1') def test_ignore(self): conn, child_conn = multiprocessing.Pipe() @@ -5856,6 +5990,7 @@ class TestIgnoreEINTR(unittest.TestCase): a = l.accept() a.send('welcome') + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipUnless(hasattr(signal, 'SIGUSR1'), 'requires SIGUSR1') def test_ignore_listener(self): conn, child_conn = multiprocessing.Pipe() @@ -5890,6 +6025,7 @@ class TestStartMethod(unittest.TestCase): p.join() self.assertEqual(child_method, ctx.get_start_method()) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_context(self): for method in ('fork', 'spawn', 'forkserver'): try: @@ -5906,6 +6042,7 @@ class TestStartMethod(unittest.TestCase): def _dummy_func(): pass + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_spawn_dont_set_context(self): # Run a process with spawn or forkserver context may change # the global start method, see gh-109263. @@ -5943,6 +6080,7 @@ class TestStartMethod(unittest.TestCase): with self.assertRaisesRegex(TypeError, 'module_names must be a list of strings'): ctx.set_forkserver_preload([1, 2, 3]) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_set_get(self): multiprocessing.set_forkserver_preload(PRELOAD) count = 0 @@ -6000,6 +6138,7 @@ class TestStartMethod(unittest.TestCase): print(err) self.fail("failed spawning forkserver or grandchild") + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @unittest.skipIf(sys.platform == "win32", "Only Spawn on windows so no risk of mixing") @only_run_in_spawn_testsuite("avoids redundant testing.") @@ -6033,6 +6172,7 @@ class TestStartMethod(unittest.TestCase): process.start() process.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_nested_startmethod(self): # gh-108520: Regression test to ensure that child process can send its # arguments to another process @@ -6189,6 +6329,7 @@ class TestResourceTracker(unittest.TestCase): reused &= _resource_tracker._check_alive() conn.send(reused) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_resource_tracker_reused(self): from multiprocessing.resource_tracker import _resource_tracker _resource_tracker.ensure_running() @@ -6467,6 +6608,7 @@ class TestSimpleQueue(unittest.TestCase): with self.assertRaisesRegex(OSError, 'is closed'): q.empty() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_empty(self): queue = multiprocessing.SimpleQueue() child_can_start = multiprocessing.Event() @@ -6574,7 +6716,8 @@ class TestSyncManagerTypes(unittest.TestCase): def setUp(self): self.manager = self.manager_class() - self.manager.start() + with warnings_helper.ignore_fork_in_thread_deprecation_warnings(): + self.manager.start() self.proc = None def tearDown(self): @@ -6623,6 +6766,7 @@ class TestSyncManagerTypes(unittest.TestCase): obj.clear() obj.wait(0.001) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_event(self): o = self.manager.Event() o.set() @@ -6635,6 +6779,7 @@ class TestSyncManagerTypes(unittest.TestCase): obj.acquire() obj.locked() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_lock(self, lname="Lock"): o = getattr(self.manager, lname)() self.run_worker(self._test_lock, o) @@ -6647,6 +6792,7 @@ class TestSyncManagerTypes(unittest.TestCase): obj.release() obj.locked() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_rlock(self, lname="RLock"): o = getattr(self.manager, lname)() self.run_worker(self._test_rlock, o) @@ -6655,6 +6801,7 @@ class TestSyncManagerTypes(unittest.TestCase): def _test_semaphore(cls, obj): obj.acquire() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_semaphore(self, sname="Semaphore"): o = getattr(self.manager, sname)() self.run_worker(self._test_semaphore, o) @@ -6668,6 +6815,7 @@ class TestSyncManagerTypes(unittest.TestCase): obj.acquire() obj.release() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_condition(self): o = self.manager.Condition() self.run_worker(self._test_condition, o) @@ -6677,6 +6825,7 @@ class TestSyncManagerTypes(unittest.TestCase): assert obj.parties == 5 obj.reset() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_barrier(self): o = self.manager.Barrier(5) self.run_worker(self._test_barrier, o) @@ -6687,6 +6836,7 @@ class TestSyncManagerTypes(unittest.TestCase): with obj: pass + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_pool(self): o = self.manager.Pool(processes=4) self.run_worker(self._test_pool, o) @@ -6701,6 +6851,7 @@ class TestSyncManagerTypes(unittest.TestCase): assert obj.get() == 6 assert obj.empty() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_queue(self, qname="Queue"): o = getattr(self.manager, qname)(2) o.put(5) @@ -6709,6 +6860,7 @@ class TestSyncManagerTypes(unittest.TestCase): assert o.empty() assert not o.full() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_joinable_queue(self): self.test_queue("JoinableQueue") @@ -6743,6 +6895,7 @@ class TestSyncManagerTypes(unittest.TestCase): obj.clear() case.assertEqual(len(obj), 0) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_list(self): o = self.manager.list() o.append(5) @@ -6784,6 +6937,7 @@ class TestSyncManagerTypes(unittest.TestCase): obj.clear() case.assertEqual(len(obj), 0) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_dict(self): o = self.manager.dict() o['foo'] = 5 @@ -6798,6 +6952,7 @@ class TestSyncManagerTypes(unittest.TestCase): case.assertEqual(obj.get(), 1) obj.set(2) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_value(self): o = self.manager.Value('i', 1) self.run_worker(self._test_value, o) @@ -6812,6 +6967,7 @@ class TestSyncManagerTypes(unittest.TestCase): case.assertEqual(len(obj), 2) case.assertListEqual(list(obj), [0, 1]) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_array(self): o = self.manager.Array('i', [0, 1]) self.run_worker(self._test_array, o) @@ -6822,6 +6978,7 @@ class TestSyncManagerTypes(unittest.TestCase): case.assertEqual(obj.x, 0) case.assertEqual(obj.y, 1) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_namespace(self): o = self.manager.Namespace() o.x = 0 @@ -6941,6 +7098,7 @@ class TestSyncManagerTypes(unittest.TestCase): case.assertGreater(obj, {'a'}) case.assertGreaterEqual(obj, {'a', 'b'}) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_set(self): o = self.manager.set() self.run_worker(self._test_set_operator_symbols, o) @@ -6958,6 +7116,7 @@ class TestSyncManagerTypes(unittest.TestCase): self.assertSetEqual(o, {"a", "b", "c"}) self.assertRaises(RemoteError, self.manager.set, 1234) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_set_contain_all_method(self): o = self.manager.set() set_methods = { @@ -7011,6 +7170,7 @@ class _TestAtExit(BaseTestCase): f.write("deadbeef") atexit.register(exit_handler) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_atexit(self): # gh-83856 with os_helper.temp_dir() as temp_dir: @@ -7148,6 +7308,7 @@ class MiscTestCase(unittest.TestCase): self.assertEqual("332833500", out.decode('utf-8').strip()) self.assertFalse(err, msg=err.decode('utf-8')) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_forked_thread_not_started(self): # gh-134381: Ensure that a thread that has not been started yet in # the parent process can be started within a forked child process. @@ -7298,8 +7459,9 @@ class ManagerMixin(BaseMixin): @classmethod def setUpClass(cls): - super().setUpClass() - cls.manager = multiprocessing.Manager() + with warnings_helper.ignore_fork_in_thread_deprecation_warnings(): + super().setUpClass() + cls.manager = multiprocessing.Manager() @classmethod def tearDownClass(cls): @@ -7468,6 +7630,64 @@ class SemLockTests(unittest.TestCase): _multiprocessing.sem_unlink(name) +@unittest.skipIf(sys.platform != "linux", "Linux only") +class ForkInThreads(unittest.TestCase): + + def test_fork(self): + code = """ + import os, sys, threading, time + + t = threading.Thread(target=time.sleep, args=(1,), daemon=True) + t.start() + + assert threading.active_count() == 2 + + pid = os.fork() + if pid < 0: + print("Fork failed") + elif pid == 0: + print("In child") + sys.exit(0) + print("In parent") + """ + + res = assert_python_ok("-c", code, PYTHONWARNINGS='always') + self.assertIn(b'In child', res.out) + self.assertIn(b'In parent', res.out) + self.assertIn(b'DeprecationWarning', res.err) + self.assertIn(b'is multi-threaded, use of fork() may lead to deadlocks in the child', res.err) + + res = assert_python_failure("-c", code, PYTHONWARNINGS='error') + self.assertIn(b'DeprecationWarning', res.err) + self.assertIn(b'is multi-threaded, use of fork() may lead to deadlocks in the child', res.err) + + def test_forkpty(self): + code = """ + import os, sys, threading, time + + t = threading.Thread(target=time.sleep, args=(1,), daemon=True) + t.start() + + assert threading.active_count() == 2 + + pid, _ = os.forkpty() + if pid < 0: + print(f"forkpty failed") + elif pid == 0: + print(f"In child") + sys.exit(0) + print(f"In parent") + """ + + res = assert_python_ok("-c", code, PYTHONWARNINGS='always') + self.assertIn(b'In parent', res.out) + self.assertIn(b'DeprecationWarning', res.err) + self.assertIn(b'is multi-threaded, use of forkpty() may lead to deadlocks in the child', res.err) + + res = assert_python_failure("-c", code, PYTHONWARNINGS='error') + self.assertIn(b'DeprecationWarning', res.err) + self.assertIn(b'is multi-threaded, use of forkpty() may lead to deadlocks in the child', res.err) + @unittest.skipUnless(HAS_SHMEM, "requires multiprocessing.shared_memory") class TestSharedMemoryNames(unittest.TestCase): @subTests('use_simple_format', (True, False)) @@ -7511,7 +7731,7 @@ class TestSharedMemoryNames(unittest.TestCase): print("SUCCESS") """ % use_simple_format) - rc, out, err = script_helper.assert_python_ok("-c", test_script) + rc, out, err = assert_python_ok("-c", test_script) self.assertIn(b"SUCCESS", out) self.assertNotIn(b"traceback", err.lower(), err) self.assertNotIn(b"resource_tracker.py", err, err) diff --git a/Lib/test/test_concurrent_futures/executor.py b/Lib/test/test_concurrent_futures/executor.py index 6a26d695b..a37c4d45f 100644 --- a/Lib/test/test_concurrent_futures/executor.py +++ b/Lib/test/test_concurrent_futures/executor.py @@ -1,12 +1,11 @@ import itertools import threading import time -import unittest import weakref from concurrent import futures from operator import add from test import support -from test.support import Py_GIL_DISABLED +from test.support import Py_GIL_DISABLED, warnings_helper def mul(x, y): @@ -41,12 +40,15 @@ class FalseyLenException(Exception): class ExecutorTest: + # Executor.shutdown() and context manager usage is tested by # ExecutorShutdownTest. + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_submit(self): future = self.executor.submit(pow, 2, 8) self.assertEqual(256, future.result()) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_submit_keyword(self): future = self.executor.submit(mul, 2, y=8) self.assertEqual(16, future.result()) @@ -57,6 +59,7 @@ class ExecutorTest: with self.assertRaises(TypeError): self.executor.submit(arg=1) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_map(self): self.assertEqual( list(self.executor.map(pow, range(10), range(10))), @@ -66,12 +69,15 @@ class ExecutorTest: list(self.executor.map(pow, range(10), range(10), chunksize=3)), list(map(pow, range(10), range(10)))) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_map_exception(self): i = self.executor.map(divmod, [1, 1, 1, 1], [2, 3, 0, 5]) self.assertEqual(i.__next__(), (0, 1)) self.assertEqual(i.__next__(), (0, 1)) - self.assertRaises(ZeroDivisionError, i.__next__) + with self.assertRaises(ZeroDivisionError): + i.__next__() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @support.requires_resource('walltime') def test_map_timeout(self): results = [] @@ -85,7 +91,9 @@ class ExecutorTest: else: self.fail('expected TimeoutError') - self.assertEqual([None, None], results) + # gh-110097: On heavily loaded systems, the launch of the worker may + # take longer than the specified timeout. + self.assertIn(results, ([None, None], [None], [])) def test_map_buffersize_type_validation(self): for buffersize in ("foo", 2.0): @@ -96,7 +104,6 @@ class ExecutorTest: ): self.executor.map(str, range(4), buffersize=buffersize) - def test_map_buffersize_value_validation(self): for buffersize in (0, -1): with self.subTest(buffersize=buffersize): @@ -106,7 +113,7 @@ class ExecutorTest: ): self.executor.map(str, range(4), buffersize=buffersize) - + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_map_buffersize(self): ints = range(4) for buffersize in (1, 2, len(ints), len(ints) * 2): @@ -114,7 +121,7 @@ class ExecutorTest: res = self.executor.map(str, ints, buffersize=buffersize) self.assertListEqual(list(res), ["0", "1", "2", "3"]) - + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_map_buffersize_on_multiple_iterables(self): ints = range(4) for buffersize in (1, 2, len(ints), len(ints) * 2): @@ -122,19 +129,34 @@ class ExecutorTest: res = self.executor.map(add, ints, ints, buffersize=buffersize) self.assertListEqual(list(res), [0, 2, 4, 6]) - + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_map_buffersize_on_infinite_iterable(self): res = self.executor.map(str, itertools.count(), buffersize=2) self.assertEqual(next(res, None), "0") self.assertEqual(next(res, None), "1") self.assertEqual(next(res, None), "2") + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_map_buffersize_on_multiple_infinite_iterables(self): + res = self.executor.map( + add, + itertools.count(), + itertools.count(), + buffersize=2 + ) + self.assertEqual(next(res, None), 0) + self.assertEqual(next(res, None), 2) + self.assertEqual(next(res, None), 4) def test_map_buffersize_on_empty_iterable(self): res = self.executor.map(str, [], buffersize=2) self.assertIsNone(next(res, None)) + def test_map_buffersize_without_iterable(self): + res = self.executor.map(str, buffersize=2) + self.assertIsNone(next(res, None)) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_map_buffersize_when_buffer_is_full(self): ints = iter(range(4)) buffersize = 2 @@ -146,7 +168,7 @@ class ExecutorTest: msg="should have fetched only `buffersize` elements from `ints`.", ) - + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_shutdown_race_issue12456(self): # Issue #12456: race condition at shutdown where trying to post a # sentinel in the call queue blocks (the queue is full while processes @@ -154,6 +176,7 @@ class ExecutorTest: self.executor.map(str, [2] * (self.worker_count + 1)) self.executor.shutdown() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @support.cpython_only def test_no_stale_references(self): # Issue #16284: check that the executors don't unnecessarily hang onto @@ -198,6 +221,7 @@ class ExecutorTest: "than 0"): self.executor_type(max_workers=number) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_free_reference(self): # Issue #14406: Result iterator should not keep an internal # reference to result objects. @@ -210,6 +234,7 @@ class ExecutorTest: if wr() is None: break + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_swallows_falsey_exceptions(self): # see gh-132063: Prevent exceptions that evaluate as falsey # from being ignored. diff --git a/Lib/test/test_concurrent_futures/test_as_completed.py b/Lib/test/test_concurrent_futures/test_as_completed.py index c90b0021d..31c7bb3eb 100644 --- a/Lib/test/test_concurrent_futures/test_as_completed.py +++ b/Lib/test/test_concurrent_futures/test_as_completed.py @@ -7,6 +7,7 @@ from concurrent.futures._base import ( CANCELLED_AND_NOTIFIED, FINISHED, Future) from test import support +from test.support import warnings_helper from .util import ( PENDING_FUTURE, RUNNING_FUTURE, @@ -19,6 +20,7 @@ def mul(x, y): class AsCompletedTests: + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_no_timeout(self): future1 = self.executor.submit(mul, 2, 21) future2 = self.executor.submit(mul, 7, 6) @@ -35,6 +37,7 @@ class AsCompletedTests: future1, future2]), completed) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_future_times_out(self): """Test ``futures.as_completed`` timing out before completing it's final future.""" @@ -62,6 +65,7 @@ class AsCompletedTests: # Check that ``future`` wasn't completed. self.assertEqual(completed_futures, already_completed) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_duplicate_futures(self): # Issue 20367. Duplicate futures should not raise exceptions or give # duplicate responses. diff --git a/Lib/test/test_concurrent_futures/test_deadlock.py b/Lib/test/test_concurrent_futures/test_deadlock.py index dcc1d6856..5cd84f7e4 100644 --- a/Lib/test/test_concurrent_futures/test_deadlock.py +++ b/Lib/test/test_concurrent_futures/test_deadlock.py @@ -10,6 +10,7 @@ from concurrent import futures from concurrent.futures.process import BrokenProcessPool, _ThreadWakeup from test import support +from test.support import warnings_helper from .util import ( create_executor_tests, setup_module, @@ -111,9 +112,9 @@ class ExecutorDeadlockTest: print(f"\nTraceback:\n {tb}", file=sys.__stderr__) self.fail(f"Executor deadlock:\n\n{tb}") - - def _check_crash(self, error, func, *args, ignore_stderr=False): - # test for deadlock caused by crashes in a pool + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def _check_error(self, error, func, *args, ignore_stderr=False): + # test for deadlock caused by crashes or exiting in a pool self.executor.shutdown(wait=True) executor = self.executor_type( @@ -138,49 +139,52 @@ class ExecutorDeadlockTest: def test_error_at_task_pickle(self): # Check problem occurring while pickling a task in # the task_handler thread - self._check_crash(PicklingError, id, ErrorAtPickle()) + self._check_error(PicklingError, id, ErrorAtPickle()) def test_exit_at_task_unpickle(self): # Check problem occurring while unpickling a task on workers - self._check_crash(BrokenProcessPool, id, ExitAtUnpickle()) + self._check_error(BrokenProcessPool, id, ExitAtUnpickle()) def test_error_at_task_unpickle(self): # gh-109832: Restore stderr overridden by _raise_error_ignore_stderr() self.addCleanup(setattr, sys, 'stderr', sys.stderr) # Check problem occurring while unpickling a task on workers - self._check_crash(BrokenProcessPool, id, ErrorAtUnpickle()) + self._check_error(BrokenProcessPool, id, ErrorAtUnpickle()) + @support.skip_if_sanitizer("UBSan: explicit SIGSEV not allowed", ub=True) def test_crash_at_task_unpickle(self): # Check problem occurring while unpickling a task on workers - self._check_crash(BrokenProcessPool, id, CrashAtUnpickle()) + self._check_error(BrokenProcessPool, id, CrashAtUnpickle()) + @support.skip_if_sanitizer("UBSan: explicit SIGSEV not allowed", ub=True) def test_crash_during_func_exec_on_worker(self): # Check problem occurring during func execution on workers - self._check_crash(BrokenProcessPool, _crash) + self._check_error(BrokenProcessPool, _crash) def test_exit_during_func_exec_on_worker(self): # Check problem occurring during func execution on workers - self._check_crash(SystemExit, _exit) + self._check_error(SystemExit, _exit) def test_error_during_func_exec_on_worker(self): # Check problem occurring during func execution on workers - self._check_crash(RuntimeError, _raise_error, RuntimeError) + self._check_error(RuntimeError, _raise_error, RuntimeError) + @support.skip_if_sanitizer("UBSan: explicit SIGSEV not allowed", ub=True) def test_crash_during_result_pickle_on_worker(self): # Check problem occurring while pickling a task result # on workers - self._check_crash(BrokenProcessPool, _return_instance, CrashAtPickle) + self._check_error(BrokenProcessPool, _return_instance, CrashAtPickle) def test_exit_during_result_pickle_on_worker(self): # Check problem occurring while pickling a task result # on workers - self._check_crash(SystemExit, _return_instance, ExitAtPickle) + self._check_error(SystemExit, _return_instance, ExitAtPickle) def test_error_during_result_pickle_on_worker(self): # Check problem occurring while pickling a task result # on workers - self._check_crash(PicklingError, _return_instance, ErrorAtPickle) + self._check_error(PicklingError, _return_instance, ErrorAtPickle) def test_error_during_result_unpickle_in_result_handler(self): # gh-109832: Restore stderr overridden by _raise_error_ignore_stderr() @@ -188,15 +192,17 @@ class ExecutorDeadlockTest: # Check problem occurring while unpickling a task in # the result_handler thread - self._check_crash(BrokenProcessPool, + self._check_error(BrokenProcessPool, _return_instance, ErrorAtUnpickle, ignore_stderr=True) def test_exit_during_result_unpickle_in_result_handler(self): # Check problem occurring while unpickling a task in # the result_handler thread - self._check_crash(BrokenProcessPool, _return_instance, ExitAtUnpickle) + self._check_error(BrokenProcessPool, _return_instance, ExitAtUnpickle) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + @support.skip_if_sanitizer("UBSan: explicit SIGSEV not allowed", ub=True) def test_shutdown_deadlock(self): # Test that the pool calling shutdown do not cause deadlock # if a worker fails after the shutdown call. @@ -209,6 +215,7 @@ class ExecutorDeadlockTest: with self.assertRaises(BrokenProcessPool): f.result() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_shutdown_deadlock_pickle(self): # Test that the pool calling shutdown with wait=False does not cause # a deadlock if a task fails at pickle after the shutdown call. @@ -235,8 +242,10 @@ class ExecutorDeadlockTest: # dangling threads executor_manager.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + @support.skip_if_sanitizer("UBSan: explicit SIGSEV not allowed", ub=True) def test_crash_big_data(self): - # Test that there is a clean exception instad of a deadlock when a + # Test that there is a clean exception instead of a deadlock when a # child process crashes while some data is being written into the # queue. # https://github.com/python/cpython/issues/94777 @@ -250,6 +259,7 @@ class ExecutorDeadlockTest: executor.shutdown(wait=True) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_gh105829_should_not_deadlock_if_wakeup_pipe_full(self): # Issue #105829: The _ExecutorManagerThread wakeup pipe could # fill up and block. See: https://github.com/python/cpython/issues/105829 diff --git a/Lib/test/test_concurrent_futures/test_init.py b/Lib/test/test_concurrent_futures/test_init.py index df6409293..c9fedc721 100644 --- a/Lib/test/test_concurrent_futures/test_init.py +++ b/Lib/test/test_concurrent_futures/test_init.py @@ -1,5 +1,6 @@ import contextlib import logging +import multiprocessing import queue import time import unittest @@ -11,6 +12,7 @@ from concurrent.futures.process import _check_system_limits from logging.handlers import QueueHandler from test import support +from test.support import warnings_helper from .util import ExecutorMixin, create_executor_tests, setup_module @@ -20,6 +22,10 @@ INITIALIZER_STATUS = 'uninitialized' def init(x): global INITIALIZER_STATUS INITIALIZER_STATUS = x + # InterpreterPoolInitializerTest.test_initializer fails + # if we don't have a LOAD_GLOBAL. (It could be any global.) + # We will address this separately. + INITIALIZER_STATUS def get_init_status(): return INITIALIZER_STATUS @@ -44,6 +50,7 @@ class InitializerMixin(ExecutorMixin): initargs=('initialized',)) super().setUp() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_initializer(self): futures = [self.executor.submit(get_init_status) for _ in range(self.worker_count)] @@ -70,6 +77,7 @@ class FailingInitializerMixin(ExecutorMixin): self.executor_kwargs = dict(initializer=init_fail) super().setUp() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_initializer(self): with self._assert_logged('ValueError: error in initializer'): try: @@ -140,6 +148,8 @@ class FailingInitializerResourcesTest(unittest.TestCase): self._test(ProcessPoolSpawnFailingInitializerTest) @support.skip_if_sanitizer("TSAN doesn't support threads after fork", thread=True) + @unittest.skipUnless("forkserver" in multiprocessing.get_all_start_methods(), + "forkserver start method is not available") def test_forkserver(self): self._test(ProcessPoolForkserverFailingInitializerTest) diff --git a/Lib/test/test_concurrent_futures/test_interpreter_pool.py b/Lib/test/test_concurrent_futures/test_interpreter_pool.py index b8f1a0f3a..853a363f1 100644 --- a/Lib/test/test_concurrent_futures/test_interpreter_pool.py +++ b/Lib/test/test_concurrent_futures/test_interpreter_pool.py @@ -129,12 +129,12 @@ class InterpreterPoolExecutorTest( """ os.write(w, b'\0') - executor = self.executor_type(initializer=initscript) - before_init = os.read(r, 100) - fut = executor.submit(script) - after_init = read_msg(r) - fut.result() - after_run = read_msg(r) + with self.executor_type(initializer=initscript) as executor: + before_init = os.read(r, 100) + fut = executor.submit(script) + after_init = read_msg(r) + fut.result() + after_run = read_msg(r) self.assertEqual(before_init, b'\0') self.assertEqual(after_init, msg1) @@ -151,11 +151,11 @@ class InterpreterPoolExecutorTest( r, w = self.pipe() os.write(w, b'\0') - executor = self.executor_type( - initializer=write_msg, initargs=(w, msg)) - before = os.read(r, 100) - executor.submit(mul, 10, 10) - after = read_msg(r) + with self.executor_type( + initializer=write_msg, initargs=(w, msg)) as executor: + before = os.read(r, 100) + executor.submit(mul, 10, 10) + after = read_msg(r) self.assertEqual(before, b'\0') self.assertEqual(after, msg) @@ -261,11 +261,10 @@ class InterpreterPoolExecutorTest( import os os.write({w}, __name__.encode('utf-8') + b'\\0') """ - executor = self.executor_type() - - fut = executor.submit(script) - res = fut.result() - after = read_msg(r) + with self.executor_type() as executor: + fut = executor.submit(script) + res = fut.result() + after = read_msg(r) self.assertEqual(after, b'__main__') self.assertIs(res, None) @@ -279,25 +278,24 @@ class InterpreterPoolExecutorTest( spam += 1 return spam - executor = self.executor_type() + with self.executor_type() as executor: + fut = executor.submit(task1) + with self.assertRaises(_interpreters.NotShareableError): + fut.result() - fut = executor.submit(task1) - with self.assertRaises(_interpreters.NotShareableError): - fut.result() - - fut = executor.submit(task2) - with self.assertRaises(_interpreters.NotShareableError): - fut.result() + fut = executor.submit(task2) + with self.assertRaises(_interpreters.NotShareableError): + fut.result() def test_submit_local_instance(self): class Spam: def __init__(self): self.value = True - executor = self.executor_type() - fut = executor.submit(Spam) - with self.assertRaises(_interpreters.NotShareableError): - fut.result() + with self.executor_type() as executor: + fut = executor.submit(Spam) + with self.assertRaises(_interpreters.NotShareableError): + fut.result() def test_submit_instance_method(self): class Spam: @@ -305,15 +303,15 @@ class InterpreterPoolExecutorTest( return True spam = Spam() - executor = self.executor_type() - fut = executor.submit(spam.run) - with self.assertRaises(_interpreters.NotShareableError): - fut.result() + with self.executor_type() as executor: + fut = executor.submit(spam.run) + with self.assertRaises(_interpreters.NotShareableError): + fut.result() def test_submit_func_globals(self): - executor = self.executor_type() - fut = executor.submit(get_current_name) - name = fut.result() + with self.executor_type() as executor: + fut = executor.submit(get_current_name) + name = fut.result() self.assertEqual(name, __name__) self.assertNotEqual(name, '__main__') diff --git a/Lib/test/test_concurrent_futures/test_process_pool.py b/Lib/test/test_concurrent_futures/test_process_pool.py index e99abaf65..c7d489c4f 100644 --- a/Lib/test/test_concurrent_futures/test_process_pool.py +++ b/Lib/test/test_concurrent_futures/test_process_pool.py @@ -3,13 +3,15 @@ import queue import sys import threading import time +import traceback import unittest import unittest.mock +import weakref from concurrent import futures from concurrent.futures.process import BrokenProcessPool from test import support -from test.support import hashlib_helper +from test.support import hashlib_helper, threading_helper, warnings_helper from test.test_importlib.metadata.fixtures import parameterize from .executor import ExecutorTest, mul @@ -25,7 +27,6 @@ class EventfulGCObj(): def __del__(self): self.event.set() - TERMINATE_WORKERS = futures.ProcessPoolExecutor.terminate_workers.__name__ KILL_WORKERS = futures.ProcessPoolExecutor.kill_workers.__name__ FORCE_SHUTDOWN_PARAMS = [ @@ -42,6 +43,12 @@ def _put_wait_put(queue, event): queue.put('finished') +def _report_wait_return(queue, event, value): + """ Used as part of _run_stranded_worker_exit_test """ + queue.put(value) + event.wait() + return value + class ProcessPoolExecutorTest(ExecutorTest): @@ -51,6 +58,7 @@ class ProcessPoolExecutorTest(ExecutorTest): "max_workers must be <= 61"): futures.ProcessPoolExecutor(max_workers=62) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_killed_child(self): # When a child process is abruptly terminated, the whole pool gets # "broken". @@ -63,6 +71,33 @@ class ProcessPoolExecutorTest(ExecutorTest): # Submitting other jobs fails as well. self.assertRaises(BrokenProcessPool, self.executor.submit, pow, 2, 8) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() + def test_broken_process_pool_traceback(self): + # When a child process is abruptly terminated, the whole pool gets + # "broken", and a BrokenProcessPool exception should be created + # for each future instead of sharing one exception among all futures. + event = self.create_event() + futures = [self.executor.submit(event.wait) for _ in range(3)] + p = next(iter(self.executor._processes.values())) + p.terminate() + for fut in futures: + # Don't use assertRaises(): it clears the traceback off exc. + try: + fut.result() + except BrokenProcessPool as exc: + tb = exc.__traceback__ + else: + self.fail("BrokenProcessPool not raised") + count = sum( + 1 + for frame_summary in traceback.extract_tb(tb) + if frame_summary.filename == __file__ + ) + # This code file should appear exactly once in the traceback. + # A shared exception would accumulate a frame per result() call. + self.assertEqual(count, 1) + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_map_chunksize(self): def bad_map(): list(self.executor.map(pow, range(40), range(40), chunksize=-1)) @@ -83,6 +118,7 @@ class ProcessPoolExecutorTest(ExecutorTest): def _test_traceback(cls): raise RuntimeError(123) # some comment + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_traceback(self): # We want ensure that the traceback from the child process is # contained in the traceback raised in the main process. @@ -105,6 +141,22 @@ class ProcessPoolExecutorTest(ExecutorTest): self.assertIn('raise RuntimeError(123) # some comment', f1.getvalue()) + def test_traceback_when_child_process_terminates_abruptly(self): + # gh-139462 enhancement - BrokenProcessPool exceptions + # should describe which process terminated. + exit_code = 99 + with self.executor_type(max_workers=1) as executor: + future = executor.submit(os._exit, exit_code) + with self.assertRaises(BrokenProcessPool) as bpe: + future.result() + + cause = bpe.exception.__cause__ + self.assertIsInstance(cause, futures.process._RemoteTraceback) + self.assertIn( + f"terminated abruptly with exit code {exit_code}", cause.tb + ) + + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @hashlib_helper.requires_hashdigest('md5') def test_ressources_gced_in_workers(self): # Ensure that argument for a job are correctly gc-ed after the job @@ -125,6 +177,7 @@ class ProcessPoolExecutorTest(ExecutorTest): mgr.shutdown() mgr.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_saturation(self): executor = self.executor mp_context = self.get_context() @@ -195,6 +248,153 @@ class ProcessPoolExecutorTest(ExecutorTest): executor = self.executor_type(1, max_tasks_per_child=3) self.assertEqual(executor._mp_context.get_start_method(), "spawn") + def test_max_tasks_per_child_pending_tasks_gh115634(self): + # gh-115634: A worker exiting at its max_tasks_per_child limit left a + # stale token in the idle worker semaphore, so no replacement worker + # was spawned and the remaining queued tasks deadlocked. Submit more + # tasks than the pool can run at once so a backlog is queued while + # workers hit their task limit. + context = self.get_context() + if context.get_start_method(allow_none=False) == "fork": + raise unittest.SkipTest("Incompatible with the fork start method.") + + for max_workers, max_tasks, num_tasks in [(1, 2, 6), (2, 2, 8)]: + with self.subTest(max_workers=max_workers, max_tasks=max_tasks): + executor = self.executor_type( + max_workers, mp_context=context, + max_tasks_per_child=max_tasks) + try: + futures = [executor.submit(mul, i, 2) + for i in range(num_tasks)] + # If the deadlock regresses, the result() calls time out, + # and the shutdown below hangs until the test timeout. + results = [f.result(timeout=support.SHORT_TIMEOUT) + for f in futures] + self.assertEqual(results, + [i * 2 for i in range(num_tasks)]) + finally: + executor.shutdown(wait=True, cancel_futures=True) + + def _run_stranded_worker_exit_test(self, *, shutdown, drop_reference): + # A worker exits upon reaching its max_tasks_per_child limit while + # more submitted work is queued. While the executor object is + # alive a replacement worker must be spawned and the remaining + # work executed; once it has been garbage collected no replacement + # is possible and the remaining futures must fail promptly instead + # of never resolving. + context = self.get_context() + if context.get_start_method(allow_none=False) == "fork": + raise unittest.SkipTest("Incompatible with the fork start method.") + manager = context.Manager() + self.addCleanup(manager.join) + self.addCleanup(manager.shutdown) + started = manager.Queue() + gate = manager.Event() + + executor = self.executor_type( + 1, mp_context=context, max_tasks_per_child=1) + futs = [executor.submit(_report_wait_return, started, gate, i) + for i in range(3)] + self.addCleanup(threading_helper.join_thread, + executor._executor_manager_thread) + # Wait until the worker is inside the first task so that it exits + # at its task limit only after the executor has been shut down + # and/or garbage collected below. + self.assertEqual(started.get(timeout=support.SHORT_TIMEOUT), 0) + if shutdown: + executor.shutdown(wait=False) + if drop_reference: + executor_ref = weakref.ref(executor) + executor = None + for _ in support.sleeping_retry(support.SHORT_TIMEOUT): + support.gc_collect() + if executor_ref() is None: + break + gate.set() + + self.assertEqual(futs[0].result(timeout=support.SHORT_TIMEOUT), 0) + if drop_reference: + for fut in futs[1:]: + with self.assertRaisesRegex(BrokenProcessPool, + "garbage collected"): + fut.result(timeout=support.SHORT_TIMEOUT) + else: + results = [f.result(timeout=support.SHORT_TIMEOUT) + for f in futs[1:]] + self.assertEqual(results, [1, 2]) + + def test_shutdown_no_wait_max_tasks_gh119592(self): + # gh-119592: shutdown(wait=False) used to clear executor state that + # worker replacement relied on. A worker exiting at its + # max_tasks_per_child limit afterwards could not be replaced, so + # the remaining submitted work never ran, and a racing worker exit + # could crash the manager thread on the partially cleared state. + for drop_reference in (False, True): + with self.subTest(drop_reference=drop_reference): + self._run_stranded_worker_exit_test( + shutdown=True, drop_reference=drop_reference) + + def test_gc_during_max_tasks_worker_exit_gh152967(self): + # gh-152967: If the executor was garbage collected without + # shutdown() while its last worker exited at its + # max_tasks_per_child limit, no replacement worker could be + # spawned and the remaining futures were never resolved. + self._run_stranded_worker_exit_test( + shutdown=False, drop_reference=True) + + def _run_unreplaceable_worker_exit_test(self, *, error_regex, + force_shutting_down=False, + failing_spawn=False): + # Drive a max_tasks_per_child worker exit while worker + # replacement is impossible; the queued futures must fail + # promptly with a BrokenProcessPool explaining why. + context = self.get_context() + if context.get_start_method(allow_none=False) == "fork": + raise unittest.SkipTest("Incompatible with the fork start method.") + manager = context.Manager() + self.addCleanup(manager.join) + self.addCleanup(manager.shutdown) + started = manager.Queue() + gate = manager.Event() + + executor = self.executor_type( + 1, mp_context=context, max_tasks_per_child=1) + futs = [executor.submit(_report_wait_return, started, gate, i) + for i in range(3)] + self.addCleanup(threading_helper.join_thread, + executor._executor_manager_thread) + self.assertEqual(started.get(timeout=support.SHORT_TIMEOUT), 0) + if force_shutting_down: + with executor._shutdown_lock: + executor._force_shutting_down = True + if failing_spawn: + spawn_patch = unittest.mock.patch( + "concurrent.futures.process._spawn_worker", + side_effect=OSError("spawn failed")) + spawn_patch.start() + self.addCleanup(spawn_patch.stop) + gate.set() + + self.assertEqual(futs[0].result(timeout=support.SHORT_TIMEOUT), 0) + for fut in futs[1:]: + with self.assertRaisesRegex(BrokenProcessPool, error_regex): + fut.result(timeout=support.SHORT_TIMEOUT) + + def test_force_shutdown_during_max_tasks_worker_exit(self): + # A worker exiting at its max_tasks_per_child limit during + # terminate_workers()/kill_workers() must not be replaced (the + # replacement would escape the kill); queued futures fail instead. + self._run_unreplaceable_worker_exit_test( + force_shutting_down=True, + error_regex="forcefully shut down") + + def test_failed_worker_replacement_breaks_pool(self): + # If no replacement worker can be started and no workers remain, + # the pool must break rather than strand the queued futures. + self._run_unreplaceable_worker_exit_test( + failing_spawn=True, + error_regex="could not be started") + def test_max_tasks_early_shutdown(self): context = self.get_context() if context.get_start_method(allow_none=False) == "fork": @@ -210,6 +410,7 @@ class ProcessPoolExecutorTest(ExecutorTest): for i, future in enumerate(futures): self.assertEqual(future.result(), mul(i, i)) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_python_finalization_error(self): # gh-109047: Catch RuntimeError on thread creation # during Python finalization. @@ -238,7 +439,6 @@ class ProcessPoolExecutorTest(ExecutorTest): list(executor.map(mul, [(2, 3)] * 10)) executor.shutdown() - def test_terminate_workers(self): mock_fn = unittest.mock.Mock() with self.executor_type(max_workers=1) as executor: @@ -261,6 +461,7 @@ class ProcessPoolExecutorTest(ExecutorTest): executor._force_shutdown, operation='invalid operation'), + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() @parameterize(*FORCE_SHUTDOWN_PARAMS) def test_force_shutdown_workers(self, function_name): manager = self.get_context().Manager() @@ -350,5 +551,6 @@ create_executor_tests(globals(), ProcessPoolExecutorTest, def setUpModule(): setup_module() + if __name__ == "__main__": unittest.main() diff --git a/Lib/test/test_concurrent_futures/test_shutdown.py b/Lib/test/test_concurrent_futures/test_shutdown.py index 99b315b47..129b5cb50 100644 --- a/Lib/test/test_concurrent_futures/test_shutdown.py +++ b/Lib/test/test_concurrent_futures/test_shutdown.py @@ -6,6 +6,7 @@ import unittest from concurrent import futures from test import support +from test.support import warnings_helper from test.support.script_helper import assert_python_ok from .util import ( @@ -49,6 +50,7 @@ class ExecutorShutdownTest: self.assertFalse(err) self.assertEqual(out.strip(), b"apple") + @support.force_not_colorized def test_submit_after_interpreter_shutdown(self): # Test the atexit hook for shutdown of worker threads and processes rc, out, err = assert_python_ok('-c', """if 1: @@ -77,12 +79,14 @@ class ExecutorShutdownTest: self.assertIn("RuntimeError: cannot schedule new futures", err.decode()) self.assertEqual(out.strip(), b"runtime-error") + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_hang_issue12364(self): fs = [self.executor.submit(time.sleep, 0.1) for _ in range(50)] self.executor.shutdown() for f in fs: f.result() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_cancel_futures(self): assert self.worker_count <= 5, "test needs few workers" fs = [self.executor.submit(time.sleep, .1) for _ in range(50)] @@ -111,23 +115,25 @@ class ExecutorShutdownTest: See https://github.com/python/cpython/issues/83386. """ - if self.executor_type == futures.ProcessPoolExecutor: - raise unittest.SkipTest( - "Hangs, see https://github.com/python/cpython/issues/83386") - rc, out, err = assert_python_ok('-c', """if True: from concurrent.futures import {executor_type} from test.test_concurrent_futures.test_shutdown import sleep_and_print if __name__ == "__main__": - if {context!r}: multiprocessing.set_start_method({context!r}) - t = {executor_type}(max_workers=3) + context = '{context}' + if context == "": + t = {executor_type}(max_workers=3) + else: + from multiprocessing import get_context + t = {executor_type}(max_workers=3, + mp_context=get_context(context)) t.submit(sleep_and_print, 1.0, "apple") t.shutdown(wait=False) """.format(executor_type=self.executor_type.__name__, - context=getattr(self, 'ctx', None))) + context=getattr(self, 'ctx', ''))) self.assertFalse(err) self.assertEqual(out.strip(), b"apple") + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_hang_gh94440(self): """shutdown(wait=True) doesn't hang when a future was submitted and quickly canceled right before shutdown. @@ -171,6 +177,7 @@ class ThreadPoolShutdownTest(ThreadPoolMixin, ExecutorShutdownTest, BaseTestCase for t in self.executor._threads: t.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_context_manager_shutdown(self): with futures.ThreadPoolExecutor(max_workers=5) as e: executor = e @@ -180,6 +187,7 @@ class ThreadPoolShutdownTest(ThreadPoolMixin, ExecutorShutdownTest, BaseTestCase for t in executor._threads: t.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_del_shutdown(self): executor = futures.ThreadPoolExecutor(max_workers=5) res = executor.map(abs, range(-5, 5)) @@ -193,6 +201,7 @@ class ThreadPoolShutdownTest(ThreadPoolMixin, ExecutorShutdownTest, BaseTestCase # executor got shutdown. assert all([r == abs(v) for r, v in zip(res, range(-5, 5))]) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_shutdown_no_wait(self): # Ensure that the executor cleans up the threads when calling # shutdown with wait=False @@ -207,7 +216,7 @@ class ThreadPoolShutdownTest(ThreadPoolMixin, ExecutorShutdownTest, BaseTestCase # executor got shutdown. assert all([r == abs(v) for r, v in zip(res, range(-5, 5))]) - + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_thread_names_assigned(self): executor = futures.ThreadPoolExecutor( max_workers=5, thread_name_prefix='SpecialPool') @@ -220,6 +229,7 @@ class ThreadPoolShutdownTest(ThreadPoolMixin, ExecutorShutdownTest, BaseTestCase self.assertRegex(t.name, r'^SpecialPool_[0-4]$') t.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_thread_names_default(self): executor = futures.ThreadPoolExecutor(max_workers=5) executor.map(abs, range(-5, 5)) @@ -234,8 +244,6 @@ class ThreadPoolShutdownTest(ThreadPoolMixin, ExecutorShutdownTest, BaseTestCase t.join() def test_cancel_futures_wait_false(self): - # Can only be reliably tested for TPE, since PPE often hangs with - # `wait=False` (even without *cancel_futures*). rc, out, err = assert_python_ok('-c', """if True: from concurrent.futures import ThreadPoolExecutor from test.test_concurrent_futures.test_shutdown import sleep_and_print @@ -253,6 +261,7 @@ class ThreadPoolShutdownTest(ThreadPoolMixin, ExecutorShutdownTest, BaseTestCase class ProcessPoolShutdownTest(ExecutorShutdownTest): + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_processes_terminate(self): def acquire_lock(lock): lock.acquire() @@ -276,6 +285,7 @@ class ProcessPoolShutdownTest(ExecutorShutdownTest): for p in processes.values(): p.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_context_manager_shutdown(self): with futures.ProcessPoolExecutor( max_workers=5, mp_context=self.get_context()) as e: @@ -286,6 +296,7 @@ class ProcessPoolShutdownTest(ExecutorShutdownTest): for p in processes.values(): p.join() + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_del_shutdown(self): executor = futures.ProcessPoolExecutor( max_workers=5, mp_context=self.get_context()) @@ -308,6 +319,7 @@ class ProcessPoolShutdownTest(ExecutorShutdownTest): # executor got shutdown. assert all([r == abs(v) for r, v in zip(res, range(-5, 5))]) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_shutdown_no_wait(self): # Ensure that the executor cleans up the processes when calling # shutdown with wait=False diff --git a/Lib/test/test_concurrent_futures/test_thread_pool.py b/Lib/test/test_concurrent_futures/test_thread_pool.py index 4324241b3..226b7acb6 100644 --- a/Lib/test/test_concurrent_futures/test_thread_pool.py +++ b/Lib/test/test_concurrent_futures/test_thread_pool.py @@ -7,6 +7,7 @@ import threading import unittest from concurrent import futures from test import support +from test.support import warnings_helper from .executor import ExecutorTest, mul from .util import BaseTestCase, ThreadPoolMixin, setup_module @@ -53,6 +54,7 @@ class ThreadPoolExecutorTest(ThreadPoolMixin, ExecutorTest, BaseTestCase): @support.requires_fork() @unittest.skipUnless(hasattr(os, 'register_at_fork'), 'need os.register_at_fork') @support.requires_resource('cpu') + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_hang_global_shutdown_lock(self): # bpo-45021: _global_shutdown_lock should be reinitialized in the child # process, otherwise it will never exit @@ -68,6 +70,7 @@ class ThreadPoolExecutorTest(ThreadPoolMixin, ExecutorTest, BaseTestCase): @support.requires_fork() @unittest.skipUnless(hasattr(os, 'register_at_fork'), 'need os.register_at_fork') + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_process_fork_from_a_threadpool(self): # bpo-43944: clear concurrent.futures.thread._threads_queues after fork, # otherwise child process will try to join parent thread diff --git a/Lib/test/test_concurrent_futures/test_wait.py b/Lib/test/test_concurrent_futures/test_wait.py index 6749a690f..b8250cec7 100644 --- a/Lib/test/test_concurrent_futures/test_wait.py +++ b/Lib/test/test_concurrent_futures/test_wait.py @@ -3,7 +3,7 @@ import threading import unittest from concurrent import futures from test import support -from test.support import threading_helper +from test.support import threading_helper, warnings_helper from .util import ( CANCELLED_FUTURE, CANCELLED_AND_NOTIFIED_FUTURE, EXCEPTION_FUTURE, @@ -22,6 +22,7 @@ def wait_and_raise(e): class WaitTests: + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_20369(self): # See https://bugs.python.org/issue20369 future = self.executor.submit(mul, 1, 2) @@ -30,7 +31,7 @@ class WaitTests: self.assertEqual({future}, done) self.assertEqual(set(), not_done) - + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_first_completed(self): event = self.create_event() future1 = self.executor.submit(mul, 21, 2) @@ -47,6 +48,7 @@ class WaitTests: event.set() future2.result() # wait for job to finish + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_first_completed_some_already_completed(self): event = self.create_event() future1 = self.executor.submit(event.wait) @@ -64,6 +66,7 @@ class WaitTests: event.set() future1.result() # wait for job to finish + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_first_exception(self): event1 = self.create_event() event2 = self.create_event() @@ -93,6 +96,7 @@ class WaitTests: event2.set() future3.result() # wait for job to finish + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_first_exception_some_already_complete(self): event = self.create_event() future1 = self.executor.submit(divmod, 21, 0) @@ -114,6 +118,7 @@ class WaitTests: event.set() future2.result() # wait for job to finish + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_first_exception_one_already_failed(self): event = self.create_event() future1 = self.executor.submit(event.wait) @@ -129,6 +134,7 @@ class WaitTests: event.set() future1.result() # wait for job to finish + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_all_completed(self): future1 = self.executor.submit(divmod, 2, 0) future2 = self.executor.submit(mul, 2, 21) @@ -148,6 +154,7 @@ class WaitTests: future2]), finished) self.assertEqual(set(), pending) + @warnings_helper.ignore_fork_in_thread_deprecation_warnings() def test_timeout(self): short_timeout = 0.050 @@ -200,5 +207,6 @@ create_executor_tests(globals(), WaitTests, def setUpModule(): setup_module() + if __name__ == "__main__": unittest.main() diff --git a/Lib/test/test_concurrent_futures/util.py b/Lib/test/test_concurrent_futures/util.py index 1ea5a16d8..1dad78f08 100644 --- a/Lib/test/test_concurrent_futures/util.py +++ b/Lib/test/test_concurrent_futures/util.py @@ -10,7 +10,7 @@ from concurrent.futures._base import ( from concurrent.futures.process import _check_system_limits from test import support -from test.support import threading_helper +from test.support import threading_helper, warnings_helper def create_future(state=PENDING, exception=None, result=None): @@ -51,7 +51,8 @@ class ExecutorMixin: max_workers=self.worker_count, mp_context=self.get_context(), **self.executor_kwargs) - self.manager = self.get_context().Manager() + with warnings_helper.ignore_fork_in_thread_deprecation_warnings(): + self.manager = self.get_context().Manager() else: self.executor = self.executor_type( max_workers=self.worker_count, @@ -138,6 +139,8 @@ class ProcessPoolForkserverMixin(ExecutorMixin): self.skipTest("require unix system") if support.check_sanitizer(thread=True): self.skipTest("TSAN doesn't support threads after fork") + if "forkserver" not in multiprocessing.get_all_start_methods(): + self.skipTest("forkserver start method is not available") return super().get_context() def create_event(self): @@ -146,6 +149,7 @@ class ProcessPoolForkserverMixin(ExecutorMixin): def create_executor_tests(remote_globals, mixin, bases=(BaseTestCase,), executor_mixins=(ThreadPoolMixin, + InterpreterPoolMixin, ProcessPoolForkMixin, ProcessPoolForkserverMixin, ProcessPoolSpawnMixin)): diff --git a/Lib/test/test_multiprocessing_forkserver/__init__.py b/Lib/test/test_multiprocessing_forkserver/__init__.py index d91715a34..162adf207 100644 --- a/Lib/test/test_multiprocessing_forkserver/__init__.py +++ b/Lib/test/test_multiprocessing_forkserver/__init__.py @@ -1,3 +1,4 @@ +import multiprocessing import os.path import sys import unittest @@ -9,5 +10,13 @@ if support.PGO: if sys.platform == "win32": raise unittest.SkipTest("forkserver is not available on Windows") +if not support.has_fork_support: + raise unittest.SkipTest("requires working os.fork()") + +# The forkserver start method requires passing file descriptors over a Unix +# socket, which is not available on every platform (e.g. Solaris/illumos). +if "forkserver" not in multiprocessing.get_all_start_methods(): + raise unittest.SkipTest("forkserver start method is not available") + def load_tests(*args): return support.load_package_tests(os.path.dirname(__file__), *args) diff --git a/Lib/test/test_multiprocessing_forkserver/test_preload.py b/Lib/test/test_multiprocessing_forkserver/test_preload.py new file mode 100644 index 000000000..f8e119aa3 --- /dev/null +++ b/Lib/test/test_multiprocessing_forkserver/test_preload.py @@ -0,0 +1,230 @@ +"""Tests for forkserver preload functionality.""" + +import contextlib +import multiprocessing +import os +import shutil +import sys +import tempfile +import unittest +from multiprocessing import forkserver, spawn + + +class TestForkserverPreload(unittest.TestCase): + """Tests for forkserver preload functionality.""" + + def setUp(self): + self._saved_warnoptions = sys.warnoptions.copy() + # Remove warning options that would convert ImportWarning to errors: + # - 'error' converts all warnings to errors + # - 'error::ImportWarning' specifically converts ImportWarning + # Keep other specific options like 'error::BytesWarning' that + # subprocess's _args_from_interpreter_flags() expects to remove + sys.warnoptions[:] = [ + opt for opt in sys.warnoptions + if opt not in ('error', 'error::ImportWarning') + ] + self.ctx = multiprocessing.get_context('forkserver') + forkserver._forkserver._stop() + + def tearDown(self): + sys.warnoptions[:] = self._saved_warnoptions + forkserver._forkserver._stop() + + @staticmethod + def _send_value(conn, value): + """Send value through connection. Static method to be picklable as Process target.""" + conn.send(value) + + @contextlib.contextmanager + def capture_forkserver_stderr(self): + """Capture stderr from forkserver by preloading a module that redirects it. + + Yields (module_name, capture_file_path). The capture file can be read + after the forkserver has processed preloads. This works because + forkserver.main() calls util._flush_std_streams() after preloading, + ensuring captured output is written before we read it. + """ + tmpdir = tempfile.mkdtemp() + capture_module = os.path.join(tmpdir, '_capture_stderr.py') + capture_file = os.path.join(tmpdir, 'stderr.txt') + try: + with open(capture_module, 'w') as f: + # Use line buffering (buffering=1) to ensure warnings are written. + # Enable ImportWarning since it's ignored by default. + f.write( + f'import sys, warnings; ' + f'sys.stderr = open({capture_file!r}, "w", buffering=1); ' + f'warnings.filterwarnings("always", category=ImportWarning)\n' + ) + sys.path.insert(0, tmpdir) + yield '_capture_stderr', capture_file + finally: + sys.path.remove(tmpdir) + shutil.rmtree(tmpdir, ignore_errors=True) + + def test_preload_on_error_ignore_default(self): + """Test that invalid modules are silently ignored by default.""" + self.ctx.set_forkserver_preload(['nonexistent_module_xyz']) + + r, w = self.ctx.Pipe(duplex=False) + p = self.ctx.Process(target=self._send_value, args=(w, 42)) + p.start() + w.close() + result = r.recv() + r.close() + p.join() + + self.assertEqual(result, 42) + self.assertEqual(p.exitcode, 0) + + def test_preload_on_error_ignore_explicit(self): + """Test that invalid modules are silently ignored with on_error='ignore'.""" + self.ctx.set_forkserver_preload(['nonexistent_module_xyz'], on_error='ignore') + + r, w = self.ctx.Pipe(duplex=False) + p = self.ctx.Process(target=self._send_value, args=(w, 99)) + p.start() + w.close() + result = r.recv() + r.close() + p.join() + + self.assertEqual(result, 99) + self.assertEqual(p.exitcode, 0) + + def test_preload_on_error_warn(self): + """Test that invalid modules emit warnings with on_error='warn'.""" + with self.capture_forkserver_stderr() as (capture_mod, stderr_file): + self.ctx.set_forkserver_preload( + [capture_mod, 'nonexistent_module_xyz'], on_error='warn') + + r, w = self.ctx.Pipe(duplex=False) + p = self.ctx.Process(target=self._send_value, args=(w, 123)) + p.start() + w.close() + result = r.recv() + r.close() + p.join() + + self.assertEqual(result, 123) + self.assertEqual(p.exitcode, 0) + + with open(stderr_file) as f: + stderr_output = f.read() + self.assertIn('nonexistent_module_xyz', stderr_output) + self.assertIn('ImportWarning', stderr_output) + + def test_preload_on_error_fail_breaks_context(self): + """Test that invalid modules with on_error='fail' breaks the forkserver.""" + with self.capture_forkserver_stderr() as (capture_mod, stderr_file): + self.ctx.set_forkserver_preload( + [capture_mod, 'nonexistent_module_xyz'], on_error='fail') + + r, w = self.ctx.Pipe(duplex=False) + try: + p = self.ctx.Process(target=self._send_value, args=(w, 42)) + with self.assertRaises((EOFError, ConnectionError, BrokenPipeError)) as cm: + p.start() + notes = getattr(cm.exception, '__notes__', []) + self.assertTrue(notes, "Expected exception to have __notes__") + self.assertIn('Forkserver process may have crashed', notes[0]) + + with open(stderr_file) as f: + stderr_output = f.read() + self.assertIn('nonexistent_module_xyz', stderr_output) + self.assertIn('ModuleNotFoundError', stderr_output) + finally: + w.close() + r.close() + + def test_preload_valid_modules_with_on_error_fail(self): + """Test that valid modules work fine with on_error='fail'.""" + self.ctx.set_forkserver_preload(['os', 'sys'], on_error='fail') + + r, w = self.ctx.Pipe(duplex=False) + p = self.ctx.Process(target=self._send_value, args=(w, 'success')) + p.start() + w.close() + result = r.recv() + r.close() + p.join() + + self.assertEqual(result, 'success') + self.assertEqual(p.exitcode, 0) + + def test_preload_invalid_on_error_value(self): + """Test that invalid on_error values raise ValueError.""" + with self.assertRaises(ValueError) as cm: + self.ctx.set_forkserver_preload(['os'], on_error='invalid') + self.assertIn("on_error must be 'ignore', 'warn', or 'fail'", str(cm.exception)) + + +class TestHandlePreload(unittest.TestCase): + """Unit tests for _handle_preload() function.""" + + def setUp(self): + self._saved_main = sys.modules['__main__'] + + def tearDown(self): + spawn.old_main_modules.clear() + sys.modules['__main__'] = self._saved_main + + def test_handle_preload_main_on_error_fail(self): + """Test that __main__ import failures raise with on_error='fail'.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.py') as f: + f.write('raise RuntimeError("test error in __main__")\n') + f.flush() + with self.assertRaises(RuntimeError) as cm: + forkserver._handle_preload(['__main__'], main_path=f.name, on_error='fail') + self.assertIn("test error in __main__", str(cm.exception)) + + def test_handle_preload_main_on_error_warn(self): + """Test that __main__ import failures warn with on_error='warn'.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.py') as f: + f.write('raise ImportError("test import error")\n') + f.flush() + with self.assertWarns(ImportWarning) as cm: + forkserver._handle_preload(['__main__'], main_path=f.name, on_error='warn') + self.assertIn("Failed to preload __main__", str(cm.warning)) + self.assertIn("test import error", str(cm.warning)) + + def test_handle_preload_main_on_error_ignore(self): + """Test that __main__ import failures are ignored with on_error='ignore'.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.py') as f: + f.write('raise ImportError("test import error")\n') + f.flush() + forkserver._handle_preload(['__main__'], main_path=f.name, on_error='ignore') + + def test_handle_preload_main_valid(self): + """Test that valid __main__ preload works.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.py') as f: + f.write('test_var = 42\n') + f.flush() + forkserver._handle_preload(['__main__'], main_path=f.name, on_error='fail') + + def test_handle_preload_module_on_error_fail(self): + """Test that module import failures raise with on_error='fail'.""" + with self.assertRaises(ModuleNotFoundError): + forkserver._handle_preload(['nonexistent_test_module_xyz'], on_error='fail') + + def test_handle_preload_module_on_error_warn(self): + """Test that module import failures warn with on_error='warn'.""" + with self.assertWarns(ImportWarning) as cm: + forkserver._handle_preload(['nonexistent_test_module_xyz'], on_error='warn') + self.assertIn("Failed to preload module", str(cm.warning)) + + def test_handle_preload_module_on_error_ignore(self): + """Test that module import failures are ignored with on_error='ignore'.""" + forkserver._handle_preload(['nonexistent_test_module_xyz'], on_error='ignore') + + def test_handle_preload_combined(self): + """Test preloading both __main__ and modules.""" + with tempfile.NamedTemporaryFile(mode='w', suffix='.py') as f: + f.write('import sys\n') + f.flush() + forkserver._handle_preload(['__main__', 'os', 'sys'], main_path=f.name, on_error='fail') + + +if __name__ == '__main__': + unittest.main() diff --git a/Lib/test/test_multiprocessing_main_handling.py b/Lib/test/test_multiprocessing_main_handling.py index 6b30a8931..fbd44fb0d 100644 --- a/Lib/test/test_multiprocessing_main_handling.py +++ b/Lib/test/test_multiprocessing_main_handling.py @@ -196,9 +196,8 @@ class MultiProcessingCmdLineMixin(): def test_script_compiled(self): with os_helper.temp_dir() as script_dir: script_name = _make_test_script(script_dir, 'script') - py_compile.compile(script_name, doraise=True) + pyc_file = import_helper.make_legacy_pyc(script_name, allow_compile=True) os.remove(script_name) - pyc_file = import_helper.make_legacy_pyc(script_name) self._check_script(pyc_file) def test_directory(self): @@ -213,9 +212,8 @@ class MultiProcessingCmdLineMixin(): with os_helper.temp_dir() as script_dir: script_name = _make_test_script(script_dir, '__main__', source=source) - py_compile.compile(script_name, doraise=True) + pyc_file = import_helper.make_legacy_pyc(script_name, allow_compile=True) os.remove(script_name) - pyc_file = import_helper.make_legacy_pyc(script_name) self._check_script(script_dir) def test_zipfile(self): @@ -231,7 +229,8 @@ class MultiProcessingCmdLineMixin(): with os_helper.temp_dir() as script_dir: script_name = _make_test_script(script_dir, '__main__', source=source) - compiled_name = py_compile.compile(script_name, doraise=True) + compiled_name = script_name + 'c' + py_compile.compile(script_name, compiled_name, doraise=True) zip_name, run_name = make_zip_script(script_dir, 'test_zip', compiled_name) self._check_script(zip_name) @@ -273,9 +272,8 @@ class MultiProcessingCmdLineMixin(): make_pkg(pkg_dir) script_name = _make_test_script(pkg_dir, '__main__', source=source) - compiled_name = py_compile.compile(script_name, doraise=True) + pyc_file = import_helper.make_legacy_pyc(script_name, allow_compile=True) os.remove(script_name) - pyc_file = import_helper.make_legacy_pyc(script_name) launch_name = _make_launch_script(script_dir, 'launch', 'test_pkg') self._check_script(launch_name) diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py index 4a7e52cbc..9c737e39e 100644 --- a/Lib/test/test_subprocess.py +++ b/Lib/test/test_subprocess.py @@ -1130,6 +1130,39 @@ class ProcessTestCase(BaseTestCase): p.kill() p.wait() + def test_communicate_timeout_resume_partial_write(self): + """Resume writing input after a partial-write TimeoutExpired. + + Exercises the _input_offset bookkeeping across the + _communicate_io_posix factoring: a first communicate() must time out + mid-write, and a subsequent communicate() must finish delivering the + remaining bytes so the child receives the full input intact. + """ + # 1 MiB easily exceeds typical pipe buffers (~64 KiB) so writing + # blocks once the buffer fills before the child starts reading. + input_data = bytes(range(256)) * 4096 # 1 MiB, distinctive pattern + self.assertEqual(len(input_data), 1024 * 1024) + + p = subprocess.Popen( + [sys.executable, "-c", + "import sys, time; " + "time.sleep(0.5); " + "sys.stdout.buffer.write(sys.stdin.buffer.read())"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + try: + with self.assertRaises(subprocess.TimeoutExpired): + p.communicate(input_data, timeout=0.05) + + # Resume: no new input, generous timeout to avoid CI flakes. + stdout, stderr = p.communicate(timeout=support.LONG_TIMEOUT) + self.assertEqual(len(stdout), len(input_data)) + self.assertEqual(stdout, input_data) + finally: + p.kill() + p.wait() + # Test for the fd leak reported in http://bugs.python.org/issue2791. def test_communicate_pipe_fd_leak(self): for stdin_pipe in (False, True): @@ -1463,6 +1496,8 @@ class ProcessTestCase(BaseTestCase): def test_wait_timeout(self): p = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(0.3)"]) + with self.assertRaises(subprocess.TimeoutExpired) as c: + p.wait(timeout=0) with self.assertRaises(subprocess.TimeoutExpired) as c: p.wait(timeout=0.0001) self.assertIn("0.0001", str(c.exception)) # For coverage of __str__. @@ -2016,8 +2051,9 @@ class RunFuncTestCase(BaseTestCase): run("echo hello", shell=True, text=True) check_output("echo hello", shell=True, text=True) """) + env = support.make_clean_env() cp = subprocess.run([sys.executable, "-Xwarn_default_encoding", "-c", code], - capture_output=True) + capture_output=True, env=env) lines = cp.stderr.splitlines() self.assertEqual(len(lines), 2, lines) self.assertStartsWith(lines[0], b":2: EncodingWarning: ") @@ -4139,5 +4175,148 @@ class ContextManagerTests(BaseTestCase): self.assertTrue(proc.stdin.closed) + +class FastWaitTestCase(BaseTestCase): + """Tests for efficient (pidfd_open() + poll() / kqueue()) process + waiting in subprocess.Popen.wait(). + """ + CAN_USE_PIDFD_OPEN = subprocess._CAN_USE_PIDFD_OPEN + CAN_USE_KQUEUE = subprocess._CAN_USE_KQUEUE + COMMAND = [sys.executable, "-c", "import time; time.sleep(0.3)"] + WAIT_TIMEOUT = 0.0001 # 0.1 ms + + def assert_fast_waitpid_error(self, patch_point): + # Emulate a case where pidfd_open() or kqueue() fails. + # Busy-poll wait should be used as fallback. + exc = OSError(errno.EMFILE, os.strerror(errno.EMFILE)) + with mock.patch(patch_point, side_effect=exc) as m: + p = subprocess.Popen(self.COMMAND) + with self.assertRaises(subprocess.TimeoutExpired): + p.wait(self.WAIT_TIMEOUT) + self.assertEqual(p.wait(timeout=support.SHORT_TIMEOUT), 0) + self.assertTrue(m.called) + + @unittest.skipIf(not CAN_USE_PIDFD_OPEN, reason="needs pidfd_open()") + def test_wait_pidfd_open_error(self): + self.assert_fast_waitpid_error("os.pidfd_open") + + @unittest.skipIf(not CAN_USE_KQUEUE, reason="needs kqueue() for proc") + def test_wait_kqueue_error(self): + self.assert_fast_waitpid_error("select.kqueue") + + @unittest.skipIf(not CAN_USE_KQUEUE, reason="needs kqueue() for proc") + def test_kqueue_control_error(self): + # Emulate a case where kqueue.control() fails. Busy-poll wait + # should be used as fallback. + p = subprocess.Popen(self.COMMAND) + kq_mock = mock.Mock() + kq_mock.control.side_effect = OSError( + errno.EPERM, os.strerror(errno.EPERM) + ) + kq_mock.close = mock.Mock() + + with mock.patch("select.kqueue", return_value=kq_mock) as m: + with self.assertRaises(subprocess.TimeoutExpired): + p.wait(self.WAIT_TIMEOUT) + self.assertEqual(p.wait(timeout=support.SHORT_TIMEOUT), 0) + self.assertTrue(m.called) + + def assert_wait_race_condition(self, patch_target, real_func): + # Call pidfd_open() / kqueue(), then terminate the process. + # Make sure that the wait call (poll() / kqueue.control()) + # still works for a terminated PID. + p = subprocess.Popen(self.COMMAND) + + def wrapper(*args, **kwargs): + ret = real_func(*args, **kwargs) + try: + os.kill(p.pid, signal.SIGTERM) + os.waitpid(p.pid, 0) + except OSError: + pass + return ret + + with mock.patch(patch_target, side_effect=wrapper) as m: + status = p.wait(timeout=support.SHORT_TIMEOUT) + self.assertTrue(m.called) + self.assertEqual(status, 0) + + @unittest.skipIf(not CAN_USE_PIDFD_OPEN, reason="needs pidfd_open()") + def test_pidfd_open_race(self): + self.assert_wait_race_condition("os.pidfd_open", os.pidfd_open) + + @unittest.skipIf(not CAN_USE_KQUEUE, reason="needs kqueue() for proc") + def test_kqueue_race(self): + self.assert_wait_race_condition("select.kqueue", select.kqueue) + + def assert_notification_without_immediate_reap(self, patch_target): + # Verify fallback to busy polling when poll() / kqueue() + # succeeds, but waitpid(pid, WNOHANG) returns (0, 0). + def waitpid_wrapper(pid, flags): + nonlocal ncalls + ncalls += 1 + if ncalls == 1: + return (0, 0) + return real_waitpid(pid, flags) + + ncalls = 0 + real_waitpid = os.waitpid + with mock.patch.object(subprocess.Popen, patch_target, return_value=True) as m1: + with mock.patch("os.waitpid", side_effect=waitpid_wrapper) as m2: + p = subprocess.Popen(self.COMMAND) + with self.assertRaises(subprocess.TimeoutExpired): + p.wait(self.WAIT_TIMEOUT) + self.assertEqual(p.wait(timeout=support.SHORT_TIMEOUT), 0) + self.assertTrue(m1.called) + self.assertTrue(m2.called) + + @unittest.skipIf(not CAN_USE_PIDFD_OPEN, reason="needs pidfd_open()") + def test_pidfd_open_notification_without_immediate_reap(self): + self.assert_notification_without_immediate_reap("_wait_pidfd") + + @unittest.skipIf(not CAN_USE_KQUEUE, reason="needs kqueue() for proc") + def test_kqueue_notification_without_immediate_reap(self): + self.assert_notification_without_immediate_reap("_wait_kqueue") + + @unittest.skipUnless( + CAN_USE_PIDFD_OPEN or CAN_USE_KQUEUE, + "fast wait mechanism not available" + ) + def test_fast_path_avoid_busy_loop(self): + # assert that the busy loop is not called as long as the fast + # wait is available + with mock.patch('time.sleep') as m: + p = subprocess.Popen(self.COMMAND) + with self.assertRaises(subprocess.TimeoutExpired): + p.wait(self.WAIT_TIMEOUT) + self.assertEqual(p.wait(timeout=support.LONG_TIMEOUT), 0) + self.assertFalse(m.called) + + @unittest.skipIf(mswindows, "requires the POSIX wait implementation") + def test_wait_huge_timeout(self): + # gh-154836: very large timeout values used to overflow the C + # timestamp conversion in poll() / kqueue.control() and raise + # OverflowError / TypeError. + for timeout in (10**10, sys.maxsize, float('inf')): + with self.subTest(timeout=timeout): + p = subprocess.Popen(ZERO_RETURN_CMD) + self.assertEqual(p.wait(timeout=timeout), 0) + + @unittest.skipIf(mswindows, "requires the POSIX wait implementation") + def test_run_huge_timeout(self): + # gh-154836: same as test_wait_huge_timeout, via the + # subprocess.run() / communicate() code path. + cp = subprocess.run(ZERO_RETURN_CMD, timeout=1e10) + self.assertEqual(cp.returncode, 0) + + @unittest.skipIf(mswindows, "requires the POSIX wait implementation") + def test_wait_slices_do_not_expire_early(self): + # A clamped wait slice must not raise TimeoutExpired before the + # real deadline: with a tiny slice limit, a process that + # outlives many slices must still be waited for successfully. + with mock.patch.object(subprocess, "_MAXIMUM_WAIT_TIMEOUT", 0.01): + p = subprocess.Popen(self.COMMAND) # sleeps 0.3s + self.assertEqual(p.wait(timeout=support.SHORT_TIMEOUT), 0) + if __name__ == "__main__": unittest.main()