Python code coverage for Lib/test/_test_multiprocessing.py

#	count	content
1	n/a	#
2	n/a	# Unit tests for the multiprocessing package
3	n/a	#
4	n/a
5	n/a	import unittest
6	n/a	import queue as pyqueue
7	n/a	import time
8	n/a	import io
9	n/a	import itertools
10	n/a	import sys
11	n/a	import os
12	n/a	import gc
13	n/a	import errno
14	n/a	import signal
15	n/a	import array
16	n/a	import socket
17	n/a	import random
18	n/a	import logging
19	n/a	import struct
20	n/a	import operator
21	n/a	import test.support
22	n/a	import test.support.script_helper
23	n/a
24	n/a
25	n/a	# Skip tests if _multiprocessing wasn't built.
26	n/a	_multiprocessing = test.support.import_module('_multiprocessing')
27	n/a	# Skip tests if sem_open implementation is broken.
28	n/a	test.support.import_module('multiprocessing.synchronize')
29	n/a	# import threading after _multiprocessing to raise a more relevant error
30	n/a	# message: "No module named _multiprocessing". _multiprocessing is not compiled
31	n/a	# without thread support.
32	n/a	import threading
33	n/a
34	n/a	import multiprocessing.dummy
35	n/a	import multiprocessing.connection
36	n/a	import multiprocessing.managers
37	n/a	import multiprocessing.heap
38	n/a	import multiprocessing.pool
39	n/a
40	n/a	from multiprocessing import util
41	n/a
42	n/a	try:
43	n/a	from multiprocessing import reduction
44	n/a	HAS_REDUCTION = reduction.HAVE_SEND_HANDLE
45	n/a	except ImportError:
46	n/a	HAS_REDUCTION = False
47	n/a
48	n/a	try:
49	n/a	from multiprocessing.sharedctypes import Value, copy
50	n/a	HAS_SHAREDCTYPES = True
51	n/a	except ImportError:
52	n/a	HAS_SHAREDCTYPES = False
53	n/a
54	n/a	try:
55	n/a	import msvcrt
56	n/a	except ImportError:
57	n/a	msvcrt = None
58	n/a
59	n/a	#
60	n/a	#
61	n/a	#
62	n/a
63	n/a	def latin(s):
64	n/a	return s.encode('latin')
65	n/a
66	n/a	#
67	n/a	# Constants
68	n/a	#
69	n/a
70	n/a	LOG_LEVEL = util.SUBWARNING
71	n/a	#LOG_LEVEL = logging.DEBUG
72	n/a
73	n/a	DELTA = 0.1
74	n/a	CHECK_TIMINGS = False # making true makes tests take a lot longer
75	n/a	# and can sometimes cause some non-serious
76	n/a	# failures because some calls block a bit
77	n/a	# longer than expected
78	n/a	if CHECK_TIMINGS:
79	n/a	TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.82, 0.35, 1.4
80	n/a	else:
81	n/a	TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.1, 0.1, 0.1
82	n/a
83	n/a	HAVE_GETVALUE = not getattr(_multiprocessing,
84	n/a	'HAVE_BROKEN_SEM_GETVALUE', False)
85	n/a
86	n/a	WIN32 = (sys.platform == "win32")
87	n/a
88	n/a	from multiprocessing.connection import wait
89	n/a
90	n/a	def wait_for_handle(handle, timeout):
91	n/a	if timeout is not None and timeout < 0.0:
92	n/a	timeout = None
93	n/a	return wait([handle], timeout)
94	n/a
95	n/a	try:
96	n/a	MAXFD = os.sysconf("SC_OPEN_MAX")
97	n/a	except:
98	n/a	MAXFD = 256
99	n/a
100	n/a	# To speed up tests when using the forkserver, we can preload these:
101	n/a	PRELOAD = ['__main__', 'test.test_multiprocessing_forkserver']
102	n/a
103	n/a	#
104	n/a	# Some tests require ctypes
105	n/a	#
106	n/a
107	n/a	try:
108	n/a	from ctypes import Structure, c_int, c_double
109	n/a	except ImportError:
110	n/a	Structure = object
111	n/a	c_int = c_double = None
112	n/a
113	n/a
114	n/a	def check_enough_semaphores():
115	n/a	"""Check that the system supports enough semaphores to run the test."""
116	n/a	# minimum number of semaphores available according to POSIX
117	n/a	nsems_min = 256
118	n/a	try:
119	n/a	nsems = os.sysconf("SC_SEM_NSEMS_MAX")
120	n/a	except (AttributeError, ValueError):
121	n/a	# sysconf not available or setting not available
122	n/a	return
123	n/a	if nsems == -1 or nsems >= nsems_min:
124	n/a	return
125	n/a	raise unittest.SkipTest("The OS doesn't support enough semaphores "
126	n/a	"to run the test (required: %d)." % nsems_min)
127	n/a
128	n/a
129	n/a	#
130	n/a	# Creates a wrapper for a function which records the time it takes to finish
131	n/a	#
132	n/a
133	n/a	class TimingWrapper(object):
134	n/a
135	n/a	def __init__(self, func):
136	n/a	self.func = func
137	n/a	self.elapsed = None
138	n/a
139	n/a	def __call__(self, args, *kwds):
140	n/a	t = time.time()
141	n/a	try:
142	n/a	return self.func(args, *kwds)
143	n/a	finally:
144	n/a	self.elapsed = time.time() - t
145	n/a
146	n/a	#
147	n/a	# Base class for test cases
148	n/a	#
149	n/a
150	n/a	class BaseTestCase(object):
151	n/a
152	n/a	ALLOWED_TYPES = ('processes', 'manager', 'threads')
153	n/a
154	n/a	def assertTimingAlmostEqual(self, a, b):
155	n/a	if CHECK_TIMINGS:
156	n/a	self.assertAlmostEqual(a, b, 1)
157	n/a
158	n/a	def assertReturnsIfImplemented(self, value, func, *args):
159	n/a	try:
160	n/a	res = func(*args)
161	n/a	except NotImplementedError:
162	n/a	pass
163	n/a	else:
164	n/a	return self.assertEqual(value, res)
165	n/a
166	n/a	# For the sanity of Windows users, rather than crashing or freezing in
167	n/a	# multiple ways.
168	n/a	def __reduce__(self, *args):
169	n/a	raise NotImplementedError("shouldn't try to pickle a test case")
170	n/a
171	n/a	__reduce_ex__ = __reduce__
172	n/a
173	n/a	#
174	n/a	# Return the value of a semaphore
175	n/a	#
176	n/a
177	n/a	def get_value(self):
178	n/a	try:
179	n/a	return self.get_value()
180	n/a	except AttributeError:
181	n/a	try:
182	n/a	return self._Semaphore__value
183	n/a	except AttributeError:
184	n/a	try:
185	n/a	return self._value
186	n/a	except AttributeError:
187	n/a	raise NotImplementedError
188	n/a
189	n/a	#
190	n/a	# Testcases
191	n/a	#
192	n/a
193	n/a	class _TestProcess(BaseTestCase):
194	n/a
195	n/a	ALLOWED_TYPES = ('processes', 'threads')
196	n/a
197	n/a	def test_current(self):
198	n/a	if self.TYPE == 'threads':
199	n/a	self.skipTest('test not appropriate for {}'.format(self.TYPE))
200	n/a
201	n/a	current = self.current_process()
202	n/a	authkey = current.authkey
203	n/a
204	n/a	self.assertTrue(current.is_alive())
205	n/a	self.assertTrue(not current.daemon)
206	n/a	self.assertIsInstance(authkey, bytes)
207	n/a	self.assertTrue(len(authkey) > 0)
208	n/a	self.assertEqual(current.ident, os.getpid())
209	n/a	self.assertEqual(current.exitcode, None)
210	n/a
211	n/a	def test_daemon_argument(self):
212	n/a	if self.TYPE == "threads":
213	n/a	self.skipTest('test not appropriate for {}'.format(self.TYPE))
214	n/a
215	n/a	# By default uses the current process's daemon flag.
216	n/a	proc0 = self.Process(target=self._test)
217	n/a	self.assertEqual(proc0.daemon, self.current_process().daemon)
218	n/a	proc1 = self.Process(target=self._test, daemon=True)
219	n/a	self.assertTrue(proc1.daemon)
220	n/a	proc2 = self.Process(target=self._test, daemon=False)
221	n/a	self.assertFalse(proc2.daemon)
222	n/a
223	n/a	@classmethod
224	n/a	def _test(cls, q, args, *kwds):
225	n/a	current = cls.current_process()
226	n/a	q.put(args)
227	n/a	q.put(kwds)
228	n/a	q.put(current.name)
229	n/a	if cls.TYPE != 'threads':
230	n/a	q.put(bytes(current.authkey))
231	n/a	q.put(current.pid)
232	n/a
233	n/a	def test_process(self):
234	n/a	q = self.Queue(1)
235	n/a	e = self.Event()
236	n/a	args = (q, 1, 2)
237	n/a	kwargs = {'hello':23, 'bye':2.54}
238	n/a	name = 'SomeProcess'
239	n/a	p = self.Process(
240	n/a	target=self._test, args=args, kwargs=kwargs, name=name
241	n/a	)
242	n/a	p.daemon = True
243	n/a	current = self.current_process()
244	n/a
245	n/a	if self.TYPE != 'threads':
246	n/a	self.assertEqual(p.authkey, current.authkey)
247	n/a	self.assertEqual(p.is_alive(), False)
248	n/a	self.assertEqual(p.daemon, True)
249	n/a	self.assertNotIn(p, self.active_children())
250	n/a	self.assertTrue(type(self.active_children()) is list)
251	n/a	self.assertEqual(p.exitcode, None)
252	n/a
253	n/a	p.start()
254	n/a
255	n/a	self.assertEqual(p.exitcode, None)
256	n/a	self.assertEqual(p.is_alive(), True)
257	n/a	self.assertIn(p, self.active_children())
258	n/a
259	n/a	self.assertEqual(q.get(), args[1:])
260	n/a	self.assertEqual(q.get(), kwargs)
261	n/a	self.assertEqual(q.get(), p.name)
262	n/a	if self.TYPE != 'threads':
263	n/a	self.assertEqual(q.get(), current.authkey)
264	n/a	self.assertEqual(q.get(), p.pid)
265	n/a
266	n/a	p.join()
267	n/a
268	n/a	self.assertEqual(p.exitcode, 0)
269	n/a	self.assertEqual(p.is_alive(), False)
270	n/a	self.assertNotIn(p, self.active_children())
271	n/a
272	n/a	@classmethod
273	n/a	def _test_terminate(cls):
274	n/a	time.sleep(100)
275	n/a
276	n/a	def test_terminate(self):
277	n/a	if self.TYPE == 'threads':
278	n/a	self.skipTest('test not appropriate for {}'.format(self.TYPE))
279	n/a
280	n/a	p = self.Process(target=self._test_terminate)
281	n/a	p.daemon = True
282	n/a	p.start()
283	n/a
284	n/a	self.assertEqual(p.is_alive(), True)
285	n/a	self.assertIn(p, self.active_children())
286	n/a	self.assertEqual(p.exitcode, None)
287	n/a
288	n/a	join = TimingWrapper(p.join)
289	n/a
290	n/a	self.assertEqual(join(0), None)
291	n/a	self.assertTimingAlmostEqual(join.elapsed, 0.0)
292	n/a	self.assertEqual(p.is_alive(), True)
293	n/a
294	n/a	self.assertEqual(join(-1), None)
295	n/a	self.assertTimingAlmostEqual(join.elapsed, 0.0)
296	n/a	self.assertEqual(p.is_alive(), True)
297	n/a
298	n/a	# XXX maybe terminating too soon causes the problems on Gentoo...
299	n/a	time.sleep(1)
300	n/a
301	n/a	p.terminate()
302	n/a
303	n/a	if hasattr(signal, 'alarm'):
304	n/a	# On the Gentoo buildbot waitpid() often seems to block forever.
305	n/a	# We use alarm() to interrupt it if it blocks for too long.
306	n/a	def handler(*args):
307	n/a	raise RuntimeError('join took too long: %s' % p)
308	n/a	old_handler = signal.signal(signal.SIGALRM, handler)
309	n/a	try:
310	n/a	signal.alarm(10)
311	n/a	self.assertEqual(join(), None)
312	n/a	finally:
313	n/a	signal.alarm(0)
314	n/a	signal.signal(signal.SIGALRM, old_handler)
315	n/a	else:
316	n/a	self.assertEqual(join(), None)
317	n/a
318	n/a	self.assertTimingAlmostEqual(join.elapsed, 0.0)
319	n/a
320	n/a	self.assertEqual(p.is_alive(), False)
321	n/a	self.assertNotIn(p, self.active_children())
322	n/a
323	n/a	p.join()
324	n/a
325	n/a	# XXX sometimes get p.exitcode == 0 on Windows ...
326	n/a	#self.assertEqual(p.exitcode, -signal.SIGTERM)
327	n/a
328	n/a	def test_cpu_count(self):
329	n/a	try:
330	n/a	cpus = multiprocessing.cpu_count()
331	n/a	except NotImplementedError:
332	n/a	cpus = 1
333	n/a	self.assertTrue(type(cpus) is int)
334	n/a	self.assertTrue(cpus >= 1)
335	n/a
336	n/a	def test_active_children(self):
337	n/a	self.assertEqual(type(self.active_children()), list)
338	n/a
339	n/a	p = self.Process(target=time.sleep, args=(DELTA,))
340	n/a	self.assertNotIn(p, self.active_children())
341	n/a
342	n/a	p.daemon = True
343	n/a	p.start()
344	n/a	self.assertIn(p, self.active_children())
345	n/a
346	n/a	p.join()
347	n/a	self.assertNotIn(p, self.active_children())
348	n/a
349	n/a	@classmethod
350	n/a	def _test_recursion(cls, wconn, id):
351	n/a	wconn.send(id)
352	n/a	if len(id) < 2:
353	n/a	for i in range(2):
354	n/a	p = cls.Process(
355	n/a	target=cls._test_recursion, args=(wconn, id+[i])
356	n/a	)
357	n/a	p.start()
358	n/a	p.join()
359	n/a
360	n/a	def test_recursion(self):
361	n/a	rconn, wconn = self.Pipe(duplex=False)
362	n/a	self._test_recursion(wconn, [])
363	n/a
364	n/a	time.sleep(DELTA)
365	n/a	result = []
366	n/a	while rconn.poll():
367	n/a	result.append(rconn.recv())
368	n/a
369	n/a	expected = [
370	n/a	[],
371	n/a	[0],
372	n/a	[0, 0],
373	n/a	[0, 1],
374	n/a	[1],
375	n/a	[1, 0],
376	n/a	[1, 1]
377	n/a	]
378	n/a	self.assertEqual(result, expected)
379	n/a
380	n/a	@classmethod
381	n/a	def _test_sentinel(cls, event):
382	n/a	event.wait(10.0)
383	n/a
384	n/a	def test_sentinel(self):
385	n/a	if self.TYPE == "threads":
386	n/a	self.skipTest('test not appropriate for {}'.format(self.TYPE))
387	n/a	event = self.Event()
388	n/a	p = self.Process(target=self._test_sentinel, args=(event,))
389	n/a	with self.assertRaises(ValueError):
390	n/a	p.sentinel
391	n/a	p.start()
392	n/a	self.addCleanup(p.join)
393	n/a	sentinel = p.sentinel
394	n/a	self.assertIsInstance(sentinel, int)
395	n/a	self.assertFalse(wait_for_handle(sentinel, timeout=0.0))
396	n/a	event.set()
397	n/a	p.join()
398	n/a	self.assertTrue(wait_for_handle(sentinel, timeout=1))
399	n/a
400	n/a	#
401	n/a	#
402	n/a	#
403	n/a
404	n/a	class _UpperCaser(multiprocessing.Process):
405	n/a
406	n/a	def __init__(self):
407	n/a	multiprocessing.Process.__init__(self)
408	n/a	self.child_conn, self.parent_conn = multiprocessing.Pipe()
409	n/a
410	n/a	def run(self):
411	n/a	self.parent_conn.close()
412	n/a	for s in iter(self.child_conn.recv, None):
413	n/a	self.child_conn.send(s.upper())
414	n/a	self.child_conn.close()
415	n/a
416	n/a	def submit(self, s):
417	n/a	assert type(s) is str
418	n/a	self.parent_conn.send(s)
419	n/a	return self.parent_conn.recv()
420	n/a
421	n/a	def stop(self):
422	n/a	self.parent_conn.send(None)
423	n/a	self.parent_conn.close()
424	n/a	self.child_conn.close()
425	n/a
426	n/a	class _TestSubclassingProcess(BaseTestCase):
427	n/a
428	n/a	ALLOWED_TYPES = ('processes',)
429	n/a
430	n/a	def test_subclassing(self):
431	n/a	uppercaser = _UpperCaser()
432	n/a	uppercaser.daemon = True
433	n/a	uppercaser.start()
434	n/a	self.assertEqual(uppercaser.submit('hello'), 'HELLO')
435	n/a	self.assertEqual(uppercaser.submit('world'), 'WORLD')
436	n/a	uppercaser.stop()
437	n/a	uppercaser.join()
438	n/a
439	n/a	def test_stderr_flush(self):
440	n/a	# sys.stderr is flushed at process shutdown (issue #13812)
441	n/a	if self.TYPE == "threads":
442	n/a	self.skipTest('test not appropriate for {}'.format(self.TYPE))
443	n/a
444	n/a	testfn = test.support.TESTFN
445	n/a	self.addCleanup(test.support.unlink, testfn)
446	n/a	proc = self.Process(target=self._test_stderr_flush, args=(testfn,))
447	n/a	proc.start()
448	n/a	proc.join()
449	n/a	with open(testfn, 'r') as f:
450	n/a	err = f.read()
451	n/a	# The whole traceback was printed
452	n/a	self.assertIn("ZeroDivisionError", err)
453	n/a	self.assertIn("test_multiprocessing.py", err)
454	n/a	self.assertIn("1/0 # MARKER", err)
455	n/a
456	n/a	@classmethod
457	n/a	def _test_stderr_flush(cls, testfn):
458	n/a	fd = os.open(testfn, os.O_WRONLY \| os.O_CREAT \| os.O_EXCL)
459	n/a	sys.stderr = open(fd, 'w', closefd=False)
460	n/a	1/0 # MARKER
461	n/a
462	n/a
463	n/a	@classmethod
464	n/a	def _test_sys_exit(cls, reason, testfn):
465	n/a	fd = os.open(testfn, os.O_WRONLY \| os.O_CREAT \| os.O_EXCL)
466	n/a	sys.stderr = open(fd, 'w', closefd=False)
467	n/a	sys.exit(reason)
468	n/a
469	n/a	def test_sys_exit(self):
470	n/a	# See Issue 13854
471	n/a	if self.TYPE == 'threads':
472	n/a	self.skipTest('test not appropriate for {}'.format(self.TYPE))
473	n/a
474	n/a	testfn = test.support.TESTFN
475	n/a	self.addCleanup(test.support.unlink, testfn)
476	n/a
477	n/a	for reason in (
478	n/a	[1, 2, 3],
479	n/a	'ignore this',
480	n/a	):
481	n/a	p = self.Process(target=self._test_sys_exit, args=(reason, testfn))
482	n/a	p.daemon = True
483	n/a	p.start()
484	n/a	p.join(5)
485	n/a	self.assertEqual(p.exitcode, 1)
486	n/a
487	n/a	with open(testfn, 'r') as f:
488	n/a	content = f.read()
489	n/a	self.assertEqual(content.rstrip(), str(reason))
490	n/a
491	n/a	os.unlink(testfn)
492	n/a
493	n/a	for reason in (True, False, 8):
494	n/a	p = self.Process(target=sys.exit, args=(reason,))
495	n/a	p.daemon = True
496	n/a	p.start()
497	n/a	p.join(5)
498	n/a	self.assertEqual(p.exitcode, reason)
499	n/a
500	n/a	#
501	n/a	#
502	n/a	#
503	n/a
504	n/a	def queue_empty(q):
505	n/a	if hasattr(q, 'empty'):
506	n/a	return q.empty()
507	n/a	else:
508	n/a	return q.qsize() == 0
509	n/a
510	n/a	def queue_full(q, maxsize):
511	n/a	if hasattr(q, 'full'):
512	n/a	return q.full()
513	n/a	else:
514	n/a	return q.qsize() == maxsize
515	n/a
516	n/a
517	n/a	class _TestQueue(BaseTestCase):
518	n/a
519	n/a
520	n/a	@classmethod
521	n/a	def _test_put(cls, queue, child_can_start, parent_can_continue):
522	n/a	child_can_start.wait()
523	n/a	for i in range(6):
524	n/a	queue.get()
525	n/a	parent_can_continue.set()
526	n/a
527	n/a	def test_put(self):
528	n/a	MAXSIZE = 6
529	n/a	queue = self.Queue(maxsize=MAXSIZE)
530	n/a	child_can_start = self.Event()
531	n/a	parent_can_continue = self.Event()
532	n/a
533	n/a	proc = self.Process(
534	n/a	target=self._test_put,
535	n/a	args=(queue, child_can_start, parent_can_continue)
536	n/a	)
537	n/a	proc.daemon = True
538	n/a	proc.start()
539	n/a
540	n/a	self.assertEqual(queue_empty(queue), True)
541	n/a	self.assertEqual(queue_full(queue, MAXSIZE), False)
542	n/a
543	n/a	queue.put(1)
544	n/a	queue.put(2, True)
545	n/a	queue.put(3, True, None)
546	n/a	queue.put(4, False)
547	n/a	queue.put(5, False, None)
548	n/a	queue.put_nowait(6)
549	n/a
550	n/a	# the values may be in buffer but not yet in pipe so sleep a bit
551	n/a	time.sleep(DELTA)
552	n/a
553	n/a	self.assertEqual(queue_empty(queue), False)
554	n/a	self.assertEqual(queue_full(queue, MAXSIZE), True)
555	n/a
556	n/a	put = TimingWrapper(queue.put)
557	n/a	put_nowait = TimingWrapper(queue.put_nowait)
558	n/a
559	n/a	self.assertRaises(pyqueue.Full, put, 7, False)
560	n/a	self.assertTimingAlmostEqual(put.elapsed, 0)
561	n/a
562	n/a	self.assertRaises(pyqueue.Full, put, 7, False, None)
563	n/a	self.assertTimingAlmostEqual(put.elapsed, 0)
564	n/a
565	n/a	self.assertRaises(pyqueue.Full, put_nowait, 7)
566	n/a	self.assertTimingAlmostEqual(put_nowait.elapsed, 0)
567	n/a
568	n/a	self.assertRaises(pyqueue.Full, put, 7, True, TIMEOUT1)
569	n/a	self.assertTimingAlmostEqual(put.elapsed, TIMEOUT1)
570	n/a
571	n/a	self.assertRaises(pyqueue.Full, put, 7, False, TIMEOUT2)
572	n/a	self.assertTimingAlmostEqual(put.elapsed, 0)
573	n/a
574	n/a	self.assertRaises(pyqueue.Full, put, 7, True, timeout=TIMEOUT3)
575	n/a	self.assertTimingAlmostEqual(put.elapsed, TIMEOUT3)
576	n/a
577	n/a	child_can_start.set()
578	n/a	parent_can_continue.wait()
579	n/a
580	n/a	self.assertEqual(queue_empty(queue), True)
581	n/a	self.assertEqual(queue_full(queue, MAXSIZE), False)
582	n/a
583	n/a	proc.join()
584	n/a
585	n/a	@classmethod
586	n/a	def _test_get(cls, queue, child_can_start, parent_can_continue):
587	n/a	child_can_start.wait()
588	n/a	#queue.put(1)
589	n/a	queue.put(2)
590	n/a	queue.put(3)
591	n/a	queue.put(4)
592	n/a	queue.put(5)
593	n/a	parent_can_continue.set()
594	n/a
595	n/a	def test_get(self):
596	n/a	queue = self.Queue()
597	n/a	child_can_start = self.Event()
598	n/a	parent_can_continue = self.Event()
599	n/a
600	n/a	proc = self.Process(
601	n/a	target=self._test_get,
602	n/a	args=(queue, child_can_start, parent_can_continue)
603	n/a	)
604	n/a	proc.daemon = True
605	n/a	proc.start()
606	n/a
607	n/a	self.assertEqual(queue_empty(queue), True)
608	n/a
609	n/a	child_can_start.set()
610	n/a	parent_can_continue.wait()
611	n/a
612	n/a	time.sleep(DELTA)
613	n/a	self.assertEqual(queue_empty(queue), False)
614	n/a
615	n/a	# Hangs unexpectedly, remove for now
616	n/a	#self.assertEqual(queue.get(), 1)
617	n/a	self.assertEqual(queue.get(True, None), 2)
618	n/a	self.assertEqual(queue.get(True), 3)
619	n/a	self.assertEqual(queue.get(timeout=1), 4)
620	n/a	self.assertEqual(queue.get_nowait(), 5)
621	n/a
622	n/a	self.assertEqual(queue_empty(queue), True)
623	n/a
624	n/a	get = TimingWrapper(queue.get)
625	n/a	get_nowait = TimingWrapper(queue.get_nowait)
626	n/a
627	n/a	self.assertRaises(pyqueue.Empty, get, False)
628	n/a	self.assertTimingAlmostEqual(get.elapsed, 0)
629	n/a
630	n/a	self.assertRaises(pyqueue.Empty, get, False, None)
631	n/a	self.assertTimingAlmostEqual(get.elapsed, 0)
632	n/a
633	n/a	self.assertRaises(pyqueue.Empty, get_nowait)
634	n/a	self.assertTimingAlmostEqual(get_nowait.elapsed, 0)
635	n/a
636	n/a	self.assertRaises(pyqueue.Empty, get, True, TIMEOUT1)
637	n/a	self.assertTimingAlmostEqual(get.elapsed, TIMEOUT1)
638	n/a
639	n/a	self.assertRaises(pyqueue.Empty, get, False, TIMEOUT2)
640	n/a	self.assertTimingAlmostEqual(get.elapsed, 0)
641	n/a
642	n/a	self.assertRaises(pyqueue.Empty, get, timeout=TIMEOUT3)
643	n/a	self.assertTimingAlmostEqual(get.elapsed, TIMEOUT3)
644	n/a
645	n/a	proc.join()
646	n/a
647	n/a	@classmethod
648	n/a	def _test_fork(cls, queue):
649	n/a	for i in range(10, 20):
650	n/a	queue.put(i)
651	n/a	# note that at this point the items may only be buffered, so the
652	n/a	# process cannot shutdown until the feeder thread has finished
653	n/a	# pushing items onto the pipe.
654	n/a
655	n/a	def test_fork(self):
656	n/a	# Old versions of Queue would fail to create a new feeder
657	n/a	# thread for a forked process if the original process had its
658	n/a	# own feeder thread. This test checks that this no longer
659	n/a	# happens.
660	n/a
661	n/a	queue = self.Queue()
662	n/a
663	n/a	# put items on queue so that main process starts a feeder thread
664	n/a	for i in range(10):
665	n/a	queue.put(i)
666	n/a
667	n/a	# wait to make sure thread starts before we fork a new process
668	n/a	time.sleep(DELTA)
669	n/a
670	n/a	# fork process
671	n/a	p = self.Process(target=self._test_fork, args=(queue,))
672	n/a	p.daemon = True
673	n/a	p.start()
674	n/a
675	n/a	# check that all expected items are in the queue
676	n/a	for i in range(20):
677	n/a	self.assertEqual(queue.get(), i)
678	n/a	self.assertRaises(pyqueue.Empty, queue.get, False)
679	n/a
680	n/a	p.join()
681	n/a
682	n/a	def test_qsize(self):
683	n/a	q = self.Queue()
684	n/a	try:
685	n/a	self.assertEqual(q.qsize(), 0)
686	n/a	except NotImplementedError:
687	n/a	self.skipTest('qsize method not implemented')
688	n/a	q.put(1)
689	n/a	self.assertEqual(q.qsize(), 1)
690	n/a	q.put(5)
691	n/a	self.assertEqual(q.qsize(), 2)
692	n/a	q.get()
693	n/a	self.assertEqual(q.qsize(), 1)
694	n/a	q.get()
695	n/a	self.assertEqual(q.qsize(), 0)
696	n/a
697	n/a	@classmethod
698	n/a	def _test_task_done(cls, q):
699	n/a	for obj in iter(q.get, None):
700	n/a	time.sleep(DELTA)
701	n/a	q.task_done()
702	n/a
703	n/a	def test_task_done(self):
704	n/a	queue = self.JoinableQueue()
705	n/a
706	n/a	workers = [self.Process(target=self._test_task_done, args=(queue,))
707	n/a	for i in range(4)]
708	n/a
709	n/a	for p in workers:
710	n/a	p.daemon = True
711	n/a	p.start()
712	n/a
713	n/a	for i in range(10):
714	n/a	queue.put(i)
715	n/a
716	n/a	queue.join()
717	n/a
718	n/a	for p in workers:
719	n/a	queue.put(None)
720	n/a
721	n/a	for p in workers:
722	n/a	p.join()
723	n/a
724	n/a	def test_no_import_lock_contention(self):
725	n/a	with test.support.temp_cwd():
726	n/a	module_name = 'imported_by_an_imported_module'
727	n/a	with open(module_name + '.py', 'w') as f:
728	n/a	f.write("""if 1:
729	n/a	import multiprocessing
730	n/a
731	n/a	q = multiprocessing.Queue()
732	n/a	q.put('knock knock')
733	n/a	q.get(timeout=3)
734	n/a	q.close()
735	n/a	del q
736	n/a	""")
737	n/a
738	n/a	with test.support.DirsOnSysPath(os.getcwd()):
739	n/a	try:
740	n/a	__import__(module_name)
741	n/a	except pyqueue.Empty:
742	n/a	self.fail("Probable regression on import lock contention;"
743	n/a	" see Issue #22853")
744	n/a
745	n/a	def test_timeout(self):
746	n/a	q = multiprocessing.Queue()
747	n/a	start = time.time()
748	n/a	self.assertRaises(pyqueue.Empty, q.get, True, 0.200)
749	n/a	delta = time.time() - start
750	n/a	# Tolerate a delta of 30 ms because of the bad clock resolution on
751	n/a	# Windows (usually 15.6 ms)
752	n/a	self.assertGreaterEqual(delta, 0.170)
753	n/a
754	n/a	#
755	n/a	#
756	n/a	#
757	n/a
758	n/a	class _TestLock(BaseTestCase):
759	n/a
760	n/a	def test_lock(self):
761	n/a	lock = self.Lock()
762	n/a	self.assertEqual(lock.acquire(), True)
763	n/a	self.assertEqual(lock.acquire(False), False)
764	n/a	self.assertEqual(lock.release(), None)
765	n/a	self.assertRaises((ValueError, threading.ThreadError), lock.release)
766	n/a
767	n/a	def test_rlock(self):
768	n/a	lock = self.RLock()
769	n/a	self.assertEqual(lock.acquire(), True)
770	n/a	self.assertEqual(lock.acquire(), True)
771	n/a	self.assertEqual(lock.acquire(), True)
772	n/a	self.assertEqual(lock.release(), None)
773	n/a	self.assertEqual(lock.release(), None)
774	n/a	self.assertEqual(lock.release(), None)
775	n/a	self.assertRaises((AssertionError, RuntimeError), lock.release)
776	n/a
777	n/a	def test_lock_context(self):
778	n/a	with self.Lock():
779	n/a	pass
780	n/a
781	n/a
782	n/a	class _TestSemaphore(BaseTestCase):
783	n/a
784	n/a	def _test_semaphore(self, sem):
785	n/a	self.assertReturnsIfImplemented(2, get_value, sem)
786	n/a	self.assertEqual(sem.acquire(), True)
787	n/a	self.assertReturnsIfImplemented(1, get_value, sem)
788	n/a	self.assertEqual(sem.acquire(), True)
789	n/a	self.assertReturnsIfImplemented(0, get_value, sem)
790	n/a	self.assertEqual(sem.acquire(False), False)
791	n/a	self.assertReturnsIfImplemented(0, get_value, sem)
792	n/a	self.assertEqual(sem.release(), None)
793	n/a	self.assertReturnsIfImplemented(1, get_value, sem)
794	n/a	self.assertEqual(sem.release(), None)
795	n/a	self.assertReturnsIfImplemented(2, get_value, sem)
796	n/a
797	n/a	def test_semaphore(self):
798	n/a	sem = self.Semaphore(2)
799	n/a	self._test_semaphore(sem)
800	n/a	self.assertEqual(sem.release(), None)
801	n/a	self.assertReturnsIfImplemented(3, get_value, sem)
802	n/a	self.assertEqual(sem.release(), None)
803	n/a	self.assertReturnsIfImplemented(4, get_value, sem)
804	n/a
805	n/a	def test_bounded_semaphore(self):
806	n/a	sem = self.BoundedSemaphore(2)
807	n/a	self._test_semaphore(sem)
808	n/a	# Currently fails on OS/X
809	n/a	#if HAVE_GETVALUE:
810	n/a	# self.assertRaises(ValueError, sem.release)
811	n/a	# self.assertReturnsIfImplemented(2, get_value, sem)
812	n/a
813	n/a	def test_timeout(self):
814	n/a	if self.TYPE != 'processes':
815	n/a	self.skipTest('test not appropriate for {}'.format(self.TYPE))
816	n/a
817	n/a	sem = self.Semaphore(0)
818	n/a	acquire = TimingWrapper(sem.acquire)
819	n/a
820	n/a	self.assertEqual(acquire(False), False)
821	n/a	self.assertTimingAlmostEqual(acquire.elapsed, 0.0)
822	n/a
823	n/a	self.assertEqual(acquire(False, None), False)
824	n/a	self.assertTimingAlmostEqual(acquire.elapsed, 0.0)
825	n/a
826	n/a	self.assertEqual(acquire(False, TIMEOUT1), False)
827	n/a	self.assertTimingAlmostEqual(acquire.elapsed, 0)
828	n/a
829	n/a	self.assertEqual(acquire(True, TIMEOUT2), False)
830	n/a	self.assertTimingAlmostEqual(acquire.elapsed, TIMEOUT2)
831	n/a
832	n/a	self.assertEqual(acquire(timeout=TIMEOUT3), False)
833	n/a	self.assertTimingAlmostEqual(acquire.elapsed, TIMEOUT3)
834	n/a
835	n/a
836	n/a	class _TestCondition(BaseTestCase):
837	n/a
838	n/a	@classmethod
839	n/a	def f(cls, cond, sleeping, woken, timeout=None):
840	n/a	cond.acquire()
841	n/a	sleeping.release()
842	n/a	cond.wait(timeout)
843	n/a	woken.release()
844	n/a	cond.release()
845	n/a
846	n/a	def check_invariant(self, cond):
847	n/a	# this is only supposed to succeed when there are no sleepers
848	n/a	if self.TYPE == 'processes':
849	n/a	try:
850	n/a	sleepers = (cond._sleeping_count.get_value() -
851	n/a	cond._woken_count.get_value())
852	n/a	self.assertEqual(sleepers, 0)
853	n/a	self.assertEqual(cond._wait_semaphore.get_value(), 0)
854	n/a	except NotImplementedError:
855	n/a	pass
856	n/a
857	n/a	def test_notify(self):
858	n/a	cond = self.Condition()
859	n/a	sleeping = self.Semaphore(0)
860	n/a	woken = self.Semaphore(0)
861	n/a
862	n/a	p = self.Process(target=self.f, args=(cond, sleeping, woken))
863	n/a	p.daemon = True
864	n/a	p.start()
865	n/a
866	n/a	p = threading.Thread(target=self.f, args=(cond, sleeping, woken))
867	n/a	p.daemon = True
868	n/a	p.start()
869	n/a
870	n/a	# wait for both children to start sleeping
871	n/a	sleeping.acquire()
872	n/a	sleeping.acquire()
873	n/a
874	n/a	# check no process/thread has woken up
875	n/a	time.sleep(DELTA)
876	n/a	self.assertReturnsIfImplemented(0, get_value, woken)
877	n/a
878	n/a	# wake up one process/thread
879	n/a	cond.acquire()
880	n/a	cond.notify()
881	n/a	cond.release()
882	n/a
883	n/a	# check one process/thread has woken up
884	n/a	time.sleep(DELTA)
885	n/a	self.assertReturnsIfImplemented(1, get_value, woken)
886	n/a
887	n/a	# wake up another
888	n/a	cond.acquire()
889	n/a	cond.notify()
890	n/a	cond.release()
891	n/a
892	n/a	# check other has woken up
893	n/a	time.sleep(DELTA)
894	n/a	self.assertReturnsIfImplemented(2, get_value, woken)
895	n/a
896	n/a	# check state is not mucked up
897	n/a	self.check_invariant(cond)
898	n/a	p.join()
899	n/a
900	n/a	def test_notify_all(self):
901	n/a	cond = self.Condition()
902	n/a	sleeping = self.Semaphore(0)
903	n/a	woken = self.Semaphore(0)
904	n/a
905	n/a	# start some threads/processes which will timeout
906	n/a	for i in range(3):
907	n/a	p = self.Process(target=self.f,
908	n/a	args=(cond, sleeping, woken, TIMEOUT1))
909	n/a	p.daemon = True
910	n/a	p.start()
911	n/a
912	n/a	t = threading.Thread(target=self.f,
913	n/a	args=(cond, sleeping, woken, TIMEOUT1))
914	n/a	t.daemon = True
915	n/a	t.start()
916	n/a
917	n/a	# wait for them all to sleep
918	n/a	for i in range(6):
919	n/a	sleeping.acquire()
920	n/a
921	n/a	# check they have all timed out
922	n/a	for i in range(6):
923	n/a	woken.acquire()
924	n/a	self.assertReturnsIfImplemented(0, get_value, woken)
925	n/a
926	n/a	# check state is not mucked up
927	n/a	self.check_invariant(cond)
928	n/a
929	n/a	# start some more threads/processes
930	n/a	for i in range(3):
931	n/a	p = self.Process(target=self.f, args=(cond, sleeping, woken))
932	n/a	p.daemon = True
933	n/a	p.start()
934	n/a
935	n/a	t = threading.Thread(target=self.f, args=(cond, sleeping, woken))
936	n/a	t.daemon = True
937	n/a	t.start()
938	n/a
939	n/a	# wait for them to all sleep
940	n/a	for i in range(6):
941	n/a	sleeping.acquire()
942	n/a
943	n/a	# check no process/thread has woken up
944	n/a	time.sleep(DELTA)
945	n/a	self.assertReturnsIfImplemented(0, get_value, woken)
946	n/a
947	n/a	# wake them all up
948	n/a	cond.acquire()
949	n/a	cond.notify_all()
950	n/a	cond.release()
951	n/a
952	n/a	# check they have all woken
953	n/a	for i in range(10):
954	n/a	try:
955	n/a	if get_value(woken) == 6:
956	n/a	break
957	n/a	except NotImplementedError:
958	n/a	break
959	n/a	time.sleep(DELTA)
960	n/a	self.assertReturnsIfImplemented(6, get_value, woken)
961	n/a
962	n/a	# check state is not mucked up
963	n/a	self.check_invariant(cond)
964	n/a
965	n/a	def test_timeout(self):
966	n/a	cond = self.Condition()
967	n/a	wait = TimingWrapper(cond.wait)
968	n/a	cond.acquire()
969	n/a	res = wait(TIMEOUT1)
970	n/a	cond.release()
971	n/a	self.assertEqual(res, False)
972	n/a	self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1)
973	n/a
974	n/a	@classmethod
975	n/a	def _test_waitfor_f(cls, cond, state):
976	n/a	with cond:
977	n/a	state.value = 0
978	n/a	cond.notify()
979	n/a	result = cond.wait_for(lambda : state.value==4)
980	n/a	if not result or state.value != 4:
981	n/a	sys.exit(1)
982	n/a
983	n/a	@unittest.skipUnless(HAS_SHAREDCTYPES, 'needs sharedctypes')
984	n/a	def test_waitfor(self):
985	n/a	# based on test in test/lock_tests.py
986	n/a	cond = self.Condition()
987	n/a	state = self.Value('i', -1)
988	n/a
989	n/a	p = self.Process(target=self._test_waitfor_f, args=(cond, state))
990	n/a	p.daemon = True
991	n/a	p.start()
992	n/a
993	n/a	with cond:
994	n/a	result = cond.wait_for(lambda : state.value==0)
995	n/a	self.assertTrue(result)
996	n/a	self.assertEqual(state.value, 0)
997	n/a
998	n/a	for i in range(4):
999	n/a	time.sleep(0.01)
1000	n/a	with cond:
1001	n/a	state.value += 1
1002	n/a	cond.notify()
1003	n/a
1004	n/a	p.join(5)
1005	n/a	self.assertFalse(p.is_alive())
1006	n/a	self.assertEqual(p.exitcode, 0)
1007	n/a
1008	n/a	@classmethod
1009	n/a	def _test_waitfor_timeout_f(cls, cond, state, success, sem):
1010	n/a	sem.release()
1011	n/a	with cond:
1012	n/a	expected = 0.1
1013	n/a	dt = time.time()
1014	n/a	result = cond.wait_for(lambda : state.value==4, timeout=expected)
1015	n/a	dt = time.time() - dt
1016	n/a	# borrow logic in assertTimeout() from test/lock_tests.py
1017	n/a	if not result and expected * 0.6 < dt < expected * 10.0:
1018	n/a	success.value = True
1019	n/a
1020	n/a	@unittest.skipUnless(HAS_SHAREDCTYPES, 'needs sharedctypes')
1021	n/a	def test_waitfor_timeout(self):
1022	n/a	# based on test in test/lock_tests.py
1023	n/a	cond = self.Condition()
1024	n/a	state = self.Value('i', 0)
1025	n/a	success = self.Value('i', False)
1026	n/a	sem = self.Semaphore(0)
1027	n/a
1028	n/a	p = self.Process(target=self._test_waitfor_timeout_f,
1029	n/a	args=(cond, state, success, sem))
1030	n/a	p.daemon = True
1031	n/a	p.start()
1032	n/a	self.assertTrue(sem.acquire(timeout=10))
1033	n/a
1034	n/a	# Only increment 3 times, so state == 4 is never reached.
1035	n/a	for i in range(3):
1036	n/a	time.sleep(0.01)
1037	n/a	with cond:
1038	n/a	state.value += 1
1039	n/a	cond.notify()
1040	n/a
1041	n/a	p.join(5)
1042	n/a	self.assertTrue(success.value)
1043	n/a
1044	n/a	@classmethod
1045	n/a	def _test_wait_result(cls, c, pid):
1046	n/a	with c:
1047	n/a	c.notify()
1048	n/a	time.sleep(1)
1049	n/a	if pid is not None:
1050	n/a	os.kill(pid, signal.SIGINT)
1051	n/a
1052	n/a	def test_wait_result(self):
1053	n/a	if isinstance(self, ProcessesMixin) and sys.platform != 'win32':
1054	n/a	pid = os.getpid()
1055	n/a	else:
1056	n/a	pid = None
1057	n/a
1058	n/a	c = self.Condition()
1059	n/a	with c:
1060	n/a	self.assertFalse(c.wait(0))
1061	n/a	self.assertFalse(c.wait(0.1))
1062	n/a
1063	n/a	p = self.Process(target=self._test_wait_result, args=(c, pid))
1064	n/a	p.start()
1065	n/a
1066	n/a	self.assertTrue(c.wait(10))
1067	n/a	if pid is not None:
1068	n/a	self.assertRaises(KeyboardInterrupt, c.wait, 10)
1069	n/a
1070	n/a	p.join()
1071	n/a
1072	n/a
1073	n/a	class _TestEvent(BaseTestCase):
1074	n/a
1075	n/a	@classmethod
1076	n/a	def _test_event(cls, event):
1077	n/a	time.sleep(TIMEOUT2)
1078	n/a	event.set()
1079	n/a
1080	n/a	def test_event(self):
1081	n/a	event = self.Event()
1082	n/a	wait = TimingWrapper(event.wait)
1083	n/a
1084	n/a	# Removed temporarily, due to API shear, this does not
1085	n/a	# work with threading._Event objects. is_set == isSet
1086	n/a	self.assertEqual(event.is_set(), False)
1087	n/a
1088	n/a	# Removed, threading.Event.wait() will return the value of the __flag
1089	n/a	# instead of None. API Shear with the semaphore backed mp.Event
1090	n/a	self.assertEqual(wait(0.0), False)
1091	n/a	self.assertTimingAlmostEqual(wait.elapsed, 0.0)
1092	n/a	self.assertEqual(wait(TIMEOUT1), False)
1093	n/a	self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1)
1094	n/a
1095	n/a	event.set()
1096	n/a
1097	n/a	# See note above on the API differences
1098	n/a	self.assertEqual(event.is_set(), True)
1099	n/a	self.assertEqual(wait(), True)
1100	n/a	self.assertTimingAlmostEqual(wait.elapsed, 0.0)
1101	n/a	self.assertEqual(wait(TIMEOUT1), True)
1102	n/a	self.assertTimingAlmostEqual(wait.elapsed, 0.0)
1103	n/a	# self.assertEqual(event.is_set(), True)
1104	n/a
1105	n/a	event.clear()
1106	n/a
1107	n/a	#self.assertEqual(event.is_set(), False)
1108	n/a
1109	n/a	p = self.Process(target=self._test_event, args=(event,))
1110	n/a	p.daemon = True
1111	n/a	p.start()
1112	n/a	self.assertEqual(wait(), True)
1113	n/a
1114	n/a	#
1115	n/a	# Tests for Barrier - adapted from tests in test/lock_tests.py
1116	n/a	#
1117	n/a
1118	n/a	# Many of the tests for threading.Barrier use a list as an atomic
1119	n/a	# counter: a value is appended to increment the counter, and the
1120	n/a	# length of the list gives the value. We use the class DummyList
1121	n/a	# for the same purpose.
1122	n/a
1123	n/a	class _DummyList(object):
1124	n/a
1125	n/a	def __init__(self):
1126	n/a	wrapper = multiprocessing.heap.BufferWrapper(struct.calcsize('i'))
1127	n/a	lock = multiprocessing.Lock()
1128	n/a	self.__setstate__((wrapper, lock))
1129	n/a	self._lengthbuf[0] = 0
1130	n/a
1131	n/a	def __setstate__(self, state):
1132	n/a	(self._wrapper, self._lock) = state
1133	n/a	self._lengthbuf = self._wrapper.create_memoryview().cast('i')
1134	n/a
1135	n/a	def __getstate__(self):
1136	n/a	return (self._wrapper, self._lock)
1137	n/a
1138	n/a	def append(self, _):
1139	n/a	with self._lock:
1140	n/a	self._lengthbuf[0] += 1
1141	n/a
1142	n/a	def __len__(self):
1143	n/a	with self._lock:
1144	n/a	return self._lengthbuf[0]
1145	n/a
1146	n/a	def _wait():
1147	n/a	# A crude wait/yield function not relying on synchronization primitives.
1148	n/a	time.sleep(0.01)
1149	n/a
1150	n/a
1151	n/a	class Bunch(object):
1152	n/a	"""
1153	n/a	A bunch of threads.
1154	n/a	"""
1155	n/a	def __init__(self, namespace, f, args, n, wait_before_exit=False):
1156	n/a	"""
1157	n/a	Construct a bunch of `n` threads running the same function `f`.
1158	n/a	If `wait_before_exit` is True, the threads won't terminate until
1159	n/a	do_finish() is called.
1160	n/a	"""
1161	n/a	self.f = f
1162	n/a	self.args = args
1163	n/a	self.n = n
1164	n/a	self.started = namespace.DummyList()
1165	n/a	self.finished = namespace.DummyList()
1166	n/a	self._can_exit = namespace.Event()
1167	n/a	if not wait_before_exit:
1168	n/a	self._can_exit.set()
1169	n/a	for i in range(n):
1170	n/a	p = namespace.Process(target=self.task)
1171	n/a	p.daemon = True
1172	n/a	p.start()
1173	n/a
1174	n/a	def task(self):
1175	n/a	pid = os.getpid()
1176	n/a	self.started.append(pid)
1177	n/a	try:
1178	n/a	self.f(*self.args)
1179	n/a	finally:
1180	n/a	self.finished.append(pid)
1181	n/a	self._can_exit.wait(30)
1182	n/a	assert self._can_exit.is_set()
1183	n/a
1184	n/a	def wait_for_started(self):
1185	n/a	while len(self.started) < self.n:
1186	n/a	_wait()
1187	n/a
1188	n/a	def wait_for_finished(self):
1189	n/a	while len(self.finished) < self.n:
1190	n/a	_wait()
1191	n/a
1192	n/a	def do_finish(self):
1193	n/a	self._can_exit.set()
1194	n/a
1195	n/a
1196	n/a	class AppendTrue(object):
1197	n/a	def __init__(self, obj):
1198	n/a	self.obj = obj
1199	n/a	def __call__(self):
1200	n/a	self.obj.append(True)
1201	n/a
1202	n/a
1203	n/a	class _TestBarrier(BaseTestCase):
1204	n/a	"""
1205	n/a	Tests for Barrier objects.
1206	n/a	"""
1207	n/a	N = 5
1208	n/a	defaultTimeout = 30.0 # XXX Slow Windows buildbots need generous timeout
1209	n/a
1210	n/a	def setUp(self):
1211	n/a	self.barrier = self.Barrier(self.N, timeout=self.defaultTimeout)
1212	n/a
1213	n/a	def tearDown(self):
1214	n/a	self.barrier.abort()
1215	n/a	self.barrier = None
1216	n/a
1217	n/a	def DummyList(self):
1218	n/a	if self.TYPE == 'threads':
1219	n/a	return []
1220	n/a	elif self.TYPE == 'manager':
1221	n/a	return self.manager.list()
1222	n/a	else:
1223	n/a	return _DummyList()
1224	n/a
1225	n/a	def run_threads(self, f, args):
1226	n/a	b = Bunch(self, f, args, self.N-1)
1227	n/a	f(*args)
1228	n/a	b.wait_for_finished()
1229	n/a
1230	n/a	@classmethod
1231	n/a	def multipass(cls, barrier, results, n):
1232	n/a	m = barrier.parties
1233	n/a	assert m == cls.N
1234	n/a	for i in range(n):
1235	n/a	results[0].append(True)
1236	n/a	assert len(results[1]) == i * m
1237	n/a	barrier.wait()
1238	n/a	results[1].append(True)
1239	n/a	assert len(results[0]) == (i + 1) * m
1240	n/a	barrier.wait()
1241	n/a	try:
1242	n/a	assert barrier.n_waiting == 0
1243	n/a	except NotImplementedError:
1244	n/a	pass
1245	n/a	assert not barrier.broken
1246	n/a
1247	n/a	def test_barrier(self, passes=1):
1248	n/a	"""
1249	n/a	Test that a barrier is passed in lockstep
1250	n/a	"""
1251	n/a	results = [self.DummyList(), self.DummyList()]
1252	n/a	self.run_threads(self.multipass, (self.barrier, results, passes))
1253	n/a
1254	n/a	def test_barrier_10(self):
1255	n/a	"""
1256	n/a	Test that a barrier works for 10 consecutive runs
1257	n/a	"""
1258	n/a	return self.test_barrier(10)
1259	n/a
1260	n/a	@classmethod
1261	n/a	def _test_wait_return_f(cls, barrier, queue):
1262	n/a	res = barrier.wait()
1263	n/a	queue.put(res)
1264	n/a
1265	n/a	def test_wait_return(self):
1266	n/a	"""
1267	n/a	test the return value from barrier.wait
1268	n/a	"""
1269	n/a	queue = self.Queue()
1270	n/a	self.run_threads(self._test_wait_return_f, (self.barrier, queue))
1271	n/a	results = [queue.get() for i in range(self.N)]
1272	n/a	self.assertEqual(results.count(0), 1)
1273	n/a
1274	n/a	@classmethod
1275	n/a	def _test_action_f(cls, barrier, results):
1276	n/a	barrier.wait()
1277	n/a	if len(results) != 1:
1278	n/a	raise RuntimeError
1279	n/a
1280	n/a	def test_action(self):
1281	n/a	"""
1282	n/a	Test the 'action' callback
1283	n/a	"""
1284	n/a	results = self.DummyList()
1285	n/a	barrier = self.Barrier(self.N, action=AppendTrue(results))
1286	n/a	self.run_threads(self._test_action_f, (barrier, results))
1287	n/a	self.assertEqual(len(results), 1)
1288	n/a
1289	n/a	@classmethod
1290	n/a	def _test_abort_f(cls, barrier, results1, results2):
1291	n/a	try:
1292	n/a	i = barrier.wait()
1293	n/a	if i == cls.N//2:
1294	n/a	raise RuntimeError
1295	n/a	barrier.wait()
1296	n/a	results1.append(True)
1297	n/a	except threading.BrokenBarrierError:
1298	n/a	results2.append(True)
1299	n/a	except RuntimeError:
1300	n/a	barrier.abort()
1301	n/a
1302	n/a	def test_abort(self):
1303	n/a	"""
1304	n/a	Test that an abort will put the barrier in a broken state
1305	n/a	"""
1306	n/a	results1 = self.DummyList()
1307	n/a	results2 = self.DummyList()
1308	n/a	self.run_threads(self._test_abort_f,
1309	n/a	(self.barrier, results1, results2))
1310	n/a	self.assertEqual(len(results1), 0)
1311	n/a	self.assertEqual(len(results2), self.N-1)
1312	n/a	self.assertTrue(self.barrier.broken)
1313	n/a
1314	n/a	@classmethod
1315	n/a	def _test_reset_f(cls, barrier, results1, results2, results3):
1316	n/a	i = barrier.wait()
1317	n/a	if i == cls.N//2:
1318	n/a	# Wait until the other threads are all in the barrier.
1319	n/a	while barrier.n_waiting < cls.N-1:
1320	n/a	time.sleep(0.001)
1321	n/a	barrier.reset()
1322	n/a	else:
1323	n/a	try:
1324	n/a	barrier.wait()
1325	n/a	results1.append(True)
1326	n/a	except threading.BrokenBarrierError:
1327	n/a	results2.append(True)
1328	n/a	# Now, pass the barrier again
1329	n/a	barrier.wait()
1330	n/a	results3.append(True)
1331	n/a
1332	n/a	def test_reset(self):
1333	n/a	"""
1334	n/a	Test that a 'reset' on a barrier frees the waiting threads
1335	n/a	"""
1336	n/a	results1 = self.DummyList()
1337	n/a	results2 = self.DummyList()
1338	n/a	results3 = self.DummyList()
1339	n/a	self.run_threads(self._test_reset_f,
1340	n/a	(self.barrier, results1, results2, results3))
1341	n/a	self.assertEqual(len(results1), 0)
1342	n/a	self.assertEqual(len(results2), self.N-1)
1343	n/a	self.assertEqual(len(results3), self.N)
1344	n/a
1345	n/a	@classmethod
1346	n/a	def _test_abort_and_reset_f(cls, barrier, barrier2,
1347	n/a	results1, results2, results3):
1348	n/a	try:
1349	n/a	i = barrier.wait()
1350	n/a	if i == cls.N//2:
1351	n/a	raise RuntimeError
1352	n/a	barrier.wait()
1353	n/a	results1.append(True)
1354	n/a	except threading.BrokenBarrierError:
1355	n/a	results2.append(True)
1356	n/a	except RuntimeError:
1357	n/a	barrier.abort()
1358	n/a	# Synchronize and reset the barrier. Must synchronize first so
1359	n/a	# that everyone has left it when we reset, and after so that no
1360	n/a	# one enters it before the reset.
1361	n/a	if barrier2.wait() == cls.N//2:
1362	n/a	barrier.reset()
1363	n/a	barrier2.wait()
1364	n/a	barrier.wait()
1365	n/a	results3.append(True)
1366	n/a
1367	n/a	def test_abort_and_reset(self):
1368	n/a	"""
1369	n/a	Test that a barrier can be reset after being broken.
1370	n/a	"""
1371	n/a	results1 = self.DummyList()
1372	n/a	results2 = self.DummyList()
1373	n/a	results3 = self.DummyList()
1374	n/a	barrier2 = self.Barrier(self.N)
1375	n/a
1376	n/a	self.run_threads(self._test_abort_and_reset_f,
1377	n/a	(self.barrier, barrier2, results1, results2, results3))
1378	n/a	self.assertEqual(len(results1), 0)
1379	n/a	self.assertEqual(len(results2), self.N-1)
1380	n/a	self.assertEqual(len(results3), self.N)
1381	n/a
1382	n/a	@classmethod
1383	n/a	def _test_timeout_f(cls, barrier, results):
1384	n/a	i = barrier.wait()
1385	n/a	if i == cls.N//2:
1386	n/a	# One thread is late!
1387	n/a	time.sleep(1.0)
1388	n/a	try:
1389	n/a	barrier.wait(0.5)
1390	n/a	except threading.BrokenBarrierError:
1391	n/a	results.append(True)
1392	n/a
1393	n/a	def test_timeout(self):
1394	n/a	"""
1395	n/a	Test wait(timeout)
1396	n/a	"""
1397	n/a	results = self.DummyList()
1398	n/a	self.run_threads(self._test_timeout_f, (self.barrier, results))
1399	n/a	self.assertEqual(len(results), self.barrier.parties)
1400	n/a
1401	n/a	@classmethod
1402	n/a	def _test_default_timeout_f(cls, barrier, results):
1403	n/a	i = barrier.wait(cls.defaultTimeout)
1404	n/a	if i == cls.N//2:
1405	n/a	# One thread is later than the default timeout
1406	n/a	time.sleep(1.0)
1407	n/a	try:
1408	n/a	barrier.wait()
1409	n/a	except threading.BrokenBarrierError:
1410	n/a	results.append(True)
1411	n/a
1412	n/a	def test_default_timeout(self):
1413	n/a	"""
1414	n/a	Test the barrier's default timeout
1415	n/a	"""
1416	n/a	barrier = self.Barrier(self.N, timeout=0.5)
1417	n/a	results = self.DummyList()
1418	n/a	self.run_threads(self._test_default_timeout_f, (barrier, results))
1419	n/a	self.assertEqual(len(results), barrier.parties)
1420	n/a
1421	n/a	def test_single_thread(self):
1422	n/a	b = self.Barrier(1)
1423	n/a	b.wait()
1424	n/a	b.wait()
1425	n/a
1426	n/a	@classmethod
1427	n/a	def _test_thousand_f(cls, barrier, passes, conn, lock):
1428	n/a	for i in range(passes):
1429	n/a	barrier.wait()
1430	n/a	with lock:
1431	n/a	conn.send(i)
1432	n/a
1433	n/a	def test_thousand(self):
1434	n/a	if self.TYPE == 'manager':
1435	n/a	self.skipTest('test not appropriate for {}'.format(self.TYPE))
1436	n/a	passes = 1000
1437	n/a	lock = self.Lock()
1438	n/a	conn, child_conn = self.Pipe(False)
1439	n/a	for j in range(self.N):
1440	n/a	p = self.Process(target=self._test_thousand_f,
1441	n/a	args=(self.barrier, passes, child_conn, lock))
1442	n/a	p.start()
1443	n/a
1444	n/a	for i in range(passes):
1445	n/a	for j in range(self.N):
1446	n/a	self.assertEqual(conn.recv(), i)
1447	n/a
1448	n/a	#
1449	n/a	#
1450	n/a	#
1451	n/a
1452	n/a	class _TestValue(BaseTestCase):
1453	n/a
1454	n/a	ALLOWED_TYPES = ('processes',)
1455	n/a
1456	n/a	codes_values = [
1457	n/a	('i', 4343, 24234),
1458	n/a	('d', 3.625, -4.25),
1459	n/a	('h', -232, 234),
1460	n/a	('c', latin('x'), latin('y'))
1461	n/a	]
1462	n/a
1463	n/a	def setUp(self):
1464	n/a	if not HAS_SHAREDCTYPES:
1465	n/a	self.skipTest("requires multiprocessing.sharedctypes")
1466	n/a
1467	n/a	@classmethod
1468	n/a	def _test(cls, values):
1469	n/a	for sv, cv in zip(values, cls.codes_values):
1470	n/a	sv.value = cv[2]
1471	n/a
1472	n/a
1473	n/a	def test_value(self, raw=False):
1474	n/a	if raw:
1475	n/a	values = [self.RawValue(code, value)
1476	n/a	for code, value, _ in self.codes_values]
1477	n/a	else:
1478	n/a	values = [self.Value(code, value)
1479	n/a	for code, value, _ in self.codes_values]
1480	n/a
1481	n/a	for sv, cv in zip(values, self.codes_values):
1482	n/a	self.assertEqual(sv.value, cv[1])
1483	n/a
1484	n/a	proc = self.Process(target=self._test, args=(values,))
1485	n/a	proc.daemon = True
1486	n/a	proc.start()
1487	n/a	proc.join()
1488	n/a
1489	n/a	for sv, cv in zip(values, self.codes_values):
1490	n/a	self.assertEqual(sv.value, cv[2])
1491	n/a
1492	n/a	def test_rawvalue(self):
1493	n/a	self.test_value(raw=True)
1494	n/a
1495	n/a	def test_getobj_getlock(self):
1496	n/a	val1 = self.Value('i', 5)
1497	n/a	lock1 = val1.get_lock()
1498	n/a	obj1 = val1.get_obj()
1499	n/a
1500	n/a	val2 = self.Value('i', 5, lock=None)
1501	n/a	lock2 = val2.get_lock()
1502	n/a	obj2 = val2.get_obj()
1503	n/a
1504	n/a	lock = self.Lock()
1505	n/a	val3 = self.Value('i', 5, lock=lock)
1506	n/a	lock3 = val3.get_lock()
1507	n/a	obj3 = val3.get_obj()
1508	n/a	self.assertEqual(lock, lock3)
1509	n/a
1510	n/a	arr4 = self.Value('i', 5, lock=False)
1511	n/a	self.assertFalse(hasattr(arr4, 'get_lock'))
1512	n/a	self.assertFalse(hasattr(arr4, 'get_obj'))
1513	n/a
1514	n/a	self.assertRaises(AttributeError, self.Value, 'i', 5, lock='navalue')
1515	n/a
1516	n/a	arr5 = self.RawValue('i', 5)
1517	n/a	self.assertFalse(hasattr(arr5, 'get_lock'))
1518	n/a	self.assertFalse(hasattr(arr5, 'get_obj'))
1519	n/a
1520	n/a
1521	n/a	class _TestArray(BaseTestCase):
1522	n/a
1523	n/a	ALLOWED_TYPES = ('processes',)
1524	n/a
1525	n/a	@classmethod
1526	n/a	def f(cls, seq):
1527	n/a	for i in range(1, len(seq)):
1528	n/a	seq[i] += seq[i-1]
1529	n/a
1530	n/a	@unittest.skipIf(c_int is None, "requires _ctypes")
1531	n/a	def test_array(self, raw=False):
1532	n/a	seq = [680, 626, 934, 821, 150, 233, 548, 982, 714, 831]
1533	n/a	if raw:
1534	n/a	arr = self.RawArray('i', seq)
1535	n/a	else:
1536	n/a	arr = self.Array('i', seq)
1537	n/a
1538	n/a	self.assertEqual(len(arr), len(seq))
1539	n/a	self.assertEqual(arr[3], seq[3])
1540	n/a	self.assertEqual(list(arr[2:7]), list(seq[2:7]))
1541	n/a
1542	n/a	arr[4:8] = seq[4:8] = array.array('i', [1, 2, 3, 4])
1543	n/a
1544	n/a	self.assertEqual(list(arr[:]), seq)
1545	n/a
1546	n/a	self.f(seq)
1547	n/a
1548	n/a	p = self.Process(target=self.f, args=(arr,))
1549	n/a	p.daemon = True
1550	n/a	p.start()
1551	n/a	p.join()
1552	n/a
1553	n/a	self.assertEqual(list(arr[:]), seq)
1554	n/a
1555	n/a	@unittest.skipIf(c_int is None, "requires _ctypes")
1556	n/a	def test_array_from_size(self):
1557	n/a	size = 10
1558	n/a	# Test for zeroing (see issue #11675).
1559	n/a	# The repetition below strengthens the test by increasing the chances
1560	n/a	# of previously allocated non-zero memory being used for the new array
1561	n/a	# on the 2nd and 3rd loops.
1562	n/a	for _ in range(3):
1563	n/a	arr = self.Array('i', size)
1564	n/a	self.assertEqual(len(arr), size)
1565	n/a	self.assertEqual(list(arr), [0] * size)
1566	n/a	arr[:] = range(10)
1567	n/a	self.assertEqual(list(arr), list(range(10)))
1568	n/a	del arr
1569	n/a
1570	n/a	@unittest.skipIf(c_int is None, "requires _ctypes")
1571	n/a	def test_rawarray(self):
1572	n/a	self.test_array(raw=True)
1573	n/a
1574	n/a	@unittest.skipIf(c_int is None, "requires _ctypes")
1575	n/a	def test_getobj_getlock_obj(self):
1576	n/a	arr1 = self.Array('i', list(range(10)))
1577	n/a	lock1 = arr1.get_lock()
1578	n/a	obj1 = arr1.get_obj()
1579	n/a
1580	n/a	arr2 = self.Array('i', list(range(10)), lock=None)
1581	n/a	lock2 = arr2.get_lock()
1582	n/a	obj2 = arr2.get_obj()
1583	n/a
1584	n/a	lock = self.Lock()
1585	n/a	arr3 = self.Array('i', list(range(10)), lock=lock)
1586	n/a	lock3 = arr3.get_lock()
1587	n/a	obj3 = arr3.get_obj()
1588	n/a	self.assertEqual(lock, lock3)
1589	n/a
1590	n/a	arr4 = self.Array('i', range(10), lock=False)
1591	n/a	self.assertFalse(hasattr(arr4, 'get_lock'))
1592	n/a	self.assertFalse(hasattr(arr4, 'get_obj'))
1593	n/a	self.assertRaises(AttributeError,
1594	n/a	self.Array, 'i', range(10), lock='notalock')
1595	n/a
1596	n/a	arr5 = self.RawArray('i', range(10))
1597	n/a	self.assertFalse(hasattr(arr5, 'get_lock'))
1598	n/a	self.assertFalse(hasattr(arr5, 'get_obj'))
1599	n/a
1600	n/a	#
1601	n/a	#
1602	n/a	#
1603	n/a
1604	n/a	class _TestContainers(BaseTestCase):
1605	n/a
1606	n/a	ALLOWED_TYPES = ('manager',)
1607	n/a
1608	n/a	def test_list(self):
1609	n/a	a = self.list(list(range(10)))
1610	n/a	self.assertEqual(a[:], list(range(10)))
1611	n/a
1612	n/a	b = self.list()
1613	n/a	self.assertEqual(b[:], [])
1614	n/a
1615	n/a	b.extend(list(range(5)))
1616	n/a	self.assertEqual(b[:], list(range(5)))
1617	n/a
1618	n/a	self.assertEqual(b[2], 2)
1619	n/a	self.assertEqual(b[2:10], [2,3,4])
1620	n/a
1621	n/a	b *= 2
1622	n/a	self.assertEqual(b[:], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4])
1623	n/a
1624	n/a	self.assertEqual(b + [5, 6], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4, 5, 6])
1625	n/a
1626	n/a	self.assertEqual(a[:], list(range(10)))
1627	n/a
1628	n/a	d = [a, b]
1629	n/a	e = self.list(d)
1630	n/a	self.assertEqual(
1631	n/a	[element[:] for element in e],
1632	n/a	[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4]]
1633	n/a	)
1634	n/a
1635	n/a	f = self.list([a])
1636	n/a	a.append('hello')
1637	n/a	self.assertEqual(f[0][:], [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 'hello'])
1638	n/a
1639	n/a	def test_list_proxy_in_list(self):
1640	n/a	a = self.list([self.list(range(3)) for _i in range(3)])
1641	n/a	self.assertEqual([inner[:] for inner in a], [[0, 1, 2]] * 3)
1642	n/a
1643	n/a	a[0][-1] = 55
1644	n/a	self.assertEqual(a[0][:], [0, 1, 55])
1645	n/a	for i in range(1, 3):
1646	n/a	self.assertEqual(a[i][:], [0, 1, 2])
1647	n/a
1648	n/a	self.assertEqual(a[1].pop(), 2)
1649	n/a	self.assertEqual(len(a[1]), 2)
1650	n/a	for i in range(0, 3, 2):
1651	n/a	self.assertEqual(len(a[i]), 3)
1652	n/a
1653	n/a	del a
1654	n/a
1655	n/a	b = self.list()
1656	n/a	b.append(b)
1657	n/a	del b
1658	n/a
1659	n/a	def test_dict(self):
1660	n/a	d = self.dict()
1661	n/a	indices = list(range(65, 70))
1662	n/a	for i in indices:
1663	n/a	d[i] = chr(i)
1664	n/a	self.assertEqual(d.copy(), dict((i, chr(i)) for i in indices))
1665	n/a	self.assertEqual(sorted(d.keys()), indices)
1666	n/a	self.assertEqual(sorted(d.values()), [chr(i) for i in indices])
1667	n/a	self.assertEqual(sorted(d.items()), [(i, chr(i)) for i in indices])
1668	n/a
1669	n/a	def test_dict_proxy_nested(self):
1670	n/a	pets = self.dict(ferrets=2, hamsters=4)
1671	n/a	supplies = self.dict(water=10, feed=3)
1672	n/a	d = self.dict(pets=pets, supplies=supplies)
1673	n/a
1674	n/a	self.assertEqual(supplies['water'], 10)
1675	n/a	self.assertEqual(d['supplies']['water'], 10)
1676	n/a
1677	n/a	d['supplies']['blankets'] = 5
1678	n/a	self.assertEqual(supplies['blankets'], 5)
1679	n/a	self.assertEqual(d['supplies']['blankets'], 5)
1680	n/a
1681	n/a	d['supplies']['water'] = 7
1682	n/a	self.assertEqual(supplies['water'], 7)
1683	n/a	self.assertEqual(d['supplies']['water'], 7)
1684	n/a
1685	n/a	del pets
1686	n/a	del supplies
1687	n/a	self.assertEqual(d['pets']['ferrets'], 2)
1688	n/a	d['supplies']['blankets'] = 11
1689	n/a	self.assertEqual(d['supplies']['blankets'], 11)
1690	n/a
1691	n/a	pets = d['pets']
1692	n/a	supplies = d['supplies']
1693	n/a	supplies['water'] = 7
1694	n/a	self.assertEqual(supplies['water'], 7)
1695	n/a	self.assertEqual(d['supplies']['water'], 7)
1696	n/a
1697	n/a	d.clear()
1698	n/a	self.assertEqual(len(d), 0)
1699	n/a	self.assertEqual(supplies['water'], 7)
1700	n/a	self.assertEqual(pets['hamsters'], 4)
1701	n/a
1702	n/a	l = self.list([pets, supplies])
1703	n/a	l[0]['marmots'] = 1
1704	n/a	self.assertEqual(pets['marmots'], 1)
1705	n/a	self.assertEqual(l[0]['marmots'], 1)
1706	n/a
1707	n/a	del pets
1708	n/a	del supplies
1709	n/a	self.assertEqual(l[0]['marmots'], 1)
1710	n/a
1711	n/a	outer = self.list([[88, 99], l])
1712	n/a	self.assertIsInstance(outer[0], list) # Not a ListProxy
1713	n/a	self.assertEqual(outer[-1][-1]['feed'], 3)
1714	n/a
1715	n/a	def test_namespace(self):
1716	n/a	n = self.Namespace()
1717	n/a	n.name = 'Bob'
1718	n/a	n.job = 'Builder'
1719	n/a	n._hidden = 'hidden'
1720	n/a	self.assertEqual((n.name, n.job), ('Bob', 'Builder'))
1721	n/a	del n.job
1722	n/a	self.assertEqual(str(n), "Namespace(name='Bob')")
1723	n/a	self.assertTrue(hasattr(n, 'name'))
1724	n/a	self.assertTrue(not hasattr(n, 'job'))
1725	n/a
1726	n/a	#
1727	n/a	#
1728	n/a	#
1729	n/a
1730	n/a	def sqr(x, wait=0.0):
1731	n/a	time.sleep(wait)
1732	n/a	return x*x
1733	n/a
1734	n/a	def mul(x, y):
1735	n/a	return x*y
1736	n/a
1737	n/a	def raise_large_valuerror(wait):
1738	n/a	time.sleep(wait)
1739	n/a	raise ValueError("x" * 1024**2)
1740	n/a
1741	n/a	class SayWhenError(ValueError): pass
1742	n/a
1743	n/a	def exception_throwing_generator(total, when):
1744	n/a	for i in range(total):
1745	n/a	if i == when:
1746	n/a	raise SayWhenError("Somebody said when")
1747	n/a	yield i
1748	n/a
1749	n/a	class _TestPool(BaseTestCase):
1750	n/a
1751	n/a	@classmethod
1752	n/a	def setUpClass(cls):
1753	n/a	super().setUpClass()
1754	n/a	cls.pool = cls.Pool(4)
1755	n/a
1756	n/a	@classmethod
1757	n/a	def tearDownClass(cls):
1758	n/a	cls.pool.terminate()
1759	n/a	cls.pool.join()
1760	n/a	cls.pool = None
1761	n/a	super().tearDownClass()
1762	n/a
1763	n/a	def test_apply(self):
1764	n/a	papply = self.pool.apply
1765	n/a	self.assertEqual(papply(sqr, (5,)), sqr(5))
1766	n/a	self.assertEqual(papply(sqr, (), {'x':3}), sqr(x=3))
1767	n/a
1768	n/a	def test_map(self):
1769	n/a	pmap = self.pool.map
1770	n/a	self.assertEqual(pmap(sqr, list(range(10))), list(map(sqr, list(range(10)))))
1771	n/a	self.assertEqual(pmap(sqr, list(range(100)), chunksize=20),
1772	n/a	list(map(sqr, list(range(100)))))
1773	n/a
1774	n/a	def test_starmap(self):
1775	n/a	psmap = self.pool.starmap
1776	n/a	tuples = list(zip(range(10), range(9,-1, -1)))
1777	n/a	self.assertEqual(psmap(mul, tuples),
1778	n/a	list(itertools.starmap(mul, tuples)))
1779	n/a	tuples = list(zip(range(100), range(99,-1, -1)))
1780	n/a	self.assertEqual(psmap(mul, tuples, chunksize=20),
1781	n/a	list(itertools.starmap(mul, tuples)))
1782	n/a
1783	n/a	def test_starmap_async(self):
1784	n/a	tuples = list(zip(range(100), range(99,-1, -1)))
1785	n/a	self.assertEqual(self.pool.starmap_async(mul, tuples).get(),
1786	n/a	list(itertools.starmap(mul, tuples)))
1787	n/a
1788	n/a	def test_map_async(self):
1789	n/a	self.assertEqual(self.pool.map_async(sqr, list(range(10))).get(),
1790	n/a	list(map(sqr, list(range(10)))))
1791	n/a
1792	n/a	def test_map_async_callbacks(self):
1793	n/a	call_args = self.manager.list() if self.TYPE == 'manager' else []
1794	n/a	self.pool.map_async(int, ['1'],
1795	n/a	callback=call_args.append,
1796	n/a	error_callback=call_args.append).wait()
1797	n/a	self.assertEqual(1, len(call_args))
1798	n/a	self.assertEqual([1], call_args[0])
1799	n/a	self.pool.map_async(int, ['a'],
1800	n/a	callback=call_args.append,
1801	n/a	error_callback=call_args.append).wait()
1802	n/a	self.assertEqual(2, len(call_args))
1803	n/a	self.assertIsInstance(call_args[1], ValueError)
1804	n/a
1805	n/a	def test_map_unplicklable(self):
1806	n/a	# Issue #19425 -- failure to pickle should not cause a hang
1807	n/a	if self.TYPE == 'threads':
1808	n/a	self.skipTest('test not appropriate for {}'.format(self.TYPE))
1809	n/a	class A(object):
1810	n/a	def __reduce__(self):
1811	n/a	raise RuntimeError('cannot pickle')
1812	n/a	with self.assertRaises(RuntimeError):
1813	n/a	self.pool.map(sqr, [A()]*10)
1814	n/a
1815	n/a	def test_map_chunksize(self):
1816	n/a	try:
1817	n/a	self.pool.map_async(sqr, [], chunksize=1).get(timeout=TIMEOUT1)
1818	n/a	except multiprocessing.TimeoutError:
1819	n/a	self.fail("pool.map_async with chunksize stalled on null list")
1820	n/a
1821	n/a	def test_async(self):
1822	n/a	res = self.pool.apply_async(sqr, (7, TIMEOUT1,))
1823	n/a	get = TimingWrapper(res.get)
1824	n/a	self.assertEqual(get(), 49)
1825	n/a	self.assertTimingAlmostEqual(get.elapsed, TIMEOUT1)
1826	n/a
1827	n/a	def test_async_timeout(self):
1828	n/a	res = self.pool.apply_async(sqr, (6, TIMEOUT2 + 1.0))
1829	n/a	get = TimingWrapper(res.get)
1830	n/a	self.assertRaises(multiprocessing.TimeoutError, get, timeout=TIMEOUT2)
1831	n/a	self.assertTimingAlmostEqual(get.elapsed, TIMEOUT2)
1832	n/a
1833	n/a	def test_imap(self):
1834	n/a	it = self.pool.imap(sqr, list(range(10)))
1835	n/a	self.assertEqual(list(it), list(map(sqr, list(range(10)))))
1836	n/a
1837	n/a	it = self.pool.imap(sqr, list(range(10)))
1838	n/a	for i in range(10):
1839	n/a	self.assertEqual(next(it), i*i)
1840	n/a	self.assertRaises(StopIteration, it.__next__)
1841	n/a
1842	n/a	it = self.pool.imap(sqr, list(range(1000)), chunksize=100)
1843	n/a	for i in range(1000):
1844	n/a	self.assertEqual(next(it), i*i)
1845	n/a	self.assertRaises(StopIteration, it.__next__)
1846	n/a
1847	n/a	def test_imap_handle_iterable_exception(self):
1848	n/a	if self.TYPE == 'manager':
1849	n/a	self.skipTest('test not appropriate for {}'.format(self.TYPE))
1850	n/a
1851	n/a	it = self.pool.imap(sqr, exception_throwing_generator(10, 3), 1)
1852	n/a	for i in range(3):
1853	n/a	self.assertEqual(next(it), i*i)
1854	n/a	self.assertRaises(SayWhenError, it.__next__)
1855	n/a
1856	n/a	# SayWhenError seen at start of problematic chunk's results
1857	n/a	it = self.pool.imap(sqr, exception_throwing_generator(20, 7), 2)
1858	n/a	for i in range(6):
1859	n/a	self.assertEqual(next(it), i*i)
1860	n/a	self.assertRaises(SayWhenError, it.__next__)
1861	n/a	it = self.pool.imap(sqr, exception_throwing_generator(20, 7), 4)
1862	n/a	for i in range(4):
1863	n/a	self.assertEqual(next(it), i*i)
1864	n/a	self.assertRaises(SayWhenError, it.__next__)
1865	n/a
1866	n/a	def test_imap_unordered(self):
1867	n/a	it = self.pool.imap_unordered(sqr, list(range(1000)))
1868	n/a	self.assertEqual(sorted(it), list(map(sqr, list(range(1000)))))
1869	n/a
1870	n/a	it = self.pool.imap_unordered(sqr, list(range(1000)), chunksize=53)
1871	n/a	self.assertEqual(sorted(it), list(map(sqr, list(range(1000)))))
1872	n/a
1873	n/a	def test_imap_unordered_handle_iterable_exception(self):
1874	n/a	if self.TYPE == 'manager':
1875	n/a	self.skipTest('test not appropriate for {}'.format(self.TYPE))
1876	n/a
1877	n/a	it = self.pool.imap_unordered(sqr,
1878	n/a	exception_throwing_generator(10, 3),
1879	n/a	1)
1880	n/a	expected_values = list(map(sqr, list(range(10))))
1881	n/a	with self.assertRaises(SayWhenError):
1882	n/a	# imap_unordered makes it difficult to anticipate the SayWhenError
1883	n/a	for i in range(10):
1884	n/a	value = next(it)
1885	n/a	self.assertIn(value, expected_values)
1886	n/a	expected_values.remove(value)
1887	n/a
1888	n/a	it = self.pool.imap_unordered(sqr,
1889	n/a	exception_throwing_generator(20, 7),
1890	n/a	2)
1891	n/a	expected_values = list(map(sqr, list(range(20))))
1892	n/a	with self.assertRaises(SayWhenError):
1893	n/a	for i in range(20):
1894	n/a	value = next(it)
1895	n/a	self.assertIn(value, expected_values)
1896	n/a	expected_values.remove(value)
1897	n/a
1898	n/a	def test_make_pool(self):
1899	n/a	expected_error = (RemoteError if self.TYPE == 'manager'
1900	n/a	else ValueError)
1901	n/a
1902	n/a	self.assertRaises(expected_error, self.Pool, -1)
1903	n/a	self.assertRaises(expected_error, self.Pool, 0)
1904	n/a
1905	n/a	if self.TYPE != 'manager':
1906	n/a	p = self.Pool(3)
1907	n/a	try:
1908	n/a	self.assertEqual(3, len(p._pool))
1909	n/a	finally:
1910	n/a	p.close()
1911	n/a	p.join()
1912	n/a
1913	n/a	def test_terminate(self):
1914	n/a	result = self.pool.map_async(
1915	n/a	time.sleep, [0.1 for i in range(10000)], chunksize=1
1916	n/a	)
1917	n/a	self.pool.terminate()
1918	n/a	join = TimingWrapper(self.pool.join)
1919	n/a	join()
1920	n/a	# Sanity check the pool didn't wait for all tasks to finish
1921	n/a	self.assertLess(join.elapsed, 2.0)
1922	n/a
1923	n/a	def test_empty_iterable(self):
1924	n/a	# See Issue 12157
1925	n/a	p = self.Pool(1)
1926	n/a
1927	n/a	self.assertEqual(p.map(sqr, []), [])
1928	n/a	self.assertEqual(list(p.imap(sqr, [])), [])
1929	n/a	self.assertEqual(list(p.imap_unordered(sqr, [])), [])
1930	n/a	self.assertEqual(p.map_async(sqr, []).get(), [])
1931	n/a
1932	n/a	p.close()
1933	n/a	p.join()
1934	n/a
1935	n/a	def test_context(self):
1936	n/a	if self.TYPE == 'processes':
1937	n/a	L = list(range(10))
1938	n/a	expected = [sqr(i) for i in L]
1939	n/a	with self.Pool(2) as p:
1940	n/a	r = p.map_async(sqr, L)
1941	n/a	self.assertEqual(r.get(), expected)
1942	n/a	self.assertRaises(ValueError, p.map_async, sqr, L)
1943	n/a
1944	n/a	@classmethod
1945	n/a	def _test_traceback(cls):
1946	n/a	raise RuntimeError(123) # some comment
1947	n/a
1948	n/a	def test_traceback(self):
1949	n/a	# We want ensure that the traceback from the child process is
1950	n/a	# contained in the traceback raised in the main process.
1951	n/a	if self.TYPE == 'processes':
1952	n/a	with self.Pool(1) as p:
1953	n/a	try:
1954	n/a	p.apply(self._test_traceback)
1955	n/a	except Exception as e:
1956	n/a	exc = e
1957	n/a	else:
1958	n/a	raise AssertionError('expected RuntimeError')
1959	n/a	self.assertIs(type(exc), RuntimeError)
1960	n/a	self.assertEqual(exc.args, (123,))
1961	n/a	cause = exc.__cause__
1962	n/a	self.assertIs(type(cause), multiprocessing.pool.RemoteTraceback)
1963	n/a	self.assertIn('raise RuntimeError(123) # some comment', cause.tb)
1964	n/a
1965	n/a	with test.support.captured_stderr() as f1:
1966	n/a	try:
1967	n/a	raise exc
1968	n/a	except RuntimeError:
1969	n/a	sys.excepthook(*sys.exc_info())
1970	n/a	self.assertIn('raise RuntimeError(123) # some comment',
1971	n/a	f1.getvalue())
1972	n/a
1973	n/a	@classmethod
1974	n/a	def _test_wrapped_exception(cls):
1975	n/a	raise RuntimeError('foo')
1976	n/a
1977	n/a	def test_wrapped_exception(self):
1978	n/a	# Issue #20980: Should not wrap exception when using thread pool
1979	n/a	with self.Pool(1) as p:
1980	n/a	with self.assertRaises(RuntimeError):
1981	n/a	p.apply(self._test_wrapped_exception)
1982	n/a
1983	n/a	def test_map_no_failfast(self):
1984	n/a	# Issue #23992: the fail-fast behaviour when an exception is raised
1985	n/a	# during map() would make Pool.join() deadlock, because a worker
1986	n/a	# process would fill the result queue (after the result handler thread
1987	n/a	# terminated, hence not draining it anymore).
1988	n/a
1989	n/a	t_start = time.time()
1990	n/a
1991	n/a	with self.assertRaises(ValueError):
1992	n/a	with self.Pool(2) as p:
1993	n/a	try:
1994	n/a	p.map(raise_large_valuerror, [0, 1])
1995	n/a	finally:
1996	n/a	time.sleep(0.5)
1997	n/a	p.close()
1998	n/a	p.join()
1999	n/a
2000	n/a	# check that we indeed waited for all jobs
2001	n/a	self.assertGreater(time.time() - t_start, 0.9)
2002	n/a
2003	n/a
2004	n/a	def raising():
2005	n/a	raise KeyError("key")
2006	n/a
2007	n/a	def unpickleable_result():
2008	n/a	return lambda: 42
2009	n/a
2010	n/a	class _TestPoolWorkerErrors(BaseTestCase):
2011	n/a	ALLOWED_TYPES = ('processes', )
2012	n/a
2013	n/a	def test_async_error_callback(self):
2014	n/a	p = multiprocessing.Pool(2)
2015	n/a
2016	n/a	scratchpad = [None]
2017	n/a	def errback(exc):
2018	n/a	scratchpad[0] = exc
2019	n/a
2020	n/a	res = p.apply_async(raising, error_callback=errback)
2021	n/a	self.assertRaises(KeyError, res.get)
2022	n/a	self.assertTrue(scratchpad[0])
2023	n/a	self.assertIsInstance(scratchpad[0], KeyError)
2024	n/a
2025	n/a	p.close()
2026	n/a	p.join()
2027	n/a
2028	n/a	def test_unpickleable_result(self):
2029	n/a	from multiprocessing.pool import MaybeEncodingError
2030	n/a	p = multiprocessing.Pool(2)
2031	n/a
2032	n/a	# Make sure we don't lose pool processes because of encoding errors.
2033	n/a	for iteration in range(20):
2034	n/a
2035	n/a	scratchpad = [None]
2036	n/a	def errback(exc):
2037	n/a	scratchpad[0] = exc
2038	n/a
2039	n/a	res = p.apply_async(unpickleable_result, error_callback=errback)
2040	n/a	self.assertRaises(MaybeEncodingError, res.get)
2041	n/a	wrapped = scratchpad[0]
2042	n/a	self.assertTrue(wrapped)
2043	n/a	self.assertIsInstance(scratchpad[0], MaybeEncodingError)
2044	n/a	self.assertIsNotNone(wrapped.exc)
2045	n/a	self.assertIsNotNone(wrapped.value)
2046	n/a
2047	n/a	p.close()
2048	n/a	p.join()
2049	n/a
2050	n/a	class _TestPoolWorkerLifetime(BaseTestCase):
2051	n/a	ALLOWED_TYPES = ('processes', )
2052	n/a
2053	n/a	def test_pool_worker_lifetime(self):
2054	n/a	p = multiprocessing.Pool(3, maxtasksperchild=10)
2055	n/a	self.assertEqual(3, len(p._pool))
2056	n/a	origworkerpids = [w.pid for w in p._pool]
2057	n/a	# Run many tasks so each worker gets replaced (hopefully)
2058	n/a	results = []
2059	n/a	for i in range(100):
2060	n/a	results.append(p.apply_async(sqr, (i, )))
2061	n/a	# Fetch the results and verify we got the right answers,
2062	n/a	# also ensuring all the tasks have completed.
2063	n/a	for (j, res) in enumerate(results):
2064	n/a	self.assertEqual(res.get(), sqr(j))
2065	n/a	# Refill the pool
2066	n/a	p._repopulate_pool()
2067	n/a	# Wait until all workers are alive
2068	n/a	# (countdown * DELTA = 5 seconds max startup process time)
2069	n/a	countdown = 50
2070	n/a	while countdown and not all(w.is_alive() for w in p._pool):
2071	n/a	countdown -= 1
2072	n/a	time.sleep(DELTA)
2073	n/a	finalworkerpids = [w.pid for w in p._pool]
2074	n/a	# All pids should be assigned. See issue #7805.
2075	n/a	self.assertNotIn(None, origworkerpids)
2076	n/a	self.assertNotIn(None, finalworkerpids)
2077	n/a	# Finally, check that the worker pids have changed
2078	n/a	self.assertNotEqual(sorted(origworkerpids), sorted(finalworkerpids))
2079	n/a	p.close()
2080	n/a	p.join()
2081	n/a
2082	n/a	def test_pool_worker_lifetime_early_close(self):
2083	n/a	# Issue #10332: closing a pool whose workers have limited lifetimes
2084	n/a	# before all the tasks completed would make join() hang.
2085	n/a	p = multiprocessing.Pool(3, maxtasksperchild=1)
2086	n/a	results = []
2087	n/a	for i in range(6):
2088	n/a	results.append(p.apply_async(sqr, (i, 0.3)))
2089	n/a	p.close()
2090	n/a	p.join()
2091	n/a	# check the results
2092	n/a	for (j, res) in enumerate(results):
2093	n/a	self.assertEqual(res.get(), sqr(j))
2094	n/a
2095	n/a	#
2096	n/a	# Test of creating a customized manager class
2097	n/a	#
2098	n/a
2099	n/a	from multiprocessing.managers import BaseManager, BaseProxy, RemoteError
2100	n/a
2101	n/a	class FooBar(object):
2102	n/a	def f(self):
2103	n/a	return 'f()'
2104	n/a	def g(self):
2105	n/a	raise ValueError
2106	n/a	def _h(self):
2107	n/a	return '_h()'
2108	n/a
2109	n/a	def baz():
2110	n/a	for i in range(10):
2111	n/a	yield i*i
2112	n/a
2113	n/a	class IteratorProxy(BaseProxy):
2114	n/a	_exposed_ = ('__next__',)
2115	n/a	def __iter__(self):
2116	n/a	return self
2117	n/a	def __next__(self):
2118	n/a	return self._callmethod('__next__')
2119	n/a
2120	n/a	class MyManager(BaseManager):
2121	n/a	pass
2122	n/a
2123	n/a	MyManager.register('Foo', callable=FooBar)
2124	n/a	MyManager.register('Bar', callable=FooBar, exposed=('f', '_h'))
2125	n/a	MyManager.register('baz', callable=baz, proxytype=IteratorProxy)
2126	n/a
2127	n/a
2128	n/a	class _TestMyManager(BaseTestCase):
2129	n/a
2130	n/a	ALLOWED_TYPES = ('manager',)
2131	n/a
2132	n/a	def test_mymanager(self):
2133	n/a	manager = MyManager()
2134	n/a	manager.start()
2135	n/a	self.common(manager)
2136	n/a	manager.shutdown()
2137	n/a
2138	n/a	# If the manager process exited cleanly then the exitcode
2139	n/a	# will be zero. Otherwise (after a short timeout)
2140	n/a	# terminate() is used, resulting in an exitcode of -SIGTERM.
2141	n/a	self.assertEqual(manager._process.exitcode, 0)
2142	n/a
2143	n/a	def test_mymanager_context(self):
2144	n/a	with MyManager() as manager:
2145	n/a	self.common(manager)
2146	n/a	self.assertEqual(manager._process.exitcode, 0)
2147	n/a
2148	n/a	def test_mymanager_context_prestarted(self):
2149	n/a	manager = MyManager()
2150	n/a	manager.start()
2151	n/a	with manager:
2152	n/a	self.common(manager)
2153	n/a	self.assertEqual(manager._process.exitcode, 0)
2154	n/a
2155	n/a	def common(self, manager):
2156	n/a	foo = manager.Foo()
2157	n/a	bar = manager.Bar()
2158	n/a	baz = manager.baz()
2159	n/a
2160	n/a	foo_methods = [name for name in ('f', 'g', '_h') if hasattr(foo, name)]
2161	n/a	bar_methods = [name for name in ('f', 'g', '_h') if hasattr(bar, name)]
2162	n/a
2163	n/a	self.assertEqual(foo_methods, ['f', 'g'])
2164	n/a	self.assertEqual(bar_methods, ['f', '_h'])
2165	n/a
2166	n/a	self.assertEqual(foo.f(), 'f()')
2167	n/a	self.assertRaises(ValueError, foo.g)
2168	n/a	self.assertEqual(foo._callmethod('f'), 'f()')
2169	n/a	self.assertRaises(RemoteError, foo._callmethod, '_h')
2170	n/a
2171	n/a	self.assertEqual(bar.f(), 'f()')
2172	n/a	self.assertEqual(bar._h(), '_h()')
2173	n/a	self.assertEqual(bar._callmethod('f'), 'f()')
2174	n/a	self.assertEqual(bar._callmethod('_h'), '_h()')
2175	n/a
2176	n/a	self.assertEqual(list(baz), [i*i for i in range(10)])
2177	n/a
2178	n/a
2179	n/a	#
2180	n/a	# Test of connecting to a remote server and using xmlrpclib for serialization
2181	n/a	#
2182	n/a
2183	n/a	_queue = pyqueue.Queue()
2184	n/a	def get_queue():
2185	n/a	return _queue
2186	n/a
2187	n/a	class QueueManager(BaseManager):
2188	n/a	'''manager class used by server process'''
2189	n/a	QueueManager.register('get_queue', callable=get_queue)
2190	n/a
2191	n/a	class QueueManager2(BaseManager):
2192	n/a	'''manager class which specifies the same interface as QueueManager'''
2193	n/a	QueueManager2.register('get_queue')
2194	n/a
2195	n/a
2196	n/a	SERIALIZER = 'xmlrpclib'
2197	n/a
2198	n/a	class _TestRemoteManager(BaseTestCase):
2199	n/a
2200	n/a	ALLOWED_TYPES = ('manager',)
2201	n/a	values = ['hello world', None, True, 2.25,
2202	n/a	'hall\xe5 v\xe4rlden',
2203	n/a	'\u043f\u0440\u0438\u0432\u0456\u0442 \u0441\u0432\u0456\u0442',
2204	n/a	b'hall\xe5 v\xe4rlden',
2205	n/a	]
2206	n/a	result = values[:]
2207	n/a
2208	n/a	@classmethod
2209	n/a	def _putter(cls, address, authkey):
2210	n/a	manager = QueueManager2(
2211	n/a	address=address, authkey=authkey, serializer=SERIALIZER
2212	n/a	)
2213	n/a	manager.connect()
2214	n/a	queue = manager.get_queue()
2215	n/a	# Note that xmlrpclib will deserialize object as a list not a tuple
2216	n/a	queue.put(tuple(cls.values))
2217	n/a
2218	n/a	def test_remote(self):
2219	n/a	authkey = os.urandom(32)
2220	n/a
2221	n/a	manager = QueueManager(
2222	n/a	address=(test.support.HOST, 0), authkey=authkey, serializer=SERIALIZER
2223	n/a	)
2224	n/a	manager.start()
2225	n/a
2226	n/a	p = self.Process(target=self._putter, args=(manager.address, authkey))
2227	n/a	p.daemon = True
2228	n/a	p.start()
2229	n/a
2230	n/a	manager2 = QueueManager2(
2231	n/a	address=manager.address, authkey=authkey, serializer=SERIALIZER
2232	n/a	)
2233	n/a	manager2.connect()
2234	n/a	queue = manager2.get_queue()
2235	n/a
2236	n/a	self.assertEqual(queue.get(), self.result)
2237	n/a
2238	n/a	# Because we are using xmlrpclib for serialization instead of
2239	n/a	# pickle this will cause a serialization error.
2240	n/a	self.assertRaises(Exception, queue.put, time.sleep)
2241	n/a
2242	n/a	# Make queue finalizer run before the server is stopped
2243	n/a	del queue
2244	n/a	manager.shutdown()
2245	n/a
2246	n/a	class _TestManagerRestart(BaseTestCase):
2247	n/a
2248	n/a	@classmethod
2249	n/a	def _putter(cls, address, authkey):
2250	n/a	manager = QueueManager(
2251	n/a	address=address, authkey=authkey, serializer=SERIALIZER)
2252	n/a	manager.connect()
2253	n/a	queue = manager.get_queue()
2254	n/a	queue.put('hello world')
2255	n/a
2256	n/a	def test_rapid_restart(self):
2257	n/a	authkey = os.urandom(32)
2258	n/a	manager = QueueManager(
2259	n/a	address=(test.support.HOST, 0), authkey=authkey, serializer=SERIALIZER)
2260	n/a	srvr = manager.get_server()
2261	n/a	addr = srvr.address
2262	n/a	# Close the connection.Listener socket which gets opened as a part
2263	n/a	# of manager.get_server(). It's not needed for the test.
2264	n/a	srvr.listener.close()
2265	n/a	manager.start()
2266	n/a
2267	n/a	p = self.Process(target=self._putter, args=(manager.address, authkey))
2268	n/a	p.daemon = True
2269	n/a	p.start()
2270	n/a	queue = manager.get_queue()
2271	n/a	self.assertEqual(queue.get(), 'hello world')
2272	n/a	del queue
2273	n/a	manager.shutdown()
2274	n/a	manager = QueueManager(
2275	n/a	address=addr, authkey=authkey, serializer=SERIALIZER)
2276	n/a	try:
2277	n/a	manager.start()
2278	n/a	except OSError as e:
2279	n/a	if e.errno != errno.EADDRINUSE:
2280	n/a	raise
2281	n/a	# Retry after some time, in case the old socket was lingering
2282	n/a	# (sporadic failure on buildbots)
2283	n/a	time.sleep(1.0)
2284	n/a	manager = QueueManager(
2285	n/a	address=addr, authkey=authkey, serializer=SERIALIZER)
2286	n/a	manager.shutdown()
2287	n/a
2288	n/a	#
2289	n/a	#
2290	n/a	#
2291	n/a
2292	n/a	SENTINEL = latin('')
2293	n/a
2294	n/a	class _TestConnection(BaseTestCase):
2295	n/a
2296	n/a	ALLOWED_TYPES = ('processes', 'threads')
2297	n/a
2298	n/a	@classmethod
2299	n/a	def _echo(cls, conn):
2300	n/a	for msg in iter(conn.recv_bytes, SENTINEL):
2301	n/a	conn.send_bytes(msg)
2302	n/a	conn.close()
2303	n/a
2304	n/a	def test_connection(self):
2305	n/a	conn, child_conn = self.Pipe()
2306	n/a
2307	n/a	p = self.Process(target=self._echo, args=(child_conn,))
2308	n/a	p.daemon = True
2309	n/a	p.start()
2310	n/a
2311	n/a	seq = [1, 2.25, None]
2312	n/a	msg = latin('hello world')
2313	n/a	longmsg = msg * 10
2314	n/a	arr = array.array('i', list(range(4)))
2315	n/a
2316	n/a	if self.TYPE == 'processes':
2317	n/a	self.assertEqual(type(conn.fileno()), int)
2318	n/a
2319	n/a	self.assertEqual(conn.send(seq), None)
2320	n/a	self.assertEqual(conn.recv(), seq)
2321	n/a
2322	n/a	self.assertEqual(conn.send_bytes(msg), None)
2323	n/a	self.assertEqual(conn.recv_bytes(), msg)
2324	n/a
2325	n/a	if self.TYPE == 'processes':
2326	n/a	buffer = array.array('i', [0]*10)
2327	n/a	expected = list(arr) + [0] * (10 - len(arr))
2328	n/a	self.assertEqual(conn.send_bytes(arr), None)
2329	n/a	self.assertEqual(conn.recv_bytes_into(buffer),
2330	n/a	len(arr) * buffer.itemsize)
2331	n/a	self.assertEqual(list(buffer), expected)
2332	n/a
2333	n/a	buffer = array.array('i', [0]*10)
2334	n/a	expected = [0] * 3 + list(arr) + [0] * (10 - 3 - len(arr))
2335	n/a	self.assertEqual(conn.send_bytes(arr), None)
2336	n/a	self.assertEqual(conn.recv_bytes_into(buffer, 3 * buffer.itemsize),
2337	n/a	len(arr) * buffer.itemsize)
2338	n/a	self.assertEqual(list(buffer), expected)
2339	n/a
2340	n/a	buffer = bytearray(latin(' ' * 40))
2341	n/a	self.assertEqual(conn.send_bytes(longmsg), None)
2342	n/a	try:
2343	n/a	res = conn.recv_bytes_into(buffer)
2344	n/a	except multiprocessing.BufferTooShort as e:
2345	n/a	self.assertEqual(e.args, (longmsg,))
2346	n/a	else:
2347	n/a	self.fail('expected BufferTooShort, got %s' % res)
2348	n/a
2349	n/a	poll = TimingWrapper(conn.poll)
2350	n/a
2351	n/a	self.assertEqual(poll(), False)
2352	n/a	self.assertTimingAlmostEqual(poll.elapsed, 0)
2353	n/a
2354	n/a	self.assertEqual(poll(-1), False)
2355	n/a	self.assertTimingAlmostEqual(poll.elapsed, 0)
2356	n/a
2357	n/a	self.assertEqual(poll(TIMEOUT1), False)
2358	n/a	self.assertTimingAlmostEqual(poll.elapsed, TIMEOUT1)
2359	n/a
2360	n/a	conn.send(None)
2361	n/a	time.sleep(.1)
2362	n/a
2363	n/a	self.assertEqual(poll(TIMEOUT1), True)
2364	n/a	self.assertTimingAlmostEqual(poll.elapsed, 0)
2365	n/a
2366	n/a	self.assertEqual(conn.recv(), None)
2367	n/a
2368	n/a	really_big_msg = latin('X') * (1024 * 1024 * 16) # 16Mb
2369	n/a	conn.send_bytes(really_big_msg)
2370	n/a	self.assertEqual(conn.recv_bytes(), really_big_msg)
2371	n/a
2372	n/a	conn.send_bytes(SENTINEL) # tell child to quit
2373	n/a	child_conn.close()
2374	n/a
2375	n/a	if self.TYPE == 'processes':
2376	n/a	self.assertEqual(conn.readable, True)
2377	n/a	self.assertEqual(conn.writable, True)
2378	n/a	self.assertRaises(EOFError, conn.recv)
2379	n/a	self.assertRaises(EOFError, conn.recv_bytes)
2380	n/a
2381	n/a	p.join()
2382	n/a
2383	n/a	def test_duplex_false(self):
2384	n/a	reader, writer = self.Pipe(duplex=False)
2385	n/a	self.assertEqual(writer.send(1), None)
2386	n/a	self.assertEqual(reader.recv(), 1)
2387	n/a	if self.TYPE == 'processes':
2388	n/a	self.assertEqual(reader.readable, True)
2389	n/a	self.assertEqual(reader.writable, False)
2390	n/a	self.assertEqual(writer.readable, False)
2391	n/a	self.assertEqual(writer.writable, True)
2392	n/a	self.assertRaises(OSError, reader.send, 2)
2393	n/a	self.assertRaises(OSError, writer.recv)
2394	n/a	self.assertRaises(OSError, writer.poll)
2395	n/a
2396	n/a	def test_spawn_close(self):
2397	n/a	# We test that a pipe connection can be closed by parent
2398	n/a	# process immediately after child is spawned. On Windows this
2399	n/a	# would have sometimes failed on old versions because
2400	n/a	# child_conn would be closed before the child got a chance to
2401	n/a	# duplicate it.
2402	n/a	conn, child_conn = self.Pipe()
2403	n/a
2404	n/a	p = self.Process(target=self._echo, args=(child_conn,))
2405	n/a	p.daemon = True
2406	n/a	p.start()
2407	n/a	child_conn.close() # this might complete before child initializes
2408	n/a
2409	n/a	msg = latin('hello')
2410	n/a	conn.send_bytes(msg)
2411	n/a	self.assertEqual(conn.recv_bytes(), msg)
2412	n/a
2413	n/a	conn.send_bytes(SENTINEL)
2414	n/a	conn.close()
2415	n/a	p.join()
2416	n/a
2417	n/a	def test_sendbytes(self):
2418	n/a	if self.TYPE != 'processes':
2419	n/a	self.skipTest('test not appropriate for {}'.format(self.TYPE))
2420	n/a
2421	n/a	msg = latin('abcdefghijklmnopqrstuvwxyz')
2422	n/a	a, b = self.Pipe()
2423	n/a
2424	n/a	a.send_bytes(msg)
2425	n/a	self.assertEqual(b.recv_bytes(), msg)
2426	n/a
2427	n/a	a.send_bytes(msg, 5)
2428	n/a	self.assertEqual(b.recv_bytes(), msg[5:])
2429	n/a
2430	n/a	a.send_bytes(msg, 7, 8)
2431	n/a	self.assertEqual(b.recv_bytes(), msg[7:7+8])
2432	n/a
2433	n/a	a.send_bytes(msg, 26)
2434	n/a	self.assertEqual(b.recv_bytes(), latin(''))
2435	n/a
2436	n/a	a.send_bytes(msg, 26, 0)
2437	n/a	self.assertEqual(b.recv_bytes(), latin(''))
2438	n/a
2439	n/a	self.assertRaises(ValueError, a.send_bytes, msg, 27)
2440	n/a
2441	n/a	self.assertRaises(ValueError, a.send_bytes, msg, 22, 5)
2442	n/a
2443	n/a	self.assertRaises(ValueError, a.send_bytes, msg, 26, 1)
2444	n/a
2445	n/a	self.assertRaises(ValueError, a.send_bytes, msg, -1)
2446	n/a
2447	n/a	self.assertRaises(ValueError, a.send_bytes, msg, 4, -1)
2448	n/a
2449	n/a	@classmethod
2450	n/a	def _is_fd_assigned(cls, fd):
2451	n/a	try:
2452	n/a	os.fstat(fd)
2453	n/a	except OSError as e:
2454	n/a	if e.errno == errno.EBADF:
2455	n/a	return False
2456	n/a	raise
2457	n/a	else:
2458	n/a	return True
2459	n/a
2460	n/a	@classmethod
2461	n/a	def _writefd(cls, conn, data, create_dummy_fds=False):
2462	n/a	if create_dummy_fds:
2463	n/a	for i in range(0, 256):
2464	n/a	if not cls._is_fd_assigned(i):
2465	n/a	os.dup2(conn.fileno(), i)
2466	n/a	fd = reduction.recv_handle(conn)
2467	n/a	if msvcrt:
2468	n/a	fd = msvcrt.open_osfhandle(fd, os.O_WRONLY)
2469	n/a	os.write(fd, data)
2470	n/a	os.close(fd)
2471	n/a
2472	n/a	@unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
2473	n/a	def test_fd_transfer(self):
2474	n/a	if self.TYPE != 'processes':
2475	n/a	self.skipTest("only makes sense with processes")
2476	n/a	conn, child_conn = self.Pipe(duplex=True)
2477	n/a
2478	n/a	p = self.Process(target=self._writefd, args=(child_conn, b"foo"))
2479	n/a	p.daemon = True
2480	n/a	p.start()
2481	n/a	self.addCleanup(test.support.unlink, test.support.TESTFN)
2482	n/a	with open(test.support.TESTFN, "wb") as f:
2483	n/a	fd = f.fileno()
2484	n/a	if msvcrt:
2485	n/a	fd = msvcrt.get_osfhandle(fd)
2486	n/a	reduction.send_handle(conn, fd, p.pid)
2487	n/a	p.join()
2488	n/a	with open(test.support.TESTFN, "rb") as f:
2489	n/a	self.assertEqual(f.read(), b"foo")
2490	n/a
2491	n/a	@unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
2492	n/a	@unittest.skipIf(sys.platform == "win32",
2493	n/a	"test semantics don't make sense on Windows")
2494	n/a	@unittest.skipIf(MAXFD <= 256,
2495	n/a	"largest assignable fd number is too small")
2496	n/a	@unittest.skipUnless(hasattr(os, "dup2"),
2497	n/a	"test needs os.dup2()")
2498	n/a	def test_large_fd_transfer(self):
2499	n/a	# With fd > 256 (issue #11657)
2500	n/a	if self.TYPE != 'processes':
2501	n/a	self.skipTest("only makes sense with processes")
2502	n/a	conn, child_conn = self.Pipe(duplex=True)
2503	n/a
2504	n/a	p = self.Process(target=self._writefd, args=(child_conn, b"bar", True))
2505	n/a	p.daemon = True
2506	n/a	p.start()
2507	n/a	self.addCleanup(test.support.unlink, test.support.TESTFN)
2508	n/a	with open(test.support.TESTFN, "wb") as f:
2509	n/a	fd = f.fileno()
2510	n/a	for newfd in range(256, MAXFD):
2511	n/a	if not self._is_fd_assigned(newfd):
2512	n/a	break
2513	n/a	else:
2514	n/a	self.fail("could not find an unassigned large file descriptor")
2515	n/a	os.dup2(fd, newfd)
2516	n/a	try:
2517	n/a	reduction.send_handle(conn, newfd, p.pid)
2518	n/a	finally:
2519	n/a	os.close(newfd)
2520	n/a	p.join()
2521	n/a	with open(test.support.TESTFN, "rb") as f:
2522	n/a	self.assertEqual(f.read(), b"bar")
2523	n/a
2524	n/a	@classmethod
2525	n/a	def _send_data_without_fd(self, conn):
2526	n/a	os.write(conn.fileno(), b"\0")
2527	n/a
2528	n/a	@unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
2529	n/a	@unittest.skipIf(sys.platform == "win32", "doesn't make sense on Windows")
2530	n/a	def test_missing_fd_transfer(self):
2531	n/a	# Check that exception is raised when received data is not
2532	n/a	# accompanied by a file descriptor in ancillary data.
2533	n/a	if self.TYPE != 'processes':
2534	n/a	self.skipTest("only makes sense with processes")
2535	n/a	conn, child_conn = self.Pipe(duplex=True)
2536	n/a
2537	n/a	p = self.Process(target=self._send_data_without_fd, args=(child_conn,))
2538	n/a	p.daemon = True
2539	n/a	p.start()
2540	n/a	self.assertRaises(RuntimeError, reduction.recv_handle, conn)
2541	n/a	p.join()
2542	n/a
2543	n/a	def test_context(self):
2544	n/a	a, b = self.Pipe()
2545	n/a
2546	n/a	with a, b:
2547	n/a	a.send(1729)
2548	n/a	self.assertEqual(b.recv(), 1729)
2549	n/a	if self.TYPE == 'processes':
2550	n/a	self.assertFalse(a.closed)
2551	n/a	self.assertFalse(b.closed)
2552	n/a
2553	n/a	if self.TYPE == 'processes':
2554	n/a	self.assertTrue(a.closed)
2555	n/a	self.assertTrue(b.closed)
2556	n/a	self.assertRaises(OSError, a.recv)
2557	n/a	self.assertRaises(OSError, b.recv)
2558	n/a
2559	n/a	class _TestListener(BaseTestCase):
2560	n/a
2561	n/a	ALLOWED_TYPES = ('processes',)
2562	n/a
2563	n/a	def test_multiple_bind(self):
2564	n/a	for family in self.connection.families:
2565	n/a	l = self.connection.Listener(family=family)
2566	n/a	self.addCleanup(l.close)
2567	n/a	self.assertRaises(OSError, self.connection.Listener,
2568	n/a	l.address, family)
2569	n/a
2570	n/a	def test_context(self):
2571	n/a	with self.connection.Listener() as l:
2572	n/a	with self.connection.Client(l.address) as c:
2573	n/a	with l.accept() as d:
2574	n/a	c.send(1729)
2575	n/a	self.assertEqual(d.recv(), 1729)
2576	n/a
2577	n/a	if self.TYPE == 'processes':
2578	n/a	self.assertRaises(OSError, l.accept)
2579	n/a
2580	n/a	class _TestListenerClient(BaseTestCase):
2581	n/a
2582	n/a	ALLOWED_TYPES = ('processes', 'threads')
2583	n/a
2584	n/a	@classmethod
2585	n/a	def _test(cls, address):
2586	n/a	conn = cls.connection.Client(address)
2587	n/a	conn.send('hello')
2588	n/a	conn.close()
2589	n/a
2590	n/a	def test_listener_client(self):
2591	n/a	for family in self.connection.families:
2592	n/a	l = self.connection.Listener(family=family)
2593	n/a	p = self.Process(target=self._test, args=(l.address,))
2594	n/a	p.daemon = True
2595	n/a	p.start()
2596	n/a	conn = l.accept()
2597	n/a	self.assertEqual(conn.recv(), 'hello')
2598	n/a	p.join()
2599	n/a	l.close()
2600	n/a
2601	n/a	def test_issue14725(self):
2602	n/a	l = self.connection.Listener()
2603	n/a	p = self.Process(target=self._test, args=(l.address,))
2604	n/a	p.daemon = True
2605	n/a	p.start()
2606	n/a	time.sleep(1)
2607	n/a	# On Windows the client process should by now have connected,
2608	n/a	# written data and closed the pipe handle by now. This causes
2609	n/a	# ConnectNamdedPipe() to fail with ERROR_NO_DATA. See Issue
2610	n/a	# 14725.
2611	n/a	conn = l.accept()
2612	n/a	self.assertEqual(conn.recv(), 'hello')
2613	n/a	conn.close()
2614	n/a	p.join()
2615	n/a	l.close()
2616	n/a
2617	n/a	def test_issue16955(self):
2618	n/a	for fam in self.connection.families:
2619	n/a	l = self.connection.Listener(family=fam)
2620	n/a	c = self.connection.Client(l.address)
2621	n/a	a = l.accept()
2622	n/a	a.send_bytes(b"hello")
2623	n/a	self.assertTrue(c.poll(1))
2624	n/a	a.close()
2625	n/a	c.close()
2626	n/a	l.close()
2627	n/a
2628	n/a	class _TestPoll(BaseTestCase):
2629	n/a
2630	n/a	ALLOWED_TYPES = ('processes', 'threads')
2631	n/a
2632	n/a	def test_empty_string(self):
2633	n/a	a, b = self.Pipe()
2634	n/a	self.assertEqual(a.poll(), False)
2635	n/a	b.send_bytes(b'')
2636	n/a	self.assertEqual(a.poll(), True)
2637	n/a	self.assertEqual(a.poll(), True)
2638	n/a
2639	n/a	@classmethod
2640	n/a	def _child_strings(cls, conn, strings):
2641	n/a	for s in strings:
2642	n/a	time.sleep(0.1)
2643	n/a	conn.send_bytes(s)
2644	n/a	conn.close()
2645	n/a
2646	n/a	def test_strings(self):
2647	n/a	strings = (b'hello', b'', b'a', b'b', b'', b'bye', b'', b'lop')
2648	n/a	a, b = self.Pipe()
2649	n/a	p = self.Process(target=self._child_strings, args=(b, strings))
2650	n/a	p.start()
2651	n/a
2652	n/a	for s in strings:
2653	n/a	for i in range(200):
2654	n/a	if a.poll(0.01):
2655	n/a	break
2656	n/a	x = a.recv_bytes()
2657	n/a	self.assertEqual(s, x)
2658	n/a
2659	n/a	p.join()
2660	n/a
2661	n/a	@classmethod
2662	n/a	def _child_boundaries(cls, r):
2663	n/a	# Polling may "pull" a message in to the child process, but we
2664	n/a	# don't want it to pull only part of a message, as that would
2665	n/a	# corrupt the pipe for any other processes which might later
2666	n/a	# read from it.
2667	n/a	r.poll(5)
2668	n/a
2669	n/a	def test_boundaries(self):
2670	n/a	r, w = self.Pipe(False)
2671	n/a	p = self.Process(target=self._child_boundaries, args=(r,))
2672	n/a	p.start()
2673	n/a	time.sleep(2)
2674	n/a	L = [b"first", b"second"]
2675	n/a	for obj in L:
2676	n/a	w.send_bytes(obj)
2677	n/a	w.close()
2678	n/a	p.join()
2679	n/a	self.assertIn(r.recv_bytes(), L)
2680	n/a
2681	n/a	@classmethod
2682	n/a	def _child_dont_merge(cls, b):
2683	n/a	b.send_bytes(b'a')
2684	n/a	b.send_bytes(b'b')
2685	n/a	b.send_bytes(b'cd')
2686	n/a
2687	n/a	def test_dont_merge(self):
2688	n/a	a, b = self.Pipe()
2689	n/a	self.assertEqual(a.poll(0.0), False)
2690	n/a	self.assertEqual(a.poll(0.1), False)
2691	n/a
2692	n/a	p = self.Process(target=self._child_dont_merge, args=(b,))
2693	n/a	p.start()
2694	n/a
2695	n/a	self.assertEqual(a.recv_bytes(), b'a')
2696	n/a	self.assertEqual(a.poll(1.0), True)
2697	n/a	self.assertEqual(a.poll(1.0), True)
2698	n/a	self.assertEqual(a.recv_bytes(), b'b')
2699	n/a	self.assertEqual(a.poll(1.0), True)
2700	n/a	self.assertEqual(a.poll(1.0), True)
2701	n/a	self.assertEqual(a.poll(0.0), True)
2702	n/a	self.assertEqual(a.recv_bytes(), b'cd')
2703	n/a
2704	n/a	p.join()
2705	n/a
2706	n/a	#
2707	n/a	# Test of sending connection and socket objects between processes
2708	n/a	#
2709	n/a
2710	n/a	@unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
2711	n/a	class _TestPicklingConnections(BaseTestCase):
2712	n/a
2713	n/a	ALLOWED_TYPES = ('processes',)
2714	n/a
2715	n/a	@classmethod
2716	n/a	def tearDownClass(cls):
2717	n/a	from multiprocessing import resource_sharer
2718	n/a	resource_sharer.stop(timeout=5)
2719	n/a
2720	n/a	@classmethod
2721	n/a	def _listener(cls, conn, families):
2722	n/a	for fam in families:
2723	n/a	l = cls.connection.Listener(family=fam)
2724	n/a	conn.send(l.address)
2725	n/a	new_conn = l.accept()
2726	n/a	conn.send(new_conn)
2727	n/a	new_conn.close()
2728	n/a	l.close()
2729	n/a
2730	n/a	l = socket.socket()
2731	n/a	l.bind((test.support.HOST, 0))
2732	n/a	l.listen()
2733	n/a	conn.send(l.getsockname())
2734	n/a	new_conn, addr = l.accept()
2735	n/a	conn.send(new_conn)
2736	n/a	new_conn.close()
2737	n/a	l.close()
2738	n/a
2739	n/a	conn.recv()
2740	n/a
2741	n/a	@classmethod
2742	n/a	def _remote(cls, conn):
2743	n/a	for (address, msg) in iter(conn.recv, None):
2744	n/a	client = cls.connection.Client(address)
2745	n/a	client.send(msg.upper())
2746	n/a	client.close()
2747	n/a
2748	n/a	address, msg = conn.recv()
2749	n/a	client = socket.socket()
2750	n/a	client.connect(address)
2751	n/a	client.sendall(msg.upper())
2752	n/a	client.close()
2753	n/a
2754	n/a	conn.close()
2755	n/a
2756	n/a	def test_pickling(self):
2757	n/a	families = self.connection.families
2758	n/a
2759	n/a	lconn, lconn0 = self.Pipe()
2760	n/a	lp = self.Process(target=self._listener, args=(lconn0, families))
2761	n/a	lp.daemon = True
2762	n/a	lp.start()
2763	n/a	lconn0.close()
2764	n/a
2765	n/a	rconn, rconn0 = self.Pipe()
2766	n/a	rp = self.Process(target=self._remote, args=(rconn0,))
2767	n/a	rp.daemon = True
2768	n/a	rp.start()
2769	n/a	rconn0.close()
2770	n/a
2771	n/a	for fam in families:
2772	n/a	msg = ('This connection uses family %s' % fam).encode('ascii')
2773	n/a	address = lconn.recv()
2774	n/a	rconn.send((address, msg))
2775	n/a	new_conn = lconn.recv()
2776	n/a	self.assertEqual(new_conn.recv(), msg.upper())
2777	n/a
2778	n/a	rconn.send(None)
2779	n/a
2780	n/a	msg = latin('This connection uses a normal socket')
2781	n/a	address = lconn.recv()
2782	n/a	rconn.send((address, msg))
2783	n/a	new_conn = lconn.recv()
2784	n/a	buf = []
2785	n/a	while True:
2786	n/a	s = new_conn.recv(100)
2787	n/a	if not s:
2788	n/a	break
2789	n/a	buf.append(s)
2790	n/a	buf = b''.join(buf)
2791	n/a	self.assertEqual(buf, msg.upper())
2792	n/a	new_conn.close()
2793	n/a
2794	n/a	lconn.send(None)
2795	n/a
2796	n/a	rconn.close()
2797	n/a	lconn.close()
2798	n/a
2799	n/a	lp.join()
2800	n/a	rp.join()
2801	n/a
2802	n/a	@classmethod
2803	n/a	def child_access(cls, conn):
2804	n/a	w = conn.recv()
2805	n/a	w.send('all is well')
2806	n/a	w.close()
2807	n/a
2808	n/a	r = conn.recv()
2809	n/a	msg = r.recv()
2810	n/a	conn.send(msg*2)
2811	n/a
2812	n/a	conn.close()
2813	n/a
2814	n/a	def test_access(self):
2815	n/a	# On Windows, if we do not specify a destination pid when
2816	n/a	# using DupHandle then we need to be careful to use the
2817	n/a	# correct access flags for DuplicateHandle(), or else
2818	n/a	# DupHandle.detach() will raise PermissionError. For example,
2819	n/a	# for a read only pipe handle we should use
2820	n/a	# access=FILE_GENERIC_READ. (Unfortunately
2821	n/a	# DUPLICATE_SAME_ACCESS does not work.)
2822	n/a	conn, child_conn = self.Pipe()
2823	n/a	p = self.Process(target=self.child_access, args=(child_conn,))
2824	n/a	p.daemon = True
2825	n/a	p.start()
2826	n/a	child_conn.close()
2827	n/a
2828	n/a	r, w = self.Pipe(duplex=False)
2829	n/a	conn.send(w)
2830	n/a	w.close()
2831	n/a	self.assertEqual(r.recv(), 'all is well')
2832	n/a	r.close()
2833	n/a
2834	n/a	r, w = self.Pipe(duplex=False)
2835	n/a	conn.send(r)
2836	n/a	r.close()
2837	n/a	w.send('foobar')
2838	n/a	w.close()
2839	n/a	self.assertEqual(conn.recv(), 'foobar'*2)
2840	n/a
2841	n/a	#
2842	n/a	#
2843	n/a	#
2844	n/a
2845	n/a	class _TestHeap(BaseTestCase):
2846	n/a
2847	n/a	ALLOWED_TYPES = ('processes',)
2848	n/a
2849	n/a	def test_heap(self):
2850	n/a	iterations = 5000
2851	n/a	maxblocks = 50
2852	n/a	blocks = []
2853	n/a
2854	n/a	# create and destroy lots of blocks of different sizes
2855	n/a	for i in range(iterations):
2856	n/a	size = int(random.lognormvariate(0, 1) * 1000)
2857	n/a	b = multiprocessing.heap.BufferWrapper(size)
2858	n/a	blocks.append(b)
2859	n/a	if len(blocks) > maxblocks:
2860	n/a	i = random.randrange(maxblocks)
2861	n/a	del blocks[i]
2862	n/a
2863	n/a	# get the heap object
2864	n/a	heap = multiprocessing.heap.BufferWrapper._heap
2865	n/a
2866	n/a	# verify the state of the heap
2867	n/a	all = []
2868	n/a	occupied = 0
2869	n/a	heap._lock.acquire()
2870	n/a	self.addCleanup(heap._lock.release)
2871	n/a	for L in list(heap._len_to_seq.values()):
2872	n/a	for arena, start, stop in L:
2873	n/a	all.append((heap._arenas.index(arena), start, stop,
2874	n/a	stop-start, 'free'))
2875	n/a	for arena, start, stop in heap._allocated_blocks:
2876	n/a	all.append((heap._arenas.index(arena), start, stop,
2877	n/a	stop-start, 'occupied'))
2878	n/a	occupied += (stop-start)
2879	n/a
2880	n/a	all.sort()
2881	n/a
2882	n/a	for i in range(len(all)-1):
2883	n/a	(arena, start, stop) = all[i][:3]
2884	n/a	(narena, nstart, nstop) = all[i+1][:3]
2885	n/a	self.assertTrue((arena != narena and nstart == 0) or
2886	n/a	(stop == nstart))
2887	n/a
2888	n/a	def test_free_from_gc(self):
2889	n/a	# Check that freeing of blocks by the garbage collector doesn't deadlock
2890	n/a	# (issue #12352).
2891	n/a	# Make sure the GC is enabled, and set lower collection thresholds to
2892	n/a	# make collections more frequent (and increase the probability of
2893	n/a	# deadlock).
2894	n/a	if not gc.isenabled():
2895	n/a	gc.enable()
2896	n/a	self.addCleanup(gc.disable)
2897	n/a	thresholds = gc.get_threshold()
2898	n/a	self.addCleanup(gc.set_threshold, *thresholds)
2899	n/a	gc.set_threshold(10)
2900	n/a
2901	n/a	# perform numerous block allocations, with cyclic references to make
2902	n/a	# sure objects are collected asynchronously by the gc
2903	n/a	for i in range(5000):
2904	n/a	a = multiprocessing.heap.BufferWrapper(1)
2905	n/a	b = multiprocessing.heap.BufferWrapper(1)
2906	n/a	# circular references
2907	n/a	a.buddy = b
2908	n/a	b.buddy = a
2909	n/a
2910	n/a	#
2911	n/a	#
2912	n/a	#
2913	n/a
2914	n/a	class _Foo(Structure):
2915	n/a	_fields_ = [
2916	n/a	('x', c_int),
2917	n/a	('y', c_double)
2918	n/a	]
2919	n/a
2920	n/a	class _TestSharedCTypes(BaseTestCase):
2921	n/a
2922	n/a	ALLOWED_TYPES = ('processes',)
2923	n/a
2924	n/a	def setUp(self):
2925	n/a	if not HAS_SHAREDCTYPES:
2926	n/a	self.skipTest("requires multiprocessing.sharedctypes")
2927	n/a
2928	n/a	@classmethod
2929	n/a	def _double(cls, x, y, foo, arr, string):
2930	n/a	x.value *= 2
2931	n/a	y.value *= 2
2932	n/a	foo.x *= 2
2933	n/a	foo.y *= 2
2934	n/a	string.value *= 2
2935	n/a	for i in range(len(arr)):
2936	n/a	arr[i] *= 2
2937	n/a
2938	n/a	def test_sharedctypes(self, lock=False):
2939	n/a	x = Value('i', 7, lock=lock)
2940	n/a	y = Value(c_double, 1.0/3.0, lock=lock)
2941	n/a	foo = Value(_Foo, 3, 2, lock=lock)
2942	n/a	arr = self.Array('d', list(range(10)), lock=lock)
2943	n/a	string = self.Array('c', 20, lock=lock)
2944	n/a	string.value = latin('hello')
2945	n/a
2946	n/a	p = self.Process(target=self._double, args=(x, y, foo, arr, string))
2947	n/a	p.daemon = True
2948	n/a	p.start()
2949	n/a	p.join()
2950	n/a
2951	n/a	self.assertEqual(x.value, 14)
2952	n/a	self.assertAlmostEqual(y.value, 2.0/3.0)
2953	n/a	self.assertEqual(foo.x, 6)
2954	n/a	self.assertAlmostEqual(foo.y, 4.0)
2955	n/a	for i in range(10):
2956	n/a	self.assertAlmostEqual(arr[i], i*2)
2957	n/a	self.assertEqual(string.value, latin('hellohello'))
2958	n/a
2959	n/a	def test_synchronize(self):
2960	n/a	self.test_sharedctypes(lock=True)
2961	n/a
2962	n/a	def test_copy(self):
2963	n/a	foo = _Foo(2, 5.0)
2964	n/a	bar = copy(foo)
2965	n/a	foo.x = 0
2966	n/a	foo.y = 0
2967	n/a	self.assertEqual(bar.x, 2)
2968	n/a	self.assertAlmostEqual(bar.y, 5.0)
2969	n/a
2970	n/a	#
2971	n/a	#
2972	n/a	#
2973	n/a
2974	n/a	class _TestFinalize(BaseTestCase):
2975	n/a
2976	n/a	ALLOWED_TYPES = ('processes',)
2977	n/a
2978	n/a	@classmethod
2979	n/a	def _test_finalize(cls, conn):
2980	n/a	class Foo(object):
2981	n/a	pass
2982	n/a
2983	n/a	a = Foo()
2984	n/a	util.Finalize(a, conn.send, args=('a',))
2985	n/a	del a # triggers callback for a
2986	n/a
2987	n/a	b = Foo()
2988	n/a	close_b = util.Finalize(b, conn.send, args=('b',))
2989	n/a	close_b() # triggers callback for b
2990	n/a	close_b() # does nothing because callback has already been called
2991	n/a	del b # does nothing because callback has already been called
2992	n/a
2993	n/a	c = Foo()
2994	n/a	util.Finalize(c, conn.send, args=('c',))
2995	n/a
2996	n/a	d10 = Foo()
2997	n/a	util.Finalize(d10, conn.send, args=('d10',), exitpriority=1)
2998	n/a
2999	n/a	d01 = Foo()
3000	n/a	util.Finalize(d01, conn.send, args=('d01',), exitpriority=0)
3001	n/a	d02 = Foo()
3002	n/a	util.Finalize(d02, conn.send, args=('d02',), exitpriority=0)
3003	n/a	d03 = Foo()
3004	n/a	util.Finalize(d03, conn.send, args=('d03',), exitpriority=0)
3005	n/a
3006	n/a	util.Finalize(None, conn.send, args=('e',), exitpriority=-10)
3007	n/a
3008	n/a	util.Finalize(None, conn.send, args=('STOP',), exitpriority=-100)
3009	n/a
3010	n/a	# call multiprocessing's cleanup function then exit process without
3011	n/a	# garbage collecting locals
3012	n/a	util._exit_function()
3013	n/a	conn.close()
3014	n/a	os._exit(0)
3015	n/a
3016	n/a	def test_finalize(self):
3017	n/a	conn, child_conn = self.Pipe()
3018	n/a
3019	n/a	p = self.Process(target=self._test_finalize, args=(child_conn,))
3020	n/a	p.daemon = True
3021	n/a	p.start()
3022	n/a	p.join()
3023	n/a
3024	n/a	result = [obj for obj in iter(conn.recv, 'STOP')]
3025	n/a	self.assertEqual(result, ['a', 'b', 'd10', 'd03', 'd02', 'd01', 'e'])
3026	n/a
3027	n/a	#
3028	n/a	# Test that from ... import * works for each module
3029	n/a	#
3030	n/a
3031	n/a	class _TestImportStar(unittest.TestCase):
3032	n/a
3033	n/a	def get_module_names(self):
3034	n/a	import glob
3035	n/a	folder = os.path.dirname(multiprocessing.__file__)
3036	n/a	pattern = os.path.join(folder, '*.py')
3037	n/a	files = glob.glob(pattern)
3038	n/a	modules = [os.path.splitext(os.path.split(f)[1])[0] for f in files]
3039	n/a	modules = ['multiprocessing.' + m for m in modules]
3040	n/a	modules.remove('multiprocessing.__init__')
3041	n/a	modules.append('multiprocessing')
3042	n/a	return modules
3043	n/a
3044	n/a	def test_import(self):
3045	n/a	modules = self.get_module_names()
3046	n/a	if sys.platform == 'win32':
3047	n/a	modules.remove('multiprocessing.popen_fork')
3048	n/a	modules.remove('multiprocessing.popen_forkserver')
3049	n/a	modules.remove('multiprocessing.popen_spawn_posix')
3050	n/a	else:
3051	n/a	modules.remove('multiprocessing.popen_spawn_win32')
3052	n/a	if not HAS_REDUCTION:
3053	n/a	modules.remove('multiprocessing.popen_forkserver')
3054	n/a
3055	n/a	if c_int is None:
3056	n/a	# This module requires _ctypes
3057	n/a	modules.remove('multiprocessing.sharedctypes')
3058	n/a
3059	n/a	for name in modules:
3060	n/a	__import__(name)
3061	n/a	mod = sys.modules[name]
3062	n/a	self.assertTrue(hasattr(mod, '__all__'), name)
3063	n/a
3064	n/a	for attr in mod.__all__:
3065	n/a	self.assertTrue(
3066	n/a	hasattr(mod, attr),
3067	n/a	'%r does not have attribute %r' % (mod, attr)
3068	n/a	)
3069	n/a
3070	n/a	#
3071	n/a	# Quick test that logging works -- does not test logging output
3072	n/a	#
3073	n/a
3074	n/a	class _TestLogging(BaseTestCase):
3075	n/a
3076	n/a	ALLOWED_TYPES = ('processes',)
3077	n/a
3078	n/a	def test_enable_logging(self):
3079	n/a	logger = multiprocessing.get_logger()
3080	n/a	logger.setLevel(util.SUBWARNING)
3081	n/a	self.assertTrue(logger is not None)
3082	n/a	logger.debug('this will not be printed')
3083	n/a	logger.info('nor will this')
3084	n/a	logger.setLevel(LOG_LEVEL)
3085	n/a
3086	n/a	@classmethod
3087	n/a	def _test_level(cls, conn):
3088	n/a	logger = multiprocessing.get_logger()
3089	n/a	conn.send(logger.getEffectiveLevel())
3090	n/a
3091	n/a	def test_level(self):
3092	n/a	LEVEL1 = 32
3093	n/a	LEVEL2 = 37
3094	n/a
3095	n/a	logger = multiprocessing.get_logger()
3096	n/a	root_logger = logging.getLogger()
3097	n/a	root_level = root_logger.level
3098	n/a
3099	n/a	reader, writer = multiprocessing.Pipe(duplex=False)
3100	n/a
3101	n/a	logger.setLevel(LEVEL1)
3102	n/a	p = self.Process(target=self._test_level, args=(writer,))
3103	n/a	p.daemon = True
3104	n/a	p.start()
3105	n/a	self.assertEqual(LEVEL1, reader.recv())
3106	n/a
3107	n/a	logger.setLevel(logging.NOTSET)
3108	n/a	root_logger.setLevel(LEVEL2)
3109	n/a	p = self.Process(target=self._test_level, args=(writer,))
3110	n/a	p.daemon = True
3111	n/a	p.start()
3112	n/a	self.assertEqual(LEVEL2, reader.recv())
3113	n/a
3114	n/a	root_logger.setLevel(root_level)
3115	n/a	logger.setLevel(level=LOG_LEVEL)
3116	n/a
3117	n/a
3118	n/a	# class _TestLoggingProcessName(BaseTestCase):
3119	n/a	#
3120	n/a	# def handle(self, record):
3121	n/a	# assert record.processName == multiprocessing.current_process().name
3122	n/a	# self.__handled = True
3123	n/a	#
3124	n/a	# def test_logging(self):
3125	n/a	# handler = logging.Handler()
3126	n/a	# handler.handle = self.handle
3127	n/a	# self.__handled = False
3128	n/a	# # Bypass getLogger() and side-effects
3129	n/a	# logger = logging.getLoggerClass()(
3130	n/a	# 'multiprocessing.test.TestLoggingProcessName')
3131	n/a	# logger.addHandler(handler)
3132	n/a	# logger.propagate = False
3133	n/a	#
3134	n/a	# logger.warn('foo')
3135	n/a	# assert self.__handled
3136	n/a
3137	n/a	#
3138	n/a	# Check that Process.join() retries if os.waitpid() fails with EINTR
3139	n/a	#
3140	n/a
3141	n/a	class _TestPollEintr(BaseTestCase):
3142	n/a
3143	n/a	ALLOWED_TYPES = ('processes',)
3144	n/a
3145	n/a	@classmethod
3146	n/a	def _killer(cls, pid):
3147	n/a	time.sleep(0.1)
3148	n/a	os.kill(pid, signal.SIGUSR1)
3149	n/a
3150	n/a	@unittest.skipUnless(hasattr(signal, 'SIGUSR1'), 'requires SIGUSR1')
3151	n/a	def test_poll_eintr(self):
3152	n/a	got_signal = [False]
3153	n/a	def record(*args):
3154	n/a	got_signal[0] = True
3155	n/a	pid = os.getpid()
3156	n/a	oldhandler = signal.signal(signal.SIGUSR1, record)
3157	n/a	try:
3158	n/a	killer = self.Process(target=self._killer, args=(pid,))
3159	n/a	killer.start()
3160	n/a	try:
3161	n/a	p = self.Process(target=time.sleep, args=(2,))
3162	n/a	p.start()
3163	n/a	p.join()
3164	n/a	finally:
3165	n/a	killer.join()
3166	n/a	self.assertTrue(got_signal[0])
3167	n/a	self.assertEqual(p.exitcode, 0)
3168	n/a	finally:
3169	n/a	signal.signal(signal.SIGUSR1, oldhandler)
3170	n/a
3171	n/a	#
3172	n/a	# Test to verify handle verification, see issue 3321
3173	n/a	#
3174	n/a
3175	n/a	class TestInvalidHandle(unittest.TestCase):
3176	n/a
3177	n/a	@unittest.skipIf(WIN32, "skipped on Windows")
3178	n/a	def test_invalid_handles(self):
3179	n/a	conn = multiprocessing.connection.Connection(44977608)
3180	n/a	# check that poll() doesn't crash
3181	n/a	try:
3182	n/a	conn.poll()
3183	n/a	except (ValueError, OSError):
3184	n/a	pass
3185	n/a	finally:
3186	n/a	# Hack private attribute _handle to avoid printing an error
3187	n/a	# in conn.__del__
3188	n/a	conn._handle = None
3189	n/a	self.assertRaises((ValueError, OSError),
3190	n/a	multiprocessing.connection.Connection, -1)
3191	n/a
3192	n/a
3193	n/a
3194	n/a	class OtherTest(unittest.TestCase):
3195	n/a	# TODO: add more tests for deliver/answer challenge.
3196	n/a	def test_deliver_challenge_auth_failure(self):
3197	n/a	class _FakeConnection(object):
3198	n/a	def recv_bytes(self, size):
3199	n/a	return b'something bogus'
3200	n/a	def send_bytes(self, data):
3201	n/a	pass
3202	n/a	self.assertRaises(multiprocessing.AuthenticationError,
3203	n/a	multiprocessing.connection.deliver_challenge,
3204	n/a	_FakeConnection(), b'abc')
3205	n/a
3206	n/a	def test_answer_challenge_auth_failure(self):
3207	n/a	class _FakeConnection(object):
3208	n/a	def __init__(self):
3209	n/a	self.count = 0
3210	n/a	def recv_bytes(self, size):
3211	n/a	self.count += 1
3212	n/a	if self.count == 1:
3213	n/a	return multiprocessing.connection.CHALLENGE
3214	n/a	elif self.count == 2:
3215	n/a	return b'something bogus'
3216	n/a	return b''
3217	n/a	def send_bytes(self, data):
3218	n/a	pass
3219	n/a	self.assertRaises(multiprocessing.AuthenticationError,
3220	n/a	multiprocessing.connection.answer_challenge,
3221	n/a	_FakeConnection(), b'abc')
3222	n/a
3223	n/a	#
3224	n/a	# Test Manager.start()/Pool.__init__() initializer feature - see issue 5585
3225	n/a	#
3226	n/a
3227	n/a	def initializer(ns):
3228	n/a	ns.test += 1
3229	n/a
3230	n/a	class TestInitializers(unittest.TestCase):
3231	n/a	def setUp(self):
3232	n/a	self.mgr = multiprocessing.Manager()
3233	n/a	self.ns = self.mgr.Namespace()
3234	n/a	self.ns.test = 0
3235	n/a
3236	n/a	def tearDown(self):
3237	n/a	self.mgr.shutdown()
3238	n/a	self.mgr.join()
3239	n/a
3240	n/a	def test_manager_initializer(self):
3241	n/a	m = multiprocessing.managers.SyncManager()
3242	n/a	self.assertRaises(TypeError, m.start, 1)
3243	n/a	m.start(initializer, (self.ns,))
3244	n/a	self.assertEqual(self.ns.test, 1)
3245	n/a	m.shutdown()
3246	n/a	m.join()
3247	n/a
3248	n/a	def test_pool_initializer(self):
3249	n/a	self.assertRaises(TypeError, multiprocessing.Pool, initializer=1)
3250	n/a	p = multiprocessing.Pool(1, initializer, (self.ns,))
3251	n/a	p.close()
3252	n/a	p.join()
3253	n/a	self.assertEqual(self.ns.test, 1)
3254	n/a
3255	n/a	#
3256	n/a	# Issue 5155, 5313, 5331: Test process in processes
3257	n/a	# Verifies os.close(sys.stdin.fileno) vs. sys.stdin.close() behavior
3258	n/a	#
3259	n/a
3260	n/a	def _this_sub_process(q):
3261	n/a	try:
3262	n/a	item = q.get(block=False)
3263	n/a	except pyqueue.Empty:
3264	n/a	pass
3265	n/a
3266	n/a	def _test_process(q):
3267	n/a	queue = multiprocessing.Queue()
3268	n/a	subProc = multiprocessing.Process(target=_this_sub_process, args=(queue,))
3269	n/a	subProc.daemon = True
3270	n/a	subProc.start()
3271	n/a	subProc.join()
3272	n/a
3273	n/a	def _afunc(x):
3274	n/a	return x*x
3275	n/a
3276	n/a	def pool_in_process():
3277	n/a	pool = multiprocessing.Pool(processes=4)
3278	n/a	x = pool.map(_afunc, [1, 2, 3, 4, 5, 6, 7])
3279	n/a	pool.close()
3280	n/a	pool.join()
3281	n/a
3282	n/a	class _file_like(object):
3283	n/a	def __init__(self, delegate):
3284	n/a	self._delegate = delegate
3285	n/a	self._pid = None
3286	n/a
3287	n/a	@property
3288	n/a	def cache(self):
3289	n/a	pid = os.getpid()
3290	n/a	# There are no race conditions since fork keeps only the running thread
3291	n/a	if pid != self._pid:
3292	n/a	self._pid = pid
3293	n/a	self._cache = []
3294	n/a	return self._cache
3295	n/a
3296	n/a	def write(self, data):
3297	n/a	self.cache.append(data)
3298	n/a
3299	n/a	def flush(self):
3300	n/a	self._delegate.write(''.join(self.cache))
3301	n/a	self._cache = []
3302	n/a
3303	n/a	class TestStdinBadfiledescriptor(unittest.TestCase):
3304	n/a
3305	n/a	def test_queue_in_process(self):
3306	n/a	queue = multiprocessing.Queue()
3307	n/a	proc = multiprocessing.Process(target=_test_process, args=(queue,))
3308	n/a	proc.start()
3309	n/a	proc.join()
3310	n/a
3311	n/a	def test_pool_in_process(self):
3312	n/a	p = multiprocessing.Process(target=pool_in_process)
3313	n/a	p.start()
3314	n/a	p.join()
3315	n/a
3316	n/a	def test_flushing(self):
3317	n/a	sio = io.StringIO()
3318	n/a	flike = _file_like(sio)
3319	n/a	flike.write('foo')
3320	n/a	proc = multiprocessing.Process(target=lambda: flike.flush())
3321	n/a	flike.flush()
3322	n/a	assert sio.getvalue() == 'foo'
3323	n/a
3324	n/a
3325	n/a	class TestWait(unittest.TestCase):
3326	n/a
3327	n/a	@classmethod
3328	n/a	def _child_test_wait(cls, w, slow):
3329	n/a	for i in range(10):
3330	n/a	if slow:
3331	n/a	time.sleep(random.random()*0.1)
3332	n/a	w.send((i, os.getpid()))
3333	n/a	w.close()
3334	n/a
3335	n/a	def test_wait(self, slow=False):
3336	n/a	from multiprocessing.connection import wait
3337	n/a	readers = []
3338	n/a	procs = []
3339	n/a	messages = []
3340	n/a
3341	n/a	for i in range(4):
3342	n/a	r, w = multiprocessing.Pipe(duplex=False)
3343	n/a	p = multiprocessing.Process(target=self._child_test_wait, args=(w, slow))
3344	n/a	p.daemon = True
3345	n/a	p.start()
3346	n/a	w.close()
3347	n/a	readers.append(r)
3348	n/a	procs.append(p)
3349	n/a	self.addCleanup(p.join)
3350	n/a
3351	n/a	while readers:
3352	n/a	for r in wait(readers):
3353	n/a	try:
3354	n/a	msg = r.recv()
3355	n/a	except EOFError:
3356	n/a	readers.remove(r)
3357	n/a	r.close()
3358	n/a	else:
3359	n/a	messages.append(msg)
3360	n/a
3361	n/a	messages.sort()
3362	n/a	expected = sorted((i, p.pid) for i in range(10) for p in procs)
3363	n/a	self.assertEqual(messages, expected)
3364	n/a
3365	n/a	@classmethod
3366	n/a	def _child_test_wait_socket(cls, address, slow):
3367	n/a	s = socket.socket()
3368	n/a	s.connect(address)
3369	n/a	for i in range(10):
3370	n/a	if slow:
3371	n/a	time.sleep(random.random()*0.1)
3372	n/a	s.sendall(('%s\n' % i).encode('ascii'))
3373	n/a	s.close()
3374	n/a
3375	n/a	def test_wait_socket(self, slow=False):
3376	n/a	from multiprocessing.connection import wait
3377	n/a	l = socket.socket()
3378	n/a	l.bind((test.support.HOST, 0))
3379	n/a	l.listen()
3380	n/a	addr = l.getsockname()
3381	n/a	readers = []
3382	n/a	procs = []
3383	n/a	dic = {}
3384	n/a
3385	n/a	for i in range(4):
3386	n/a	p = multiprocessing.Process(target=self._child_test_wait_socket,
3387	n/a	args=(addr, slow))
3388	n/a	p.daemon = True
3389	n/a	p.start()
3390	n/a	procs.append(p)
3391	n/a	self.addCleanup(p.join)
3392	n/a
3393	n/a	for i in range(4):
3394	n/a	r, _ = l.accept()
3395	n/a	readers.append(r)
3396	n/a	dic[r] = []
3397	n/a	l.close()
3398	n/a
3399	n/a	while readers:
3400	n/a	for r in wait(readers):
3401	n/a	msg = r.recv(32)
3402	n/a	if not msg:
3403	n/a	readers.remove(r)
3404	n/a	r.close()
3405	n/a	else:
3406	n/a	dic[r].append(msg)
3407	n/a
3408	n/a	expected = ''.join('%s\n' % i for i in range(10)).encode('ascii')
3409	n/a	for v in dic.values():
3410	n/a	self.assertEqual(b''.join(v), expected)
3411	n/a
3412	n/a	def test_wait_slow(self):
3413	n/a	self.test_wait(True)
3414	n/a
3415	n/a	def test_wait_socket_slow(self):
3416	n/a	self.test_wait_socket(True)
3417	n/a
3418	n/a	def test_wait_timeout(self):
3419	n/a	from multiprocessing.connection import wait
3420	n/a
3421	n/a	expected = 5
3422	n/a	a, b = multiprocessing.Pipe()
3423	n/a
3424	n/a	start = time.time()
3425	n/a	res = wait([a, b], expected)
3426	n/a	delta = time.time() - start
3427	n/a
3428	n/a	self.assertEqual(res, [])
3429	n/a	self.assertLess(delta, expected * 2)
3430	n/a	self.assertGreater(delta, expected * 0.5)
3431	n/a
3432	n/a	b.send(None)
3433	n/a
3434	n/a	start = time.time()
3435	n/a	res = wait([a, b], 20)
3436	n/a	delta = time.time() - start
3437	n/a
3438	n/a	self.assertEqual(res, [a])
3439	n/a	self.assertLess(delta, 0.4)
3440	n/a
3441	n/a	@classmethod
3442	n/a	def signal_and_sleep(cls, sem, period):
3443	n/a	sem.release()
3444	n/a	time.sleep(period)
3445	n/a
3446	n/a	def test_wait_integer(self):
3447	n/a	from multiprocessing.connection import wait
3448	n/a
3449	n/a	expected = 3
3450	n/a	sorted_ = lambda l: sorted(l, key=lambda x: id(x))
3451	n/a	sem = multiprocessing.Semaphore(0)
3452	n/a	a, b = multiprocessing.Pipe()
3453	n/a	p = multiprocessing.Process(target=self.signal_and_sleep,
3454	n/a	args=(sem, expected))
3455	n/a
3456	n/a	p.start()
3457	n/a	self.assertIsInstance(p.sentinel, int)
3458	n/a	self.assertTrue(sem.acquire(timeout=20))
3459	n/a
3460	n/a	start = time.time()
3461	n/a	res = wait([a, p.sentinel, b], expected + 20)
3462	n/a	delta = time.time() - start
3463	n/a
3464	n/a	self.assertEqual(res, [p.sentinel])
3465	n/a	self.assertLess(delta, expected + 2)
3466	n/a	self.assertGreater(delta, expected - 2)
3467	n/a
3468	n/a	a.send(None)
3469	n/a
3470	n/a	start = time.time()
3471	n/a	res = wait([a, p.sentinel, b], 20)
3472	n/a	delta = time.time() - start
3473	n/a
3474	n/a	self.assertEqual(sorted_(res), sorted_([p.sentinel, b]))
3475	n/a	self.assertLess(delta, 0.4)
3476	n/a
3477	n/a	b.send(None)
3478	n/a
3479	n/a	start = time.time()
3480	n/a	res = wait([a, p.sentinel, b], 20)
3481	n/a	delta = time.time() - start
3482	n/a
3483	n/a	self.assertEqual(sorted_(res), sorted_([a, p.sentinel, b]))
3484	n/a	self.assertLess(delta, 0.4)
3485	n/a
3486	n/a	p.terminate()
3487	n/a	p.join()
3488	n/a
3489	n/a	def test_neg_timeout(self):
3490	n/a	from multiprocessing.connection import wait
3491	n/a	a, b = multiprocessing.Pipe()
3492	n/a	t = time.time()
3493	n/a	res = wait([a], timeout=-1)
3494	n/a	t = time.time() - t
3495	n/a	self.assertEqual(res, [])
3496	n/a	self.assertLess(t, 1)
3497	n/a	a.close()
3498	n/a	b.close()
3499	n/a
3500	n/a	#
3501	n/a	# Issue 14151: Test invalid family on invalid environment
3502	n/a	#
3503	n/a
3504	n/a	class TestInvalidFamily(unittest.TestCase):
3505	n/a
3506	n/a	@unittest.skipIf(WIN32, "skipped on Windows")
3507	n/a	def test_invalid_family(self):
3508	n/a	with self.assertRaises(ValueError):
3509	n/a	multiprocessing.connection.Listener(r'\\.\test')
3510	n/a
3511	n/a	@unittest.skipUnless(WIN32, "skipped on non-Windows platforms")
3512	n/a	def test_invalid_family_win32(self):
3513	n/a	with self.assertRaises(ValueError):
3514	n/a	multiprocessing.connection.Listener('/var/test.pipe')
3515	n/a
3516	n/a	#
3517	n/a	# Issue 12098: check sys.flags of child matches that for parent
3518	n/a	#
3519	n/a
3520	n/a	class TestFlags(unittest.TestCase):
3521	n/a	@classmethod
3522	n/a	def run_in_grandchild(cls, conn):
3523	n/a	conn.send(tuple(sys.flags))
3524	n/a
3525	n/a	@classmethod
3526	n/a	def run_in_child(cls):
3527	n/a	import json
3528	n/a	r, w = multiprocessing.Pipe(duplex=False)
3529	n/a	p = multiprocessing.Process(target=cls.run_in_grandchild, args=(w,))
3530	n/a	p.start()
3531	n/a	grandchild_flags = r.recv()
3532	n/a	p.join()
3533	n/a	r.close()
3534	n/a	w.close()
3535	n/a	flags = (tuple(sys.flags), grandchild_flags)
3536	n/a	print(json.dumps(flags))
3537	n/a
3538	n/a	def test_flags(self):
3539	n/a	import json, subprocess
3540	n/a	# start child process using unusual flags
3541	n/a	prog = ('from test._test_multiprocessing import TestFlags; ' +
3542	n/a	'TestFlags.run_in_child()')
3543	n/a	data = subprocess.check_output(
3544	n/a	[sys.executable, '-E', '-S', '-O', '-c', prog])
3545	n/a	child_flags, grandchild_flags = json.loads(data.decode('ascii'))
3546	n/a	self.assertEqual(child_flags, grandchild_flags)
3547	n/a
3548	n/a	#
3549	n/a	# Test interaction with socket timeouts - see Issue #6056
3550	n/a	#
3551	n/a
3552	n/a	class TestTimeouts(unittest.TestCase):
3553	n/a	@classmethod
3554	n/a	def _test_timeout(cls, child, address):
3555	n/a	time.sleep(1)
3556	n/a	child.send(123)
3557	n/a	child.close()
3558	n/a	conn = multiprocessing.connection.Client(address)
3559	n/a	conn.send(456)
3560	n/a	conn.close()
3561	n/a
3562	n/a	def test_timeout(self):
3563	n/a	old_timeout = socket.getdefaulttimeout()
3564	n/a	try:
3565	n/a	socket.setdefaulttimeout(0.1)
3566	n/a	parent, child = multiprocessing.Pipe(duplex=True)
3567	n/a	l = multiprocessing.connection.Listener(family='AF_INET')
3568	n/a	p = multiprocessing.Process(target=self._test_timeout,
3569	n/a	args=(child, l.address))
3570	n/a	p.start()
3571	n/a	child.close()
3572	n/a	self.assertEqual(parent.recv(), 123)
3573	n/a	parent.close()
3574	n/a	conn = l.accept()
3575	n/a	self.assertEqual(conn.recv(), 456)
3576	n/a	conn.close()
3577	n/a	l.close()
3578	n/a	p.join(10)
3579	n/a	finally:
3580	n/a	socket.setdefaulttimeout(old_timeout)
3581	n/a
3582	n/a	#
3583	n/a	# Test what happens with no "if __name__ == '__main__'"
3584	n/a	#
3585	n/a
3586	n/a	class TestNoForkBomb(unittest.TestCase):
3587	n/a	def test_noforkbomb(self):
3588	n/a	sm = multiprocessing.get_start_method()
3589	n/a	name = os.path.join(os.path.dirname(__file__), 'mp_fork_bomb.py')
3590	n/a	if sm != 'fork':
3591	n/a	rc, out, err = test.support.script_helper.assert_python_failure(name, sm)
3592	n/a	self.assertEqual(out, b'')
3593	n/a	self.assertIn(b'RuntimeError', err)
3594	n/a	else:
3595	n/a	rc, out, err = test.support.script_helper.assert_python_ok(name, sm)
3596	n/a	self.assertEqual(out.rstrip(), b'123')
3597	n/a	self.assertEqual(err, b'')
3598	n/a
3599	n/a	#
3600	n/a	# Issue #17555: ForkAwareThreadLock
3601	n/a	#
3602	n/a
3603	n/a	class TestForkAwareThreadLock(unittest.TestCase):
3604	n/a	# We recurisvely start processes. Issue #17555 meant that the
3605	n/a	# after fork registry would get duplicate entries for the same
3606	n/a	# lock. The size of the registry at generation n was ~2**n.
3607	n/a
3608	n/a	@classmethod
3609	n/a	def child(cls, n, conn):
3610	n/a	if n > 1:
3611	n/a	p = multiprocessing.Process(target=cls.child, args=(n-1, conn))
3612	n/a	p.start()
3613	n/a	conn.close()
3614	n/a	p.join(timeout=5)
3615	n/a	else:
3616	n/a	conn.send(len(util._afterfork_registry))
3617	n/a	conn.close()
3618	n/a
3619	n/a	def test_lock(self):
3620	n/a	r, w = multiprocessing.Pipe(False)
3621	n/a	l = util.ForkAwareThreadLock()
3622	n/a	old_size = len(util._afterfork_registry)
3623	n/a	p = multiprocessing.Process(target=self.child, args=(5, w))
3624	n/a	p.start()
3625	n/a	w.close()
3626	n/a	new_size = r.recv()
3627	n/a	p.join(timeout=5)
3628	n/a	self.assertLessEqual(new_size, old_size)
3629	n/a
3630	n/a	#
3631	n/a	# Check that non-forked child processes do not inherit unneeded fds/handles
3632	n/a	#
3633	n/a
3634	n/a	class TestCloseFds(unittest.TestCase):
3635	n/a
3636	n/a	def get_high_socket_fd(self):
3637	n/a	if WIN32:
3638	n/a	# The child process will not have any socket handles, so
3639	n/a	# calling socket.fromfd() should produce WSAENOTSOCK even
3640	n/a	# if there is a handle of the same number.
3641	n/a	return socket.socket().detach()
3642	n/a	else:
3643	n/a	# We want to produce a socket with an fd high enough that a
3644	n/a	# freshly created child process will not have any fds as high.
3645	n/a	fd = socket.socket().detach()
3646	n/a	to_close = []
3647	n/a	while fd < 50:
3648	n/a	to_close.append(fd)
3649	n/a	fd = os.dup(fd)
3650	n/a	for x in to_close:
3651	n/a	os.close(x)
3652	n/a	return fd
3653	n/a
3654	n/a	def close(self, fd):
3655	n/a	if WIN32:
3656	n/a	socket.socket(fileno=fd).close()
3657	n/a	else:
3658	n/a	os.close(fd)
3659	n/a
3660	n/a	@classmethod
3661	n/a	def _test_closefds(cls, conn, fd):
3662	n/a	try:
3663	n/a	s = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM)
3664	n/a	except Exception as e:
3665	n/a	conn.send(e)
3666	n/a	else:
3667	n/a	s.close()
3668	n/a	conn.send(None)
3669	n/a
3670	n/a	def test_closefd(self):
3671	n/a	if not HAS_REDUCTION:
3672	n/a	raise unittest.SkipTest('requires fd pickling')
3673	n/a
3674	n/a	reader, writer = multiprocessing.Pipe()
3675	n/a	fd = self.get_high_socket_fd()
3676	n/a	try:
3677	n/a	p = multiprocessing.Process(target=self._test_closefds,
3678	n/a	args=(writer, fd))
3679	n/a	p.start()
3680	n/a	writer.close()
3681	n/a	e = reader.recv()
3682	n/a	p.join(timeout=5)
3683	n/a	finally:
3684	n/a	self.close(fd)
3685	n/a	writer.close()
3686	n/a	reader.close()
3687	n/a
3688	n/a	if multiprocessing.get_start_method() == 'fork':
3689	n/a	self.assertIs(e, None)
3690	n/a	else:
3691	n/a	WSAENOTSOCK = 10038
3692	n/a	self.assertIsInstance(e, OSError)
3693	n/a	self.assertTrue(e.errno == errno.EBADF or
3694	n/a	e.winerror == WSAENOTSOCK, e)
3695	n/a
3696	n/a	#
3697	n/a	# Issue #17097: EINTR should be ignored by recv(), send(), accept() etc
3698	n/a	#
3699	n/a
3700	n/a	class TestIgnoreEINTR(unittest.TestCase):
3701	n/a
3702	n/a	@classmethod
3703	n/a	def _test_ignore(cls, conn):
3704	n/a	def handler(signum, frame):
3705	n/a	pass
3706	n/a	signal.signal(signal.SIGUSR1, handler)
3707	n/a	conn.send('ready')
3708	n/a	x = conn.recv()
3709	n/a	conn.send(x)
3710	n/a	conn.send_bytes(b'x'(10241024)) # sending 1 MB should block
3711	n/a
3712	n/a	@unittest.skipUnless(hasattr(signal, 'SIGUSR1'), 'requires SIGUSR1')
3713	n/a	def test_ignore(self):
3714	n/a	conn, child_conn = multiprocessing.Pipe()
3715	n/a	try:
3716	n/a	p = multiprocessing.Process(target=self._test_ignore,
3717	n/a	args=(child_conn,))
3718	n/a	p.daemon = True
3719	n/a	p.start()
3720	n/a	child_conn.close()
3721	n/a	self.assertEqual(conn.recv(), 'ready')
3722	n/a	time.sleep(0.1)
3723	n/a	os.kill(p.pid, signal.SIGUSR1)
3724	n/a	time.sleep(0.1)
3725	n/a	conn.send(1234)
3726	n/a	self.assertEqual(conn.recv(), 1234)
3727	n/a	time.sleep(0.1)
3728	n/a	os.kill(p.pid, signal.SIGUSR1)
3729	n/a	self.assertEqual(conn.recv_bytes(), b'x'(10241024))
3730	n/a	time.sleep(0.1)
3731	n/a	p.join()
3732	n/a	finally:
3733	n/a	conn.close()
3734	n/a
3735	n/a	@classmethod
3736	n/a	def _test_ignore_listener(cls, conn):
3737	n/a	def handler(signum, frame):
3738	n/a	pass
3739	n/a	signal.signal(signal.SIGUSR1, handler)
3740	n/a	with multiprocessing.connection.Listener() as l:
3741	n/a	conn.send(l.address)
3742	n/a	a = l.accept()
3743	n/a	a.send('welcome')
3744	n/a
3745	n/a	@unittest.skipUnless(hasattr(signal, 'SIGUSR1'), 'requires SIGUSR1')
3746	n/a	def test_ignore_listener(self):
3747	n/a	conn, child_conn = multiprocessing.Pipe()
3748	n/a	try:
3749	n/a	p = multiprocessing.Process(target=self._test_ignore_listener,
3750	n/a	args=(child_conn,))
3751	n/a	p.daemon = True
3752	n/a	p.start()
3753	n/a	child_conn.close()
3754	n/a	address = conn.recv()
3755	n/a	time.sleep(0.1)
3756	n/a	os.kill(p.pid, signal.SIGUSR1)
3757	n/a	time.sleep(0.1)
3758	n/a	client = multiprocessing.connection.Client(address)
3759	n/a	self.assertEqual(client.recv(), 'welcome')
3760	n/a	p.join()
3761	n/a	finally:
3762	n/a	conn.close()
3763	n/a
3764	n/a	class TestStartMethod(unittest.TestCase):
3765	n/a	@classmethod
3766	n/a	def _check_context(cls, conn):
3767	n/a	conn.send(multiprocessing.get_start_method())
3768	n/a
3769	n/a	def check_context(self, ctx):
3770	n/a	r, w = ctx.Pipe(duplex=False)
3771	n/a	p = ctx.Process(target=self._check_context, args=(w,))
3772	n/a	p.start()
3773	n/a	w.close()
3774	n/a	child_method = r.recv()
3775	n/a	r.close()
3776	n/a	p.join()
3777	n/a	self.assertEqual(child_method, ctx.get_start_method())
3778	n/a
3779	n/a	def test_context(self):
3780	n/a	for method in ('fork', 'spawn', 'forkserver'):
3781	n/a	try:
3782	n/a	ctx = multiprocessing.get_context(method)
3783	n/a	except ValueError:
3784	n/a	continue
3785	n/a	self.assertEqual(ctx.get_start_method(), method)
3786	n/a	self.assertIs(ctx.get_context(), ctx)
3787	n/a	self.assertRaises(ValueError, ctx.set_start_method, 'spawn')
3788	n/a	self.assertRaises(ValueError, ctx.set_start_method, None)
3789	n/a	self.check_context(ctx)
3790	n/a
3791	n/a	def test_set_get(self):
3792	n/a	multiprocessing.set_forkserver_preload(PRELOAD)
3793	n/a	count = 0
3794	n/a	old_method = multiprocessing.get_start_method()
3795	n/a	try:
3796	n/a	for method in ('fork', 'spawn', 'forkserver'):
3797	n/a	try:
3798	n/a	multiprocessing.set_start_method(method, force=True)
3799	n/a	except ValueError:
3800	n/a	continue
3801	n/a	self.assertEqual(multiprocessing.get_start_method(), method)
3802	n/a	ctx = multiprocessing.get_context()
3803	n/a	self.assertEqual(ctx.get_start_method(), method)
3804	n/a	self.assertTrue(type(ctx).__name__.lower().startswith(method))
3805	n/a	self.assertTrue(
3806	n/a	ctx.Process.__name__.lower().startswith(method))
3807	n/a	self.check_context(multiprocessing)
3808	n/a	count += 1
3809	n/a	finally:
3810	n/a	multiprocessing.set_start_method(old_method, force=True)
3811	n/a	self.assertGreaterEqual(count, 1)
3812	n/a
3813	n/a	def test_get_all(self):
3814	n/a	methods = multiprocessing.get_all_start_methods()
3815	n/a	if sys.platform == 'win32':
3816	n/a	self.assertEqual(methods, ['spawn'])
3817	n/a	else:
3818	n/a	self.assertTrue(methods == ['fork', 'spawn'] or
3819	n/a	methods == ['fork', 'spawn', 'forkserver'])
3820	n/a
3821	n/a	def test_preload_resources(self):
3822	n/a	if multiprocessing.get_start_method() != 'forkserver':
3823	n/a	self.skipTest("test only relevant for 'forkserver' method")
3824	n/a	name = os.path.join(os.path.dirname(__file__), 'mp_preload.py')
3825	n/a	rc, out, err = test.support.script_helper.assert_python_ok(name)
3826	n/a	out = out.decode()
3827	n/a	err = err.decode()
3828	n/a	if out.rstrip() != 'ok' or err != '':
3829	n/a	print(out)
3830	n/a	print(err)
3831	n/a	self.fail("failed spawning forkserver or grandchild")
3832	n/a
3833	n/a
3834	n/a	#
3835	n/a	# Check that killing process does not leak named semaphores
3836	n/a	#
3837	n/a
3838	n/a	@unittest.skipIf(sys.platform == "win32",
3839	n/a	"test semantics don't make sense on Windows")
3840	n/a	class TestSemaphoreTracker(unittest.TestCase):
3841	n/a	def test_semaphore_tracker(self):
3842	n/a	import subprocess
3843	n/a	cmd = '''if 1:
3844	n/a	import multiprocessing as mp, time, os
3845	n/a	mp.set_start_method("spawn")
3846	n/a	lock1 = mp.Lock()
3847	n/a	lock2 = mp.Lock()
3848	n/a	os.write(%d, lock1._semlock.name.encode("ascii") + b"\\n")
3849	n/a	os.write(%d, lock2._semlock.name.encode("ascii") + b"\\n")
3850	n/a	time.sleep(10)
3851	n/a	'''
3852	n/a	r, w = os.pipe()
3853	n/a	p = subprocess.Popen([sys.executable,
3854	n/a	'-c', cmd % (w, w)],
3855	n/a	pass_fds=[w],
3856	n/a	stderr=subprocess.PIPE)
3857	n/a	os.close(w)
3858	n/a	with open(r, 'rb', closefd=True) as f:
3859	n/a	name1 = f.readline().rstrip().decode('ascii')
3860	n/a	name2 = f.readline().rstrip().decode('ascii')
3861	n/a	_multiprocessing.sem_unlink(name1)
3862	n/a	p.terminate()
3863	n/a	p.wait()
3864	n/a	time.sleep(2.0)
3865	n/a	with self.assertRaises(OSError) as ctx:
3866	n/a	_multiprocessing.sem_unlink(name2)
3867	n/a	# docs say it should be ENOENT, but OSX seems to give EINVAL
3868	n/a	self.assertIn(ctx.exception.errno, (errno.ENOENT, errno.EINVAL))
3869	n/a	err = p.stderr.read().decode('utf-8')
3870	n/a	p.stderr.close()
3871	n/a	expected = 'semaphore_tracker: There appear to be 2 leaked semaphores'
3872	n/a	self.assertRegex(err, expected)
3873	n/a	self.assertRegex(err, r'semaphore_tracker: %r: \[Errno' % name1)
3874	n/a
3875	n/a	#
3876	n/a	# Mixins
3877	n/a	#
3878	n/a
3879	n/a	class ProcessesMixin(object):
3880	n/a	TYPE = 'processes'
3881	n/a	Process = multiprocessing.Process
3882	n/a	connection = multiprocessing.connection
3883	n/a	current_process = staticmethod(multiprocessing.current_process)
3884	n/a	active_children = staticmethod(multiprocessing.active_children)
3885	n/a	Pool = staticmethod(multiprocessing.Pool)
3886	n/a	Pipe = staticmethod(multiprocessing.Pipe)
3887	n/a	Queue = staticmethod(multiprocessing.Queue)
3888	n/a	JoinableQueue = staticmethod(multiprocessing.JoinableQueue)
3889	n/a	Lock = staticmethod(multiprocessing.Lock)
3890	n/a	RLock = staticmethod(multiprocessing.RLock)
3891	n/a	Semaphore = staticmethod(multiprocessing.Semaphore)
3892	n/a	BoundedSemaphore = staticmethod(multiprocessing.BoundedSemaphore)
3893	n/a	Condition = staticmethod(multiprocessing.Condition)
3894	n/a	Event = staticmethod(multiprocessing.Event)
3895	n/a	Barrier = staticmethod(multiprocessing.Barrier)
3896	n/a	Value = staticmethod(multiprocessing.Value)
3897	n/a	Array = staticmethod(multiprocessing.Array)
3898	n/a	RawValue = staticmethod(multiprocessing.RawValue)
3899	n/a	RawArray = staticmethod(multiprocessing.RawArray)
3900	n/a
3901	n/a
3902	n/a	class ManagerMixin(object):
3903	n/a	TYPE = 'manager'
3904	n/a	Process = multiprocessing.Process
3905	n/a	Queue = property(operator.attrgetter('manager.Queue'))
3906	n/a	JoinableQueue = property(operator.attrgetter('manager.JoinableQueue'))
3907	n/a	Lock = property(operator.attrgetter('manager.Lock'))
3908	n/a	RLock = property(operator.attrgetter('manager.RLock'))
3909	n/a	Semaphore = property(operator.attrgetter('manager.Semaphore'))
3910	n/a	BoundedSemaphore = property(operator.attrgetter('manager.BoundedSemaphore'))
3911	n/a	Condition = property(operator.attrgetter('manager.Condition'))
3912	n/a	Event = property(operator.attrgetter('manager.Event'))
3913	n/a	Barrier = property(operator.attrgetter('manager.Barrier'))
3914	n/a	Value = property(operator.attrgetter('manager.Value'))
3915	n/a	Array = property(operator.attrgetter('manager.Array'))
3916	n/a	list = property(operator.attrgetter('manager.list'))
3917	n/a	dict = property(operator.attrgetter('manager.dict'))
3918	n/a	Namespace = property(operator.attrgetter('manager.Namespace'))
3919	n/a
3920	n/a	@classmethod
3921	n/a	def Pool(cls, args, *kwds):
3922	n/a	return cls.manager.Pool(args, *kwds)
3923	n/a
3924	n/a	@classmethod
3925	n/a	def setUpClass(cls):
3926	n/a	cls.manager = multiprocessing.Manager()
3927	n/a
3928	n/a	@classmethod
3929	n/a	def tearDownClass(cls):
3930	n/a	# only the manager process should be returned by active_children()
3931	n/a	# but this can take a bit on slow machines, so wait a few seconds
3932	n/a	# if there are other children too (see #17395)
3933	n/a	t = 0.01
3934	n/a	while len(multiprocessing.active_children()) > 1 and t < 5:
3935	n/a	time.sleep(t)
3936	n/a	t *= 2
3937	n/a	gc.collect() # do garbage collection
3938	n/a	if cls.manager._number_of_objects() != 0:
3939	n/a	# This is not really an error since some tests do not
3940	n/a	# ensure that all processes which hold a reference to a
3941	n/a	# managed object have been joined.
3942	n/a	print('Shared objects which still exist at manager shutdown:')
3943	n/a	print(cls.manager._debug_info())
3944	n/a	cls.manager.shutdown()
3945	n/a	cls.manager.join()
3946	n/a	cls.manager = None
3947	n/a
3948	n/a
3949	n/a	class ThreadsMixin(object):
3950	n/a	TYPE = 'threads'
3951	n/a	Process = multiprocessing.dummy.Process
3952	n/a	connection = multiprocessing.dummy.connection
3953	n/a	current_process = staticmethod(multiprocessing.dummy.current_process)
3954	n/a	active_children = staticmethod(multiprocessing.dummy.active_children)
3955	n/a	Pool = staticmethod(multiprocessing.dummy.Pool)
3956	n/a	Pipe = staticmethod(multiprocessing.dummy.Pipe)
3957	n/a	Queue = staticmethod(multiprocessing.dummy.Queue)
3958	n/a	JoinableQueue = staticmethod(multiprocessing.dummy.JoinableQueue)
3959	n/a	Lock = staticmethod(multiprocessing.dummy.Lock)
3960	n/a	RLock = staticmethod(multiprocessing.dummy.RLock)
3961	n/a	Semaphore = staticmethod(multiprocessing.dummy.Semaphore)
3962	n/a	BoundedSemaphore = staticmethod(multiprocessing.dummy.BoundedSemaphore)
3963	n/a	Condition = staticmethod(multiprocessing.dummy.Condition)
3964	n/a	Event = staticmethod(multiprocessing.dummy.Event)
3965	n/a	Barrier = staticmethod(multiprocessing.dummy.Barrier)
3966	n/a	Value = staticmethod(multiprocessing.dummy.Value)
3967	n/a	Array = staticmethod(multiprocessing.dummy.Array)
3968	n/a
3969	n/a	#
3970	n/a	# Functions used to create test cases from the base ones in this module
3971	n/a	#
3972	n/a
3973	n/a	def install_tests_in_module_dict(remote_globs, start_method):
3974	n/a	__module__ = remote_globs['__name__']
3975	n/a	local_globs = globals()
3976	n/a	ALL_TYPES = {'processes', 'threads', 'manager'}
3977	n/a
3978	n/a	for name, base in local_globs.items():
3979	n/a	if not isinstance(base, type):
3980	n/a	continue
3981	n/a	if issubclass(base, BaseTestCase):
3982	n/a	if base is BaseTestCase:
3983	n/a	continue
3984	n/a	assert set(base.ALLOWED_TYPES) <= ALL_TYPES, base.ALLOWED_TYPES
3985	n/a	for type_ in base.ALLOWED_TYPES:
3986	n/a	newname = 'With' + type_.capitalize() + name[1:]
3987	n/a	Mixin = local_globs[type_.capitalize() + 'Mixin']
3988	n/a	class Temp(base, Mixin, unittest.TestCase):
3989	n/a	pass
3990	n/a	Temp.__name__ = Temp.__qualname__ = newname
3991	n/a	Temp.__module__ = __module__
3992	n/a	remote_globs[newname] = Temp
3993	n/a	elif issubclass(base, unittest.TestCase):
3994	n/a	class Temp(base, object):
3995	n/a	pass
3996	n/a	Temp.__name__ = Temp.__qualname__ = name
3997	n/a	Temp.__module__ = __module__
3998	n/a	remote_globs[name] = Temp
3999	n/a
4000	n/a	dangling = [None, None]
4001	n/a	old_start_method = [None]
4002	n/a
4003	n/a	def setUpModule():
4004	n/a	multiprocessing.set_forkserver_preload(PRELOAD)
4005	n/a	multiprocessing.process._cleanup()
4006	n/a	dangling[0] = multiprocessing.process._dangling.copy()
4007	n/a	dangling[1] = threading._dangling.copy()
4008	n/a	old_start_method[0] = multiprocessing.get_start_method(allow_none=True)
4009	n/a	try:
4010	n/a	multiprocessing.set_start_method(start_method, force=True)
4011	n/a	except ValueError:
4012	n/a	raise unittest.SkipTest(start_method +
4013	n/a	' start method not supported')
4014	n/a
4015	n/a	if sys.platform.startswith("linux"):
4016	n/a	try:
4017	n/a	lock = multiprocessing.RLock()
4018	n/a	except OSError:
4019	n/a	raise unittest.SkipTest("OSError raises on RLock creation, "
4020	n/a	"see issue 3111!")
4021	n/a	check_enough_semaphores()
4022	n/a	util.get_temp_dir() # creates temp directory
4023	n/a	multiprocessing.get_logger().setLevel(LOG_LEVEL)
4024	n/a
4025	n/a	def tearDownModule():
4026	n/a	multiprocessing.set_start_method(old_start_method[0], force=True)
4027	n/a	# pause a bit so we don't get warning about dangling threads/processes
4028	n/a	time.sleep(0.5)
4029	n/a	multiprocessing.process._cleanup()
4030	n/a	gc.collect()
4031	n/a	tmp = set(multiprocessing.process._dangling) - set(dangling[0])
4032	n/a	if tmp:
4033	n/a	print('Dangling processes:', tmp, file=sys.stderr)
4034	n/a	del tmp
4035	n/a	tmp = set(threading._dangling) - set(dangling[1])
4036	n/a	if tmp:
4037	n/a	print('Dangling threads:', tmp, file=sys.stderr)
4038	n/a
4039	n/a	remote_globs['setUpModule'] = setUpModule
4040	n/a	remote_globs['tearDownModule'] = tearDownModule