Building Robust Object-Oriented Python Applications and Libraries/

...

/

Queues

Queues

Learn how to use the queue.Queue for inter-process communication and building a concurrent text content search engine 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:

Press + to interact
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 multiprocessing.Queue objects; one for incoming queries and one to send outgoing results. These queues automatically pickle the data in the queue and pass it into the subprocess over a pipe. These two queues are set up in the main process and passed through the pipes into the search function inside the child processes.

Type hints and queue usage

The type hints reflect the way mypy wants details about the structure of data in each queue. ...