Skip to content

Instantly share code, notes, and snippets.

@Z4HRA-S
Created July 16, 2021 07:20
Show Gist options
  • Save Z4HRA-S/5fb1ca3133ce199bd82a4c395344545c to your computer and use it in GitHub Desktop.
Save Z4HRA-S/5fb1ca3133ce199bd82a4c395344545c to your computer and use it in GitHub Desktop.
Here is a module for profiling CPU and memory usage in python.The CumulativeProfile class is used as a decorator for monitoring CPU Consumption and RAM Utilization of a function.
from datetime import datetime
import cProfile
import time
import logging
import multiprocessing as mp
import psutil
class CumulativeProfile:
def __init__(self, func, v=0):
self.__name__ = func.__name__
self.func = func
self.calls = 0
self.total_time = 0
self.cpu = []
self.memory = []
self.queue = mp.Queue()
self.verbosity = 10 ** v
logging.basicConfig(
filename='logs/{}_cumulative_profile_{}.log'.format(func.__name__,
datetime.today().strftime('%Y-%d-%m-%H-%M-%S')),
filemode='w',
level=logging.INFO)
def __wrapper__(self, *args, **kwargs):
result = self.func(*args, **kwargs)
self.queue.put(result)
def __call__(self, *args, **kwargs):
process = mp.Process(target=self.__wrapper__, args=args, kwargs=kwargs)
start = time.time()
process.start()
working_pr = psutil.Process(process.pid)
while working_pr.status() == 'running':
cpu_perc = working_pr.cpu_percent()
if cpu_perc != 0:
self.cpu.append(cpu_perc)
mem_perc = working_pr.memory_full_info().uss
if mem_perc != 0:
self.memory.append(mem_perc)
time.sleep(0.0001)
execute_time = time.time() - start
result = self.queue.get()
process.join()
self.calls += 1
self.total_time += execute_time
logging.info("=" * 100)
logging.info("Total calls of {}: {}".format(self.func.__name__, self.calls))
logging.info("Total time of execution: {} s".format(self.total_time))
logging.info("Average run time: {:0.2f} s".format(self.total_time / self.calls))
if self.cpu:
logging.info("CPU usage (Percent): Average:{:0.2f}, Max:{}, Min:{}".format(sum(self.cpu) / len(self.cpu),
max(self.cpu), min(self.cpu)))
if self.memory:
logging.info("Memory usage (Mb): Average:{:0.2f}, Max:{:0.2f}, Min:{:0.2f}".format(
(sum(self.memory) / len(self.memory)) / (1024 ** 2),
max(self.memory) / (1024 ** 2), min(self.memory) / (1024 ** 2)))
if self.calls % self.verbosity == 0 and self.calls > self.verbosity:
print("{}th call of {} took {} seconds to execute".format(self.calls, self.func.__name__, execute_time))
return result
def __del__(self):
if self.calls:
print("=" * 50)
print("Total calls of {}: {}".format(self.func.__name__, self.calls))
print("Total time of execution: {} s".format(self.total_time))
print("Average run time: {:0.2f} s".format(self.total_time / self.calls))
if len(self.cpu):
print("CPU usage (Percent): Average:{:0.2f}, Max:{}, Min:{}".format(sum(self.cpu) / len(self.cpu),
max(self.cpu), min(self.cpu)))
if len(self.memory):
print("Memory usage (Mib): Average:{:0.2f}, Max:{:0.2f}, Min:{:0.2f}".format(
(sum(self.memory) / len(self.memory)) / (1024 ** 2),
max(self.memory) / (1024 ** 2), min(self.memory) / (1024 ** 2)))
print()
class CProfile:
def __init__(self, func):
self.func = func
def __call__(self, *args, **kwargs):
profiler = cProfile.Profile()
profiler.enable()
result = self.func(*args, **kwargs)
profiler.disable()
profiler.dump_stats(
'logs/{}_cProfile_{}.prof'.format(self.func.__name__, datetime.today().strftime('%Y-%d-%m-%H-%M-%S')))
return result
@Z4HRA-S
Copy link
Author

Z4HRA-S commented Jul 16, 2021

Unfortunately, sometimes (which I don't know exactly when :) ) the passed function won't run and goes to sleep. I have checked the process pid with strace and here is the result:
futex(0x562c8d8a3f64, FUTEX_WAIT_PRIVATE, 320, NULL
I can't figure out the problem.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment