Repository: apify/crawlee-python
File: src/crawlee/_autoscaling/_types.py (Python)

Code before:
    When of type `ByteSize` then it is used as fixed memory size. When of type `Ratio` then it allows for dynamic memory
    scaling based on the available system memory.
    """

    system_wide_memory_size: ByteSize | None
    """Total memory available in the whole system."""

    max_used_memory_ratio: float
    """The maximum acceptable ratio of `current_size` to `max_memory_size`."""

    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    """The time at which the system load information was measured."""

    @property
    def is_overloaded(self) -> bool:
        """Indicate whether the memory is considered as overloaded."""
        if self.system_wide_memory_size is not None and self.system_wide_used_size is not None:
            system_wide_utilization = self.system_wide_used_size / self.system_wide_memory_size
            if system_wide_utilization > SYSTEM_WIDE_MEMORY_OVERLOAD_THRESHOLD:
                return True

        if isinstance(self.max_memory_size, Ratio):
            # The snapshot overload is decided not when the snapshot was taken, but when `is_overload` property is
            # accessed. This allows for dynamic memory scaling. The same memory snapshot that used to be overloaded in
            # the past can become non-overloaded if the available memory was increased.
            max_memory_size = ByteSize(int(get_memory_info().total_size.bytes * self.max_memory_size.value))
        else:
            max_memory_size = self.max_memory_size

        return (self.current_size / max_memory_size) > self.max_used_memory_ratio


@dataclass
class EventLoopSnapshot:
    """Snapshot of the state of the event loop."""

    delay: timedelta
    """The current delay of the event loop."""

    max_delay: timedelta
    """The maximum delay that is considered acceptable."""

    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    """The time at which the system load information was measured."""

    @property
    def max_delay_exceeded(self) -> timedelta:
        """The amount of time by which the delay exceeds the maximum delay."""
        return max(self.delay - self.max_delay, timedelta(seconds=0))

    @property

Review comment (suggestion):
I believe `get_memory_info()` should be wrapped in `asyncio.to_thread` since it uses `psutil`.

Also, I don't think historical snapshots should be affected by memory scaling - the overload was real at the moment the snapshot was taken. Once it's older than `_SNAPSHOT_HISTORY`, it won't influence the `Snapshotter` anyway. The 30-second inertia from old memory values seems reasonable to me.

Code after:

    max_memory_size: ByteSize
    """The maximum memory that can be used by `AutoscaledPool`."""

    system_wide_memory_size: ByteSize | None
    """Total memory available in the whole system."""

    max_used_memory_ratio: float
    """The maximum acceptable ratio of `current_size` to `max_memory_size`."""

    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    """The time at which the system load information was measured."""

    @property
    def is_overloaded(self) -> bool:
        """Indicate whether the memory is considered as overloaded."""
        if self.system_wide_memory_size is not None and self.system_wide_used_size is not None:
            system_wide_utilization = self.system_wide_used_size / self.system_wide_memory_size
            if system_wide_utilization > SYSTEM_WIDE_MEMORY_OVERLOAD_THRESHOLD:
                return True

        return (self.current_size / self.max_memory_size) > self.max_used_memory_ratio


@dataclass
class EventLoopSnapshot:
    """Snapshot of the state of the event loop."""

    delay: timedelta
    """The current delay of the event loop."""

    max_delay: timedelta
    """The maximum delay that is considered acceptable."""

    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    """The time at which the system load information was measured."""

    @property
    def max_delay_exceeded(self) -> timedelta:
        """The amount of time by which the delay exceeds the maximum delay."""
        return max(self.delay - self.max_delay, timedelta(seconds=0))

    @property
    def is_overloaded(self) -> bool:
        """Indicate whether the event loop is considered as overloaded."""
        return self.delay > self.max_delay


@dataclass
class ClientSnapshot:
    """Snapshot of the state of the client."""