Python systems coding reference

Start with the contract

1. What are the public methods and return values?
2. Which inputs are invalid?
3. What ordering is promised?
4. What blocks, and what ends the wait?
5. What happens after closure, failure or a duplicate request?
6. Must repeated queries leave stored state unchanged?

Choose the data structure

from bisect import bisect_left, bisect_right
from collections import Counter, defaultdict, deque
from concurrent.futures import ThreadPoolExecutor, as_completed
from heapq import heapify, heappop, heappush
from queue import Queue
from threading import Condition, Event, Lock, Semaphore, Thread
import asyncio
import time

Binary search, time lookup and heaps

bisect_left(values, target)   # first index with value >= target
bisect_right(values, target)  # first index with value > target

left, right = 0, len(values)
while left < right:
    mid = (left + right) // 2
    if values[mid] >= target:
        right = mid
    else:
        left = mid + 1
return left

Latest value at or before a timestamp:

i = bisect_right(entries, timestamp, key=lambda pair: pair[0])
return "" if i == 0 else entries[i - 1][1]

Python's heapq is a min-heap:

heapify(values)               # O(n)
heappush(values, item)        # O(log n)
smallest = heappop(values)    # O(log n)
smallest = values[0]          # O(1), no removal

heappush(heap, (run_at, priority, sequence, job_id))

Reading values[0] is O(1) because the minimum is already at the root. Removing it is O(log n): heappop() moves the final element to the root and sifts it down through at most the height of the heap.

The sequence number makes ties deterministic and avoids comparing non-orderable payloads.

Dependency ordering with Kahn's algorithm

def execution_order(tasks, dependencies):
    graph = {task: [] for task in tasks}
    indegree = {task: 0 for task in tasks}

    for task, prerequisite in dependencies:
        graph[prerequisite].append(task)
        indegree[task] += 1

    ready = deque(task for task in tasks if indegree[task] == 0)
    order = []

    while ready:
        task = ready.popleft()
        order.append(task)
        for dependent in graph[task]:
            indegree[dependent] -= 1
            if indegree[dependent] == 0:
                ready.append(dependent)

    return order if len(order) == len(tasks) else []

O(V + E) time and space. Stateful query methods copy or recompute indegrees; Kahn's working counts must not consume stored graph state.

Threads, locks and conditions

thread = Thread(target=work, args=(value,))
thread.start()
thread.join(timeout=2)
if thread.is_alive():
    raise RuntimeError("worker did not terminate")

start() creates another execution path. join() waits for termination; its timeout does not kill the worker. The GIL does not make a multi-step shared-state transition atomic.

with condition:
    while not ready():
        condition.wait()

    change_shared_state()
    condition.notify_all()

waiter sleeps and releases lock
→ notifier changes state and notifies while still holding lock
→ notifier releases lock
→ notified waiters compete for lock
→ winner reacquires it; only then does wait() return
→ while predicate is checked again

Primitive selection

Closeable bounded queue

# Producer
with condition:
    while len(items) >= capacity and not closed:
        condition.wait()
    if closed:
        raise QueueClosed
    items.append(item)
    condition.notify_all()

# Consumer: drain after closure
with condition:
    while not items and not closed:
        condition.wait()
    if not items:
        raise QueueClosed
    item = items.popleft()
    condition.notify_all()
    return item

# Idempotent closure
with condition:
    if closed:
        return
    closed = True
    condition.notify_all()

An event does not replace this condition. Producers wait for “space or closed”; consumers wait for “item or closed.” Closure belongs beside the deque under the same lock.

Total timeout

deadline = None if timeout is None else time.monotonic() + timeout

with condition:
    while not ready() and not closed:
        if deadline is None:
            condition.wait()
        else:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return False
            condition.wait(remaining)

    if closed:
        raise QueueClosed

The loop exits because the predicate succeeded, terminal state was observed, or the one total deadline expired.

Standard thread queue: queue.Queue

