Pool deadlocked until timeout?

Author: Pijukatel

src/crawlee/_autoscaling/autoscaled_pool.py

@@ -237,20 +237,21 @@ 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.debug('Not scheduling new tasks - system is overloaded')
+                    logger.info('Not scheduling new tasks - system is overloaded')
                 elif self._is_paused:
-                    logger.debug('Not scheduling new tasks - the autoscaled pool is paused')
+                    logger.info('Not scheduling new tasks - the autoscaled pool is paused')
                 elif self.current_concurrency >= self.desired_concurrency:
-                    logger.debug('Not scheduling new tasks - already running at desired concurrency')
+                    logger.info('Not scheduling new tasks - already running at desired concurrency')
                 elif not await self._is_task_ready_function():
                     logger.debug('Not scheduling new task - no task is ready')
                 else:
-                    logger.debug('Scheduling a new task')
+                    logger.info('Scheduling a new task')
                     worker_task = asyncio.create_task(self._worker_task(), name='autoscaled pool worker task')
                     worker_task.add_done_callback(lambda task: self._reap_worker_task(task, run))
                     run.worker_tasks.append(worker_task)
 
                     if math.isfinite(self._max_tasks_per_minute):
+                        logger.info('Deadlock sleep????')
                         await asyncio.sleep(60 / self._max_tasks_per_minute)
 
                     continue

src/crawlee/crawlers/_basic/_basic_crawler.py

@@ -1369,7 +1369,10 @@ async def __is_task_ready_function(self) -> bool:
             return False
 
         request_manager = await self.get_request_manager()
-        return not await request_manager.is_empty()
+        is_ready = not await request_manager.is_empty()
+        if is_ready:
+            self.log.info('There is a request to process')
+        return is_ready
 
     async def __run_task_function(self) -> None:
         request_manager = await self.get_request_manager()

tests/unit/events/test_local_event_manager.py

@@ -17,15 +17,15 @@ def listener() -> AsyncMock:
     async def async_listener(payload: Any) -> None:
         pass
 
-    al = AsyncMock()
-    update_wrapper(al, async_listener)
-    return al
+    return AsyncMock(target=async_listener)
 
 
 async def test_emit_system_info_event(listener: AsyncMock) -> None:
-    async with LocalEventManager(system_info_interval=timedelta(milliseconds=50)) as event_manager:
+    system_info_interval = timedelta(milliseconds=50)
+    test_tolerance_coefficient = 10
+    async with LocalEventManager(system_info_interval=system_info_interval) as event_manager:
         event_manager.on(event=Event.SYSTEM_INFO, listener=listener)
-        await asyncio.sleep(0.2)
+        await asyncio.sleep(system_info_interval.total_seconds() * test_tolerance_coefficient)
 
     assert listener.call_count >= 1
     assert isinstance(listener.call_args[0][0], EventSystemInfoData)