Measuring Before Optimising

Warm-up: from Day 5

From memory first.

1. Predict both lines. Job 2 fails while job 3 is still waiting.

Warm-up: predict Python

import asyncio

async def job(n):
    await asyncio.sleep(0.01 * n)
    if n == 2:
        raise ValueError(f"job {n}")
    return n

async def main():
    try:
        async with asyncio.TaskGroup() as group:
            tasks = [group.create_task(job(n)) for n in (1, 2, 3)]
    except* ValueError as errors:
        print("failed:", [str(e) for e in errors.exceptions])
    print([t.cancelled() for t in tasks])

asyncio.run(main())

Check your prediction

failed: ['job 2']
[False, False, True]

Job 1 finished before anything went wrong. When job 2 raised, the group cancelled job 3, and reported the failure as an ExceptionGroup, which except* unpacks. From Tasks, Groups and Exception Groups.

2. Fix. At most two fetches should run at once: this should print peak: 2.

Warm-up: fix Python

import asyncio

running = 0
peak = 0

async def fetch(n):
    global running, peak
    limit = asyncio.Semaphore(2)
    async with limit:
        running += 1
        peak = max(peak, running)
        await asyncio.sleep(0.01)
        running -= 1

async def main():
    await asyncio.gather(*(fetch(n) for n in range(6)))
    print("peak:", peak)

asyncio.run(main())

Check your fix

import asyncio

running = 0
peak = 0

async def fetch(n, limit):
    global running, peak
    async with limit:
        running += 1
        peak = max(peak, running)
        await asyncio.sleep(0.01)
        running -= 1

async def main():
    limit = asyncio.Semaphore(2)
    await asyncio.gather(*(fetch(n, limit) for n in range(6)))
    print("peak:", peak)

asyncio.run(main())

peak: 2

Each call made its own semaphore, so nothing was shared and nothing was limited. A limit only works when every task waits on the same one. From Bounding Concurrency.

3. Explain. No code: why is a bounded queue, on its own, not enough to bound how much a pipeline holds in memory?

Check your explanation

Values a worker has produced but can’t yet hand to a full queue are in memory too, and so is any input read eagerly up front. The real bound counts them all — workers + queue_size + 1 — and it only holds if input is pulled lazily. From Day 5 Checkpoint: Backpressure and the Leak.

Today

Every experienced programmer has optimised the wrong thing. The defence isn’t intuition — it’s a measurement you can repeat, taken on a workload that resembles what your users actually do.

Today is about producing those measurements for the library, and then acting on exactly what they say.

A workload big enough to be measurable

Timing one call tells you almost nothing: at microsecond scale you are mostly measuring the clock, the interpreter warming up, and whatever else the machine is doing.

How noisy a single measurement is Python

import time


def work(items):
    return sum(index * index for index in range(items))


for run in range(3):
    start = time.perf_counter()
    work(100)
    print(f"100 items, run {run}: {(time.perf_counter() - start) * 1e6:8.1f} us")

print()
for run in range(3):
    start = time.perf_counter()
    work(200_000)
    print(f"200k items, run {run}: {(time.perf_counter() - start) * 1e3:8.2f} ms")

The small runs vary by large factors between identical calls. The big ones agree to a few percent. That ratio — noise to signal — is the only reason to prefer a big workload; it isn’t about being realistic, it’s about being readable.

time.perf_counter() is the right clock: monotonic and the highest resolution available. Never use time.time() for durations — it can jump backwards when the system clock is adjusted.

timeit for the small stuff

For a single expression, timeit handles repetition and picks the best of several runs: python -m timeit -s "setup" "expression". Best-of is deliberate — you want the least-interrupted run, not the average, because interruptions only ever add time.

Reading a profile

cProfile records every call. Run it over the whole benchmark and you profile everything — the setup, and both the no-latency and latency runs:

python -m cProfile -s tottime benchmark.py

That’s rarely what you want. To profile one path in isolation — here, loading 2000 rows one at a time, with no latency and the setup excluded — profile just that call:

python -c "import benchmark, cProfile; \
           c = benchmark.connect(); ids = benchmark.fill(c); \
           benchmark.Post._instances.clear(); \
           cProfile.run('benchmark.one_by_one(c, ids)', sort='tottime')"

Its output, trimmed to the first few rows (line numbers are from an earlier day’s models.py; yours will differ):

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
     2000    0.002    0.000    0.010    0.000 models.py:97(get)
     2000    0.001    0.000    0.001    0.000 {method 'execute' of 'sqlite3.Connection' objects}
     2000    0.001    0.000    0.001    0.000 models.py:87(_remember)
     2000    0.001    0.000    0.004    0.000 models.py:43(__init__)
     2000    0.001    0.000    0.001    0.000 {method 'fetchone' of 'sqlite3.Cursor' objects}
     2000    0.001    0.000    0.001    0.000 fields.py:71(check)
     4000    0.001    0.000    0.003    0.000 fields.py:42(__set__)
     2000    0.001    0.000    0.001    0.000 fields.py:89(check)

Four columns matter:

ncalls — how many times. Often the most informative number on the page: __set__ runs 4000 times because each of 2000 posts has two fields.
tottime — time in that function excluding what it called. Sort by this to find where the CPU actually goes.
cumtime — time including callees. Sort by this to find which high-level operation is expensive.
percall — the same, divided by ncalls.

Read the profile above and there is no smoking gun: no single line dominates, and the work is spread evenly across query, construction and validation. That is a genuine and common result, and it means “make the code faster” is the wrong question. The right one is “why are we doing this 2000 times?”

