📬 Pattern 3: Queues & Pipes

Why You Need This

Recall the one fact that drives everything (from Processes 101): a child process has its own private memory. It cannot see a variable you mutated in the parent, and you cannot see what it computed. There is no shared list to append to, no shared dict to read. If two processes need to cooperate, they must pass messages across an OS boundary.

multiprocessing.Queue: the Process-Safe FIFO

mp.Queue is a first-in-first-out queue that is safe for multiple producers and multiple consumers at once — the locking is built in, so you never wrap put() or get() in your own lock. It's the default tool for fanning work out to several long-lived worker processes and collecting their results.

What Happens Under the Hood

This surprises people coming from other languages: put(item) does not immediately write to the pipe. It pickles item, drops the bytes into an in-process buffer, and a background "feeder" thread flushes that buffer into an OS pipe. The consumer's get() reads bytes off the pipe and unpickles them. Two consequences fall out of this design:

• Items must be picklable — a lambda, an open socket, or a DB connection will raise on put() (or in the feeder thread).
• put() can return before the item is actually on the pipe. This is why joining a producer too early can deadlock (next section).

Producer / Consumer with Sentinel Shutdown

A queue has no "I'm done" signal of its own. The idiomatic way to tell consumers to stop is a sentinel — a unique marker value (conventionally None) that means "no more work." You put() one sentinel per consumer so each one sees its own stop signal.

import multiprocessing as mp

SENTINEL = None   # marker meaning "no more work"

def producer(task_queue, items):
    for item in items:
        task_queue.put(item)          # pickled + shipped through the pipe

def consumer(task_queue, result_queue):
    while True:
        item = task_queue.get()       # blocks until an item is available
        if item is SENTINEL:          # our shutdown signal
            break
        result_queue.put(item * item) # do the work, ship the result back

if __name__ == "__main__":
    N_WORKERS = 3
    task_queue = mp.Queue()
    result_queue = mp.Queue()

    workers = [
        mp.Process(target=consumer, args=(task_queue, result_queue))
        for _ in range(N_WORKERS)
    ]
    for w in workers:
        w.start()

    # feed 10 jobs, then one sentinel PER worker so each one exits
    for n in range(10):
        task_queue.put(n)
    for _ in range(N_WORKERS):
        task_queue.put(SENTINEL)

    # drain ALL results BEFORE joining (see the gotcha below)
    results = [result_queue.get() for _ in range(10)]

    for w in workers:
        w.join()

    print(sorted(results))   # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

The Gotcha That Bites Everyone: Drain Before You Join

Why it's easy to miss: with tiny results the pipe buffer never fills, so join()-then-drain appears to work in testing and only deadlocks in production once the payloads grow. Get the ordering right from day one.

JoinableQueue: "Wait Until All Work Is Consumed"

Sometimes you don't want to count results — you just want to block until every task you enqueued has been processed. JoinableQueue adds two methods for that: a consumer calls task_done() after finishing each item, and the producer calls the queue's join() to block until the count of task_done() calls matches the count of put() calls.

import multiprocessing as mp

def worker(task_queue):
    while True:
        item = task_queue.get()
        if item is None:
            task_queue.task_done()    # even the sentinel must be acked
            break
        # ... do the work ...
        task_queue.task_done()        # signal: this item is fully handled

if __name__ == "__main__":
    task_queue = mp.JoinableQueue()
    workers = [mp.Process(target=worker, args=(task_queue,)) for _ in range(3)]
    for w in workers:
        w.start()

    for n in range(10):
        task_queue.put(n)
    for _ in workers:
        task_queue.put(None)          # one sentinel per worker

    task_queue.join()                 # blocks until every task_done() has fired
    for w in workers:
        w.join()
    print("all work consumed")

Pipe(): the Fast Two-Endpoint Channel

mp.Pipe() returns a pair of Connection objects — the two ends of a single channel. It's duplex by default (both ends can send() and recv()), and like Queue it pickles whatever you send. For a channel between exactly two processes it's noticeably faster than a Queue, because there's no feeder thread or internal locking in the way.

import multiprocessing as mp

def child(conn):
    msg = conn.recv()                 # blocks until something arrives (unpickled)
    conn.send(f"got: {msg}")          # reply back down the same pipe
    conn.close()

if __name__ == "__main__":
    parent_conn, child_conn = mp.Pipe()   # two ends of one channel
    p = mp.Process(target=child, args=(child_conn,))
    p.start()

    parent_conn.send("hello")
    print(parent_conn.recv())         # got: hello
    p.join()

Which Channel? mp.Queue vs Pipe vs queue.Queue

When to Reach for Raw Queues at All