Python concurrency from first principles

1. One program, multiple threads

A normal Python program begins with one execution path: the main thread. Sequential code has an unavoidable order:

main thread: put(1) → put(2) → get()

The call to get() cannot begin until put(2) returns. Creating worker threads introduces several paths that the scheduler may interleave:

producer thread: put(1) → put(2) → put(3)
consumer thread: get()  → get()  → get()
main thread:     start both, then wait for both

Correct concurrent code must work under every allowed interleaving, not only the order seen during one test run.

The “calling thread”

The calling thread is whichever thread makes a particular call. If a producer executes condition.wait(), the producer is the calling thread. If the main thread calls worker.join(), the main thread is the one that waits.

Shared state creates the problem

Threads in one process can access the same Python objects. That makes communication easy, but it also creates races. Consider two producers running an unprotected check:

if len(items) == 0:
    items.append(item)

Both producers can observe an empty collection before either appends. A supposed one-slot mailbox can end with two items. This is a race condition: correctness depends on unpredictable timing.

2. A lock protects an invariant

A lock admits only one thread at a time into a protected critical section. Choose that section by identifying the invariant that must remain true while shared state changes.

For a one-slot mailbox, the invariant is:

0 ≤ len(items) ≤ 1

For a bounded queue:

0 ≤ len(items) ≤ capacity

Every check and mutation relevant to that invariant must occur while holding the same lock. Otherwise another thread can change the state between checking it and acting on it.

A public observer such as size() should acquire the lock because callers may invoke it independently:

def size(self) -> int:
    with self.condition:
        return len(self.items)

The return value is still only a snapshot. Another thread may change the queue immediately after the method releases the lock.

3. A condition is a lock plus a waiting room

A Condition combines an underlying lock with a place where threads can sleep until shared state may have changed.

Predicates are questions about state

A predicate is an expression whose answer is true or false. For a bounded queue:

# Producer: is there room?
len(items) < capacity

# Consumer: is there an item?
len(items) > 0

A thread may proceed only when its own predicate is true.

What wait() actually does

A thread cannot sleep while keeping the lock: nobody else could acquire it to create the state being awaited. condition.wait() therefore:

1. Releases the condition lock.
2. Blocks the calling thread.
3. Becomes eligible to wake after a notification or timeout.
4. Reacquires the lock before returning.
5. Lets the caller check its predicate again while protected.

consumer owns lock and sees no item
→ wait() puts consumer in waiting set, releases lock, sleeps
→ producer acquires lock, appends item, calls notify_all()
→ consumer is eligible to wake, but producer still owns lock
→ producer releases lock
→ notified consumers compete to reacquire it
→ winner acquires lock; only now does wait() return
→ while predicate is checked again under the lock

There may be several notified contenders, but there is no promised FIFO “awakened queue”. Notification makes a waiter eligible to compete for the lock. It does not hand over the lock or reserve the item.

In code, wait() means: the current state blocks this operation; release the lock, sleep, then reacquire the lock and check again.

Why waiting uses while, not if

Suppose two consumers wait on an empty queue. A producer adds one item and wakes both. They do not enter the critical section simultaneously; they compete to reacquire the one lock. The winner removes the item. The second consumer eventually acquires the lock and finds the queue empty again.

while not self.items:
    self.condition.wait()

Notification means “something may have changed”, not “your predicate is now true”. A notification is also not stored for future waiters. Correctness lives in the shared state and predicate.

Condition protocol

1. Acquire the lock.
2. Check the predicate in a loop.
3. Wait if the operation is not allowed.
4. Change the shared state.
5. Notify waiters after the change.
6. Release the lock.

4. Starting, joining and observing threads

producer_thread = Thread(target=producer)
consumer_thread = Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

producer_thread.join(timeout=2)
consumer_thread.join(timeout=2)

Constructing a Thread does not run its target. start() creates a separate execution path and returns without waiting for the target to finish. It does not promise whether the new worker or the calling thread gets the next scheduler turn.

Thread(target=producer)    # pass the function
Thread(target=producer())  # call it now in the current thread: wrong shape

join() blocks the calling thread until the target thread terminates or the timeout expires. The name describes separate control-flow paths meeting again:

main  ───── start worker ───── join(worker) ───── continue
                 ╲                ╱
worker            ─── do work ───

A join timeout limits how long the caller waits. It neither cancels nor kills the worker. Since join() always returns None, tests inspect liveness separately:

worker.join(timeout=2)
assert not worker.is_alive(), "worker is stuck"

