Problem 82 · Concurrency & Python

Asyncio Crawler with Bounded Concurrency

HardClassic DSA & systems

01Problem

Implement an asyncio crawler that, given an async fetch(url) -> list[str], visits all same-host pages while never running more than N fetches at once.

02Key insight

asyncio gives cheap concurrency for IO-bound work on a single thread, so 'seen' needs no lock (no await between check and add). An asyncio.Semaphore bounds in-flight requests; hold it only around the fetch, not while recursing, to avoid deadlock.

03Solution

solution.pypython27 lines

import asyncio
from typing import Awaitable, Callable
from urllib.parse import urldefrag, urljoin, urlparse


async def crawl_async(start_url: str, fetch: Callable[[str], Awaitable[list[str]]],
                      max_concurrency: int = 5) -> set[str]:
    host = urlparse(start_url).hostname
    seen: set[str] = {start_url}
    sem = asyncio.Semaphore(max_concurrency)

    async def visit(url: str) -> None:
        async with sem:  # bound only the IO, release before spawning children
            try:
                links = await fetch(url)
            except Exception:
                return
        children = []
        for link in links:
            child, _ = urldefrag(urljoin(url, link))
            if urlparse(child).hostname == host and child not in seen:
                seen.add(child)  # check+add has no await between: atomic on the event loop
                children.append(visit(child))
        await asyncio.gather(*children)

    await visit(start_url)
    return seen

Try it first.

TimeO(V+E)

SpaceO(V) plus pending coroutines

tests.py · 2 passing

tests.pypython14 lines

import asyncio
pages = {"http://s.io/": [f"/p{i}" for i in range(20)] + ["http://other.io/"]}
for i in range(20):
    pages[f"http://s.io/p{i}"] = ["/", f"/p{(i + 1) % 20}"]
state = {"active": 0, "peak": 0}
async def fake_fetch(url):
    state["active"] += 1
    state["peak"] = max(state["peak"], state["active"])
    await asyncio.sleep(0.001)
    state["active"] -= 1
    return pages.get(url, [])
res = asyncio.run(crawl_async("http://s.io/", fake_fetch, max_concurrency=3))
assert len(res) == 21 and "http://other.io/" not in res
assert state["peak"] == 3

04Staff-level follow-up

What happens with millions of URLs? Recursive gather holds a coroutine per pending page; switch to a bounded asyncio.Queue with N worker tasks so memory is O(N + frontier) and back-pressure is explicit.