add ConcurrencySystem to share one concurrency budget between crawlers

Author: Mantisus

docs/examples/code_examples/resuming_paused_crawl.py

@@ -16,8 +16,9 @@
 
 async def main() -> None:
     crawler = BeautifulSoupCrawler(
-        # Let's slow down the crawler for a demonstration
-        concurrency_settings=ConcurrencySettings(max_tasks_per_minute=20)
+        # Let's slow down the crawler for a demonstration. It processes 2 requests
+        # and then waits for a minute, which leaves time to interrupt it.
+        concurrency_settings=ConcurrencySettings(max_tasks_per_minute=2)
     )
 
     @crawler.router.default_handler

docs/examples/fill_and_submit_web_form.mdx

@@ -102,7 +102,6 @@ Finally, run your crawler. Your logs should show something like this:
   "url": "https://httpbin.org/post"
 }
 
-[crawlee._autoscaling.autoscaled_pool] INFO  Waiting for remaining tasks to finish
 [HttpCrawler] INFO  Final request statistics:
 ┌───────────────────────────────┬─────────┐
 │ requests_finished             │ 1       │

docs/guides/architecture_overview.mdx

@@ -351,12 +351,12 @@ The <ApiLink to="class/EventManager">`EventManager`</ApiLink> is responsible for
 Crawlee provides several implementations of the event manager:
 
 - <ApiLink to="class/EventManager">`EventManager`</ApiLink> is the base class for event management in Crawlee.
