Cancellation and Concurrency Limits
Make deadlines, cleanup, and backpressure part of a concurrent operation’s contract.
By the end, you’ll be able to…
- Propagate CancelledError after cleanup.
- Apply timeout scopes at meaningful boundaries.
- Use semaphores and bounded queues for different limits.
01Cancellation asks a task to stop
Task cancellation requests that CancelledError be raised at an opportunity for the task to resume. It is cooperative, so a task stuck in blocking code cannot respond promptly. Cleanup belongs in finally or a context manager. If you catch CancelledError to record a message, normally re-raise it so the caller and structured concurrency machinery observe cancellation correctly.
CancelledError is a BaseException subclass in modern Python. A routine except Exception usually leaves cancellation alone, which is often desirable. Swallowing cancellation can make TaskGroup shutdown and timeout scopes behave incorrectly. Cleanup should be bounded too: an endless cleanup operation can prevent a program from exiting even after its main operation has been cancelled.
02A deadline is a scope
asyncio.timeout bounds an async block and converts the cancellation it initiates into TimeoutError outside that context. Catch the TimeoutError outside the block when providing fallback behavior. Decide whether a deadline applies per request, to an entire batch, or to both; ten sequential requests with individual timeouts can still exceed a user’s overall latency budget.
A timeout is not a transaction rollback. A remote server may have accepted a write before the client times out. Retries therefore require idempotency or reconciliation. Likewise, cancelling a task awaiting to_thread does not stop the thread’s underlying operation. Associate a deadline with resources and effects explicitly instead of interpreting a timeout as proof that nothing happened.
03Limit access to scarce resources
An asyncio.Semaphore limits how many tasks enter a protected async region at once. Hold it around the operation that consumes the scarce resource, then release it through async with even if that operation fails. The semaphore represents a policy such as at most three active requests; it is not a requests-per-second rate limiter.
Creating one task per item in a million-item input still creates a million task objects even when a semaphore admits only three. For large or unbounded streams, use a bounded asyncio.Queue and a fixed number of workers. The queue provides backpressure: the producer waits when the buffer fills instead of accumulating unlimited pending work.
04Test shutdown, not only success
A concurrent feature is incomplete until it can stop. Test a timeout during work, cancellation while waiting for a permit, a failure in one sibling, and resource release after an exception. Prefer events and explicit handshakes to timing-sensitive tests that hope a task has started after an arbitrary sleep. A deterministic start signal makes cleanup assertions much more reliable.
Choose separate limits for active operations, queued items, response size, and total elapsed time. Log meaningful failure context while excluding secrets. Document whether partial results are returned, discarded, or durably saved. These policies turn a collection of tasks into an understandable service operation, and they give maintainers acceptance criteria beyond merely observing that a happy-path demo finishes.
The ideas worth keeping.
- Cancellation must propagate after bounded cleanup.
- Timeout scopes bound waits, not external side effects.
- Use fixed workers and bounded queues when pending work can grow.
Make it yours.
Type and test your code right here in the interactive workspace. Run your solution, check the output, and compare with the worked guide when ready.
Release after failure
Use a semaphore with one permit. Raise inside its protected block, then prove a later operation can acquire it.
Explore a worked solution
import asyncioasync def main(): semaphore = asyncio.Semaphore(1) try: async with semaphore: raise ValueError('fixture failure') except ValueError: pass async with asyncio.timeout(0.1): async with semaphore: return 'reacquired'assert asyncio.run(main()) == 'reacquired'The async context manager releases the permit during exception unwinding. The timeout makes a leaked permit fail the check instead of hanging forever.
A bounded worker queue
Process [1, 2, 3] with two workers and a queue whose capacity is two. Use one sentinel per worker and call task_done for every retrieved item.
Explore a worked solution
import asyncioasync def process(): queue = asyncio.Queue(maxsize=2) results = [] async def worker(): while True: value = await queue.get() try: if value is None: return results.append(value * 2) finally: queue.task_done() async with asyncio.TaskGroup() as group: for _ in range(2): group.create_task(worker()) for value in [1, 2, 3]: await queue.put(value) for _ in range(2): await queue.put(None) await queue.join() return sorted(results)assert asyncio.run(process()) == [2, 4, 6]A fixed worker count bounds tasks and maxsize bounds buffered items. Sentinels belong to this finite demonstration; production shutdown must also handle producer failure and cancellation.
What stayed with you?
Check the ideas while they’re fresh. Your answers stay in this page and feedback appears as you go.
Release after failure
Use a semaphore with one permit. Raise inside its protected block, then prove a later operation can acquire it.
[Process exited with code 0 - No output]
Keep exploring.
The official references behind this chapter, when you want a second angle or the full details.
Pause. Put the pieces together.
You’ve reached the end of Concurrency with Clear Boundaries. Before moving on, use this task to check what you can do independently.
Explain your model choice, surface worker failures, and verify cleanup when work is cancelled.
Try it in your own environment and write down the result. This is a self-check; reading completion does not evaluate your code.
Review this moduleOne chapter closer.
Reading is a start. Trying it yourself makes the connection.