Queue is Python's built-in synchronized queue for communication between threads. It contains the locking and condition logic needed to block producers while full and consumers while empty.

from queue import Queue

jobs = Queue(maxsize=2)
jobs.put(job)       # calling thread blocks while full
job = jobs.get()    # calling thread blocks while empty

Unfinished-work accounting

put()       opens a work ticket
get()       assigns the ticket
task_done() closes the ticket
join()      waits until no tickets remain open

jobs.join() waits for work acknowledgement. thread.join() waits for a worker function to terminate. A persistent worker loop commonly receives one sentinel per worker and calls task_done() in finally.

Semaphore: counted permits

A Semaphore stores a count of interchangeable permits. In this example, permits is a semaphore initialized with three permits, so at most three callers may use the resource concurrently.

from threading import Semaphore

permits = Semaphore(3)

permits.acquire()
try:
    use_resource()
finally:
    permits.release()

acquire() consumes one permit, blocking when the count is zero. release() returns one permit. This snippet limits concurrent access to a resource; it is not itself a queue.

Using semaphores to implement a bounded queue

from collections import deque
from threading import Lock, Semaphore

items = deque()
data_lock = Lock()
empty_slots = Semaphore(capacity)   # producer may reserve space
available_items = Semaphore(0)     # consumer waits for published work

# Producer: acquire empty slot → append under lock → release available item
# Consumer: acquire available item → pop under lock → release empty slot

The two semaphores count different facts. A separate lock protects the deque while it is changed.

Event: a persistent signal

An Event is a thread-safe boolean signal. Once set, current and future waiters can observe it until it is cleared.

from threading import Event

ready = Event()

def worker():
    ready.wait()       # block until configuration is ready
    use_configuration()

load_configuration()
ready.set()            # current and future waiters may proceed

Use an event for a simple fact such as “configuration is ready” or “stop requested.” It does not protect a deque or replace a condition that waits on a compound predicate.

Thread pool and futures

A thread pool runs independent blocking functions on a bounded set of reusable worker threads. Each submission returns a Future representing its eventual result or exception.

from concurrent.futures import ThreadPoolExecutor, as_completed

future_to_input = {}
with ThreadPoolExecutor(max_workers=4) as pool:
    for value in values:
        future = pool.submit(process, value)
        future_to_input[future] = value

    for future in as_completed(future_to_input):
        value = future_to_input[future]
        try:
            result = future.result()
        except Exception as exc:
            errors.append((value, exc))
        else:
            results.append((value, result))

A future's result() returns its value or re-raises the worker exception. Submission, start and completion order are separate.

Minimum asyncio

async def main():
    task_a = asyncio.create_task(worker("A", 0.2))
    task_b = asyncio.create_task(worker("B", 0.1))
    result_a = await task_a
    result_b = await task_b

asyncio.run(main())

The event loop is the scheduler. create_task() makes a coroutine eligible to run. An incomplete await suspends only the current task. time.sleep() blocks the loop thread; await asyncio.sleep() yields it.

await jobs.put(item)
item = await jobs.get()
jobs.task_done()
await jobs.join()

asyncio.Queue coordinates event-loop tasks and is not a cross-thread queue. Cancellation is cooperative: request it with task.cancel(), then await the task so cleanup runs and the terminal state is observed.

Final correctness pass

• Are all relevant reads and writes protected by the same lock?
• Does every wait use a while predicate?
• Is shared state changed before notification?
• Can every blocked caller observe closure?
• Can two locks be acquired in opposite orders?
• Can an exception leak a lock, permit or work ticket?
• Does timeout logic use one monotonic deadline?
• Do graph queries preserve stored indegrees?
• Are duplicate and missing identifiers defined?
• Are time and space complexity stated?
• Is the correct base runnable before another extension begins?

Official Python documentation

• collections.deque
• bisect
• heapq
• threading
• queue
• concurrent.futures
• asyncio