Introduction

The error "cannot schedule new futures after shutdown" or a deadlock with ThreadPoolExecutor(max_workers=1) occurs when a task submitted to a single-threaded executor submits another task to the same executor and waits for it. Since there is only one worker thread, the first task occupies it while waiting for the second task to complete, but the second task cannot start because the only thread is blocked. This is a classic deadlock pattern. The solutions are: increase the thread pool size, avoid nested submit().result() calls, or restructure the code to eliminate the dependency.

The Deadlock Pattern

1from concurrent.futures import ThreadPoolExecutor
2
3executor = ThreadPoolExecutor(max_workers=1)
4
5def task_a():
6    print("Task A started")
7    # Submit task_b to the same single-thread executor
8    future = executor.submit(task_b)
9    # Wait for task_b — but task_b can't start because task_a holds the only thread
10    result = future.result()  # DEADLOCK: hangs forever
11    print(f"Task A got: {result}")
12    return "done"
13
14def task_b():
15    print("Task B started")
16    return "result from B"
17
18# This will hang forever
19future_a = executor.submit(task_a)
20print(future_a.result())  # never completes

Task A is running on the single worker thread. It submits Task B to the same executor and calls .result(), blocking until Task B finishes. But Task B is queued and cannot start because the only thread is occupied by Task A, which is waiting for Task B.

Fix 1: Increase Thread Pool Size

1from concurrent.futures import ThreadPoolExecutor
2
3# More threads prevent the deadlock
4executor = ThreadPoolExecutor(max_workers=2)
5
6def task_a():
7    print("Task A started")
8    future = executor.submit(task_b)
9    result = future.result()  # works — task_b runs on the second thread
10    print(f"Task A got: {result}")
11    return "done"
12
13def task_b():
14    print("Task B started")
15    return "result from B"
16
17future_a = executor.submit(task_a)
18print(future_a.result())  # "done"

With 2 or more workers, Task B can run on a different thread while Task A waits. However, if nesting goes deeper (Task B submits Task C), you need even more threads. This approach does not scale for arbitrary nesting depths.

Fix 2: Avoid Nested Submissions

1from concurrent.futures import ThreadPoolExecutor
2
3executor = ThreadPoolExecutor(max_workers=1)
4
5def task_a():
6    print("Task A: doing work")
7    return "data from A"
8
9def task_b(data):
10    print(f"Task B: processing {data}")
11    return f"result from B using {data}"
12
13# Chain tasks outside the executor — no nesting
14future_a = executor.submit(task_a)
15result_a = future_a.result()
16
17future_b = executor.submit(task_b, result_a)
18result_b = future_b.result()
19print(result_b)  # "result from B using data from A"

Submitting tasks sequentially from the main thread avoids the deadlock because each task completes before the next is submitted. The executor's single thread is never simultaneously occupied by two dependent tasks.

Fix 3: Use asyncio Instead

1import asyncio
2
3async def task_a():
4    print("Task A started")
5    result = await task_b()  # cooperative — doesn't block a thread
6    print(f"Task A got: {result}")
7    return "done"
8
9async def task_b():
10    print("Task B started")
11    await asyncio.sleep(0.1)  # simulate I/O
12    return "result from B"
13
14# asyncio event loop handles nested awaits without deadlock
15result = asyncio.run(task_a())
16print(result)  # "done"

asyncio uses cooperative multitasking — await suspends the coroutine without blocking a thread, so nested calls naturally work. For I/O-bound workloads, asyncio is a better fit than thread pools.

Fix 4: Separate Executor for Nested Tasks

1from concurrent.futures import ThreadPoolExecutor
2
3outer_executor = ThreadPoolExecutor(max_workers=1)
4inner_executor = ThreadPoolExecutor(max_workers=2)
5
6def task_a():
7    print("Task A started")
8    # Submit to a DIFFERENT executor
9    future = inner_executor.submit(task_b)
10    result = future.result()  # works — different executor, different threads
11    print(f"Task A got: {result}")
12    return "done"
13
14def task_b():
15    print("Task B started")
16    return "result from B"
17
18future_a = outer_executor.submit(task_a)
19print(future_a.result())  # "done"

Using separate executors for different levels of task nesting avoids the deadlock because each level has its own thread pool.

The Same Issue with ProcessPoolExecutor

1from concurrent.futures import ProcessPoolExecutor
2
3# The same deadlock can happen with processes
4executor = ProcessPoolExecutor(max_workers=1)
5
6def task_a():
7    # This will also deadlock with a single-worker process pool
8    future = executor.submit(task_b)
9    return future.result()
10
11def task_b():
12    return "result"
13
14# Additionally, ProcessPoolExecutor cannot submit from within a worker
15# because the executor object is not picklable across processes

ProcessPoolExecutor has the same deadlock potential plus the additional restriction that you cannot easily pass the executor to child processes (it is not picklable).

Common Pitfalls

• Calling .result() on a future from inside the same executor: This is the direct cause of the deadlock. The calling task holds the thread, and the awaited task needs a thread from the same pool. Either increase pool size, submit sequentially from outside, or use a different executor.
• Assuming max_workers=1 behaves like a queue: A single-thread executor does serialize tasks, but only when tasks are independent. If Task A depends on Task B (which is also submitted to the executor), the serialization becomes a deadlock because A cannot complete until B runs, and B cannot start until A completes.
• Using executor inside a callback or decorator: If a function wrapped in executor.submit() internally calls another executor.submit() (e.g., through a decorator or middleware), the nesting is hidden and the deadlock is not obvious from reading the top-level code. Audit the call chain for hidden submissions.
• Not shutting down executors properly: Forgetting executor.shutdown(wait=True) or using the executor as a context manager (with ThreadPoolExecutor() as e:) can leave orphaned threads. A deadlocked executor will hang on shutdown forever — add timeouts to .result() calls to detect deadlocks.
• Increasing max_workers without bound to "fix" nesting: While more workers prevent the immediate deadlock, each additional nesting level requires another thread. For deep nesting or recursive patterns, the thread count grows unboundedly. Restructure the code to eliminate nesting instead.

Summary

• A single-thread executor deadlocks when a running task submits and waits for another task on the same executor
• Fix by increasing max_workers, chaining tasks sequentially from outside the executor, or using separate executors
• Use asyncio for I/O-bound workloads — cooperative await does not block threads
• Add timeouts to .result(timeout=N) to detect and recover from deadlocks
• Audit the full call chain to find hidden nested executor.submit() calls
• Avoid recursive or deeply nested submissions — restructure as sequential task chains