Thread Pools

Introduction

Thread pools manage a collection of reusable worker threads, avoiding the overhead of creating new threads for each task. concurrent.futures provides ThreadPoolExecutor.

ThreadPoolExecutor Basics

from concurrent.futures import ThreadPoolExecutor
import time

def task(n):
    time.sleep(1)
    return f"Task {n} done"

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(task, i) for i in range(8)]
    for future in futures:
        print(future.result())

Map and Submit

from concurrent.futures import ThreadPoolExecutor

def process(n):
    return n * 2

with ThreadPoolExecutor(max_workers=3) as executor:
    # map returns results in order
    results = list(executor.map(process, range(5)))
    print(results)  # [0, 2, 4, 6, 8]
    
    # submit returns futures
    futures = [executor.submit(process, i) for i in range(5)]
    for f in futures:
        print(f.result())

Callbacks and Futures

from concurrent.futures import ThreadPoolExecutor, Future

def done_callback(future):
    try:
        result = future.result()
        print(f"Completed: {result}")
    except Exception as e:
        print(f"Failed: {e}")

def task(n):
    if n == 3:
        raise ValueError("Error in task 3")
    return n * 2

with ThreadPoolExecutor(max_workers=2) as executor:
    future = executor.submit(task, 3)
    future.add_done_callback(done_callback)

ThreadPoolExecutor Configuration

from concurrent.futures import ThreadPoolExecutor
import threading

# Configure thread pool
executor = ThreadPoolExecutor(
    max_workers=4,
    thread_name_prefix="worker",
    initializer=init_worker,
    initargs=(arg,)
)

def init_worker(arg):
    # Initialize thread-local state
    pass

# Shutdown gracefully
executor.shutdown(wait=True, cancel_futures=False)

Using with Async

import asyncio
from concurrent.futures import ThreadPoolExecutor
import time

def blocking_task():
    time.sleep(1)
    return "Done"

async def main():
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor() as executor:
        result = await loop.run_in_executor(executor, blocking_task)
        print(result)

asyncio.run(main())

Practice Problems

1. Use ThreadPoolExecutor for parallel execution
2. Process multiple tasks with map
3. Handle future callbacks
4. Integrate with asyncio
5. Configure thread pool settings