def monitor_cpu_ram_memory()

in optimum_benchmark/trackers/memory.py [0:0]

• 20 lines of code
• 8 McCabe index (conditional complexity)

def monitor_cpu_ram_memory(monitored_pid: int, connection: Connection):
    stop = False
    max_used_memory = 0
    monitored_process = psutil.Process(monitored_pid)

    if monitored_process.is_running():
        try:
            connection.send(0)
        except Exception:
            exit(0)

    while monitored_process.is_running() and not stop:
        meminfo_attr = "memory_info" if hasattr(monitored_process, "memory_info") else "get_memory_info"
        used_memory = getattr(monitored_process, meminfo_attr)()[0]
        max_used_memory = max(max_used_memory, used_memory)
        stop = connection.poll(MEMORY_CONSUMPTION_SAMPLING_RATE)

    if monitored_process.is_running():
        try:
            connection.send(max_used_memory / 1e6)  # convert to MB
        except Exception:
            exit(0)

    connection.close()