test_check_mem.py

import threading
import psutil
import time


N = 1000
S = 0.01

a = list(range(N))
stop = True


def check_mem():
    p = psutil.Process()
    if hasattr(p, "memory_info"):
        mf = getattr(p, "memory_info")
    else:
        mf = getattr(p, 'get_memory_info')
    while not stop:
        mf()
        time.sleep(S)
    print("stopped")


def start_check():
    global stop
    stop = False
    t = threading.Thread(target=check_mem)
    t.start()


def stop_check():
    global stop
    stop = True
    time.sleep(10 * S)


def work():
    return sum([i * i for i in a])


def worker(msg):
    t = time.time()
    for i in range(100000):
        work()

    t = time.time() - t
    print(msg, t)
    return t


def test(s):
    global S
    S = s
    print("sleep", s)
    a = worker("no check")
    start_check()
    time.sleep(1)
    b = worker("check")
    print("chec/nocheck = ", b/a)
    stop_check()


def test_all():
    test(0.1)
    test(0.01)
    test(0.001)


test_all()