blob: e4031f6927756496c9233adb2fa91aead9976862 [file] [log] [blame]
Benjamin Peterson90f5ba52010-03-11 22:53:45 +00001#!/usr/bin/env python3
Georg Brandl86b2fb92008-07-16 03:43:04 +00002
Benjamin Petersone711caf2008-06-11 16:44:04 +00003#
4# Unit tests for the multiprocessing package
5#
6
7import unittest
Benjamin Petersone711caf2008-06-11 16:44:04 +00008import queue as pyqueue
9import time
Alexandre Vassalottic57a84f2009-07-17 12:07:01 +000010import io
Antoine Pitroude911b22011-12-21 11:03:24 +010011import itertools
Benjamin Petersone711caf2008-06-11 16:44:04 +000012import sys
13import os
14import gc
Antoine Pitrouc824e9a2011-04-05 18:11:33 +020015import errno
Benjamin Petersone711caf2008-06-11 16:44:04 +000016import signal
17import array
Benjamin Petersone711caf2008-06-11 16:44:04 +000018import socket
19import random
20import logging
Richard Oudkerk3730a172012-06-15 18:26:07 +010021import struct
R. David Murraya21e4ca2009-03-31 23:16:50 +000022import test.support
Benjamin Petersone711caf2008-06-11 16:44:04 +000023
Benjamin Petersone5384b02008-10-04 22:00:42 +000024
R. David Murraya21e4ca2009-03-31 23:16:50 +000025# Skip tests if _multiprocessing wasn't built.
26_multiprocessing = test.support.import_module('_multiprocessing')
27# Skip tests if sem_open implementation is broken.
28test.support.import_module('multiprocessing.synchronize')
Victor Stinner45df8202010-04-28 22:31:17 +000029# import threading after _multiprocessing to raise a more revelant error
30# message: "No module named _multiprocessing". _multiprocessing is not compiled
31# without thread support.
32import threading
Benjamin Petersone5384b02008-10-04 22:00:42 +000033
Benjamin Petersone711caf2008-06-11 16:44:04 +000034import multiprocessing.dummy
35import multiprocessing.connection
36import multiprocessing.managers
37import multiprocessing.heap
Benjamin Petersone711caf2008-06-11 16:44:04 +000038import multiprocessing.pool
Benjamin Petersone711caf2008-06-11 16:44:04 +000039
Charles-François Natalibc8f0822011-09-20 20:36:51 +020040from multiprocessing import util
41
42try:
43 from multiprocessing import reduction
44 HAS_REDUCTION = True
45except ImportError:
46 HAS_REDUCTION = False
Benjamin Petersone711caf2008-06-11 16:44:04 +000047
Brian Curtinafa88b52010-10-07 01:12:19 +000048try:
49 from multiprocessing.sharedctypes import Value, copy
50 HAS_SHAREDCTYPES = True
51except ImportError:
52 HAS_SHAREDCTYPES = False
53
Antoine Pitroubcb39d42011-08-23 19:46:22 +020054try:
55 import msvcrt
56except ImportError:
57 msvcrt = None
58
Benjamin Petersone711caf2008-06-11 16:44:04 +000059#
60#
61#
62
Benjamin Peterson2bc91df2008-07-13 18:45:30 +000063def latin(s):
64 return s.encode('latin')
Benjamin Petersone711caf2008-06-11 16:44:04 +000065
Benjamin Petersone711caf2008-06-11 16:44:04 +000066#
67# Constants
68#
69
70LOG_LEVEL = util.SUBWARNING
Jesse Noller1f0b6582010-01-27 03:36:01 +000071#LOG_LEVEL = logging.DEBUG
Benjamin Petersone711caf2008-06-11 16:44:04 +000072
73DELTA = 0.1
74CHECK_TIMINGS = False # making true makes tests take a lot longer
75 # and can sometimes cause some non-serious
76 # failures because some calls block a bit
77 # longer than expected
78if CHECK_TIMINGS:
79 TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.82, 0.35, 1.4
80else:
81 TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.1, 0.1, 0.1
82
83HAVE_GETVALUE = not getattr(_multiprocessing,
84 'HAVE_BROKEN_SEM_GETVALUE', False)
85
Jesse Noller6214edd2009-01-19 16:23:53 +000086WIN32 = (sys.platform == "win32")
Antoine Pitrou176f07d2011-06-06 19:35:31 +020087
Richard Oudkerk59d54042012-05-10 16:11:12 +010088from multiprocessing.connection import wait
Antoine Pitrou176f07d2011-06-06 19:35:31 +020089
Richard Oudkerk59d54042012-05-10 16:11:12 +010090def wait_for_handle(handle, timeout):
91 if timeout is not None and timeout < 0.0:
92 timeout = None
93 return wait([handle], timeout)
Jesse Noller6214edd2009-01-19 16:23:53 +000094
Antoine Pitroubcb39d42011-08-23 19:46:22 +020095try:
96 MAXFD = os.sysconf("SC_OPEN_MAX")
97except:
98 MAXFD = 256
99
Benjamin Petersone711caf2008-06-11 16:44:04 +0000100#
Florent Xiclunafd1b0932010-03-28 00:25:02 +0000101# Some tests require ctypes
102#
103
104try:
Florent Xiclunaaa171062010-08-14 15:56:42 +0000105 from ctypes import Structure, c_int, c_double
Florent Xiclunafd1b0932010-03-28 00:25:02 +0000106except ImportError:
107 Structure = object
108 c_int = c_double = None
109
Charles-François Natali221ef672011-11-22 18:55:22 +0100110
111def check_enough_semaphores():
112 """Check that the system supports enough semaphores to run the test."""
113 # minimum number of semaphores available according to POSIX
114 nsems_min = 256
115 try:
116 nsems = os.sysconf("SC_SEM_NSEMS_MAX")
117 except (AttributeError, ValueError):
118 # sysconf not available or setting not available
119 return
120 if nsems == -1 or nsems >= nsems_min:
121 return
122 raise unittest.SkipTest("The OS doesn't support enough semaphores "
123 "to run the test (required: %d)." % nsems_min)
124
125
Florent Xiclunafd1b0932010-03-28 00:25:02 +0000126#
Benjamin Petersone711caf2008-06-11 16:44:04 +0000127# Creates a wrapper for a function which records the time it takes to finish
128#
129
130class TimingWrapper(object):
131
132 def __init__(self, func):
133 self.func = func
134 self.elapsed = None
135
136 def __call__(self, *args, **kwds):
137 t = time.time()
138 try:
139 return self.func(*args, **kwds)
140 finally:
141 self.elapsed = time.time() - t
142
143#
144# Base class for test cases
145#
146
147class BaseTestCase(object):
148
149 ALLOWED_TYPES = ('processes', 'manager', 'threads')
150
151 def assertTimingAlmostEqual(self, a, b):
152 if CHECK_TIMINGS:
153 self.assertAlmostEqual(a, b, 1)
154
155 def assertReturnsIfImplemented(self, value, func, *args):
156 try:
157 res = func(*args)
158 except NotImplementedError:
159 pass
160 else:
161 return self.assertEqual(value, res)
162
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +0000163 # For the sanity of Windows users, rather than crashing or freezing in
164 # multiple ways.
165 def __reduce__(self, *args):
166 raise NotImplementedError("shouldn't try to pickle a test case")
167
168 __reduce_ex__ = __reduce__
169
Benjamin Petersone711caf2008-06-11 16:44:04 +0000170#
171# Return the value of a semaphore
172#
173
174def get_value(self):
175 try:
176 return self.get_value()
177 except AttributeError:
178 try:
179 return self._Semaphore__value
180 except AttributeError:
181 try:
182 return self._value
183 except AttributeError:
184 raise NotImplementedError
185
186#
187# Testcases
188#
189
190class _TestProcess(BaseTestCase):
191
192 ALLOWED_TYPES = ('processes', 'threads')
193
194 def test_current(self):
195 if self.TYPE == 'threads':
196 return
197
198 current = self.current_process()
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000199 authkey = current.authkey
Benjamin Petersone711caf2008-06-11 16:44:04 +0000200
201 self.assertTrue(current.is_alive())
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000202 self.assertTrue(not current.daemon)
Ezio Melottie9615932010-01-24 19:26:24 +0000203 self.assertIsInstance(authkey, bytes)
Benjamin Petersone711caf2008-06-11 16:44:04 +0000204 self.assertTrue(len(authkey) > 0)
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000205 self.assertEqual(current.ident, os.getpid())
206 self.assertEqual(current.exitcode, None)
Benjamin Petersone711caf2008-06-11 16:44:04 +0000207
Antoine Pitrou0bd4deb2011-02-25 22:07:43 +0000208 def test_daemon_argument(self):
209 if self.TYPE == "threads":
210 return
211
212 # By default uses the current process's daemon flag.
213 proc0 = self.Process(target=self._test)
Antoine Pitrouec785222011-03-02 00:15:44 +0000214 self.assertEqual(proc0.daemon, self.current_process().daemon)
Antoine Pitrou0bd4deb2011-02-25 22:07:43 +0000215 proc1 = self.Process(target=self._test, daemon=True)
216 self.assertTrue(proc1.daemon)
217 proc2 = self.Process(target=self._test, daemon=False)
218 self.assertFalse(proc2.daemon)
219
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +0000220 @classmethod
221 def _test(cls, q, *args, **kwds):
222 current = cls.current_process()
Benjamin Petersone711caf2008-06-11 16:44:04 +0000223 q.put(args)
224 q.put(kwds)
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000225 q.put(current.name)
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +0000226 if cls.TYPE != 'threads':
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000227 q.put(bytes(current.authkey))
Benjamin Petersone711caf2008-06-11 16:44:04 +0000228 q.put(current.pid)
229
230 def test_process(self):
231 q = self.Queue(1)
232 e = self.Event()
233 args = (q, 1, 2)
234 kwargs = {'hello':23, 'bye':2.54}
235 name = 'SomeProcess'
236 p = self.Process(
237 target=self._test, args=args, kwargs=kwargs, name=name
238 )
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000239 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +0000240 current = self.current_process()
241
242 if self.TYPE != 'threads':
Ezio Melottib3aedd42010-11-20 19:04:17 +0000243 self.assertEqual(p.authkey, current.authkey)
244 self.assertEqual(p.is_alive(), False)
245 self.assertEqual(p.daemon, True)
Benjamin Peterson577473f2010-01-19 00:09:57 +0000246 self.assertNotIn(p, self.active_children())
Benjamin Petersone711caf2008-06-11 16:44:04 +0000247 self.assertTrue(type(self.active_children()) is list)
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000248 self.assertEqual(p.exitcode, None)
Benjamin Petersone711caf2008-06-11 16:44:04 +0000249
250 p.start()
251
Ezio Melottib3aedd42010-11-20 19:04:17 +0000252 self.assertEqual(p.exitcode, None)
253 self.assertEqual(p.is_alive(), True)
Benjamin Peterson577473f2010-01-19 00:09:57 +0000254 self.assertIn(p, self.active_children())
Benjamin Petersone711caf2008-06-11 16:44:04 +0000255
Ezio Melottib3aedd42010-11-20 19:04:17 +0000256 self.assertEqual(q.get(), args[1:])
257 self.assertEqual(q.get(), kwargs)
258 self.assertEqual(q.get(), p.name)
Benjamin Petersone711caf2008-06-11 16:44:04 +0000259 if self.TYPE != 'threads':
Ezio Melottib3aedd42010-11-20 19:04:17 +0000260 self.assertEqual(q.get(), current.authkey)
261 self.assertEqual(q.get(), p.pid)
Benjamin Petersone711caf2008-06-11 16:44:04 +0000262
263 p.join()
264
Ezio Melottib3aedd42010-11-20 19:04:17 +0000265 self.assertEqual(p.exitcode, 0)
266 self.assertEqual(p.is_alive(), False)
Benjamin Peterson577473f2010-01-19 00:09:57 +0000267 self.assertNotIn(p, self.active_children())
Benjamin Petersone711caf2008-06-11 16:44:04 +0000268
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +0000269 @classmethod
270 def _test_terminate(cls):
Benjamin Petersone711caf2008-06-11 16:44:04 +0000271 time.sleep(1000)
272
273 def test_terminate(self):
274 if self.TYPE == 'threads':
275 return
276
277 p = self.Process(target=self._test_terminate)
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000278 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +0000279 p.start()
280
281 self.assertEqual(p.is_alive(), True)
Benjamin Peterson577473f2010-01-19 00:09:57 +0000282 self.assertIn(p, self.active_children())
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000283 self.assertEqual(p.exitcode, None)
Benjamin Petersone711caf2008-06-11 16:44:04 +0000284
Richard Oudkerk59d54042012-05-10 16:11:12 +0100285 join = TimingWrapper(p.join)
286
287 self.assertEqual(join(0), None)
288 self.assertTimingAlmostEqual(join.elapsed, 0.0)
289 self.assertEqual(p.is_alive(), True)
290
291 self.assertEqual(join(-1), None)
292 self.assertTimingAlmostEqual(join.elapsed, 0.0)
293 self.assertEqual(p.is_alive(), True)
294
Benjamin Petersone711caf2008-06-11 16:44:04 +0000295 p.terminate()
296
Benjamin Petersone711caf2008-06-11 16:44:04 +0000297 self.assertEqual(join(), None)
298 self.assertTimingAlmostEqual(join.elapsed, 0.0)
299
300 self.assertEqual(p.is_alive(), False)
Benjamin Peterson577473f2010-01-19 00:09:57 +0000301 self.assertNotIn(p, self.active_children())
Benjamin Petersone711caf2008-06-11 16:44:04 +0000302
303 p.join()
304
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000305 # XXX sometimes get p.exitcode == 0 on Windows ...
306 #self.assertEqual(p.exitcode, -signal.SIGTERM)
Benjamin Petersone711caf2008-06-11 16:44:04 +0000307
308 def test_cpu_count(self):
309 try:
310 cpus = multiprocessing.cpu_count()
311 except NotImplementedError:
312 cpus = 1
313 self.assertTrue(type(cpus) is int)
314 self.assertTrue(cpus >= 1)
315
316 def test_active_children(self):
317 self.assertEqual(type(self.active_children()), list)
318
319 p = self.Process(target=time.sleep, args=(DELTA,))
Benjamin Peterson577473f2010-01-19 00:09:57 +0000320 self.assertNotIn(p, self.active_children())
Benjamin Petersone711caf2008-06-11 16:44:04 +0000321
Jesus Cea94f964f2011-09-09 20:26:57 +0200322 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +0000323 p.start()
Benjamin Peterson577473f2010-01-19 00:09:57 +0000324 self.assertIn(p, self.active_children())
Benjamin Petersone711caf2008-06-11 16:44:04 +0000325
326 p.join()
Benjamin Peterson577473f2010-01-19 00:09:57 +0000327 self.assertNotIn(p, self.active_children())
Benjamin Petersone711caf2008-06-11 16:44:04 +0000328
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +0000329 @classmethod
330 def _test_recursion(cls, wconn, id):
Benjamin Petersone711caf2008-06-11 16:44:04 +0000331 from multiprocessing import forking
332 wconn.send(id)
333 if len(id) < 2:
334 for i in range(2):
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +0000335 p = cls.Process(
336 target=cls._test_recursion, args=(wconn, id+[i])
Benjamin Petersone711caf2008-06-11 16:44:04 +0000337 )
338 p.start()
339 p.join()
340
341 def test_recursion(self):
342 rconn, wconn = self.Pipe(duplex=False)
343 self._test_recursion(wconn, [])
344
345 time.sleep(DELTA)
346 result = []
347 while rconn.poll():
348 result.append(rconn.recv())
349
350 expected = [
351 [],
352 [0],
353 [0, 0],
354 [0, 1],
355 [1],
356 [1, 0],
357 [1, 1]
358 ]
359 self.assertEqual(result, expected)
360
Antoine Pitrou176f07d2011-06-06 19:35:31 +0200361 @classmethod
362 def _test_sentinel(cls, event):
363 event.wait(10.0)
364
365 def test_sentinel(self):
366 if self.TYPE == "threads":
367 return
368 event = self.Event()
369 p = self.Process(target=self._test_sentinel, args=(event,))
370 with self.assertRaises(ValueError):
371 p.sentinel
372 p.start()
373 self.addCleanup(p.join)
374 sentinel = p.sentinel
375 self.assertIsInstance(sentinel, int)
376 self.assertFalse(wait_for_handle(sentinel, timeout=0.0))
377 event.set()
378 p.join()
379 self.assertTrue(wait_for_handle(sentinel, timeout=DELTA))
380
Benjamin Petersone711caf2008-06-11 16:44:04 +0000381#
382#
383#
384
385class _UpperCaser(multiprocessing.Process):
386
387 def __init__(self):
388 multiprocessing.Process.__init__(self)
389 self.child_conn, self.parent_conn = multiprocessing.Pipe()
390
391 def run(self):
392 self.parent_conn.close()
393 for s in iter(self.child_conn.recv, None):
394 self.child_conn.send(s.upper())
395 self.child_conn.close()
396
397 def submit(self, s):
398 assert type(s) is str
399 self.parent_conn.send(s)
400 return self.parent_conn.recv()
401
402 def stop(self):
403 self.parent_conn.send(None)
404 self.parent_conn.close()
405 self.child_conn.close()
406
407class _TestSubclassingProcess(BaseTestCase):
408
409 ALLOWED_TYPES = ('processes',)
410
411 def test_subclassing(self):
412 uppercaser = _UpperCaser()
Jesus Cea94f964f2011-09-09 20:26:57 +0200413 uppercaser.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +0000414 uppercaser.start()
415 self.assertEqual(uppercaser.submit('hello'), 'HELLO')
416 self.assertEqual(uppercaser.submit('world'), 'WORLD')
417 uppercaser.stop()
418 uppercaser.join()
419
Antoine Pitrou84a0fbf2012-01-27 10:52:37 +0100420 def test_stderr_flush(self):
421 # sys.stderr is flushed at process shutdown (issue #13812)
422 if self.TYPE == "threads":
423 return
424
425 testfn = test.support.TESTFN
426 self.addCleanup(test.support.unlink, testfn)
427 proc = self.Process(target=self._test_stderr_flush, args=(testfn,))
428 proc.start()
429 proc.join()
430 with open(testfn, 'r') as f:
431 err = f.read()
432 # The whole traceback was printed
433 self.assertIn("ZeroDivisionError", err)
434 self.assertIn("test_multiprocessing.py", err)
435 self.assertIn("1/0 # MARKER", err)
436
437 @classmethod
438 def _test_stderr_flush(cls, testfn):
439 sys.stderr = open(testfn, 'w')
440 1/0 # MARKER
441
442
Richard Oudkerk29471de2012-06-06 19:04:57 +0100443 @classmethod
444 def _test_sys_exit(cls, reason, testfn):
445 sys.stderr = open(testfn, 'w')
446 sys.exit(reason)
447
448 def test_sys_exit(self):
449 # See Issue 13854
450 if self.TYPE == 'threads':
451 return
452
453 testfn = test.support.TESTFN
454 self.addCleanup(test.support.unlink, testfn)
455
456 for reason, code in (([1, 2, 3], 1), ('ignore this', 0)):
457 p = self.Process(target=self._test_sys_exit, args=(reason, testfn))
458 p.daemon = True
459 p.start()
460 p.join(5)
461 self.assertEqual(p.exitcode, code)
462
463 with open(testfn, 'r') as f:
464 self.assertEqual(f.read().rstrip(), str(reason))
465
466 for reason in (True, False, 8):
467 p = self.Process(target=sys.exit, args=(reason,))
468 p.daemon = True
469 p.start()
470 p.join(5)
471 self.assertEqual(p.exitcode, reason)
472
Benjamin Petersone711caf2008-06-11 16:44:04 +0000473#
474#
475#
476
477def queue_empty(q):
478 if hasattr(q, 'empty'):
479 return q.empty()
480 else:
481 return q.qsize() == 0
482
483def queue_full(q, maxsize):
484 if hasattr(q, 'full'):
485 return q.full()
486 else:
487 return q.qsize() == maxsize
488
489
490class _TestQueue(BaseTestCase):
491
492
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +0000493 @classmethod
494 def _test_put(cls, queue, child_can_start, parent_can_continue):
Benjamin Petersone711caf2008-06-11 16:44:04 +0000495 child_can_start.wait()
496 for i in range(6):
497 queue.get()
498 parent_can_continue.set()
499
500 def test_put(self):
501 MAXSIZE = 6
502 queue = self.Queue(maxsize=MAXSIZE)
503 child_can_start = self.Event()
504 parent_can_continue = self.Event()
505
506 proc = self.Process(
507 target=self._test_put,
508 args=(queue, child_can_start, parent_can_continue)
509 )
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000510 proc.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +0000511 proc.start()
512
513 self.assertEqual(queue_empty(queue), True)
514 self.assertEqual(queue_full(queue, MAXSIZE), False)
515
516 queue.put(1)
517 queue.put(2, True)
518 queue.put(3, True, None)
519 queue.put(4, False)
520 queue.put(5, False, None)
521 queue.put_nowait(6)
522
523 # the values may be in buffer but not yet in pipe so sleep a bit
524 time.sleep(DELTA)
525
526 self.assertEqual(queue_empty(queue), False)
527 self.assertEqual(queue_full(queue, MAXSIZE), True)
528
529 put = TimingWrapper(queue.put)
530 put_nowait = TimingWrapper(queue.put_nowait)
531
532 self.assertRaises(pyqueue.Full, put, 7, False)
533 self.assertTimingAlmostEqual(put.elapsed, 0)
534
535 self.assertRaises(pyqueue.Full, put, 7, False, None)
536 self.assertTimingAlmostEqual(put.elapsed, 0)
537
538 self.assertRaises(pyqueue.Full, put_nowait, 7)
539 self.assertTimingAlmostEqual(put_nowait.elapsed, 0)
540
541 self.assertRaises(pyqueue.Full, put, 7, True, TIMEOUT1)
542 self.assertTimingAlmostEqual(put.elapsed, TIMEOUT1)
543
544 self.assertRaises(pyqueue.Full, put, 7, False, TIMEOUT2)
545 self.assertTimingAlmostEqual(put.elapsed, 0)
546
547 self.assertRaises(pyqueue.Full, put, 7, True, timeout=TIMEOUT3)
548 self.assertTimingAlmostEqual(put.elapsed, TIMEOUT3)
549
550 child_can_start.set()
551 parent_can_continue.wait()
552
553 self.assertEqual(queue_empty(queue), True)
554 self.assertEqual(queue_full(queue, MAXSIZE), False)
555
556 proc.join()
557
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +0000558 @classmethod
559 def _test_get(cls, queue, child_can_start, parent_can_continue):
Benjamin Petersone711caf2008-06-11 16:44:04 +0000560 child_can_start.wait()
Benjamin Petersonb29cbbc2008-06-16 20:57:14 +0000561 #queue.put(1)
Benjamin Petersone711caf2008-06-11 16:44:04 +0000562 queue.put(2)
563 queue.put(3)
564 queue.put(4)
565 queue.put(5)
566 parent_can_continue.set()
567
568 def test_get(self):
569 queue = self.Queue()
570 child_can_start = self.Event()
571 parent_can_continue = self.Event()
572
573 proc = self.Process(
574 target=self._test_get,
575 args=(queue, child_can_start, parent_can_continue)
576 )
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000577 proc.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +0000578 proc.start()
579
580 self.assertEqual(queue_empty(queue), True)
581
582 child_can_start.set()
583 parent_can_continue.wait()
584
585 time.sleep(DELTA)
586 self.assertEqual(queue_empty(queue), False)
587
Benjamin Petersonb29cbbc2008-06-16 20:57:14 +0000588 # Hangs unexpectedly, remove for now
589 #self.assertEqual(queue.get(), 1)
Benjamin Petersone711caf2008-06-11 16:44:04 +0000590 self.assertEqual(queue.get(True, None), 2)
591 self.assertEqual(queue.get(True), 3)
592 self.assertEqual(queue.get(timeout=1), 4)
593 self.assertEqual(queue.get_nowait(), 5)
594
595 self.assertEqual(queue_empty(queue), True)
596
597 get = TimingWrapper(queue.get)
598 get_nowait = TimingWrapper(queue.get_nowait)
599
600 self.assertRaises(pyqueue.Empty, get, False)
601 self.assertTimingAlmostEqual(get.elapsed, 0)
602
603 self.assertRaises(pyqueue.Empty, get, False, None)
604 self.assertTimingAlmostEqual(get.elapsed, 0)
605
606 self.assertRaises(pyqueue.Empty, get_nowait)
607 self.assertTimingAlmostEqual(get_nowait.elapsed, 0)
608
609 self.assertRaises(pyqueue.Empty, get, True, TIMEOUT1)
610 self.assertTimingAlmostEqual(get.elapsed, TIMEOUT1)
611
612 self.assertRaises(pyqueue.Empty, get, False, TIMEOUT2)
613 self.assertTimingAlmostEqual(get.elapsed, 0)
614
615 self.assertRaises(pyqueue.Empty, get, timeout=TIMEOUT3)
616 self.assertTimingAlmostEqual(get.elapsed, TIMEOUT3)
617
618 proc.join()
619
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +0000620 @classmethod
621 def _test_fork(cls, queue):
Benjamin Petersone711caf2008-06-11 16:44:04 +0000622 for i in range(10, 20):
623 queue.put(i)
624 # note that at this point the items may only be buffered, so the
625 # process cannot shutdown until the feeder thread has finished
626 # pushing items onto the pipe.
627
628 def test_fork(self):
629 # Old versions of Queue would fail to create a new feeder
630 # thread for a forked process if the original process had its
631 # own feeder thread. This test checks that this no longer
632 # happens.
633
634 queue = self.Queue()
635
636 # put items on queue so that main process starts a feeder thread
637 for i in range(10):
638 queue.put(i)
639
640 # wait to make sure thread starts before we fork a new process
641 time.sleep(DELTA)
642
643 # fork process
644 p = self.Process(target=self._test_fork, args=(queue,))
Jesus Cea94f964f2011-09-09 20:26:57 +0200645 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +0000646 p.start()
647
648 # check that all expected items are in the queue
649 for i in range(20):
650 self.assertEqual(queue.get(), i)
651 self.assertRaises(pyqueue.Empty, queue.get, False)
652
653 p.join()
654
655 def test_qsize(self):
656 q = self.Queue()
657 try:
658 self.assertEqual(q.qsize(), 0)
659 except NotImplementedError:
660 return
661 q.put(1)
662 self.assertEqual(q.qsize(), 1)
663 q.put(5)
664 self.assertEqual(q.qsize(), 2)
665 q.get()
666 self.assertEqual(q.qsize(), 1)
667 q.get()
668 self.assertEqual(q.qsize(), 0)
669
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +0000670 @classmethod
671 def _test_task_done(cls, q):
Benjamin Petersone711caf2008-06-11 16:44:04 +0000672 for obj in iter(q.get, None):
673 time.sleep(DELTA)
674 q.task_done()
675
676 def test_task_done(self):
677 queue = self.JoinableQueue()
678
679 if sys.version_info < (2, 5) and not hasattr(queue, 'task_done'):
Florent Xiclunafd1b0932010-03-28 00:25:02 +0000680 self.skipTest("requires 'queue.task_done()' method")
Benjamin Petersone711caf2008-06-11 16:44:04 +0000681
682 workers = [self.Process(target=self._test_task_done, args=(queue,))
683 for i in range(4)]
684
685 for p in workers:
Jesus Cea94f964f2011-09-09 20:26:57 +0200686 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +0000687 p.start()
688
689 for i in range(10):
690 queue.put(i)
691
692 queue.join()
693
694 for p in workers:
695 queue.put(None)
696
697 for p in workers:
698 p.join()
699
700#
701#
702#
703
704class _TestLock(BaseTestCase):
705
706 def test_lock(self):
707 lock = self.Lock()
708 self.assertEqual(lock.acquire(), True)
709 self.assertEqual(lock.acquire(False), False)
710 self.assertEqual(lock.release(), None)
711 self.assertRaises((ValueError, threading.ThreadError), lock.release)
712
713 def test_rlock(self):
714 lock = self.RLock()
715 self.assertEqual(lock.acquire(), True)
716 self.assertEqual(lock.acquire(), True)
717 self.assertEqual(lock.acquire(), True)
718 self.assertEqual(lock.release(), None)
719 self.assertEqual(lock.release(), None)
720 self.assertEqual(lock.release(), None)
721 self.assertRaises((AssertionError, RuntimeError), lock.release)
722
Jesse Nollerf8d00852009-03-31 03:25:07 +0000723 def test_lock_context(self):
724 with self.Lock():
725 pass
726
Benjamin Petersone711caf2008-06-11 16:44:04 +0000727
728class _TestSemaphore(BaseTestCase):
729
730 def _test_semaphore(self, sem):
731 self.assertReturnsIfImplemented(2, get_value, sem)
732 self.assertEqual(sem.acquire(), True)
733 self.assertReturnsIfImplemented(1, get_value, sem)
734 self.assertEqual(sem.acquire(), True)
735 self.assertReturnsIfImplemented(0, get_value, sem)
736 self.assertEqual(sem.acquire(False), False)
737 self.assertReturnsIfImplemented(0, get_value, sem)
738 self.assertEqual(sem.release(), None)
739 self.assertReturnsIfImplemented(1, get_value, sem)
740 self.assertEqual(sem.release(), None)
741 self.assertReturnsIfImplemented(2, get_value, sem)
742
743 def test_semaphore(self):
744 sem = self.Semaphore(2)
745 self._test_semaphore(sem)
746 self.assertEqual(sem.release(), None)
747 self.assertReturnsIfImplemented(3, get_value, sem)
748 self.assertEqual(sem.release(), None)
749 self.assertReturnsIfImplemented(4, get_value, sem)
750
751 def test_bounded_semaphore(self):
752 sem = self.BoundedSemaphore(2)
753 self._test_semaphore(sem)
754 # Currently fails on OS/X
755 #if HAVE_GETVALUE:
756 # self.assertRaises(ValueError, sem.release)
757 # self.assertReturnsIfImplemented(2, get_value, sem)
758
759 def test_timeout(self):
760 if self.TYPE != 'processes':
761 return
762
763 sem = self.Semaphore(0)
764 acquire = TimingWrapper(sem.acquire)
765
766 self.assertEqual(acquire(False), False)
767 self.assertTimingAlmostEqual(acquire.elapsed, 0.0)
768
769 self.assertEqual(acquire(False, None), False)
770 self.assertTimingAlmostEqual(acquire.elapsed, 0.0)
771
772 self.assertEqual(acquire(False, TIMEOUT1), False)
773 self.assertTimingAlmostEqual(acquire.elapsed, 0)
774
775 self.assertEqual(acquire(True, TIMEOUT2), False)
776 self.assertTimingAlmostEqual(acquire.elapsed, TIMEOUT2)
777
778 self.assertEqual(acquire(timeout=TIMEOUT3), False)
779 self.assertTimingAlmostEqual(acquire.elapsed, TIMEOUT3)
780
781
782class _TestCondition(BaseTestCase):
783
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +0000784 @classmethod
785 def f(cls, cond, sleeping, woken, timeout=None):
Benjamin Petersone711caf2008-06-11 16:44:04 +0000786 cond.acquire()
787 sleeping.release()
788 cond.wait(timeout)
789 woken.release()
790 cond.release()
791
792 def check_invariant(self, cond):
793 # this is only supposed to succeed when there are no sleepers
794 if self.TYPE == 'processes':
795 try:
796 sleepers = (cond._sleeping_count.get_value() -
797 cond._woken_count.get_value())
798 self.assertEqual(sleepers, 0)
799 self.assertEqual(cond._wait_semaphore.get_value(), 0)
800 except NotImplementedError:
801 pass
802
803 def test_notify(self):
804 cond = self.Condition()
805 sleeping = self.Semaphore(0)
806 woken = self.Semaphore(0)
807
808 p = self.Process(target=self.f, args=(cond, sleeping, woken))
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000809 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +0000810 p.start()
811
812 p = threading.Thread(target=self.f, args=(cond, sleeping, woken))
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000813 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +0000814 p.start()
815
816 # wait for both children to start sleeping
817 sleeping.acquire()
818 sleeping.acquire()
819
820 # check no process/thread has woken up
821 time.sleep(DELTA)
822 self.assertReturnsIfImplemented(0, get_value, woken)
823
824 # wake up one process/thread
825 cond.acquire()
826 cond.notify()
827 cond.release()
828
829 # check one process/thread has woken up
830 time.sleep(DELTA)
831 self.assertReturnsIfImplemented(1, get_value, woken)
832
833 # wake up another
834 cond.acquire()
835 cond.notify()
836 cond.release()
837
838 # check other has woken up
839 time.sleep(DELTA)
840 self.assertReturnsIfImplemented(2, get_value, woken)
841
842 # check state is not mucked up
843 self.check_invariant(cond)
844 p.join()
845
846 def test_notify_all(self):
847 cond = self.Condition()
848 sleeping = self.Semaphore(0)
849 woken = self.Semaphore(0)
850
851 # start some threads/processes which will timeout
852 for i in range(3):
853 p = self.Process(target=self.f,
854 args=(cond, sleeping, woken, TIMEOUT1))
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000855 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +0000856 p.start()
857
858 t = threading.Thread(target=self.f,
859 args=(cond, sleeping, woken, TIMEOUT1))
Benjamin Peterson72753702008-08-18 18:09:21 +0000860 t.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +0000861 t.start()
862
863 # wait for them all to sleep
864 for i in range(6):
865 sleeping.acquire()
866
867 # check they have all timed out
868 for i in range(6):
869 woken.acquire()
870 self.assertReturnsIfImplemented(0, get_value, woken)
871
872 # check state is not mucked up
873 self.check_invariant(cond)
874
875 # start some more threads/processes
876 for i in range(3):
877 p = self.Process(target=self.f, args=(cond, sleeping, woken))
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +0000878 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +0000879 p.start()
880
881 t = threading.Thread(target=self.f, args=(cond, sleeping, woken))
Benjamin Peterson72753702008-08-18 18:09:21 +0000882 t.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +0000883 t.start()
884
885 # wait for them to all sleep
886 for i in range(6):
887 sleeping.acquire()
888
889 # check no process/thread has woken up
890 time.sleep(DELTA)
891 self.assertReturnsIfImplemented(0, get_value, woken)
892
893 # wake them all up
894 cond.acquire()
895 cond.notify_all()
896 cond.release()
897
898 # check they have all woken
Antoine Pitrouf25a8de2011-04-16 21:02:01 +0200899 for i in range(10):
900 try:
901 if get_value(woken) == 6:
902 break
903 except NotImplementedError:
904 break
905 time.sleep(DELTA)
Benjamin Petersone711caf2008-06-11 16:44:04 +0000906 self.assertReturnsIfImplemented(6, get_value, woken)
907
908 # check state is not mucked up
909 self.check_invariant(cond)
910
911 def test_timeout(self):
912 cond = self.Condition()
913 wait = TimingWrapper(cond.wait)
914 cond.acquire()
915 res = wait(TIMEOUT1)
916 cond.release()
Georg Brandl65ffae02010-10-28 09:24:56 +0000917 self.assertEqual(res, False)
Benjamin Petersone711caf2008-06-11 16:44:04 +0000918 self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1)
919
Charles-François Natalic8ce7152012-04-17 18:45:57 +0200920 @classmethod
921 def _test_waitfor_f(cls, cond, state):
922 with cond:
923 state.value = 0
924 cond.notify()
925 result = cond.wait_for(lambda : state.value==4)
926 if not result or state.value != 4:
927 sys.exit(1)
928
929 @unittest.skipUnless(HAS_SHAREDCTYPES, 'needs sharedctypes')
930 def test_waitfor(self):
931 # based on test in test/lock_tests.py
932 cond = self.Condition()
933 state = self.Value('i', -1)
934
935 p = self.Process(target=self._test_waitfor_f, args=(cond, state))
936 p.daemon = True
937 p.start()
938
939 with cond:
940 result = cond.wait_for(lambda : state.value==0)
941 self.assertTrue(result)
942 self.assertEqual(state.value, 0)
943
944 for i in range(4):
945 time.sleep(0.01)
946 with cond:
947 state.value += 1
948 cond.notify()
949
950 p.join(5)
951 self.assertFalse(p.is_alive())
952 self.assertEqual(p.exitcode, 0)
953
954 @classmethod
Richard Oudkerk6dbca362012-05-06 16:46:36 +0100955 def _test_waitfor_timeout_f(cls, cond, state, success, sem):
956 sem.release()
Charles-François Natalic8ce7152012-04-17 18:45:57 +0200957 with cond:
958 expected = 0.1
959 dt = time.time()
960 result = cond.wait_for(lambda : state.value==4, timeout=expected)
961 dt = time.time() - dt
962 # borrow logic in assertTimeout() from test/lock_tests.py
963 if not result and expected * 0.6 < dt < expected * 10.0:
964 success.value = True
965
966 @unittest.skipUnless(HAS_SHAREDCTYPES, 'needs sharedctypes')
967 def test_waitfor_timeout(self):
968 # based on test in test/lock_tests.py
969 cond = self.Condition()
970 state = self.Value('i', 0)
971 success = self.Value('i', False)
Richard Oudkerk6dbca362012-05-06 16:46:36 +0100972 sem = self.Semaphore(0)
Charles-François Natalic8ce7152012-04-17 18:45:57 +0200973
974 p = self.Process(target=self._test_waitfor_timeout_f,
Richard Oudkerk6dbca362012-05-06 16:46:36 +0100975 args=(cond, state, success, sem))
Charles-François Natalic8ce7152012-04-17 18:45:57 +0200976 p.daemon = True
977 p.start()
Richard Oudkerk6dbca362012-05-06 16:46:36 +0100978 self.assertTrue(sem.acquire(timeout=10))
Charles-François Natalic8ce7152012-04-17 18:45:57 +0200979
980 # Only increment 3 times, so state == 4 is never reached.
981 for i in range(3):
982 time.sleep(0.01)
983 with cond:
984 state.value += 1
985 cond.notify()
986
987 p.join(5)
988 self.assertTrue(success.value)
989
Richard Oudkerk98449932012-06-05 13:15:29 +0100990 @classmethod
991 def _test_wait_result(cls, c, pid):
992 with c:
993 c.notify()
994 time.sleep(1)
995 if pid is not None:
996 os.kill(pid, signal.SIGINT)
997
998 def test_wait_result(self):
999 if isinstance(self, ProcessesMixin) and sys.platform != 'win32':
1000 pid = os.getpid()
1001 else:
1002 pid = None
1003
1004 c = self.Condition()
1005 with c:
1006 self.assertFalse(c.wait(0))
1007 self.assertFalse(c.wait(0.1))
1008
1009 p = self.Process(target=self._test_wait_result, args=(c, pid))
1010 p.start()
1011
1012 self.assertTrue(c.wait(10))
1013 if pid is not None:
1014 self.assertRaises(KeyboardInterrupt, c.wait, 10)
1015
1016 p.join()
1017
Benjamin Petersone711caf2008-06-11 16:44:04 +00001018
1019class _TestEvent(BaseTestCase):
1020
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00001021 @classmethod
1022 def _test_event(cls, event):
Benjamin Petersone711caf2008-06-11 16:44:04 +00001023 time.sleep(TIMEOUT2)
1024 event.set()
1025
1026 def test_event(self):
1027 event = self.Event()
1028 wait = TimingWrapper(event.wait)
1029
Ezio Melotti13925002011-03-16 11:05:33 +02001030 # Removed temporarily, due to API shear, this does not
Benjamin Petersone711caf2008-06-11 16:44:04 +00001031 # work with threading._Event objects. is_set == isSet
Benjamin Peterson965ce872009-04-05 21:24:58 +00001032 self.assertEqual(event.is_set(), False)
Benjamin Petersone711caf2008-06-11 16:44:04 +00001033
Benjamin Peterson965ce872009-04-05 21:24:58 +00001034 # Removed, threading.Event.wait() will return the value of the __flag
1035 # instead of None. API Shear with the semaphore backed mp.Event
1036 self.assertEqual(wait(0.0), False)
Benjamin Petersone711caf2008-06-11 16:44:04 +00001037 self.assertTimingAlmostEqual(wait.elapsed, 0.0)
Benjamin Peterson965ce872009-04-05 21:24:58 +00001038 self.assertEqual(wait(TIMEOUT1), False)
Benjamin Petersone711caf2008-06-11 16:44:04 +00001039 self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1)
1040
1041 event.set()
1042
1043 # See note above on the API differences
Benjamin Peterson965ce872009-04-05 21:24:58 +00001044 self.assertEqual(event.is_set(), True)
1045 self.assertEqual(wait(), True)
Benjamin Petersone711caf2008-06-11 16:44:04 +00001046 self.assertTimingAlmostEqual(wait.elapsed, 0.0)
Benjamin Peterson965ce872009-04-05 21:24:58 +00001047 self.assertEqual(wait(TIMEOUT1), True)
Benjamin Petersone711caf2008-06-11 16:44:04 +00001048 self.assertTimingAlmostEqual(wait.elapsed, 0.0)
1049 # self.assertEqual(event.is_set(), True)
1050
1051 event.clear()
1052
1053 #self.assertEqual(event.is_set(), False)
1054
Jesus Cea94f964f2011-09-09 20:26:57 +02001055 p = self.Process(target=self._test_event, args=(event,))
1056 p.daemon = True
1057 p.start()
Benjamin Peterson965ce872009-04-05 21:24:58 +00001058 self.assertEqual(wait(), True)
Benjamin Petersone711caf2008-06-11 16:44:04 +00001059
1060#
Richard Oudkerk3730a172012-06-15 18:26:07 +01001061# Tests for Barrier - adapted from tests in test/lock_tests.py
1062#
1063
1064# Many of the tests for threading.Barrier use a list as an atomic
1065# counter: a value is appended to increment the counter, and the
1066# length of the list gives the value. We use the class DummyList
1067# for the same purpose.
1068
1069class _DummyList(object):
1070
1071 def __init__(self):
1072 wrapper = multiprocessing.heap.BufferWrapper(struct.calcsize('i'))
1073 lock = multiprocessing.Lock()
1074 self.__setstate__((wrapper, lock))
1075 self._lengthbuf[0] = 0
1076
1077 def __setstate__(self, state):
1078 (self._wrapper, self._lock) = state
1079 self._lengthbuf = self._wrapper.create_memoryview().cast('i')
1080
1081 def __getstate__(self):
1082 return (self._wrapper, self._lock)
1083
1084 def append(self, _):
1085 with self._lock:
1086 self._lengthbuf[0] += 1
1087
1088 def __len__(self):
1089 with self._lock:
1090 return self._lengthbuf[0]
1091
1092def _wait():
1093 # A crude wait/yield function not relying on synchronization primitives.
1094 time.sleep(0.01)
1095
1096
1097class Bunch(object):
1098 """
1099 A bunch of threads.
1100 """
1101 def __init__(self, namespace, f, args, n, wait_before_exit=False):
1102 """
1103 Construct a bunch of `n` threads running the same function `f`.
1104 If `wait_before_exit` is True, the threads won't terminate until
1105 do_finish() is called.
1106 """
1107 self.f = f
1108 self.args = args
1109 self.n = n
1110 self.started = namespace.DummyList()
1111 self.finished = namespace.DummyList()
Richard Oudkerk0f523462012-06-15 19:18:30 +01001112 self._can_exit = namespace.Event()
1113 if not wait_before_exit:
1114 self._can_exit.set()
Richard Oudkerk3730a172012-06-15 18:26:07 +01001115 for i in range(n):
Richard Oudkerk0f523462012-06-15 19:18:30 +01001116 p = namespace.Process(target=self.task)
1117 p.daemon = True
1118 p.start()
Richard Oudkerk3730a172012-06-15 18:26:07 +01001119
1120 def task(self):
1121 pid = os.getpid()
1122 self.started.append(pid)
1123 try:
1124 self.f(*self.args)
1125 finally:
1126 self.finished.append(pid)
Richard Oudkerk0f523462012-06-15 19:18:30 +01001127 self._can_exit.wait(30)
1128 assert self._can_exit.is_set()
Richard Oudkerk3730a172012-06-15 18:26:07 +01001129
1130 def wait_for_started(self):
1131 while len(self.started) < self.n:
1132 _wait()
1133
1134 def wait_for_finished(self):
1135 while len(self.finished) < self.n:
1136 _wait()
1137
1138 def do_finish(self):
Richard Oudkerk0f523462012-06-15 19:18:30 +01001139 self._can_exit.set()
Richard Oudkerk3730a172012-06-15 18:26:07 +01001140
1141
1142class AppendTrue(object):
1143 def __init__(self, obj):
1144 self.obj = obj
1145 def __call__(self):
1146 self.obj.append(True)
1147
1148
1149class _TestBarrier(BaseTestCase):
1150 """
1151 Tests for Barrier objects.
1152 """
1153 N = 5
1154 defaultTimeout = 10.0 # XXX Slow Windows buildbots need generous timeout
1155
1156 def setUp(self):
1157 self.barrier = self.Barrier(self.N, timeout=self.defaultTimeout)
1158
1159 def tearDown(self):
1160 self.barrier.abort()
1161 self.barrier = None
1162
1163 def DummyList(self):
1164 if self.TYPE == 'threads':
1165 return []
1166 elif self.TYPE == 'manager':
1167 return self.manager.list()
1168 else:
1169 return _DummyList()
1170
1171 def run_threads(self, f, args):
1172 b = Bunch(self, f, args, self.N-1)
1173 f(*args)
1174 b.wait_for_finished()
1175
1176 @classmethod
1177 def multipass(cls, barrier, results, n):
1178 m = barrier.parties
1179 assert m == cls.N
1180 for i in range(n):
1181 results[0].append(True)
1182 assert len(results[1]) == i * m
1183 barrier.wait()
1184 results[1].append(True)
1185 assert len(results[0]) == (i + 1) * m
1186 barrier.wait()
1187 try:
1188 assert barrier.n_waiting == 0
1189 except NotImplementedError:
1190 pass
1191 assert not barrier.broken
1192
1193 def test_barrier(self, passes=1):
1194 """
1195 Test that a barrier is passed in lockstep
1196 """
1197 results = [self.DummyList(), self.DummyList()]
1198 self.run_threads(self.multipass, (self.barrier, results, passes))
1199
1200 def test_barrier_10(self):
1201 """
1202 Test that a barrier works for 10 consecutive runs
1203 """
1204 return self.test_barrier(10)
1205
1206 @classmethod
1207 def _test_wait_return_f(cls, barrier, queue):
1208 res = barrier.wait()
1209 queue.put(res)
1210
1211 def test_wait_return(self):
1212 """
1213 test the return value from barrier.wait
1214 """
1215 queue = self.Queue()
1216 self.run_threads(self._test_wait_return_f, (self.barrier, queue))
1217 results = [queue.get() for i in range(self.N)]
1218 self.assertEqual(results.count(0), 1)
1219
1220 @classmethod
1221 def _test_action_f(cls, barrier, results):
1222 barrier.wait()
1223 if len(results) != 1:
1224 raise RuntimeError
1225
1226 def test_action(self):
1227 """
1228 Test the 'action' callback
1229 """
1230 results = self.DummyList()
1231 barrier = self.Barrier(self.N, action=AppendTrue(results))
1232 self.run_threads(self._test_action_f, (barrier, results))
1233 self.assertEqual(len(results), 1)
1234
1235 @classmethod
1236 def _test_abort_f(cls, barrier, results1, results2):
1237 try:
1238 i = barrier.wait()
1239 if i == cls.N//2:
1240 raise RuntimeError
1241 barrier.wait()
1242 results1.append(True)
1243 except threading.BrokenBarrierError:
1244 results2.append(True)
1245 except RuntimeError:
1246 barrier.abort()
1247
1248 def test_abort(self):
1249 """
1250 Test that an abort will put the barrier in a broken state
1251 """
1252 results1 = self.DummyList()
1253 results2 = self.DummyList()
1254 self.run_threads(self._test_abort_f,
1255 (self.barrier, results1, results2))
1256 self.assertEqual(len(results1), 0)
1257 self.assertEqual(len(results2), self.N-1)
1258 self.assertTrue(self.barrier.broken)
1259
1260 @classmethod
1261 def _test_reset_f(cls, barrier, results1, results2, results3):
1262 i = barrier.wait()
1263 if i == cls.N//2:
1264 # Wait until the other threads are all in the barrier.
1265 while barrier.n_waiting < cls.N-1:
1266 time.sleep(0.001)
1267 barrier.reset()
1268 else:
1269 try:
1270 barrier.wait()
1271 results1.append(True)
1272 except threading.BrokenBarrierError:
1273 results2.append(True)
1274 # Now, pass the barrier again
1275 barrier.wait()
1276 results3.append(True)
1277
1278 def test_reset(self):
1279 """
1280 Test that a 'reset' on a barrier frees the waiting threads
1281 """
1282 results1 = self.DummyList()
1283 results2 = self.DummyList()
1284 results3 = self.DummyList()
1285 self.run_threads(self._test_reset_f,
1286 (self.barrier, results1, results2, results3))
1287 self.assertEqual(len(results1), 0)
1288 self.assertEqual(len(results2), self.N-1)
1289 self.assertEqual(len(results3), self.N)
1290
1291 @classmethod
1292 def _test_abort_and_reset_f(cls, barrier, barrier2,
1293 results1, results2, results3):
1294 try:
1295 i = barrier.wait()
1296 if i == cls.N//2:
1297 raise RuntimeError
1298 barrier.wait()
1299 results1.append(True)
1300 except threading.BrokenBarrierError:
1301 results2.append(True)
1302 except RuntimeError:
1303 barrier.abort()
1304 # Synchronize and reset the barrier. Must synchronize first so
1305 # that everyone has left it when we reset, and after so that no
1306 # one enters it before the reset.
1307 if barrier2.wait() == cls.N//2:
1308 barrier.reset()
1309 barrier2.wait()
1310 barrier.wait()
1311 results3.append(True)
1312
1313 def test_abort_and_reset(self):
1314 """
1315 Test that a barrier can be reset after being broken.
1316 """
1317 results1 = self.DummyList()
1318 results2 = self.DummyList()
1319 results3 = self.DummyList()
1320 barrier2 = self.Barrier(self.N)
1321
1322 self.run_threads(self._test_abort_and_reset_f,
1323 (self.barrier, barrier2, results1, results2, results3))
1324 self.assertEqual(len(results1), 0)
1325 self.assertEqual(len(results2), self.N-1)
1326 self.assertEqual(len(results3), self.N)
1327
1328 @classmethod
1329 def _test_timeout_f(cls, barrier, results):
1330 i = barrier.wait(20)
1331 if i == cls.N//2:
1332 # One thread is late!
1333 time.sleep(4.0)
1334 try:
1335 barrier.wait(0.5)
1336 except threading.BrokenBarrierError:
1337 results.append(True)
1338
1339 def test_timeout(self):
1340 """
1341 Test wait(timeout)
1342 """
1343 results = self.DummyList()
1344 self.run_threads(self._test_timeout_f, (self.barrier, results))
1345 self.assertEqual(len(results), self.barrier.parties)
1346
1347 @classmethod
1348 def _test_default_timeout_f(cls, barrier, results):
1349 i = barrier.wait(20)
1350 if i == cls.N//2:
1351 # One thread is later than the default timeout
1352 time.sleep(4.0)
1353 try:
1354 barrier.wait()
1355 except threading.BrokenBarrierError:
1356 results.append(True)
1357
1358 def test_default_timeout(self):
1359 """
1360 Test the barrier's default timeout
1361 """
1362 barrier = self.Barrier(self.N, timeout=1.0)
1363 results = self.DummyList()
1364 self.run_threads(self._test_default_timeout_f, (barrier, results))
1365 self.assertEqual(len(results), barrier.parties)
1366
1367 def test_single_thread(self):
1368 b = self.Barrier(1)
1369 b.wait()
1370 b.wait()
1371
1372 @classmethod
1373 def _test_thousand_f(cls, barrier, passes, conn, lock):
1374 for i in range(passes):
1375 barrier.wait()
1376 with lock:
1377 conn.send(i)
1378
1379 def test_thousand(self):
1380 if self.TYPE == 'manager':
1381 return
1382 passes = 1000
1383 lock = self.Lock()
1384 conn, child_conn = self.Pipe(False)
1385 for j in range(self.N):
1386 p = self.Process(target=self._test_thousand_f,
1387 args=(self.barrier, passes, child_conn, lock))
1388 p.start()
1389
1390 for i in range(passes):
1391 for j in range(self.N):
1392 self.assertEqual(conn.recv(), i)
1393
1394#
Benjamin Petersone711caf2008-06-11 16:44:04 +00001395#
1396#
1397
1398class _TestValue(BaseTestCase):
1399
Florent Xiclunafd1b0932010-03-28 00:25:02 +00001400 ALLOWED_TYPES = ('processes',)
1401
Benjamin Petersone711caf2008-06-11 16:44:04 +00001402 codes_values = [
1403 ('i', 4343, 24234),
1404 ('d', 3.625, -4.25),
1405 ('h', -232, 234),
1406 ('c', latin('x'), latin('y'))
1407 ]
1408
Antoine Pitrou7744e2a2010-11-22 16:26:21 +00001409 def setUp(self):
1410 if not HAS_SHAREDCTYPES:
1411 self.skipTest("requires multiprocessing.sharedctypes")
1412
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00001413 @classmethod
1414 def _test(cls, values):
1415 for sv, cv in zip(values, cls.codes_values):
Benjamin Petersone711caf2008-06-11 16:44:04 +00001416 sv.value = cv[2]
1417
1418
1419 def test_value(self, raw=False):
Benjamin Petersone711caf2008-06-11 16:44:04 +00001420 if raw:
1421 values = [self.RawValue(code, value)
1422 for code, value, _ in self.codes_values]
1423 else:
1424 values = [self.Value(code, value)
1425 for code, value, _ in self.codes_values]
1426
1427 for sv, cv in zip(values, self.codes_values):
1428 self.assertEqual(sv.value, cv[1])
1429
1430 proc = self.Process(target=self._test, args=(values,))
Jesus Cea94f964f2011-09-09 20:26:57 +02001431 proc.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00001432 proc.start()
1433 proc.join()
1434
1435 for sv, cv in zip(values, self.codes_values):
1436 self.assertEqual(sv.value, cv[2])
1437
1438 def test_rawvalue(self):
1439 self.test_value(raw=True)
1440
1441 def test_getobj_getlock(self):
Benjamin Petersone711caf2008-06-11 16:44:04 +00001442 val1 = self.Value('i', 5)
1443 lock1 = val1.get_lock()
1444 obj1 = val1.get_obj()
1445
1446 val2 = self.Value('i', 5, lock=None)
1447 lock2 = val2.get_lock()
1448 obj2 = val2.get_obj()
1449
1450 lock = self.Lock()
1451 val3 = self.Value('i', 5, lock=lock)
1452 lock3 = val3.get_lock()
1453 obj3 = val3.get_obj()
1454 self.assertEqual(lock, lock3)
1455
Jesse Nollerb0516a62009-01-18 03:11:38 +00001456 arr4 = self.Value('i', 5, lock=False)
Benjamin Petersone711caf2008-06-11 16:44:04 +00001457 self.assertFalse(hasattr(arr4, 'get_lock'))
1458 self.assertFalse(hasattr(arr4, 'get_obj'))
1459
Jesse Nollerb0516a62009-01-18 03:11:38 +00001460 self.assertRaises(AttributeError, self.Value, 'i', 5, lock='navalue')
1461
1462 arr5 = self.RawValue('i', 5)
1463 self.assertFalse(hasattr(arr5, 'get_lock'))
1464 self.assertFalse(hasattr(arr5, 'get_obj'))
1465
Benjamin Petersone711caf2008-06-11 16:44:04 +00001466
1467class _TestArray(BaseTestCase):
1468
Florent Xiclunafd1b0932010-03-28 00:25:02 +00001469 ALLOWED_TYPES = ('processes',)
1470
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00001471 @classmethod
1472 def f(cls, seq):
Benjamin Petersone711caf2008-06-11 16:44:04 +00001473 for i in range(1, len(seq)):
1474 seq[i] += seq[i-1]
1475
Florent Xiclunafd1b0932010-03-28 00:25:02 +00001476 @unittest.skipIf(c_int is None, "requires _ctypes")
Benjamin Petersone711caf2008-06-11 16:44:04 +00001477 def test_array(self, raw=False):
Benjamin Petersone711caf2008-06-11 16:44:04 +00001478 seq = [680, 626, 934, 821, 150, 233, 548, 982, 714, 831]
1479 if raw:
1480 arr = self.RawArray('i', seq)
1481 else:
1482 arr = self.Array('i', seq)
1483
1484 self.assertEqual(len(arr), len(seq))
1485 self.assertEqual(arr[3], seq[3])
1486 self.assertEqual(list(arr[2:7]), list(seq[2:7]))
1487
1488 arr[4:8] = seq[4:8] = array.array('i', [1, 2, 3, 4])
1489
1490 self.assertEqual(list(arr[:]), seq)
1491
1492 self.f(seq)
1493
1494 p = self.Process(target=self.f, args=(arr,))
Jesus Cea94f964f2011-09-09 20:26:57 +02001495 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00001496 p.start()
1497 p.join()
1498
1499 self.assertEqual(list(arr[:]), seq)
1500
Florent Xiclunafd1b0932010-03-28 00:25:02 +00001501 @unittest.skipIf(c_int is None, "requires _ctypes")
Mark Dickinson89461ef2011-03-26 10:19:03 +00001502 def test_array_from_size(self):
1503 size = 10
1504 # Test for zeroing (see issue #11675).
1505 # The repetition below strengthens the test by increasing the chances
1506 # of previously allocated non-zero memory being used for the new array
1507 # on the 2nd and 3rd loops.
1508 for _ in range(3):
1509 arr = self.Array('i', size)
1510 self.assertEqual(len(arr), size)
1511 self.assertEqual(list(arr), [0] * size)
1512 arr[:] = range(10)
1513 self.assertEqual(list(arr), list(range(10)))
1514 del arr
1515
1516 @unittest.skipIf(c_int is None, "requires _ctypes")
Benjamin Petersone711caf2008-06-11 16:44:04 +00001517 def test_rawarray(self):
1518 self.test_array(raw=True)
1519
Florent Xiclunafd1b0932010-03-28 00:25:02 +00001520 @unittest.skipIf(c_int is None, "requires _ctypes")
Benjamin Petersone711caf2008-06-11 16:44:04 +00001521 def test_getobj_getlock_obj(self):
Benjamin Petersone711caf2008-06-11 16:44:04 +00001522 arr1 = self.Array('i', list(range(10)))
1523 lock1 = arr1.get_lock()
1524 obj1 = arr1.get_obj()
1525
1526 arr2 = self.Array('i', list(range(10)), lock=None)
1527 lock2 = arr2.get_lock()
1528 obj2 = arr2.get_obj()
1529
1530 lock = self.Lock()
1531 arr3 = self.Array('i', list(range(10)), lock=lock)
1532 lock3 = arr3.get_lock()
1533 obj3 = arr3.get_obj()
1534 self.assertEqual(lock, lock3)
1535
Jesse Nollerb0516a62009-01-18 03:11:38 +00001536 arr4 = self.Array('i', range(10), lock=False)
Benjamin Petersone711caf2008-06-11 16:44:04 +00001537 self.assertFalse(hasattr(arr4, 'get_lock'))
1538 self.assertFalse(hasattr(arr4, 'get_obj'))
Jesse Nollerb0516a62009-01-18 03:11:38 +00001539 self.assertRaises(AttributeError,
1540 self.Array, 'i', range(10), lock='notalock')
1541
1542 arr5 = self.RawArray('i', range(10))
1543 self.assertFalse(hasattr(arr5, 'get_lock'))
1544 self.assertFalse(hasattr(arr5, 'get_obj'))
Benjamin Petersone711caf2008-06-11 16:44:04 +00001545
1546#
1547#
1548#
1549
1550class _TestContainers(BaseTestCase):
1551
1552 ALLOWED_TYPES = ('manager',)
1553
1554 def test_list(self):
1555 a = self.list(list(range(10)))
1556 self.assertEqual(a[:], list(range(10)))
1557
1558 b = self.list()
1559 self.assertEqual(b[:], [])
1560
1561 b.extend(list(range(5)))
1562 self.assertEqual(b[:], list(range(5)))
1563
1564 self.assertEqual(b[2], 2)
1565 self.assertEqual(b[2:10], [2,3,4])
1566
1567 b *= 2
1568 self.assertEqual(b[:], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4])
1569
1570 self.assertEqual(b + [5, 6], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4, 5, 6])
1571
1572 self.assertEqual(a[:], list(range(10)))
1573
1574 d = [a, b]
1575 e = self.list(d)
1576 self.assertEqual(
1577 e[:],
1578 [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4]]
1579 )
1580
1581 f = self.list([a])
1582 a.append('hello')
1583 self.assertEqual(f[:], [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 'hello']])
1584
1585 def test_dict(self):
1586 d = self.dict()
1587 indices = list(range(65, 70))
1588 for i in indices:
1589 d[i] = chr(i)
1590 self.assertEqual(d.copy(), dict((i, chr(i)) for i in indices))
1591 self.assertEqual(sorted(d.keys()), indices)
1592 self.assertEqual(sorted(d.values()), [chr(i) for i in indices])
1593 self.assertEqual(sorted(d.items()), [(i, chr(i)) for i in indices])
1594
1595 def test_namespace(self):
1596 n = self.Namespace()
1597 n.name = 'Bob'
1598 n.job = 'Builder'
1599 n._hidden = 'hidden'
1600 self.assertEqual((n.name, n.job), ('Bob', 'Builder'))
1601 del n.job
1602 self.assertEqual(str(n), "Namespace(name='Bob')")
1603 self.assertTrue(hasattr(n, 'name'))
1604 self.assertTrue(not hasattr(n, 'job'))
1605
1606#
1607#
1608#
1609
1610def sqr(x, wait=0.0):
1611 time.sleep(wait)
1612 return x*x
Ask Solem2afcbf22010-11-09 20:55:52 +00001613
Antoine Pitroude911b22011-12-21 11:03:24 +01001614def mul(x, y):
1615 return x*y
1616
Benjamin Petersone711caf2008-06-11 16:44:04 +00001617class _TestPool(BaseTestCase):
1618
1619 def test_apply(self):
1620 papply = self.pool.apply
1621 self.assertEqual(papply(sqr, (5,)), sqr(5))
1622 self.assertEqual(papply(sqr, (), {'x':3}), sqr(x=3))
1623
1624 def test_map(self):
1625 pmap = self.pool.map
1626 self.assertEqual(pmap(sqr, list(range(10))), list(map(sqr, list(range(10)))))
1627 self.assertEqual(pmap(sqr, list(range(100)), chunksize=20),
1628 list(map(sqr, list(range(100)))))
1629
Antoine Pitroude911b22011-12-21 11:03:24 +01001630 def test_starmap(self):
1631 psmap = self.pool.starmap
1632 tuples = list(zip(range(10), range(9,-1, -1)))
1633 self.assertEqual(psmap(mul, tuples),
1634 list(itertools.starmap(mul, tuples)))
1635 tuples = list(zip(range(100), range(99,-1, -1)))
1636 self.assertEqual(psmap(mul, tuples, chunksize=20),
1637 list(itertools.starmap(mul, tuples)))
1638
1639 def test_starmap_async(self):
1640 tuples = list(zip(range(100), range(99,-1, -1)))
1641 self.assertEqual(self.pool.starmap_async(mul, tuples).get(),
1642 list(itertools.starmap(mul, tuples)))
1643
Alexandre Vassalottie52e3782009-07-17 09:18:18 +00001644 def test_map_chunksize(self):
1645 try:
1646 self.pool.map_async(sqr, [], chunksize=1).get(timeout=TIMEOUT1)
1647 except multiprocessing.TimeoutError:
1648 self.fail("pool.map_async with chunksize stalled on null list")
1649
Benjamin Petersone711caf2008-06-11 16:44:04 +00001650 def test_async(self):
1651 res = self.pool.apply_async(sqr, (7, TIMEOUT1,))
1652 get = TimingWrapper(res.get)
1653 self.assertEqual(get(), 49)
1654 self.assertTimingAlmostEqual(get.elapsed, TIMEOUT1)
1655
1656 def test_async_timeout(self):
1657 res = self.pool.apply_async(sqr, (6, TIMEOUT2 + 0.2))
1658 get = TimingWrapper(res.get)
1659 self.assertRaises(multiprocessing.TimeoutError, get, timeout=TIMEOUT2)
1660 self.assertTimingAlmostEqual(get.elapsed, TIMEOUT2)
1661
1662 def test_imap(self):
1663 it = self.pool.imap(sqr, list(range(10)))
1664 self.assertEqual(list(it), list(map(sqr, list(range(10)))))
1665
1666 it = self.pool.imap(sqr, list(range(10)))
1667 for i in range(10):
1668 self.assertEqual(next(it), i*i)
1669 self.assertRaises(StopIteration, it.__next__)
1670
1671 it = self.pool.imap(sqr, list(range(1000)), chunksize=100)
1672 for i in range(1000):
1673 self.assertEqual(next(it), i*i)
1674 self.assertRaises(StopIteration, it.__next__)
1675
1676 def test_imap_unordered(self):
1677 it = self.pool.imap_unordered(sqr, list(range(1000)))
1678 self.assertEqual(sorted(it), list(map(sqr, list(range(1000)))))
1679
1680 it = self.pool.imap_unordered(sqr, list(range(1000)), chunksize=53)
1681 self.assertEqual(sorted(it), list(map(sqr, list(range(1000)))))
1682
1683 def test_make_pool(self):
Victor Stinner2fae27b2011-06-20 17:53:35 +02001684 self.assertRaises(ValueError, multiprocessing.Pool, -1)
1685 self.assertRaises(ValueError, multiprocessing.Pool, 0)
1686
Benjamin Petersone711caf2008-06-11 16:44:04 +00001687 p = multiprocessing.Pool(3)
1688 self.assertEqual(3, len(p._pool))
1689 p.close()
1690 p.join()
1691
1692 def test_terminate(self):
1693 if self.TYPE == 'manager':
1694 # On Unix a forked process increfs each shared object to
1695 # which its parent process held a reference. If the
1696 # forked process gets terminated then there is likely to
1697 # be a reference leak. So to prevent
1698 # _TestZZZNumberOfObjects from failing we skip this test
1699 # when using a manager.
1700 return
1701
1702 result = self.pool.map_async(
1703 time.sleep, [0.1 for i in range(10000)], chunksize=1
1704 )
1705 self.pool.terminate()
1706 join = TimingWrapper(self.pool.join)
1707 join()
Victor Stinner900189b2011-03-24 16:39:07 +01001708 self.assertLess(join.elapsed, 0.5)
Jesse Noller1f0b6582010-01-27 03:36:01 +00001709
Richard Oudkerke41682b2012-06-06 19:04:57 +01001710 def test_empty_iterable(self):
1711 # See Issue 12157
1712 p = self.Pool(1)
1713
1714 self.assertEqual(p.map(sqr, []), [])
1715 self.assertEqual(list(p.imap(sqr, [])), [])
1716 self.assertEqual(list(p.imap_unordered(sqr, [])), [])
1717 self.assertEqual(p.map_async(sqr, []).get(), [])
1718
1719 p.close()
1720 p.join()
1721
Ask Solem2afcbf22010-11-09 20:55:52 +00001722def raising():
1723 raise KeyError("key")
Jesse Noller1f0b6582010-01-27 03:36:01 +00001724
Ask Solem2afcbf22010-11-09 20:55:52 +00001725def unpickleable_result():
1726 return lambda: 42
1727
1728class _TestPoolWorkerErrors(BaseTestCase):
Jesse Noller1f0b6582010-01-27 03:36:01 +00001729 ALLOWED_TYPES = ('processes', )
Ask Solem2afcbf22010-11-09 20:55:52 +00001730
1731 def test_async_error_callback(self):
1732 p = multiprocessing.Pool(2)
1733
1734 scratchpad = [None]
1735 def errback(exc):
1736 scratchpad[0] = exc
1737
1738 res = p.apply_async(raising, error_callback=errback)
1739 self.assertRaises(KeyError, res.get)
1740 self.assertTrue(scratchpad[0])
1741 self.assertIsInstance(scratchpad[0], KeyError)
1742
1743 p.close()
1744 p.join()
1745
1746 def test_unpickleable_result(self):
1747 from multiprocessing.pool import MaybeEncodingError
1748 p = multiprocessing.Pool(2)
1749
1750 # Make sure we don't lose pool processes because of encoding errors.
1751 for iteration in range(20):
1752
1753 scratchpad = [None]
1754 def errback(exc):
1755 scratchpad[0] = exc
1756
1757 res = p.apply_async(unpickleable_result, error_callback=errback)
1758 self.assertRaises(MaybeEncodingError, res.get)
1759 wrapped = scratchpad[0]
1760 self.assertTrue(wrapped)
1761 self.assertIsInstance(scratchpad[0], MaybeEncodingError)
1762 self.assertIsNotNone(wrapped.exc)
1763 self.assertIsNotNone(wrapped.value)
1764
1765 p.close()
1766 p.join()
1767
1768class _TestPoolWorkerLifetime(BaseTestCase):
1769 ALLOWED_TYPES = ('processes', )
1770
Jesse Noller1f0b6582010-01-27 03:36:01 +00001771 def test_pool_worker_lifetime(self):
1772 p = multiprocessing.Pool(3, maxtasksperchild=10)
1773 self.assertEqual(3, len(p._pool))
1774 origworkerpids = [w.pid for w in p._pool]
1775 # Run many tasks so each worker gets replaced (hopefully)
1776 results = []
1777 for i in range(100):
1778 results.append(p.apply_async(sqr, (i, )))
1779 # Fetch the results and verify we got the right answers,
1780 # also ensuring all the tasks have completed.
1781 for (j, res) in enumerate(results):
1782 self.assertEqual(res.get(), sqr(j))
1783 # Refill the pool
1784 p._repopulate_pool()
Florent Xiclunafb190f62010-03-04 16:10:10 +00001785 # Wait until all workers are alive
Antoine Pitrou540ab062011-04-06 22:51:17 +02001786 # (countdown * DELTA = 5 seconds max startup process time)
1787 countdown = 50
Florent Xiclunafb190f62010-03-04 16:10:10 +00001788 while countdown and not all(w.is_alive() for w in p._pool):
1789 countdown -= 1
1790 time.sleep(DELTA)
Jesse Noller1f0b6582010-01-27 03:36:01 +00001791 finalworkerpids = [w.pid for w in p._pool]
Florent Xiclunafb190f62010-03-04 16:10:10 +00001792 # All pids should be assigned. See issue #7805.
1793 self.assertNotIn(None, origworkerpids)
1794 self.assertNotIn(None, finalworkerpids)
1795 # Finally, check that the worker pids have changed
Jesse Noller1f0b6582010-01-27 03:36:01 +00001796 self.assertNotEqual(sorted(origworkerpids), sorted(finalworkerpids))
1797 p.close()
1798 p.join()
1799
Charles-François Natalif8859e12011-10-24 18:45:29 +02001800 def test_pool_worker_lifetime_early_close(self):
1801 # Issue #10332: closing a pool whose workers have limited lifetimes
1802 # before all the tasks completed would make join() hang.
1803 p = multiprocessing.Pool(3, maxtasksperchild=1)
1804 results = []
1805 for i in range(6):
1806 results.append(p.apply_async(sqr, (i, 0.3)))
1807 p.close()
1808 p.join()
1809 # check the results
1810 for (j, res) in enumerate(results):
1811 self.assertEqual(res.get(), sqr(j))
1812
1813
Benjamin Petersone711caf2008-06-11 16:44:04 +00001814#
1815# Test that manager has expected number of shared objects left
1816#
1817
1818class _TestZZZNumberOfObjects(BaseTestCase):
1819 # Because test cases are sorted alphabetically, this one will get
1820 # run after all the other tests for the manager. It tests that
1821 # there have been no "reference leaks" for the manager's shared
1822 # objects. Note the comment in _TestPool.test_terminate().
1823 ALLOWED_TYPES = ('manager',)
1824
1825 def test_number_of_objects(self):
1826 EXPECTED_NUMBER = 1 # the pool object is still alive
1827 multiprocessing.active_children() # discard dead process objs
1828 gc.collect() # do garbage collection
1829 refs = self.manager._number_of_objects()
Jesse Noller63b3a972009-01-21 02:15:48 +00001830 debug_info = self.manager._debug_info()
Benjamin Petersone711caf2008-06-11 16:44:04 +00001831 if refs != EXPECTED_NUMBER:
Georg Brandl3dbca812008-07-23 16:10:53 +00001832 print(self.manager._debug_info())
Jesse Noller63b3a972009-01-21 02:15:48 +00001833 print(debug_info)
Benjamin Petersone711caf2008-06-11 16:44:04 +00001834
1835 self.assertEqual(refs, EXPECTED_NUMBER)
1836
1837#
1838# Test of creating a customized manager class
1839#
1840
1841from multiprocessing.managers import BaseManager, BaseProxy, RemoteError
1842
1843class FooBar(object):
1844 def f(self):
1845 return 'f()'
1846 def g(self):
1847 raise ValueError
1848 def _h(self):
1849 return '_h()'
1850
1851def baz():
1852 for i in range(10):
1853 yield i*i
1854
1855class IteratorProxy(BaseProxy):
Florent Xiclunaaa171062010-08-14 15:56:42 +00001856 _exposed_ = ('__next__',)
Benjamin Petersone711caf2008-06-11 16:44:04 +00001857 def __iter__(self):
1858 return self
1859 def __next__(self):
Benjamin Petersone711caf2008-06-11 16:44:04 +00001860 return self._callmethod('__next__')
1861
1862class MyManager(BaseManager):
1863 pass
1864
1865MyManager.register('Foo', callable=FooBar)
1866MyManager.register('Bar', callable=FooBar, exposed=('f', '_h'))
1867MyManager.register('baz', callable=baz, proxytype=IteratorProxy)
1868
1869
1870class _TestMyManager(BaseTestCase):
1871
1872 ALLOWED_TYPES = ('manager',)
1873
1874 def test_mymanager(self):
1875 manager = MyManager()
1876 manager.start()
1877
1878 foo = manager.Foo()
1879 bar = manager.Bar()
1880 baz = manager.baz()
1881
1882 foo_methods = [name for name in ('f', 'g', '_h') if hasattr(foo, name)]
1883 bar_methods = [name for name in ('f', 'g', '_h') if hasattr(bar, name)]
1884
1885 self.assertEqual(foo_methods, ['f', 'g'])
1886 self.assertEqual(bar_methods, ['f', '_h'])
1887
1888 self.assertEqual(foo.f(), 'f()')
1889 self.assertRaises(ValueError, foo.g)
1890 self.assertEqual(foo._callmethod('f'), 'f()')
1891 self.assertRaises(RemoteError, foo._callmethod, '_h')
1892
1893 self.assertEqual(bar.f(), 'f()')
1894 self.assertEqual(bar._h(), '_h()')
1895 self.assertEqual(bar._callmethod('f'), 'f()')
1896 self.assertEqual(bar._callmethod('_h'), '_h()')
1897
1898 self.assertEqual(list(baz), [i*i for i in range(10)])
1899
1900 manager.shutdown()
1901
Richard Oudkerk73d9a292012-06-14 15:30:10 +01001902 # If the manager process exited cleanly then the exitcode
1903 # will be zero. Otherwise (after a short timeout)
1904 # terminate() is used, resulting in an exitcode of -SIGTERM.
1905 self.assertEqual(manager._process.exitcode, 0)
1906
Benjamin Petersone711caf2008-06-11 16:44:04 +00001907#
1908# Test of connecting to a remote server and using xmlrpclib for serialization
1909#
1910
1911_queue = pyqueue.Queue()
1912def get_queue():
1913 return _queue
1914
1915class QueueManager(BaseManager):
1916 '''manager class used by server process'''
1917QueueManager.register('get_queue', callable=get_queue)
1918
1919class QueueManager2(BaseManager):
1920 '''manager class which specifies the same interface as QueueManager'''
1921QueueManager2.register('get_queue')
1922
1923
1924SERIALIZER = 'xmlrpclib'
1925
1926class _TestRemoteManager(BaseTestCase):
1927
1928 ALLOWED_TYPES = ('manager',)
1929
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00001930 @classmethod
1931 def _putter(cls, address, authkey):
Benjamin Petersone711caf2008-06-11 16:44:04 +00001932 manager = QueueManager2(
1933 address=address, authkey=authkey, serializer=SERIALIZER
1934 )
1935 manager.connect()
1936 queue = manager.get_queue()
1937 queue.put(('hello world', None, True, 2.25))
1938
1939 def test_remote(self):
1940 authkey = os.urandom(32)
1941
1942 manager = QueueManager(
1943 address=('localhost', 0), authkey=authkey, serializer=SERIALIZER
1944 )
1945 manager.start()
1946
1947 p = self.Process(target=self._putter, args=(manager.address, authkey))
Jesus Cea94f964f2011-09-09 20:26:57 +02001948 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00001949 p.start()
1950
1951 manager2 = QueueManager2(
1952 address=manager.address, authkey=authkey, serializer=SERIALIZER
1953 )
1954 manager2.connect()
1955 queue = manager2.get_queue()
1956
1957 # Note that xmlrpclib will deserialize object as a list not a tuple
1958 self.assertEqual(queue.get(), ['hello world', None, True, 2.25])
1959
1960 # Because we are using xmlrpclib for serialization instead of
1961 # pickle this will cause a serialization error.
1962 self.assertRaises(Exception, queue.put, time.sleep)
1963
1964 # Make queue finalizer run before the server is stopped
1965 del queue
1966 manager.shutdown()
1967
Jesse Nollerc5d28a02009-03-30 16:37:36 +00001968class _TestManagerRestart(BaseTestCase):
1969
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00001970 @classmethod
1971 def _putter(cls, address, authkey):
Jesse Nollerc5d28a02009-03-30 16:37:36 +00001972 manager = QueueManager(
1973 address=address, authkey=authkey, serializer=SERIALIZER)
1974 manager.connect()
1975 queue = manager.get_queue()
1976 queue.put('hello world')
1977
1978 def test_rapid_restart(self):
1979 authkey = os.urandom(32)
1980 manager = QueueManager(
Antoine Pitrou043bad02010-04-30 23:20:15 +00001981 address=('localhost', 0), authkey=authkey, serializer=SERIALIZER)
Brian Curtin50be1ca2010-11-01 05:10:44 +00001982 srvr = manager.get_server()
1983 addr = srvr.address
1984 # Close the connection.Listener socket which gets opened as a part
1985 # of manager.get_server(). It's not needed for the test.
1986 srvr.listener.close()
Jesse Nollerc5d28a02009-03-30 16:37:36 +00001987 manager.start()
1988
1989 p = self.Process(target=self._putter, args=(manager.address, authkey))
Jesus Cea94f964f2011-09-09 20:26:57 +02001990 p.daemon = True
Jesse Nollerc5d28a02009-03-30 16:37:36 +00001991 p.start()
1992 queue = manager.get_queue()
1993 self.assertEqual(queue.get(), 'hello world')
Jesse Noller35d1f002009-03-30 22:59:27 +00001994 del queue
Jesse Nollerc5d28a02009-03-30 16:37:36 +00001995 manager.shutdown()
1996 manager = QueueManager(
Antoine Pitrou043bad02010-04-30 23:20:15 +00001997 address=addr, authkey=authkey, serializer=SERIALIZER)
Antoine Pitrouc824e9a2011-04-05 18:11:33 +02001998 try:
1999 manager.start()
2000 except IOError as e:
2001 if e.errno != errno.EADDRINUSE:
2002 raise
2003 # Retry after some time, in case the old socket was lingering
2004 # (sporadic failure on buildbots)
2005 time.sleep(1.0)
2006 manager = QueueManager(
2007 address=addr, authkey=authkey, serializer=SERIALIZER)
Jesse Noller35d1f002009-03-30 22:59:27 +00002008 manager.shutdown()
Jesse Nollerc5d28a02009-03-30 16:37:36 +00002009
Benjamin Petersone711caf2008-06-11 16:44:04 +00002010#
2011#
2012#
2013
2014SENTINEL = latin('')
2015
2016class _TestConnection(BaseTestCase):
2017
2018 ALLOWED_TYPES = ('processes', 'threads')
2019
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00002020 @classmethod
2021 def _echo(cls, conn):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002022 for msg in iter(conn.recv_bytes, SENTINEL):
2023 conn.send_bytes(msg)
2024 conn.close()
2025
2026 def test_connection(self):
2027 conn, child_conn = self.Pipe()
2028
2029 p = self.Process(target=self._echo, args=(child_conn,))
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +00002030 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00002031 p.start()
2032
2033 seq = [1, 2.25, None]
2034 msg = latin('hello world')
2035 longmsg = msg * 10
2036 arr = array.array('i', list(range(4)))
2037
2038 if self.TYPE == 'processes':
2039 self.assertEqual(type(conn.fileno()), int)
2040
2041 self.assertEqual(conn.send(seq), None)
2042 self.assertEqual(conn.recv(), seq)
2043
2044 self.assertEqual(conn.send_bytes(msg), None)
2045 self.assertEqual(conn.recv_bytes(), msg)
2046
2047 if self.TYPE == 'processes':
2048 buffer = array.array('i', [0]*10)
2049 expected = list(arr) + [0] * (10 - len(arr))
2050 self.assertEqual(conn.send_bytes(arr), None)
2051 self.assertEqual(conn.recv_bytes_into(buffer),
2052 len(arr) * buffer.itemsize)
2053 self.assertEqual(list(buffer), expected)
2054
2055 buffer = array.array('i', [0]*10)
2056 expected = [0] * 3 + list(arr) + [0] * (10 - 3 - len(arr))
2057 self.assertEqual(conn.send_bytes(arr), None)
2058 self.assertEqual(conn.recv_bytes_into(buffer, 3 * buffer.itemsize),
2059 len(arr) * buffer.itemsize)
2060 self.assertEqual(list(buffer), expected)
2061
2062 buffer = bytearray(latin(' ' * 40))
2063 self.assertEqual(conn.send_bytes(longmsg), None)
2064 try:
2065 res = conn.recv_bytes_into(buffer)
2066 except multiprocessing.BufferTooShort as e:
2067 self.assertEqual(e.args, (longmsg,))
2068 else:
2069 self.fail('expected BufferTooShort, got %s' % res)
2070
2071 poll = TimingWrapper(conn.poll)
2072
2073 self.assertEqual(poll(), False)
2074 self.assertTimingAlmostEqual(poll.elapsed, 0)
2075
Richard Oudkerk59d54042012-05-10 16:11:12 +01002076 self.assertEqual(poll(-1), False)
2077 self.assertTimingAlmostEqual(poll.elapsed, 0)
2078
Benjamin Petersone711caf2008-06-11 16:44:04 +00002079 self.assertEqual(poll(TIMEOUT1), False)
2080 self.assertTimingAlmostEqual(poll.elapsed, TIMEOUT1)
2081
2082 conn.send(None)
2083
2084 self.assertEqual(poll(TIMEOUT1), True)
2085 self.assertTimingAlmostEqual(poll.elapsed, 0)
2086
2087 self.assertEqual(conn.recv(), None)
2088
2089 really_big_msg = latin('X') * (1024 * 1024 * 16) # 16Mb
2090 conn.send_bytes(really_big_msg)
2091 self.assertEqual(conn.recv_bytes(), really_big_msg)
2092
2093 conn.send_bytes(SENTINEL) # tell child to quit
2094 child_conn.close()
2095
2096 if self.TYPE == 'processes':
2097 self.assertEqual(conn.readable, True)
2098 self.assertEqual(conn.writable, True)
2099 self.assertRaises(EOFError, conn.recv)
2100 self.assertRaises(EOFError, conn.recv_bytes)
2101
2102 p.join()
2103
2104 def test_duplex_false(self):
2105 reader, writer = self.Pipe(duplex=False)
2106 self.assertEqual(writer.send(1), None)
2107 self.assertEqual(reader.recv(), 1)
2108 if self.TYPE == 'processes':
2109 self.assertEqual(reader.readable, True)
2110 self.assertEqual(reader.writable, False)
2111 self.assertEqual(writer.readable, False)
2112 self.assertEqual(writer.writable, True)
2113 self.assertRaises(IOError, reader.send, 2)
2114 self.assertRaises(IOError, writer.recv)
2115 self.assertRaises(IOError, writer.poll)
2116
2117 def test_spawn_close(self):
2118 # We test that a pipe connection can be closed by parent
2119 # process immediately after child is spawned. On Windows this
2120 # would have sometimes failed on old versions because
2121 # child_conn would be closed before the child got a chance to
2122 # duplicate it.
2123 conn, child_conn = self.Pipe()
2124
2125 p = self.Process(target=self._echo, args=(child_conn,))
Jesus Cea94f964f2011-09-09 20:26:57 +02002126 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00002127 p.start()
2128 child_conn.close() # this might complete before child initializes
2129
2130 msg = latin('hello')
2131 conn.send_bytes(msg)
2132 self.assertEqual(conn.recv_bytes(), msg)
2133
2134 conn.send_bytes(SENTINEL)
2135 conn.close()
2136 p.join()
2137
2138 def test_sendbytes(self):
2139 if self.TYPE != 'processes':
2140 return
2141
2142 msg = latin('abcdefghijklmnopqrstuvwxyz')
2143 a, b = self.Pipe()
2144
2145 a.send_bytes(msg)
2146 self.assertEqual(b.recv_bytes(), msg)
2147
2148 a.send_bytes(msg, 5)
2149 self.assertEqual(b.recv_bytes(), msg[5:])
2150
2151 a.send_bytes(msg, 7, 8)
2152 self.assertEqual(b.recv_bytes(), msg[7:7+8])
2153
2154 a.send_bytes(msg, 26)
2155 self.assertEqual(b.recv_bytes(), latin(''))
2156
2157 a.send_bytes(msg, 26, 0)
2158 self.assertEqual(b.recv_bytes(), latin(''))
2159
2160 self.assertRaises(ValueError, a.send_bytes, msg, 27)
2161
2162 self.assertRaises(ValueError, a.send_bytes, msg, 22, 5)
2163
2164 self.assertRaises(ValueError, a.send_bytes, msg, 26, 1)
2165
2166 self.assertRaises(ValueError, a.send_bytes, msg, -1)
2167
2168 self.assertRaises(ValueError, a.send_bytes, msg, 4, -1)
2169
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002170 @classmethod
2171 def _is_fd_assigned(cls, fd):
2172 try:
2173 os.fstat(fd)
2174 except OSError as e:
2175 if e.errno == errno.EBADF:
2176 return False
2177 raise
2178 else:
2179 return True
2180
2181 @classmethod
2182 def _writefd(cls, conn, data, create_dummy_fds=False):
2183 if create_dummy_fds:
2184 for i in range(0, 256):
2185 if not cls._is_fd_assigned(i):
2186 os.dup2(conn.fileno(), i)
2187 fd = reduction.recv_handle(conn)
2188 if msvcrt:
2189 fd = msvcrt.open_osfhandle(fd, os.O_WRONLY)
2190 os.write(fd, data)
2191 os.close(fd)
2192
Charles-François Natalibc8f0822011-09-20 20:36:51 +02002193 @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002194 def test_fd_transfer(self):
2195 if self.TYPE != 'processes':
2196 self.skipTest("only makes sense with processes")
2197 conn, child_conn = self.Pipe(duplex=True)
2198
2199 p = self.Process(target=self._writefd, args=(child_conn, b"foo"))
Jesus Cea94f964f2011-09-09 20:26:57 +02002200 p.daemon = True
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002201 p.start()
Victor Stinnerd0b10a62011-09-21 01:10:29 +02002202 self.addCleanup(test.support.unlink, test.support.TESTFN)
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002203 with open(test.support.TESTFN, "wb") as f:
2204 fd = f.fileno()
2205 if msvcrt:
2206 fd = msvcrt.get_osfhandle(fd)
2207 reduction.send_handle(conn, fd, p.pid)
2208 p.join()
2209 with open(test.support.TESTFN, "rb") as f:
2210 self.assertEqual(f.read(), b"foo")
2211
Charles-François Natalibc8f0822011-09-20 20:36:51 +02002212 @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002213 @unittest.skipIf(sys.platform == "win32",
2214 "test semantics don't make sense on Windows")
2215 @unittest.skipIf(MAXFD <= 256,
2216 "largest assignable fd number is too small")
2217 @unittest.skipUnless(hasattr(os, "dup2"),
2218 "test needs os.dup2()")
2219 def test_large_fd_transfer(self):
2220 # With fd > 256 (issue #11657)
2221 if self.TYPE != 'processes':
2222 self.skipTest("only makes sense with processes")
2223 conn, child_conn = self.Pipe(duplex=True)
2224
2225 p = self.Process(target=self._writefd, args=(child_conn, b"bar", True))
Jesus Cea94f964f2011-09-09 20:26:57 +02002226 p.daemon = True
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002227 p.start()
Victor Stinnerd0b10a62011-09-21 01:10:29 +02002228 self.addCleanup(test.support.unlink, test.support.TESTFN)
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002229 with open(test.support.TESTFN, "wb") as f:
2230 fd = f.fileno()
2231 for newfd in range(256, MAXFD):
2232 if not self._is_fd_assigned(newfd):
2233 break
2234 else:
2235 self.fail("could not find an unassigned large file descriptor")
2236 os.dup2(fd, newfd)
2237 try:
2238 reduction.send_handle(conn, newfd, p.pid)
2239 finally:
2240 os.close(newfd)
2241 p.join()
2242 with open(test.support.TESTFN, "rb") as f:
2243 self.assertEqual(f.read(), b"bar")
2244
Jesus Cea4507e642011-09-21 03:53:25 +02002245 @classmethod
2246 def _send_data_without_fd(self, conn):
2247 os.write(conn.fileno(), b"\0")
2248
Charles-François Natalie51c8da2011-09-21 18:48:21 +02002249 @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
Jesus Cea4507e642011-09-21 03:53:25 +02002250 @unittest.skipIf(sys.platform == "win32", "doesn't make sense on Windows")
2251 def test_missing_fd_transfer(self):
2252 # Check that exception is raised when received data is not
2253 # accompanied by a file descriptor in ancillary data.
2254 if self.TYPE != 'processes':
2255 self.skipTest("only makes sense with processes")
2256 conn, child_conn = self.Pipe(duplex=True)
2257
2258 p = self.Process(target=self._send_data_without_fd, args=(child_conn,))
2259 p.daemon = True
2260 p.start()
2261 self.assertRaises(RuntimeError, reduction.recv_handle, conn)
2262 p.join()
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002263
Charles-François Natalied4a8fc2012-02-08 21:15:58 +01002264class _TestListener(BaseTestCase):
2265
2266 ALLOWED_TYPES = ('processes')
2267
2268 def test_multiple_bind(self):
2269 for family in self.connection.families:
2270 l = self.connection.Listener(family=family)
2271 self.addCleanup(l.close)
2272 self.assertRaises(OSError, self.connection.Listener,
2273 l.address, family)
2274
Benjamin Petersone711caf2008-06-11 16:44:04 +00002275class _TestListenerClient(BaseTestCase):
2276
2277 ALLOWED_TYPES = ('processes', 'threads')
2278
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00002279 @classmethod
2280 def _test(cls, address):
2281 conn = cls.connection.Client(address)
Benjamin Petersone711caf2008-06-11 16:44:04 +00002282 conn.send('hello')
2283 conn.close()
2284
2285 def test_listener_client(self):
2286 for family in self.connection.families:
2287 l = self.connection.Listener(family=family)
2288 p = self.Process(target=self._test, args=(l.address,))
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +00002289 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00002290 p.start()
2291 conn = l.accept()
2292 self.assertEqual(conn.recv(), 'hello')
2293 p.join()
2294 l.close()
Charles-François Natalied4a8fc2012-02-08 21:15:58 +01002295
Richard Oudkerkfdb8dcf2012-05-05 19:45:37 +01002296 def test_issue14725(self):
2297 l = self.connection.Listener()
2298 p = self.Process(target=self._test, args=(l.address,))
2299 p.daemon = True
2300 p.start()
2301 time.sleep(1)
2302 # On Windows the client process should by now have connected,
2303 # written data and closed the pipe handle by now. This causes
2304 # ConnectNamdedPipe() to fail with ERROR_NO_DATA. See Issue
2305 # 14725.
2306 conn = l.accept()
2307 self.assertEqual(conn.recv(), 'hello')
2308 conn.close()
2309 p.join()
2310 l.close()
2311
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01002312class _TestPoll(unittest.TestCase):
2313
2314 ALLOWED_TYPES = ('processes', 'threads')
2315
2316 def test_empty_string(self):
2317 a, b = self.Pipe()
2318 self.assertEqual(a.poll(), False)
2319 b.send_bytes(b'')
2320 self.assertEqual(a.poll(), True)
2321 self.assertEqual(a.poll(), True)
2322
2323 @classmethod
2324 def _child_strings(cls, conn, strings):
2325 for s in strings:
2326 time.sleep(0.1)
2327 conn.send_bytes(s)
2328 conn.close()
2329
2330 def test_strings(self):
2331 strings = (b'hello', b'', b'a', b'b', b'', b'bye', b'', b'lop')
2332 a, b = self.Pipe()
2333 p = self.Process(target=self._child_strings, args=(b, strings))
2334 p.start()
2335
2336 for s in strings:
2337 for i in range(200):
2338 if a.poll(0.01):
2339 break
2340 x = a.recv_bytes()
2341 self.assertEqual(s, x)
2342
2343 p.join()
2344
2345 @classmethod
2346 def _child_boundaries(cls, r):
2347 # Polling may "pull" a message in to the child process, but we
2348 # don't want it to pull only part of a message, as that would
2349 # corrupt the pipe for any other processes which might later
2350 # read from it.
2351 r.poll(5)
2352
2353 def test_boundaries(self):
2354 r, w = self.Pipe(False)
2355 p = self.Process(target=self._child_boundaries, args=(r,))
2356 p.start()
2357 time.sleep(2)
2358 L = [b"first", b"second"]
2359 for obj in L:
2360 w.send_bytes(obj)
2361 w.close()
2362 p.join()
2363 self.assertIn(r.recv_bytes(), L)
2364
2365 @classmethod
2366 def _child_dont_merge(cls, b):
2367 b.send_bytes(b'a')
2368 b.send_bytes(b'b')
2369 b.send_bytes(b'cd')
2370
2371 def test_dont_merge(self):
2372 a, b = self.Pipe()
2373 self.assertEqual(a.poll(0.0), False)
2374 self.assertEqual(a.poll(0.1), False)
2375
2376 p = self.Process(target=self._child_dont_merge, args=(b,))
2377 p.start()
2378
2379 self.assertEqual(a.recv_bytes(), b'a')
2380 self.assertEqual(a.poll(1.0), True)
2381 self.assertEqual(a.poll(1.0), True)
2382 self.assertEqual(a.recv_bytes(), b'b')
2383 self.assertEqual(a.poll(1.0), True)
2384 self.assertEqual(a.poll(1.0), True)
2385 self.assertEqual(a.poll(0.0), True)
2386 self.assertEqual(a.recv_bytes(), b'cd')
2387
2388 p.join()
2389
Benjamin Petersone711caf2008-06-11 16:44:04 +00002390#
2391# Test of sending connection and socket objects between processes
2392#
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002393
Richard Oudkerk24524192012-04-30 14:48:51 +01002394# Intermittent fails on Mac OS X -- see Issue14669 and Issue12958
2395@unittest.skipIf(sys.platform == "darwin", "fd passing unreliable on Mac OS X")
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002396@unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
Benjamin Petersone711caf2008-06-11 16:44:04 +00002397class _TestPicklingConnections(BaseTestCase):
2398
2399 ALLOWED_TYPES = ('processes',)
2400
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002401 @classmethod
Antoine Pitrou92ff4e12012-04-27 23:51:03 +02002402 def tearDownClass(cls):
2403 from multiprocessing.reduction import resource_sharer
2404 resource_sharer.stop(timeout=5)
2405
2406 @classmethod
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002407 def _listener(cls, conn, families):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002408 for fam in families:
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002409 l = cls.connection.Listener(family=fam)
Benjamin Petersone711caf2008-06-11 16:44:04 +00002410 conn.send(l.address)
2411 new_conn = l.accept()
2412 conn.send(new_conn)
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002413 new_conn.close()
2414 l.close()
Benjamin Petersone711caf2008-06-11 16:44:04 +00002415
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002416 l = socket.socket()
2417 l.bind(('localhost', 0))
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002418 l.listen(1)
Richard Oudkerk5d73c172012-05-08 22:24:47 +01002419 conn.send(l.getsockname())
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002420 new_conn, addr = l.accept()
2421 conn.send(new_conn)
2422 new_conn.close()
2423 l.close()
Benjamin Petersone711caf2008-06-11 16:44:04 +00002424
2425 conn.recv()
2426
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002427 @classmethod
2428 def _remote(cls, conn):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002429 for (address, msg) in iter(conn.recv, None):
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002430 client = cls.connection.Client(address)
Benjamin Petersone711caf2008-06-11 16:44:04 +00002431 client.send(msg.upper())
2432 client.close()
2433
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002434 address, msg = conn.recv()
2435 client = socket.socket()
2436 client.connect(address)
2437 client.sendall(msg.upper())
2438 client.close()
Benjamin Petersone711caf2008-06-11 16:44:04 +00002439
2440 conn.close()
2441
2442 def test_pickling(self):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002443 families = self.connection.families
2444
2445 lconn, lconn0 = self.Pipe()
2446 lp = self.Process(target=self._listener, args=(lconn0, families))
Jesus Cea94f964f2011-09-09 20:26:57 +02002447 lp.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00002448 lp.start()
2449 lconn0.close()
2450
2451 rconn, rconn0 = self.Pipe()
2452 rp = self.Process(target=self._remote, args=(rconn0,))
Jesus Cea94f964f2011-09-09 20:26:57 +02002453 rp.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00002454 rp.start()
2455 rconn0.close()
2456
2457 for fam in families:
2458 msg = ('This connection uses family %s' % fam).encode('ascii')
2459 address = lconn.recv()
2460 rconn.send((address, msg))
2461 new_conn = lconn.recv()
2462 self.assertEqual(new_conn.recv(), msg.upper())
2463
2464 rconn.send(None)
2465
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002466 msg = latin('This connection uses a normal socket')
2467 address = lconn.recv()
2468 rconn.send((address, msg))
2469 new_conn = lconn.recv()
Richard Oudkerk4460c342012-04-30 14:48:50 +01002470 buf = []
2471 while True:
2472 s = new_conn.recv(100)
2473 if not s:
2474 break
2475 buf.append(s)
2476 buf = b''.join(buf)
2477 self.assertEqual(buf, msg.upper())
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002478 new_conn.close()
Benjamin Petersone711caf2008-06-11 16:44:04 +00002479
2480 lconn.send(None)
2481
2482 rconn.close()
2483 lconn.close()
2484
2485 lp.join()
2486 rp.join()
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002487
2488 @classmethod
2489 def child_access(cls, conn):
2490 w = conn.recv()
2491 w.send('all is well')
2492 w.close()
2493
2494 r = conn.recv()
2495 msg = r.recv()
2496 conn.send(msg*2)
2497
2498 conn.close()
2499
2500 def test_access(self):
2501 # On Windows, if we do not specify a destination pid when
2502 # using DupHandle then we need to be careful to use the
2503 # correct access flags for DuplicateHandle(), or else
2504 # DupHandle.detach() will raise PermissionError. For example,
2505 # for a read only pipe handle we should use
2506 # access=FILE_GENERIC_READ. (Unfortunately
2507 # DUPLICATE_SAME_ACCESS does not work.)
2508 conn, child_conn = self.Pipe()
2509 p = self.Process(target=self.child_access, args=(child_conn,))
2510 p.daemon = True
2511 p.start()
2512 child_conn.close()
2513
2514 r, w = self.Pipe(duplex=False)
2515 conn.send(w)
2516 w.close()
2517 self.assertEqual(r.recv(), 'all is well')
2518 r.close()
2519
2520 r, w = self.Pipe(duplex=False)
2521 conn.send(r)
2522 r.close()
2523 w.send('foobar')
2524 w.close()
2525 self.assertEqual(conn.recv(), 'foobar'*2)
2526
Benjamin Petersone711caf2008-06-11 16:44:04 +00002527#
2528#
2529#
2530
2531class _TestHeap(BaseTestCase):
2532
2533 ALLOWED_TYPES = ('processes',)
2534
2535 def test_heap(self):
2536 iterations = 5000
2537 maxblocks = 50
2538 blocks = []
2539
2540 # create and destroy lots of blocks of different sizes
2541 for i in range(iterations):
2542 size = int(random.lognormvariate(0, 1) * 1000)
2543 b = multiprocessing.heap.BufferWrapper(size)
2544 blocks.append(b)
2545 if len(blocks) > maxblocks:
2546 i = random.randrange(maxblocks)
2547 del blocks[i]
2548
2549 # get the heap object
2550 heap = multiprocessing.heap.BufferWrapper._heap
2551
2552 # verify the state of the heap
2553 all = []
2554 occupied = 0
Charles-François Natali778db492011-07-02 14:35:49 +02002555 heap._lock.acquire()
2556 self.addCleanup(heap._lock.release)
Benjamin Petersone711caf2008-06-11 16:44:04 +00002557 for L in list(heap._len_to_seq.values()):
2558 for arena, start, stop in L:
2559 all.append((heap._arenas.index(arena), start, stop,
2560 stop-start, 'free'))
2561 for arena, start, stop in heap._allocated_blocks:
2562 all.append((heap._arenas.index(arena), start, stop,
2563 stop-start, 'occupied'))
2564 occupied += (stop-start)
2565
2566 all.sort()
2567
2568 for i in range(len(all)-1):
2569 (arena, start, stop) = all[i][:3]
2570 (narena, nstart, nstop) = all[i+1][:3]
2571 self.assertTrue((arena != narena and nstart == 0) or
2572 (stop == nstart))
2573
Charles-François Natali778db492011-07-02 14:35:49 +02002574 def test_free_from_gc(self):
2575 # Check that freeing of blocks by the garbage collector doesn't deadlock
2576 # (issue #12352).
2577 # Make sure the GC is enabled, and set lower collection thresholds to
2578 # make collections more frequent (and increase the probability of
2579 # deadlock).
2580 if not gc.isenabled():
2581 gc.enable()
2582 self.addCleanup(gc.disable)
2583 thresholds = gc.get_threshold()
2584 self.addCleanup(gc.set_threshold, *thresholds)
2585 gc.set_threshold(10)
2586
2587 # perform numerous block allocations, with cyclic references to make
2588 # sure objects are collected asynchronously by the gc
2589 for i in range(5000):
2590 a = multiprocessing.heap.BufferWrapper(1)
2591 b = multiprocessing.heap.BufferWrapper(1)
2592 # circular references
2593 a.buddy = b
2594 b.buddy = a
2595
Benjamin Petersone711caf2008-06-11 16:44:04 +00002596#
2597#
2598#
2599
Benjamin Petersone711caf2008-06-11 16:44:04 +00002600class _Foo(Structure):
2601 _fields_ = [
2602 ('x', c_int),
2603 ('y', c_double)
2604 ]
2605
2606class _TestSharedCTypes(BaseTestCase):
2607
2608 ALLOWED_TYPES = ('processes',)
2609
Antoine Pitrou7744e2a2010-11-22 16:26:21 +00002610 def setUp(self):
2611 if not HAS_SHAREDCTYPES:
2612 self.skipTest("requires multiprocessing.sharedctypes")
2613
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00002614 @classmethod
2615 def _double(cls, x, y, foo, arr, string):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002616 x.value *= 2
2617 y.value *= 2
2618 foo.x *= 2
2619 foo.y *= 2
2620 string.value *= 2
2621 for i in range(len(arr)):
2622 arr[i] *= 2
2623
2624 def test_sharedctypes(self, lock=False):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002625 x = Value('i', 7, lock=lock)
Georg Brandl89fad142010-03-14 10:23:39 +00002626 y = Value(c_double, 1.0/3.0, lock=lock)
Benjamin Petersone711caf2008-06-11 16:44:04 +00002627 foo = Value(_Foo, 3, 2, lock=lock)
Georg Brandl89fad142010-03-14 10:23:39 +00002628 arr = self.Array('d', list(range(10)), lock=lock)
2629 string = self.Array('c', 20, lock=lock)
Brian Curtinafa88b52010-10-07 01:12:19 +00002630 string.value = latin('hello')
Benjamin Petersone711caf2008-06-11 16:44:04 +00002631
2632 p = self.Process(target=self._double, args=(x, y, foo, arr, string))
Jesus Cea94f964f2011-09-09 20:26:57 +02002633 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00002634 p.start()
2635 p.join()
2636
2637 self.assertEqual(x.value, 14)
2638 self.assertAlmostEqual(y.value, 2.0/3.0)
2639 self.assertEqual(foo.x, 6)
2640 self.assertAlmostEqual(foo.y, 4.0)
2641 for i in range(10):
2642 self.assertAlmostEqual(arr[i], i*2)
2643 self.assertEqual(string.value, latin('hellohello'))
2644
2645 def test_synchronize(self):
2646 self.test_sharedctypes(lock=True)
2647
2648 def test_copy(self):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002649 foo = _Foo(2, 5.0)
Brian Curtinafa88b52010-10-07 01:12:19 +00002650 bar = copy(foo)
Benjamin Petersone711caf2008-06-11 16:44:04 +00002651 foo.x = 0
2652 foo.y = 0
2653 self.assertEqual(bar.x, 2)
2654 self.assertAlmostEqual(bar.y, 5.0)
2655
2656#
2657#
2658#
2659
2660class _TestFinalize(BaseTestCase):
2661
2662 ALLOWED_TYPES = ('processes',)
2663
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00002664 @classmethod
2665 def _test_finalize(cls, conn):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002666 class Foo(object):
2667 pass
2668
2669 a = Foo()
2670 util.Finalize(a, conn.send, args=('a',))
2671 del a # triggers callback for a
2672
2673 b = Foo()
2674 close_b = util.Finalize(b, conn.send, args=('b',))
2675 close_b() # triggers callback for b
2676 close_b() # does nothing because callback has already been called
2677 del b # does nothing because callback has already been called
2678
2679 c = Foo()
2680 util.Finalize(c, conn.send, args=('c',))
2681
2682 d10 = Foo()
2683 util.Finalize(d10, conn.send, args=('d10',), exitpriority=1)
2684
2685 d01 = Foo()
2686 util.Finalize(d01, conn.send, args=('d01',), exitpriority=0)
2687 d02 = Foo()
2688 util.Finalize(d02, conn.send, args=('d02',), exitpriority=0)
2689 d03 = Foo()
2690 util.Finalize(d03, conn.send, args=('d03',), exitpriority=0)
2691
2692 util.Finalize(None, conn.send, args=('e',), exitpriority=-10)
2693
2694 util.Finalize(None, conn.send, args=('STOP',), exitpriority=-100)
2695
Ezio Melotti13925002011-03-16 11:05:33 +02002696 # call multiprocessing's cleanup function then exit process without
Benjamin Petersone711caf2008-06-11 16:44:04 +00002697 # garbage collecting locals
2698 util._exit_function()
2699 conn.close()
2700 os._exit(0)
2701
2702 def test_finalize(self):
2703 conn, child_conn = self.Pipe()
2704
2705 p = self.Process(target=self._test_finalize, args=(child_conn,))
Jesus Cea94f964f2011-09-09 20:26:57 +02002706 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00002707 p.start()
2708 p.join()
2709
2710 result = [obj for obj in iter(conn.recv, 'STOP')]
2711 self.assertEqual(result, ['a', 'b', 'd10', 'd03', 'd02', 'd01', 'e'])
2712
2713#
2714# Test that from ... import * works for each module
2715#
2716
2717class _TestImportStar(BaseTestCase):
2718
2719 ALLOWED_TYPES = ('processes',)
2720
2721 def test_import(self):
Florent Xiclunafd1b0932010-03-28 00:25:02 +00002722 modules = [
Benjamin Petersone711caf2008-06-11 16:44:04 +00002723 'multiprocessing', 'multiprocessing.connection',
2724 'multiprocessing.heap', 'multiprocessing.managers',
2725 'multiprocessing.pool', 'multiprocessing.process',
Benjamin Petersone711caf2008-06-11 16:44:04 +00002726 'multiprocessing.synchronize', 'multiprocessing.util'
Florent Xiclunafd1b0932010-03-28 00:25:02 +00002727 ]
2728
Charles-François Natalibc8f0822011-09-20 20:36:51 +02002729 if HAS_REDUCTION:
2730 modules.append('multiprocessing.reduction')
2731
Florent Xiclunafd1b0932010-03-28 00:25:02 +00002732 if c_int is not None:
2733 # This module requires _ctypes
2734 modules.append('multiprocessing.sharedctypes')
Benjamin Petersone711caf2008-06-11 16:44:04 +00002735
2736 for name in modules:
2737 __import__(name)
2738 mod = sys.modules[name]
2739
2740 for attr in getattr(mod, '__all__', ()):
2741 self.assertTrue(
2742 hasattr(mod, attr),
2743 '%r does not have attribute %r' % (mod, attr)
2744 )
2745
2746#
2747# Quick test that logging works -- does not test logging output
2748#
2749
2750class _TestLogging(BaseTestCase):
2751
2752 ALLOWED_TYPES = ('processes',)
2753
2754 def test_enable_logging(self):
2755 logger = multiprocessing.get_logger()
2756 logger.setLevel(util.SUBWARNING)
2757 self.assertTrue(logger is not None)
2758 logger.debug('this will not be printed')
2759 logger.info('nor will this')
2760 logger.setLevel(LOG_LEVEL)
2761
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00002762 @classmethod
2763 def _test_level(cls, conn):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002764 logger = multiprocessing.get_logger()
2765 conn.send(logger.getEffectiveLevel())
2766
2767 def test_level(self):
2768 LEVEL1 = 32
2769 LEVEL2 = 37
2770
2771 logger = multiprocessing.get_logger()
2772 root_logger = logging.getLogger()
2773 root_level = root_logger.level
2774
2775 reader, writer = multiprocessing.Pipe(duplex=False)
2776
2777 logger.setLevel(LEVEL1)
Jesus Cea94f964f2011-09-09 20:26:57 +02002778 p = self.Process(target=self._test_level, args=(writer,))
2779 p.daemon = True
2780 p.start()
Benjamin Petersone711caf2008-06-11 16:44:04 +00002781 self.assertEqual(LEVEL1, reader.recv())
2782
2783 logger.setLevel(logging.NOTSET)
2784 root_logger.setLevel(LEVEL2)
Jesus Cea94f964f2011-09-09 20:26:57 +02002785 p = self.Process(target=self._test_level, args=(writer,))
2786 p.daemon = True
2787 p.start()
Benjamin Petersone711caf2008-06-11 16:44:04 +00002788 self.assertEqual(LEVEL2, reader.recv())
2789
2790 root_logger.setLevel(root_level)
2791 logger.setLevel(level=LOG_LEVEL)
2792
Jesse Nollerb9a49b72009-11-21 18:09:38 +00002793
Jesse Nollerf4ae35f2009-11-24 14:22:24 +00002794# class _TestLoggingProcessName(BaseTestCase):
2795#
2796# def handle(self, record):
2797# assert record.processName == multiprocessing.current_process().name
2798# self.__handled = True
2799#
2800# def test_logging(self):
2801# handler = logging.Handler()
2802# handler.handle = self.handle
2803# self.__handled = False
2804# # Bypass getLogger() and side-effects
2805# logger = logging.getLoggerClass()(
2806# 'multiprocessing.test.TestLoggingProcessName')
2807# logger.addHandler(handler)
2808# logger.propagate = False
2809#
2810# logger.warn('foo')
2811# assert self.__handled
Jesse Nollerb9a49b72009-11-21 18:09:38 +00002812
Benjamin Petersone711caf2008-06-11 16:44:04 +00002813#
Jesse Noller6214edd2009-01-19 16:23:53 +00002814# Test to verify handle verification, see issue 3321
2815#
2816
2817class TestInvalidHandle(unittest.TestCase):
2818
Florent Xiclunafd1b0932010-03-28 00:25:02 +00002819 @unittest.skipIf(WIN32, "skipped on Windows")
Jesse Noller6214edd2009-01-19 16:23:53 +00002820 def test_invalid_handles(self):
Antoine Pitrou87cf2202011-05-09 17:04:27 +02002821 conn = multiprocessing.connection.Connection(44977608)
2822 try:
2823 self.assertRaises((ValueError, IOError), conn.poll)
2824 finally:
2825 # Hack private attribute _handle to avoid printing an error
2826 # in conn.__del__
2827 conn._handle = None
2828 self.assertRaises((ValueError, IOError),
2829 multiprocessing.connection.Connection, -1)
Florent Xiclunafd1b0932010-03-28 00:25:02 +00002830
Jesse Noller6214edd2009-01-19 16:23:53 +00002831#
Benjamin Petersone711caf2008-06-11 16:44:04 +00002832# Functions used to create test cases from the base ones in this module
2833#
2834
2835def get_attributes(Source, names):
2836 d = {}
2837 for name in names:
2838 obj = getattr(Source, name)
2839 if type(obj) == type(get_attributes):
2840 obj = staticmethod(obj)
2841 d[name] = obj
2842 return d
2843
2844def create_test_cases(Mixin, type):
2845 result = {}
2846 glob = globals()
Florent Xiclunafd1b0932010-03-28 00:25:02 +00002847 Type = type.capitalize()
Benjamin Petersone711caf2008-06-11 16:44:04 +00002848
2849 for name in list(glob.keys()):
2850 if name.startswith('_Test'):
2851 base = glob[name]
2852 if type in base.ALLOWED_TYPES:
2853 newname = 'With' + Type + name[1:]
2854 class Temp(base, unittest.TestCase, Mixin):
2855 pass
2856 result[newname] = Temp
2857 Temp.__name__ = newname
2858 Temp.__module__ = Mixin.__module__
2859 return result
2860
2861#
2862# Create test cases
2863#
2864
2865class ProcessesMixin(object):
2866 TYPE = 'processes'
2867 Process = multiprocessing.Process
2868 locals().update(get_attributes(multiprocessing, (
2869 'Queue', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore',
Richard Oudkerk3730a172012-06-15 18:26:07 +01002870 'Condition', 'Event', 'Barrier', 'Value', 'Array', 'RawValue',
Benjamin Petersone711caf2008-06-11 16:44:04 +00002871 'RawArray', 'current_process', 'active_children', 'Pipe',
Richard Oudkerke41682b2012-06-06 19:04:57 +01002872 'connection', 'JoinableQueue', 'Pool'
Benjamin Petersone711caf2008-06-11 16:44:04 +00002873 )))
2874
2875testcases_processes = create_test_cases(ProcessesMixin, type='processes')
2876globals().update(testcases_processes)
2877
2878
2879class ManagerMixin(object):
2880 TYPE = 'manager'
2881 Process = multiprocessing.Process
2882 manager = object.__new__(multiprocessing.managers.SyncManager)
2883 locals().update(get_attributes(manager, (
2884 'Queue', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore',
Richard Oudkerk3730a172012-06-15 18:26:07 +01002885 'Condition', 'Event', 'Barrier', 'Value', 'Array', 'list', 'dict',
Richard Oudkerke41682b2012-06-06 19:04:57 +01002886 'Namespace', 'JoinableQueue', 'Pool'
Benjamin Petersone711caf2008-06-11 16:44:04 +00002887 )))
2888
2889testcases_manager = create_test_cases(ManagerMixin, type='manager')
2890globals().update(testcases_manager)
2891
2892
2893class ThreadsMixin(object):
2894 TYPE = 'threads'
2895 Process = multiprocessing.dummy.Process
2896 locals().update(get_attributes(multiprocessing.dummy, (
2897 'Queue', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore',
Richard Oudkerk3730a172012-06-15 18:26:07 +01002898 'Condition', 'Event', 'Barrier', 'Value', 'Array', 'current_process',
Benjamin Petersone711caf2008-06-11 16:44:04 +00002899 'active_children', 'Pipe', 'connection', 'dict', 'list',
Richard Oudkerke41682b2012-06-06 19:04:57 +01002900 'Namespace', 'JoinableQueue', 'Pool'
Benjamin Petersone711caf2008-06-11 16:44:04 +00002901 )))
2902
2903testcases_threads = create_test_cases(ThreadsMixin, type='threads')
2904globals().update(testcases_threads)
2905
Neal Norwitz5d6415e2008-08-25 01:53:32 +00002906class OtherTest(unittest.TestCase):
2907 # TODO: add more tests for deliver/answer challenge.
2908 def test_deliver_challenge_auth_failure(self):
2909 class _FakeConnection(object):
2910 def recv_bytes(self, size):
Neal Norwitzec105ad2008-08-25 03:05:54 +00002911 return b'something bogus'
Neal Norwitz5d6415e2008-08-25 01:53:32 +00002912 def send_bytes(self, data):
2913 pass
2914 self.assertRaises(multiprocessing.AuthenticationError,
2915 multiprocessing.connection.deliver_challenge,
2916 _FakeConnection(), b'abc')
2917
2918 def test_answer_challenge_auth_failure(self):
2919 class _FakeConnection(object):
2920 def __init__(self):
2921 self.count = 0
2922 def recv_bytes(self, size):
2923 self.count += 1
2924 if self.count == 1:
2925 return multiprocessing.connection.CHALLENGE
2926 elif self.count == 2:
Neal Norwitzec105ad2008-08-25 03:05:54 +00002927 return b'something bogus'
2928 return b''
Neal Norwitz5d6415e2008-08-25 01:53:32 +00002929 def send_bytes(self, data):
2930 pass
2931 self.assertRaises(multiprocessing.AuthenticationError,
2932 multiprocessing.connection.answer_challenge,
2933 _FakeConnection(), b'abc')
2934
Benjamin Petersonf47ed4a2009-04-11 20:45:40 +00002935#
2936# Test Manager.start()/Pool.__init__() initializer feature - see issue 5585
2937#
2938
2939def initializer(ns):
2940 ns.test += 1
2941
2942class TestInitializers(unittest.TestCase):
2943 def setUp(self):
2944 self.mgr = multiprocessing.Manager()
2945 self.ns = self.mgr.Namespace()
2946 self.ns.test = 0
2947
2948 def tearDown(self):
2949 self.mgr.shutdown()
Richard Oudkerka6becaa2012-05-03 18:29:02 +01002950 self.mgr.join()
Benjamin Petersonf47ed4a2009-04-11 20:45:40 +00002951
2952 def test_manager_initializer(self):
2953 m = multiprocessing.managers.SyncManager()
2954 self.assertRaises(TypeError, m.start, 1)
2955 m.start(initializer, (self.ns,))
2956 self.assertEqual(self.ns.test, 1)
2957 m.shutdown()
Richard Oudkerka6becaa2012-05-03 18:29:02 +01002958 m.join()
Benjamin Petersonf47ed4a2009-04-11 20:45:40 +00002959
2960 def test_pool_initializer(self):
2961 self.assertRaises(TypeError, multiprocessing.Pool, initializer=1)
2962 p = multiprocessing.Pool(1, initializer, (self.ns,))
2963 p.close()
2964 p.join()
2965 self.assertEqual(self.ns.test, 1)
2966
Alexandre Vassalottic57a84f2009-07-17 12:07:01 +00002967#
2968# Issue 5155, 5313, 5331: Test process in processes
2969# Verifies os.close(sys.stdin.fileno) vs. sys.stdin.close() behavior
2970#
2971
2972def _ThisSubProcess(q):
2973 try:
2974 item = q.get(block=False)
2975 except pyqueue.Empty:
2976 pass
2977
2978def _TestProcess(q):
2979 queue = multiprocessing.Queue()
2980 subProc = multiprocessing.Process(target=_ThisSubProcess, args=(queue,))
Jesus Cea94f964f2011-09-09 20:26:57 +02002981 subProc.daemon = True
Alexandre Vassalottic57a84f2009-07-17 12:07:01 +00002982 subProc.start()
2983 subProc.join()
2984
2985def _afunc(x):
2986 return x*x
2987
2988def pool_in_process():
2989 pool = multiprocessing.Pool(processes=4)
2990 x = pool.map(_afunc, [1, 2, 3, 4, 5, 6, 7])
Richard Oudkerk225cb8d2012-05-02 19:36:11 +01002991 pool.close()
2992 pool.join()
Alexandre Vassalottic57a84f2009-07-17 12:07:01 +00002993
2994class _file_like(object):
2995 def __init__(self, delegate):
2996 self._delegate = delegate
2997 self._pid = None
2998
2999 @property
3000 def cache(self):
3001 pid = os.getpid()
3002 # There are no race conditions since fork keeps only the running thread
3003 if pid != self._pid:
3004 self._pid = pid
3005 self._cache = []
3006 return self._cache
3007
3008 def write(self, data):
3009 self.cache.append(data)
3010
3011 def flush(self):
3012 self._delegate.write(''.join(self.cache))
3013 self._cache = []
3014
3015class TestStdinBadfiledescriptor(unittest.TestCase):
3016
3017 def test_queue_in_process(self):
3018 queue = multiprocessing.Queue()
3019 proc = multiprocessing.Process(target=_TestProcess, args=(queue,))
3020 proc.start()
3021 proc.join()
3022
3023 def test_pool_in_process(self):
3024 p = multiprocessing.Process(target=pool_in_process)
3025 p.start()
3026 p.join()
3027
3028 def test_flushing(self):
3029 sio = io.StringIO()
3030 flike = _file_like(sio)
3031 flike.write('foo')
3032 proc = multiprocessing.Process(target=lambda: flike.flush())
3033 flike.flush()
3034 assert sio.getvalue() == 'foo'
3035
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003036
3037class TestWait(unittest.TestCase):
3038
3039 @classmethod
3040 def _child_test_wait(cls, w, slow):
3041 for i in range(10):
3042 if slow:
3043 time.sleep(random.random()*0.1)
3044 w.send((i, os.getpid()))
3045 w.close()
3046
3047 def test_wait(self, slow=False):
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003048 from multiprocessing.connection import wait
3049 readers = []
3050 procs = []
3051 messages = []
3052
3053 for i in range(4):
Antoine Pitrou5bb9a8f2012-03-06 13:43:24 +01003054 r, w = multiprocessing.Pipe(duplex=False)
3055 p = multiprocessing.Process(target=self._child_test_wait, args=(w, slow))
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003056 p.daemon = True
3057 p.start()
3058 w.close()
3059 readers.append(r)
3060 procs.append(p)
Antoine Pitrou6c64cc12012-03-06 13:42:35 +01003061 self.addCleanup(p.join)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003062
3063 while readers:
3064 for r in wait(readers):
3065 try:
3066 msg = r.recv()
3067 except EOFError:
3068 readers.remove(r)
3069 r.close()
3070 else:
3071 messages.append(msg)
3072
3073 messages.sort()
3074 expected = sorted((i, p.pid) for i in range(10) for p in procs)
3075 self.assertEqual(messages, expected)
3076
3077 @classmethod
3078 def _child_test_wait_socket(cls, address, slow):
3079 s = socket.socket()
3080 s.connect(address)
3081 for i in range(10):
3082 if slow:
3083 time.sleep(random.random()*0.1)
3084 s.sendall(('%s\n' % i).encode('ascii'))
3085 s.close()
3086
3087 def test_wait_socket(self, slow=False):
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003088 from multiprocessing.connection import wait
3089 l = socket.socket()
3090 l.bind(('', 0))
3091 l.listen(4)
3092 addr = ('localhost', l.getsockname()[1])
3093 readers = []
3094 procs = []
3095 dic = {}
3096
3097 for i in range(4):
Antoine Pitrou5bb9a8f2012-03-06 13:43:24 +01003098 p = multiprocessing.Process(target=self._child_test_wait_socket,
3099 args=(addr, slow))
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003100 p.daemon = True
3101 p.start()
3102 procs.append(p)
Antoine Pitrou6c64cc12012-03-06 13:42:35 +01003103 self.addCleanup(p.join)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003104
3105 for i in range(4):
3106 r, _ = l.accept()
3107 readers.append(r)
3108 dic[r] = []
3109 l.close()
3110
3111 while readers:
3112 for r in wait(readers):
3113 msg = r.recv(32)
3114 if not msg:
3115 readers.remove(r)
3116 r.close()
3117 else:
3118 dic[r].append(msg)
3119
3120 expected = ''.join('%s\n' % i for i in range(10)).encode('ascii')
3121 for v in dic.values():
3122 self.assertEqual(b''.join(v), expected)
3123
3124 def test_wait_slow(self):
3125 self.test_wait(True)
3126
3127 def test_wait_socket_slow(self):
Richard Oudkerk104b3f42012-05-08 16:08:07 +01003128 self.test_wait_socket(True)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003129
3130 def test_wait_timeout(self):
3131 from multiprocessing.connection import wait
3132
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003133 expected = 5
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003134 a, b = multiprocessing.Pipe()
3135
3136 start = time.time()
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003137 res = wait([a, b], expected)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003138 delta = time.time() - start
3139
3140 self.assertEqual(res, [])
Richard Oudkerk6dbca362012-05-06 16:46:36 +01003141 self.assertLess(delta, expected * 2)
3142 self.assertGreater(delta, expected * 0.5)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003143
3144 b.send(None)
3145
3146 start = time.time()
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003147 res = wait([a, b], 20)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003148 delta = time.time() - start
3149
3150 self.assertEqual(res, [a])
Antoine Pitrou37749772012-03-09 18:40:15 +01003151 self.assertLess(delta, 0.4)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003152
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003153 @classmethod
3154 def signal_and_sleep(cls, sem, period):
3155 sem.release()
3156 time.sleep(period)
3157
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003158 def test_wait_integer(self):
3159 from multiprocessing.connection import wait
3160
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003161 expected = 3
3162 sem = multiprocessing.Semaphore(0)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003163 a, b = multiprocessing.Pipe()
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003164 p = multiprocessing.Process(target=self.signal_and_sleep,
3165 args=(sem, expected))
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003166
3167 p.start()
3168 self.assertIsInstance(p.sentinel, int)
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003169 self.assertTrue(sem.acquire(timeout=20))
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003170
3171 start = time.time()
3172 res = wait([a, p.sentinel, b], expected + 20)
3173 delta = time.time() - start
3174
3175 self.assertEqual(res, [p.sentinel])
Antoine Pitrou37749772012-03-09 18:40:15 +01003176 self.assertLess(delta, expected + 2)
3177 self.assertGreater(delta, expected - 2)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003178
3179 a.send(None)
3180
3181 start = time.time()
3182 res = wait([a, p.sentinel, b], 20)
3183 delta = time.time() - start
3184
3185 self.assertEqual(res, [p.sentinel, b])
Antoine Pitrou37749772012-03-09 18:40:15 +01003186 self.assertLess(delta, 0.4)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003187
3188 b.send(None)
3189
3190 start = time.time()
3191 res = wait([a, p.sentinel, b], 20)
3192 delta = time.time() - start
3193
3194 self.assertEqual(res, [a, p.sentinel, b])
Antoine Pitrou37749772012-03-09 18:40:15 +01003195 self.assertLess(delta, 0.4)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003196
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003197 p.terminate()
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003198 p.join()
3199
Richard Oudkerk59d54042012-05-10 16:11:12 +01003200 def test_neg_timeout(self):
3201 from multiprocessing.connection import wait
3202 a, b = multiprocessing.Pipe()
3203 t = time.time()
3204 res = wait([a], timeout=-1)
3205 t = time.time() - t
3206 self.assertEqual(res, [])
3207 self.assertLess(t, 1)
3208 a.close()
3209 b.close()
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003210
Antoine Pitrou709176f2012-04-01 17:19:09 +02003211#
3212# Issue 14151: Test invalid family on invalid environment
3213#
3214
3215class TestInvalidFamily(unittest.TestCase):
3216
3217 @unittest.skipIf(WIN32, "skipped on Windows")
3218 def test_invalid_family(self):
3219 with self.assertRaises(ValueError):
3220 multiprocessing.connection.Listener(r'\\.\test')
3221
Antoine Pitrou6d20cba2012-04-03 20:12:23 +02003222 @unittest.skipUnless(WIN32, "skipped on non-Windows platforms")
3223 def test_invalid_family_win32(self):
3224 with self.assertRaises(ValueError):
3225 multiprocessing.connection.Listener('/var/test.pipe')
Antoine Pitrou93bba8f2012-04-01 17:25:49 +02003226
Richard Oudkerk77c84f22012-05-18 14:28:02 +01003227#
3228# Issue 12098: check sys.flags of child matches that for parent
3229#
3230
3231class TestFlags(unittest.TestCase):
3232 @classmethod
3233 def run_in_grandchild(cls, conn):
3234 conn.send(tuple(sys.flags))
3235
3236 @classmethod
3237 def run_in_child(cls):
3238 import json
3239 r, w = multiprocessing.Pipe(duplex=False)
3240 p = multiprocessing.Process(target=cls.run_in_grandchild, args=(w,))
3241 p.start()
3242 grandchild_flags = r.recv()
3243 p.join()
3244 r.close()
3245 w.close()
3246 flags = (tuple(sys.flags), grandchild_flags)
3247 print(json.dumps(flags))
3248
3249 def test_flags(self):
3250 import json, subprocess
3251 # start child process using unusual flags
3252 prog = ('from test.test_multiprocessing import TestFlags; ' +
3253 'TestFlags.run_in_child()')
3254 data = subprocess.check_output(
3255 [sys.executable, '-E', '-S', '-O', '-c', prog])
3256 child_flags, grandchild_flags = json.loads(data.decode('ascii'))
3257 self.assertEqual(child_flags, grandchild_flags)
3258
Alexandre Vassalottic57a84f2009-07-17 12:07:01 +00003259testcases_other = [OtherTest, TestInvalidHandle, TestInitializers,
Richard Oudkerk77c84f22012-05-18 14:28:02 +01003260 TestStdinBadfiledescriptor, TestWait, TestInvalidFamily,
3261 TestFlags]
Neal Norwitz5d6415e2008-08-25 01:53:32 +00003262
Benjamin Petersone711caf2008-06-11 16:44:04 +00003263#
3264#
3265#
3266
3267def test_main(run=None):
Jesse Nollerd00df3c2008-06-18 14:22:48 +00003268 if sys.platform.startswith("linux"):
3269 try:
3270 lock = multiprocessing.RLock()
3271 except OSError:
Benjamin Petersone549ead2009-03-28 21:42:05 +00003272 raise unittest.SkipTest("OSError raises on RLock creation, see issue 3111!")
Benjamin Peterson3c0dd062008-06-17 22:43:48 +00003273
Charles-François Natali221ef672011-11-22 18:55:22 +01003274 check_enough_semaphores()
3275
Benjamin Petersone711caf2008-06-11 16:44:04 +00003276 if run is None:
3277 from test.support import run_unittest as run
3278
3279 util.get_temp_dir() # creates temp directory for use by all processes
3280
3281 multiprocessing.get_logger().setLevel(LOG_LEVEL)
3282
Benjamin Peterson41181742008-07-02 20:22:54 +00003283 ProcessesMixin.pool = multiprocessing.Pool(4)
3284 ThreadsMixin.pool = multiprocessing.dummy.Pool(4)
3285 ManagerMixin.manager.__init__()
3286 ManagerMixin.manager.start()
3287 ManagerMixin.pool = ManagerMixin.manager.Pool(4)
Benjamin Petersone711caf2008-06-11 16:44:04 +00003288
3289 testcases = (
Benjamin Peterson41181742008-07-02 20:22:54 +00003290 sorted(testcases_processes.values(), key=lambda tc:tc.__name__) +
3291 sorted(testcases_threads.values(), key=lambda tc:tc.__name__) +
Neal Norwitz5d6415e2008-08-25 01:53:32 +00003292 sorted(testcases_manager.values(), key=lambda tc:tc.__name__) +
3293 testcases_other
Benjamin Petersone711caf2008-06-11 16:44:04 +00003294 )
3295
3296 loadTestsFromTestCase = unittest.defaultTestLoader.loadTestsFromTestCase
3297 suite = unittest.TestSuite(loadTestsFromTestCase(tc) for tc in testcases)
Richard Oudkerk225cb8d2012-05-02 19:36:11 +01003298 try:
3299 run(suite)
3300 finally:
3301 ThreadsMixin.pool.terminate()
3302 ProcessesMixin.pool.terminate()
3303 ManagerMixin.pool.terminate()
3304 ManagerMixin.pool.join()
3305 ManagerMixin.manager.shutdown()
Richard Oudkerka6becaa2012-05-03 18:29:02 +01003306 ManagerMixin.manager.join()
Richard Oudkerk225cb8d2012-05-02 19:36:11 +01003307 ThreadsMixin.pool.join()
3308 ProcessesMixin.pool.join()
3309 del ProcessesMixin.pool, ThreadsMixin.pool, ManagerMixin.pool
Benjamin Petersone711caf2008-06-11 16:44:04 +00003310
3311def main():
3312 test_main(unittest.TextTestRunner(verbosity=2).run)
3313
3314if __name__ == '__main__':
3315 main()