Function test_dynamic_memory

tests/unit/_autoscaling/test_snapshotter.py:391–447

Test dynamic memory scaling scenario where the system-wide memory can change. Create two memory snapshots. They have same memory usage, but different available memory. First snapshot is created with insufficient memory, so it is overloaded. Second snapshot is created with sufficient mem

(
    *,
    default_cpu_info: CpuInfo,
    event_manager: LocalEventManager,
    dynamic_memory: bool,
)

@pytest.mark.parametrize('dynamic_memory', [True, False])
async def test_dynamic_memory(
    *,
    default_cpu_info: CpuInfo,
    event_manager: LocalEventManager,
    dynamic_memory: bool,
) -> None:
    """Test dynamic memory scaling scenario where the system-wide memory can change.

    Create two memory snapshots. They have same memory usage, but different available memory.
    First snapshot is created with insufficient memory, so it is overloaded.
    Second snapshot is created with sufficient memory.

    Based on the Snapshotter configuration, it will either take into account the increased available memory or not.
    """
    _initial_memory_info = get_memory_info()
    ratio_just_below_system_wide_overload = 0.99 * SYSTEM_WIDE_MEMORY_OVERLOAD_THRESHOLD

    memory_mbytes = 0 if dynamic_memory else floor(_initial_memory_info.total_size.to_mb())

    service_locator.set_event_manager(event_manager)

    async with Snapshotter.from_config(
        Configuration(memory_mbytes=memory_mbytes, available_memory_ratio=ratio_just_below_system_wide_overload)
    ) as snapshotter:
        # Default state, memory usage exactly at the overload threshold -> overloaded, but not system-wide overloaded
        memory_infos = [
            # Overloaded sample
            MemoryInfo(
                total_size=_initial_memory_info.total_size,
                current_size=_initial_memory_info.total_size * ratio_just_below_system_wide_overload,
                system_wide_used_size=_initial_memory_info.total_size * ratio_just_below_system_wide_overload,
            ),
            # Same as first sample, with twice as memory available in the system
            MemoryInfo(
                total_size=_initial_memory_info.total_size * 2,  # Simulate increased total memory
                current_size=_initial_memory_info.total_size * ratio_just_below_system_wide_overload,
                system_wide_used_size=_initial_memory_info.total_size * ratio_just_below_system_wide_overload,
            ),
        ]

        for memory_info in memory_infos:
            event_manager.emit(
                event=Event.SYSTEM_INFO,
                event_data=EventSystemInfoData(
                    cpu_info=default_cpu_info,
                    memory_info=memory_info,
                ),
            )

        await event_manager.wait_for_all_listeners_to_complete()

        memory_samples = snapshotter.get_memory_sample()
        assert len(memory_samples) == 2
        # First sample will be overloaded.
        assert memory_samples[0].is_overloaded
        # Second sample can reflect the increased available memory based on the configuration used to create Snapshotter
        assert memory_samples[1].is_overloaded == (not dynamic_memory)