From 8f84707d2cf3d708a1b5984056344dff078effa2 Mon Sep 17 00:00:00 2001 From: chelsealong Date: Wed, 9 Sep 2026 13:20:23 +0000 Subject: [PATCH] fix(deep_crawling): O(n^2) parent lookup in BFS, duplicate enqueue in BestFirst BFSDeepCrawlStrategy re-scanned the entire current_level list once per fetched result to find that result's parent URL, making per-level bookkeeping O(n^2) instead of O(n). Build a url->parent dict once per level instead. BestFirstCrawlingStrategy.link_discovery only checked `visited` without updating it, so a URL discovered by two sibling pages processed in the same batch was scored and pushed onto the priority queue twice before either copy was dequeued. Track newly discovered URLs in a separate `_enqueued` set (kept distinct from `visited`, which the dequeue loop relies on to mean "already processed") so a repeat discovery within the same crawl is skipped. Fixes #2242 --- crawl4ai/deep_crawling/bff_strategy.py | 16 +- crawl4ai/deep_crawling/bfs_strategy.py | 8 +- .../test_deep_crawl_efficiency.py | 157 ++++++++++++++++++ 3 files changed, 176 insertions(+), 5 deletions(-) create mode 100644 tests/deep_crawling/test_deep_crawl_efficiency.py diff --git a/crawl4ai/deep_crawling/bff_strategy.py b/crawl4ai/deep_crawling/bff_strategy.py index 511fde692..2a068f4d6 100644 --- a/crawl4ai/deep_crawling/bff_strategy.py +++ b/crawl4ai/deep_crawling/bff_strategy.py @@ -71,6 +71,10 @@ def __init__( self._last_state: Optional[Dict[str, Any]] = None # Shadow list for queue items (only used when on_state_change is set) self._queue_shadow: Optional[List[Tuple[float, int, str, Optional[str]]]] = None + # URLs already scored/enqueued this crawl. Kept separate from `visited` + # (which tracks dequeued URLs) so a URL discovered by two sibling pages + # in the same batch is only pushed onto the priority queue once. + self._enqueued: Set[str] = set() async def can_process_url(self, url: str, depth: int) -> bool: """ @@ -177,12 +181,16 @@ async def link_discovery( for link in links: url = link.get("href") base_url = normalize_url_for_deep_crawl(url, source_url) - if base_url in visited: + if base_url in visited or base_url in self._enqueued: continue if not await self.can_process_url(base_url, new_depth): self.stats.urls_skipped += 1 continue - + + # Mark as enqueued now (not just when it's later put on the queue) + # so a second inbound link to the same URL, discovered from a + # sibling page later in this batch, is skipped here too. + self._enqueued.add(base_url) valid_links.append(base_url) # Record the new depths and add to next_links @@ -216,6 +224,9 @@ async def _arun_best_first( queue_items = self._resume_state.get("queue_items", []) for item in queue_items: await queue.put((item["score"], item["depth"], item["url"], item["parent_url"])) + # Already-visited and already-queued URLs are both settled; new + # discoveries should skip them. + self._enqueued = visited | {item["url"] for item in queue_items} # Initialize shadow list if callback is set if self._on_state_change: self._queue_shadow = [ @@ -228,6 +239,7 @@ async def _arun_best_first( await queue.put((-initial_score, 0, start_url, None)) visited: Set[str] = set() depths: Dict[str, int] = {start_url: 0} + self._enqueued = {start_url} # Initialize shadow list if callback is set if self._on_state_change: self._queue_shadow = [(-initial_score, 0, start_url, None)] diff --git a/crawl4ai/deep_crawling/bfs_strategy.py b/crawl4ai/deep_crawling/bfs_strategy.py index dfb759272..af95d16e7 100644 --- a/crawl4ai/deep_crawling/bfs_strategy.py +++ b/crawl4ai/deep_crawling/bfs_strategy.py @@ -248,6 +248,7 @@ async def _arun_batch( next_level: List[Tuple[str, Optional[str]]] = [] urls = [url for url, _ in current_level] + parent_by_url: Dict[str, Optional[str]] = dict(current_level) # Clone the config to disable deep crawling recursion and enforce batch mode. batch_config = config.clone(deep_crawl_strategy=None, stream=False) @@ -258,7 +259,7 @@ async def _arun_batch( depth = depths.get(url, 0) result.metadata = result.metadata or {} result.metadata["depth"] = depth - parent_url = next((parent for (u, parent) in current_level if u == url), None) + parent_url = parent_by_url.get(url) result.metadata["parent_url"] = parent_url results.append(result) @@ -336,11 +337,12 @@ async def _arun_stream( next_level: List[Tuple[str, Optional[str]]] = [] urls = [url for url, _ in current_level] + parent_by_url: Dict[str, Optional[str]] = dict(current_level) visited.update(urls) stream_config = config.clone(deep_crawl_strategy=None, stream=True) stream_gen = await crawler.arun_many(urls=urls, config=stream_config) - + # Keep track of processed results for this batch results_count = 0 async for result in stream_gen: @@ -348,7 +350,7 @@ async def _arun_stream( depth = depths.get(url, 0) result.metadata = result.metadata or {} result.metadata["depth"] = depth - parent_url = next((parent for (u, parent) in current_level if u == url), None) + parent_url = parent_by_url.get(url) result.metadata["parent_url"] = parent_url # Count only successful crawls diff --git a/tests/deep_crawling/test_deep_crawl_efficiency.py b/tests/deep_crawling/test_deep_crawl_efficiency.py new file mode 100644 index 000000000..ebb6dac11 --- /dev/null +++ b/tests/deep_crawling/test_deep_crawl_efficiency.py @@ -0,0 +1,157 @@ +""" +Regression tests for GH issue #2242: + +1. BFSDeepCrawlStrategy re-scanned the entire current level (a Python list) + once per fetched result to find that result's parent URL, making the + per-level bookkeeping O(n^2) instead of O(n). +2. BestFirstCrawlingStrategy.link_discovery only checked `visited` without + updating it, so a URL discovered by two sibling pages in the same batch + was scored and enqueued twice instead of once. +""" + +import asyncio +import time + +import pytest +from unittest.mock import MagicMock + +from crawl4ai.deep_crawling import BFSDeepCrawlStrategy, BestFirstCrawlingStrategy + + +def create_mock_config(stream=False): + config = MagicMock() + config.stream = stream + + def clone_config(**kwargs): + new_config = MagicMock() + new_config.stream = kwargs.get("stream", stream) + new_config.clone = MagicMock(side_effect=clone_config) + return new_config + + config.clone = MagicMock(side_effect=clone_config) + return config + + +def create_fanout_crawler(start_url, num_children): + """Mock crawler where `start_url` links to `num_children` leaf pages.""" + + async def mock_arun_many(urls, config): + results = [] + for url in urls: + result = MagicMock() + result.url = url + result.success = True + result.metadata = {} + if url == start_url: + links = [ + {"href": f"{start_url}/child{i}"} for i in range(num_children) + ] + else: + links = [] + result.links = {"internal": links, "external": []} + results.append(result) + return results + + crawler = MagicMock() + crawler.arun_many = mock_arun_many + return crawler + + +class TestBFSParentLookupPerformance: + """ + The parent lookup used to be `next((parent for (u, parent) in + current_level if u == url), None)`, executed once per result in the + level: O(n) work per result, O(n^2) for the whole level. Measure the + strategy's own wall-clock growth between a small and a large level - + a self-relative check that doesn't depend on absolute machine speed. + Linear behavior keeps the ratio close to the level-size ratio (8x); + quadratic behavior pushes it towards its square (64x). + """ + + @pytest.mark.asyncio + async def test_parent_lookup_scales_linearly_with_level_size(self): + start_url = "https://example.com/start" + small_n, large_n = 2500, 20000 + + async def timed_run(num_children): + strategy = BFSDeepCrawlStrategy(max_depth=2) + crawler = create_fanout_crawler(start_url, num_children) + config = create_mock_config(stream=False) + t0 = time.perf_counter() + results = await strategy._arun_batch(start_url, crawler, config) + elapsed = time.perf_counter() - t0 + assert len(results) == num_children + 1 + return elapsed + + small_time = await timed_run(small_n) + large_time = await timed_run(large_n) + + growth = large_time / small_time + # Level size grows 8x; O(n) bookkeeping keeps growth near 8x while + # O(n^2) bookkeeping pushes it towards 64x. 12x cleanly separates + # the two on this workload (measured ~8.3x fixed, ~16.9x unfixed). + assert growth < 12, ( + f"per-level bookkeeping does not scale linearly: {small_n} " + f"URLs took {small_time:.3f}s, {large_n} URLs took " + f"{large_time:.3f}s ({growth:.1f}x for an {large_n / small_n:.0f}x " + "increase in level size)" + ) + + +class CountingScorer: + """Records how many times each URL is scored.""" + + def __init__(self): + self.call_counts = {} + + def score(self, url): + self.call_counts[url] = self.call_counts.get(url, 0) + 1 + return 0.0 + + +class TestBestFirstDuplicateEnqueue: + """ + `link_discovery` checked `if base_url in visited: continue` but never + added newly discovered URLs to `visited`. When two sibling pages in the + same batch link to the same third URL, that URL was scored and pushed + onto the priority queue twice before either copy was ever dequeued. + """ + + @pytest.mark.asyncio + async def test_shared_link_is_scored_and_enqueued_once(self): + start_url = "https://example.com/start" + page_a = "https://example.com/a" + page_b = "https://example.com/b" + shared = "https://example.com/shared" + + async def mock_arun_many(urls, config): + async def gen(): + for url in urls: + result = MagicMock() + result.url = url + result.success = True + result.metadata = {} + if url == start_url: + links = [{"href": page_a}, {"href": page_b}] + elif url in (page_a, page_b): + links = [{"href": shared}] + else: + links = [] + result.links = {"internal": links, "external": []} + yield result + + return gen() + + crawler = MagicMock() + crawler.arun_many = mock_arun_many + config = create_mock_config(stream=True) + + scorer = CountingScorer() + strategy = BestFirstCrawlingStrategy(max_depth=2, url_scorer=scorer) + + results = [] + async for result in strategy._arun_best_first(start_url, crawler, config): + results.append(result) + + assert scorer.call_counts.get(shared, 0) == 1 + assert sum(1 for r in results if r.url == shared) == 1