The profiler changes what it measures

cProfile adds overhead to every call, so function-call-heavy code looks relatively worse under it than in production. Use it to find where the time goes; use plain timing to decide whether a change actually helped.

Counters beat clocks for tests

Wall-clock assertions in a test suite are a menace: they pass on your laptop and fail on a busy CI machine, and people learn to ignore them.

Count the operation instead. For a data library, the number of queries is both the thing that matters and perfectly deterministic:

class CountingConnection:
    def __init__(self, connection):
        self.real = connection
        self.queries: list[str] = []

    def execute(self, sql, parameters=(), /):
        self.queries.append(sql)
        return self.real.execute(sql, parameters)

That’s the whole trick, and it works because of Day 3: Connection is a protocol, so anything with execute and commit can stand in for the real thing.

    def test_get_many_costs_one_query(self) -> None:
        Post.get_many(self.connection, list(range(1, 11)))
        self.assertEqual(self.connection.selects, 1)

That test fails the day someone reintroduces a query per row — on any machine, at any load, in milliseconds.

Exercise

Count the work, not the time

summarise builds a report and is accidentally quadratic: it re-scans every record for every record. Add the counter that catches that, then fix it — a sort plus two linear passes turns O(n²) into O(n log n). The tests describe what the counter is allowed to grow like.

import unittest


class Counter:
    """TODO: count how many times `comparisons` is incremented."""


def summarise(records):
    """Return (name, rank) for each record, where rank is how many records
    have more views. Currently O(n^2): it re-scans for every record."""
    result = []
    for record in records:
        rank = 1
        for other in records:
            if other["views"] > record["views"]:
                rank += 1
        result.append((record["name"], rank))
    return result


def make(count):
    return [{"name": f"post {index}", "views": index} for index in range(count)]


class Summaries(unittest.TestCase):
    def test_it_ranks_correctly(self):
        records = [{"name": "a", "views": 5}, {"name": "b", "views": 9},
                   {"name": "c", "views": 1}]
        self.assertEqual(sorted(summarise(records)), [("a", 2), ("b", 1), ("c", 3)])

    def test_the_work_grows_with_the_input_not_its_square(self):
        small = Counter()
        summarise(make(50), counter=small)
        large = Counter()
        summarise(make(500), counter=large)
        # 10x the input should be about 10x the work, not 100x.
        self.assertLess(large.comparisons, small.comparisons * 20)

    def test_ties_share_a_rank(self):
        records = [{"name": "a", "views": 5}, {"name": "b", "views": 5}]
        self.assertEqual(sorted(summarise(records)), [("a", 1), ("b", 1)])


unittest.TextTestRunner(verbosity=2).run(
    unittest.defaultTestLoader.loadTestsFromTestCase(Summaries)
)

Hint 1: the idea

Two separate jobs. The Counter only has to count — one attribute, one method that adds to it. The fix to summarise is the idea in the exercise text: instead of comparing every record with every other, sort once, give each distinct number of views the rank of its first position, then look ranks up.

Hint 2: a first step

Write Counter first, and give summarise a counter=None parameter that counts each pass of the existing nested loop — then watch the growth test fail for the right reason. Then replace the nested loop: sorted by views, highest first, and a dictionary from views to rank. Ties share the first rank their views reached.

Show one solution

class Counter:
    def __init__(self):
        self.comparisons = 0

    def count(self, n=1):
        self.comparisons += n


def summarise(records, counter=None):
    counter = counter or Counter()
    ordered = sorted(records, key=lambda record: -record["views"])   # O(n log n)

    result = []
    rank_by_views = {}
    for position, record in enumerate(ordered, start=1):
        counter.count()                          # one linear pass to assign ranks
        rank_by_views.setdefault(record["views"], position)

    for record in records:
        counter.count()                          # one linear pass to build the result
        result.append((record["name"], rank_by_views[record["views"]]))
    return result

Sorting once and then looking ranks up in a dict replaces the nested scan. The sort is O(n log n) in the worst case — Python’s Timsort is O(n) on input that’s already ordered, which these test records happen to be, so don’t read the test’s numbers as proof of the worst case — and the two passes after it are linear.

Now be honest about what the counter measures. It counts the two explicit passes — 2n — and it does not count the comparisons sorted does inside, because sorted doesn’t expose them. So the counter is a proxy: it stays linear as long as you don’t reintroduce a nested loop, which is exactly the regression worth guarding against. At 50 records the naive version does 2,500 comparisons in code you can see; the improved version’s counted work is 100, plus a few hundred the sort does out of sight.

That caveat is the real lesson: a self-incremented counter catches a quadratic regression only in the work you actually instrument. An uncounted nested loop slips right past it, and it can’t see inside library calls. Here that’s fine, because a reintroduced O(n²) scan shows up in the counted passes; if the thing you feared lived inside sorted, you’d need a different measurement. assertLess(large, small * 20) is then a shape assertion on the counted work: 10x the input, well under 20x the count, and it fails loudly if someone brings the nested scan back.

counter=None with a fresh default keeps the counter optional, so production callers don’t have to know it exists — and it’s a keyword argument, so the signature stays free to change (Day 3).

What you learned

Small measurements are mostly noise; size the workload until repeated runs agree.
perf_counter for durations, timeit for expressions, cProfile for where the time goes — sorted by tottime for hot code, cumtime for expensive operations.
ncalls often answers the question before the timings do.
Tests should assert on counts, not seconds.