Revert all but executor changes

Author: Pijukatel

.github/workflows/run_code_checks.yaml

@@ -40,17 +40,12 @@ jobs:
       HTTPBIN_URL: 'asd'
 
     steps:
-      - name: Os cleanup
+      - name: macOS cleanup
         if: runner.os == 'macOS'
+        # Disable Spotlight indexing and try to kill all useless processes that could drain CPU during tests
         run: |
           sudo mdutil -i off /
-          sudo killall ecosystemanalyticsd
-          sudo killall Finder || true
-          sudo killall mds || true
-          sudo killall mds_stores || true
-          sudo killall mds_worker || true
-          sudo killall mdworker || true
-          sudo killall mdworker_shared || true
+          sudo killall Finder spindump ecosystemanalyticsd SystemUIServer NotificationCenter mds mds_stores mds_worker mdworker mdworker_shared || true
 
       - name: Checkout repository
         uses: actions/checkout@v4

Makefile

@@ -31,7 +31,7 @@ type-check:
 
 unit-tests:
 	uv run pytest --numprocesses=1 -vv tests/unit -m "run_alone"
-	#uv run pytest --numprocesses=auto -vv tests/unit -m "not run_alone"
+	uv run pytest --numprocesses=auto -vv tests/unit -m "not run_alone"
 
 unit-tests-cov:
 	uv run pytest --numprocesses=1 -vv --cov=src/crawlee tests/unit -m "run_alone"

src/crawlee/_autoscaling/autoscaled_pool.py

@@ -46,7 +46,7 @@ class AutoscaledPool:
     _AUTOSCALE_INTERVAL = timedelta(seconds=10)
     """Interval at which the autoscaled pool adjusts the desired concurrency based on the latest system status."""
 
-    _LOGGING_INTERVAL = timedelta(seconds=0.5)
+    _LOGGING_INTERVAL = timedelta(minutes=1)
     """Interval at which the autoscaled pool logs its current state."""
 
     _DESIRED_CONCURRENCY_RATIO = 0.9
@@ -61,9 +61,6 @@ class AutoscaledPool:
     _TASK_TIMEOUT: timedelta | None = None
     """Timeout within which the `run_task_function` must complete."""
 
-    _OVERLOADED_BACKOFF_TIME: timedelta = timedelta(seconds=1)
-    """When overloaded, Autoscaled pool waits this long before rechecking system status."""
-
     def __init__(
         self,
         *,
@@ -126,33 +123,19 @@ async def run(self) -> None:
         )
 
         try:
-            logger.info(f'Await result')
             await run.result
-            logger.info(f'Finished naturally, {run.worker_tasks=}, {run.result.result()=}')
         except AbortError:
-            logger.info('AbortError')
             orchestrator.cancel()
             for task in run.worker_tasks:
                 if not task.done():
                     task.cancel()
-        except Exception as exc:
-            logger.error('Something sinister happened', exc_info=exc)
-            raise
-        except BaseException as exc:
-            logger.error('BaseException happened', exc_info=exc)
-            raise
-
         finally:
-            logger.error('finally')
             with suppress(asyncio.CancelledError):
-                logger.error('self._autoscale_task.stop()')
                 await self._autoscale_task.stop()
             with suppress(asyncio.CancelledError):
-                logger.error('await self._log_system_status_task.stop()')
                 await self._log_system_status_task.stop()
 
             if not orchestrator.done():
-                logger.error('not orchestrator.done()')
                 orchestrator.cancel()
             elif not orchestrator.cancelled() and orchestrator.exception() is not None:
                 logger.error('Exception in worker task orchestrator', exc_info=orchestrator.exception())
