DEV Community: Kosuke Suzuki

Parallel monitoring of CPU usage with Python's psutil

Kosuke Suzuki — Tue, 09 Aug 2022 07:41:15 +0000

I wrote a script that periodically monitors CPU utilization while running some process in Python.

Multithreading allows periodic execution without stopping the main process. It may be useful when you want to report time series data of heavy computer load. The full code is at the bottom.

Library Installation

The library psutil can be used to get the CPU utilization in Python. You can install it via pip.

pip install psutil

The basic usage is as follows.

# percpu=True returns a list of utilization (%) per core
>>> psutil.cpu_percent(interval=1, percpu=True)
[7.0, 1.0, 8.9, 0.0, 5.0, 1.0, 4.0, 0.0]

# percpu=False returns an average utilization (%)
>>> psutil.cpu_percent(interval=1, percpu=False)
8.9

# Smaller interval(sec) shortens the measurement time, but also increases the error.
>>> psutil.cpu_percent(interval=0.5, percpu=True) 
[6.0, 0.0, 7.8, 1.9, 4.0, 0.0, 5.8, 0.0]

Parallel processing

This time, we will implement parallel processing using the native library threading.
Thread instances can be created and executed as follows.

m = threading.Thread(target=monitor_cpu,args=((initial_time,)))
m.start()

As arguments for threading.Thread(), target is a method to execute and args is the argument for the method.

monitor_cpu is the method for monitoring.

Monitoring method

Manage monitoring threads using threading.Event.
threading.Event is used to make a thread wait until an event occurs, and then resume the waiting thread when an event is raised by another thread.

We use 2 events.

wait(timeout) : make the current thread wait until an event occurs or timeout seconds elapse.
set() : fires an event.

def monitor_cpu(initial_time):
    print("START monitor_cpu")
    while not event.wait(1):  # Wait for 1 second, if no event occurs, execute inside the loop and wait again
        elapsed_time = time.time() - initial_time
        cpu_percent = psutil.cpu_percent(percpu=True)
        cpu_percent = '\t'.join(["{:10.4f}".format(v) for v in cpu_percent])
        print("time:", int(elapsed_time), cpu_percent)
    print("END monitor_cpu")  # When an event occurs, exit the loop and terminate execution


if __name__=="__main__":
    event = threading.Event()
    initial_time = time.time()
    m = threading.Thread(target=monitor_cpu,args=((initial_time,)))
    m.start()

    # Main process

    event.set()  # Fires an event

Inside monitor_cpu, an infinite loop runs with an execution interval of 1 second. When an event occurs, it breaks the loop and terminates execution.

def monitor_cpu(initial_time):
    print("START monitor_cpu")
    while flag:
        time.sleep(1)
        elapsed_time = time.time() - initial_time
        cpu_percent = psutil.cpu_percent(percpu=True)
        cpu_percent = '\t'.join(["{:10.4f}".format(v) for v in cpu_percent])
        print("time:", int(elapsed_time), cpu_percent)
    print("END monitor_cpu")

if __name__=="__main__":
    event = threading.Event()
    initial_time = time.time()
    flag = True
    m = threading.Thread(target=monitor_cpu,args=((initial_time,)))
    m.start()
    tmp = 0
    for i in range(100000000):
        tmp = i+i
    flag = False

Entire code

import threading
import time
import psutil

def monitor_cpu(initial_time):
    print("START monitor_cpu")
    while not event.wait(1):
        elapsed_time = time.time() - initial_time
        cpu_percent = psutil.cpu_percent(percpu=True)
        cpu_percent = '\t'.join(["{:10.4f}".format(v) for v in cpu_percent])
        print("time:", int(elapsed_time), cpu_percent)
    print("END monitor_cpu")

if __name__=="__main__":
    event = threading.Event()
    initial_time = time.time()
    m = threading.Thread(target=monitor_cpu,args=((initial_time,)))
    m.start()
    tmp = 0
    for i in range(100000000):
        tmp = i+i
    event.set()

Results

From left to right: time (seconds), CPU1 utilization (%), CPU2 utilization ... CPU8 utilization.

START monitor_cpu
time: 1  52.0     1.0    29.7     1.0    28.7     2.0    30.4     0.0
time: 2  45.0     3.0    36.3     3.0    32.4     1.0    29.4     1.0
time: 3  43.6     2.0    31.0     1.0    33.7     0.0    27.7     2.0
time: 4  45.5     2.0    25.2     1.0    22.8     0.0    19.8     0.0
time: 5  41.6     1.0    26.0     1.0    26.7     1.0    20.8     1.0
time: 6  46.1     5.0    34.7     3.0    38.0     3.0    31.7     4.0
time: 7  63.0    10.0    52.5    10.0    55.4    10.9    55.0    10.0
time: 8  51.5     6.9    36.0     4.9    41.2     5.0    39.6     4.9
time: 9  55.0     0.0    20.6     1.0    26.0     0.0    14.9     0.0
time: 10  49.5    2.9    22.8     1.0    25.2     1.0    23.5     1.0
time: 11  43.6    1.0    32.7     0.0    28.3     0.0    23.8     1.0
time: 12  47.5    2.9    22.8     1.0    20.6     2.0    25.0     2.0
END monitor_cpu

