Thread-safe queues

Producer-consumer handoff

queue.Queue synchronizes put and get. FIFO determines removal order, not completion order across consumers. Processing state outside the queue needs its own synchronization.

Threaded queues pass object references, not snapshots:

from queue import Queue

messages = Queue()
message = {"status": "new"}
messages.put(message)
message["status"] = "changed"
received = messages.get()
print(received is message, received["status"])  # True changed
messages.task_done()
messages.join()

Choose an ownership policy: immutable messages, an explicit copy before submission, or a producer that stops accessing the submitted object. Queue synchronization does not protect later edits to the payload.

Capacity and acknowledgment

• get() waits when empty; a positive maxsize makes put() wait when full. This bounds the backlog and applies backpressure.
• maxsize=0 means unbounded capacity, not zero available slots.
• get_nowait() and put_nowait() raise Empty and Full instead of waiting.

Removing an item frees storage capacity before the processing attempt is acknowledged:

from queue import Full, Queue

work = Queue(maxsize=1)
work.put("scan")
try:
    work.put_nowait("index")
except Full:
    print("backlog full")
first = work.get()
work.put_nowait("index")
print(first, work.get())  # scan index
work.task_done()
work.task_done()
work.join()

join() waits for matching acknowledgments, not for an empty queue or terminated workers. A normal worker calls task_done() once after each attempt, typically in finally. Record success or failure separately; acknowledgment alone is not business success.

Shutdown

A sentinel is a reserved stop message. For a finite batch, finish production before enqueuing enough sentinels for all consumers. Acknowledge each retrieved sentinel, exit the worker, then join worker threads as well as queue work.

A shared unique object's identity works as a thread sentinel; process serialization requires a different contract. With multiple producers, coordinate their completion so one cannot stop consumers prematurely. Timeout-based stop checks also need a policy for abandoned work.

Gotchas

• empty() and qsize() cannot guarantee the next operation will succeed; another worker may change the queue immediately.
• Missing acknowledgments or a dead worker can leave join() blocked. Extra task_done() calls raise ValueError.
• maxsize bounds queued items, not all in-flight work or payload bytes.
• SimpleQueue omits capacity and task tracking. Use process-aware queues across processes.

This lesson requires native Python.