09 · Bounded work: queues, deadlines and ownership

A queue does not create processing capacity. It trades waiting time and memory for tolerance of bursts. If a notebook importer accepts work faster than it completes work indefinitely, its backlog grows indefinitely unless it rejects, drops, coalesces or blocks some arrivals. This is a systems problem before it is a Kubernetes or cloud problem.

Choose the boundary and the units

Distinguish three counts: received requests, admitted jobs and completed jobs. A rejected request is not a completed job. A retried request may refer to a job already completed. A queue-length graph without those definitions can hide lost work or double-counted success.

For a deliberately simple deterministic model, assume a queue initially empty, 12 jobs arriving each second, and one worker completing eight jobs per second. Assume arrivals are admitted before service in each discrete time step, service is always available, and every job has the same service cost. Without a limit, backlog grows by four jobs per step. After ten steps it contains 40 jobs; no amount of optimism about an “asynchronous architecture” changes the arithmetic.

def queue_model(arrivals, service_per_tick, capacity):
    if (type(service_per_tick) is not int or service_per_tick < 0
            or type(capacity) is not int or capacity < 0):
        raise ValueError("nonnegative integer service and capacity required")
    queued = accepted = completed = rejected = 0
    history = []
    for count in arrivals:
        if type(count) is not int or count < 0:
            raise ValueError("nonnegative integer arrivals required")
        admitted = min(count, capacity - queued)
        accepted += admitted
        rejected += count - admitted
        queued += admitted
        done = min(queued, service_per_tick)
        queued -= done
        completed += done
        assert accepted == completed + queued
        history.append(queued)
    return {"accepted": accepted, "completed": completed,
            "rejected": rejected, "queued": queued, "history": history}
 
unbounded_for_this_run = queue_model([12] * 10, 8, 1000)
assert unbounded_for_this_run["queued"] == 40
bounded = queue_model([12] * 10, 8, 20)
assert bounded["accepted"] == 92
assert bounded["completed"] == 80
assert bounded["rejected"] == 28
assert bounded["queued"] == 12
assert bounded["history"] == [4, 8, 12, 12, 12, 12, 12, 12, 12, 12]

Why does the bounded queue end at 12 rather than 20? Capacity is checked before eight jobs finish each tick. The transient pre-service queue reaches 20, then drains to 12. Reversing arrival and service order changes the finite trace. State that order instead of presenting a made-up simulation as a measured workload.

If service lasts exactly one eighth of a second per job with one worker and no further arrivals, 40 queued jobs need five seconds to drain. Real jobs have variable costs, parallel workers, failures and scheduling overhead. A mean throughput does not establish a p99 waiting time, and a count limit is not a byte limit. Twenty jobs with 100 MB payloads are a very different memory commitment from twenty small IDs.

Bounded admission is a behavior, not just a number

Python's official queue.Queue reference documents a maxsize bound, blocking insertion, nonblocking insertion with Full, and the approximate nature of qsize()/empty()/full() under concurrency. Use the operation's result, not a separate size check, to decide whether admission succeeded.

from queue import Queue, Full, Empty
 
work = Queue(maxsize=2)
accepted, rejected = [], []
for name in ["ingest", "validate", "publish"]:
    try:
        work.put_nowait(name)
    except Full:
        rejected.append(name)
    else:
        accepted.append(name)
assert accepted == ["ingest", "validate"]
assert rejected == ["publish"]
 
finished = []
while True:
    try:
        item = work.get_nowait()
    except Empty:
        break
    try:
        finished.append(item)  # stand-in for successful processing
    finally:
        work.task_done()
work.join()
assert finished == ["ingest", "validate"]

This example is sequential. task_done() tells the queue that handling of a retrieved item is accounted for; it does not persist the result or prove the business job succeeded. In a real worker, define how failure is recorded and whether retry is rescheduled before acknowledging queue bookkeeping. join() waits for that bookkeeping to reach zero. It cannot discover whether a worker lied.

Also note a subtle API mismatch: the model above defines capacity zero as admitting nothing, while Queue(maxsize=0) means an unbounded queue. Read the concrete API, not just a variable's name. A bounded deque is different again: appending to a full bounded deque discards an item from the opposite end, which may silently lose jobs if used as an admission queue.

A process owns resources; a timeout limits waiting

A process executes a program with operating-system-managed resources. A thread is an execution stream within a process; choosing threads does not provide independent durable job state. Connections, open files and worker slots should have a named owner and a release path. The SQL lesson uses finally: con.close() because the connection must close whether assertions succeed or fail.

Do not assume that “the caller stopped waiting” means “the work stopped.” The Python subprocess documentation explicitly states that a communicate() timeout does not kill the child process; the caller must perform cleanup. By contrast, subprocess.run(..., timeout=...) kills and waits for the direct child when its timeout expires, while process creation itself may not be interruptible. This distinction is source-backed here, not demonstrated by starting long-lived processes. Cleanup of a process tree or remote job is a separate problem.

The same reasoning applies to an HTTP deadline: after a timeout, a server may continue working. Retry safety belongs to the operation's identity and effect boundary, not to the timer. A cancellation token is a request for cooperating code to stop, not a time machine that reverses already committed work.

Exercises

1. After a burst, arrivals become four jobs per tick while service remains eight. How quickly does a backlog of 40 drain in the unbounded model?
2. Name a workload where dropping older queued work is valid and one where it is not.
3. Why can if not work.full(): work.put_nowait(job) still raise Full in a concurrent program?
4. What measurements would you need before promising a latency percentile?

Exit artifact: a conservation check, a bounded admission policy, and a resource-ownership note. These models do not validate OS scheduling, thread safety of application state, real latency, cloud capacity or a distributed queue.

Next: Build and assess the local notebook capstone.