Python code coverage for Doc/includes/mp_pool.py

#	count	content
1	n/a	import multiprocessing
2	n/a	import time
3	n/a	import random
4	n/a	import sys
5	n/a
6	n/a	#
7	n/a	# Functions used by test code
8	n/a	#
9	n/a
10	n/a	def calculate(func, args):
11	n/a	result = func(*args)
12	n/a	return '%s says that %s%s = %s' % (
13	n/a	multiprocessing.current_process().name,
14	n/a	func.__name__, args, result
15	n/a	)
16	n/a
17	n/a	def calculatestar(args):
18	n/a	return calculate(*args)
19	n/a
20	n/a	def mul(a, b):
21	n/a	time.sleep(0.5 * random.random())
22	n/a	return a * b
23	n/a
24	n/a	def plus(a, b):
25	n/a	time.sleep(0.5 * random.random())
26	n/a	return a + b
27	n/a
28	n/a	def f(x):
29	n/a	return 1.0 / (x - 5.0)
30	n/a
31	n/a	def pow3(x):
32	n/a	return x ** 3
33	n/a
34	n/a	def noop(x):
35	n/a	pass
36	n/a
37	n/a	#
38	n/a	# Test code
39	n/a	#
40	n/a
41	n/a	def test():
42	n/a	PROCESSES = 4
43	n/a	print('Creating pool with %d processes\n' % PROCESSES)
44	n/a
45	n/a	with multiprocessing.Pool(PROCESSES) as pool:
46	n/a	#
47	n/a	# Tests
48	n/a	#
49	n/a
50	n/a	TASKS = [(mul, (i, 7)) for i in range(10)] + \
51	n/a	[(plus, (i, 8)) for i in range(10)]
52	n/a
53	n/a	results = [pool.apply_async(calculate, t) for t in TASKS]
54	n/a	imap_it = pool.imap(calculatestar, TASKS)
55	n/a	imap_unordered_it = pool.imap_unordered(calculatestar, TASKS)
56	n/a
57	n/a	print('Ordered results using pool.apply_async():')
58	n/a	for r in results:
59	n/a	print('\t', r.get())
60	n/a	print()
61	n/a
62	n/a	print('Ordered results using pool.imap():')
63	n/a	for x in imap_it:
64	n/a	print('\t', x)
65	n/a	print()
66	n/a
67	n/a	print('Unordered results using pool.imap_unordered():')
68	n/a	for x in imap_unordered_it:
69	n/a	print('\t', x)
70	n/a	print()
71	n/a
72	n/a	print('Ordered results using pool.map() --- will block till complete:')
73	n/a	for x in pool.map(calculatestar, TASKS):
74	n/a	print('\t', x)
75	n/a	print()
76	n/a
77	n/a	#
78	n/a	# Test error handling
79	n/a	#
80	n/a
81	n/a	print('Testing error handling:')
82	n/a
83	n/a	try:
84	n/a	print(pool.apply(f, (5,)))
85	n/a	except ZeroDivisionError:
86	n/a	print('\tGot ZeroDivisionError as expected from pool.apply()')
87	n/a	else:
88	n/a	raise AssertionError('expected ZeroDivisionError')
89	n/a
90	n/a	try:
91	n/a	print(pool.map(f, list(range(10))))
92	n/a	except ZeroDivisionError:
93	n/a	print('\tGot ZeroDivisionError as expected from pool.map()')
94	n/a	else:
95	n/a	raise AssertionError('expected ZeroDivisionError')
96	n/a
97	n/a	try:
98	n/a	print(list(pool.imap(f, list(range(10)))))
99	n/a	except ZeroDivisionError:
100	n/a	print('\tGot ZeroDivisionError as expected from list(pool.imap())')
101	n/a	else:
102	n/a	raise AssertionError('expected ZeroDivisionError')
103	n/a
104	n/a	it = pool.imap(f, list(range(10)))
105	n/a	for i in range(10):
106	n/a	try:
107	n/a	x = next(it)
108	n/a	except ZeroDivisionError:
109	n/a	if i == 5:
110	n/a	pass
111	n/a	except StopIteration:
112	n/a	break
113	n/a	else:
114	n/a	if i == 5:
115	n/a	raise AssertionError('expected ZeroDivisionError')
116	n/a
117	n/a	assert i == 9
118	n/a	print('\tGot ZeroDivisionError as expected from IMapIterator.next()')
119	n/a	print()
120	n/a
121	n/a	#
122	n/a	# Testing timeouts
123	n/a	#
124	n/a
125	n/a	print('Testing ApplyResult.get() with timeout:', end=' ')
126	n/a	res = pool.apply_async(calculate, TASKS[0])
127	n/a	while 1:
128	n/a	sys.stdout.flush()
129	n/a	try:
130	n/a	sys.stdout.write('\n\t%s' % res.get(0.02))
131	n/a	break
132	n/a	except multiprocessing.TimeoutError:
133	n/a	sys.stdout.write('.')
134	n/a	print()
135	n/a	print()
136	n/a
137	n/a	print('Testing IMapIterator.next() with timeout:', end=' ')
138	n/a	it = pool.imap(calculatestar, TASKS)
139	n/a	while 1:
140	n/a	sys.stdout.flush()
141	n/a	try:
142	n/a	sys.stdout.write('\n\t%s' % it.next(0.02))
143	n/a	except StopIteration:
144	n/a	break
145	n/a	except multiprocessing.TimeoutError:
146	n/a	sys.stdout.write('.')
147	n/a	print()
148	n/a	print()
149	n/a
150	n/a
151	n/a	if __name__ == '__main__':
152	n/a	multiprocessing.freeze_support()
153	n/a	test()