From 1e92635a188315255ae57c68eb05fa142cfcd713 Mon Sep 17 00:00:00 2001 From: Adrian Date: Sun, 9 Aug 2026 11:01:06 +0200 Subject: [PATCH] Benchmark item processing and item concurrency (#7954) --- tests/benchmarks/test_crawl.py | 115 +++++++++++++++++++++++++++++++-- 1 file changed, 110 insertions(+), 5 deletions(-) diff --git a/tests/benchmarks/test_crawl.py b/tests/benchmarks/test_crawl.py index d3bf0fdd6..f79b66a5b 100644 --- a/tests/benchmarks/test_crawl.py +++ b/tests/benchmarks/test_crawl.py @@ -1,5 +1,7 @@ from __future__ import annotations +import asyncio +from collections import Counter from typing import TYPE_CHECKING, Any from urllib.parse import urlencode @@ -29,11 +31,23 @@ LINKS_PER_PAGE = 5 REQUESTS = 200 BROAD_DEEP_PAGES = 10 -# Requests per crawl and delay of the benchmark that measures delayed requests, -# where wall time, unlike in the other benchmarks, is a function of the delay. +# Requests per crawl and delay of the benchmarks that wait, where wall time, +# unlike in the other benchmarks, is a function of the delay. DELAYED_REQUESTS = 50 DELAY = 0.005 +# Requests per crawl and items per response of the benchmarks that measure item +# processing, which reaches fewer pages than the other benchmarks because every +# page costs it several items. +ITEM_REQUESTS = 20 +ITEMS_PER_RESPONSE = 100 + +# Item concurrency limits of the benchmarks that measure item processing. The +# high limit is above the number of items that a response yields in any of +# them. +HIGH_CONCURRENT_ITEMS = 1000 +DELAYED_CONCURRENT_ITEMS = 50 + NULL_SETTINGS: dict[str, Any] = { "DOWNLOAD_HANDLERS": {"http": NullDownloadHandler}, "LOG_ENABLED": False, @@ -63,7 +77,8 @@ class _FollowSpider(Spider): class _TreeSpider(Spider): - """Crawl *pages* pages on each of *domains* hostnames. + """Crawl *pages* pages on each of *domains* hostnames, yielding *items* + items from every page. Pages are numbered from 1, and page *n* links to pages *2n* and *2n+1*, so that requests also reach the scheduler from callbacks, and not only from @@ -73,6 +88,7 @@ class _TreeSpider(Spider): name = "benchmark-tree" domains: int = 1 pages: int = 1 + items: int = 0 async def start(self) -> AsyncIterator[Any]: for domain in range(self.domains): @@ -83,6 +99,8 @@ class _TreeSpider(Spider): for child in (page * 2, page * 2 + 1): if child <= self.pages: yield Request(response.urljoin(f"/{child}")) + for _ in range(self.items): + yield _Page(url=response.url) class _Pipeline: @@ -90,12 +108,48 @@ class _Pipeline: return item -def _crawl_tree(settings: dict[str, Any], *, domains: int, pages: int) -> Crawler: +class _DelayedPipeline: + """Item pipeline that waits, so that the item concurrency limit applies. + + The peak number of items of a same response in progress is tracked in the + ``benchmark/peak_items`` stat. Items are counted per response because the + limit is per response, and the items of a response are processed while + later responses are already being downloaded. + """ + + def __init__(self, crawler: Crawler): + self._crawler = crawler + self._active: Counter[str] = Counter() + + @classmethod + def from_crawler(cls, crawler: Crawler) -> _DelayedPipeline: + return cls(crawler) + + async def process_item(self, item: Any) -> Any: + url = item["url"] + self._active[url] += 1 + assert self._crawler.stats + self._crawler.stats.max_value("benchmark/peak_items", self._active[url]) + try: + await asyncio.sleep(DELAY) + return item + finally: + self._active[url] -= 1 + + +def _crawl_tree( + settings: dict[str, Any], *, domains: int, pages: int, items: int = 0 +) -> Crawler: crawler = crawl( - _TreeSpider, {**NULL_SETTINGS, **settings}, domains=domains, pages=pages + _TreeSpider, + {**NULL_SETTINGS, **settings}, + domains=domains, + pages=pages, + items=items, ) assert crawler.stats assert crawler.stats.get_value("downloader/response_count") == domains * pages + assert crawler.stats.get_value("item_scraped_count", 0) == domains * pages * items return crawler @@ -163,3 +217,54 @@ def test_overhead_delay(benchmark: BenchmarkFixture) -> None: """ settings = {"DOWNLOAD_DELAY": DELAY, "RANDOMIZE_DOWNLOAD_DELAY": False} benchmark(lambda: _crawl_tree(settings, domains=1, pages=DELAYED_REQUESTS)) + + +@pytest.mark.parametrize( + ("items", "settings"), + [ + pytest.param(1, {}, id="single"), + pytest.param(ITEMS_PER_RESPONSE, {}, id="many"), + pytest.param( + 1, + {"CONCURRENT_ITEMS": HIGH_CONCURRENT_ITEMS}, + id="high-limit", + ), + ], +) +def test_overhead_items( + benchmark: BenchmarkFixture, items: int, settings: dict[str, Any] +) -> None: + """Overhead of sending the items of a callback through the item pipeline. + + The single and many scenarios, which use the default + :setting:`CONCURRENT_ITEMS` value, measure how that overhead grows with the + number of items that a response yields. The high-limit scenario instead + raises :setting:`CONCURRENT_ITEMS` well above that number. + """ + benchmark( + lambda: _crawl_tree(settings, domains=1, pages=ITEM_REQUESTS, items=items) + ) + + +def test_overhead_item_concurrency(benchmark: BenchmarkFixture) -> None: + """Overhead of a crawl where item processing waits. + + Every response yields more items than :setting:`CONCURRENT_ITEMS` allows in + parallel, so that the item pipeline gets them in several batches, and wall + time, unlike in most of the other benchmarks, is a function of the delay. + """ + settings = { + "CONCURRENT_ITEMS": DELAYED_CONCURRENT_ITEMS, + "ITEM_PIPELINES": {_DelayedPipeline: 100}, + } + + def run() -> None: + crawler = _crawl_tree( + settings, domains=1, pages=ITEM_REQUESTS, items=ITEMS_PER_RESPONSE + ) + assert crawler.stats + assert ( + crawler.stats.get_value("benchmark/peak_items") == DELAYED_CONCURRENT_ITEMS + ) + + benchmark(run)