Problem 81 · Concurrency & Python

Concurrent Web Crawler (Thread Pool)

HardClassic DSA & systems

01Problem

Given a start URL and a blocking fetch(url) -> list[str] of links, crawl every reachable page on the same hostname concurrently using a thread pool and return the set of visited URLs.

02Key insight

Crawling is IO-bound, so threads scale despite the GIL (the GIL is released during network waits). Keep all mutation of the 'seen' set on the coordinating thread and use wait(FIRST_COMPLETED) so workers only fetch; this avoids locks and duplicate fetches.

03Solution

solution.pypython31 lines

from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait
from typing import Callable
from urllib.parse import urldefrag, urljoin, urlparse


def _normalize(base: str, link: str) -> str:
    url, _ = urldefrag(urljoin(base, link))  # resolve relative, drop #fragment
    parts = urlparse(url)
    return parts._replace(path="/").geturl() if not parts.path else url  # http://a.com == http://a.com/


def crawl(start_url: str, fetch: Callable[[str], list[str]], max_workers: int = 8) -> set[str]:
    start_url = _normalize(start_url, start_url)  # same canonical form as discovered links
    host = urlparse(start_url).hostname
    seen: set[str] = {start_url}
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        pending: dict[Future, str] = {pool.submit(fetch, start_url): start_url}
        while pending:
            done, _ = wait(pending, return_when=FIRST_COMPLETED)
            for fut in done:
                base = pending.pop(fut)
                try:
                    links = fut.result()
                except Exception:
                    continue  # one bad page must not abort the whole crawl
                for link in links:
                    url = _normalize(base, link)
                    if urlparse(url).hostname == host and url not in seen:
                        seen.add(url)  # only the coordinator thread mutates `seen`
                        pending[pool.submit(fetch, url)] = url
    return seen

Try it first.

TimeO(V+E) fetches/link checks

SpaceO(V)

tests.py · 5 passing

tests.pypython21 lines

import threading
graph = {
    "http://a.com/": ["/x", "http://a.com/y#top", "http://b.com/z"],
    "http://a.com/x": ["/y", "/"],
    "http://a.com/y": ["/broken"],
}
calls = []
lock = threading.Lock()
def fake_fetch(url):
    with lock:
        calls.append(url)
    if url.endswith("/broken"):
        raise IOError("500")
    return graph.get(url, [])
res = crawl("http://a.com/", fake_fetch, max_workers=4)
assert res == {"http://a.com/", "http://a.com/x", "http://a.com/y", "http://a.com/broken"}
assert len(calls) == len(set(calls)) == 4  # no URL fetched twice
assert all("b.com" not in u for u in res)
calls.clear()
assert crawl("http://a.com", lambda u: calls.append(u) or ["/", "#top"]) == {"http://a.com/"}
assert calls == ["http://a.com/"]  # unnormalized start URL is not fetched twice

04Staff-level follow-up

How do you make it distributed and polite? Partition the frontier by hash(host) across workers so per-host rate limits and robots.txt live in one place, and persist the frontier/seen set (e.g. a KV store or Bloom filter) for crash recovery.