Skip to main content

Lib/queue.py

cpython 3.14 @ ab2d84fe1023/Lib/queue.py

queue.py provides the Queue, LifoQueue, PriorityQueue, and SimpleQueue classes. All variants share a common thread-safe base built on threading.Condition objects. The module is designed for producer-consumer patterns: producers call put(), consumers call get(), and task_done()/join() let callers block until the backlog is fully processed.

Map​

LinesSymbolRolegopy
1-10importsthreading, heapq, collections.dequen/a
11-20__all__Public API listn/a
21-30Empty, FullException classesn/a
31-90Queue.__init__, Queue.qsize, Queue.empty, Queue.fullBase queue setup and sizing predicatesNot yet ported
91-160Queue.put, Queue.put_nowait, Queue.get, Queue.get_nowaitBlocking and non-blocking insertion/retrievalNot yet ported
161-200Queue.task_done, Queue.joinCompletion trackingNot yet ported
201-220Queue._init, Queue._qsize, Queue._put, Queue._getInternal deque-backed storage hooksNot yet ported
221-240LifoQueueStack variant; overrides _init, _put, _getNot yet ported
241-260PriorityQueueHeap variant; overrides _init, _put, _getNot yet ported
261-280SimpleQueueLock-free single-condition simplified queueNot yet ported

Reading​

Initialization and condition variables (lines 31 to 70)​

cpython 3.14 @ ab2d84fe1023/Lib/queue.py#L31-70

The constructor allocates two threading.Condition objects that share the same underlying mutex. not_empty is notified by put() to wake blocked get() callers; not_full is notified by get() to wake blocked put() callers. An all_tasks_done condition tracks the outstanding work count for join().

def __init__(self, maxsize=0):
    self.mutex = threading.Lock()
    self.not_empty = threading.Condition(self.mutex)
    self.not_full = threading.Condition(self.mutex)
    self.all_tasks_done = threading.Condition(self.mutex)
    self.unfinished_tasks = 0
    self._init(maxsize)

maxsize <= 0 means unbounded. The three conditions share self.mutex, so any one with block holds the same lock that the other two conditions also use.

Blocking put and get (lines 91 to 160)​

cpython 3.14 @ ab2d84fe1023/Lib/queue.py#L91-160

put() acquires the mutex through the not_full condition and waits in a while self._qsize() >= self.maxsize loop. The loop guard protects against spurious wakeups. Once space exists it delegates to _put(item) and then notifies not_empty so any blocked get() can proceed.

def put(self, item, block=True, timeout=None):
    with self.not_full:
        if self.maxsize > 0:
            if not block:
                if self._qsize() >= self.maxsize:
                    raise Full
            elif timeout is None:
                while self._qsize() >= self.maxsize:
                    self.not_full.wait()
            elif timeout < 0:
                raise ValueError("'timeout' must be a non-negative number")
            else:
                endtime = time() + timeout
                while self._qsize() >= self.maxsize:
                    remaining = endtime - time()
                    if remaining <= 0.0:
                        raise Full
                    self.not_full.wait(remaining)
        self._put(item)
        self.unfinished_tasks += 1
        self.not_empty.notify()

get() mirrors this pattern using not_empty.wait() and then calls _get(), followed by not_full.notify().

task_done and join (lines 161 to 200)​

cpython 3.14 @ ab2d84fe1023/Lib/queue.py#L161-200

task_done() decrements unfinished_tasks and, when it reaches zero, broadcasts on all_tasks_done to release any thread blocked in join(). Calling task_done() more times than items were put raises ValueError.

def task_done(self):
    with self.all_tasks_done:
        unfinished = self.unfinished_tasks - 1
        if unfinished <= 0:
            if unfinished < 0:
                raise ValueError('task_done() called too many times')
            self.all_tasks_done.notify_all()
        self.unfinished_tasks = unfinished

def join(self):
    with self.all_tasks_done:
        while self.unfinished_tasks:
            self.all_tasks_done.wait()

join() is safe to call from any thread. It returns immediately if no tasks are outstanding.

Storage hook overrides (lines 201 to 260)​

cpython 3.14 @ ab2d84fe1023/Lib/queue.py#L201-260

The base Queue stores items in a collections.deque. The three subclasses override only the four internal hooks, leaving all locking logic in the base class.

# Queue (FIFO)
def _init(self, maxsize): self.queue = deque()
def _qsize(self):         return len(self.queue)
def _put(self, item):     self.queue.append(item)
def _get(self):           return self.queue.popleft()

# LifoQueue (LIFO/stack)
def _init(self, maxsize): self.queue = []
def _put(self, item):     self.queue.append(item)
def _get(self):           return self.queue.pop()

# PriorityQueue (min-heap)
def _init(self, maxsize): self.queue = []
def _put(self, item):     heappush(self.queue, item)
def _get(self):           return heappop(self.queue)

SimpleQueue is a completely separate class backed by a single threading.Condition; it has no task_done, no join, and no maxsize.

gopy mirror​

Not yet ported. The Go standard library provides sync.Mutex and sync.Cond which map directly onto CPython's threading.Lock and threading.Condition. A future port would implement Queue as a Go struct with notEmpty and notFull sync.Cond fields sharing a single sync.Mutex, and would expose Put, Get, TaskDone, and Join methods. LifoQueue and PriorityQueue would embed the base struct and override the storage hooks via interface methods or embedded function fields.

CPython 3.14 changes​

CPython 3.14 made no breaking changes to queue.py. The module gained minor type annotation improvements and the SimpleQueue class received a small internal refactor to reduce lock contention on single-core interpreters taking advantage of the free-threaded build (--disable-gil). Under the free-threaded build, SimpleQueue uses a _thread.RLock variant rather than a plain Lock.