Jun 10, 2025 · Apr 30, 2025 · Apr 30, 2025 · Apr 30, 2025 · Apr 30, 2025 · Apr 30, 2025
diff --git a/Lib/concurrent/futures/process.py b/Lib/concurrent/futures/process.py
                self._executor_manager_thread_wakeup

    def _adjust_process_count(self):
        # gh-132969
        if self._processes is None:
            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
diff --git a/Lib/test/test_concurrent_futures/test_shutdown.py b/Lib/test/test_concurrent_futures/test_shutdown.py
    sys.stdout.flush()


 def failing_task_132969(n: int) -> int:
    raise ValueError("failing task")


 def good_task_132969(n: int) -> int:
    time.sleep(0.1 * n)
    return n



 class ExecutorShutdownTest:
    def test_run_after_shutdown(self):
        self.executor.shutdown()
        # shutdown.
        assert all([r == abs(v) for r, v in zip(res, range(-5, 5))])

    def _run_test_issue_132969(self, max_workers: int) -> int:
        if sys.platform == "win32":
            raise unittest.SkipTest("skip test since forkserver is not available on Windows")

        # max_workers=2 will repro exception
        # max_workers=4 will repro exception and then hang

        import multiprocessing as mp

        # Repro conditions
        #   max_tasks_per_child=1
        #   a task ends abnormally
        #   shutdown(wait=False) is called
        executor = futures.ProcessPoolExecutor(
                max_workers=max_workers,
                max_tasks_per_child=1,
                mp_context=mp.get_context("forkserver"))
        f1 = executor.submit(good_task_132969, 1)
        f2 = executor.submit(failing_task_132969, 2)
        f3 = executor.submit(good_task_132969, 3)
        result:int = 0
        try:
            result += f1.result()
            result += f2.result()
            result += f3.result()
        except ValueError:
            # stop processing results upon first exception
            pass

        executor.shutdown(wait=False)
        return result

    def test_shutdown_len_exception_132969(self):
        result = self._run_test_issue_132969(2)
        self.assertEqual(result, 1)

    def test_shutdown_process_hang_132969(self):
        result = self._run_test_issue_132969(4)
        self.assertEqual(result, 1)


 create_executor_tests(globals(), ProcessPoolShutdownTest,
                      executor_mixins=(ProcessPoolForkMixin,
diff --git a/Misc/NEWS.d/next/Library/2025-04-30-19-32-18.gh-issue-132969.EagQ3G.rst b/Misc/NEWS.d/next/Library/2025-04-30-19-32-18.gh-issue-132969.EagQ3G.rst
 Fixes error+hang when ProcessPoolExecutor shutdown called with wait=False and a task ended abnormally.
Original file line number	Diff line number	Diff line change
Expand Up		@@ -755,6 +755,10 @@ def _start_executor_manager_thread(self):
		self._executor_manager_thread_wakeup

		def _adjust_process_count(self):
		# gh-132969
ogbiggles marked this conversation as resolved. Show resolvedHide resolved
		if self._processes is None:
		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
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -20,6 +20,16 @@ def sleep_and_print(t, msg):
		sys.stdout.flush()


		def failing_task_132969(n: int) -> int:
ogbiggles marked this conversation as resolved. Show resolvedHide resolved
		raise ValueError("failing task")


		def good_task_132969(n: int) -> int:
		time.sleep(0.1 * n)
		return n



		class ExecutorShutdownTest:
		def test_run_after_shutdown(self):
		self.executor.shutdown()
Expand DownExpand Up		@@ -330,6 +340,46 @@ def test_shutdown_no_wait(self):
		# shutdown.
		assert all([r == abs(v) for r, v in zip(res, range(-5, 5))])

		def _run_test_issue_132969(self, max_workers: int) -> int:
		if sys.platform == "win32":
		raise unittest.SkipTest("skip test since forkserver is not available on Windows")
ogbiggles marked this conversation as resolved. Show resolvedHide resolved

		# max_workers=2 will repro exception
		# max_workers=4 will repro exception and then hang

		import multiprocessing as mp
ogbiggles marked this conversation as resolved. Show resolvedHide resolved

		# Repro conditions
		# max_tasks_per_child=1
		# a task ends abnormally
		# shutdown(wait=False) is called
		executor = futures.ProcessPoolExecutor(
		max_workers=max_workers,
		max_tasks_per_child=1,
		mp_context=mp.get_context("forkserver"))
ogbiggles marked this conversation as resolved. Show resolvedHide resolved
		f1 = executor.submit(good_task_132969, 1)
		f2 = executor.submit(failing_task_132969, 2)
		f3 = executor.submit(good_task_132969, 3)
		result:int = 0
ogbiggles marked this conversation as resolved. Show resolvedHide resolved
		try:
		result += f1.result()
		result += f2.result()
		result += f3.result()
		except ValueError:
		# stop processing results upon first exception
		pass

		executor.shutdown(wait=False)
		return result

		def test_shutdown_len_exception_132969(self):
ogbiggles marked this conversation as resolved. Show resolvedHide resolved
		result = self._run_test_issue_132969(2)
		self.assertEqual(result, 1)

		def test_shutdown_process_hang_132969(self):
		result = self._run_test_issue_132969(4)
		self.assertEqual(result, 1)


		create_executor_tests(globals(), ProcessPoolShutdownTest,
		executor_mixins=(ProcessPoolForkMixin,
Expand Down
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1 @@
		Fixes error+hang when ProcessPoolExecutor shutdown called with wait=False and a task ended abnormally.
ogbiggles marked this conversation as resolved. Show resolvedHide resolved