A sequential blocking-queue test can deadlock even when the queue is correct:

mailbox.put(1)
mailbox.put(2)  # the only thread blocks here
mailbox.get()   # never reached

Releasing the lock inside wait() does not create another execution path. A complementary worker is still required.

5. From a mailbox to a bounded queue

A one-slot mailbox and a bounded queue use the same protocol. Only the producer predicate changes:

while len(self.items) == self.capacity:
    self.condition.wait()

A complete condition-based queue has a symmetric shape:

def put(self, item):
    with self.condition:
        while len(self.items) == self.capacity:
            self.condition.wait()
        self.items.append(item)
        self.condition.notify_all()

def get(self):
    with self.condition:
        while not self.items:
            self.condition.wait()
        item = self.items.popleft()
        self.condition.notify_all()
        return item

Tests should verify promises the synchronization actually makes: every produced item arrives exactly once, no thread remains alive, and the final queue is empty. They should not invent a global ordering guarantee that the scheduler never promised.

6. Semaphores are counted permits

A semaphore represents a number of available permits:

acquire(): counter > 0 → subtract one and proceed
           counter = 0 → wait

release(): add one and potentially wake a waiter

The literal analogy is a car park handing out entry tokens. With three tokens, at most three cars may enter. Returning a token allows another car to proceed.

permits = Semaphore(2)

assert permits.acquire(blocking=False) is True   # 2 → 1
assert permits.acquire(blocking=False) is True   # 1 → 0
assert permits.acquire(blocking=False) is False  # remains 0
permits.release()                                # 0 → 1

Why a bounded queue uses two semaphores

empty_slots     starts at capacity
available_items starts at 0

A producer transfers one permit from empty slots to available items. A consumer performs the inverse:

producer: acquire empty slot → append → release available item
consumer: acquire item       → remove → release empty slot

The counts are not safely derivable from each other during an in-flight operation. After a producer reserves a slot but before it appends, that slot is no longer available—but no consumable item exists yet. A permit is a reservation, not merely a report of deque length.

A separate lock still protects the deque. Semaphores control how many operations may proceed; they do not make a multi-step mutation mutually exclusive.

def put(self, item):
    self.empty_slots.acquire()
    with self.lock:
        self.items.append(item)
    self.available_items.release()

Acquire the permit before the data lock. Waiting for a permit while holding the deque lock could prevent the complementary operation from acquiring that lock and creating the resource being awaited.

7. Shutdown is a state transition

A blocking queue works because a temporarily impossible operation may become possible later:

full queue  → a consumer may create space
empty queue → a producer may publish an item

Shutdown removes that promise. If an empty queue is permanently closed, a consumer must not wait for an item that can never arrive. A producer already waiting for space must also wake and learn that new work is no longer accepted.

Add a lifecycle flag under the same condition lock that protects the queue:

def close(self) -> None:
    with self.condition:
        if self.closed:
            return

        self.closed = True
        self.condition.notify_all()

Calling close() repeatedly has the same effect as calling it once. This is idempotent behaviour.

8. Waiting predicates now have two exits

Before shutdown, a producer waits only for space. After shutdown exists, the loop must end for either of two reasons: space appeared, or the queue closed.

while len(self.items) == self.capacity and not self.closed:
    self.condition.wait()

if self.closed:
    raise QueueClosed

self.items.append(item)
self.condition.notify_all()

The check after the loop matters. Waking does not prove that the producer received space; it may mean closure made waiting pointless.

The consumer has a different rule

Closure should not discard work already accepted. A consumer waits only while the queue is both empty and open:

while not self.items and not self.closed:
    self.condition.wait()

if not self.items:
    raise QueueClosed

item = self.items.popleft()
self.condition.notify_all()
return item

• If an item exists, return it even when the queue is closed.
• If no item exists, the loop can have ended because closure means no future item can arrive.

9. Closure must wake every waiter

A normal queue mutation may enable one complementary operation. Closure is different: it changes the meaning of waiting for every producer and every consumer.

That is why close() uses notify_all(). Waking only one thread could leave the others asleep forever even though the queue will never change again.

Closure still does not kill threads. It is a cooperative protocol:

1. One thread publishes the terminal state.
2. Blocked workers wake and reacquire the condition lock one at a time.
3. Each worker re-checks its predicate and returns or raises.
4. The coordinating thread joins the workers.

A worker stops because its target function finishes—not because close() or join() forcibly terminates it.

10. Test safety, liveness and errors separately

Safety