-- <ApiLink to="class/LocalEventManager">`LocalEventManager`</ApiLink> extends the base event manager for local environments by automatically emitting `SYSTEM_INFO` events at regular intervals. This provides real-time system metrics including CPU usage and memory consumption, which are essential for internal components like the <ApiLink to="class/Snapshotter">`Snapshotter`</ApiLink> and <ApiLink to="class/AutoscaledPool">`AutoscaledPool`</ApiLink>.
+- <ApiLink to="class/LocalEventManager">`LocalEventManager`</ApiLink> extends the base event manager for local environments by automatically emitting `SYSTEM_INFO` events at regular intervals. This provides real-time system metrics including CPU usage and memory consumption, which are essential for internal components like the <ApiLink to="class/Snapshotter">`Snapshotter`</ApiLink> and <ApiLink to="class/ConcurrencySystem">`ConcurrencySystem`</ApiLink>.
 - [`ApifyEventManager`](https://docs.apify.com/sdk/python/reference/class/ApifyEventManager) - Manages events on the [Apify platform](https://apify.com/) (cloud-based). It is implemented in the [Apify SDK](https://docs.apify.com/sdk/python/).
 
 :::info
 
-You can learn more about <ApiLink to="class/Snapshotter">`Snapshotter`</ApiLink> and <ApiLink to="class/AutoscaledPool">`AutoscaledPool`</ApiLink> and their configuration in the [Scaling crawlers guide](./scaling-crawlers).
+You can learn more about <ApiLink to="class/Snapshotter">`Snapshotter`</ApiLink> and <ApiLink to="class/ConcurrencySystem">`ConcurrencySystem`</ApiLink> and their configuration in the [Scaling crawlers guide](./scaling-crawlers).
 
 :::
 

docs/guides/code_examples/scaling_crawlers/shared_concurrency_system_example.py

@@ -0,0 +1,48 @@
+import asyncio
+
+from crawlee import ConcurrencySettings
+from crawlee.autoscaling import ConcurrencySystem
+from crawlee.crawlers import (
+    BeautifulSoupCrawler,
+    BeautifulSoupCrawlingContext,
+    ParselCrawler,
+    ParselCrawlingContext,
+)
+from crawlee.storages import RequestQueue
+
+
+async def main() -> None:
+    concurrency_settings = ConcurrencySettings(max_concurrency=10)
+
+    # Each crawler needs a request queue of its own, otherwise they would both
+    # consume the default one and process each other's requests.
+    first_queue = await RequestQueue.open(alias='first-crawler')
+    second_queue = await RequestQueue.open(alias='second-crawler')
+
+    # One budget for both crawlers, instead of one each.
+    async with ConcurrencySystem(concurrency_settings) as concurrency_system:
+        first_crawler = BeautifulSoupCrawler(
+            concurrency_system=concurrency_system,
+            request_manager=first_queue,
+        )
+        second_crawler = ParselCrawler(
+            concurrency_system=concurrency_system,
+            request_manager=second_queue,
+        )
+
+        @first_crawler.router.default_handler
+        async def first_handler(context: BeautifulSoupCrawlingContext) -> None:
+            context.log.info(f'Processing {context.request.url}...')
+
+        @second_crawler.router.default_handler
+        async def second_handler(context: ParselCrawlingContext) -> None:
+            context.log.info(f'Processing {context.request.url}...')
+
+        await asyncio.gather(
+            first_crawler.run(['https://crawlee.dev/']),
+            second_crawler.run(['https://crawlee.dev/python/']),
+        )
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

docs/guides/scaling_crawlers.mdx

@@ -9,6 +9,7 @@ import RunnableCodeBlock from '@site/src/components/RunnableCodeBlock';
 
 import MaxTasksPerMinuteExample from '!!raw-loader!roa-loader!./code_examples/scaling_crawlers/max_tasks_per_minute_example.py';
 import MinAndMaxConcurrencyExample from '!!raw-loader!roa-loader!./code_examples/scaling_crawlers/min_and_max_concurrency_example.py';
+import SharedConcurrencySystemExample from '!!raw-loader!roa-loader!./code_examples/scaling_crawlers/shared_concurrency_system_example.py';
 
 As we build our crawler, we may want to control how many tasks it performs at any given time. In other words, how many requests it makes to the web we are trying to scrape. Crawlee offers several options to fine-tune the number of parallel tasks, limit the number of requests per minute, and optimize scaling based on available system resources.
 
@@ -20,7 +21,7 @@ All of these options are available across all crawlers provided by Crawlee. In t
 
 ## Max tasks per minute
 
-The `max_tasks_per_minute` setting in <ApiLink to="class/ConcurrencySettings">`ConcurrencySettings`</ApiLink> controls how many total tasks the crawler can process per minute. It ensures that tasks are spread evenly throughout the minute, preventing a sudden burst at the `max_concurrency` limit followed by idle time. By default, this is set to `Infinity`, meaning the crawler can run at full speed, limited only by `max_concurrency`. Use this option if you want to throttle your crawler to avoid overwhelming the target website with continuous requests.
+The `max_tasks_per_minute` setting in <ApiLink to="class/ConcurrencySettings">`ConcurrencySettings`</ApiLink> controls how many total tasks the crawler can process per minute. The tasks started over the last 60 seconds are counted, so the crawler may spend the whole limit early in the minute, as fast as its concurrency allows. It then waits until the oldest of those starts ages out of the window. By default, this is set to `Infinity`, meaning the crawler can run at full speed, limited only by `max_concurrency`. Use this option if you want to throttle your crawler to avoid overwhelming the target website with continuous requests.
 
 <RunnableCodeBlock className="language-python" language="python">
     {MaxTasksPerMinuteExample}
@@ -44,9 +45,15 @@ The `desired_concurrency` option in the <ApiLink to="class/ConcurrencySettings">
     {MinAndMaxConcurrencyExample}
 </RunnableCodeBlock>
 
-## Autoscaled pool
+## Concurrency system
 
-The <ApiLink to="class/AutoscaledPool">`AutoscaledPool`</ApiLink> manages a pool of asynchronous, resource-intensive tasks that run in parallel. It keeps `min_concurrency` tasks running even while the system is overloaded, and starts additional tasks only when there is enough free CPU and memory. To monitor system resources, it leverages the <ApiLink to="class/Snapshotter">`Snapshotter`</ApiLink> and <ApiLink to="class/SystemStatus">`SystemStatus`</ApiLink> classes. If any task raises an exception, the error is propagated, and the pool is stopped. Every crawler uses an <ApiLink to="class/AutoscaledPool">`AutoscaledPool`</ApiLink> under the hood.
+Every crawler runs its requests through an <ApiLink to="class/AutoscaledPool">`AutoscaledPool`</ApiLink>, and the pool asks a <ApiLink to="class/ConcurrencySystem">`ConcurrencySystem`</ApiLink> whether there is room for one more request. The system is what keeps `min_concurrency` requests running even while the machine is overloaded, starts further ones only when there is enough free CPU and memory, and scales the budget up and down as the load changes. To monitor system resources, it leverages the <ApiLink to="class/Snapshotter">`Snapshotter`</ApiLink> and <ApiLink to="class/SystemStatus">`SystemStatus`</ApiLink> classes. If any task raises an exception, the error is propagated and the pool is stopped.
+
+The crawler builds its own system from the `ConcurrencySettings` described above. Build one yourself and pass it as `concurrency_system` to go beyond those settings, or to share a single budget between several crawlers that would otherwise each scale against the whole machine and oversubscribe it. The budget goes to whoever asks for it first, so a crawler that joins a saturated system late may have to wait. Browser crawlers default to a lower `desired_concurrency` than the rest, and a supplied system replaces that default too, so set it in the `ConcurrencySettings` you build the system from. Holding the instance also lets you retune `min_concurrency` and `max_concurrency` while the crawl runs. The lifecycle of such a system belongs to you, so keep it open until every crawler using it has finished.
+
+<RunnableCodeBlock className="language-python" language="python">
+    {SharedConcurrencySystemExample}
+</RunnableCodeBlock>
 
 ## Running under a resource limit
 

docs/guides/scrapy_migration.mdx

@@ -195,7 +195,7 @@ The default (unnamed) <ApiLink to="class/Dataset">`Dataset`</ApiLink> stays on d
 
 ## Concurrency and throttling
 
-Both Scrapy and Crawlee have settings to influence the scraping throughput, but they differ in their approach. Crawlee reads <ApiLink to="class/ConcurrencySettings">`ConcurrencySettings`</ApiLink> and scales concurrency automatically based on system resources, as described in the [Scaling crawlers guide](./scaling-crawlers). Use <ApiLink to="class/ConcurrencySettings#__init__">`max_tasks_per_minute`</ApiLink> to cap the request rate. It's close to `DOWNLOAD_DELAY`, but it limits the whole pool rather than adding a fixed delay per domain.
+Both Scrapy and Crawlee have settings to influence the scraping throughput, but they differ in their approach. Crawlee reads <ApiLink to="class/ConcurrencySettings">`ConcurrencySettings`</ApiLink> and scales concurrency automatically based on system resources, as described in the [Scaling crawlers guide](./scaling-crawlers). Use <ApiLink to="class/ConcurrencySettings#__init__">`max_tasks_per_minute`</ApiLink> to cap the request rate. It's close to `DOWNLOAD_DELAY`, but it limits the whole pool rather than adding a fixed delay per domain. Requests aren't spaced out either. They start as fast as the concurrency allows until the cap for the last 60 seconds is reached, and then wait for that window to move on.
 
 <Tabs groupId="scrapy-migration-concurrency">
     <TabItem value="scrapy" label="Scrapy">

docs/upgrading/upgrading_to_v2.md

@@ -60,3 +60,65 @@ h2 = await context.query_selector_one('h2', timeout=timedelta(seconds=5))
 The storage clients now write their JSON data with camelCase keys, the same as Crawlee JS. This covers the metadata, request and record metadata files of `FileSystemStorageClient`, the request and record metadata JSON stored by `RedisStorageClient`, and the request JSON stored by `SqlStorageClient`. For example, `item_count` in a dataset's `__metadata__.json` becomes `itemCount`. The storage metadata documents of `RedisStorageClient` keep their snake_case keys.
 
 Data written by v1 still loads, so an existing storage directory, Redis instance or database keeps working. Any tooling that reads these files directly needs to handle the camelCase keys.
+
+## Concurrency is managed by `ConcurrencySystem`
+
+Load monitoring and concurrency scaling moved from `AutoscaledPool` to the new `ConcurrencySystem`. A crawler builds a default system from its `concurrency_settings` for each run, so a crawler configured through `concurrency_settings` works as before. To cap the combined concurrency of several crawlers, pass one shared `ConcurrencySystem` to all of them through the new `concurrency_system` option.
+
+The autoscaling classes now live in the public `crawlee.autoscaling` package, which replaces the private `crawlee._autoscaling`. Import `AutoscaledPool`, `ConcurrencySystem`, `ConcurrencyConsumer`, `SystemInfo` and `LoadRatioInfo` from there. `Snapshotter` and `SystemStatus` stay internal.
+
+`AutoscaledPool` now only runs the task loop. If you create a pool yourself, note that:
+
+- `system_status` and `concurrency_settings` are replaced by `concurrency_system`, which has to be active when the pool runs, and `consumer`, which identifies the pool to the system.
+- `run_task_function` receives the id the task is booked under.
+- `current_concurrency` comes from the concurrency system, so a shared system may count the tasks of other pools too. The pool's own tasks are counted by the new `own_concurrency`.
+
+For example:
+
+```python
+# v1
+from crawlee._autoscaling import AutoscaledPool, Snapshotter, SystemStatus
+
+async with Snapshotter.from_config() as snapshotter:
+    pool = AutoscaledPool(
+        system_status=SystemStatus(snapshotter),
+        concurrency_settings=ConcurrencySettings(max_concurrency=10),
+        run_task_function=run_task,
+        is_task_ready_function=is_task_ready,
+        is_finished_function=is_finished,
+    )
+    await pool.run()
+
+# v2
+from crawlee.autoscaling import AutoscaledPool, ConcurrencyConsumer, ConcurrencySystem
+
+settings = ConcurrencySettings(max_concurrency=10)
+
+async with ConcurrencySystem(settings) as concurrency_system:
+    pool = AutoscaledPool(
+        concurrency_system=concurrency_system,
+        consumer=ConcurrencyConsumer(id='my-pool'),
+        run_task_function=run_task,  # Now called with the task id.
+        is_task_ready_function=is_task_ready,
+        is_finished_function=is_finished,
+    )
+    await pool.run()
+```
+
+## `max_tasks_per_minute` counts the last minute
+
+`ConcurrencySettings.max_tasks_per_minute` now caps the tasks started over the last 60 seconds, the same as Crawlee JS. In v1, the pool spaced task starts evenly, `60 / max_tasks_per_minute` seconds apart. In v2, tasks start as fast as the concurrency allows until the cap is reached. The next ones wait until the oldest starts age out of the window. The rate over a minute stays the same, but tasks start in bursts. For details, see [Max tasks per minute](../guides/scaling-crawlers#max-tasks-per-minute).
+
+The value has to be a whole number of at least 1, or infinity, which is the default. A fractional value, such as `0.5` or `7.5`, now raises `ValueError`.
+
+## Autoscaling logs moved to DEBUG
+
+The periodic log of the autoscaling state and the `Waiting for remaining tasks to finish` message are now logged at DEBUG instead of INFO. The crawler's status message at INFO still reports the desired concurrency.
+
+The state log also has a new format. Instead of raw ratios for every resource, it names only the overloaded ones, with the share of time each spent over its threshold and its limit:
+
+```text
+current concurrency 3, desired 5, overloaded: cpu 58% of the time (limit 40%)
+```
+
+To read the concurrency values in code, use `crawler.concurrency_system`.

src/crawlee/_autoscaling/__init__.py

@@ -100,7 +100,7 @@ def __len__(self) -> int:
 
 @docs_group('Configuration')
 class ConcurrencySettings:
-    """Concurrency settings for AutoscaledPool."""
+    """Concurrency settings for a `ConcurrencySystem`."""
 
     def __init__(
         self,
@@ -116,8 +116,8 @@ def __init__(
             min_concurrency: The minimum number of tasks running in parallel. If you set this value too high
                 with respect to the available system memory and CPU, your code might run extremely slow or crash.
             max_concurrency: The maximum number of tasks running in parallel.
-            max_tasks_per_minute: The maximum number of tasks per minute the pool can run. By default, this is set
-                to infinity, but you can pass any positive, non-zero number.
+            max_tasks_per_minute: The maximum number of tasks per minute the crawler can run. By default, this is set
+                to infinity, but you can pass any whole number of at least 1.
             desired_concurrency: The desired number of tasks that should be running parallel on the start of the pool,
                 if there is a large enough supply of them.
         """
@@ -133,8 +133,10 @@ def __init__(
         if desired_concurrency > max_concurrency:
             raise ValueError('desired_concurrency cannot be greater than max_concurrency')
 
-        if max_tasks_per_minute <= 0:
-            raise ValueError('max_tasks_per_minute must be positive')
+        if max_tasks_per_minute != float('inf') and (
+            max_tasks_per_minute < 1 or not float(max_tasks_per_minute).is_integer()
+        ):
+            raise ValueError('max_tasks_per_minute must be a whole number of at least 1, or infinity')
 
         self.min_concurrency = min_concurrency
         self.max_concurrency = max_concurrency

src/crawlee/_types.py

@@ -0,0 +1,11 @@
+from ._autoscaled_pool import AutoscaledPool
+from ._concurrency_system import ConcurrencySystem
+from ._types import ConcurrencyConsumer, LoadRatioInfo, SystemInfo
+
+__all__ = [
+    'AutoscaledPool',
+    'ConcurrencyConsumer',
+    'ConcurrencySystem',
+    'LoadRatioInfo',
+    'SystemInfo',
+]