Align process lifecycle libraries with Python 3.15 (#9201)

Migrate ten canonical production modules and fifteen complete canonical test files, retaining existing RustPython guards. Nine modules change behavior; heap removes duplicated allocation checks.

Assisted-by: OpenAI Codex:model version unavailable

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}")

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()

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()

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:

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()
 

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,))

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."

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.

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)
 

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

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.

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.

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

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)
 

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__')

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()

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

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

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()

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)):

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)

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()

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)
 

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"<string>: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()