A document-processing service handles ten uploads comfortably, then slows down or crashes when a customer submits thousands. Adding workers helps briefly, but memory keeps climbing and failures leave jobs stuck. The confusion comes from treating memory management, concurrency, and testing as separate concerns. They meet at one practical question: how much unfinished work can the system safely hold?

The core idea: give unfinished work a boundary

Memory management means controlling the lifetime and size of data and releasing resources when work finishes. Concurrency means keeping multiple operations in progress; parallelism means executing operations simultaneously. Concurrency can improve throughput, the amount of work completed per unit of time, but it also creates more places to retain data.

Think of a restaurant kitchen. Order tickets wait on a rack, cooks take tickets, and each cook needs counter space. Adding cooks does little if the ticket rack grows indefinitely or every ticket arrives with an entire grocery delivery attached.

A queue is that ticket rack: it holds work until a worker, a task responsible for processing jobs, takes it. An event can be the kitchen's readiness bell: it signals a condition without storing separate orders. Limit the rack, limit the cooks, and limit each cook's workspace. Then test those boundaries, including what happens when a cook fails.

Follow a document through the system

Our running example accepts text documents, counts their bytes, and computes a SHA-256 digest, a fixed-length fingerprint of the contents. A digest helps detect changes; it does not authenticate who supplied a document.

The service follows five steps:

  1. Accept a stable document. Finish the upload before scanning it. Apply size limits and authorize access to its storage location; changing a file during scanning can produce a result without a consistent snapshot.
  2. Queue a reference. Store a file path or document identifier, rather than the document's bytes. A queue with capacity four holds at most four waiting references.
  3. Wait when capacity is exhausted. A producer, the code submitting jobs, pauses until a worker removes an item. This is backpressure: downstream capacity controls upstream admission.
  4. Process incrementally. Three workers read documents in 64 KiB (65,536-byte) chunks, updating byte counts and digests. Each worker keeps only a small portion of its current document in memory.
  5. Complete and drain. Deliver each result, mark the queue item finished, and stop workers after accepted work drains. A failure must reach the caller instead of leaving it waiting forever.
uploaded files -> bounded queue -> 3 workers -> results
                 paths only       chunked reads

Memory needs several limits

A bounded queue limits item count, not all the memory of a process, a running program. Four file paths are cheap; four complete documents might consume gigabytes.

Memory source Boundary in this example Waiting work Four references Active work Three workers Input held by each worker 64 KiB reads, plus buffering and temporary allocations Finished results Emitted incrementally, without a growing collection

These boundaries still leave memory for Python itself, execution overhead, and file buffering. A callback that saves every result in a list can reintroduce unbounded growth.

Python automatically manages object memory. CPython, the usual Python implementation, commonly uses reference counting, which tracks references to each object, plus garbage collection of unreachable reference cycles, where objects refer back to one another. Collection cannot reclaim documents still referenced by a cache, a store kept for reuse, or a results list. Released objects do not guarantee that process memory immediately returns to the operating system.

Files and sockets also need explicit closure. The example's with statement uses a context manager, which performs cleanup when its block exits, including after an exception.

Choose how workers execute

An event loop schedules cooperating asynchronous tasks, units of work that can pause while waiting. An await allows other tasks to run when its operation actually suspends; merely declaring a function async does not make blocking work harmless.

A thread is an execution path within a process that shares its memory. Parsing means interpreting a document's structure.

Pattern Useful for the document service Main cost or constraint Sequential processing Small batches with simple requirements One slow document delays the next Async workers Waiting on asynchronous storage or network clients Blocking work stalls the event loop Threads Existing blocking file or network libraries Shared memory needs coordination Processes Substantial processor work, such as some parsers Separate memory and data-transfer overhead

Threads can race when they modify shared state: the result depends on timing. A lock, which permits one holder at a time, can protect such updates. Passing results back to one owner often makes coordination simpler.

Processor-intensive Python code often benefits from processes. Libraries implemented in compiled code and Python build configurations can change whether threads provide parallel execution, so measure the actual parser.

Hands-on: a bounded file scanner

Use Python 3.11 or newer, which provides TaskGroup and asyncio.timeout, and a Bash environment with sort. Python's standard-library asyncio module coordinates asynchronous work; no third-party packages are required. Save this as pipeline.py:

import asyncio
import hashlib
from pathlib import Path

CHUNK_SIZE = 64 * 1024


def inspect_file(path):
    size = 0
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        while chunk := stream.read(CHUNK_SIZE):
            size += len(chunk)
            digest.update(chunk)
    return path.name, size, digest.hexdigest()


