Usually I use shell command time
. My purpose is to test if data is small, medium, large or very large set, how much time and memory usage will be.
Any tools for Linux or just Python to do this?
Have a look at timeit, the python profiler and pycallgraph. Also make sure to have a look at the comment below by nikicc
mentioning "SnakeViz". It gives you yet another visualisation of profiling data which can be helpful.
def test():
"""Stupid test function"""
lst = []
for i in range(100):
lst.append(i)
if __name__ == '__main__':
import timeit
print(timeit.timeit("test()", setup="from __main__ import test"))
# For Python>=3.5 one can also write:
print(timeit.timeit("test()", globals=locals()))
Essentially, you can pass it python code as a string parameter, and it will run in the specified amount of times and prints the execution time. The important bits from the docs:
timeit.timeit(stmt='pass', setup='pass', timer=<default timer>, number=1000000, globals=None)
Create aTimer
instance with the given statement, setup code and timer function and run itstimeit
method with number executions. The optional globals argument specifies a namespace in which to execute the code.
... and:
Timer.timeit(number=1000000)
Time number executions of the main statement. This executes the setup statement once, and then returns the time it takes to execute the main statement a number of times, measured in seconds as a float. The argument is the number of times through the loop, defaulting to one million. The main statement, the setup statement and the timer function to be used are passed to the constructor.Note: By default,
timeit
temporarily turns offgarbage collection
during the timing. The advantage of this approach is that it makes independent timings more comparable. This disadvantage is that GC may be an important component of the performance of the function being measured. If so, GC can be re-enabled as the first statement in the setup string. For example:
timeit.Timer('for i in xrange(10): oct(i)', 'gc.enable()').timeit()
Profiling will give you a much more detailed idea about what's going on. Here's the "instant example" from the official docs:
import cProfile
import re
cProfile.run('re.compile("foo|bar")')
Which will give you:
197 function calls (192 primitive calls) in 0.002 seconds
Ordered by: standard name
ncalls tottime percall cumtime percall filename:lineno(function)
1 0.000 0.000 0.001 0.001 <string>:1(<module>)
1 0.000 0.000 0.001 0.001 re.py:212(compile)
1 0.000 0.000 0.001 0.001 re.py:268(_compile)
1 0.000 0.000 0.000 0.000 sre_compile.py:172(_compile_charset)
1 0.000 0.000 0.000 0.000 sre_compile.py:201(_optimize_charset)
4 0.000 0.000 0.000 0.000 sre_compile.py:25(_identityfunction)
3/1 0.000 0.000 0.000 0.000 sre_compile.py:33(_compile)
Both of these modules should give you an idea about where to look for bottlenecks.
Also, to get to grips with the output of profile
, have a look at this post
NOTE pycallgraph has been officially abandoned since Feb. 2018. As of Dec. 2020 it was still working on Python 3.6 though. As long as there are no core changes in how python exposes the profiling API it should remain a helpful tool though.
This module uses graphviz to create callgraphs like the following:
You can easily see which paths used up the most time by colour. You can either create them using the pycallgraph API, or using a packaged script:
pycallgraph graphviz -- ./mypythonscript.py
The overhead is quite considerable though. So for already long-running processes, creating the graph can take some time.
python -m cProfile -o results.prof myscript.py
. The oputput file can then be very nicely presented in a browser by a program called SnakeViz using snakeviz results.prof
pip install pycallgraph
. If I run the command above on my script I get the following error 'pycallgraph' is not recognized as an internal or external command, operable program or batch file.
Any idea why that is the case?
timeit.timeit(stmt='print(1)')
As you can see, I simply did not specify anything for the timer. You only need to specify it if you need a non-standard timing function. And that is probably out of scope of this answer and you should have a look at the official Python docs.
I use a simple decorator to time the func
import time
def st_time(func):
"""
st decorator to calculate the total time of a func
"""
def st_func(*args, **keyArgs):
t1 = time.time()
r = func(*args, **keyArgs)
t2 = time.time()
print("Function=%s, Time=%s" % (func.__name__, t2 - t1))
return r
return st_func
The timeit
module was slow and weird, so I wrote this:
def timereps(reps, func):
from time import time
start = time()
for i in range(0, reps):
func()
end = time()
return (end - start) / reps
Example:
import os
listdir_time = timereps(10000, lambda: os.listdir('/'))
print "python can do %d os.listdir('/') per second" % (1 / listdir_time)
For me, it says:
python can do 40925 os.listdir('/') per second
This is a primitive sort of benchmarking, but it's good enough.
I usually do a quick time ./script.py
to see how long it takes. That does not show you the memory though, at least not as a default. You can use /usr/bin/time -v ./script.py
to get a lot of information, including memory usage.
/usr/bin/time
with -v
option is not available as default in many distros, has to be installed. sudo apt-get install time
in debian, ubuntu, etc. pacman -S time
archlinux
Commented
Oct 6, 2013 at 20:40
Memory Profiler for all your memory needs.
https://pypi.python.org/pypi/memory_profiler
Run a pip install:
pip install memory_profiler
Import the library:
import memory_profiler
Add a decorator to the item you wish to profile:
@profile
def my_func():
a = [1] * (10 ** 6)
b = [2] * (2 * 10 ** 7)
del b
return a
if __name__ == '__main__':
my_func()
Execute the code:
python -m memory_profiler example.py
Recieve the output:
Line # Mem usage Increment Line Contents
==============================================
3 @profile
4 5.97 MB 0.00 MB def my_func():
5 13.61 MB 7.64 MB a = [1] * (10 ** 6)
6 166.20 MB 152.59 MB b = [2] * (2 * 10 ** 7)
7 13.61 MB -152.59 MB del b
8 13.61 MB 0.00 MB return a
Examples are from the docs, linked above.
pip install line_profiler
@profile
decorator before function. For example:@profile
def function(base, index, shift):
addend = index << shift
result = base + addend
return result
kernprof -l <file_name>
to create an instance of line_profiler. For example:kernprof -l test.py
kernprof will print Wrote profile results to <file_name>.lprof
on success. For example:
Wrote profile results to test.py.lprof
python -m line_profiler <file_name>.lprof
to print benchmark results. For example:python -m line_profiler test.py.lprof
You will see detailed info about each line of code:
Timer unit: 1e-06 s
Total time: 0.0021632 s
File: test.py
Function: function at line 1
Line # Hits Time Per Hit % Time Line Contents
==============================================================
1 @profile
2 def function(base, index, shift):
3 1000 796.4 0.8 36.8 addend = index << shift
4 1000 745.9 0.7 34.5 result = base + addend
5 1000 620.9 0.6 28.7 return result
pip install memory_profiler
@profile
decorator before function. For example:@profile
def function():
result = []
for i in range(10000):
result.append(i)
return result
python -m memory_profiler <file_name>
to print benchmark results. For example:python -m memory_profiler test.py
You will see detailed info about each line of code:
Filename: test.py
Line # Mem usage Increment Occurences Line Contents
============================================================
1 40.246 MiB 40.246 MiB 1 @profile
2 def function():
3 40.246 MiB 0.000 MiB 1 result = []
4 40.758 MiB 0.008 MiB 10001 for i in range(10000):
5 40.758 MiB 0.504 MiB 10000 result.append(i)
6 40.758 MiB 0.000 MiB 1 return result
Call a function many times to minimize environment impact.
snakeviz
interactive viewer for cProfile
https://github.com/jiffyclub/snakeviz/
cProfile was mentioned at https://stackoverflow.com/a/1593034/895245 and snakeviz was mentioned in a comment, but I wanted to highlight it further.
It is very hard to debug program performance just by looking at cprofile
/ pstats
output, because they can only total times per function out of the box.
However, what we really need in general is to see a nested view containing the stack traces of each call to actually find the main bottlenecks easily.
And this is exactly what snakeviz provides via its default "icicle" view.
First you have to dump the cProfile data to a binary file, and then you can snakeviz on that
pip install -u snakeviz
python -m cProfile -o results.prof myscript.py
snakeviz results.prof
This prints an URL to stdout which you can open on your browser, which contains the desired output that looks like this:
and you can then:
More profile oriented question: How can you profile a Python script?
If you don't want to write boilerplate code for timeit and get easy to analyze results, take a look at benchmarkit. Also it saves history of previous runs, so it is easy to compare the same function over the course of development.
# pip install benchmarkit
from benchmarkit import benchmark, benchmark_run
N = 10000
seq_list = list(range(N))
seq_set = set(range(N))
SAVE_PATH = '/tmp/benchmark_time.jsonl'
@benchmark(num_iters=100, save_params=True)
def search_in_list(num_items=N):
return num_items - 1 in seq_list
@benchmark(num_iters=100, save_params=True)
def search_in_set(num_items=N):
return num_items - 1 in seq_set
benchmark_results = benchmark_run(
[search_in_list, search_in_set],
SAVE_PATH,
comment='initial benchmark search',
)
Prints to terminal and returns list of dictionaries with data for the last run. Command line entrypoints also available.
If you change N=1000000
and rerun
Have a look at nose and at one of its plugins, this one in particular.
Once installed, nose is a script in your path, and that you can call in a directory which contains some python scripts:
$: nosetests
This will look in all the python files in the current directory and will execute any function that it recognizes as a test: for example, it recognizes any function with the word test_ in its name as a test.
So you can just create a python script called test_yourfunction.py and write something like this in it:
$: cat > test_yourfunction.py
def test_smallinput():
yourfunction(smallinput)
def test_mediuminput():
yourfunction(mediuminput)
def test_largeinput():
yourfunction(largeinput)
Then you have to run
$: nosetest --with-profile --profile-stats-file yourstatsprofile.prof testyourfunction.py
and to read the profile file, use this python line:
python -c "import hotshot.stats ; stats = hotshot.stats.load('yourstatsprofile.prof') ; stats.sort_stats('time', 'calls') ; stats.print_stats(200)"
nose
relies on hotshot. It's no longer maintained since Python 2.5 and is only kept "for specialized usage"
There are several ways to benchmark Python scripts. One simple way to do this is by using the timeit module, which provides a simple way to measure the execution time of small code snippets. However, if you are looking for a more comprehensive benchmark that includes memory usage, you can use the memory_profiler package to measure memory usage.
To visualize your benchmarks, you can use the plotly library, which allows you to create interactive plots. You can create a line chart to display the execution time and memory usage for different input sizes.
Here's an example code snippet to benchmark two different implementations of a function that takes a matrix, row and column as inputs:
import timeit
import random
import numpy as np
from plotly.subplots import make_subplots
import plotly.graph_objects as go
from memory_profiler import memory_usage
from memory_profiler import profile
from my.package.module import real_func_1, real_func_2
@profile
def func_impl_1(matrix, row, column):
return real_func_1(matrix, row, column)
@profile
def func_impl_2(matrix, row, column):
return real_func_2(matrix, row, column)
# Analysis range
x = list(range(3, 100))
# Time results
y1 = []
y2 = []
# Memory results
m1 = []
m2 = []
for i in x:
# Random choice of parameters
A = np.random.rand(i, i)
rx = random.randint(0, i-1)
ry = random.randint(0, i-1)
t1 = 0
t2 = 0
m1_ = 0
m2_ = 0
for _ in range(10):
t1 += timeit.timeit(
lambda: func_impl_1(A, rx, ry),
number=1,
)
t2 += timeit.timeit(
lambda: func_impl_2(A, rx, ry),
number=1,
)
m1_ += max(memory_usage(
(lambda: func_impl_1(A, rx, ry),)
))
m2_ += max(memory_usage(
(lambda: func_impl_2(A, rx, ry),)
))
y1.append(t1/100)
y2.append(t2/100)
m1.append(m1_/100)
m2.append(m2_/100)
# Title of first graph:
fig = make_subplots(rows=2, cols=1, shared_xaxes=True, subplot_titles=("Time", "Memory"))
fig.add_trace(go.Scatter(x=x, y=y1, name='func_impl_1 time', legendgroup='1'), row=1, col=1)
fig.add_trace(go.Scatter(x=x, y=y2, name='func_impl_2 time', legendgroup='1'), row=1, col=1)
fig.add_trace(go.Scatter(x=x, y=m1, name='func_impl_1 memory', legendgroup='2'), row=2, col=1)
fig.add_trace(go.Scatter(x=x, y=m2, name='func_impl_2 memory', legendgroup='2'), row=2, col=1)
fig.update_layout(
title="Performance of the functions",
xaxis_title="Matrix size",
)
fig.update_yaxes(title_text="Time (s)", row=1, col=1)
fig.update_yaxes(title_text="Max Memory usage (MB)", row=2, col=1)
fig.show()
Looking at the graph, it seems like both functions have similar memory usage, which is good to know. In terms of runtime, func_impl_2 seems to be generally faster than func_impl_1, which is also a positive finding. However, the difference in performance between the two functions is quite small, and there is a point where the performance of func_impl_1 surpasses that of func_impl_2 for very small input sizes. This may indicate that the simpler implementation of func_impl_1 is still a viable option for smaller inputs, even though func_impl_2 is generally faster. Overall, the graphs provide valuable insights into the performance of these functions and can help with decision-making when choosing which implementation to use in different scenarios.
The easy way to quickly test any function is to use this syntax :
%timeit my_code
For instance :
%timeit a = 1
13.4 ns ± 0.781 ns per loop (mean ± std. dev. of 7 runs, 100000000 loops each)
Be carefull timeit
is very slow, it take 12 second on my medium processor to just initialize (or maybe run the function). you can test this accepted answer
def test():
lst = []
for i in range(100):
lst.append(i)
if __name__ == '__main__':
import timeit
print(timeit.timeit("test()", setup="from __main__ import test")) # 12 second
for simple thing I will use time
instead, on my PC it return the result 0.0
import time
def test():
lst = []
for i in range(100):
lst.append(i)
t1 = time.time()
test()
result = time.time() - t1
print(result) # 0.000000xxxx
timeit
runs your function many times, to average out noise. The number of repeats is an option, see Benchmarking run times in python or the later part of the accepted answer on this question.
Commented
May 30, 2019 at 22:58
Based on Danyun Liu's answer with some convenience features, perhaps it is useful to someone.
def stopwatch(repeat=1, autorun=True):
"""
stopwatch decorator to calculate the total time of a function
"""
import timeit
import functools
def outer_func(func):
@functools.wraps(func)
def time_func(*args, **kwargs):
t1 = timeit.default_timer()
for _ in range(repeat):
r = func(*args, **kwargs)
t2 = timeit.default_timer()
print(f"Function={func.__name__}, Time={t2 - t1}")
return r
if autorun:
try:
time_func()
except TypeError:
raise Exception(f"{time_func.__name__}: autorun only works with no parameters, you may want to use @stopwatch(autorun=False)") from None
return time_func
if callable(repeat):
func = repeat
repeat = 1
return outer_func(func)
return outer_func
Some tests:
def is_in_set(x):
return x in {"linux", "darwin"}
def is_in_list(x):
return x in ["linux", "darwin"]
@stopwatch
def run_once():
import time
time.sleep(0.5)
@stopwatch(autorun=False)
def run_manually():
import time
time.sleep(0.5)
run_manually()
@stopwatch(repeat=10000000)
def repeat_set():
is_in_set("windows")
is_in_set("darwin")
@stopwatch(repeat=10000000)
def repeat_list():
is_in_list("windows")
is_in_list("darwin")
@stopwatch
def should_fail(x):
pass
Result:
Function=run_once, Time=0.5005391679987952
Function=run_manually, Time=0.500624185999186
Function=repeat_set, Time=1.7064883739985817
Function=repeat_list, Time=1.8905151920007484
Traceback (most recent call last):
(some more traceback here...)
Exception: should_fail: autorun only works with no parameters, you may want to use @stopwatch(autorun=False)
I wrote a tool to do concurrency stress test on a given function, and the out put is similar to Apache AB. may be this is what you want:
import itertools
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
def create_counter():
"""
Atomic counter
"""
return itertools.count()
def get_and_increase(counter):
return next(counter)
class Context:
def __init__(self, num_threads, target_fun):
self.failed_counter = create_counter()
self.start_barrier = threading.Barrier(num_threads)
self.target_fun = target_fun
class Worker:
def __init__(self, context, num_per_thread):
self.context = context
self.num_per_thread = num_per_thread
self.time_takes_arr = []
def run(self):
target_fun = self.context.target_fun
start_barrier = self.context.start_barrier
failed_counter = self.context.failed_counter
time_takes_arr = self.time_takes_arr
start_barrier.wait()
for i in range(self.num_per_thread):
start = time.time_ns()
try:
target_fun()
except Exception as err:
get_and_increase(failed_counter)
finally:
time_takes_arr.append(time.time_ns() - start)
def test(num, num_threads, target_fun, num_warm_up=0):
if num_warm_up > 0:
for i in range(num_warm_up):
target_fun()
executor = ThreadPoolExecutor(max_workers=num_threads)
num_per_thread = num // num_threads
context = Context(num_threads, target_fun)
workers = [Worker(context, num_per_thread) for _ in range(num_threads)]
for worker in workers:
executor.submit(lambda: worker.run())
executor.shutdown(wait=True)
act_num = num_per_thread * num_threads
failed_num = get_and_increase(context.failed_counter)
time_takes_all = []
time_takes_thread = []
for worker in workers:
time_takes_arr = worker.time_takes_arr
time_takes_thread.append(sum(time_takes_arr))
time_takes_all.extend(time_takes_arr)
time_takes_all.sort()
time_takes_thread.sort()
total_time_takes = time_takes_thread[-1]
return num_threads, total_time_takes, act_num, failed_num, time_takes_all
def format(test_name, num_threads, total_time_takes, act_num, failed_num, time_takes_all):
idx50 = act_num // 2
idx66 = act_num * 66 // 100
idx75 = act_num * 75 // 100
idx80 = act_num * 80 // 100
idx90 = act_num * 90 // 100
idx95 = act_num * 95 // 100
idx98 = act_num * 98 // 100
idx99 = act_num * 99 // 100
time_sum = sum(time_takes_all)
tps = 1000_000_000 * num_threads * (act_num / time_sum)
time_takes_avg = time_sum / len(time_takes_all) / 1000_000
return f"""{test_name} test result:
Concurrency Level: {num_threads:d}
Time taken for tests: {total_time_takes / 1000000:.6f} ms
Complete Tasks: {act_num:,}
Failed Tasks: {failed_num:,}
Tasks per second: {tps:,.2f}
Time per task: {time_takes_avg:.9f} ms
Time per task: {time_takes_avg / num_threads} ms (across all concurrent tasks)
Shortest task: {time_takes_all[0] / 1000000:.9f} ms
Percentage of the tasks served within a certain time (ms)
50% {time_takes_all[idx50] / 1000000:.6f}
66% {time_takes_all[idx66] / 1000000:.6f}
75% {time_takes_all[idx75] / 1000000:.6f}
80% {time_takes_all[idx80] / 1000000:.6f}
90% {time_takes_all[idx90] / 1000000:.6f}
95% {time_takes_all[idx95] / 1000000:.6f}
98% {time_takes_all[idx98] / 1000000:.6f}
99% {time_takes_all[idx99] / 1000000:.6f}
100% {time_takes_all[-1] / 1000000:.6f} (longest task)"""
def test_and_print(test_name, num, num_threads, target_fun, num_warm_up=10):
num_threads, total_time_takes, act_num, failed_num, time_takes_all = test(num, num_threads, target_fun, num_warm_up)
print(format(test_name, num_threads, total_time_takes, act_num, failed_num, time_takes_all))
if __name__ == "__main__":
c1 = create_counter()
def task():
get_and_increase(c1)
test_and_print("My test",
7000000, 50
, task
, 20)
print()
print(f"c1:{get_and_increase(c1)}")
and the output will be:
My test test result:
Concurrency Level: 50
Time taken for tests: 1544.781000 ms
Complete Tasks: 7,000,000
Failed Tasks: 0
Tasks per second: 10,133,790.35
Time per task: 0.004933988 ms
Time per task: 9.867976000000001e-05 ms (across all concurrent tasks)
Shortest task: -0.003000000 ms
Percentage of the tasks served within a certain time (ms)
50% 0.000000
66% 0.000000
75% 0.000000
80% 0.000000
90% 0.001000
95% 0.001000
98% 0.001000
99% 0.001000
100% 1120.871000 (longest task)