blob: d79110abbacbe61f1ff45895b4b98252d401b4fc [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().
Richard Oudkerk3049f122012-06-15 20:08:29 +01001823
1824 # If some other test using ManagerMixin.manager fails, then the
1825 # raised exception may keep alive a frame which holds a reference
1826 # to a managed object. This will cause test_number_of_objects to
1827 # also fail.
Benjamin Petersone711caf2008-06-11 16:44:04 +00001828 ALLOWED_TYPES = ('manager',)
1829
1830 def test_number_of_objects(self):
1831 EXPECTED_NUMBER = 1 # the pool object is still alive
1832 multiprocessing.active_children() # discard dead process objs
1833 gc.collect() # do garbage collection
1834 refs = self.manager._number_of_objects()
Jesse Noller63b3a972009-01-21 02:15:48 +00001835 debug_info = self.manager._debug_info()
Benjamin Petersone711caf2008-06-11 16:44:04 +00001836 if refs != EXPECTED_NUMBER:
Georg Brandl3dbca812008-07-23 16:10:53 +00001837 print(self.manager._debug_info())
Jesse Noller63b3a972009-01-21 02:15:48 +00001838 print(debug_info)
Benjamin Petersone711caf2008-06-11 16:44:04 +00001839
1840 self.assertEqual(refs, EXPECTED_NUMBER)
1841
1842#
1843# Test of creating a customized manager class
1844#
1845
1846from multiprocessing.managers import BaseManager, BaseProxy, RemoteError
1847
1848class FooBar(object):
1849 def f(self):
1850 return 'f()'
1851 def g(self):
1852 raise ValueError
1853 def _h(self):
1854 return '_h()'
1855
1856def baz():
1857 for i in range(10):
1858 yield i*i
1859
1860class IteratorProxy(BaseProxy):
Florent Xiclunaaa171062010-08-14 15:56:42 +00001861 _exposed_ = ('__next__',)
Benjamin Petersone711caf2008-06-11 16:44:04 +00001862 def __iter__(self):
1863 return self
1864 def __next__(self):
Benjamin Petersone711caf2008-06-11 16:44:04 +00001865 return self._callmethod('__next__')
1866
1867class MyManager(BaseManager):
1868 pass
1869
1870MyManager.register('Foo', callable=FooBar)
1871MyManager.register('Bar', callable=FooBar, exposed=('f', '_h'))
1872MyManager.register('baz', callable=baz, proxytype=IteratorProxy)
1873
1874
1875class _TestMyManager(BaseTestCase):
1876
1877 ALLOWED_TYPES = ('manager',)
1878
1879 def test_mymanager(self):
1880 manager = MyManager()
1881 manager.start()
1882
1883 foo = manager.Foo()
1884 bar = manager.Bar()
1885 baz = manager.baz()
1886
1887 foo_methods = [name for name in ('f', 'g', '_h') if hasattr(foo, name)]
1888 bar_methods = [name for name in ('f', 'g', '_h') if hasattr(bar, name)]
1889
1890 self.assertEqual(foo_methods, ['f', 'g'])
1891 self.assertEqual(bar_methods, ['f', '_h'])
1892
1893 self.assertEqual(foo.f(), 'f()')
1894 self.assertRaises(ValueError, foo.g)
1895 self.assertEqual(foo._callmethod('f'), 'f()')
1896 self.assertRaises(RemoteError, foo._callmethod, '_h')
1897
1898 self.assertEqual(bar.f(), 'f()')
1899 self.assertEqual(bar._h(), '_h()')
1900 self.assertEqual(bar._callmethod('f'), 'f()')
1901 self.assertEqual(bar._callmethod('_h'), '_h()')
1902
1903 self.assertEqual(list(baz), [i*i for i in range(10)])
1904
1905 manager.shutdown()
1906
Richard Oudkerk73d9a292012-06-14 15:30:10 +01001907 # If the manager process exited cleanly then the exitcode
1908 # will be zero. Otherwise (after a short timeout)
1909 # terminate() is used, resulting in an exitcode of -SIGTERM.
1910 self.assertEqual(manager._process.exitcode, 0)
1911
Benjamin Petersone711caf2008-06-11 16:44:04 +00001912#
1913# Test of connecting to a remote server and using xmlrpclib for serialization
1914#
1915
1916_queue = pyqueue.Queue()
1917def get_queue():
1918 return _queue
1919
1920class QueueManager(BaseManager):
1921 '''manager class used by server process'''
1922QueueManager.register('get_queue', callable=get_queue)
1923
1924class QueueManager2(BaseManager):
1925 '''manager class which specifies the same interface as QueueManager'''
1926QueueManager2.register('get_queue')
1927
1928
1929SERIALIZER = 'xmlrpclib'
1930
1931class _TestRemoteManager(BaseTestCase):
1932
1933 ALLOWED_TYPES = ('manager',)
1934
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00001935 @classmethod
1936 def _putter(cls, address, authkey):
Benjamin Petersone711caf2008-06-11 16:44:04 +00001937 manager = QueueManager2(
1938 address=address, authkey=authkey, serializer=SERIALIZER
1939 )
1940 manager.connect()
1941 queue = manager.get_queue()
1942 queue.put(('hello world', None, True, 2.25))
1943
1944 def test_remote(self):
1945 authkey = os.urandom(32)
1946
1947 manager = QueueManager(
1948 address=('localhost', 0), authkey=authkey, serializer=SERIALIZER
1949 )
1950 manager.start()
1951
1952 p = self.Process(target=self._putter, args=(manager.address, authkey))
Jesus Cea94f964f2011-09-09 20:26:57 +02001953 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00001954 p.start()
1955
1956 manager2 = QueueManager2(
1957 address=manager.address, authkey=authkey, serializer=SERIALIZER
1958 )
1959 manager2.connect()
1960 queue = manager2.get_queue()
1961
1962 # Note that xmlrpclib will deserialize object as a list not a tuple
1963 self.assertEqual(queue.get(), ['hello world', None, True, 2.25])
1964
1965 # Because we are using xmlrpclib for serialization instead of
1966 # pickle this will cause a serialization error.
1967 self.assertRaises(Exception, queue.put, time.sleep)
1968
1969 # Make queue finalizer run before the server is stopped
1970 del queue
1971 manager.shutdown()
1972
Jesse Nollerc5d28a02009-03-30 16:37:36 +00001973class _TestManagerRestart(BaseTestCase):
1974
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00001975 @classmethod
1976 def _putter(cls, address, authkey):
Jesse Nollerc5d28a02009-03-30 16:37:36 +00001977 manager = QueueManager(
1978 address=address, authkey=authkey, serializer=SERIALIZER)
1979 manager.connect()
1980 queue = manager.get_queue()
1981 queue.put('hello world')
1982
1983 def test_rapid_restart(self):
1984 authkey = os.urandom(32)
1985 manager = QueueManager(
Antoine Pitrou043bad02010-04-30 23:20:15 +00001986 address=('localhost', 0), authkey=authkey, serializer=SERIALIZER)
Brian Curtin50be1ca2010-11-01 05:10:44 +00001987 srvr = manager.get_server()
1988 addr = srvr.address
1989 # Close the connection.Listener socket which gets opened as a part
1990 # of manager.get_server(). It's not needed for the test.
1991 srvr.listener.close()
Jesse Nollerc5d28a02009-03-30 16:37:36 +00001992 manager.start()
1993
1994 p = self.Process(target=self._putter, args=(manager.address, authkey))
Jesus Cea94f964f2011-09-09 20:26:57 +02001995 p.daemon = True
Jesse Nollerc5d28a02009-03-30 16:37:36 +00001996 p.start()
1997 queue = manager.get_queue()
1998 self.assertEqual(queue.get(), 'hello world')
Jesse Noller35d1f002009-03-30 22:59:27 +00001999 del queue
Jesse Nollerc5d28a02009-03-30 16:37:36 +00002000 manager.shutdown()
2001 manager = QueueManager(
Antoine Pitrou043bad02010-04-30 23:20:15 +00002002 address=addr, authkey=authkey, serializer=SERIALIZER)
Antoine Pitrouc824e9a2011-04-05 18:11:33 +02002003 try:
2004 manager.start()
2005 except IOError as e:
2006 if e.errno != errno.EADDRINUSE:
2007 raise
2008 # Retry after some time, in case the old socket was lingering
2009 # (sporadic failure on buildbots)
2010 time.sleep(1.0)
2011 manager = QueueManager(
2012 address=addr, authkey=authkey, serializer=SERIALIZER)
Jesse Noller35d1f002009-03-30 22:59:27 +00002013 manager.shutdown()
Jesse Nollerc5d28a02009-03-30 16:37:36 +00002014
Benjamin Petersone711caf2008-06-11 16:44:04 +00002015#
2016#
2017#
2018
2019SENTINEL = latin('')
2020
2021class _TestConnection(BaseTestCase):
2022
2023 ALLOWED_TYPES = ('processes', 'threads')
2024
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00002025 @classmethod
2026 def _echo(cls, conn):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002027 for msg in iter(conn.recv_bytes, SENTINEL):
2028 conn.send_bytes(msg)
2029 conn.close()
2030
2031 def test_connection(self):
2032 conn, child_conn = self.Pipe()
2033
2034 p = self.Process(target=self._echo, args=(child_conn,))
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +00002035 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00002036 p.start()
2037
2038 seq = [1, 2.25, None]
2039 msg = latin('hello world')
2040 longmsg = msg * 10
2041 arr = array.array('i', list(range(4)))
2042
2043 if self.TYPE == 'processes':
2044 self.assertEqual(type(conn.fileno()), int)
2045
2046 self.assertEqual(conn.send(seq), None)
2047 self.assertEqual(conn.recv(), seq)
2048
2049 self.assertEqual(conn.send_bytes(msg), None)
2050 self.assertEqual(conn.recv_bytes(), msg)
2051
2052 if self.TYPE == 'processes':
2053 buffer = array.array('i', [0]*10)
2054 expected = list(arr) + [0] * (10 - len(arr))
2055 self.assertEqual(conn.send_bytes(arr), None)
2056 self.assertEqual(conn.recv_bytes_into(buffer),
2057 len(arr) * buffer.itemsize)
2058 self.assertEqual(list(buffer), expected)
2059
2060 buffer = array.array('i', [0]*10)
2061 expected = [0] * 3 + list(arr) + [0] * (10 - 3 - len(arr))
2062 self.assertEqual(conn.send_bytes(arr), None)
2063 self.assertEqual(conn.recv_bytes_into(buffer, 3 * buffer.itemsize),
2064 len(arr) * buffer.itemsize)
2065 self.assertEqual(list(buffer), expected)
2066
2067 buffer = bytearray(latin(' ' * 40))
2068 self.assertEqual(conn.send_bytes(longmsg), None)
2069 try:
2070 res = conn.recv_bytes_into(buffer)
2071 except multiprocessing.BufferTooShort as e:
2072 self.assertEqual(e.args, (longmsg,))
2073 else:
2074 self.fail('expected BufferTooShort, got %s' % res)
2075
2076 poll = TimingWrapper(conn.poll)
2077
2078 self.assertEqual(poll(), False)
2079 self.assertTimingAlmostEqual(poll.elapsed, 0)
2080
Richard Oudkerk59d54042012-05-10 16:11:12 +01002081 self.assertEqual(poll(-1), False)
2082 self.assertTimingAlmostEqual(poll.elapsed, 0)
2083
Benjamin Petersone711caf2008-06-11 16:44:04 +00002084 self.assertEqual(poll(TIMEOUT1), False)
2085 self.assertTimingAlmostEqual(poll.elapsed, TIMEOUT1)
2086
2087 conn.send(None)
2088
2089 self.assertEqual(poll(TIMEOUT1), True)
2090 self.assertTimingAlmostEqual(poll.elapsed, 0)
2091
2092 self.assertEqual(conn.recv(), None)
2093
2094 really_big_msg = latin('X') * (1024 * 1024 * 16) # 16Mb
2095 conn.send_bytes(really_big_msg)
2096 self.assertEqual(conn.recv_bytes(), really_big_msg)
2097
2098 conn.send_bytes(SENTINEL) # tell child to quit
2099 child_conn.close()
2100
2101 if self.TYPE == 'processes':
2102 self.assertEqual(conn.readable, True)
2103 self.assertEqual(conn.writable, True)
2104 self.assertRaises(EOFError, conn.recv)
2105 self.assertRaises(EOFError, conn.recv_bytes)
2106
2107 p.join()
2108
2109 def test_duplex_false(self):
2110 reader, writer = self.Pipe(duplex=False)
2111 self.assertEqual(writer.send(1), None)
2112 self.assertEqual(reader.recv(), 1)
2113 if self.TYPE == 'processes':
2114 self.assertEqual(reader.readable, True)
2115 self.assertEqual(reader.writable, False)
2116 self.assertEqual(writer.readable, False)
2117 self.assertEqual(writer.writable, True)
2118 self.assertRaises(IOError, reader.send, 2)
2119 self.assertRaises(IOError, writer.recv)
2120 self.assertRaises(IOError, writer.poll)
2121
2122 def test_spawn_close(self):
2123 # We test that a pipe connection can be closed by parent
2124 # process immediately after child is spawned. On Windows this
2125 # would have sometimes failed on old versions because
2126 # child_conn would be closed before the child got a chance to
2127 # duplicate it.
2128 conn, child_conn = self.Pipe()
2129
2130 p = self.Process(target=self._echo, args=(child_conn,))
Jesus Cea94f964f2011-09-09 20:26:57 +02002131 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00002132 p.start()
2133 child_conn.close() # this might complete before child initializes
2134
2135 msg = latin('hello')
2136 conn.send_bytes(msg)
2137 self.assertEqual(conn.recv_bytes(), msg)
2138
2139 conn.send_bytes(SENTINEL)
2140 conn.close()
2141 p.join()
2142
2143 def test_sendbytes(self):
2144 if self.TYPE != 'processes':
2145 return
2146
2147 msg = latin('abcdefghijklmnopqrstuvwxyz')
2148 a, b = self.Pipe()
2149
2150 a.send_bytes(msg)
2151 self.assertEqual(b.recv_bytes(), msg)
2152
2153 a.send_bytes(msg, 5)
2154 self.assertEqual(b.recv_bytes(), msg[5:])
2155
2156 a.send_bytes(msg, 7, 8)
2157 self.assertEqual(b.recv_bytes(), msg[7:7+8])
2158
2159 a.send_bytes(msg, 26)
2160 self.assertEqual(b.recv_bytes(), latin(''))
2161
2162 a.send_bytes(msg, 26, 0)
2163 self.assertEqual(b.recv_bytes(), latin(''))
2164
2165 self.assertRaises(ValueError, a.send_bytes, msg, 27)
2166
2167 self.assertRaises(ValueError, a.send_bytes, msg, 22, 5)
2168
2169 self.assertRaises(ValueError, a.send_bytes, msg, 26, 1)
2170
2171 self.assertRaises(ValueError, a.send_bytes, msg, -1)
2172
2173 self.assertRaises(ValueError, a.send_bytes, msg, 4, -1)
2174
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002175 @classmethod
2176 def _is_fd_assigned(cls, fd):
2177 try:
2178 os.fstat(fd)
2179 except OSError as e:
2180 if e.errno == errno.EBADF:
2181 return False
2182 raise
2183 else:
2184 return True
2185
2186 @classmethod
2187 def _writefd(cls, conn, data, create_dummy_fds=False):
2188 if create_dummy_fds:
2189 for i in range(0, 256):
2190 if not cls._is_fd_assigned(i):
2191 os.dup2(conn.fileno(), i)
2192 fd = reduction.recv_handle(conn)
2193 if msvcrt:
2194 fd = msvcrt.open_osfhandle(fd, os.O_WRONLY)
2195 os.write(fd, data)
2196 os.close(fd)
2197
Charles-François Natalibc8f0822011-09-20 20:36:51 +02002198 @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002199 def test_fd_transfer(self):
2200 if self.TYPE != 'processes':
2201 self.skipTest("only makes sense with processes")
2202 conn, child_conn = self.Pipe(duplex=True)
2203
2204 p = self.Process(target=self._writefd, args=(child_conn, b"foo"))
Jesus Cea94f964f2011-09-09 20:26:57 +02002205 p.daemon = True
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002206 p.start()
Victor Stinnerd0b10a62011-09-21 01:10:29 +02002207 self.addCleanup(test.support.unlink, test.support.TESTFN)
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002208 with open(test.support.TESTFN, "wb") as f:
2209 fd = f.fileno()
2210 if msvcrt:
2211 fd = msvcrt.get_osfhandle(fd)
2212 reduction.send_handle(conn, fd, p.pid)
2213 p.join()
2214 with open(test.support.TESTFN, "rb") as f:
2215 self.assertEqual(f.read(), b"foo")
2216
Charles-François Natalibc8f0822011-09-20 20:36:51 +02002217 @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002218 @unittest.skipIf(sys.platform == "win32",
2219 "test semantics don't make sense on Windows")
2220 @unittest.skipIf(MAXFD <= 256,
2221 "largest assignable fd number is too small")
2222 @unittest.skipUnless(hasattr(os, "dup2"),
2223 "test needs os.dup2()")
2224 def test_large_fd_transfer(self):
2225 # With fd > 256 (issue #11657)
2226 if self.TYPE != 'processes':
2227 self.skipTest("only makes sense with processes")
2228 conn, child_conn = self.Pipe(duplex=True)
2229
2230 p = self.Process(target=self._writefd, args=(child_conn, b"bar", True))
Jesus Cea94f964f2011-09-09 20:26:57 +02002231 p.daemon = True
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002232 p.start()
Victor Stinnerd0b10a62011-09-21 01:10:29 +02002233 self.addCleanup(test.support.unlink, test.support.TESTFN)
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002234 with open(test.support.TESTFN, "wb") as f:
2235 fd = f.fileno()
2236 for newfd in range(256, MAXFD):
2237 if not self._is_fd_assigned(newfd):
2238 break
2239 else:
2240 self.fail("could not find an unassigned large file descriptor")
2241 os.dup2(fd, newfd)
2242 try:
2243 reduction.send_handle(conn, newfd, p.pid)
2244 finally:
2245 os.close(newfd)
2246 p.join()
2247 with open(test.support.TESTFN, "rb") as f:
2248 self.assertEqual(f.read(), b"bar")
2249
Jesus Cea4507e642011-09-21 03:53:25 +02002250 @classmethod
2251 def _send_data_without_fd(self, conn):
2252 os.write(conn.fileno(), b"\0")
2253
Charles-François Natalie51c8da2011-09-21 18:48:21 +02002254 @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
Jesus Cea4507e642011-09-21 03:53:25 +02002255 @unittest.skipIf(sys.platform == "win32", "doesn't make sense on Windows")
2256 def test_missing_fd_transfer(self):
2257 # Check that exception is raised when received data is not
2258 # accompanied by a file descriptor in ancillary data.
2259 if self.TYPE != 'processes':
2260 self.skipTest("only makes sense with processes")
2261 conn, child_conn = self.Pipe(duplex=True)
2262
2263 p = self.Process(target=self._send_data_without_fd, args=(child_conn,))
2264 p.daemon = True
2265 p.start()
2266 self.assertRaises(RuntimeError, reduction.recv_handle, conn)
2267 p.join()
Antoine Pitroubcb39d42011-08-23 19:46:22 +02002268
Charles-François Natalied4a8fc2012-02-08 21:15:58 +01002269class _TestListener(BaseTestCase):
2270
2271 ALLOWED_TYPES = ('processes')
2272
2273 def test_multiple_bind(self):
2274 for family in self.connection.families:
2275 l = self.connection.Listener(family=family)
2276 self.addCleanup(l.close)
2277 self.assertRaises(OSError, self.connection.Listener,
2278 l.address, family)
2279
Benjamin Petersone711caf2008-06-11 16:44:04 +00002280class _TestListenerClient(BaseTestCase):
2281
2282 ALLOWED_TYPES = ('processes', 'threads')
2283
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00002284 @classmethod
2285 def _test(cls, address):
2286 conn = cls.connection.Client(address)
Benjamin Petersone711caf2008-06-11 16:44:04 +00002287 conn.send('hello')
2288 conn.close()
2289
2290 def test_listener_client(self):
2291 for family in self.connection.families:
2292 l = self.connection.Listener(family=family)
2293 p = self.Process(target=self._test, args=(l.address,))
Benjamin Peterson58ea9fe2008-08-19 19:17:39 +00002294 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00002295 p.start()
2296 conn = l.accept()
2297 self.assertEqual(conn.recv(), 'hello')
2298 p.join()
2299 l.close()
Charles-François Natalied4a8fc2012-02-08 21:15:58 +01002300
Richard Oudkerkfdb8dcf2012-05-05 19:45:37 +01002301 def test_issue14725(self):
2302 l = self.connection.Listener()
2303 p = self.Process(target=self._test, args=(l.address,))
2304 p.daemon = True
2305 p.start()
2306 time.sleep(1)
2307 # On Windows the client process should by now have connected,
2308 # written data and closed the pipe handle by now. This causes
2309 # ConnectNamdedPipe() to fail with ERROR_NO_DATA. See Issue
2310 # 14725.
2311 conn = l.accept()
2312 self.assertEqual(conn.recv(), 'hello')
2313 conn.close()
2314 p.join()
2315 l.close()
2316
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01002317class _TestPoll(unittest.TestCase):
2318
2319 ALLOWED_TYPES = ('processes', 'threads')
2320
2321 def test_empty_string(self):
2322 a, b = self.Pipe()
2323 self.assertEqual(a.poll(), False)
2324 b.send_bytes(b'')
2325 self.assertEqual(a.poll(), True)
2326 self.assertEqual(a.poll(), True)
2327
2328 @classmethod
2329 def _child_strings(cls, conn, strings):
2330 for s in strings:
2331 time.sleep(0.1)
2332 conn.send_bytes(s)
2333 conn.close()
2334
2335 def test_strings(self):
2336 strings = (b'hello', b'', b'a', b'b', b'', b'bye', b'', b'lop')
2337 a, b = self.Pipe()
2338 p = self.Process(target=self._child_strings, args=(b, strings))
2339 p.start()
2340
2341 for s in strings:
2342 for i in range(200):
2343 if a.poll(0.01):
2344 break
2345 x = a.recv_bytes()
2346 self.assertEqual(s, x)
2347
2348 p.join()
2349
2350 @classmethod
2351 def _child_boundaries(cls, r):
2352 # Polling may "pull" a message in to the child process, but we
2353 # don't want it to pull only part of a message, as that would
2354 # corrupt the pipe for any other processes which might later
2355 # read from it.
2356 r.poll(5)
2357
2358 def test_boundaries(self):
2359 r, w = self.Pipe(False)
2360 p = self.Process(target=self._child_boundaries, args=(r,))
2361 p.start()
2362 time.sleep(2)
2363 L = [b"first", b"second"]
2364 for obj in L:
2365 w.send_bytes(obj)
2366 w.close()
2367 p.join()
2368 self.assertIn(r.recv_bytes(), L)
2369
2370 @classmethod
2371 def _child_dont_merge(cls, b):
2372 b.send_bytes(b'a')
2373 b.send_bytes(b'b')
2374 b.send_bytes(b'cd')
2375
2376 def test_dont_merge(self):
2377 a, b = self.Pipe()
2378 self.assertEqual(a.poll(0.0), False)
2379 self.assertEqual(a.poll(0.1), False)
2380
2381 p = self.Process(target=self._child_dont_merge, args=(b,))
2382 p.start()
2383
2384 self.assertEqual(a.recv_bytes(), b'a')
2385 self.assertEqual(a.poll(1.0), True)
2386 self.assertEqual(a.poll(1.0), True)
2387 self.assertEqual(a.recv_bytes(), b'b')
2388 self.assertEqual(a.poll(1.0), True)
2389 self.assertEqual(a.poll(1.0), True)
2390 self.assertEqual(a.poll(0.0), True)
2391 self.assertEqual(a.recv_bytes(), b'cd')
2392
2393 p.join()
2394
Benjamin Petersone711caf2008-06-11 16:44:04 +00002395#
2396# Test of sending connection and socket objects between processes
2397#
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002398
Richard Oudkerk24524192012-04-30 14:48:51 +01002399# Intermittent fails on Mac OS X -- see Issue14669 and Issue12958
2400@unittest.skipIf(sys.platform == "darwin", "fd passing unreliable on Mac OS X")
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002401@unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
Benjamin Petersone711caf2008-06-11 16:44:04 +00002402class _TestPicklingConnections(BaseTestCase):
2403
2404 ALLOWED_TYPES = ('processes',)
2405
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002406 @classmethod
Antoine Pitrou92ff4e12012-04-27 23:51:03 +02002407 def tearDownClass(cls):
2408 from multiprocessing.reduction import resource_sharer
2409 resource_sharer.stop(timeout=5)
2410
2411 @classmethod
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002412 def _listener(cls, conn, families):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002413 for fam in families:
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002414 l = cls.connection.Listener(family=fam)
Benjamin Petersone711caf2008-06-11 16:44:04 +00002415 conn.send(l.address)
2416 new_conn = l.accept()
2417 conn.send(new_conn)
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002418 new_conn.close()
2419 l.close()
Benjamin Petersone711caf2008-06-11 16:44:04 +00002420
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002421 l = socket.socket()
2422 l.bind(('localhost', 0))
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002423 l.listen(1)
Richard Oudkerk5d73c172012-05-08 22:24:47 +01002424 conn.send(l.getsockname())
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002425 new_conn, addr = l.accept()
2426 conn.send(new_conn)
2427 new_conn.close()
2428 l.close()
Benjamin Petersone711caf2008-06-11 16:44:04 +00002429
2430 conn.recv()
2431
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002432 @classmethod
2433 def _remote(cls, conn):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002434 for (address, msg) in iter(conn.recv, None):
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002435 client = cls.connection.Client(address)
Benjamin Petersone711caf2008-06-11 16:44:04 +00002436 client.send(msg.upper())
2437 client.close()
2438
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002439 address, msg = conn.recv()
2440 client = socket.socket()
2441 client.connect(address)
2442 client.sendall(msg.upper())
2443 client.close()
Benjamin Petersone711caf2008-06-11 16:44:04 +00002444
2445 conn.close()
2446
2447 def test_pickling(self):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002448 families = self.connection.families
2449
2450 lconn, lconn0 = self.Pipe()
2451 lp = self.Process(target=self._listener, args=(lconn0, families))
Jesus Cea94f964f2011-09-09 20:26:57 +02002452 lp.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00002453 lp.start()
2454 lconn0.close()
2455
2456 rconn, rconn0 = self.Pipe()
2457 rp = self.Process(target=self._remote, args=(rconn0,))
Jesus Cea94f964f2011-09-09 20:26:57 +02002458 rp.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00002459 rp.start()
2460 rconn0.close()
2461
2462 for fam in families:
2463 msg = ('This connection uses family %s' % fam).encode('ascii')
2464 address = lconn.recv()
2465 rconn.send((address, msg))
2466 new_conn = lconn.recv()
2467 self.assertEqual(new_conn.recv(), msg.upper())
2468
2469 rconn.send(None)
2470
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002471 msg = latin('This connection uses a normal socket')
2472 address = lconn.recv()
2473 rconn.send((address, msg))
2474 new_conn = lconn.recv()
Richard Oudkerk4460c342012-04-30 14:48:50 +01002475 buf = []
2476 while True:
2477 s = new_conn.recv(100)
2478 if not s:
2479 break
2480 buf.append(s)
2481 buf = b''.join(buf)
2482 self.assertEqual(buf, msg.upper())
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002483 new_conn.close()
Benjamin Petersone711caf2008-06-11 16:44:04 +00002484
2485 lconn.send(None)
2486
2487 rconn.close()
2488 lconn.close()
2489
2490 lp.join()
2491 rp.join()
Antoine Pitrou5438ed12012-04-24 22:56:57 +02002492
2493 @classmethod
2494 def child_access(cls, conn):
2495 w = conn.recv()
2496 w.send('all is well')
2497 w.close()
2498
2499 r = conn.recv()
2500 msg = r.recv()
2501 conn.send(msg*2)
2502
2503 conn.close()
2504
2505 def test_access(self):
2506 # On Windows, if we do not specify a destination pid when
2507 # using DupHandle then we need to be careful to use the
2508 # correct access flags for DuplicateHandle(), or else
2509 # DupHandle.detach() will raise PermissionError. For example,
2510 # for a read only pipe handle we should use
2511 # access=FILE_GENERIC_READ. (Unfortunately
2512 # DUPLICATE_SAME_ACCESS does not work.)
2513 conn, child_conn = self.Pipe()
2514 p = self.Process(target=self.child_access, args=(child_conn,))
2515 p.daemon = True
2516 p.start()
2517 child_conn.close()
2518
2519 r, w = self.Pipe(duplex=False)
2520 conn.send(w)
2521 w.close()
2522 self.assertEqual(r.recv(), 'all is well')
2523 r.close()
2524
2525 r, w = self.Pipe(duplex=False)
2526 conn.send(r)
2527 r.close()
2528 w.send('foobar')
2529 w.close()
2530 self.assertEqual(conn.recv(), 'foobar'*2)
2531
Benjamin Petersone711caf2008-06-11 16:44:04 +00002532#
2533#
2534#
2535
2536class _TestHeap(BaseTestCase):
2537
2538 ALLOWED_TYPES = ('processes',)
2539
2540 def test_heap(self):
2541 iterations = 5000
2542 maxblocks = 50
2543 blocks = []
2544
2545 # create and destroy lots of blocks of different sizes
2546 for i in range(iterations):
2547 size = int(random.lognormvariate(0, 1) * 1000)
2548 b = multiprocessing.heap.BufferWrapper(size)
2549 blocks.append(b)
2550 if len(blocks) > maxblocks:
2551 i = random.randrange(maxblocks)
2552 del blocks[i]
2553
2554 # get the heap object
2555 heap = multiprocessing.heap.BufferWrapper._heap
2556
2557 # verify the state of the heap
2558 all = []
2559 occupied = 0
Charles-François Natali778db492011-07-02 14:35:49 +02002560 heap._lock.acquire()
2561 self.addCleanup(heap._lock.release)
Benjamin Petersone711caf2008-06-11 16:44:04 +00002562 for L in list(heap._len_to_seq.values()):
2563 for arena, start, stop in L:
2564 all.append((heap._arenas.index(arena), start, stop,
2565 stop-start, 'free'))
2566 for arena, start, stop in heap._allocated_blocks:
2567 all.append((heap._arenas.index(arena), start, stop,
2568 stop-start, 'occupied'))
2569 occupied += (stop-start)
2570
2571 all.sort()
2572
2573 for i in range(len(all)-1):
2574 (arena, start, stop) = all[i][:3]
2575 (narena, nstart, nstop) = all[i+1][:3]
2576 self.assertTrue((arena != narena and nstart == 0) or
2577 (stop == nstart))
2578
Charles-François Natali778db492011-07-02 14:35:49 +02002579 def test_free_from_gc(self):
2580 # Check that freeing of blocks by the garbage collector doesn't deadlock
2581 # (issue #12352).
2582 # Make sure the GC is enabled, and set lower collection thresholds to
2583 # make collections more frequent (and increase the probability of
2584 # deadlock).
2585 if not gc.isenabled():
2586 gc.enable()
2587 self.addCleanup(gc.disable)
2588 thresholds = gc.get_threshold()
2589 self.addCleanup(gc.set_threshold, *thresholds)
2590 gc.set_threshold(10)
2591
2592 # perform numerous block allocations, with cyclic references to make
2593 # sure objects are collected asynchronously by the gc
2594 for i in range(5000):
2595 a = multiprocessing.heap.BufferWrapper(1)
2596 b = multiprocessing.heap.BufferWrapper(1)
2597 # circular references
2598 a.buddy = b
2599 b.buddy = a
2600
Benjamin Petersone711caf2008-06-11 16:44:04 +00002601#
2602#
2603#
2604
Benjamin Petersone711caf2008-06-11 16:44:04 +00002605class _Foo(Structure):
2606 _fields_ = [
2607 ('x', c_int),
2608 ('y', c_double)
2609 ]
2610
2611class _TestSharedCTypes(BaseTestCase):
2612
2613 ALLOWED_TYPES = ('processes',)
2614
Antoine Pitrou7744e2a2010-11-22 16:26:21 +00002615 def setUp(self):
2616 if not HAS_SHAREDCTYPES:
2617 self.skipTest("requires multiprocessing.sharedctypes")
2618
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00002619 @classmethod
2620 def _double(cls, x, y, foo, arr, string):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002621 x.value *= 2
2622 y.value *= 2
2623 foo.x *= 2
2624 foo.y *= 2
2625 string.value *= 2
2626 for i in range(len(arr)):
2627 arr[i] *= 2
2628
2629 def test_sharedctypes(self, lock=False):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002630 x = Value('i', 7, lock=lock)
Georg Brandl89fad142010-03-14 10:23:39 +00002631 y = Value(c_double, 1.0/3.0, lock=lock)
Benjamin Petersone711caf2008-06-11 16:44:04 +00002632 foo = Value(_Foo, 3, 2, lock=lock)
Georg Brandl89fad142010-03-14 10:23:39 +00002633 arr = self.Array('d', list(range(10)), lock=lock)
2634 string = self.Array('c', 20, lock=lock)
Brian Curtinafa88b52010-10-07 01:12:19 +00002635 string.value = latin('hello')
Benjamin Petersone711caf2008-06-11 16:44:04 +00002636
2637 p = self.Process(target=self._double, args=(x, y, foo, arr, string))
Jesus Cea94f964f2011-09-09 20:26:57 +02002638 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00002639 p.start()
2640 p.join()
2641
2642 self.assertEqual(x.value, 14)
2643 self.assertAlmostEqual(y.value, 2.0/3.0)
2644 self.assertEqual(foo.x, 6)
2645 self.assertAlmostEqual(foo.y, 4.0)
2646 for i in range(10):
2647 self.assertAlmostEqual(arr[i], i*2)
2648 self.assertEqual(string.value, latin('hellohello'))
2649
2650 def test_synchronize(self):
2651 self.test_sharedctypes(lock=True)
2652
2653 def test_copy(self):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002654 foo = _Foo(2, 5.0)
Brian Curtinafa88b52010-10-07 01:12:19 +00002655 bar = copy(foo)
Benjamin Petersone711caf2008-06-11 16:44:04 +00002656 foo.x = 0
2657 foo.y = 0
2658 self.assertEqual(bar.x, 2)
2659 self.assertAlmostEqual(bar.y, 5.0)
2660
2661#
2662#
2663#
2664
2665class _TestFinalize(BaseTestCase):
2666
2667 ALLOWED_TYPES = ('processes',)
2668
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00002669 @classmethod
2670 def _test_finalize(cls, conn):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002671 class Foo(object):
2672 pass
2673
2674 a = Foo()
2675 util.Finalize(a, conn.send, args=('a',))
2676 del a # triggers callback for a
2677
2678 b = Foo()
2679 close_b = util.Finalize(b, conn.send, args=('b',))
2680 close_b() # triggers callback for b
2681 close_b() # does nothing because callback has already been called
2682 del b # does nothing because callback has already been called
2683
2684 c = Foo()
2685 util.Finalize(c, conn.send, args=('c',))
2686
2687 d10 = Foo()
2688 util.Finalize(d10, conn.send, args=('d10',), exitpriority=1)
2689
2690 d01 = Foo()
2691 util.Finalize(d01, conn.send, args=('d01',), exitpriority=0)
2692 d02 = Foo()
2693 util.Finalize(d02, conn.send, args=('d02',), exitpriority=0)
2694 d03 = Foo()
2695 util.Finalize(d03, conn.send, args=('d03',), exitpriority=0)
2696
2697 util.Finalize(None, conn.send, args=('e',), exitpriority=-10)
2698
2699 util.Finalize(None, conn.send, args=('STOP',), exitpriority=-100)
2700
Ezio Melotti13925002011-03-16 11:05:33 +02002701 # call multiprocessing's cleanup function then exit process without
Benjamin Petersone711caf2008-06-11 16:44:04 +00002702 # garbage collecting locals
2703 util._exit_function()
2704 conn.close()
2705 os._exit(0)
2706
2707 def test_finalize(self):
2708 conn, child_conn = self.Pipe()
2709
2710 p = self.Process(target=self._test_finalize, args=(child_conn,))
Jesus Cea94f964f2011-09-09 20:26:57 +02002711 p.daemon = True
Benjamin Petersone711caf2008-06-11 16:44:04 +00002712 p.start()
2713 p.join()
2714
2715 result = [obj for obj in iter(conn.recv, 'STOP')]
2716 self.assertEqual(result, ['a', 'b', 'd10', 'd03', 'd02', 'd01', 'e'])
2717
2718#
2719# Test that from ... import * works for each module
2720#
2721
2722class _TestImportStar(BaseTestCase):
2723
2724 ALLOWED_TYPES = ('processes',)
2725
2726 def test_import(self):
Florent Xiclunafd1b0932010-03-28 00:25:02 +00002727 modules = [
Benjamin Petersone711caf2008-06-11 16:44:04 +00002728 'multiprocessing', 'multiprocessing.connection',
2729 'multiprocessing.heap', 'multiprocessing.managers',
2730 'multiprocessing.pool', 'multiprocessing.process',
Benjamin Petersone711caf2008-06-11 16:44:04 +00002731 'multiprocessing.synchronize', 'multiprocessing.util'
Florent Xiclunafd1b0932010-03-28 00:25:02 +00002732 ]
2733
Charles-François Natalibc8f0822011-09-20 20:36:51 +02002734 if HAS_REDUCTION:
2735 modules.append('multiprocessing.reduction')
2736
Florent Xiclunafd1b0932010-03-28 00:25:02 +00002737 if c_int is not None:
2738 # This module requires _ctypes
2739 modules.append('multiprocessing.sharedctypes')
Benjamin Petersone711caf2008-06-11 16:44:04 +00002740
2741 for name in modules:
2742 __import__(name)
2743 mod = sys.modules[name]
2744
2745 for attr in getattr(mod, '__all__', ()):
2746 self.assertTrue(
2747 hasattr(mod, attr),
2748 '%r does not have attribute %r' % (mod, attr)
2749 )
2750
2751#
2752# Quick test that logging works -- does not test logging output
2753#
2754
2755class _TestLogging(BaseTestCase):
2756
2757 ALLOWED_TYPES = ('processes',)
2758
2759 def test_enable_logging(self):
2760 logger = multiprocessing.get_logger()
2761 logger.setLevel(util.SUBWARNING)
2762 self.assertTrue(logger is not None)
2763 logger.debug('this will not be printed')
2764 logger.info('nor will this')
2765 logger.setLevel(LOG_LEVEL)
2766
Antoine Pitrou0d1b38c2010-11-02 23:50:11 +00002767 @classmethod
2768 def _test_level(cls, conn):
Benjamin Petersone711caf2008-06-11 16:44:04 +00002769 logger = multiprocessing.get_logger()
2770 conn.send(logger.getEffectiveLevel())
2771
2772 def test_level(self):
2773 LEVEL1 = 32
2774 LEVEL2 = 37
2775
2776 logger = multiprocessing.get_logger()
2777 root_logger = logging.getLogger()
2778 root_level = root_logger.level
2779
2780 reader, writer = multiprocessing.Pipe(duplex=False)
2781
2782 logger.setLevel(LEVEL1)
Jesus Cea94f964f2011-09-09 20:26:57 +02002783 p = self.Process(target=self._test_level, args=(writer,))
2784 p.daemon = True
2785 p.start()
Benjamin Petersone711caf2008-06-11 16:44:04 +00002786 self.assertEqual(LEVEL1, reader.recv())
2787
2788 logger.setLevel(logging.NOTSET)
2789 root_logger.setLevel(LEVEL2)
Jesus Cea94f964f2011-09-09 20:26:57 +02002790 p = self.Process(target=self._test_level, args=(writer,))
2791 p.daemon = True
2792 p.start()
Benjamin Petersone711caf2008-06-11 16:44:04 +00002793 self.assertEqual(LEVEL2, reader.recv())
2794
2795 root_logger.setLevel(root_level)
2796 logger.setLevel(level=LOG_LEVEL)
2797
Jesse Nollerb9a49b72009-11-21 18:09:38 +00002798
Jesse Nollerf4ae35f2009-11-24 14:22:24 +00002799# class _TestLoggingProcessName(BaseTestCase):
2800#
2801# def handle(self, record):
2802# assert record.processName == multiprocessing.current_process().name
2803# self.__handled = True
2804#
2805# def test_logging(self):
2806# handler = logging.Handler()
2807# handler.handle = self.handle
2808# self.__handled = False
2809# # Bypass getLogger() and side-effects
2810# logger = logging.getLoggerClass()(
2811# 'multiprocessing.test.TestLoggingProcessName')
2812# logger.addHandler(handler)
2813# logger.propagate = False
2814#
2815# logger.warn('foo')
2816# assert self.__handled
Jesse Nollerb9a49b72009-11-21 18:09:38 +00002817
Benjamin Petersone711caf2008-06-11 16:44:04 +00002818#
Jesse Noller6214edd2009-01-19 16:23:53 +00002819# Test to verify handle verification, see issue 3321
2820#
2821
2822class TestInvalidHandle(unittest.TestCase):
2823
Florent Xiclunafd1b0932010-03-28 00:25:02 +00002824 @unittest.skipIf(WIN32, "skipped on Windows")
Jesse Noller6214edd2009-01-19 16:23:53 +00002825 def test_invalid_handles(self):
Antoine Pitrou87cf2202011-05-09 17:04:27 +02002826 conn = multiprocessing.connection.Connection(44977608)
2827 try:
2828 self.assertRaises((ValueError, IOError), conn.poll)
2829 finally:
2830 # Hack private attribute _handle to avoid printing an error
2831 # in conn.__del__
2832 conn._handle = None
2833 self.assertRaises((ValueError, IOError),
2834 multiprocessing.connection.Connection, -1)
Florent Xiclunafd1b0932010-03-28 00:25:02 +00002835
Jesse Noller6214edd2009-01-19 16:23:53 +00002836#
Benjamin Petersone711caf2008-06-11 16:44:04 +00002837# Functions used to create test cases from the base ones in this module
2838#
2839
2840def get_attributes(Source, names):
2841 d = {}
2842 for name in names:
2843 obj = getattr(Source, name)
2844 if type(obj) == type(get_attributes):
2845 obj = staticmethod(obj)
2846 d[name] = obj
2847 return d
2848
2849def create_test_cases(Mixin, type):
2850 result = {}
2851 glob = globals()
Florent Xiclunafd1b0932010-03-28 00:25:02 +00002852 Type = type.capitalize()
Benjamin Petersone711caf2008-06-11 16:44:04 +00002853
2854 for name in list(glob.keys()):
2855 if name.startswith('_Test'):
2856 base = glob[name]
2857 if type in base.ALLOWED_TYPES:
2858 newname = 'With' + Type + name[1:]
2859 class Temp(base, unittest.TestCase, Mixin):
2860 pass
2861 result[newname] = Temp
2862 Temp.__name__ = newname
2863 Temp.__module__ = Mixin.__module__
2864 return result
2865
2866#
2867# Create test cases
2868#
2869
2870class ProcessesMixin(object):
2871 TYPE = 'processes'
2872 Process = multiprocessing.Process
2873 locals().update(get_attributes(multiprocessing, (
2874 'Queue', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore',
Richard Oudkerk3730a172012-06-15 18:26:07 +01002875 'Condition', 'Event', 'Barrier', 'Value', 'Array', 'RawValue',
Benjamin Petersone711caf2008-06-11 16:44:04 +00002876 'RawArray', 'current_process', 'active_children', 'Pipe',
Richard Oudkerke41682b2012-06-06 19:04:57 +01002877 'connection', 'JoinableQueue', 'Pool'
Benjamin Petersone711caf2008-06-11 16:44:04 +00002878 )))
2879
2880testcases_processes = create_test_cases(ProcessesMixin, type='processes')
2881globals().update(testcases_processes)
2882
2883
2884class ManagerMixin(object):
2885 TYPE = 'manager'
2886 Process = multiprocessing.Process
2887 manager = object.__new__(multiprocessing.managers.SyncManager)
2888 locals().update(get_attributes(manager, (
2889 'Queue', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore',
Richard Oudkerk3730a172012-06-15 18:26:07 +01002890 'Condition', 'Event', 'Barrier', 'Value', 'Array', 'list', 'dict',
Richard Oudkerke41682b2012-06-06 19:04:57 +01002891 'Namespace', 'JoinableQueue', 'Pool'
Benjamin Petersone711caf2008-06-11 16:44:04 +00002892 )))
2893
2894testcases_manager = create_test_cases(ManagerMixin, type='manager')
2895globals().update(testcases_manager)
2896
2897
2898class ThreadsMixin(object):
2899 TYPE = 'threads'
2900 Process = multiprocessing.dummy.Process
2901 locals().update(get_attributes(multiprocessing.dummy, (
2902 'Queue', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore',
Richard Oudkerk3730a172012-06-15 18:26:07 +01002903 'Condition', 'Event', 'Barrier', 'Value', 'Array', 'current_process',
Benjamin Petersone711caf2008-06-11 16:44:04 +00002904 'active_children', 'Pipe', 'connection', 'dict', 'list',
Richard Oudkerke41682b2012-06-06 19:04:57 +01002905 'Namespace', 'JoinableQueue', 'Pool'
Benjamin Petersone711caf2008-06-11 16:44:04 +00002906 )))
2907
2908testcases_threads = create_test_cases(ThreadsMixin, type='threads')
2909globals().update(testcases_threads)
2910
Neal Norwitz5d6415e2008-08-25 01:53:32 +00002911class OtherTest(unittest.TestCase):
2912 # TODO: add more tests for deliver/answer challenge.
2913 def test_deliver_challenge_auth_failure(self):
2914 class _FakeConnection(object):
2915 def recv_bytes(self, size):
Neal Norwitzec105ad2008-08-25 03:05:54 +00002916 return b'something bogus'
Neal Norwitz5d6415e2008-08-25 01:53:32 +00002917 def send_bytes(self, data):
2918 pass
2919 self.assertRaises(multiprocessing.AuthenticationError,
2920 multiprocessing.connection.deliver_challenge,
2921 _FakeConnection(), b'abc')
2922
2923 def test_answer_challenge_auth_failure(self):
2924 class _FakeConnection(object):
2925 def __init__(self):
2926 self.count = 0
2927 def recv_bytes(self, size):
2928 self.count += 1
2929 if self.count == 1:
2930 return multiprocessing.connection.CHALLENGE
2931 elif self.count == 2:
Neal Norwitzec105ad2008-08-25 03:05:54 +00002932 return b'something bogus'
2933 return b''
Neal Norwitz5d6415e2008-08-25 01:53:32 +00002934 def send_bytes(self, data):
2935 pass
2936 self.assertRaises(multiprocessing.AuthenticationError,
2937 multiprocessing.connection.answer_challenge,
2938 _FakeConnection(), b'abc')
2939
Benjamin Petersonf47ed4a2009-04-11 20:45:40 +00002940#
2941# Test Manager.start()/Pool.__init__() initializer feature - see issue 5585
2942#
2943
2944def initializer(ns):
2945 ns.test += 1
2946
2947class TestInitializers(unittest.TestCase):
2948 def setUp(self):
2949 self.mgr = multiprocessing.Manager()
2950 self.ns = self.mgr.Namespace()
2951 self.ns.test = 0
2952
2953 def tearDown(self):
2954 self.mgr.shutdown()
Richard Oudkerka6becaa2012-05-03 18:29:02 +01002955 self.mgr.join()
Benjamin Petersonf47ed4a2009-04-11 20:45:40 +00002956
2957 def test_manager_initializer(self):
2958 m = multiprocessing.managers.SyncManager()
2959 self.assertRaises(TypeError, m.start, 1)
2960 m.start(initializer, (self.ns,))
2961 self.assertEqual(self.ns.test, 1)
2962 m.shutdown()
Richard Oudkerka6becaa2012-05-03 18:29:02 +01002963 m.join()
Benjamin Petersonf47ed4a2009-04-11 20:45:40 +00002964
2965 def test_pool_initializer(self):
2966 self.assertRaises(TypeError, multiprocessing.Pool, initializer=1)
2967 p = multiprocessing.Pool(1, initializer, (self.ns,))
2968 p.close()
2969 p.join()
2970 self.assertEqual(self.ns.test, 1)
2971
Alexandre Vassalottic57a84f2009-07-17 12:07:01 +00002972#
2973# Issue 5155, 5313, 5331: Test process in processes
2974# Verifies os.close(sys.stdin.fileno) vs. sys.stdin.close() behavior
2975#
2976
2977def _ThisSubProcess(q):
2978 try:
2979 item = q.get(block=False)
2980 except pyqueue.Empty:
2981 pass
2982
2983def _TestProcess(q):
2984 queue = multiprocessing.Queue()
2985 subProc = multiprocessing.Process(target=_ThisSubProcess, args=(queue,))
Jesus Cea94f964f2011-09-09 20:26:57 +02002986 subProc.daemon = True
Alexandre Vassalottic57a84f2009-07-17 12:07:01 +00002987 subProc.start()
2988 subProc.join()
2989
2990def _afunc(x):
2991 return x*x
2992
2993def pool_in_process():
2994 pool = multiprocessing.Pool(processes=4)
2995 x = pool.map(_afunc, [1, 2, 3, 4, 5, 6, 7])
Richard Oudkerk225cb8d2012-05-02 19:36:11 +01002996 pool.close()
2997 pool.join()
Alexandre Vassalottic57a84f2009-07-17 12:07:01 +00002998
2999class _file_like(object):
3000 def __init__(self, delegate):
3001 self._delegate = delegate
3002 self._pid = None
3003
3004 @property
3005 def cache(self):
3006 pid = os.getpid()
3007 # There are no race conditions since fork keeps only the running thread
3008 if pid != self._pid:
3009 self._pid = pid
3010 self._cache = []
3011 return self._cache
3012
3013 def write(self, data):
3014 self.cache.append(data)
3015
3016 def flush(self):
3017 self._delegate.write(''.join(self.cache))
3018 self._cache = []
3019
3020class TestStdinBadfiledescriptor(unittest.TestCase):
3021
3022 def test_queue_in_process(self):
3023 queue = multiprocessing.Queue()
3024 proc = multiprocessing.Process(target=_TestProcess, args=(queue,))
3025 proc.start()
3026 proc.join()
3027
3028 def test_pool_in_process(self):
3029 p = multiprocessing.Process(target=pool_in_process)
3030 p.start()
3031 p.join()
3032
3033 def test_flushing(self):
3034 sio = io.StringIO()
3035 flike = _file_like(sio)
3036 flike.write('foo')
3037 proc = multiprocessing.Process(target=lambda: flike.flush())
3038 flike.flush()
3039 assert sio.getvalue() == 'foo'
3040
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003041
3042class TestWait(unittest.TestCase):
3043
3044 @classmethod
3045 def _child_test_wait(cls, w, slow):
3046 for i in range(10):
3047 if slow:
3048 time.sleep(random.random()*0.1)
3049 w.send((i, os.getpid()))
3050 w.close()
3051
3052 def test_wait(self, slow=False):
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003053 from multiprocessing.connection import wait
3054 readers = []
3055 procs = []
3056 messages = []
3057
3058 for i in range(4):
Antoine Pitrou5bb9a8f2012-03-06 13:43:24 +01003059 r, w = multiprocessing.Pipe(duplex=False)
3060 p = multiprocessing.Process(target=self._child_test_wait, args=(w, slow))
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003061 p.daemon = True
3062 p.start()
3063 w.close()
3064 readers.append(r)
3065 procs.append(p)
Antoine Pitrou6c64cc12012-03-06 13:42:35 +01003066 self.addCleanup(p.join)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003067
3068 while readers:
3069 for r in wait(readers):
3070 try:
3071 msg = r.recv()
3072 except EOFError:
3073 readers.remove(r)
3074 r.close()
3075 else:
3076 messages.append(msg)
3077
3078 messages.sort()
3079 expected = sorted((i, p.pid) for i in range(10) for p in procs)
3080 self.assertEqual(messages, expected)
3081
3082 @classmethod
3083 def _child_test_wait_socket(cls, address, slow):
3084 s = socket.socket()
3085 s.connect(address)
3086 for i in range(10):
3087 if slow:
3088 time.sleep(random.random()*0.1)
3089 s.sendall(('%s\n' % i).encode('ascii'))
3090 s.close()
3091
3092 def test_wait_socket(self, slow=False):
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003093 from multiprocessing.connection import wait
3094 l = socket.socket()
3095 l.bind(('', 0))
3096 l.listen(4)
3097 addr = ('localhost', l.getsockname()[1])
3098 readers = []
3099 procs = []
3100 dic = {}
3101
3102 for i in range(4):
Antoine Pitrou5bb9a8f2012-03-06 13:43:24 +01003103 p = multiprocessing.Process(target=self._child_test_wait_socket,
3104 args=(addr, slow))
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003105 p.daemon = True
3106 p.start()
3107 procs.append(p)
Antoine Pitrou6c64cc12012-03-06 13:42:35 +01003108 self.addCleanup(p.join)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003109
3110 for i in range(4):
3111 r, _ = l.accept()
3112 readers.append(r)
3113 dic[r] = []
3114 l.close()
3115
3116 while readers:
3117 for r in wait(readers):
3118 msg = r.recv(32)
3119 if not msg:
3120 readers.remove(r)
3121 r.close()
3122 else:
3123 dic[r].append(msg)
3124
3125 expected = ''.join('%s\n' % i for i in range(10)).encode('ascii')
3126 for v in dic.values():
3127 self.assertEqual(b''.join(v), expected)
3128
3129 def test_wait_slow(self):
3130 self.test_wait(True)
3131
3132 def test_wait_socket_slow(self):
Richard Oudkerk104b3f42012-05-08 16:08:07 +01003133 self.test_wait_socket(True)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003134
3135 def test_wait_timeout(self):
3136 from multiprocessing.connection import wait
3137
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003138 expected = 5
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003139 a, b = multiprocessing.Pipe()
3140
3141 start = time.time()
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003142 res = wait([a, b], expected)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003143 delta = time.time() - start
3144
3145 self.assertEqual(res, [])
Richard Oudkerk6dbca362012-05-06 16:46:36 +01003146 self.assertLess(delta, expected * 2)
3147 self.assertGreater(delta, expected * 0.5)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003148
3149 b.send(None)
3150
3151 start = time.time()
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003152 res = wait([a, b], 20)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003153 delta = time.time() - start
3154
3155 self.assertEqual(res, [a])
Antoine Pitrou37749772012-03-09 18:40:15 +01003156 self.assertLess(delta, 0.4)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003157
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003158 @classmethod
3159 def signal_and_sleep(cls, sem, period):
3160 sem.release()
3161 time.sleep(period)
3162
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003163 def test_wait_integer(self):
3164 from multiprocessing.connection import wait
3165
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003166 expected = 3
3167 sem = multiprocessing.Semaphore(0)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003168 a, b = multiprocessing.Pipe()
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003169 p = multiprocessing.Process(target=self.signal_and_sleep,
3170 args=(sem, expected))
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003171
3172 p.start()
3173 self.assertIsInstance(p.sentinel, int)
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003174 self.assertTrue(sem.acquire(timeout=20))
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003175
3176 start = time.time()
3177 res = wait([a, p.sentinel, b], expected + 20)
3178 delta = time.time() - start
3179
3180 self.assertEqual(res, [p.sentinel])
Antoine Pitrou37749772012-03-09 18:40:15 +01003181 self.assertLess(delta, expected + 2)
3182 self.assertGreater(delta, expected - 2)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003183
3184 a.send(None)
3185
3186 start = time.time()
3187 res = wait([a, p.sentinel, b], 20)
3188 delta = time.time() - start
3189
3190 self.assertEqual(res, [p.sentinel, b])
Antoine Pitrou37749772012-03-09 18:40:15 +01003191 self.assertLess(delta, 0.4)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003192
3193 b.send(None)
3194
3195 start = time.time()
3196 res = wait([a, p.sentinel, b], 20)
3197 delta = time.time() - start
3198
3199 self.assertEqual(res, [a, p.sentinel, b])
Antoine Pitrou37749772012-03-09 18:40:15 +01003200 self.assertLess(delta, 0.4)
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003201
Richard Oudkerk009b15e2012-05-04 09:44:39 +01003202 p.terminate()
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003203 p.join()
3204
Richard Oudkerk59d54042012-05-10 16:11:12 +01003205 def test_neg_timeout(self):
3206 from multiprocessing.connection import wait
3207 a, b = multiprocessing.Pipe()
3208 t = time.time()
3209 res = wait([a], timeout=-1)
3210 t = time.time() - t
3211 self.assertEqual(res, [])
3212 self.assertLess(t, 1)
3213 a.close()
3214 b.close()
Antoine Pitroubdb1cf12012-03-05 19:28:37 +01003215
Antoine Pitrou709176f2012-04-01 17:19:09 +02003216#
3217# Issue 14151: Test invalid family on invalid environment
3218#
3219
3220class TestInvalidFamily(unittest.TestCase):
3221
3222 @unittest.skipIf(WIN32, "skipped on Windows")
3223 def test_invalid_family(self):
3224 with self.assertRaises(ValueError):
3225 multiprocessing.connection.Listener(r'\\.\test')
3226
Antoine Pitrou6d20cba2012-04-03 20:12:23 +02003227 @unittest.skipUnless(WIN32, "skipped on non-Windows platforms")
3228 def test_invalid_family_win32(self):
3229 with self.assertRaises(ValueError):
3230 multiprocessing.connection.Listener('/var/test.pipe')
Antoine Pitrou93bba8f2012-04-01 17:25:49 +02003231
Richard Oudkerk77c84f22012-05-18 14:28:02 +01003232#
3233# Issue 12098: check sys.flags of child matches that for parent
3234#
3235
3236class TestFlags(unittest.TestCase):
3237 @classmethod
3238 def run_in_grandchild(cls, conn):
3239 conn.send(tuple(sys.flags))
3240
3241 @classmethod
3242 def run_in_child(cls):
3243 import json
3244 r, w = multiprocessing.Pipe(duplex=False)
3245 p = multiprocessing.Process(target=cls.run_in_grandchild, args=(w,))
3246 p.start()
3247 grandchild_flags = r.recv()
3248 p.join()
3249 r.close()
3250 w.close()
3251 flags = (tuple(sys.flags), grandchild_flags)
3252 print(json.dumps(flags))
3253
3254 def test_flags(self):
3255 import json, subprocess
3256 # start child process using unusual flags
3257 prog = ('from test.test_multiprocessing import TestFlags; ' +
3258 'TestFlags.run_in_child()')
3259 data = subprocess.check_output(
3260 [sys.executable, '-E', '-S', '-O', '-c', prog])
3261 child_flags, grandchild_flags = json.loads(data.decode('ascii'))
3262 self.assertEqual(child_flags, grandchild_flags)
3263
Alexandre Vassalottic57a84f2009-07-17 12:07:01 +00003264testcases_other = [OtherTest, TestInvalidHandle, TestInitializers,
Richard Oudkerk77c84f22012-05-18 14:28:02 +01003265 TestStdinBadfiledescriptor, TestWait, TestInvalidFamily,
3266 TestFlags]
Neal Norwitz5d6415e2008-08-25 01:53:32 +00003267
Benjamin Petersone711caf2008-06-11 16:44:04 +00003268#
3269#
3270#
3271
3272def test_main(run=None):
Jesse Nollerd00df3c2008-06-18 14:22:48 +00003273 if sys.platform.startswith("linux"):
3274 try:
3275 lock = multiprocessing.RLock()
3276 except OSError:
Benjamin Petersone549ead2009-03-28 21:42:05 +00003277 raise unittest.SkipTest("OSError raises on RLock creation, see issue 3111!")
Benjamin Peterson3c0dd062008-06-17 22:43:48 +00003278
Charles-François Natali221ef672011-11-22 18:55:22 +01003279 check_enough_semaphores()
3280
Benjamin Petersone711caf2008-06-11 16:44:04 +00003281 if run is None:
3282 from test.support import run_unittest as run
3283
3284 util.get_temp_dir() # creates temp directory for use by all processes
3285
3286 multiprocessing.get_logger().setLevel(LOG_LEVEL)
3287
Benjamin Peterson41181742008-07-02 20:22:54 +00003288 ProcessesMixin.pool = multiprocessing.Pool(4)
3289 ThreadsMixin.pool = multiprocessing.dummy.Pool(4)
3290 ManagerMixin.manager.__init__()
3291 ManagerMixin.manager.start()
3292 ManagerMixin.pool = ManagerMixin.manager.Pool(4)
Benjamin Petersone711caf2008-06-11 16:44:04 +00003293
3294 testcases = (
Benjamin Peterson41181742008-07-02 20:22:54 +00003295 sorted(testcases_processes.values(), key=lambda tc:tc.__name__) +
3296 sorted(testcases_threads.values(), key=lambda tc:tc.__name__) +
Neal Norwitz5d6415e2008-08-25 01:53:32 +00003297 sorted(testcases_manager.values(), key=lambda tc:tc.__name__) +
3298 testcases_other
Benjamin Petersone711caf2008-06-11 16:44:04 +00003299 )
3300
3301 loadTestsFromTestCase = unittest.defaultTestLoader.loadTestsFromTestCase
3302 suite = unittest.TestSuite(loadTestsFromTestCase(tc) for tc in testcases)
Richard Oudkerk225cb8d2012-05-02 19:36:11 +01003303 try:
3304 run(suite)
3305 finally:
3306 ThreadsMixin.pool.terminate()
3307 ProcessesMixin.pool.terminate()
3308 ManagerMixin.pool.terminate()
3309 ManagerMixin.pool.join()
3310 ManagerMixin.manager.shutdown()
Richard Oudkerka6becaa2012-05-03 18:29:02 +01003311 ManagerMixin.manager.join()
Richard Oudkerk225cb8d2012-05-02 19:36:11 +01003312 ThreadsMixin.pool.join()
3313 ProcessesMixin.pool.join()
3314 del ProcessesMixin.pool, ThreadsMixin.pool, ManagerMixin.pool
Benjamin Petersone711caf2008-06-11 16:44:04 +00003315
3316def main():
3317 test_main(unittest.TextTestRunner(verbosity=2).run)
3318
3319if __name__ == '__main__':
3320 main()