async def scan(path):
    return await asyncio.to_thread(inspect_file, path)


def emit(record):
    name, size, digest = record
    print(f"{name}: {size} bytes, sha256={digest[:12]}")


async def run(paths, workers=3, backlog=4, scanner=scan, on_result=emit):
    if workers < 1 or backlog < 1:
        raise ValueError("workers and backlog must be positive")
    queue = asyncio.Queue(maxsize=backlog)

    async def consume():
        while True:
            path = await queue.get()
            try:
                if path is None:
                    return
                on_result(await scanner(path))
            finally:
                queue.task_done()

    async with asyncio.TaskGroup() as group:
        for _ in range(workers):
            group.create_task(consume())
        for path in paths:
            await queue.put(path)
        await queue.join()
        for _ in range(workers):
            await queue.put(None)


if __name__ == "__main__":
    asyncio.run(run(Path("demo-docs").glob("*.txt")))

asyncio.to_thread moves the blocking reader to a thread so it does not run directly on the event loop. TaskGroup keeps the workers within one managed scope: it waits for them and cancels remaining asynchronous work when a worker fails.

Every successful get() needs one task_done(), including the None items used to stop workers. queue.join() waits for that accounting to reach zero. It does not prove that results reached durable storage, meaning storage that survives a process restart.

Create three documents and run the scanner:

mkdir -p demo-docs
printf 'alpha\n' > demo-docs/a.txt
printf 'beta\n' > demo-docs/b.txt
printf 'gamma\n' > demo-docs/c.txt
python3 pipeline.py | sort

Expected output:

a.txt: 6 bytes, sha256=b6a98d9ce9a2
b.txt: 5 bytes, sha256=f2c82decdd71
c.txt: 6 bytes, sha256=ae9a6306a205

Completion order can vary; sort makes this tiny demonstration reproducible. The displayed digest is shortened for readability. Use the complete digest returned by inspect_file for comparisons.

There are two deliberate production limitations. emit is synchronous, so a slow result destination would block the event loop; replace it with an appropriate asynchronous writer. Cancelling an await of to_thread does not terminate an already-running reader thread, which can continue reading and delay shutdown.

Test behavior at several scales

Tests should establish properties the service promises: correct results, limited active work, and visible failures. They should not require a particular scheduling speed.

For coordination, asyncio.Event stores a boolean flag. set() makes the flag true and wakes waiters; wait() returns when it is true. Repeated set() calls do not record multiple occurrences. Our worker test uses one event to signal that three workers started and another to release them. Grouped task failures arrive as an ExceptionGroup, a container holding reported exceptions.

Save this as test_pipeline.py:

import asyncio
import hashlib
import tempfile
import unittest
from pathlib import Path

from pipeline import CHUNK_SIZE, inspect_file, run


class PipelineTests(unittest.IsolatedAsyncioTestCase):
    def test_streaming_checksum(self):
        data = b"x" * (CHUNK_SIZE + 17)
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "large.txt"
            path.write_bytes(data)
            self.assertEqual(
                inspect_file(path),
                ("large.txt", len(data), hashlib.sha256(data).hexdigest()),
            )

    async def test_worker_limit(self):
        active = peak = 0
        ready, release = asyncio.Event(), asyncio.Event()
        results = []

        async def controlled_scan(path):
            nonlocal active, peak
            active += 1
            peak = max(peak, active)
            if active == 3:
                ready.set()
            try:
                await release.wait()
                return path
            finally:
                active -= 1

        async with asyncio.timeout(5):
            async with asyncio.TaskGroup() as group:
                group.create_task(run(
                    range(12), scanner=controlled_scan,
                    on_result=results.append,
                ))
                await ready.wait()
                self.assertEqual(peak, 3)
                release.set()
        self.assertEqual(peak, 3)
        self.assertCountEqual(results, range(12))

    async def test_failure_reaches_caller(self):
        async def fail(path):
            raise OSError("unreadable")

        with self.assertRaises(ExceptionGroup) as caught:
            async with asyncio.timeout(5):
                await run(range(12), scanner=fail)
        self.assertIsInstance(caught.exception.exceptions[0], OSError)

Run:

python3 -m unittest -v test_pipeline.py

Expected results, with timing and separator lines omitted:

test_failure_reaches_caller (test_pipeline.PipelineTests.test_failure_reaches_caller) ... ok
test_streaming_checksum (test_pipeline.PipelineTests.test_streaming_checksum) ... ok
test_worker_limit (test_pipeline.PipelineTests.test_worker_limit) ... ok
OK

