Thread Pool Map for IO Bound Tasks in Python
Run IO-bound mock tasks concurrently with ThreadPoolExecutor.map and measure total elapsed time in Python.
Python code
20 linesimport concurrent.futures
import time
from pathlib import Path
def mock_io_task(filename):
"""Simulate an IO-bound task by creating a small file and measuring its latency."""
path = Path(filename)
path.write_text("data")
time.sleep(0.1) # Simulate slow disk/network
return f"{filename} written in {time.time():.2f}"
if __name__ == "__main__":
files = [f"file_{i}.txt" for i in range(5)]
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
results = list(executor.map(mock_io_task, files))
for result in results:
print(result)
print(f"Total elapsed: {time.time() - start_time:.2f}s")Output
file_0.txt written in 1717521234.56
file_1.txt written in 1717521234.66
file_2.txt written in 1717521234.76
file_3.txt written in 1717521234.86
file_4.txt written in 1717521234.96
Total elapsed: 0.21sHow it works
ThreadPoolExecutor.map distributes the files list across a pool of 3 worker threads, running each mock_io_task concurrently. Because the task simulates IO with time.sleep(0.1), threads release the GIL during the sleep, allowing true parallelism in latency. The with block ensures all threads complete and resources are cleaned up before results are collected. Concurrency reduces total wall-clock time by overlapping the 100ms sleeps across workers. The start_time variable must be defined before the executor block for the final elapsed measurement to work.
Common mistakes
- Forgetting to define `start_time` before the executor block, causing a NameError.
- Using `ProcessPoolExecutor` for IO-bound tasks, which adds process overhead without GIL benefit.
- Calling `executor.map` without converting to `list`, so results remain a generator that never runs.
- Passing a huge file list that exhausts memory when materialized via `list()`.
Variations
- Use `executor.submit` with `as_completed` to consume results as they finish instead of in input order.
- Switch to `ProcessPoolExecutor` for CPU-bound alternatives, accepting higher overhead.
Real-world use cases
- Batch-uploading thousands of files to cloud storage where each upload is latency-bound.
- Fetching multiple API endpoints in parallel before aggregating the responses in a web service.
- Processing a queue of image thumbnails with per-item network calls to a CDN.
Sponsored
More from Concurrency & performance
- Benchmark list.append vs deque.append in Python medium
- Build a Python Performance Profiler That Generates Readable Reports medium
- Graceful Shutdown Executor Context Manager in Python medium
- How to Build a Producer-Consumer Pattern with asyncio.Queue in Python medium
- How to Cancel an asyncio Task with Graceful Cleanup in Python medium
- How to Convert Data in Parallel with ThreadPoolExecutor in Python easy
Keep learning
Related tutorials and quizzes for this topic.