How to produce a CPU load for several seconds, and measure it?

[Calab]

I am trying to create a small program that will put a heavy load on the CPU, and display the load over time.

The code below should start 10 processes. Each process doing a fair load for a while. The processes will be terminated after 10 seconds. It will display the CPU load every quarter second.

I am finding that the load does start around 100%, but only stays there for about a second, then drops to around 25%. I would expect the load to stay high for the full 10 seconds.

Is there a reason the load doesn't stay near 100%? Is there a way to keep the load high?

I'm currently running this on a Windows 10 PC with Python 3.12. It will be also run on a Linux system as well.

import time
import psutil
import multiprocessing

process_count = 10
run_time = 10  # seconds

def do_work(x):
    # Simulate a CPU-intensive task
    total = 0
    for i in range(10**9):
        total += i * i
    return total

if __name__ == '__main__':
    print(f"CPU load running for {run_time} seconds...")

    args = [0 for _ in range(number_of_processes)]
    pool = multiprocessing.Pool(processes=process_count)
    processes = [pool.apply_async(do_work, args)]

    # Determine when the processes should stop
    start_time = time.time()
    end_time = start_time + run_time

    # Allow processes to run for the amount of time specified
    count = 1
    sum = 0
    while time.time() < end_time:
        # Determine CPU load for .25 seconds
        cpu = psutil.cpu_percent(interval=.25)
        sum += cpu
        print(f"{count} CPU Load: {cpu:.2f}%")
        count += 1
                
    # Kill the pool of processes
    pool.terminate()

    avg = sum / count
    print(f"CPU load completed. Average CPU usage: {avg:.2f}%")

[deanhystad]

That 100% processor load is starting the processes. You create a processing pool, but don't give the processes anything to do. Only one process is running because you only create 1 task to perform. Create more tasks and the processor utilization will climb.

import psutil
import multiprocessing


def do_work():
    # Simulate a CPU-intensive task
    total = 0
    for i in range(10**9):
        total += i * i
    return total


if __name__ == "__main__":
    pool = multiprocessing.Pool(processes=8)
    results = [pool.apply_async(do_work, ()) for _ in range(20)]
    cpu = [psutil.cpu_percent(interval=0.1) for _ in range(100)]  # Collect CPU utilization 10 times a second for 10 seconds.
    pool.terminate()

    print(cpu, sum(cpu) / len(cpu))

Output:
[100.0, 100.0, 100.0, 89.6, 87.7, 90.7, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 
100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0] 99.68

[Plyth]

Thanks for the clarification, deanhystad. I'm just joining this thread, and your explanation makes sense. I was wondering why the CPU load would drop so quickly, and it didn’t occur to me that only one task was being run in the original code. By creating multiple tasks with apply, async, it ensures that all processes have work to do and can sustain the load. I also appreciate the example showing how to measure CPU usage in parallel, very helpful for testing across systems.