Search⌘ K
AI Features
Home
Courses
Building Robust Object-Oriented Python Applications and Libraries

Queues

Explore the use of Python’s queue.Queue for managing interprocess communication in concurrent programs. Learn how to implement queues with multiprocessing, handle tasks with worker processes, and coordinate complex searches efficiently. This lesson equips you with foundational skills to build and manage parallel processing systems in Python.

We'll cover the following...

Overview

If we need more control over communication between processes, the queue.Queue data structure is useful. There are several variants offering ways to send messages from one process to one or more other processes. Any picklable object can be sent into a Queue, but remember that pickling can be a costly operation, so keep such objects small. To illustrate queues, let’s build a little search engine for text content that stores all relevant entries in memory.

Example

A particular search engine scans all files in the current directory in parallel. A process is constructed for each core on the CPU. Each of these is instructed to load some of the files into memory. Let’s look at the function that does the loading and searching:

Python 3.10.4
from __future__ import annotations
from pathlib import Path
from typing import List, Iterator, Optional, Union, TYPE_CHECKING
if TYPE_CHECKING:
  Query_Q = Queue[Union[str, None]]
  Result_Q = Queue[List[str]]
def search(
      paths: list[Path],
      query_q: Query_Q,
      results_q: Result_Q
) -> None:
  print(f"PID: {os.getpid()}, paths {len(paths)}")
  lines: List[str] = []
  for path in paths:
    lines.extend(
      l.rstrip() for l in path.read_text().splitlines())
  while True:
    if (query_text := query_q.get()) is None:
      break
    results = [l for l in lines if query_text in l]
    results_q.put(results)

Remember, the search() function is run in a separate process (in fact, it is run in cpu_count() separate processes) from the main process that created the queues. Each of these processes is started with a list of pathlib.Path objects, and two ...