Did an impossible state occur? Examples include exceeding capacity, losing or duplicating an item, or accepting a put after closure.

Liveness

Did every worker eventually make progress and terminate?

worker.join(timeout=2)
assert not worker.is_alive(), "worker is stuck"

The timeout detects a hang. It does not kill the worker.

Error visibility

An exception in a raw worker thread does not automatically propagate through Thread.join(). A small test can catch worker exceptions into a result list that the main thread inspects after joining.

Avoid correctness sleeps

sleep() may make one scheduling order more likely, but it does not prove that a thread reached a desired state. Prefer explicit coordination when a test needs to establish that a worker started. Keep join timeouts as failure detectors, not as the mechanism that makes the test correct.

11. Implementation exercise

Extend a condition-based BoundedBlockingQueue with the following contract:

• Add a small QueueClosed exception.
• Add lifecycle state protected by the existing condition lock.
• Make close() idempotent.
• Reject every new put() after closure.
• Wake producers that were blocked on a full queue.
• Let consumers drain items accepted before closure.
• Make get() fail once the closed queue is empty.
• Wake consumers that were blocked on an empty queue.
• Test blocked producer, blocked consumer, draining and repeated closure with bounded joins.

Use the existing Condition so the lifecycle predicates remain explicit. Leave out events, sentinels, standard-library queues, and async runtimes for this version.

12. Event: a persistent one-way signal

A threading.Event wraps a thread-safe boolean flag:

from threading import Event

stop_requested = Event()        # initially false
stop_requested.set()            # false → true; wake waiters
stop_requested.is_set()         # inspect the flag
stop_requested.wait(timeout=2)  # wait until true or timeout
stop_requested.clear()          # true → false

An event behaves like a persistent raised flag. Once raised, current waiters and later callers of wait() can both see it. A condition notification disappears when nobody is waiting.

Events are useful for facts such as shutdown requested, configuration ready, or workers allowed to begin. They do not protect a deque, count resources, carry a result, or explain why work failed.

model_ready = Event()

def worker():
    model_ready.wait()
    use_model()

load_model()
model_ready.set()

A worker created after set() also passes wait() immediately. The signal remains visible until it is explicitly cleared.

Why queue closure stays under the condition

The closeable queue waits on compound rules: a producer needs space or closure; a consumer needs an item or closure. A producer asleep inside condition.wait() is not awakened merely because a separate event is set. close() would still need to acquire the condition lock, change lifecycle state consistently with the deque, and notify those condition waiters.

Using an event as well would duplicate the closed state. Keeping the boolean beside the deque under one condition gives one atomic state model and one waiting mechanism. An additional event is useful only if unrelated external observers need to wait solely for closure.

For a one-way stop signal, call set() once and normally never clear it. Rapidly setting and clearing introduces a protocol in which threads can miss the brief true state.

13. Barrier: meet at a phase boundary

A barrier is a rendezvous point for a fixed number of participants:

from threading import Barrier

ready = Barrier(3, timeout=2)

def worker():
    prepare()
    ready.wait()
    perform_timed_phase()

Every participant blocks at wait() until all three arrive; they are then released. This is useful for coordinated test starts or algorithms with discrete phases.

A condition waits for a predicate over changing shared state. A barrier waits for a fixed number of participants to reach a point. If one participant never arrives, the others can wait forever, so tests should use a timeout. A timeout or abort breaks the barrier and causes waiters to raise BrokenBarrierError.

14. queue.Queue: the standard-library implementation

The standard library provides a synchronized multi-producer, multi-consumer queue:

from queue import Queue

jobs = Queue(maxsize=2)
jobs.put(job)     # block while full
job = jobs.get()  # block while empty

The custom condition-based queue exposes the invariant and waiting protocol. In application code, prefer queue.Queue unless the required contract is missing from its API.

Work retrieval is not work completion

Queue tracks unfinished work. Every put() increments an internal counter. A consumer calls task_done() only after processing the retrieved item:

def worker():
    while True:
        job = jobs.get()
        try:
            process(job)
        finally:
            jobs.task_done()

jobs.join()  # wait for the unfinished-work count to reach zero

Thread.join() → wait for one thread's function to terminate
Queue.join()  → wait for every put() to receive a task_done()

Calling task_done() in finally avoids leaving join() blocked if processing raises. Modern Python also provides Queue.shutdown(), but it was added in Python 3.13; code targeting older environments cannot assume it exists.

Avoid check-then-act logic based on empty(), full(), or qsize(). Their answers are snapshots another thread may invalidate immediately. Express the desired behaviour through blocking put() and get().