@@ -233,59 +216,48 @@ async def _worker_task_orchestrator(self, run: _AutoscaledPoolRun) -> None:
         Exits when `is_finished_function` returns True.
         """
         finished = False
-        logger.info('_worker_task_orchestrator')
+
         try:
             while not (finished := await self._is_finished_function()) and not run.result.done():
                 run.worker_tasks_updated.clear()
 
                 current_status = self._system_status.get_current_system_info()
                 if not current_status.is_system_idle:
-                    logger.info(f'Not scheduling new tasks - system is overloaded: {current_status}')
-                    await asyncio.sleep(self._OVERLOADED_BACKOFF_TIME.total_seconds())
-                    logger.info('Release the overloaded backoff')
+                    logger.debug('Not scheduling new tasks - system is overloaded')
                 elif self._is_paused:
-                    logger.info('Not scheduling new tasks - the autoscaled pool is paused')
+                    logger.debug('Not scheduling new tasks - the autoscaled pool is paused')
                 elif self.current_concurrency >= self.desired_concurrency:
-                    logger.info('Not scheduling new tasks - already running at desired concurrency')
+                    logger.debug('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.info('Scheduling a new task')
+                    logger.debug('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
 
                 with suppress(asyncio.TimeoutError):
                     await asyncio.wait_for(run.worker_tasks_updated.wait(), timeout=0.5)
-
-            logger.info("Just finishing")
-
-        except BaseException as e:
-            logger.error('What is hiding here?', exc_info=e)
-            raise
-
         finally:
-            logger.info(f'Finally pool. {finished=}, {(run.result.done())=}, {(run.result.result() if run.result.done() else None)=}')
             if finished:
-                logger.info('`is_finished_function` reports that we are finished')
+                logger.debug('`is_finished_function` reports that we are finished')
             elif run.result.done() and run.result.exception() is not None:
-                logger.info('Unhandled exception in `run_task_function`')
+                logger.debug('Unhandled exception in `run_task_function`')
 
             if run.worker_tasks:
-                logger.info('Terminating - waiting for tasks to complete')
+                logger.debug('Terminating - waiting for tasks to complete')
                 await asyncio.wait(run.worker_tasks, return_when=asyncio.ALL_COMPLETED)
-                logger.info('Worker tasks finished')
+                logger.debug('Worker tasks finished')
             else:
-                logger.info('Terminating - no running tasks to wait for')
+                logger.debug('Terminating - no running tasks to wait for')
 
             if not run.result.done():
-                run.result.set_result("Hello")
+                run.result.set_result(object())
 
     def _reap_worker_task(self, task: asyncio.Task, run: _AutoscaledPoolRun) -> None:
         """Handle cleanup and tracking of a completed worker task.

src/crawlee/_autoscaling/system_status.py

@@ -123,7 +123,6 @@ def _is_cpu_overloaded(self, sample_duration: timedelta | None = None) -> LoadRa
             CPU load ratio information.
         """
         sample = self._snapshotter.get_cpu_sample(sample_duration)
-        logger.info(sample)
         return self._is_sample_overloaded(sample, self._cpu_overload_threshold)
 
     def _is_memory_overloaded(self, sample_duration: timedelta | None = None) -> LoadRatioInfo:

src/crawlee/_log_config.py

@@ -4,7 +4,6 @@
 import logging
 import sys
 import textwrap
-from datetime import datetime
 from typing import TYPE_CHECKING, Any
 
 from colorama import Fore, Style, just_fix_windows_console
@@ -163,6 +162,6 @@ def format(self, record: logging.LogRecord) -> str:
 
         if self.include_logger_name:
             # Include logger name at the beginning of the log line
-            return f'{datetime.utcnow().strftime("%M-%S-%f")}{logger_name_string}{level_string}{log_string}{extra_string}{exception_string}'
+            return f'{logger_name_string}{level_string}{log_string}{extra_string}{exception_string}'
 
         return f'{level_string}{log_string}{extra_string}{exception_string}'

src/crawlee/_utils/system.py

@@ -5,7 +5,6 @@
 from contextlib import suppress
 from datetime import datetime, timezone
 from logging import getLogger
-from subprocess import run
 from typing import Annotated
 
 import psutil
@@ -94,13 +93,9 @@ def get_cpu_info() -> CpuInfo:
     It utilizes the `psutil` library. Function `psutil.cpu_percent()` returns a float representing the current
     system-wide CPU utilization as a percentage.
     """
-
-    cpu_percent = psutil.cpu_percent(percpu=True)
-    logger.info(f'Calling get_cpu_info()...: {cpu_percent}')
-    ratio = sum(cpu_percent)/len(cpu_percent) / 100
-    if ratio>0.95:
-        print_ps()
-    return CpuInfo(used_ratio=ratio)
+    logger.debug('Calling get_cpu_info()...')
+    cpu_percent = psutil.cpu_percent(interval=0.1)
+    return CpuInfo(used_ratio=cpu_percent / 100)
 
 
 def get_memory_info() -> MemoryInfo:
@@ -128,7 +123,3 @@ def get_memory_info() -> MemoryInfo:
         current_size=ByteSize(current_size_bytes),
         system_wide_used_size=ByteSize(vm.total - vm.available),
     )
-
-
-def print_ps():
-    run("ps -awxo pid,%cpu,comm", shell=True)

src/crawlee/crawlers/_basic/_basic_crawler.py

