-
-
Notifications
You must be signed in to change notification settings - Fork 4
Expand file tree
/
Copy pathsimple_bench.py
More file actions
79 lines (65 loc) · 2.15 KB
/
Copy pathsimple_bench.py
File metadata and controls
79 lines (65 loc) · 2.15 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
#!/usr/bin/env python3
"""Simple event loop benchmark."""
import asyncio
import time
import statistics
def run_benchmark():
results = {}
# Timer benchmark
async def bench_timers(n=100):
latencies = []
for _ in range(n):
start = time.perf_counter()
await asyncio.sleep(0.001) # 1ms
elapsed = (time.perf_counter() - start) * 1000
latencies.append(elapsed - 1.0) # overhead only
return statistics.mean(latencies), sorted(latencies)[int(n*0.95)]
# Callback benchmark
async def bench_callbacks(n=10000):
count = [0]
done = asyncio.Event()
loop = asyncio.get_running_loop()
def callback():
count[0] += 1
if count[0] < n:
loop.call_soon(callback)
else:
loop.call_soon(done.set)
start = time.perf_counter()
loop.call_soon(callback)
await done.wait()
elapsed = time.perf_counter() - start
return count[0] / elapsed
# Test Erlang loop
try:
from _erlang_impl import ErlangEventLoop
erl_loop = ErlangEventLoop()
asyncio.set_event_loop(erl_loop)
erl_timer_mean, erl_timer_p95 = erl_loop.run_until_complete(bench_timers())
erl_cb_rate = erl_loop.run_until_complete(bench_callbacks())
erl_loop.close()
results['erlang'] = {
'timer_mean': erl_timer_mean,
'timer_p95': erl_timer_p95,
'callback_rate': erl_cb_rate
}
except Exception as e:
results['erlang_error'] = str(e)
# Test standard loop
std_loop = asyncio.new_event_loop()
asyncio.set_event_loop(std_loop)
std_timer_mean, std_timer_p95 = std_loop.run_until_complete(bench_timers())
std_cb_rate = std_loop.run_until_complete(bench_callbacks())
std_loop.close()
results['standard'] = {
'timer_mean': std_timer_mean,
'timer_p95': std_timer_p95,
'callback_rate': std_cb_rate
}
return results
if __name__ == '__main__':
results = run_benchmark()
print(results)
else:
# When exec'd from Erlang
benchmark_results = run_benchmark()