pythongh-135427: Fix DeprecationWarning for os.fork when run in threads with -Werror (pythonGH-136796)

Don't ignore errors raised by `PyErr_WarnFormat` in `warn_about_fork_with_threads`
Instead, ignore the warnings in all test code that forks. (That's a lot of functions.)

In `test_support`, make `ignore_warnings` a context manager (as well as decorator),
and add a `message` argument to it.
Also add a `ignore_fork_in_thread_deprecation_warnings` helper for the deadlock-in-fork
warning.

Author: rani-pinchuk

Lib/test/_test_multiprocessing.py

@@ -1,11 +1,11 @@
 import contextlib
-import functools
 import importlib
 import re
 import sys
 import warnings
 
 
+
 def import_deprecated(name):
     """Import *name* while suppressing DeprecationWarning."""
     with warnings.catch_warnings():
@@ -42,20 +42,32 @@ def check_syntax_warning(testcase, statement, errtext='',
     testcase.assertEqual(warns, [])
 
 
-def ignore_warnings(*, category):
+@contextlib.contextmanager
+def ignore_warnings(*, category, message=''):
     """Decorator to suppress warnings.
 
-    Use of context managers to hide warnings make diffs
-    more noisy and tools like 'git blame' less useful.
+    Can also be used as a context manager. This is not preferred,
+    because it makes diffs more noisy and tools like 'git blame' less useful.
+    But, it's useful for async functions.
     """
-    def decorator(test):
-        @functools.wraps(test)
-        def wrapper(self, *args, **kwargs):
-            with warnings.catch_warnings():
-                warnings.simplefilter('ignore', category=category)
-                return test(self, *args, **kwargs)
-        return wrapper
-    return decorator
+    with warnings.catch_warnings():
+        warnings.filterwarnings('ignore', category=category, message=message)
+        yield
+
+
+@contextlib.contextmanager
+def ignore_fork_in_thread_deprecation_warnings():
+    """Suppress deprecation warnings related to forking in multi-threaded code.
+
+    See gh-135427
+
+    Can be used as decorator (preferred) or context manager.
+    """
+    with ignore_warnings(
+        message=".*fork.*may lead to deadlocks in the child.*",
+        category=DeprecationWarning,
+    ):
+        yield
 
 
 class WarningsRecorder(object):

Lib/test/support/warnings_helper.py

@@ -15,7 +15,7 @@
 from unittest import mock
 
 from test import support
-from test.support import os_helper
+from test.support import os_helper, warnings_helper
 from test.support import socket_helper
 from test.support import wait_process
 from test.support import hashlib_helper
@@ -1183,29 +1183,31 @@ async def runner():
 class TestFork(unittest.IsolatedAsyncioTestCase):
 
     async def test_fork_not_share_event_loop(self):
-        # The forked process should not share the event loop with the parent
-        loop = asyncio.get_running_loop()
-        r, w = os.pipe()
-        self.addCleanup(os.close, r)
-        self.addCleanup(os.close, w)
-        pid = os.fork()
-        if pid == 0:
-            # child
-            try:
-                loop = asyncio.get_event_loop()
-                os.write(w, b'LOOP:' + str(id(loop)).encode())
-            except RuntimeError:
-                os.write(w, b'NO LOOP')
-            except BaseException as e:
-                os.write(w, b'ERROR:' + ascii(e).encode())
-            finally:
-                os._exit(0)
-        else:
-            # parent
-            result = os.read(r, 100)
-            self.assertEqual(result, b'NO LOOP')
-            wait_process(pid, exitcode=0)
-
+        with warnings_helper.ignore_fork_in_thread_deprecation_warnings():
+            # The forked process should not share the event loop with the parent
+            loop = asyncio.get_running_loop()
+            r, w = os.pipe()
+            self.addCleanup(os.close, r)
+            self.addCleanup(os.close, w)
+            pid = os.fork()
+            if pid == 0:
+                # child
+                try:
+                    loop = asyncio.get_event_loop()
+                    os.write(w, b'LOOP:' + str(id(loop)).encode())
+                except RuntimeError:
+                    os.write(w, b'NO LOOP')
+                except BaseException as e:
+                    os.write(w, b'ERROR:' + ascii(e).encode())
+                finally:
+                    os._exit(0)
+            else:
+                # parent
+                result = os.read(r, 100)
+                self.assertEqual(result, b'NO LOOP')
+                wait_process(pid, exitcode=0)
+
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @hashlib_helper.requires_hashdigest('md5')
     @support.skip_if_sanitizer("TSAN doesn't support threads after fork", thread=True)
     def test_fork_signal_handling(self):
@@ -1253,6 +1255,7 @@ async def func():
         self.assertFalse(parent_handled.is_set())
         self.assertTrue(child_handled.is_set())
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @hashlib_helper.requires_hashdigest('md5')
     @support.skip_if_sanitizer("TSAN doesn't support threads after fork", thread=True)
     def test_fork_asyncio_run(self):
@@ -1273,6 +1276,7 @@ async def child_main():
 
         self.assertEqual(result.value, 42)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @hashlib_helper.requires_hashdigest('md5')
     @support.skip_if_sanitizer("TSAN doesn't support threads after fork", thread=True)
     def test_fork_asyncio_subprocess(self):

Lib/test/test_asyncio/test_unix_events.py

@@ -31,6 +31,7 @@
 from test import support
 from test.support import cpython_only, swap_attr
 from test.support import async_yield, run_yielding_async_fn
+from test.support import warnings_helper
 from test.support.import_helper import import_module
 from test.support.os_helper import (EnvironmentVarGuard, TESTFN, unlink)
 from test.support.script_helper import assert_python_ok
@@ -2545,6 +2546,7 @@ def run_child(self, child, terminal_input):
         finally:
             signal.signal(signal.SIGHUP, old_sighup)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def _run_child(self, child, terminal_input):
         r, w = os.pipe()  # Pipe test results from child back to parent
         try:

Lib/test/test_builtin.py

@@ -5,7 +5,7 @@
 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):
@@ -43,10 +43,12 @@ 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 @@ def test_submit_keyword(self):
         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,13 +69,15 @@ def test_map(self):
                 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))
         with self.assertRaises(ZeroDivisionError):
             i.__next__()
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @support.requires_resource('walltime')
     def test_map_timeout(self):
         results = []
