diff --git a/docs/guides/scaling_crawlers.mdx b/docs/guides/scaling_crawlers.mdx index 152d852e60..c6ea5b3977 100644 --- a/docs/guides/scaling_crawlers.mdx +++ b/docs/guides/scaling_crawlers.mdx @@ -47,3 +47,16 @@ The `desired_concurrency` option in the ## Autoscaled pool The `AutoscaledPool` manages a pool of asynchronous, resource-intensive tasks that run in parallel. It automatically starts new tasks only when there is enough free CPU and memory. To monitor system resources, it leverages the `Snapshotter` and `SystemStatus` classes. If any task raises an exception, the error is propagated, and the pool is stopped. Every crawler uses an `AutoscaledPool` under the hood. + +## Throughput autoscaled pool + +The default `AutoscaledPool` adds tasks until CPU or memory runs out. When the target website is the bottleneck, more tasks stop paying off. The site answers more slowly, and the crawler finishes fewer pages while loading the site harder. The experimental `ThroughputAutoscaledPool`, passed to a crawler as `autoscaled_pool_class`, steers by throughput instead. By Little's law, `N` tasks that each take `T` seconds finish `N / T` tasks per second, so more concurrency helps only while `T` grows slower than `N`. The pool reads throughput from windows of finished tasks and doubles the concurrency while that pays off. It then compares a level above and a level below the current concurrency, moves toward the better one, and holds once the two tie at the peak. It doesn't stay there for good. After each hold it measures the levels around the peak again, so even on a steady site the concurrency keeps moving within about 15% of it. The CPU and memory checks of the default pool still apply. + +The pool works best on long-running crawls. It learns the peak from finished tasks, so the time it spends measuring pays off only over a long run. + +Known limitations of the throughput pool: + +- Each step waits for its tasks to finish, so on slow pages the throughput pool ramps up more slowly than the default pool. +- Every finished task counts as work, so a site that answers fast with errors such as HTTP 429 looks like it can take more. +- The pool looks for a single concurrency for the whole crawl. Across many hosts, each with its own peak, such a point rarely exists, so the pool stays unstable and keeps moving instead of holding. +- Stay with the default pool when the machine is the bottleneck. When you need a fixed rate, use `max_tasks_per_minute` with the default pool, since the throughput pool only settles at whatever rate the limit lets through. diff --git a/src/crawlee/__init__.py b/src/crawlee/__init__.py index 508835e008..c1508c7af1 100644 --- a/src/crawlee/__init__.py +++ b/src/crawlee/__init__.py @@ -5,9 +5,14 @@ from ._types import ConcurrencySettings, EnqueueStrategy, HttpHeaders, RequestTransformAction, SkippedReason from ._utils.globs import Glob +# isort: split +# The snapshotter imports `service_locator` from `crawlee`, so the autoscaling package has to come after it. +from ._autoscaling import AutoscaledPool, ThroughputAutoscaledPool + __version__ = metadata.version('crawlee') __all__ = [ + 'AutoscaledPool', 'ConcurrencySettings', 'EnqueueStrategy', 'Glob', @@ -17,5 +22,6 @@ 'RequestState', 'RequestTransformAction', 'SkippedReason', + 'ThroughputAutoscaledPool', 'service_locator', ] diff --git a/src/crawlee/_autoscaling/__init__.py b/src/crawlee/_autoscaling/__init__.py index 5083a8017c..e6fd608ef0 100644 --- a/src/crawlee/_autoscaling/__init__.py +++ b/src/crawlee/_autoscaling/__init__.py @@ -1,5 +1,6 @@ from .autoscaled_pool import AutoscaledPool from .snapshotter import Snapshotter from .system_status import SystemStatus +from .throughput_autoscaled_pool import ThroughputAutoscaledPool -__all__ = ['AutoscaledPool', 'Snapshotter', 'SystemStatus'] +__all__ = ['AutoscaledPool', 'Snapshotter', 'SystemStatus', 'ThroughputAutoscaledPool'] diff --git a/src/crawlee/_autoscaling/throughput_autoscaled_pool.py b/src/crawlee/_autoscaling/throughput_autoscaled_pool.py new file mode 100644 index 0000000000..0723856cdd --- /dev/null +++ b/src/crawlee/_autoscaling/throughput_autoscaled_pool.py @@ -0,0 +1,626 @@ +from __future__ import annotations + +import math +import statistics +import time +import warnings +from collections import deque +from logging import getLogger +from typing import TYPE_CHECKING, Literal + +from typing_extensions import override + +from crawlee._autoscaling.autoscaled_pool import AutoscaledPool +from crawlee._utils.docs import docs_group + +if TYPE_CHECKING: + from collections.abc import Awaitable, Callable + + from crawlee._autoscaling.system_status import SystemStatus + from crawlee._types import ConcurrencySettings + +logger = getLogger(__name__) + +_DURATION_WINDOW = 50 +"""Recent tasks the task duration is read from, as a median.""" + +_MAX_SETTLE_TICKS = 4 +"""Most scaling ticks the pool waits for the target to meet a new level before measuring it.""" + +_MAX_ENROL_SPANS = 4 +"""Most window spans a window keeps enrolling for while it has too few members.""" + +_CHANGE_WINDOWS = 2 +"""Consecutive held windows past `hold_change_margin` on one side that end a hold.""" + +_DECISIVE_DROP = 0.5 +"""Fraction a held window's rate falls by that ends a hold on its own.""" + +_MIN_DURATION = 1e-6 +"""Seconds a task is recorded as at least, for clocks too coarse to time a quick one.""" + +_Phase = Literal['centre', 'climb', 'high', 'low'] +_Stage = Literal['settle', 'enrol', 'wait'] + + +class _Cohort: + """The tasks that started while one window was enrolling, and what has become of them.""" + + def __init__(self) -> None: + self.size = 0 + self.concurrency_sum = 0 + self.running = dict[int, float]() + self.finished = list[float]() + self.finished_total = 0.0 + + def join(self, started_at: float, concurrency: int) -> int: + """Enrol a task that started at `started_at` among `concurrency` running tasks, and return its member number.""" + self.size += 1 + self.running[self.size] = started_at + self.concurrency_sum += concurrency + return self.size + + def leave(self, member: int, elapsed: float) -> None: + """Record that a member finished after `elapsed` seconds.""" + del self.running[member] + self.finished.append(elapsed) + self.finished_total += elapsed + + @property + def concurrency(self) -> float: + """Mean concurrency the members met when they started.""" + return self.concurrency_sum / self.size if self.size else 0.0 + + def median_ceiling(self) -> float | None: + """Get the finished members' duration at the median's rank, which the median can only fall below.""" + rank = self.size // 2 + 1 + return sorted(self.finished)[rank - 1] if len(self.finished) >= rank else None + + def median(self, now: float) -> float | None: + """Get the median duration once no member still running can change it, else `None`.""" + if (upper := self.median_ceiling()) is None: + return None + + if self.running and now - max(self.running.values()) < upper: + return None + + return upper if self.size % 2 else math.sqrt(sorted(self.finished)[self.size // 2 - 1] * upper) + + +@docs_group('Autoscaling') +class ThroughputAutoscaledPool(AutoscaledPool): + """An `AutoscaledPool` that settles near the concurrency at which its targets deliver the most. + + `AutoscaledPool` climbs until a machine resource runs out, so against a target that stops answering faster under + load it keeps adding concurrency, which costs throughput and strains the target. This pool measures how many tasks + per second two levels around a centre deliver and moves the centre toward the better one. It climbs by doublings + at the start and holds for a while once the two levels tie. Every machine-resource check of the parent still + applies. + + Each step waits for its tasks to finish, so on slow pages the pool reaches a high concurrency later than + `AutoscaledPool` does. Every completion counts as delivered work, so a target that answers faster by refusing + requests reads as one that can take more. + + Tune it by subclassing and overriding the class attributes below. + + Warning: + This is an experimental feature. The behavior and interface may change in future versions. + + ### Usage + + ```python + from crawlee import ThroughputAutoscaledPool + from crawlee.crawlers import ParselCrawler + + crawler = ParselCrawler(autoscaled_pool_class=ThroughputAutoscaledPool) + ``` + """ + + dither_min = 0.15 + """Fraction the two levels sit above and below the centre once it has converged.""" + + dither_max = 0.5 + """Largest fraction the two levels spread to while the centre keeps moving one way.""" + + margin = 0.10 + """Fraction one level's rate must exceed the other's by before the centre moves toward it.""" + + startup_gain = 2.0 + """Factor the concurrency grows by per step of the opening climb.""" + + climb_margin = 0.25 + """Fraction a climb step must deliver over the level before for the climb to go on.""" + + retry_latency_margin = 0.5 + """Fraction a climb step's median latency may rise by and still be measured again when it delivered no more.""" + + reentry_gain = 1.5 + """Factor the climb resumes by when the upper level keeps winning by the climb margin.""" + + finished_fraction = 0.8 + """Share of a window's members that must finish before its rate is read.""" + + min_members = 10 + """Members a window enrols before it closes.""" + + patience_dead_times = 5.0 + """Task durations a window may take to finish before it counts as a level the target cannot serve.""" + + hold_cycles_at_peak = 6 + """Windows the pool holds its centre for once the two levels tie at the narrowest spread. Zero never holds.""" + + hold_change_margin = 0.2 + """Fraction a held window's rate may drift from the first held window's before the hold ends.""" + + hold_cycles_at_ceiling = 3 + """Windows the pool holds at its concurrency limit after the limit won, before it looks below again.""" + + def __init__( + self, + *, + system_status: SystemStatus, + concurrency_settings: ConcurrencySettings | None = None, + run_task_function: Callable[[], Awaitable], + is_task_ready_function: Callable[[], Awaitable[bool]], + is_finished_function: Callable[[], Awaitable[bool]], + ) -> None: + """Initialize a new instance. + + Args: + system_status: Provides data about system utilization (load). + concurrency_settings: Settings of concurrency levels. + run_task_function: A function that performs an asynchronous resource-intensive task. + is_task_ready_function: A function that indicates whether `run_task_function` should be called. + is_finished_function: A function that is called only when there are no tasks to be processed. If it + resolves to `True` then the pool's run finishes. + """ + super().__init__( + system_status=system_status, + concurrency_settings=concurrency_settings, + run_task_function=run_task_function, + is_task_ready_function=is_task_ready_function, + is_finished_function=is_finished_function, + ) + warnings.warn( + 'The `ThroughputAutoscaledPool` is experimental and may change or be removed in future releases.', + category=UserWarning, + stacklevel=2, + ) + + self._durations = deque[float](maxlen=_DURATION_WINDOW) + self._enrolling: _Cohort | None = None + self._window: _Cohort | None = None + + self._centre = float(self._desired_concurrency) + self._dither = self.dither_max + self._phase: _Phase = 'centre' + self._stage: _Stage = 'settle' + self._stage_since = time.monotonic() + self._dead_time_at_move = 0.0 + self._reference_duration = 0.0 + self._soft_ceiling = self._max_concurrency + # The level before the last raise, kept until a window at the raised level has been read. + self._level_before_raise: int | None = None + self._filled = True + self._rates = dict[str, float]() + self._last_direction = 0 + + # The opening climb, and a reading from its last step kept for the first comparison after it. + self._startup = True + self._climb_gain = self.startup_gain + self._climb_retried = False + self._reference: float | None = None + self._reference_median: float | None = None + self._reused = False + + # A hold at the ceiling or at a peak, and what a peak hold compares its windows with. + self._hold_cycles = 0 + self._holding_peak = False + self._hold_reference: float | None = None + self._hold_streak = 0 + + self._set_level() + + @override + async def run(self) -> None: + """Start the pool, settling its level from now and dropping any window a previous run left open.""" + self._stop_enrolling() + self._window = None + self._stage, self._stage_since = 'settle', time.monotonic() + await super().run() + + @override + async def _worker_task(self) -> None: + started_at = time.monotonic() + cohort = self._enrolling + member = cohort.join(started_at, self.current_concurrency) if cohort is not None else None + + try: + await super()._worker_task() + finally: + elapsed = max(time.monotonic() - started_at, _MIN_DURATION) + if cohort is not None and member is not None: + cohort.leave(member, elapsed) + self._durations.append(elapsed) + + @property + def _tick(self) -> float: + return self._AUTOSCALE_INTERVAL.total_seconds() + + @property + def _dead_time(self) -> float | None: + return statistics.median(self._durations) if self._durations else None + + @property + def _window_span(self) -> float: + """Seconds a window enrols the tasks that start during it: one task duration, never less than a tick.""" + return max(self._tick, self._dead_time or 0.0) + + @property + def _patience_unit(self) -> float: + """Seconds patience is counted in: the median task duration of the last level read, else the recent one.""" + if math.isinf(self._dead_time_at_move) and self._dead_time is not None: + return max(self._dead_time, self._tick) + return self._dead_time_at_move + + def _open_window(self) -> None: + self._window = self._enrolling = _Cohort() + + def _stop_enrolling(self) -> None: + self._enrolling = None + + def _ceiling(self) -> int: + return max(self._min_concurrency, min(self._max_concurrency, self._soft_ceiling)) + + def _level(self, phase: _Phase) -> int: + ceiling = self._ceiling() + if self._hold_cycles > 0: + return max(self._min_concurrency, min(ceiling, round(self._centre))) if self._holding_peak else ceiling + + if phase == 'centre': + factor = 1.0 + elif phase == 'climb': + factor = self._climb_gain + elif phase == 'high': + factor = 1 + self._dither + else: + factor = 1 / (1 + self._dither) + + level = round(self._centre * factor) + # Near the floor both levels can round to the centre, and two windows at one level can only ever tie. + if phase == 'high': + level = max(level, round(self._centre) + 1) + elif phase == 'low': + level = min(level, round(self._centre) - 1) + + return max(self._min_concurrency, min(ceiling, level)) + + def _set_level(self) -> None: + level = self._level(self._phase) + if level > self._desired_concurrency: + self._level_before_raise = self._desired_concurrency + elif level < self._desired_concurrency: + self._level_before_raise = None + self._desired_concurrency = level + # Past the knee recent durations are the congested ones, so patience comes from the last level that finished. + fallback = self._dead_time if self._dead_time is not None else float('inf') + self._dead_time_at_move = max(self._reference_duration or fallback, self._tick) + + @staticmethod + def _rate(cohort: _Cohort) -> float: + """Get tasks per second by Little's law, over the members that have finished.""" + return cohort.concurrency * len(cohort.finished) / cohort.finished_total + + def _required(self, cohort: _Cohort) -> int: + return max(1, math.floor(self.finished_fraction * cohort.size)) + + def _finished(self, cohort: _Cohort) -> bool: + return len(cohort.finished) >= self._required(cohort) + + def _losing_bound(self, cohort: _Cohort, now: float) -> float | None: + """Get the best rate the window can still be read at, if even that already loses its comparison.""" + if self._hold_reference is not None: + other, margin = self._hold_reference, self.hold_change_margin + elif self._phase == 'climb': + other, margin = self._reference, self.climb_margin + elif self._phase == 'high': + other, margin = self._rates.get('low'), self.margin + else: + return None + + if other is None or not cohort.size: + return None + + # The members still needed for a reading, and any more that would raise the rate, finish now at their ages, + # youngest first. The rate is read over finished members only, so a slow tail left running cannot lower it. + required = self._required(cohort) + count, occupied = len(cohort.finished), cohort.finished_total + bound = 0.0 + for age in sorted(now - started for started in cohort.running.values()): + count += 1 + occupied += age + if count >= required: + bound = max(bound, cohort.concurrency * count / occupied if occupied > 0 else float('inf')) + + return bound if bound < other / (1 + margin) else None + + def _lift_soft_ceiling(self) -> None: + if self._soft_ceiling < self._max_concurrency: + lifted = math.ceil(self._soft_ceiling * (1 + self._SCALE_UP_STEP_RATIO)) + self._soft_ceiling = min(self._max_concurrency, lifted) + + @override + def _autoscale(self) -> None: + now = time.monotonic() + + # A raise the machine objects to at once failed as a step, before the slower historical signal can tell. + if self._level_before_raise is not None and not self._system_status.get_current_system_info().is_system_idle: + self._on_overload(now, cut_to=self._level_before_raise) + return + + if not self._system_status.get_historical_system_info().is_system_idle: + self._on_overload(now) + return + + self._lift_soft_ceiling() + + if self._stage == 'settle': + self._settle(now) + elif self._stage == 'enrol': + self._enrol(now) + else: + self._wait(now) + + def _on_overload(self, now: float, *, cut_to: int | None = None) -> None: + """Cut to `cut_to` or by the parent's step, cap the pool at the cut, and restart the cycle there.""" + self._stop_enrolling() + self._window = None + self._level_before_raise = None + if cut_to is not None: + self._desired_concurrency = cut_to + elif self._desired_concurrency > self._min_concurrency: + step = math.ceil(self._SCALE_DOWN_STEP_RATIO * self._desired_concurrency) + self._desired_concurrency = max(self._min_concurrency, self._desired_concurrency - step) + + reason = 'right after a raise, undoing it' if cut_to is not None else 'cutting by the parent step' + logger.debug(f'Machine overloaded {reason}; desired concurrency is now {self._desired_concurrency}') + + # The cut level becomes the lower level of the next cycle, so a burst is cut at the parent's pace. Near the + # floor the centre sits a task above it, since the lower level is always at least a task below the centre. + self._soft_ceiling = self._desired_concurrency + self._centre = max(self._desired_concurrency * (1 + self.dither_min), self._desired_concurrency + 1.0) + self._dither = self.dither_min + self._end_hold() + self._rates.clear() + self._reused = False + self._last_direction = 0 + self._startup = False + self._phase, self._stage, self._stage_since = 'low', 'settle', now + self._set_level() + + def _settle(self, now: float) -> None: + """Wait for the target to meet the level, then open a window on it.""" + # The climb only waits for the pool to fill the level: a window enrolled during the fill reads its mean. + filled = not self._startup or ( + self.current_concurrency >= self._DESIRED_CONCURRENCY_RATIO * self._desired_concurrency + ) + settle = 0.0 if self._startup and filled else min(self._dead_time_at_move, _MAX_SETTLE_TICKS * self._tick) + + if now - self._stage_since < settle: + return + + if not filled: + # The request loader hands out less than this level takes, so nothing above it can be measured either. + self._end_climb(0.0, measured=False) + self._stage_since = now + self._set_level() + return + + self._open_window() + self._stage, self._stage_since = 'enrol', now + + def _enrol(self, now: float) -> None: + """Close the window once it has spanned a task duration with enough members.""" + elapsed = now - self._stage_since + span = self._window_span + size = self._window.size if self._window is not None else 0 + + if elapsed >= _MAX_ENROL_SPANS * span and size == 0: + # Nothing started because the tasks of a higher level still hold every slot. That is not a reading. + self._open_window() + self._stage_since = now + return + + if elapsed >= span and (size >= self.min_members or elapsed >= _MAX_ENROL_SPANS * span): + self._stop_enrolling() + self._stage, self._stage_since = 'wait', now + + def _wait(self, now: float) -> None: + """Read the window once enough of it has finished, or once it has already lost, or once patience runs out.""" + window = self._window + if window is None: + return + + median = window.median(now) + if self._finished(window): + rate = self._rate(window) + self._reference_duration = median or self._reference_duration + elif (bound := self._losing_bound(window, now)) is not None: + rate = bound + elif now - self._stage_since > self.patience_dead_times * self._patience_unit: + rate = 0.0 + else: + return + + # The stage moves on first, so an error while reading cannot leave the pool waiting on no window. + self._window = None + self._stage, self._stage_since = 'settle', now + + # A level counts as filled only if it was filled while the window enrolled and still is now. + wanted = self._DESIRED_CONCURRENCY_RATIO * self._desired_concurrency + self._filled = window.concurrency >= wanted and self.current_concurrency >= wanted + + self._read(rate, median if median is not None else window.median_ceiling()) + # The level this window measured has run without the machine objecting, so it no longer needs undoing. + self._level_before_raise = None + self._set_level() + + def _end_climb(self, rate: float, *, measured: bool) -> None: + cause = 'a step stopped paying off' if measured else 'the level could not be filled' + logger.debug(f'Climb ended at desired concurrency {self._desired_concurrency}: {cause}') + self._startup = False + self._phase = 'low' + self._climb_gain = self.startup_gain + if measured: + # The level the climb lost at stands in for the upper level of the first cycle. + self._rates['high'] = rate + self._reused = True + + def _read(self, rate: float, median: float | None) -> None: + if self._startup: + self._read_climb(rate, median) + return + + if self._hold_cycles > 0: + self._read_hold(rate) + return + + self._rates[self._phase] = rate + if 'high' not in self._rates or 'low' not in self._rates: + self._phase = 'low' if self._phase == 'high' else 'high' + return + + self._compare(median) + + def _read_climb(self, rate: float, median: float | None) -> None: + # A first level the target cannot serve is no base to double from. + if not self._filled or (self._reference is None and rate <= 0): + self._end_climb(rate, measured=False) + return + + if self._reference is None or rate > self._reference * (1 + self.climb_margin): + self._reference = rate + self._reference_median = median + if self._phase == 'climb': + self._centre = float(self._desired_concurrency) + self._phase = 'climb' + self._climb_retried = False + return + + # A step that delivered as much once is as likely a small window reading low as the knee, unless its latency + # rose with it, which is the plateau past the knee. + inflated = ( + median is not None + and self._reference_median is not None + and median > self._reference_median * (1 + self.retry_latency_margin) + ) + if not self._climb_retried and rate >= self._reference and not inflated: + self._climb_retried = True + return + + self._end_climb(rate, measured=True) + + def _read_hold(self, rate: float) -> None: + self._rates.clear() + self._phase = 'low' + + if not self._holding_peak: + self._hold_cycles -= 1 + # The soft ceiling may lift during the hold, and the centre follows it so the next cycle starts from there. + self._centre = float(self._ceiling()) + return + + if self._hold_reference is None: + if rate <= 0: + # The target could not serve the held level at all, so there is nothing to hold. + self._end_hold() + return + self._hold_reference = rate + else: + reference, margin = self._hold_reference, self.hold_change_margin + # The same thresholds as in `_losing_bound`, so a window read early always counts as a drop. + side = 1 if rate > reference * (1 + margin) else -1 if rate < reference / (1 + margin) else 0 + self._hold_streak = side if side * self._hold_streak <= 0 else self._hold_streak + side + if rate < reference * (1 - _DECISIVE_DROP): + self._hold_streak = -_CHANGE_WINDOWS + if abs(self._hold_streak) >= _CHANGE_WINDOWS: + self._hold_cycles = 1 + + self._hold_cycles -= 1 + if self._hold_cycles == 0: + self._end_hold() + + def _end_hold(self) -> None: + if self._holding_peak or self._hold_cycles > 0: + logger.debug(f'Hold ended; measuring around concurrency {round(self._centre)} again') + self._hold_cycles = 0 + self._holding_peak = False + self._hold_reference = None + self._hold_streak = 0 + + def _compare(self, median: float | None) -> None: + high, low = self._rates['high'], self._rates['low'] + self._rates.clear() + + if high > low * (1 + self.margin) and self._filled: + direction = 1 + elif low > high * (1 + self.margin): + direction = -1 + else: + direction = 0 + + if self._reused and direction > 0: + # The climb's reading is a phase old and from higher up, so it may lose or tie but never win. + self._reused = False + self._rates['low'] = low + self._phase = 'high' + return + self._reused = False + + self._phase = 'low' + if direction == 0: + self._on_tie() + else: + self._on_win(direction, high, low, median) + self._last_direction = direction + + def _on_tie(self) -> None: + if self._dither <= self.dither_min and self.hold_cycles_at_peak > 0: + # A step-down before the hold would have no way back up, so the hold sits between the two tied levels. + logger.debug(f'Levels tied at the narrowest spread; holding concurrency {round(self._centre)}') + self._hold_cycles = self.hold_cycles_at_peak + self._holding_peak = True + return + + # On a flat peak, or with an upper level the tasks could not fill, the lower concurrency delivers as much. + self._centre = max(float(self._min_concurrency), self._centre / (1 + self._dither / 2)) + self._dither = max(self.dither_min, self._dither / 2) + + def _on_win(self, direction: int, high: float, low: float, median: float | None) -> None: + """Move the centre toward the winning level, or resume the climb from it.""" + # The upper level was read last, so it is still the one set. + clamped = direction > 0 and self._desired_concurrency >= self._ceiling() + self._centre = float(self._desired_concurrency if direction > 0 else self._level('low')) + + if direction > 0 and not clamped and high > low * (1 + self.climb_margin) and self._last_direction > 0: + # Twice running the upper level delivered what a doubling would, so the climb ended early on noise. + self._startup = True + self._climb_gain = self.reentry_gain + self._climb_retried = False + self._reference = high + self._reference_median = median + self._phase = 'climb' + logger.debug( + f'Upper level kept winning by a climb step; climbing again from concurrency {self._centre:.0f}' + ) + return + + if clamped: + logger.debug(f'Upper level won at the concurrency limit; holding at {self._desired_concurrency}') + self._dither = self.dither_min + self._hold_cycles = self.hold_cycles_at_ceiling + self._holding_peak = False + elif direction == self._last_direction: + self._dither = min(self.dither_max, self._dither * 2) + else: + self._dither = max(self.dither_min, self._dither / 2) diff --git a/src/crawlee/crawlers/_basic/_basic_crawler.py b/src/crawlee/crawlers/_basic/_basic_crawler.py index c59751b4e1..cb66521902 100644 --- a/src/crawlee/crawlers/_basic/_basic_crawler.py +++ b/src/crawlee/crawlers/_basic/_basic_crawler.py @@ -171,6 +171,9 @@ class _BasicCrawlerOptions(TypedDict): concurrency_settings: NotRequired[ConcurrencySettings] """Settings to fine-tune concurrency levels.""" + autoscaled_pool_class: NotRequired[type[AutoscaledPool]] + """The pool class deciding how concurrency moves within `concurrency_settings`.""" + request_handler_timeout: NotRequired[timedelta] """Maximum duration allowed for a single request handler to run.""" @@ -295,6 +298,7 @@ def __init__( additional_http_error_status_codes: Iterable[int] | None = None, ignore_http_error_status_codes: Iterable[int] | None = None, concurrency_settings: ConcurrencySettings | None = None, + autoscaled_pool_class: type[AutoscaledPool] = AutoscaledPool, request_handler_timeout: timedelta = timedelta(minutes=1), statistics: Statistics[TStatisticsState] | None = None, abort_on_error: bool = False, @@ -344,6 +348,8 @@ def __init__( ignore_http_error_status_codes: HTTP status codes that are typically considered errors but should be treated as successful responses. concurrency_settings: Settings to fine-tune concurrency levels. + autoscaled_pool_class: The pool class deciding how concurrency moves within `concurrency_settings`. Pass + `ThroughputAutoscaledPool` to settle near the throughput peak of the targets. request_handler_timeout: Maximum duration allowed for a single request handler to run. statistics: A custom `Statistics` instance, allowing the use of non-default configuration. abort_on_error: If True, the crawler stops immediately when any request handler error occurs. @@ -490,7 +496,7 @@ async def persist_state_factory() -> KeyValueStore: self._robots_txt_file_cache: LRUCache[str, RobotsTxtFile] = LRUCache(maxsize=1000) self._robots_txt_lock = asyncio.Lock() self._snapshotter = Snapshotter.from_config(config) - self._autoscaled_pool = AutoscaledPool( + self._autoscaled_pool = autoscaled_pool_class( system_status=SystemStatus(self._snapshotter), concurrency_settings=concurrency_settings, is_finished_function=self.__is_finished_function, diff --git a/tests/unit/_autoscaling/test_throughput_autoscaled_pool.py b/tests/unit/_autoscaling/test_throughput_autoscaled_pool.py new file mode 100644 index 0000000000..751df6b751 --- /dev/null +++ b/tests/unit/_autoscaling/test_throughput_autoscaled_pool.py @@ -0,0 +1,867 @@ +from __future__ import annotations + +import asyncio +from collections import deque +from contextlib import asynccontextmanager, contextmanager +from functools import partial +from itertools import pairwise +from typing import TYPE_CHECKING +from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch + +import pytest + +from crawlee._autoscaling import AutoscaledPool, ThroughputAutoscaledPool +from crawlee._autoscaling.throughput_autoscaled_pool import _MAX_SETTLE_TICKS, _Cohort +from crawlee._types import ConcurrencySettings + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Iterable, Iterator + +pytestmark = pytest.mark.filterwarnings('ignore:The `ThroughputAutoscaledPool` is experimental:UserWarning') + +_TICK = AutoscaledPool._AUTOSCALE_INTERVAL.total_seconds() + +_TO_HOLD = (10.0, 10.0, 5.0, 5.0, 5.0, 5.0, 5.0, 5.0) +"""Window rates that take a pool starting at 40 to a hold at 28: a climb that ends at once, then three ties. + +The climb ends at once because a doubling at the same rate doubles the median latency, past `retry_latency_margin`. +""" + + +class _Clock: + """A monotonic clock that only moves when a test moves it.""" + + def __init__(self) -> None: + self.now = 0.0 + + def monotonic(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + self.now += seconds + + +@pytest.fixture +def clock() -> Iterator[_Clock]: + """Replace the clock the pool module reads, leaving the event loop's own clock alone.""" + fake = _Clock() + with patch('crawlee._autoscaling.throughput_autoscaled_pool.time', new=fake): + yield fake + + +@contextmanager +def _live(count: int) -> Iterator[None]: + """Patch how many tasks the pool sees running.""" + with patch.object(ThroughputAutoscaledPool, 'current_concurrency', new_callable=PropertyMock, return_value=count): + yield + + +def _status() -> MagicMock: + """Build a system status that reports an idle machine until a test says otherwise.""" + status = MagicMock() + status.get_historical_system_info.return_value = MagicMock(is_system_idle=True) + status.get_current_system_info.return_value = MagicMock(is_system_idle=True) + return status + + +def _pool( + *, + cls: type[ThroughputAutoscaledPool] = ThroughputAutoscaledPool, + desired: int = 40, + min_concurrency: int = 1, + max_concurrency: int = 100, + status: MagicMock | None = None, +) -> ThroughputAutoscaledPool: + """Build a pool that is never run.""" + return cls( + system_status=status or _status(), + concurrency_settings=ConcurrencySettings( + min_concurrency=min_concurrency, desired_concurrency=desired, max_concurrency=max_concurrency + ), + run_task_function=MagicMock(), + is_task_ready_function=MagicMock(), + is_finished_function=MagicMock(), + ) + + +def _cohort(finished: Iterable[float] = (), *, running: Iterable[float] = (), concurrency: int = 10) -> _Cohort: + """Build a closed window with `finished` durations and members that have run for `running` seconds at time zero.""" + cohort = _Cohort() + for duration in finished: + cohort.leave(cohort.join(0.0, concurrency), duration) + for age in running: + cohort.join(-age, concurrency) + return cohort + + +class _Members: + """Worker tasks of the pool that run until the test finishes them.""" + + def __init__(self, gates: list[asyncio.Event], tasks: list[asyncio.Task]) -> None: + self._gates = gates + self._tasks = tasks + + async def finish(self, count: int) -> None: + """Finish the next `count` members that are still running, at the clock's current time.""" + chosen = [(gate, task) for gate, task in zip(self._gates, self._tasks, strict=True) if not gate.is_set()] + for gate, _ in chosen[:count]: + gate.set() + await asyncio.gather(*(task for _, task in chosen[:count])) + + +@asynccontextmanager +async def _members(pool: ThroughputAutoscaledPool, count: int) -> AsyncIterator[_Members]: + """Start `count` of the pool's own worker tasks, and finish whichever are left on exit.""" + gates = [asyncio.Event() for _ in range(count)] + waiting = iter(gates) + + async def run() -> None: + await next(waiting).wait() + + with patch.object(AutoscaledPool, '_worker_task', new=AsyncMock(side_effect=run)): + tasks = [asyncio.create_task(pool._worker_task()) for _ in range(count)] + await asyncio.sleep(0) + members = _Members(gates, tasks) + try: + yield members + finally: + await members.finish(count) + + +async def _window(pool: ThroughputAutoscaledPool, clock: _Clock, durations: list[float]) -> int: + """Run one window at the pool's level, one member per duration, and return the level the pool sets after it.""" + with _live(pool.desired_concurrency): + clock.advance(_MAX_SETTLE_TICKS * _TICK) + pool._autoscale() + started_at = clock.now + + async with _members(pool, len(durations)) as members: + for duration in sorted(durations): + clock.now = started_at + duration + await members.finish(1) + + clock.now = started_at + max(_TICK, *durations) + pool._autoscale() + pool._autoscale() + + assert pool._stage == 'settle', 'the window was not read' + return pool.desired_concurrency + + +async def _levels(pool: ThroughputAutoscaledPool, clock: _Clock, *rates: float) -> list[int]: + """Run one window per rate, its members taking the duration that delivers the rate, and return each next level.""" + levels = [] + for rate in rates: + duration = pool.desired_concurrency / rate + levels.append(await _window(pool, clock, [duration] * pool.min_members)) + return levels + + +async def _stalled_window(pool: ThroughputAutoscaledPool, clock: _Clock) -> int: + """Run one window whose members never finish, until patience reads it, and return the level set after it.""" + with _live(pool.desired_concurrency): + clock.advance(_MAX_SETTLE_TICKS * _TICK) + pool._autoscale() + + async with _members(pool, pool.min_members): + clock.advance(_TICK) + pool._autoscale() + clock.advance(10_000.0) + pool._autoscale() + + assert pool._stage == 'settle', 'the window was not read' + return pool.desired_concurrency + + +_BASE_DURATION = 6.0 +"""Seconds a task takes against a target serving no more than its capacity.""" + + +def _duration(concurrency: int, capacity: int) -> float: + """Seconds a task takes at `concurrency` against a target that slows quadratically past `capacity`.""" + overload = max(0, concurrency - capacity) / capacity + return _BASE_DURATION * (1 + overload**2) + + +async def _run( + pool: ThroughputAutoscaledPool, clock: _Clock, *, capacity: int, minutes: int +) -> list[tuple[int, float]]: + """Run the pool's own worker tasks in batches of its desired concurrency. Return each batch's level and duration. + + A batch of `level` tasks that each take one duration delivers `level` tasks per duration, which is Little's law + with no scatter, so the run is exact and takes milliseconds. + """ + batches = list[tuple[int, float]]() + last_tick = clock.now + + with ( + patch.object(AutoscaledPool, '_worker_task', new=AsyncMock()) as task, + patch.object(ThroughputAutoscaledPool, 'current_concurrency', new_callable=PropertyMock) as live, + ): + while clock.now < minutes * 60: + level = pool.desired_concurrency + duration = _duration(level, capacity) + live.return_value = level + task.side_effect = partial(clock.advance, duration) + + started_at = clock.now + for _ in range(level): + clock.now = started_at + await pool._worker_task() + batches.append((level, duration)) + + if clock.now - last_tick >= _TICK: + pool._autoscale() + last_tick = clock.now + + return batches + + +def _tail(batches: list[tuple[int, float]], minutes: int) -> list[tuple[int, float]]: + """Get the trailing batches of a run that together span `minutes`.""" + tail = list[tuple[int, float]]() + for level, duration in reversed(batches): + if sum(d for _, d in tail) >= minutes * 60: + break + tail.append((level, duration)) + return tail + + +def test_median_waits_for_running_members() -> None: + """The median is withheld while a running member could still be shorter than it.""" + cohort = _cohort([1.0, 2.0, 3.0], running=[0.0]) + + assert cohort.median(2.0) is None + assert cohort.median(5.0) == pytest.approx(6**0.5, rel=0.01) + + +def test_levels_sit_around_the_centre() -> None: + """The levels are the centre, a doubling, and the centre spread up and down.""" + pool = _pool(desired=40) + + assert (pool._level('centre'), pool._level('climb'), pool._level('high'), pool._level('low')) == (40, 80, 60, 27) + + +def test_levels_stay_inside_the_settings() -> None: + """No level leaves the configured bounds.""" + assert _pool(desired=90, max_concurrency=100)._level('high') == 100 + assert _pool(desired=6, min_concurrency=5)._level('low') == 5 + + +def test_levels_collapse_when_min_equals_max() -> None: + """With one allowed concurrency every level is that concurrency.""" + pool = _pool(desired=5, min_concurrency=5, max_concurrency=5) + + assert {pool._level(phase) for phase in ('centre', 'climb', 'high', 'low')} == {5} + + +@pytest.mark.parametrize( + ('desired', 'expected'), + [ + pytest.param(2, (3, 1), id='two'), + pytest.param(3, (4, 2), id='three'), + ], +) +def test_levels_step_at_least_one_task(desired: int, expected: tuple[int, int]) -> None: + """Near the floor the two levels still differ from the centre by a task.""" + + class Narrow(ThroughputAutoscaledPool): + dither_max = 0.15 + + pool = _pool(cls=Narrow, desired=desired) + + assert (pool._level('high'), pool._level('low')) == expected + + +async def test_paying_climb_doubles_the_concurrency(clock: _Clock) -> None: + """Each window that delivers a quarter more than the last doubles the desired concurrency.""" + pool = _pool(desired=10) + + assert await _levels(pool, clock, 10.0, 20.0) == [20, 40] + + +async def test_climb_that_stops_paying_drops_below_it(clock: _Clock) -> None: + """A doubling that delivers no more ends the climb at the lower level around 10, which is 10 / 1.5.""" + pool = _pool(desired=10) + + assert await _levels(pool, clock, 10.0, 10.0) == [20, 7] + + +@pytest.mark.parametrize( + ('durations', 'expected'), + [ + pytest.param([1.0] * 9 + [11.0], 40, id='equal rate, same median'), + pytest.param([2.0] * 10, 13, id='equal rate, slower median'), + ], +) +async def test_equal_climb_step_is_measured_again(clock: _Clock, durations: list[float], expected: int) -> None: + """A doubling that delivers about as much is measured again at 40, unless its median latency rose by half.""" + pool = _pool(desired=10) + await _levels(pool, clock, 10.0, 18.0) + + assert await _window(pool, clock, durations) == expected + + +async def test_losing_climb_step_ends_the_climb(clock: _Clock) -> None: + """A doubling that delivers less ends the climb at the lower level around 20, which is 20 / 1.5.""" + pool = _pool(desired=10) + + assert await _levels(pool, clock, 10.0, 18.0, 12.0) == [20, 40, 13] + + +async def test_climb_reading_cannot_win(clock: _Clock) -> None: + """The climb's last reading beating the lower level sends the pool to measure the upper level, 30, not to it.""" + pool = _pool(desired=10) + await _levels(pool, clock, 10.0, 18.0, 12.0) + + assert await _levels(pool, clock, 6.0, 5.0) == [30, 10] + + +async def test_climb_reading_can_tie(clock: _Clock) -> None: + """A tie with the climb's last reading steps the centre from 20 to 16, measured at 13 and then 20.""" + pool = _pool(desired=10) + await _levels(pool, clock, 10.0, 18.0, 12.0) + + assert await _levels(pool, clock, 12.5, 1.0) == [13, 20] + + +async def test_climb_moves_to_the_measured_level(clock: _Clock) -> None: + """A climb step doubles the level it measured, even when the ceiling has lifted since it was set.""" + pool = _pool(desired=10) + with patch.object(pool, '_soft_ceiling', 15): + measured = await _window(pool, clock, [1.0] * 10) + + assert await _levels(pool, clock, 18.0) == [2 * measured] + + +async def test_unserved_first_level_ends_the_climb(clock: _Clock) -> None: + """A first window that runs out of patience ends the climb at the lower level around 10 instead of doubling.""" + pool = _pool(desired=10) + + with _live(10): + clock.advance(_MAX_SETTLE_TICKS * _TICK) + pool._autoscale() + + async with _members(pool, pool.min_members) as members: + clock.advance(1.0) + await members.finish(7) + clock.advance(_TICK) + pool._autoscale() + clock.advance(100.0) + pool._autoscale() + + assert pool.desired_concurrency == 7 + + +async def test_unfilled_level_ends_the_climb(clock: _Clock) -> None: + """A level the pool cannot fill ends the climb without reading a window, at the lower level around 20.""" + pool = _pool(desired=20) + + with _live(12), patch.object(pool, '_read') as read: + clock.advance(_MAX_SETTLE_TICKS * _TICK) + pool._autoscale() + + assert pool.desired_concurrency == 13 + read.assert_not_called() + + +async def test_upper_level_wins(clock: _Clock) -> None: + """The upper level 60 winning becomes the centre with the spread halved, so the next lower level is 60 / 1.25.""" + pool = _pool(desired=40) + await _levels(pool, clock, 10.0, 10.0, 5.0) + + assert await _levels(pool, clock, 130.0) == [48] + + +async def test_turn_halves_the_spread(clock: _Clock) -> None: + """A win from the other side than the last makes 48 the centre with the narrowest spread, so next is 48 / 1.15.""" + pool = _pool(desired=40) + await _levels(pool, clock, 10.0, 10.0, 5.0, 130.0) + + assert await _levels(pool, clock, 130.0, 100.0) == [75, 42] + + +async def test_same_side_doubles_the_spread(clock: _Clock) -> None: + """A second win from the same side makes 75 the centre with the spread doubled, so next is 75 / 1.5.""" + pool = _pool(desired=40) + await _levels(pool, clock, 10.0, 10.0, 5.0, 130.0) + + assert await _levels(pool, clock, 100.0, 120.0) == [75, 50] + + +async def test_two_large_wins_resume_the_climb(clock: _Clock) -> None: + """A second upper win as large as a paying climb step resumes the climb from 75 by half again.""" + pool = _pool(desired=40, max_concurrency=200) + await _levels(pool, clock, 10.0, 10.0, 5.0, 130.0) + + assert await _levels(pool, clock, 100.0, 200.0) == [75, 112] + + +async def test_tie_steps_the_centre_down(clock: _Clock) -> None: + """A tie above the narrowest spread steps the centre from 40 to 32, measured at 26 and then 40.""" + pool = _pool(desired=40) + await _levels(pool, clock, 10.0, 10.0, 5.0) + + assert await _levels(pool, clock, 5.0, 5.0) == [26, 40] + + +async def test_ties_stop_at_the_minimum(clock: _Clock) -> None: + """Ties near `min_concurrency` step the centre down to it and no further, and the level above is still measured.""" + pool = _pool(desired=6, min_concurrency=5) + + assert await _levels(pool, clock, *[10.0] * 7) == [12, 5, 5, 6, 5, 6, 5] + + +async def test_tie_at_narrowest_spread_holds(clock: _Clock) -> None: + """A tie at the narrowest spread holds the level for six windows, then measures below it again.""" + pool = _pool(desired=40) + + levels = await _levels(pool, clock, *_TO_HOLD, *[5.0] * pool.hold_cycles_at_peak) + + assert levels[len(_TO_HOLD) - 1 :] == [28] * pool.hold_cycles_at_peak + [25] + + +async def test_zero_hold_cycles_never_hold(clock: _Clock) -> None: + """A subclass can switch the hold off, and ties then keep moving the level.""" + + class NeverHolds(ThroughputAutoscaledPool): + hold_cycles_at_peak = 0 + + pool = _pool(cls=NeverHolds, desired=40) + + levels = await _levels(pool, clock, *_TO_HOLD, 5.0, 5.0, 5.0) + + assert all(level != following for level, following in pairwise(levels)) + + +@pytest.mark.parametrize( + ('rates', 'expected'), + [ + pytest.param([3.75], 28, id='one window down by a quarter'), + pytest.param([3.75, 3.75], 25, id='two windows down by a quarter'), + pytest.param([4.3, 4.3], 28, id='two windows inside the margin'), + pytest.param([4.1, 4.1], 25, id='two windows just past the margin'), + pytest.param([2.0], 25, id='one halved window'), + ], +) +async def test_drift_ends_the_hold(clock: _Clock, rates: list[float], expected: int) -> None: + """Two held windows past the margin on one side of the first, or one at half of it, end the hold at 28.""" + pool = _pool(desired=40) + await _levels(pool, clock, *_TO_HOLD, 5.0) + + assert (await _levels(pool, clock, *rates))[-1] == expected + + +async def test_unserved_first_held_window_ends_the_hold(clock: _Clock) -> None: + """A first held window that delivers nothing before patience runs out ends the hold at once.""" + pool = _pool(desired=40) + await _levels(pool, clock, *_TO_HOLD) + + assert await _stalled_window(pool, clock) == 25 + + +async def test_held_window_can_lose_early(clock: _Clock) -> None: + """A held window is given up once its bound falls below the first held window's rate by the margin.""" + pool = _pool(desired=40) + await _levels(pool, clock, *_TO_HOLD, 5.0) + + assert pool._losing_bound(_cohort([1.0, 1.0], running=[30.0, 30.0], concurrency=40), now=0.0) is not None + assert pool._losing_bound(_cohort([0.4, 0.4], running=[0.4, 0.4], concurrency=40), now=0.0) is None + + +async def test_win_at_the_ceiling_holds_there(clock: _Clock) -> None: + """An upper win at the concurrency limit holds the pool there for three windows, then looks below at 100 / 1.15.""" + pool = _pool(desired=90, max_concurrency=100) + await _levels(pool, clock, 10.0, 11.0, 11.0, 5.0) + + assert await _levels(pool, clock, 20.0, 20.0, 20.0, 20.0) == [100, 100, 100, 87] + + +@pytest.mark.parametrize( + ('cohort', 'finished'), + [ + pytest.param(_cohort([1.0] * 8, running=[0.5, 0.5]), True, id='four fifths'), + pytest.param(_cohort([1.0] * 7, running=[0.5] * 3), False, id='less'), + pytest.param(_cohort(), False, id='empty'), + ], +) +def test_window_finishes_at_four_fifths(cohort: _Cohort, finished: bool) -> None: # noqa: FBT001 + """A window is read once four fifths of its members have finished, and never while empty.""" + assert _pool()._finished(cohort) is finished + + +def test_rate_is_littles_law() -> None: + """The rate is concurrency times finished members over the time they occupied.""" + assert ThroughputAutoscaledPool._rate(_cohort([2.0] * 4, concurrency=8)) == pytest.approx(4.0) + + +async def test_upper_level_can_lose_early(clock: _Clock) -> None: + """An upper level is given up once its bound falls below the lower level's rate by the margin.""" + pool = _pool(desired=40) + await _levels(pool, clock, 10.0, 10.0, 5.0) + + assert pool._losing_bound(_cohort([1.0, 1.0], running=[30.0, 30.0], concurrency=40), now=0.0) is not None + assert pool._losing_bound(_cohort([1.0, 1.0], running=[1.0, 1.0], concurrency=40), now=0.0) is None + + +async def test_slow_tail_does_not_lose_early(clock: _Clock) -> None: + """A slow tail left running does not lose an upper level whose reading, taken without it, would win.""" + pool = _pool(desired=40) + await _levels(pool, clock, 10.0, 10.0, 5.0) + + # Eight of ten members make a reading: one more finishing at 30 seconds reads 40 * 8 / 37, above the lower level. + assert pool._losing_bound(_cohort([1.0] * 7, running=[30.0] * 3, concurrency=40), now=0.0) is None + + +async def test_young_members_past_the_reading_raise_the_bound(clock: _Clock) -> None: + """Running members past the ones a reading needs count toward the bound when they would raise the rate.""" + pool = _pool(desired=40) + await _levels(pool, clock, 10.0, 10.0, 5.0) + + # Eight members read 40 * 8 / 71, below the lower level, but all ten read 40 * 10 / 73, above it. + assert pool._losing_bound(_cohort([10.0] * 7, running=[1.0] * 3, concurrency=40), now=0.0) is None + + +async def test_climb_step_can_lose_early(clock: _Clock) -> None: + """A climb step is given up once its bound falls below the last step's rate by the climb margin.""" + pool = _pool(desired=10) + await _levels(pool, clock, 10.0) + + assert pool._losing_bound(_cohort([1.0, 1.0], running=[30.0, 30.0], concurrency=20), now=0.0) is not None + assert pool._losing_bound(_cohort([0.5, 0.5], running=[0.5, 0.5], concurrency=20), now=0.0) is None + + +def test_finished_window_is_read() -> None: + """A mostly finished window is read as its rate and sends the pool back to settling.""" + pool = _pool(desired=40) + window = _cohort([1.0] * 8, running=[0.5, 0.5], concurrency=40) + + with patch.object(pool, '_window', window), patch.object(pool, '_read') as read: + pool._wait(50.0) + + assert read.call_args.args[0] == pytest.approx(40.0) + assert pool._stage_since == 50.0 + + +async def test_losing_window_is_read_early(clock: _Clock) -> None: + """A window whose bound already loses is read before it finishes, at that bound.""" + pool = _pool(desired=40) + await _levels(pool, clock, 10.0, 10.0, 5.0) + window = _cohort([1.0, 1.0], running=[30.0, 30.0], concurrency=40) + + # The running members' ages are counted from time zero. + with patch.object(pool, '_window', window), patch.object(pool, '_read') as read: + pool._wait(0.0) + + # Three of the four members make a reading, so the best case is one running member finishing at its age. + assert read.call_args.args[0] == pytest.approx(40 * 3 / 32) + + +def test_window_out_of_patience_reads_zero() -> None: + """A window that outlasts its patience reads as a level the target cannot serve.""" + pool = _pool(desired=40) + window = _cohort([1.0], running=[0.0] * 9, concurrency=40) + + with ( + patch.object(ThroughputAutoscaledPool, '_patience_unit', new_callable=PropertyMock, return_value=10.0), + patch.object(pool, '_window', window), + patch.object(pool, '_stage_since', 0.0), + patch.object(pool, '_read') as read, + ): + pool._wait(51.0) + + assert read.call_args.args[0] == 0.0 + + +@pytest.mark.parametrize( + ('window_concurrency', 'live', 'filled'), + [ + pytest.param(40, 40, True, id='filled'), + pytest.param(20, 40, False, id='unfilled while enrolling'), + pytest.param(40, 20, False, id='unfilled when read'), + ], +) +def test_filled_needs_both_readings(window_concurrency: int, live: int, filled: bool) -> None: # noqa: FBT001 + """A level counts as filled only if it was filled while enrolling and still is when read.""" + pool = _pool(desired=40) + window = _cohort([1.0] * 10, concurrency=window_concurrency) + + with _live(live), patch.object(pool, '_window', window), patch.object(pool, '_read'): + pool._wait(0.0) + + assert pool._filled is filled + + +async def test_reading_error_does_not_stall(clock: _Clock) -> None: + """An error while reading a window still leaves the pool able to measure the next one.""" + pool = _pool(desired=40) + window = _cohort([1.0] * 10, concurrency=40) + + with ( + patch.object(pool, '_window', window), + patch.object(pool, '_read', side_effect=RuntimeError), + pytest.raises(RuntimeError), + ): + pool._wait(clock.now) + + assert await _levels(pool, clock, 10.0) == [80] + + +@pytest.mark.parametrize( + ('durations', 'expected'), + [ + pytest.param([], float('inf'), id='nothing finished'), + pytest.param([30.0], 30.0, id='recent duration'), + ], +) +def test_first_window_patience(durations: list[float], expected: float) -> None: + """Before any level is read, patience follows the recent task duration once there is one.""" + pool = _pool(desired=40) + + with patch.object(pool, '_durations', deque(durations)): + assert pool._patience_unit == expected + + +async def test_overload_cuts_and_restarts(clock: _Clock) -> None: + """An overloaded machine cuts by the parent's step and ends the climb, so a paying window no longer doubles.""" + status = _status() + status.get_historical_system_info.return_value = MagicMock(is_system_idle=False) + pool = _pool(desired=80, status=status) + + pool._autoscale() + assert pool.desired_concurrency == 76 + + status.get_historical_system_info.return_value = MagicMock(is_system_idle=True) + [level] = await _levels(pool, clock, 100.0) + assert 76 <= level < 152 + + +async def test_overload_ends_a_peak_hold(clock: _Clock) -> None: + """An overload during a hold at the peak ends the hold, so the pool measures the levels around the cut again.""" + status = _status() + pool = _pool(desired=40, status=status) + await _levels(pool, clock, *_TO_HOLD) + + status.get_historical_system_info.return_value = MagicMock(is_system_idle=False) + pool._autoscale() + assert pool.desired_concurrency == 26 + + status.get_historical_system_info.return_value = MagicMock(is_system_idle=True) + assert await _levels(pool, clock, 5.0, 5.0) == [32, 30] + + +async def test_overload_forgets_the_last_win(clock: _Clock) -> None: + """An upper win before an overload does not count toward resuming the climb after it, so 54 winning moves to 47.""" + status = _status() + pool = _pool(desired=40, status=status) + await _levels(pool, clock, 10.0, 10.0, 5.0, 130.0) + + status.get_historical_system_info.return_value = MagicMock(is_system_idle=False) + pool._autoscale() + + status.get_historical_system_info.return_value = MagicMock(is_system_idle=True) + assert await _levels(pool, clock, 5.0, 100.0) == [54, 47] + + +@pytest.mark.parametrize( + ('desired', 'expected'), + [ + pytest.param(1, 1, id='floor'), + pytest.param(3, 2, id='three'), + pytest.param(4, 3, id='four'), + ], +) +def test_overload_near_the_floor(desired: int, expected: int) -> None: + """An overload near the minimum concurrency cuts by one parent step, and at the minimum leaves the pool there.""" + status = _status() + status.get_historical_system_info.return_value = MagicMock(is_system_idle=False) + pool = _pool(desired=desired, status=status) + + pool._autoscale() + + assert pool.desired_concurrency == expected + + +async def test_overload_after_a_raise_undoes_it(clock: _Clock) -> None: + """An overload before the raised level was read returns the pool to the level before the raise.""" + status = _status() + pool = _pool(desired=40, status=status) + assert await _levels(pool, clock, 10.0) == [80] + + status.get_current_system_info.return_value = MagicMock(is_system_idle=False) + pool._autoscale() + + assert pool.desired_concurrency == 40 + + +async def test_read_window_confirms_a_raise(clock: _Clock) -> None: + """Once a window at the raised level has been read, an overload no longer undoes the raise.""" + status = _status() + pool = _pool(desired=90, max_concurrency=100, status=status) + assert await _levels(pool, clock, 10.0, 11.0, 11.0, 5.0, 20.0) == [100, 100, 60, 100, 100] + + status.get_current_system_info.return_value = MagicMock(is_system_idle=False) + pool._autoscale() + + assert pool.desired_concurrency == 100 + + +def test_overload_now_without_a_raise_keeps_the_level() -> None: + """A current overload with no raise to undo is left to the historical signal.""" + status = _status() + status.get_current_system_info.return_value = MagicMock(is_system_idle=False) + pool = _pool(desired=40, status=status) + + pool._autoscale() + + assert pool.desired_concurrency == 40 + + +def test_soft_ceiling_lifts_when_idle() -> None: + """The cap left by an overload lifts by the parent's step each idle tick.""" + status = _status() + status.get_historical_system_info.side_effect = [MagicMock(is_system_idle=value) for value in (False, True)] + pool = _pool(desired=80, status=status) + + pool._autoscale() + pool._autoscale() + + assert pool._soft_ceiling == 80 + + +def test_empty_window_reopens(clock: _Clock) -> None: + """A window nobody joined is opened again rather than read.""" + pool = _pool(desired=40) + + with _live(40): + clock.now = 100.0 + pool._autoscale() + first = pool._window + + clock.now = 141.0 + pool._autoscale() + + assert pool._stage == 'enrol' + assert pool._window is not first + assert pool._window is not None + assert pool._window.size == 0 + + +async def test_window_closes_after_four_spans(clock: _Clock) -> None: + """A window with too few members still closes after four spans.""" + pool = _pool(desired=40) + + with _live(40): + clock.now = 100.0 + pool._autoscale() + + async with _members(pool, 3): + clock.now = 141.0 + pool._autoscale() + + assert pool._stage == 'wait' + + +async def test_run_counts_settle_from_the_start(clock: _Clock) -> None: + """Time between building the pool and running it does not count toward settling.""" + pool = _pool(desired=40) + clock.now = 500.0 + + with patch.object(AutoscaledPool, 'run', new=AsyncMock()): + await pool.run() + + assert pool._stage_since == 500.0 + + +async def test_run_drops_a_window_left_open(clock: _Clock) -> None: + """A window a previous run left enrolling is not read in the next run.""" + pool = _pool(desired=40) + + with _live(40): + clock.now = 100.0 + pool._autoscale() + + assert pool._stage == 'enrol' + + with patch.object(AutoscaledPool, 'run', new=AsyncMock()): + await pool.run() + + assert (pool._stage, pool._window, pool._enrolling) == ('settle', None, None) + + +async def test_worker_records_its_duration(clock: _Clock) -> None: + """A finished task joins its window and the recent durations.""" + pool = _pool(desired=40) + + with _live(40): + clock.now = 100.0 + pool._autoscale() + + window = pool._window + + async with _members(pool, 1) as members: + clock.advance(3.0) + await members.finish(1) + + assert window is not None + assert (window.size, len(window.finished), window.finished_total) == (1, 1, 3.0) + assert list(pool._durations) == [3.0] + + +async def test_quick_task_has_a_rate(clock: _Clock) -> None: + """A task too quick for the clock to time still gives its window a rate.""" + pool = _pool(desired=40) + + with _live(40): + clock.now = 100.0 + pool._autoscale() + + window = pool._window + + with _live(40): + async with _members(pool, 1): + pass + + assert window is not None + assert ThroughputAutoscaledPool._rate(window) > 0 + + +async def test_settles_near_the_peak(clock: _Clock) -> None: + """Against a target that slows past its capacity, the pool settles near the peak and short of its limit.""" + capacity = 40 + pool = _pool(desired=10, max_concurrency=100) + peak = max(range(1, 101), key=lambda level: level / _duration(level, capacity)) + + tail = _tail(await _run(pool, clock, capacity=capacity, minutes=30), minutes=5) + + assert pool.desired_concurrency < 100 + assert all(abs(level - peak) <= 0.25 * peak for level, _ in tail) + delivered = sum(level for level, _ in tail) / sum(duration for _, duration in tail) + assert delivered >= 0.9 * peak / _duration(peak, capacity) + + +async def test_holds_at_the_limit_below_the_knee(clock: _Clock) -> None: + """Against a target that never slows below the limit, the pool climbs to the limit and stays there.""" + pool = _pool(desired=10, max_concurrency=100) + + tail = _tail(await _run(pool, clock, capacity=1000, minutes=30), minutes=5) + + levels = [level for level, _ in tail] + assert levels.count(100) >= 0.75 * len(levels) + + +def test_emits_experimental_warning() -> None: + """Building the pool warns that it is experimental.""" + with pytest.warns(UserWarning, match='experimental'): + _pool() + + +async def test_tuning_by_subclassing(clock: _Clock) -> None: + """Class attributes set on a subclass reach the pool.""" + + class Wide(ThroughputAutoscaledPool): + startup_gain = 3.0 + + pool = _pool(cls=Wide, desired=20) + + assert await _levels(pool, clock, 10.0) == [60] diff --git a/tests/unit/crawlers/_basic/test_basic_crawler.py b/tests/unit/crawlers/_basic/test_basic_crawler.py index 16eb30bfe2..2386881bd2 100644 --- a/tests/unit/crawlers/_basic/test_basic_crawler.py +++ b/tests/unit/crawlers/_basic/test_basic_crawler.py @@ -19,7 +19,7 @@ import pytest -from crawlee import ConcurrencySettings, Glob, service_locator +from crawlee import ConcurrencySettings, Glob, ThroughputAutoscaledPool, service_locator from crawlee._log_config import CrawleeLogFormatter from crawlee._request import Request, RequestState from crawlee._types import BasicCrawlingContext, EnqueueLinksKwargs, HttpMethod @@ -2590,3 +2590,11 @@ async def handler(context: BasicCrawlingContext) -> None: assert empty_during_cooldown == [True] assert len(dispatched_at) == 2 assert dispatched_at[1] - dispatched_at[0] >= 0.5 + + +def test_autoscaled_pool_class() -> None: + """The crawler builds its pool from `autoscaled_pool_class`.""" + with pytest.warns(UserWarning, match='experimental'): + crawler = BasicCrawler(autoscaled_pool_class=ThroughputAutoscaledPool) + + assert isinstance(crawler._autoscaled_pool, ThroughputAutoscaledPool)