Using it with worker threads

Queue does not create threads. Application code still creates and joins them; the queue supplies the synchronized transfer and backpressure:

jobs = Queue(maxsize=2)
stop = object()

def worker():
    while True:
        item = jobs.get()
        try:
            if item is stop:
                return
            process(item)
        finally:
            jobs.task_done()

workers = [Thread(target=worker) for _ in range(2)]
for thread in workers:
    thread.start()

for item in work:
    jobs.put(item)
for _ in workers:
    jobs.put(stop)

jobs.join()                  # all submitted work acknowledged
for thread in workers:
    thread.join()            # every worker function terminated

Submit one sentinel per worker. The sentinel is acknowledged in finally like every other queue entry, allowing the unfinished-work count to reach zero.

15. ThreadPoolExecutor and futures

Creating raw threads gives direct control. A pool instead maintains a bounded set of reusable worker threads and accepts many tasks:

from concurrent.futures import ThreadPoolExecutor, as_completed

with ThreadPoolExecutor(max_workers=4) as pool:
    futures = [pool.submit(fetch_document, doc_id) for doc_id in ids]

    for future in as_completed(futures):
        document = future.result()

submit() returns quickly with a Future, which represents one result that may not exist yet:

pending → running → completed with a value
                  → completed with an exception

future.result() waits if necessary, returns the value, and re-raises the worker exception in the calling thread. That last behaviour makes pools easier to test than raw threads: Thread.join() does not propagate the target's exception.

Submission order does not imply completion order. Use as_completed() when results should be handled as they finish.

submission order → when submit() is called
start order      → when a worker thread begins the call
completion order → when the call returns or raises

When a result or exception must be associated with its original input, retain an explicit mapping from each future to that input:

future_to_value = {}

for value in values:
    future = pool.submit(process, value)
    future_to_value[future] = value

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

Iterating over a dictionary yields its keys, so no separate list of futures is needed here. The mapping is especially useful when future.result() raises: there is no returned value to identify the failed input, but the future still selects the original input from the dictionary.

The workers execute only process(). The coordinator thread runs the as_completed() loop and is the only writer to the result and error lists, so those lists need no additional lock in this pattern.

Leaving the executor’s with block shuts the pool down and waits for submitted work by default. It stops new submissions but does not forcibly kill running functions. Cancellation usually works only before a task starts, so long-running functions still need cooperative shutdown.

A classic pool deadlock

If every worker is occupied by a task waiting for another future scheduled onto the same exhausted pool, the missing task can never start. Avoid blocking pool workers on work that requires a free worker from that same pool.

For ordinary CPython builds, thread pools are especially useful for overlapping blocking I/O such as network calls. Pure Python CPU-bound work does not usually become faster merely by adding threads because of the global interpreter lock. That performance detail does not eliminate shared-state races.

16. How the asyncio event loop runs two tasks

Start with the event loop

An event loop is a loop in the literal programming sense. It runs on one thread and repeatedly does three jobs:

1. Choose a task that is ready to run.
2. Run that task until it finishes or reaches an incomplete await.
3. When a timer or I/O operation completes, make the associated task ready again.

ready tasks: [main]

event loop chooses main
main runs until it must wait
event loop chooses another ready task
...
all requested work finishes

The loop has one Python instruction stream. Each turn belongs to one ready task. A task that suspends leaves the next turn available for another ready task.

Four names used by the API

Coroutine function

A function declared with async def.

Coroutine object

The value created when a coroutine function is called. The event loop begins executing its body after the coroutine is scheduled.

Task

A coroutine object registered with the event loop so that the loop can run it. asyncio.create_task() creates a task.

Await

Ask an asynchronous operation for its result. If the result is not ready, the current task is suspended and the loop may run another ready task. If the result is already ready, execution can continue immediately.

First complete program

import asyncio


async def worker(name: str, delay: float) -> None:
    print(name, "started")
    await asyncio.sleep(delay)
    print(name, "resumed after its timer")


async def main() -> None:
    print("main started")

    task_a = asyncio.create_task(worker("A", 0.2))
    task_b = asyncio.create_task(worker("B", 0.1))

    print("main created both tasks")

    await task_a
    await task_b

    print("main finished")


asyncio.run(main())

asyncio.run(main()) creates an event loop, gives it the main() coroutine, runs the loop until main finishes, and then closes the loop.

One execution trace

A and B are both ready after creation. The table shows the order produced by this run. Code should not rely on the start order of equally ready tasks; the timer values make B's completion occur first after both timers have been registered.

