Prevent deadlocks with default ThreadPoolTaskRunner settings (#14201)

src/prefect/task_runners.py

@@ -1,5 +1,6 @@
 import abc
 import asyncio
+import sys
 import uuid
 from concurrent.futures import ThreadPoolExecutor
 from contextvars import copy_context
@@ -205,7 +206,7 @@ class ThreadPoolTaskRunner(TaskRunner[PrefectConcurrentFuture]):
     def __init__(self, max_workers: Optional[int] = None):
         super().__init__()
         self._executor: Optional[ThreadPoolExecutor] = None
-        self._max_workers = max_workers
+        self._max_workers = sys.maxsize if max_workers is None else max_workers
 
     def duplicate(self) -> "ThreadPoolTaskRunner":
         return type(self)(max_workers=self._max_workers)

tests/test_task_runners.py

@@ -1,4 +1,5 @@
 import asyncio
+import time
 import uuid
 from concurrent.futures import Future
 from typing import Any, Iterable, Optional
@@ -9,6 +10,7 @@
 from prefect._internal.concurrency.api import create_call, from_async
 from prefect.context import TagsContext, tags
 from prefect.filesystems import LocalFileSystem
+from prefect.flows import flow
 from prefect.futures import PrefectFuture, PrefectWrappedFuture
 from prefect.results import _default_storages
 from prefect.settings import (
@@ -200,6 +202,30 @@ def test_map_with_future_resolved_to_list(self):
             results = [future.result() for future in futures]
             assert results == [(1, 1), (2, 2), (3, 3)]
 
+    def test_handles_recursively_submitted_tasks(self):
+        """
+        Regression test for https://github.com/PrefectHQ/prefect/issues/14194.
+
+        This test ensures that the ThreadPoolTaskRunner doesn't place an upper limit on the
+        number of submitted tasks active at once. The highest default max workers on a
+        ThreadPoolExecutor is 32, so this test submits 33 tasks recursively, which will
+        deadlock without the ThreadPoolTaskRunner setting the max_workers to sys.maxsize.
+        """
+
+        @task
+        def recursive_task(n):
+            if n == 0:
+                return n
+            time.sleep(0.1)
+            future = recursive_task.submit(n - 1)
+            return future.result()
+
+        @flow
+        def test_flow():
+            return recursive_task.submit(33)
+
+        assert test_flow().result() == 0
+
 
 class TestPrefectTaskRunner:
     @pytest.fixture(autouse=True)