Try to add waits to avoid busy-waiting in cases where it makes no sense

Author: Pijukatel

src/crawlee/_autoscaling/autoscaled_pool.py

@@ -61,6 +61,9 @@ class AutoscaledPool:
     _TASK_TIMEOUT: timedelta | None = None
     """Timeout within which the `run_task_function` must complete."""
 
+    _OVERLOADED_BACKOFF_TIME: timedelta = timedelta(seconds=0.5)
+    """When overloaded, Autoscaled pool waits this long before rechecking system status."""
+
     def __init__(
         self,
         *,
@@ -238,6 +241,7 @@ async def _worker_task_orchestrator(self, run: _AutoscaledPoolRun) -> None:
                 current_status = self._system_status.get_current_system_info()
                 if not current_status.is_system_idle:
                     logger.info('Not scheduling new tasks - system is overloaded')
+                    await asyncio.sleep(self._OVERLOADED_BACKOFF_TIME)
                 elif self._is_paused:
                     logger.info('Not scheduling new tasks - the autoscaled pool is paused')
                 elif self.current_concurrency >= self.desired_concurrency:
@@ -261,7 +265,7 @@ async def _worker_task_orchestrator(self, run: _AutoscaledPoolRun) -> None:
 
             logger.info("Just finishing")
 
-        except Exception as e:
+        except BaseException as e:
             logger.error('What is hiding here?', exc_info=e)
             raise
 

src/crawlee/crawlers/_basic/_basic_crawler.py

@@ -1386,6 +1386,9 @@ async def __run_task_function(self) -> None:
         )
 
         if request is None:
+            # No request to process, request manager is neither finished nor empty.
+            # All requests are locked or in progress.
+            await asyncio.sleep(0.2) # Small backoff time to avoid overloading the system through busy-waiting.
             return
 
         if not (await self._is_allowed_based_on_robots_txt_file(request.url)):

src/crawlee/storage_clients/_file_system/_request_queue_client.py

@@ -596,6 +596,7 @@ async def is_empty(self) -> bool:
         async with self._lock:
             # If we have a cached value, return it immediately.
             if self._is_empty_cache is not None:
+                logger.info(f'From cache {self._is_empty_cache=}')
                 return self._is_empty_cache
 
             state = self._state.current_value