diff options
Diffstat (limited to 'Lib/test/test_concurrent_futures')
| -rw-r--r-- | Lib/test/test_concurrent_futures/test_future.py | 57 | ||||
| -rw-r--r-- | Lib/test/test_concurrent_futures/test_init.py | 4 | ||||
| -rw-r--r-- | Lib/test/test_concurrent_futures/test_interpreter_pool.py | 265 | ||||
| -rw-r--r-- | Lib/test/test_concurrent_futures/test_shutdown.py | 58 |
4 files changed, 332 insertions, 52 deletions
diff --git a/Lib/test/test_concurrent_futures/test_future.py b/Lib/test/test_concurrent_futures/test_future.py index 4066ea1ee4b..06b11a3bacf 100644 --- a/Lib/test/test_concurrent_futures/test_future.py +++ b/Lib/test/test_concurrent_futures/test_future.py @@ -6,6 +6,7 @@ from concurrent.futures._base import ( PENDING, RUNNING, CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED, Future) from test import support +from test.support import threading_helper from .util import ( PENDING_FUTURE, RUNNING_FUTURE, CANCELLED_FUTURE, @@ -282,6 +283,62 @@ class FutureTests(BaseTestCase): self.assertEqual(f.exception(), e) + def test_get_snapshot(self): + """Test the _get_snapshot method for atomic state retrieval.""" + # Test with a pending future + f = Future() + done, cancelled, result, exception = f._get_snapshot() + self.assertFalse(done) + self.assertFalse(cancelled) + self.assertIsNone(result) + self.assertIsNone(exception) + + # Test with a finished future (successful result) + f = Future() + f.set_result(42) + done, cancelled, result, exception = f._get_snapshot() + self.assertTrue(done) + self.assertFalse(cancelled) + self.assertEqual(result, 42) + self.assertIsNone(exception) + + # Test with a finished future (exception) + f = Future() + exc = ValueError("test error") + f.set_exception(exc) + done, cancelled, result, exception = f._get_snapshot() + self.assertTrue(done) + self.assertFalse(cancelled) + self.assertIsNone(result) + self.assertIs(exception, exc) + + # Test with a cancelled future + f = Future() + f.cancel() + done, cancelled, result, exception = f._get_snapshot() + self.assertTrue(done) + self.assertTrue(cancelled) + self.assertIsNone(result) + self.assertIsNone(exception) + + # Test concurrent access (basic thread safety check) + f = Future() + f.set_result(100) + results = [] + + def get_snapshot(): + for _ in range(1000): + snapshot = f._get_snapshot() + results.append(snapshot) + + threads = [threading.Thread(target=get_snapshot) for _ in range(4)] + with threading_helper.start_threads(threads): + pass + # All snapshots should be identical for a finished future + expected = (True, False, 100, None) + for result in results: + self.assertEqual(result, expected) + def setUpModule(): setup_module() diff --git a/Lib/test/test_concurrent_futures/test_init.py b/Lib/test/test_concurrent_futures/test_init.py index df640929309..6b8484c0d5f 100644 --- a/Lib/test/test_concurrent_futures/test_init.py +++ b/Lib/test/test_concurrent_futures/test_init.py @@ -20,6 +20,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 diff --git a/Lib/test/test_concurrent_futures/test_interpreter_pool.py b/Lib/test/test_concurrent_futures/test_interpreter_pool.py index f6c62ae4b20..844dfdd6fc9 100644 --- a/Lib/test/test_concurrent_futures/test_interpreter_pool.py +++ b/Lib/test/test_concurrent_futures/test_interpreter_pool.py @@ -2,35 +2,78 @@ import asyncio import contextlib import io import os -import pickle +import sys import time import unittest -from concurrent.futures.interpreter import ( - ExecutionFailed, BrokenInterpreterPool, -) +from concurrent.futures.interpreter import BrokenInterpreterPool +from concurrent import interpreters +from concurrent.interpreters import _queues as queues import _interpreters from test import support +from test.support import os_helper +from test.support import script_helper import test.test_asyncio.utils as testasyncio_utils -from test.support.interpreters import queues from .executor import ExecutorTest, mul from .util import BaseTestCase, InterpreterPoolMixin, setup_module +WINDOWS = sys.platform.startswith('win') + + +@contextlib.contextmanager +def nonblocking(fd): + blocking = os.get_blocking(fd) + if blocking: + os.set_blocking(fd, False) + try: + yield + finally: + if blocking: + os.set_blocking(fd, blocking) + + +def read_file_with_timeout(fd, nbytes, timeout): + with nonblocking(fd): + end = time.time() + timeout + try: + return os.read(fd, nbytes) + except BlockingIOError: + pass + while time.time() < end: + try: + return os.read(fd, nbytes) + except BlockingIOError: + continue + else: + raise TimeoutError('nothing to read') + + +if not WINDOWS: + import select + def read_file_with_timeout(fd, nbytes, timeout): + r, _, _ = select.select([fd], [], [], timeout) + if fd not in r: + raise TimeoutError('nothing to read') + return os.read(fd, nbytes) + + def noop(): pass def write_msg(fd, msg): + import os os.write(fd, msg + b'\0') -def read_msg(fd): +def read_msg(fd, timeout=10.0): msg = b'' - while ch := os.read(fd, 1): - if ch == b'\0': - return msg + ch = read_file_with_timeout(fd, 1, timeout) + while ch != b'\0': msg += ch + ch = os.read(fd, 1) + return msg def get_current_name(): @@ -113,6 +156,38 @@ class InterpreterPoolExecutorTest( self.assertEqual(before, b'\0') self.assertEqual(after, msg) + def test_init_with___main___global(self): + # See https://github.com/python/cpython/pull/133957#issuecomment-2927415311. + text = """if True: + from concurrent.futures import InterpreterPoolExecutor + + INITIALIZER_STATUS = 'uninitialized' + + def init(x): + global INITIALIZER_STATUS + INITIALIZER_STATUS = x + INITIALIZER_STATUS + + def get_init_status(): + return INITIALIZER_STATUS + + if __name__ == "__main__": + exe = InterpreterPoolExecutor(initializer=init, + initargs=('initialized',)) + fut = exe.submit(get_init_status) + print(fut.result()) # 'initialized' + exe.shutdown(wait=True) + print(INITIALIZER_STATUS) # 'uninitialized' + """ + with os_helper.temp_dir() as tempdir: + filename = script_helper.make_script(tempdir, 'my-script', text) + res = script_helper.assert_python_ok(filename) + stdout = res.out.decode('utf-8').strip() + self.assertEqual(stdout.splitlines(), [ + 'initialized', + 'uninitialized', + ]) + def test_init_closure(self): count = 0 def init1(): @@ -121,10 +196,19 @@ class InterpreterPoolExecutorTest( nonlocal count count += 1 - with self.assertRaises(pickle.PicklingError): - self.executor_type(initializer=init1) - with self.assertRaises(pickle.PicklingError): - self.executor_type(initializer=init2) + with contextlib.redirect_stderr(io.StringIO()) as stderr: + with self.executor_type(initializer=init1) as executor: + fut = executor.submit(lambda: None) + self.assertIn('NotShareableError', stderr.getvalue()) + with self.assertRaises(BrokenInterpreterPool): + fut.result() + + with contextlib.redirect_stderr(io.StringIO()) as stderr: + with self.executor_type(initializer=init2) as executor: + fut = executor.submit(lambda: None) + self.assertIn('NotShareableError', stderr.getvalue()) + with self.assertRaises(BrokenInterpreterPool): + fut.result() def test_init_instance_method(self): class Spam: @@ -132,26 +216,12 @@ class InterpreterPoolExecutorTest( raise NotImplementedError spam = Spam() - with self.assertRaises(pickle.PicklingError): - self.executor_type(initializer=spam.initializer) - - def test_init_shared(self): - msg = b'eggs' - r, w = self.pipe() - script = f"""if True: - import os - if __name__ != '__main__': - import __main__ - spam = __main__.spam - os.write({w}, spam + b'\\0') - """ - - executor = self.executor_type(shared={'spam': msg}) - fut = executor.submit(exec, script) - fut.result() - after = read_msg(r) - - self.assertEqual(after, msg) + with contextlib.redirect_stderr(io.StringIO()) as stderr: + with self.executor_type(initializer=spam.initializer) as executor: + fut = executor.submit(lambda: None) + self.assertIn('NotShareableError', stderr.getvalue()) + with self.assertRaises(BrokenInterpreterPool): + fut.result() @unittest.expectedFailure def test_init_exception_in_script(self): @@ -178,8 +248,6 @@ class InterpreterPoolExecutorTest( stderr = stderr.getvalue() self.assertIn('ExecutionFailed: Exception: spam', stderr) self.assertIn('Uncaught in the interpreter:', stderr) - self.assertIn('The above exception was the direct cause of the following exception:', - stderr) @unittest.expectedFailure def test_submit_script(self): @@ -208,10 +276,14 @@ class InterpreterPoolExecutorTest( return spam executor = self.executor_type() - with self.assertRaises(pickle.PicklingError): - executor.submit(task1) - with self.assertRaises(pickle.PicklingError): - executor.submit(task2) + + fut = executor.submit(task1) + 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: @@ -219,8 +291,9 @@ class InterpreterPoolExecutorTest( self.value = True executor = self.executor_type() - with self.assertRaises(pickle.PicklingError): - executor.submit(Spam) + fut = executor.submit(Spam) + with self.assertRaises(_interpreters.NotShareableError): + fut.result() def test_submit_instance_method(self): class Spam: @@ -229,8 +302,9 @@ class InterpreterPoolExecutorTest( spam = Spam() executor = self.executor_type() - with self.assertRaises(pickle.PicklingError): - executor.submit(spam.run) + fut = executor.submit(spam.run) + with self.assertRaises(_interpreters.NotShareableError): + fut.result() def test_submit_func_globals(self): executor = self.executor_type() @@ -242,13 +316,14 @@ class InterpreterPoolExecutorTest( @unittest.expectedFailure def test_submit_exception_in_script(self): + # Scripts are not supported currently. fut = self.executor.submit('raise Exception("spam")') with self.assertRaises(Exception) as captured: fut.result() self.assertIs(type(captured.exception), Exception) self.assertEqual(str(captured.exception), 'spam') cause = captured.exception.__cause__ - self.assertIs(type(cause), ExecutionFailed) + self.assertIs(type(cause), interpreters.ExecutionFailed) for attr in ('__name__', '__qualname__', '__module__'): self.assertEqual(getattr(cause.excinfo.type, attr), getattr(Exception, attr)) @@ -261,7 +336,7 @@ class InterpreterPoolExecutorTest( self.assertIs(type(captured.exception), Exception) self.assertEqual(str(captured.exception), 'spam') cause = captured.exception.__cause__ - self.assertIs(type(cause), ExecutionFailed) + self.assertIs(type(cause), interpreters.ExecutionFailed) for attr in ('__name__', '__qualname__', '__module__'): self.assertEqual(getattr(cause.excinfo.type, attr), getattr(Exception, attr)) @@ -269,16 +344,93 @@ class InterpreterPoolExecutorTest( def test_saturation(self): blocker = queues.create() - executor = self.executor_type(4, shared=dict(blocker=blocker)) + executor = self.executor_type(4) for i in range(15 * executor._max_workers): - executor.submit(exec, 'import __main__; __main__.blocker.get()') - #executor.submit('blocker.get()') + executor.submit(blocker.get) self.assertEqual(len(executor._threads), executor._max_workers) for i in range(15 * executor._max_workers): blocker.put_nowait(None) executor.shutdown(wait=True) + def test_blocking(self): + # There is no guarantee that a worker will be created for every + # submitted task. That's because there's a race between: + # + # * a new worker thread, created when task A was just submitted, + # becoming non-idle when it picks up task A + # * after task B is added to the queue, a new worker thread + # is started only if there are no idle workers + # (the check in ThreadPoolExecutor._adjust_thread_count()) + # + # That means we must not block waiting for *all* tasks to report + # "ready" before we unblock the known-ready workers. + ready = queues.create() + blocker = queues.create() + + def run(taskid, ready, blocker): + # There can't be any globals here. + ready.put_nowait(taskid) + blocker.get() # blocking + + numtasks = 10 + futures = [] + with self.executor_type() as executor: + # Request the jobs. + for i in range(numtasks): + fut = executor.submit(run, i, ready, blocker) + futures.append(fut) + pending = numtasks + while pending > 0: + # Wait for any to be ready. + done = 0 + for _ in range(pending): + try: + ready.get(timeout=1) # blocking + except interpreters.QueueEmpty: + pass + else: + done += 1 + pending -= done + # Unblock the workers. + for _ in range(done): + blocker.put_nowait(None) + + def test_blocking_with_limited_workers(self): + # This is essentially the same as test_blocking, + # but we explicitly force a limited number of workers, + # instead of it happening implicitly sometimes due to a race. + ready = queues.create() + blocker = queues.create() + + def run(taskid, ready, blocker): + # There can't be any globals here. + ready.put_nowait(taskid) + blocker.get() # blocking + + numtasks = 10 + futures = [] + with self.executor_type(4) as executor: + # Request the jobs. + for i in range(numtasks): + fut = executor.submit(run, i, ready, blocker) + futures.append(fut) + pending = numtasks + while pending > 0: + # Wait for any to be ready. + done = 0 + for _ in range(pending): + try: + ready.get(timeout=1) # blocking + except interpreters.QueueEmpty: + pass + else: + done += 1 + pending -= done + # Unblock the workers. + for _ in range(done): + blocker.put_nowait(None) + @support.requires_gil_enabled("gh-117344: test is flaky without the GIL") def test_idle_thread_reuse(self): executor = self.executor_type() @@ -289,12 +441,21 @@ class InterpreterPoolExecutorTest( executor.shutdown(wait=True) def test_pickle_errors_propagate(self): - # GH-125864: Pickle errors happen before the script tries to execute, so the - # queue used to wait infinitely. - + # GH-125864: Pickle errors happen before the script tries to execute, + # so the queue used to wait infinitely. fut = self.executor.submit(PickleShenanigans(0)) - with self.assertRaisesRegex(RuntimeError, "gotcha"): + expected = interpreters.NotShareableError + with self.assertRaisesRegex(expected, 'args not shareable') as cm: fut.result() + self.assertRegex(str(cm.exception.__cause__), 'unpickled') + + def test_no_stale_references(self): + # Weak references don't cross between interpreters. + raise unittest.SkipTest('not applicable') + + def test_free_reference(self): + # Weak references don't cross between interpreters. + raise unittest.SkipTest('not applicable') class AsyncioTest(InterpretersMixin, testasyncio_utils.TestCase): diff --git a/Lib/test/test_concurrent_futures/test_shutdown.py b/Lib/test/test_concurrent_futures/test_shutdown.py index 7a4065afd46..99b315b47e2 100644 --- a/Lib/test/test_concurrent_futures/test_shutdown.py +++ b/Lib/test/test_concurrent_futures/test_shutdown.py @@ -330,6 +330,64 @@ class ProcessPoolShutdownTest(ExecutorShutdownTest): # shutdown. assert all([r == abs(v) for r, v in zip(res, range(-5, 5))]) + @classmethod + def _failing_task_gh_132969(cls, n): + raise ValueError("failing task") + + @classmethod + def _good_task_gh_132969(cls, n): + time.sleep(0.1 * n) + return n + + def _run_test_issue_gh_132969(self, max_workers): + # max_workers=2 will repro exception + # max_workers=4 will repro exception and then hang + + # Repro conditions + # max_tasks_per_child=1 + # a task ends abnormally + # shutdown(wait=False) is called + start_method = self.get_context().get_start_method() + if (start_method == "fork" or + (start_method == "forkserver" and sys.platform.startswith("win"))): + self.skipTest(f"Skipping test for {start_method = }") + executor = futures.ProcessPoolExecutor( + max_workers=max_workers, + max_tasks_per_child=1, + mp_context=self.get_context()) + f1 = executor.submit(ProcessPoolShutdownTest._good_task_gh_132969, 1) + f2 = executor.submit(ProcessPoolShutdownTest._failing_task_gh_132969, 2) + f3 = executor.submit(ProcessPoolShutdownTest._good_task_gh_132969, 3) + result = 0 + try: + result += f1.result() + result += f2.result() + result += f3.result() + except ValueError: + # stop processing results upon first exception + pass + + # Ensure that the executor cleans up after called + # shutdown with wait=False + executor_manager_thread = executor._executor_manager_thread + executor.shutdown(wait=False) + time.sleep(0.2) + executor_manager_thread.join() + return result + + def test_shutdown_gh_132969_case_1(self): + # gh-132969: test that exception "object of type 'NoneType' has no len()" + # is not raised when shutdown(wait=False) is called. + result = self._run_test_issue_gh_132969(2) + self.assertEqual(result, 1) + + def test_shutdown_gh_132969_case_2(self): + # gh-132969: test that process does not hang and + # exception "object of type 'NoneType' has no len()" is not raised + # when shutdown(wait=False) is called. + result = self._run_test_issue_gh_132969(4) + self.assertEqual(result, 1) + create_executor_tests(globals(), ProcessPoolShutdownTest, executor_mixins=(ProcessPoolForkMixin, |