@@ -1130,7 +1130,6 @@ async def _handle_request_retries(
             await self._statistics.error_tracker.add(error=error, context=context)
 
             if self._error_handler:
-                self.log.warning('Error handler')
                 try:
                     new_request = await self._error_handler(context, error)
                 except Exception as e:
@@ -1146,20 +1145,18 @@ async def _handle_request_retries(
             await self._mark_request_as_handled(request)
             await self._handle_failed_request(context, error)
             self._statistics.record_request_processing_failure(request.unique_key)
-        self.log.warning('_handle_request_retries DONE')
 
     async def _handle_request_error(self, context: TCrawlingContext | BasicCrawlingContext, error: Exception) -> None:
         try:
             context.request.state = RequestState.ERROR_HANDLER
-            self.log.warning('Before _handle_request_error')
+
             await wait_for(
                 partial(self._handle_request_retries, context, error),
                 timeout=self._internal_timeout,
                 timeout_message='Handling request failure timed out after '
                 f'{self._internal_timeout.total_seconds()} seconds',
                 logger=self._logger,
             )
-            self.log.warning('After _handle_request_error')
 
             context.request.state = RequestState.DONE
         except UserDefinedErrorHandlerError:
@@ -1331,17 +1328,13 @@ async def __is_finished_function(self) -> bool:
 
         if self._abort_on_error and self._failed:
             self._failed = False
-            self.log.info('_abort_on_error')
             return True
 
         if self._keep_alive:
             return False
 
         request_manager = await self.get_request_manager()
-        is_finished = await request_manager.is_finished()
-        if is_finished:
-            self.log.info('I am sure this is a lie!!!')
-        return is_finished
+        return await request_manager.is_finished()
 
     async def __is_task_ready_function(self) -> bool:
         self._stop_if_max_requests_count_exceeded()
@@ -1353,10 +1346,7 @@ async def __is_task_ready_function(self) -> bool:
             return False
 
         request_manager = await self.get_request_manager()
-        is_ready = not await request_manager.is_empty()
-        if is_ready:
-            self.log.info('There is a request to process')
-        return is_ready
+        return not await request_manager.is_empty()
 
     async def __run_task_function(self) -> None:
         request_manager = await self.get_request_manager()
@@ -1370,11 +1360,6 @@ 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.
-            self._logger.warning("Backoff: No available requests to process.")
-            await asyncio.sleep(0.2) # Small backoff time to avoid overloading the system through busy-waiting.
-            self._logger.warning("Backoff finished.")
             return
 
         if not (await self._is_allowed_based_on_robots_txt_file(request.url)):
@@ -1415,10 +1400,8 @@ async def __run_task_function(self) -> None:
             try:
                 await self._run_request_handler(context=context)
             except asyncio.TimeoutError as e:
-                context.log.info('RH error')
                 raise RequestHandlerError(e, context) from e
 
-            context.log.info('Commit resutls')
             await self._commit_request_handler_result(context)
 
             await self._mark_request_as_handled(request)
@@ -1428,7 +1411,6 @@ async def __run_task_function(self) -> None:
             if context.session and context.session.is_usable:
                 context.session.mark_good()
 
-            context.log.info('Finished processing request')
             self._statistics.record_request_processing_finish(request.unique_key)
 
         except RequestCollisionError as request_error:

src/crawlee/events/_local_event_manager.py

@@ -93,8 +93,8 @@ async def __aexit__(
 
     async def _emit_system_info_event(self) -> None:
         """Emit a system info event with the current CPU and memory usage."""
-        cpu_info = get_cpu_info()
-        memory_info = get_memory_info()
+        cpu_info = await asyncio.to_thread(get_cpu_info)
+        memory_info = await asyncio.to_thread(get_memory_info)
 
         event_data = EventSystemInfoData(cpu_info=cpu_info, memory_info=memory_info)
         self.emit(event=Event.SYSTEM_INFO, event_data=event_data)

src/crawlee/statistics/_error_tracker.py

@@ -58,7 +58,6 @@ async def add(
             early: Flag indicating that the error is added earlier than usual to have access to resources that will be
              closed before normal error collection. This prevents double reporting during normal error collection.
         """
-        logger.warning('Adding error')
         if id(error) in self._early_reported_errors:
             # Error had to be collected earlier before relevant resources are closed.
             self._early_reported_errors.remove(id(error))
@@ -99,15 +98,12 @@ async def add(
             == 1
             and context is not None
         ):
-            logger.warning('awaiting snapshot')
             # Save snapshot only on the first occurrence of the error and only if context and kvs was passed as well.
             await self._capture_error_snapshot(
                 error_message=new_error_group_message or error_group_message,
                 file_and_line=error_group_file_and_line,
                 context=context,
             )
-            logger.warning('Snapshot added')
-        logger.warning('Finished')
 
     async def _capture_error_snapshot(
         self, error_message: str, file_and_line: str, context: BasicCrawlingContext