At an incomplete await, the running task gives control back to the event loop. The loop can then choose another ready task.

main can wait while the event loop continues

main and the event loop are different things. The event loop is the scheduler. main is one task managed by that scheduler. When main reaches await task_a, only main is suspended. A and B remain eligible to run.

running: main
ready:   A, B

main reaches await task_a
waiting: main (for A)
ready:   A, B

loop runs A → A awaits its timer → A waits
loop runs B → B awaits its timer → B waits
B finishes first
A finishes and makes main ready
main resumes; awaiting the already-finished B returns immediately

create_task() registers a coroutine with the running event loop and makes it eligible to run, then returns. It does not normally interrupt the current task and execute the new task's body immediately. The current task must finish or reach an incomplete await before the loop gets another scheduling turn.

An incomplete await pauses the current task without occupying the event-loop thread. If the awaited operation is already complete, it can return immediately without a task switch.

Thread.join() → pause the calling thread until another thread terminates
await task    → pause the current task while the event loop runs other tasks

Awaiting a task also returns its result or re-raises its exception. Thread.join() returns None and does not propagate the target function's exception. Use await asyncio.wait_for(task, timeout) when an asynchronous wait needs a timeout.

Why blocking calls freeze the loop

time.sleep(1) occupies the event-loop thread for one second. During that second the loop cannot choose another task. await asyncio.sleep(1) registers a timer and suspends the current task, leaving the loop free to run other ready tasks.

Then add an asynchronous queue

An asyncio.Queue uses the same event-loop mechanism. An incomplete await queue.put(item) suspends a producer when the queue is full. An incomplete await queue.get() suspends a consumer when the queue is empty.

First understand the unfinished-work counter

A queue contains more state than its stored items. It also maintains an integer called the unfinished-work counter. It starts at zero. Every successful put() adds one. get() removes and assigns an item but leaves this counter unchanged, because the queue cannot know when the consumer has finished processing it.

unfinished = 0
finished_event = set

put(item):
    store item
    unfinished += 1
    clear finished_event

task_done():
    if unfinished <= 0:
        raise ValueError
    unfinished -= 1
    if unfinished == 0:
        set finished_event

join():
    if unfinished > 0:
        await finished_event

finished_event is an internal asyncio.Event: a persistent boolean signal with a set of waiting tasks. The queue creates it in the set state because zero jobs are initially unfinished. A successful put() clears it. The task_done() call that reduces the counter to zero sets it again, waking every task currently waiting in join(). Future calls to join() return immediately while it remains set.

The signal means that all work currently represented by the counter has been acknowledged. It does not mean that a producer or consumer coroutine has returned. A later put() can clear the signal again.

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

task_done() is completion bookkeeping, not memory cleanup. In real processing code, acknowledge an item after processing, including when processing raises:

item = await queue.get()
try:
    await process(item)
finally:
    queue.task_done()

If an acknowledgement is missing, the unfinished count remains above zero and queue.join() keeps waiting. Calling task_done() too early can let join() return while work is still running. Calling it when the counter is already zero raises ValueError, preventing the completion ledger from becoming negative. The counter is global: Python does not tie each call to a particular retrieved item, so early acknowledgement remains a logic bug even when no exception is raised.

await queue.join() waits for the unfinished counter to reach zero. It does not wait merely for the item container to become empty. A consumer may have removed the final item and still be processing it.

How this differs from a semaphore

The two mechanisms both use counters and wake waiters, but they answer different questions:

Semaphore.acquire() → wait for a permit, then decrement permits
Semaphore.release() → increment permits and possibly wake a waiter

Queue.put()          → increment unfinished work
Queue.task_done()    → decrement unfinished work; never waits
Queue.join()         → wait for unfinished work to become zero

The unfinished-work mechanism is closer to a countdown latch or a completion ledger. Queue capacity is separate: put() uses the stored-item count to wait when the queue is full.

Complete producer-consumer example

import asyncio


async def producer(queue: asyncio.Queue) -> None:
    for item in [1, 2, 3]:
        print("producer wants to put", item)
        await queue.put(item)
        print("producer put", item)


async def consumer(queue: asyncio.Queue) -> None:
    for _ in range(3):
        await asyncio.sleep(0.1)
        item = await queue.get()
        print("consumer got", item)
        queue.task_done()


async def main() -> None:
    queue = asyncio.Queue(maxsize=2)

    producer_task = asyncio.create_task(producer(queue))
    consumer_task = asyncio.create_task(consumer(queue))

    await producer_task
    await queue.join()
    await consumer_task