Multithreading in Django using threading and Queue

Kosuke Suzuki — Mon, 08 Aug 2022 07:54:00 +0000

In Django, if the process executed on a request is time-consuming, you may want to keep the process running behind and return a response anyway.

For example, sending a confirmation e-mail when a user registration is completed. If an application that takes 1 second to send an email receives 1,000 accesses per second, following things will be required.

The application does not lock during network I/O for sending mail.
The mail sending jobs are run behind and executed in the order in which they are accessed.

I had the opportunity to create an email sending function in Django, so here are my notes.

Confirmation of basic operation

First, we modified the sample code in the queue reference a little to check the process using queue and threading.

import threading
import queue
import time

q = queue.Queue()

def worker():
    while True:
        item = q.get()
        print(f'Working on {item}')
        time.sleep(1)
        print(f'Finished {item}')
        q.task_done()

# turn-on the worker thread
threading.Thread(target=worker, daemon=True).start()

# send thirty task requests to the worker
for item in range(5):
    print("Put item", item)
    q.put(item)
print('All task requests sent\n', end='')

# block until all tasks are done
q.join()
print('All work completed')

First declare q as a Queue instance in a global variable. This queue will be filled with jobs to be executed.

Next, we define the worker method as the job executor, which will retrieve the job, print it, and wait one second. Because of while true, it's always waiting,
When q.get() returns a return value, the underlying process is executed.

Next, start a new thread with threading.Thread(), passing target the method you want to execute. Then add the item as a job to the queue and the worker starts running. q.join() stops further processing until q.task_done() is called for all q.

Outpus

Put item 0
Put item 1
Put item 2
Put item 3
Put item 4
All task requests sent
Working on 0
Finished 0
Working on 1
Finished 1
Working on 2
Finished 2
Working on 3
Finished 3
Working on 4
Finished 4
All work completed

The output shows that all job additions are completed first, and the jobs are executed in the order they were added, indicating that job management and execution are performed asynchronously. It seems that sending emails can also be implemented in this way. Try it.

Asynchronous processing of tasks in Django

First, create a Django API for testing.

$ django-admin startproject mysite
$ cd mysite/
$ python manage.py startapp api

Then add the following code.

mysite/settings.py

...
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'api',
    'rest_framework'
]
...

mysite/urls.py

from django.contrib import admin
from django.urls import path
from api import views

urlpatterns = [
    path('admin/', admin.site.urls),
    path('index/', views.index)
]

api/views.py

from django.http import HttpResponse
# Create your views here.
def index(request):
    return HttpResponse("Hello, world.")

Then, start server

$ python manage.py runserver

Access http://127.0.0.1:8000/index/ and if Hello, world. is displayed, you are ready.

Now let's move on to the creation of the asynchronous mail function.

Create a new file api/mail.py and write the contents as follows.

import threading
import queue
import time

q = queue.Queue()

def worker():
    while True:
        item = q.get()
        print(f'Working on {item}')
        time.sleep(3)  # Do any processing here.
        print(f'Finished {item}')
        q.task_done()

# Five threads are set up to parallelize the process.
for _ in range(5): 
    threading.Thread(target=worker, daemon=True).start()

def add(item):
    q.put(item)

Next, modify views.py as follows.

from django.http import HttpResponse
from api.mail import add as mail_add

# Create your views here.
def index(request):
    item = request.GET.get("param")
    mail_add(item)
    return HttpResponse("Hello, world.")

Restart the server in this state and try sending a series of requests as follows.

curl http://127.0.0.1:8000/index/?param=1
curl http://127.0.0.1:8000/index/?param=2
curl http://127.0.0.1:8000/index/?param=3
curl http://127.0.0.1:8000/index/?param=4
curl http://127.0.0.1:8000/index/?param=5

Outputs

Working on 1
[15/May/2021 12:12:28] "GET /index/?param=1 HTTP/1.1" 200 13
Working on 2
[15/May/2021 12:12:28] "GET /index/?param=2 HTTP/1.1" 200 13
Working on 3
[15/May/2021 12:12:28] "GET /index/?param=3 HTTP/1.1" 200 13
Working on 4
[15/May/2021 12:12:28] "GET /index/?param=4 HTTP/1.1" 200 13
Working on 5
[15/May/2021 12:12:28] "GET /index/?param=5 HTTP/1.1" 200 13
Finished 1
Finished 2
Finished 3
Finished 4
Finished 5

Finished will be output almost simultaneously. If you rewrite the time.sleep(3) part as mail processing, an asynchronous mail sending function is completed.