Imagine a line of tasks waiting for the next available worker. A queue holds them in first-in, first-out (FIFO) order: a worker takes the next task that arrived. When several workers share a queue, they can process different tasks concurrently, so retrieval order does not necessarily determine completion order. section 5.3 section 5.4
A message queue links a task producer to its consumers without requiring them to work at the same moment. It can support asynchronous communication between threads or processes. In a thread pool, the main thread adds tasks; an available worker retrieves one, runs it, then returns for more. Reusing workers avoids creating a thread for every task. section 5.4 section 5.6
Ten tasks, four workers
Put the numbers 0 through 9 into a Python queue, then start four worker threads. Each worker calls `get()` to claim an item and processes it. The workers can claim successive items in queue order while their work overlaps. The queue coordinates who gets the next item; it does not make the work itself sequential. section 5.3
# Chapter 5/thread_pool.py
import time
import queue
import typing as T
from threading import Thread, current_thread
Callback = T.Callable[..., None]
Task = T.Tuple[Callback, T.Any, T.Any]
TaskQueue = queue.Queue
class Worker(Thread):
def __init__(self, tasks: queue.Queue[Task]):
super().__init__()
self.tasks = tasks
def run(self) -> None:
while True: #1
func, args, kargs = self.tasks.get() #1
try: #1
func(*args, **kargs) #1
except Exception as e: #1
print(e) #1
self.tasks.task_done() #1
class ThreadPool:
def __init__(self, num_threads: int):
self.tasks: TaskQueue = queue.Queue(num_threads) #2
self.num_threads = num_threads
for _ in range(self.num_threads): #3
worker = Worker(self.tasks) #3
worker.setDaemon(True) #3
worker.start() #3
def submit(self, func: Callback, *args, **kargs) -> None:
self.tasks.put((func, args, kargs))
def wait_completion(self) -> None:
self.tasks.join() #4
def cpu_waster(i: int) -> None:
name = current_thread().getName()
print(f"{name} doing {i} work")
time.sleep(3)
def main() -> None:
pool = ThreadPool(num_threads=5) #5
for i in range(20): #6
pool.submit(cpu_waster, i) #6
print("All work requests sent")
pool.wait_completion()
print("All work completed")
if __name__ == "__main__":
main()
A Python thread-pool implementation shows workers retrieving queued tasks, executing callbacks, and marking tasks done. section 5.4
Choose a queue when independent tasks should be handed to available workers or messages should be exchanged asynchronously. Other communication needs may call for shared memory, pipes, or sockets instead. section 5.4 section 5.6
