Benchmark item processing and item concurrency (#7954)

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)