feat: parallelize fast-mode BFS in the crawler bridge
The bridge BFS fetched pages one-at-a-time, so --fast only dropped the crawl-delay (sequential, zero-wait). Fast_mode's CPU*3 page-worker budget never reached the path operators actually run. Replace the popleft loop with a wave loop: each iteration pulls up to `fetcher.max_page_workers` URLs off the queue front and fetches them with asyncio.gather. Width is CPU*3 under fast_mode, 1 otherwise, so the polite path stays byte-for-byte sequential and the per-page crawl-delay still serialises same-domain fetches. Wave size is capped to the remaining max_pages budget; dedup moves from pop-time to enqueue-time so a URL linked from two parents in one wave is fetched exactly once. Measured on russell.ballestrini.net (own host, robots 404): same 12-page work 23.1s polite -> 4.0s fast (5.7x); full 243-page crawl ~25s vs the ~486s polite floor (19x). Disallow still honored; only the rate limit is lifted. Tests: peak-in-flight pins (>1 fast, ==1 polite) plus all existing BFS bound / dedup / depth / max-pages cases on the width=1 path.
This commit is contained in:
parent
aec4b544ab
commit
8298bf8618
2 changed files with 127 additions and 35 deletions
|
|
@ -217,6 +217,66 @@ def test_crawl_seed_passes_extract_links_true():
|
|||
)
|
||||
|
||||
|
||||
class _ConcurrencyFetcher:
|
||||
"""Stub that records peak in-flight fetches so a test can prove the
|
||||
BFS fans out (or doesn't). `max_page_workers` mimics what fast_mode
|
||||
sets on the real AsyncWebFetcher; the bridge keys wave width off it.
|
||||
"""
|
||||
|
||||
def __init__(self, page_links: dict[str, list[str]], max_page_workers: int):
|
||||
self.page_links = page_links
|
||||
self.max_page_workers = max_page_workers
|
||||
self._in_flight = 0
|
||||
self.peak_in_flight = 0
|
||||
|
||||
async def fetch_webpage(self, url: str, extract_links: bool = False, **kw):
|
||||
import asyncio
|
||||
|
||||
self._in_flight += 1
|
||||
self.peak_in_flight = max(self.peak_in_flight, self._in_flight)
|
||||
try:
|
||||
await asyncio.sleep(0.02) # hold the slot so overlap is observable
|
||||
return "<html></html>", self.page_links.get(url, [])
|
||||
finally:
|
||||
self._in_flight -= 1
|
||||
|
||||
|
||||
def test_crawl_seed_fast_fetches_a_wave_concurrently():
|
||||
"""fast_mode sets max_page_workers>1 → a level of sibling pages is
|
||||
fetched with multiple requests in flight, not one-at-a-time."""
|
||||
import asyncio
|
||||
|
||||
from arborist.sources.crawler import bridge
|
||||
|
||||
# Seed fans out to 6 leaf pages, all at depth 1 — one wave.
|
||||
page_links = {"https://x.com/": [f"/p{i}" for i in range(6)]}
|
||||
fetcher = _ConcurrencyFetcher(page_links, max_page_workers=6)
|
||||
out = asyncio.run(
|
||||
bridge._crawl_seed_async(
|
||||
"https://x.com/", max_depth=1, max_pages=0, fetcher=fetcher
|
||||
)
|
||||
)
|
||||
assert len(out) == 7 # seed + 6 leaves
|
||||
assert fetcher.peak_in_flight > 1, "fast crawl must overlap fetches in a wave"
|
||||
|
||||
|
||||
def test_crawl_seed_polite_fetches_one_at_a_time():
|
||||
"""max_page_workers==1 (polite default) → strictly sequential, so the
|
||||
per-page crawl-delay still serialises same-domain fetches."""
|
||||
import asyncio
|
||||
|
||||
from arborist.sources.crawler import bridge
|
||||
|
||||
page_links = {"https://x.com/": [f"/p{i}" for i in range(6)]}
|
||||
fetcher = _ConcurrencyFetcher(page_links, max_page_workers=1)
|
||||
asyncio.run(
|
||||
bridge._crawl_seed_async(
|
||||
"https://x.com/", max_depth=1, max_pages=0, fetcher=fetcher
|
||||
)
|
||||
)
|
||||
assert fetcher.peak_in_flight == 1, "polite crawl must stay one-at-a-time"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ingest_crawled: stores ETag + Last-Modified per document_root
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue