How to use ThreadPoolExecutor in Python3

Parameters

• max_workers: Specifies the maximum number of threads in the pool.
   
• thread_name_prefix (optional): Prefix for thread names, useful for debugging.
   
• initializer (optional): A callable function that runs when a thread starts.
   
• initargs (optional): Arguments to pass to the initializer.

ThreadPoolExecutor Methods

1. **submit(fn, *args, kwargs): Submits a function for execution and returns a Future object.
    
2. map(func, *iterables, timeout=None, chunksize=1): Applies a function to all items in an iterable, returning results in order.
    
3. shutdown(wait=True): Shuts down the executor, waiting for running tasks to complete.
    
4. result(timeout=None): Gets the result of a submitted task, blocking until complete.

Example 1: Basic ThreadPoolExecutor Usage

from concurrent.futures import ThreadPoolExecutor

def print_square(num):
    print(f"Square of {num} is {num * num}")

numbers = [1, 2, 3, 4, 5]

with ThreadPoolExecutor(max_workers=3) as executor:
    executor.map(print_square, numbers)

You can also try this code with Online Python Compiler

Run Code

Output:

Square of 1 is 1
Square of 2 is 4
Square of 3 is 9
Square of 4 is 16
Square of 5 is 25

Example 2: Using submit() with result()

from concurrent.futures import ThreadPoolExecutor

def multiply(x, y):
    return x * y

with ThreadPoolExecutor(max_workers=2) as executor:
    future1 = executor.submit(multiply, 2, 3)
    future2 = executor.submit(multiply, 4, 5)
    
    print(f"Result 1: {future1.result()}")
    print(f"Result 2: {future2.result()}")

You can also try this code with Online Python Compiler

Run Code

Output:

Result 1: 6
Result 2: 20

ThreadPoolExecutor Usage Patterns

• Parallel execution of independent tasks: Running tasks that do not depend on each other.
• Batch processing: Handling multiple inputs efficiently.
• I/O-bound operations: Useful for tasks involving network requests, file handling, etc.
• Web scraping: Fetching multiple web pages concurrently.

How to Configure ThreadPoolExecutor

Choosing the right max_workers value is important:

• If tasks are CPU-bound, use max_workers equal to the number of CPU cores.
   
• If tasks are I/O-bound, set max_workers to a higher value (e.g., 5-10 times the number of CPU cores).

Example:

import os
from concurrent.futures import ThreadPoolExecutor

max_threads = os.cpu_count() * 2  # Recommended for I/O-bound tasks
executor = ThreadPoolExecutor(max_workers=max_threads)

You can also try this code with Online Python Compiler

Run Code

How to Use Future Objects in Detail

A Future object represents the result of an asynchronous computation. We use submit() to execute a function and result() to get the output.

Example:

from concurrent.futures import ThreadPoolExecutor
import time

def long_task():
    time.sleep(2)
    return "Task completed"

with ThreadPoolExecutor() as executor:
    future = executor.submit(long_task)
    print(future.done())  # False, since task is running
    print(future.result())  # Waits for completion and prints result

You can also try this code with Online Python Compiler

Run Code

Output:

False
Task completed

When to Use the ThreadPoolExecutor

• Processing multiple files in parallel
   
• Handling multiple web requests at the same time
   
• Executing background tasks in GUI applications
   
• Parallelizing tasks in machine learning workflows

How Does ThreadPoolExecutor Work Internally  

The `ThreadPoolExecutor ` is part of Python's `concurrent.futures` module. It allows you to create & manage a pool of worker threads to execute tasks concurrently. Instead of creating & managing threads manually, it handles the complexity for you.  

Key Components of ThreadPoolExecutor  

1. Thread Pool: A collection of pre-initialized threads that are ready to execute tasks.  
 

2. Task Queue: Tasks are submitted to a queue, & threads pick them up for execution.  
 

3. Futures: Represents the result of an asynchronous computation. It allows you to check if the task is done or retrieve the result once it’s completed.  

How It Works  

1. When you submit a task to the `ThreadPoolExecutor`, it adds the task to the task queue.  
 

2. Worker threads in the pool pick up tasks from the queue & execute them.  
 

3. Once a task is completed, the result is stored in a `Future` object.  
 

4. You can retrieve the result using the `Future` object.  

Example Code  

Let’s see how to use `ThreadPoolExecutor` with a simple example. We’ll create a function that simulates a task & execute it using the thread pool.  

from concurrent.futures import ThreadPoolExecutor
import time

Define a function that simulates a task

def task(name):
    print(f"Task {name} started")
    time.sleep(2)   Simulate a time-consuming task
    print(f"Task {name} completed")
    return f"Result from {name}"

Create a ThreadPoolExecutor with 3 worker threads

with ThreadPoolExecutor(max_workers=3) as executor:
     Submit tasks to the executor
    future1 = executor.submit(task, "A")
    future2 = executor.submit(task, "B")
    future3 = executor.submit(task, "C")


     Retrieve results from the futures
    print(future1.result())
    print(future2.result())
    print(future3.result())