asyncio.run(main())

Trace the third put. The producer has already filled both queue slots, so it suspends. The event loop runs the consumer. Once the consumer removes an item, the producer becomes ready and completes the third put.

Why the three waits have this order

await producer_task  # all work has been submitted
await queue.join()   # all submitted work has been acknowledged
await consumer_task  # the consumer terminated; propagate any exception

Waiting for the producer before joining the queue ensures that all work has first been submitted. If queue.join() runs while the unfinished count is still zero, it can return before the producer adds anything. The producer and consumer still run concurrently while main awaits the producer, because both tasks were scheduled before that await.

queue.join() cannot replace await consumer_task because the queue does not own or observe that coroutine's lifecycle. Immediately after its final task_done(), the consumer may still need to leave its loop, perform cleanup, or raise an exception. A long-running consumer may acknowledge the last item and then wait for another item forever. Awaiting the task proves that the consumer coroutine returned and propagates any exception it raised.

In this fixed three-item example, the final acknowledgement occurs just before the consumer returns, so the two events happen close together and the final await mainly makes the lifecycle check explicit. A long-running worker normally needs a separate shutdown step: wait for queue.join(), send a sentinel or cancel according to the contract, and then await the worker task.

Cancellation: request, delivery, cleanup, observation

task.cancel() requests cancellation; it does not synchronously kill the task. At the next opportunity, normally while the task is suspended at an await, the event loop causes CancelledError to be raised inside that task.

async def worker() -> None:
    print("worker: started")
    try:
        await asyncio.sleep(10)
        print("worker: completed normally")
    finally:
        print("worker: cleanup")


async def main() -> None:
    task = asyncio.create_task(worker())
    await asyncio.sleep(0.1)

    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("main: observed cancellation")

cancel() requests cancellation
CancelledError interrupts the worker's suspended sleep
the worker's finally block runs while its stack unwinds
the worker finishes in the cancelled state
await task raises CancelledError in main
main observes the cancellation

Awaiting after cancel() separates requesting cancellation from confirming its completion. It lets worker cleanup finish and makes the final state visible to the coordinator.

Cancellation is cooperative. If a task occupies the event-loop thread with a long CPU loop containing no await, the loop cannot schedule another task. In this example, main may not even resume from its timer to call cancel() until that loop ends. A pending request is likewise normally delivered only when the task next reaches a suspension point.

A timeout uses cancellation internally

asyncio.wait_for(worker(), timeout=0.1) applies a deadline around an awaited operation. When the deadline expires, wait_for() requests cancellation of the worker and waits for that cancellation to finish. The two layers observe different exceptions:

inside worker:   CancelledError
outside wait_for: TimeoutError

worker starts
the 0.1-second deadline expires
wait_for requests worker cancellation
CancelledError interrupts the worker's sleep
the worker's finally block performs cleanup
wait_for observes completed cancellation
wait_for raises TimeoutError to main

The resulting output is:

worker: started
worker: cleanup
main: timed out
main: finished

Cancellation is the mechanism used to stop the overdue operation. TimeoutError is the API result presented to the caller that established the deadline. Since Python 3.11 this is the built-in TimeoutError; older versions raised asyncio.TimeoutError.

Check your understanding

Each statement below should follow directly from the event-loop and queue mechanics described above:

• The event loop is the scheduler; main is one task managed by it.
• create_task() makes a coroutine eligible to run but does not synchronously execute its body at the call site.
• An incomplete await suspends only the current task, so the loop can run another ready task.
• time.sleep() occupies the event-loop thread, while await asyncio.sleep() yields it.
• Queue length and unfinished work are different state.
• put() opens a work ticket, get() assigns it, task_done() closes it, and join() waits for no open tickets.
• The queue's finished event wakes join() waiters when unfinished work reaches zero; it does not mean a consumer coroutine terminated.
• Awaiting the consumer separately verifies its termination and propagates its result or exception.

17. Concurrency checklist

1. What state is shared?
2. What invariant must always remain true?
3. Which lock protects every relevant check and mutation?
4. What predicate must be true for each kind of worker to proceed?
5. Does every waiter check that predicate in a while loop?
6. Is state changed before waiters are notified?
7. Can every blocked operation observe shutdown?
8. Could all workers wait for progress that requires another blocked worker?
9. Does the test check safety, liveness and worker errors separately?
10. Do timeouts only detect hangs, with correctness established through explicit coordination?