@@ -108,26 +113,30 @@ def test_map_buffersize_value_validation(self):
                 ):
                     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):
             with self.subTest(buffersize=buffersize):
                 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):
             with self.subTest(buffersize=buffersize):
                 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,
@@ -147,6 +156,7 @@ 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
@@ -158,13 +168,15 @@ def test_map_buffersize_when_buffer_is_full(self):
             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
         # have exited).
         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
@@ -209,6 +221,7 @@ def test_max_workers_negative(self):
                                         "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.
@@ -221,6 +234,7 @@ def test_free_reference(self):
                 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/executor.py

@@ -7,6 +7,7 @@
     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 @@ def test_no_timeout(self):
                  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 @@ def test_future_times_out(self):
                 # 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_as_completed.py

@@ -10,6 +10,7 @@
 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,6 +112,7 @@ def _fail_on_deadlock(self, executor):
         print(f"\nTraceback:\n {tb}", file=sys.__stderr__)
         self.fail(f"Executor deadlock:\n\n{tb}")
 
+    @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)
@@ -199,6 +201,7 @@ def test_exit_during_result_unpickle_in_result_handler(self):
         # the result_handler thread
         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
@@ -212,6 +215,7 @@ def test_shutdown_deadlock(self):
             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.
@@ -238,6 +242,7 @@ def test_shutdown_deadlock_pickle(self):
         # 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 instead of a deadlock when a
@@ -254,6 +259,7 @@ def test_crash_big_data(self):
 
         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_deadlock.py

@@ -11,6 +11,7 @@
 from logging.handlers import QueueHandler
 
 from test import support
+from test.support import warnings_helper
 
 from .util import ExecutorMixin, create_executor_tests, setup_module
 
@@ -48,6 +49,7 @@ def setUp(self):
                                     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)]
@@ -74,6 +76,7 @@ def setUp(self):
             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: