From b68f26726ac87c5950a4258a8e29bb7ec2e0ebc1 Mon Sep 17 00:00:00 2001 From: Adrian Date: Wed, 8 Apr 2026 16:10:21 +0200 Subject: [PATCH] Fix DownloaderAwarePriorityQueue tie-breaking across slots (#7351) --- scrapy/pqueues.py | 32 ++++++++++++++++++++++++++-- tests/test_pqueues.py | 49 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 79 insertions(+), 2 deletions(-) diff --git a/scrapy/pqueues.py b/scrapy/pqueues.py index ad0b36f6b..0ad0b5d78 100644 --- a/scrapy/pqueues.py +++ b/scrapy/pqueues.py @@ -350,10 +350,38 @@ class DownloaderAwarePriorityQueue: self.crawler: Crawler = crawler self.pqueues: dict[str, ScrapyPriorityQueue] = {} # slot -> priority queue + self._last_selected_slot: str | None = None if slot_startprios: for slot, startprios in slot_startprios.items(): self.pqueues[slot] = self.pqfactory(slot, startprios) + def _next_slot(self, stats: list[tuple[int, str]], *, update_state: bool) -> str: + last = self._last_selected_slot + min_active: int | None = None + best_slot: str | None = None + best_slot_after_last: str | None = None + for active, slot in stats: + if min_active is None or active < min_active: + min_active = active + best_slot = slot + best_slot_after_last = None + if last is not None and slot > last: + best_slot_after_last = slot + elif active == min_active: + if best_slot is None or slot < best_slot: + best_slot = slot + if ( + last is not None + and slot > last + and (best_slot_after_last is None or slot < best_slot_after_last) + ): + best_slot_after_last = slot + assert best_slot is not None + slot = best_slot_after_last if best_slot_after_last is not None else best_slot + if update_state: + self._last_selected_slot = slot + return slot + def pqfactory( self, slot: str, startprios: Iterable[int] = () ) -> ScrapyPriorityQueue: @@ -371,7 +399,7 @@ class DownloaderAwarePriorityQueue: if not stats: return None - slot = min(stats)[1] + slot = self._next_slot(stats, update_state=True) queue = self.pqueues[slot] request = queue.pop() if len(queue) == 0: @@ -395,7 +423,7 @@ class DownloaderAwarePriorityQueue: stats = self._downloader_interface.stats(self.pqueues) if not stats: return None - slot = min(stats)[1] + slot = self._next_slot(stats, update_state=False) queue = self.pqueues[slot] return queue.peek() diff --git a/tests/test_pqueues.py b/tests/test_pqueues.py index 350b3e10d..7be9241b9 100644 --- a/tests/test_pqueues.py +++ b/tests/test_pqueues.py @@ -4,6 +4,7 @@ from unittest.mock import Mock import pytest import queuelib +from scrapy.core.downloader import Downloader from scrapy.http.request import Request from scrapy.pqueues import DownloaderAwarePriorityQueue, ScrapyPriorityQueue from scrapy.spiders import Spider @@ -158,6 +159,54 @@ class TestDownloaderAwarePriorityQueue: assert self.queue.pop().url == req3.url assert self.queue.peek() is None + def test_tie_breaking_rotates_slots(self): + # No active downloads are tracked in the downloader, so every slot has + # the same score and tie-breaking must not starve a slot. + req_a1 = Request("https://example.org/a1") + req_a1.meta[Downloader.DOWNLOAD_SLOT] = "slot-a" + req_b1 = Request("https://example.org/b1") + req_b1.meta[Downloader.DOWNLOAD_SLOT] = "slot-b" + req_a2 = Request("https://example.org/a2") + req_a2.meta[Downloader.DOWNLOAD_SLOT] = "slot-a" + req_b2 = Request("https://example.org/b2") + req_b2.meta[Downloader.DOWNLOAD_SLOT] = "slot-b" + + for request in (req_a1, req_b1, req_a2, req_b2): + self.queue.push(request) + + slots = [ + self.queue.pop().meta[Downloader.DOWNLOAD_SLOT], + self.queue.pop().meta[Downloader.DOWNLOAD_SLOT], + self.queue.pop().meta[Downloader.DOWNLOAD_SLOT], + self.queue.pop().meta[Downloader.DOWNLOAD_SLOT], + ] + + assert slots == ["slot-a", "slot-b", "slot-a", "slot-b"] + + def test_tie_breaking_keeps_rotation_after_selected_slot_is_deleted(self): + # If the selected slot becomes empty, rotation should continue from + # that slot marker to avoid restarting from the smallest slot. + req_a1 = Request("https://example.org/a1") + req_a1.meta[Downloader.DOWNLOAD_SLOT] = "slot-a" + req_a2 = Request("https://example.org/a2") + req_a2.meta[Downloader.DOWNLOAD_SLOT] = "slot-a" + req_b1 = Request("https://example.org/b1") + req_b1.meta[Downloader.DOWNLOAD_SLOT] = "slot-b" + req_c1 = Request("https://example.org/c1") + req_c1.meta[Downloader.DOWNLOAD_SLOT] = "slot-c" + + for request in (req_a1, req_a2, req_b1, req_c1): + self.queue.push(request) + + slots = [ + self.queue.pop().meta[Downloader.DOWNLOAD_SLOT], + self.queue.pop().meta[Downloader.DOWNLOAD_SLOT], + self.queue.pop().meta[Downloader.DOWNLOAD_SLOT], + self.queue.pop().meta[Downloader.DOWNLOAD_SLOT], + ] + + assert slots == ["slot-a", "slot-b", "slot-c", "slot-a"] + @pytest.mark.parametrize( ("input_", "output"),