The checksum test crosses a chunk boundary. The worker test verifies the concurrency limit and one result per submitted input. The failure test checks that a worker error reaches the caller even while admission is bounded.

The five-second timeouts prevent tests from hanging indefinitely; they are safeguards rather than performance targets. These tests passed on Python 3.12.

Test level What to exercise as the service grows Unit: one small behavior Empty files, chunk boundaries, parser errors, duplicate-handling rules Integration: real components together Storage permissions, completed writes, restart recovery Load: sustained and burst traffic Queue depth, oldest-job age, completion time, memory trend

Measure both normal traffic and recovery after a dependency slows down. A service may survive a burst yet accumulate memory over hours. Repeat representative loads in a controlled environment and compare trends; avoid fragile assertions about exact memory usage across platforms.

Queues, events, and their tradeoffs

A local queue distributes jobs: one worker retrieves each item. It disappears when the process exits. That suits this finite scanner, but accepted uploads that must survive a crash need durable storage and a recoverable job mechanism.

A message broker is a service that accepts and delivers messages between processes. Durability depends on its configuration and acknowledgment rules, not simply on using a broker. An acknowledgment tells it that a consumer, the receiving worker, has completed a message.

Mechanism Use it when Choose an alternative when Direct function call The caller needs an immediate result Long processing must outlive the request Bounded local queue One process needs a controlled work backlog Accepted jobs must survive restart Durable job queue Workers need independent deployment and recovery Operational complexity exceeds the benefit In-process event flag Tasks need readiness or shutdown signaling Every occurrence must carry data and be counted Published event message Several consumers react to a completed fact One consumer should perform one assigned job

An architectural event differs from an asyncio.Event flag. A published DocumentScanned message describes something that happened, with fields such as document ID, version, and full digest. A search service and an audit service that records activity can each receive it through publish/subscribe, delivery to multiple interested consumers. An in-process flag cannot replace that message history.

Durable delivery also introduces duplicates. A worker might save a digest, then crash before acknowledgment; the broker delivers the job again. Make writes idempotent, meaning repetition has the same effect as one execution, using a stable document ID and version.

Queues absorb temporary bursts; they do not create processing capacity. Events add flexibility, but separate consumers can lag behind one another. Use direct calls when their simplicity meets the requirement.

Common mistakes and their fixes

· Launching a task for every document. A semaphore, a permit counter limiting access, can restrict active processing while thousands of waiting tasks still consume memory. Fix: combine bounded admission with a fixed worker count. · Trusting garbage collection to cap memory. Live references in queues, caches, or callbacks remain live. Fix: stream large inputs, bound caches, and emit results incrementally. · Treating completion accounting as successful delivery. task_done() can run after failure, and an in-memory queue provides no recovery record. Fix: propagate failures and acknowledge durable jobs after saving the required result. · Retrying every error immediately. Permanent input errors cannot heal, and repeated retries can overload storage. Fix: distinguish temporary failures, limit attempts, and increase delays between retries. · Testing timing instead of behavior. Arbitrary sleeps pass locally and fail under a busy test runner. Fix: coordinate with events and test cancellation, shutdown, and duplicate delivery explicitly. · Trusting uploaded paths and contents. An untrusted path can reach another customer's files, and a compressed upload can expand far beyond its submitted size. Fix: use authorized storage references, cap expanded input, and avoid logging document contents.

Quick recap

  1. Bound waiting jobs, active workers, per-job input, and retained results separately.
  2. Choose concurrency around the actual work: waiting, blocking library calls, or processor-intensive computation.
  3. Use queues for payload-bearing jobs and event flags for shared conditions.
  4. Treat durable delivery, acknowledgment, and duplicate handling as explicit design decisions.
  5. Test correctness and coordination first, then validate integration, load, and recovery behavior.

Test yourself

  1. Does a queue of four items guarantee bounded memory? Answer: No. Item sizes, active workers, waiting producers, caches, and retained results can still grow. Queue references and read documents incrementally.
  2. Can three calls to asyncio.Event.set() represent three jobs? Answer: No. The event stores a flag, not an occurrence count or payload. Put three items into a queue when each job needs separate processing.
  3. Why can a successfully processed durable job run again? Answer: The worker may fail after saving its result but before acknowledgment. Use a stable job identity and idempotent writes so another delivery does not duplicate effects.

Where to go next

· Structured concurrency: keeping related tasks in a scope with predictable cancellation and shutdown; build on the example's TaskGroup. · Transactional outbox: recording a result and a pending event together in a database transaction, an all-or-nothing group of writes, then publishing it reliably. · Memory profiling: tracing allocations and retained objects to explain why the scanner's memory grows during sustained operation.