In this Code: 

1. Importing Modules: We import `ThreadPoolExecutor` from the `concurrent.futures` module & `time` for simulating delays.  
 

2. Task Function: The `task` function simulates a task that takes 2 seconds to complete.  
 

3. ThreadPoolExecutor: We create a `ThreadPoolExecutor` with 3 worker threads.  
 

4. Submitting Tasks: We submit 3 tasks to the executor using `executor.submit()`. Each task is assigned a unique name.  
 

5. Retrieving Results: We use `future.result()` to get the result of each task. This method blocks until the task is completed.  

Output  

Task A started
Task B started
Task C started
Task A completed
Task B completed
Task C completed
Result from A
Result from B
Result from C

This example shows how `ThreadPoolExecutor ` manages multiple tasks concurrently using a pool of threads.  

ThreadPoolExecutor Exception Handling

When a function raises an exception, it is captured inside the Future object.

Example:

from concurrent.futures import ThreadPoolExecutor

def divide(a, b):
    return a / b

with ThreadPoolExecutor() as executor:
    future = executor.submit(divide, 4, 0)
    try:
        print(future.result())
    except Exception as e:
        print(f"Exception: {e}")

You can also try this code with Online Python Compiler

Run Code

Output:

Exception: division by zero

ThreadPoolExecutor Best Practices  

Using `ThreadPoolExecutor` effectively requires understanding its limitations & following best practices to avoid common pitfalls. There are some key tips to help you use it efficiently:  

1. Choose the Right Number of Threads  

The `max_workers` parameter determines the number of threads in the pool. Setting it too high can lead to excessive resource usage, while setting it too low can underutilize your system’s capabilities. A good rule of thumb is to set it based on the number of CPU cores or the nature of your tasks.  

import os
from concurrent.futures import ThreadPoolExecutor

Set max_workers based on the number of CPU cores

max_workers = os.cpu_count() or 4   Default to 4 if cpu_count() returns None
executor = ThreadPoolExecutor(max_workers=max_workers)

2. Avoid Blocking the Main Thread  

When using `ThreadPoolExecutor`, ensure that your main thread isn’t blocked while waiting for results. Use `as_completed()` to process results as they become available.  

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
def task(name):
    print(f"Task {name} started")
    time.sleep(2)   Simulate a time-consuming task
    print(f"Task {name} completed")
    return f"Result from {name}"

with ThreadPoolExecutor(max_workers=3) as executor:

  futures = [executor.submit(task, f"Task-{i}") for i in range(5)]
    
    for future in as_completed(futures):
        print(future.result())

3. Handle Exceptions Gracefully  

Tasks running in threads can raise exceptions. Use `future.exception()` to handle errors without crashing your program.  

from concurrent.futures import ThreadPoolExecutor

def task(name):
    if name == "Task-2":
        raise ValueError("An error occurred in Task-2")
    return f"Result from {name}"

with ThreadPoolExecutor(max_workers=2) as executor:
    futures = [executor.submit(task, f"Task-{i}") for i in range(3)]
    
    for future in futures:
        if future.exception():
            print(f"Error: {future.exception()}")
        else:
            print(future.result())

4. Use `map()` for Simpler Workloads  

If you have a list of inputs & want to apply the same function to each, use `executor.map()` for simplicity.  

from concurrent.futures import ThreadPoolExecutor

def task(name):
    return f"Processed {name}"

with ThreadPoolExecutor(max_workers=3) as executor:
    results = executor.map(task, ["A", "B", "C", "D", "E"])
    
    for result in results:
        print(result)

5. Clean Up Resources  

Always use `ThreadPoolExecutor` in a `with` block or call `executor.shutdown()` to ensure threads are properly cleaned up.  

from concurrent.futures import ThreadPoolExecutor

def task(name):
    return f"Result from {name}"

executor = ThreadPoolExecutor(max_workers=2)
futures = [executor.submit(task, f"Task-{i}") for i in range(3)]

for future in futures:
    print(future.result())

executor.shutdown()   Clean up resources

6. Avoid Long-Running Tasks  

ThreadPoolExecutor` is best suited for short to medium-length tasks. For long-running tasks, consider using `ProcessPoolExecutor` or other concurrency models.  

Frequently Asked Questions

What is the difference between ThreadPoolExecutor and ProcessPoolExecutor?

ThreadPoolExecutor is used for I/O-bound tasks, while ProcessPoolExecutor is used for CPU-bound tasks.

What happens if max_workers is not specified?

If max_workers is not set, Python assigns a default value based on the system.

Can ThreadPoolExecutor be used for CPU-intensive tasks?

No, for CPU-intensive tasks, use ProcessPoolExecutor instead.

Conclusion

In this article, we learned how to use ThreadPoolExecutor in Python 3, which is part of the concurrent.futures module. It provides an efficient way to manage multiple threads for parallel execution of tasks, improving performance in CPU-bound and I/O-bound operations. UnderstandingThreadPoolExecutor helps in writing efficient, scalable, and concurrent applications in Python.