Python code coverage for Tools/ccbench/ccbench.py

#	count	content
1	n/a	# This file should be kept compatible with both Python 2.6 and Python >= 3.0.
2	n/a
3	n/a	from __future__ import division
4	n/a	from __future__ import print_function
5	n/a
6	n/a	"""
7	n/a	ccbench, a Python concurrency benchmark.
8	n/a	"""
9	n/a
10	n/a	import time
11	n/a	import os
12	n/a	import sys
13	n/a	import itertools
14	n/a	import threading
15	n/a	import subprocess
16	n/a	import socket
17	n/a	from optparse import OptionParser, SUPPRESS_HELP
18	n/a	import platform
19	n/a
20	n/a	# Compatibility
21	n/a	try:
22	n/a	xrange
23	n/a	except NameError:
24	n/a	xrange = range
25	n/a
26	n/a	try:
27	n/a	map = itertools.imap
28	n/a	except AttributeError:
29	n/a	pass
30	n/a
31	n/a
32	n/a	THROUGHPUT_DURATION = 2.0
33	n/a
34	n/a	LATENCY_PING_INTERVAL = 0.1
35	n/a	LATENCY_DURATION = 2.0
36	n/a
37	n/a	BANDWIDTH_PACKET_SIZE = 1024
38	n/a	BANDWIDTH_DURATION = 2.0
39	n/a
40	n/a
41	n/a	def task_pidigits():
42	n/a	"""Pi calculation (Python)"""
43	n/a	_map = map
44	n/a	_count = itertools.count
45	n/a	_islice = itertools.islice
46	n/a
47	n/a	def calc_ndigits(n):
48	n/a	# From http://shootout.alioth.debian.org/
49	n/a	def gen_x():
50	n/a	return _map(lambda k: (k, 4k + 2, 0, 2k + 1), _count(1))
51	n/a
52	n/a	def compose(a, b):
53	n/a	aq, ar, as_, at = a
54	n/a	bq, br, bs, bt = b
55	n/a	return (aq * bq,
56	n/a	aq * br + ar * bt,
57	n/a	as_ * bq + at * bs,
58	n/a	as_ * br + at * bt)
59	n/a
60	n/a	def extract(z, j):
61	n/a	q, r, s, t = z
62	n/a	return (qj + r) // (sj + t)
63	n/a
64	n/a	def pi_digits():
65	n/a	z = (1, 0, 0, 1)
66	n/a	x = gen_x()
67	n/a	while 1:
68	n/a	y = extract(z, 3)
69	n/a	while y != extract(z, 4):
70	n/a	z = compose(z, next(x))
71	n/a	y = extract(z, 3)
72	n/a	z = compose((10, -10*y, 0, 1), z)
73	n/a	yield y
74	n/a
75	n/a	return list(_islice(pi_digits(), n))
76	n/a
77	n/a	return calc_ndigits, (50, )
78	n/a
79	n/a	def task_regex():
80	n/a	"""regular expression (C)"""
81	n/a	# XXX this task gives horrendous latency results.
82	n/a	import re
83	n/a	# Taken from the `inspect` module
84	n/a	pat = re.compile(r'^(\sdef\s)\|(.(?<!\w)lambda(:\|\s))\|^(\s*@)', re.MULTILINE)
85	n/a	with open(__file__, "r") as f:
86	n/a	arg = f.read(2000)
87	n/a
88	n/a	def findall(s):
89	n/a	t = time.time()
90	n/a	try:
91	n/a	return pat.findall(s)
92	n/a	finally:
93	n/a	print(time.time() - t)
94	n/a	return pat.findall, (arg, )
95	n/a
96	n/a	def task_sort():
97	n/a	"""list sorting (C)"""
98	n/a	def list_sort(l):
99	n/a	l = l[::-1]
100	n/a	l.sort()
101	n/a
102	n/a	return list_sort, (list(range(1000)), )
103	n/a
104	n/a	def task_compress_zlib():
105	n/a	"""zlib compression (C)"""
106	n/a	import zlib
107	n/a	with open(__file__, "rb") as f:
108	n/a	arg = f.read(5000) * 3
109	n/a
110	n/a	def compress(s):
111	n/a	zlib.decompress(zlib.compress(s, 5))
112	n/a	return compress, (arg, )
113	n/a
114	n/a	def task_compress_bz2():
115	n/a	"""bz2 compression (C)"""
116	n/a	import bz2
117	n/a	with open(__file__, "rb") as f:
118	n/a	arg = f.read(3000) * 2
119	n/a
120	n/a	def compress(s):
121	n/a	bz2.compress(s)
122	n/a	return compress, (arg, )
123	n/a
124	n/a	def task_hashing():
125	n/a	"""SHA1 hashing (C)"""
126	n/a	import hashlib
127	n/a	with open(__file__, "rb") as f:
128	n/a	arg = f.read(5000) * 30
129	n/a
130	n/a	def compute(s):
131	n/a	hashlib.sha1(s).digest()
132	n/a	return compute, (arg, )
133	n/a
134	n/a
135	n/a	throughput_tasks = [task_pidigits, task_regex]
136	n/a	for mod in 'bz2', 'hashlib':
137	n/a	try:
138	n/a	globals()[mod] = __import__(mod)
139	n/a	except ImportError:
140	n/a	globals()[mod] = None
141	n/a
142	n/a	# For whatever reasons, zlib gives irregular results, so we prefer bz2 or
143	n/a	# hashlib if available.
144	n/a	# (NOTE: hashlib releases the GIL from 2.7 and 3.1 onwards)
145	n/a	if bz2 is not None:
146	n/a	throughput_tasks.append(task_compress_bz2)
147	n/a	elif hashlib is not None:
148	n/a	throughput_tasks.append(task_hashing)
149	n/a	else:
150	n/a	throughput_tasks.append(task_compress_zlib)
151	n/a
152	n/a	latency_tasks = throughput_tasks
153	n/a	bandwidth_tasks = [task_pidigits]
154	n/a
155	n/a
156	n/a	class TimedLoop:
157	n/a	def __init__(self, func, args):
158	n/a	self.func = func
159	n/a	self.args = args
160	n/a
161	n/a	def __call__(self, start_time, min_duration, end_event, do_yield=False):
162	n/a	step = 20
163	n/a	niters = 0
164	n/a	duration = 0.0
165	n/a	_time = time.time
166	n/a	_sleep = time.sleep
167	n/a	_func = self.func
168	n/a	_args = self.args
169	n/a	t1 = start_time
170	n/a	while True:
171	n/a	for i in range(step):
172	n/a	_func(*_args)
173	n/a	t2 = _time()
174	n/a	# If another thread terminated, the current measurement is invalid
175	n/a	# => return the previous one.
176	n/a	if end_event:
177	n/a	return niters, duration
178	n/a	niters += step
179	n/a	duration = t2 - start_time
180	n/a	if duration >= min_duration:
181	n/a	end_event.append(None)
182	n/a	return niters, duration
183	n/a	if t2 - t1 < 0.01:
184	n/a	# Minimize interference of measurement on overall runtime
185	n/a	step = step * 3 // 2
186	n/a	elif do_yield:
187	n/a	# OS scheduling of Python threads is sometimes so bad that we
188	n/a	# have to force thread switching ourselves, otherwise we get
189	n/a	# completely useless results.
190	n/a	_sleep(0.0001)
191	n/a	t1 = t2
192	n/a
193	n/a
194	n/a	def run_throughput_test(func, args, nthreads):
195	n/a	assert nthreads >= 1
196	n/a
197	n/a	# Warm up
198	n/a	func(*args)
199	n/a
200	n/a	results = []
201	n/a	loop = TimedLoop(func, args)
202	n/a	end_event = []
203	n/a
204	n/a	if nthreads == 1:
205	n/a	# Pure single-threaded performance, without any switching or
206	n/a	# synchronization overhead.
207	n/a	start_time = time.time()
208	n/a	results.append(loop(start_time, THROUGHPUT_DURATION,
209	n/a	end_event, do_yield=False))
210	n/a	return results
211	n/a
212	n/a	started = False
213	n/a	ready_cond = threading.Condition()
214	n/a	start_cond = threading.Condition()
215	n/a	ready = []
216	n/a
217	n/a	def run():
218	n/a	with ready_cond:
219	n/a	ready.append(None)
220	n/a	ready_cond.notify()
221	n/a	with start_cond:
222	n/a	while not started:
223	n/a	start_cond.wait()
224	n/a	results.append(loop(start_time, THROUGHPUT_DURATION,
225	n/a	end_event, do_yield=True))
226	n/a
227	n/a	threads = []
228	n/a	for i in range(nthreads):
229	n/a	threads.append(threading.Thread(target=run))
230	n/a	for t in threads:
231	n/a	t.setDaemon(True)
232	n/a	t.start()
233	n/a	# We don't want measurements to include thread startup overhead,
234	n/a	# so we arrange for timing to start after all threads are ready.
235	n/a	with ready_cond:
236	n/a	while len(ready) < nthreads:
237	n/a	ready_cond.wait()
238	n/a	with start_cond:
239	n/a	start_time = time.time()
240	n/a	started = True
241	n/a	start_cond.notify(nthreads)
242	n/a	for t in threads:
243	n/a	t.join()
244	n/a
245	n/a	return results
246	n/a
247	n/a	def run_throughput_tests(max_threads):
248	n/a	for task in throughput_tasks:
249	n/a	print(task.__doc__)
250	n/a	print()
251	n/a	func, args = task()
252	n/a	nthreads = 1
253	n/a	baseline_speed = None
254	n/a	while nthreads <= max_threads:
255	n/a	results = run_throughput_test(func, args, nthreads)
256	n/a	# Taking the max duration rather than average gives pessimistic
257	n/a	# results rather than optimistic.
258	n/a	speed = sum(r[0] for r in results) / max(r[1] for r in results)
259	n/a	print("threads=%d: %d" % (nthreads, speed), end="")
260	n/a	if baseline_speed is None:
261	n/a	print(" iterations/s.")
262	n/a	baseline_speed = speed
263	n/a	else:
264	n/a	print(" ( %d %%)" % (speed / baseline_speed * 100))
265	n/a	nthreads += 1
266	n/a	print()
267	n/a
268	n/a
269	n/a	LAT_END = "END"
270	n/a
271	n/a	def _sendto(sock, s, addr):
272	n/a	sock.sendto(s.encode('ascii'), addr)
273	n/a
274	n/a	def _recv(sock, n):
275	n/a	return sock.recv(n).decode('ascii')
276	n/a
277	n/a	def latency_client(addr, nb_pings, interval):
278	n/a	sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
279	n/a	try:
280	n/a	_time = time.time
281	n/a	_sleep = time.sleep
282	n/a	def _ping():
283	n/a	_sendto(sock, "%r\n" % _time(), addr)
284	n/a	# The first ping signals the parent process that we are ready.
285	n/a	_ping()
286	n/a	# We give the parent a bit of time to notice.
287	n/a	_sleep(1.0)
288	n/a	for i in range(nb_pings):
289	n/a	_sleep(interval)
290	n/a	_ping()
291	n/a	_sendto(sock, LAT_END + "\n", addr)
292	n/a	finally:
293	n/a	sock.close()
294	n/a
295	n/a	def run_latency_client(**kwargs):
296	n/a	cmd_line = [sys.executable, '-E', os.path.abspath(__file__)]
297	n/a	cmd_line.extend(['--latclient', repr(kwargs)])
298	n/a	return subprocess.Popen(cmd_line) #, stdin=subprocess.PIPE,
299	n/a	#stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
300	n/a
301	n/a	def run_latency_test(func, args, nthreads):
302	n/a	# Create a listening socket to receive the pings. We use UDP which should
303	n/a	# be painlessly cross-platform.
304	n/a	sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
305	n/a	sock.bind(("127.0.0.1", 0))
306	n/a	addr = sock.getsockname()
307	n/a
308	n/a	interval = LATENCY_PING_INTERVAL
309	n/a	duration = LATENCY_DURATION
310	n/a	nb_pings = int(duration / interval)
311	n/a
312	n/a	results = []
313	n/a	threads = []
314	n/a	end_event = []
315	n/a	start_cond = threading.Condition()
316	n/a	started = False
317	n/a	if nthreads > 0:
318	n/a	# Warm up
319	n/a	func(*args)
320	n/a
321	n/a	results = []
322	n/a	loop = TimedLoop(func, args)
323	n/a	ready = []
324	n/a	ready_cond = threading.Condition()
325	n/a
326	n/a	def run():
327	n/a	with ready_cond:
328	n/a	ready.append(None)
329	n/a	ready_cond.notify()
330	n/a	with start_cond:
331	n/a	while not started:
332	n/a	start_cond.wait()
333	n/a	loop(start_time, duration * 1.5, end_event, do_yield=False)
334	n/a
335	n/a	for i in range(nthreads):
336	n/a	threads.append(threading.Thread(target=run))
337	n/a	for t in threads:
338	n/a	t.setDaemon(True)
339	n/a	t.start()
340	n/a	# Wait for threads to be ready
341	n/a	with ready_cond:
342	n/a	while len(ready) < nthreads:
343	n/a	ready_cond.wait()
344	n/a
345	n/a	# Run the client and wait for the first ping(s) to arrive before
346	n/a	# unblocking the background threads.
347	n/a	chunks = []
348	n/a	process = run_latency_client(addr=sock.getsockname(),
349	n/a	nb_pings=nb_pings, interval=interval)
350	n/a	s = _recv(sock, 4096)
351	n/a	_time = time.time
352	n/a
353	n/a	with start_cond:
354	n/a	start_time = _time()
355	n/a	started = True
356	n/a	start_cond.notify(nthreads)
357	n/a
358	n/a	while LAT_END not in s:
359	n/a	s = _recv(sock, 4096)
360	n/a	t = _time()
361	n/a	chunks.append((t, s))
362	n/a
363	n/a	# Tell the background threads to stop.
364	n/a	end_event.append(None)
365	n/a	for t in threads:
366	n/a	t.join()
367	n/a	process.wait()
368	n/a	sock.close()
369	n/a
370	n/a	for recv_time, chunk in chunks:
371	n/a	# NOTE: it is assumed that a line sent by a client wasn't received
372	n/a	# in two chunks because the lines are very small.
373	n/a	for line in chunk.splitlines():
374	n/a	line = line.strip()
375	n/a	if line and line != LAT_END:
376	n/a	send_time = eval(line)
377	n/a	assert isinstance(send_time, float)
378	n/a	results.append((send_time, recv_time))
379	n/a
380	n/a	return results
381	n/a
382	n/a	def run_latency_tests(max_threads):
383	n/a	for task in latency_tasks:
384	n/a	print("Background CPU task:", task.__doc__)
385	n/a	print()
386	n/a	func, args = task()
387	n/a	nthreads = 0
388	n/a	while nthreads <= max_threads:
389	n/a	results = run_latency_test(func, args, nthreads)
390	n/a	n = len(results)
391	n/a	# We print out milliseconds
392	n/a	lats = [1000 * (t2 - t1) for (t1, t2) in results]
393	n/a	#print(list(map(int, lats)))
394	n/a	avg = sum(lats) / n
395	n/a	dev = (sum((x - avg) 2 for x in lats) / n) 0.5
396	n/a	print("CPU threads=%d: %d ms. (std dev: %d ms.)" % (nthreads, avg, dev), end="")
397	n/a	print()
398	n/a	#print(" [... from %d samples]" % n)
399	n/a	nthreads += 1
400	n/a	print()
401	n/a
402	n/a
403	n/a	BW_END = "END"
404	n/a
405	n/a	def bandwidth_client(addr, packet_size, duration):
406	n/a	sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
407	n/a	sock.bind(("127.0.0.1", 0))
408	n/a	local_addr = sock.getsockname()
409	n/a	_time = time.time
410	n/a	_sleep = time.sleep
411	n/a	def _send_chunk(msg):
412	n/a	_sendto(sock, ("%r#%s\n" % (local_addr, msg)).rjust(packet_size), addr)
413	n/a	# We give the parent some time to be ready.
414	n/a	_sleep(1.0)
415	n/a	try:
416	n/a	start_time = _time()
417	n/a	end_time = start_time + duration * 2.0
418	n/a	i = 0
419	n/a	while _time() < end_time:
420	n/a	_send_chunk(str(i))
421	n/a	s = _recv(sock, packet_size)
422	n/a	assert len(s) == packet_size
423	n/a	i += 1
424	n/a	_send_chunk(BW_END)
425	n/a	finally:
426	n/a	sock.close()
427	n/a
428	n/a	def run_bandwidth_client(**kwargs):
429	n/a	cmd_line = [sys.executable, '-E', os.path.abspath(__file__)]
430	n/a	cmd_line.extend(['--bwclient', repr(kwargs)])
431	n/a	return subprocess.Popen(cmd_line) #, stdin=subprocess.PIPE,
432	n/a	#stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
433	n/a
434	n/a	def run_bandwidth_test(func, args, nthreads):
435	n/a	# Create a listening socket to receive the packets. We use UDP which should
436	n/a	# be painlessly cross-platform.
437	n/a	with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
438	n/a	sock.bind(("127.0.0.1", 0))
439	n/a	addr = sock.getsockname()
440	n/a
441	n/a	duration = BANDWIDTH_DURATION
442	n/a	packet_size = BANDWIDTH_PACKET_SIZE
443	n/a
444	n/a	results = []
445	n/a	threads = []
446	n/a	end_event = []
447	n/a	start_cond = threading.Condition()
448	n/a	started = False
449	n/a	if nthreads > 0:
450	n/a	# Warm up
451	n/a	func(*args)
452	n/a
453	n/a	results = []
454	n/a	loop = TimedLoop(func, args)
455	n/a	ready = []
456	n/a	ready_cond = threading.Condition()
457	n/a
458	n/a	def run():
459	n/a	with ready_cond:
460	n/a	ready.append(None)
461	n/a	ready_cond.notify()
462	n/a	with start_cond:
463	n/a	while not started:
464	n/a	start_cond.wait()
465	n/a	loop(start_time, duration * 1.5, end_event, do_yield=False)
466	n/a
467	n/a	for i in range(nthreads):
468	n/a	threads.append(threading.Thread(target=run))
469	n/a	for t in threads:
470	n/a	t.setDaemon(True)
471	n/a	t.start()
472	n/a	# Wait for threads to be ready
473	n/a	with ready_cond:
474	n/a	while len(ready) < nthreads:
475	n/a	ready_cond.wait()
476	n/a
477	n/a	# Run the client and wait for the first packet to arrive before
478	n/a	# unblocking the background threads.
479	n/a	process = run_bandwidth_client(addr=addr,
480	n/a	packet_size=packet_size,
481	n/a	duration=duration)
482	n/a	_time = time.time
483	n/a	# This will also wait for the parent to be ready
484	n/a	s = _recv(sock, packet_size)
485	n/a	remote_addr = eval(s.partition('#')[0])
486	n/a
487	n/a	with start_cond:
488	n/a	start_time = _time()
489	n/a	started = True
490	n/a	start_cond.notify(nthreads)
491	n/a
492	n/a	n = 0
493	n/a	first_time = None
494	n/a	while not end_event and BW_END not in s:
495	n/a	_sendto(sock, s, remote_addr)
496	n/a	s = _recv(sock, packet_size)
497	n/a	if first_time is None:
498	n/a	first_time = _time()
499	n/a	n += 1
500	n/a	end_time = _time()
501	n/a
502	n/a	end_event.append(None)
503	n/a	for t in threads:
504	n/a	t.join()
505	n/a	process.kill()
506	n/a
507	n/a	return (n - 1) / (end_time - first_time)
508	n/a
509	n/a	def run_bandwidth_tests(max_threads):
510	n/a	for task in bandwidth_tasks:
511	n/a	print("Background CPU task:", task.__doc__)
512	n/a	print()
513	n/a	func, args = task()
514	n/a	nthreads = 0
515	n/a	baseline_speed = None
516	n/a	while nthreads <= max_threads:
517	n/a	results = run_bandwidth_test(func, args, nthreads)
518	n/a	speed = results
519	n/a	#speed = len(results) * 1.0 / results[-1][0]
520	n/a	print("CPU threads=%d: %.1f" % (nthreads, speed), end="")
521	n/a	if baseline_speed is None:
522	n/a	print(" packets/s.")
523	n/a	baseline_speed = speed
524	n/a	else:
525	n/a	print(" ( %d %%)" % (speed / baseline_speed * 100))
526	n/a	nthreads += 1
527	n/a	print()
528	n/a
529	n/a
530	n/a	def main():
531	n/a	usage = "usage: %prog [-h\|--help] [options]"
532	n/a	parser = OptionParser(usage=usage)
533	n/a	parser.add_option("-t", "--throughput",
534	n/a	action="store_true", dest="throughput", default=False,
535	n/a	help="run throughput tests")
536	n/a	parser.add_option("-l", "--latency",
537	n/a	action="store_true", dest="latency", default=False,
538	n/a	help="run latency tests")
539	n/a	parser.add_option("-b", "--bandwidth",
540	n/a	action="store_true", dest="bandwidth", default=False,
541	n/a	help="run I/O bandwidth tests")
542	n/a	parser.add_option("-i", "--interval",
543	n/a	action="store", type="int", dest="check_interval", default=None,
544	n/a	help="sys.setcheckinterval() value")
545	n/a	parser.add_option("-I", "--switch-interval",
546	n/a	action="store", type="float", dest="switch_interval", default=None,
547	n/a	help="sys.setswitchinterval() value")
548	n/a	parser.add_option("-n", "--num-threads",
549	n/a	action="store", type="int", dest="nthreads", default=4,
550	n/a	help="max number of threads in tests")
551	n/a
552	n/a	# Hidden option to run the pinging and bandwidth clients
553	n/a	parser.add_option("", "--latclient",
554	n/a	action="store", dest="latclient", default=None,
555	n/a	help=SUPPRESS_HELP)
556	n/a	parser.add_option("", "--bwclient",
557	n/a	action="store", dest="bwclient", default=None,
558	n/a	help=SUPPRESS_HELP)
559	n/a
560	n/a	options, args = parser.parse_args()
561	n/a	if args:
562	n/a	parser.error("unexpected arguments")
563	n/a
564	n/a	if options.latclient:
565	n/a	kwargs = eval(options.latclient)
566	n/a	latency_client(**kwargs)
567	n/a	return
568	n/a
569	n/a	if options.bwclient:
570	n/a	kwargs = eval(options.bwclient)
571	n/a	bandwidth_client(**kwargs)
572	n/a	return
573	n/a
574	n/a	if not options.throughput and not options.latency and not options.bandwidth:
575	n/a	options.throughput = options.latency = options.bandwidth = True
576	n/a	if options.check_interval:
577	n/a	sys.setcheckinterval(options.check_interval)
578	n/a	if options.switch_interval:
579	n/a	sys.setswitchinterval(options.switch_interval)
580	n/a
581	n/a	print("== %s %s (%s) ==" % (
582	n/a	platform.python_implementation(),
583	n/a	platform.python_version(),
584	n/a	platform.python_build()[0],
585	n/a	))
586	n/a	# Processor identification often has repeated spaces
587	n/a	cpu = ' '.join(platform.processor().split())
588	n/a	print("== %s %s on '%s' ==" % (
589	n/a	platform.machine(),
590	n/a	platform.system(),
591	n/a	cpu,
592	n/a	))
593	n/a	print()
594	n/a
595	n/a	if options.throughput:
596	n/a	print("--- Throughput ---")
597	n/a	print()
598	n/a	run_throughput_tests(options.nthreads)
599	n/a
600	n/a	if options.latency:
601	n/a	print("--- Latency ---")
602	n/a	print()
603	n/a	run_latency_tests(options.nthreads)
604	n/a
605	n/a	if options.bandwidth:
606	n/a	print("--- I/O bandwidth ---")
607	n/a	print()
608	n/a	run_bandwidth_tests(options.nthreads)
609	n/a
610	n/a	if __name__ == "__main__":
611	n/a	main()