Fix DownloaderAwarePriorityQueue tie-breaking across slots (#7351)

This commit is contained in:
Adrian 2026-04-08 16:10:21 +02:00 committed by GitHub
parent 2b174e348d
commit b68f26726a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 79 additions and 2 deletions

View File

@ -350,10 +350,38 @@ class DownloaderAwarePriorityQueue:
self.crawler: Crawler = crawler self.crawler: Crawler = crawler
self.pqueues: dict[str, ScrapyPriorityQueue] = {} # slot -> priority queue self.pqueues: dict[str, ScrapyPriorityQueue] = {} # slot -> priority queue
self._last_selected_slot: str | None = None
if slot_startprios: if slot_startprios:
for slot, startprios in slot_startprios.items(): for slot, startprios in slot_startprios.items():
self.pqueues[slot] = self.pqfactory(slot, startprios) 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( def pqfactory(
self, slot: str, startprios: Iterable[int] = () self, slot: str, startprios: Iterable[int] = ()
) -> ScrapyPriorityQueue: ) -> ScrapyPriorityQueue:
@ -371,7 +399,7 @@ class DownloaderAwarePriorityQueue:
if not stats: if not stats:
return None return None
slot = min(stats)[1] slot = self._next_slot(stats, update_state=True)
queue = self.pqueues[slot] queue = self.pqueues[slot]
request = queue.pop() request = queue.pop()
if len(queue) == 0: if len(queue) == 0:
@ -395,7 +423,7 @@ class DownloaderAwarePriorityQueue:
stats = self._downloader_interface.stats(self.pqueues) stats = self._downloader_interface.stats(self.pqueues)
if not stats: if not stats:
return None return None
slot = min(stats)[1] slot = self._next_slot(stats, update_state=False)
queue = self.pqueues[slot] queue = self.pqueues[slot]
return queue.peek() return queue.peek()

View File

@ -4,6 +4,7 @@ from unittest.mock import Mock
import pytest import pytest
import queuelib import queuelib
from scrapy.core.downloader import Downloader
from scrapy.http.request import Request from scrapy.http.request import Request
from scrapy.pqueues import DownloaderAwarePriorityQueue, ScrapyPriorityQueue from scrapy.pqueues import DownloaderAwarePriorityQueue, ScrapyPriorityQueue
from scrapy.spiders import Spider from scrapy.spiders import Spider
@ -158,6 +159,54 @@ class TestDownloaderAwarePriorityQueue:
assert self.queue.pop().url == req3.url assert self.queue.pop().url == req3.url
assert self.queue.peek() is None 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( @pytest.mark.parametrize(
("input_", "output"), ("input_", "output"),