"""Reference solution — Bounded-Concurrency Async Crawler. DO NOT READ BEFORE YOU HAVE RUN THE PROBLEM UNDER THE CLOCK. Teaching text: ../../../WARMUP.md § Chapter 10. The design decision that separates this from a naive crawler: TWO semaphores, not one. A global bound protects you; a per-host bound protects them. And the per-host bound must NOT be taken before the worker has picked up the URL, or one slow host serialises the whole frontier behind it -- workers sit blocked on a host semaphore while URLs for other hosts wait in the queue. The fix is to acquire the global slot first, then the host slot, and to keep enough workers in flight that a few blocked on a slow host cannot stall the rest. Dedupe happens at ENQUEUE time under a lock, so two workers discovering the same URL simultaneously cannot both schedule it. """ from __future__ import annotations import asyncio import random import time from urllib.parse import urlsplit, urlunsplit DEFAULT_PORTS = {"http": "80", "https": "443"} def normalize(url: str) -> str: """Strip the fragment, drop a default port, lowercase the host, and drop a trailing slash on a non-empty path. Without this, /a, /a/ and /a#x are three distinct URLs and you crawl the same page three times.""" parts = urlsplit(url) scheme = parts.scheme.lower() host = parts.hostname or "" port = parts.port if port is not None and str(port) != DEFAULT_PORTS.get(scheme): netloc = f"{host}:{port}" else: netloc = host path = parts.path or "/" if len(path) > 1 and path.endswith("/"): path = path.rstrip("/") return urlunsplit((scheme, netloc, path, parts.query, "")) def host_of(url: str) -> str: return urlsplit(url).hostname or "" class Crawler: def __init__(self, fetcher, concurrency=5, per_host=None, max_depth=0, max_attempts=1, base_backoff=0.01, cap_backoff=1.0, deadline=None, rng=None, clock=time.monotonic): if concurrency <= 0: raise ValueError("concurrency must be positive") if per_host is not None and per_host <= 0: raise ValueError("per_host must be positive") self._fetch = fetcher self.concurrency = concurrency self.per_host = per_host self.max_depth = max_depth self.max_attempts = max_attempts self.base_backoff = base_backoff self.cap_backoff = cap_backoff self.deadline = deadline self._rng = rng if rng is not None else random.Random(0) self._clock = clock self.failures = {} self.peak_in_flight = 0 self.peak_per_host = {} self._in_flight = 0 self._host_in_flight = {} # ---- the two bounds -------------------------------------------------- def _host_sem(self, host): if self.per_host is None: return None return self._host_sems.setdefault(host, asyncio.Semaphore(self.per_host)) async def _fetch_one(self, url): host = host_of(url) # Global slot FIRST, host slot second. Taking the host slot first would # let a slow host hold workers that could be serving other hosts. async with self._global_sem: sem = self._host_sem(host) if sem is None: return await self._instrumented(url, host) async with sem: return await self._instrumented(url, host) async def _instrumented(self, url, host): self._in_flight += 1 self._host_in_flight[host] = self._host_in_flight.get(host, 0) + 1 self.peak_in_flight = max(self.peak_in_flight, self._in_flight) self.peak_per_host[host] = max(self.peak_per_host.get(host, 0), self._host_in_flight[host]) try: return await self._fetch(url) finally: self._in_flight -= 1 self._host_in_flight[host] -= 1 # ---- retries --------------------------------------------------------- async def _fetch_with_retries(self, url): last = None for attempt in range(self.max_attempts): try: return await self._fetch_one(url) except asyncio.CancelledError: raise except Exception as exc: last = exc if attempt + 1 >= self.max_attempts: break ceiling = min(self.cap_backoff, self.base_backoff * (2 ** attempt)) await asyncio.sleep(self._rng.uniform(0, ceiling)) self.failures[url] = last return None # ---- the crawl ------------------------------------------------------- async def stream(self, seeds): """Yield (url, result) as they arrive, so the consumer can throttle.""" self._global_sem = asyncio.Semaphore(self.concurrency) self._host_sems = {} self.failures = {} self.peak_in_flight = 0 self.peak_per_host = {} queue = asyncio.Queue() seen = set() results = asyncio.Queue() started = self._clock() pending = 0 lock = asyncio.Lock() async def enqueue(url, depth): """Dedupe at ENQUEUE time, under a lock, so two workers that discover the same URL simultaneously cannot both schedule it.""" nonlocal pending canonical = normalize(url) async with lock: if canonical in seen or depth > self.max_depth: return seen.add(canonical) pending += 1 await queue.put((canonical, depth)) for seed in seeds: await enqueue(seed, 0) if pending == 0: # Nothing to do. Return before starting workers, or the sentinel # that ends the stream is never produced and the consumer hangs. return async def worker(): nonlocal pending while True: url, depth = await queue.get() try: if self.deadline is not None and \ self._clock() - started > self.deadline: continue outcome = await self._fetch_with_retries(url) if outcome is None: continue body, links = outcome await results.put((url, body)) if depth < self.max_depth: for link in links: await enqueue(link, depth + 1) finally: async with lock: pending -= 1 done = pending == 0 queue.task_done() if done: await results.put(None) # the sentinel workers = [asyncio.create_task(worker()) for _ in range(max(self.concurrency, 1) * 2)] try: while True: if self.deadline is not None: remaining = self.deadline - (self._clock() - started) if remaining <= 0: break try: item = await asyncio.wait_for(results.get(), remaining) except (asyncio.TimeoutError, TimeoutError): break else: item = await results.get() if item is None: break yield item finally: for task in workers: task.cancel() await asyncio.gather(*workers, return_exceptions=True) async def crawl(self, seeds): out = {} async for url, body in self.stream(seeds): out[url] = body return out