Index: Lib/test/test_multiprocessing.py |
=================================================================== |
--- Lib/test/test_multiprocessing.py (revision 0) |
+++ Lib/test/test_multiprocessing.py (revision 0) |
@@ -0,0 +1,1791 @@ |
+# |
+# Unit tests for the multiprocessing package |
+# |
+ |
+import unittest |
+import threading |
+import Queue |
+import time |
+import sys |
+import os |
+import gc |
+import signal |
+import array |
+import copy |
+import socket |
+import random |
+import logging |
+ |
+import _multiprocessing |
+import multiprocessing.dummy |
+import multiprocessing.connection |
+import multiprocessing.managers |
+import multiprocessing.heap |
+import multiprocessing.managers |
+import multiprocessing.pool |
+ |
+from multiprocessing import util |
+ |
+# |
+# |
+# |
+ |
+if sys.version_info >= (3, 0): |
+ def latin(s): |
+ return s.encode('latin') |
+else: |
+ latin = str |
+ |
+try: |
+ bytes |
+except NameError: |
+ bytes = str |
+ def bytearray(seq): |
+ return array.array('c', seq) |
+ |
+# |
+# Constants |
+# |
+ |
+LOG_LEVEL = util.SUBWARNING |
+#LOG_LEVEL = logging.WARNING |
+ |
+DELTA = 0.1 |
+CHECK_TIMINGS = False # making true makes tests take a lot longer |
+ # and can sometimes cause some non-serious |
+ # failures because some calls block a bit |
+ # longer than expected |
+if CHECK_TIMINGS: |
+ TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.82, 0.35, 1.4 |
+else: |
+ TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.1, 0.1, 0.1 |
+ |
+HAVE_GETVALUE = not getattr(_multiprocessing, |
+ 'HAVE_BROKEN_SEM_GETVALUE', False) |
+ |
+# |
+# Creates a wrapper for a function which records the time it takes to finish |
+# |
+ |
+class TimingWrapper(object): |
+ |
+ def __init__(self, func): |
+ self.func = func |
+ self.elapsed = None |
+ |
+ def __call__(self, *args, **kwds): |
+ t = time.time() |
+ try: |
+ return self.func(*args, **kwds) |
+ finally: |
+ self.elapsed = time.time() - t |
+ |
+# |
+# Base class for test cases |
+# |
+ |
+class BaseTestCase(object): |
+ |
+ ALLOWED_TYPES = ('processes', 'manager', 'threads') |
+ |
+ def assertTimingAlmostEqual(self, a, b): |
+ if CHECK_TIMINGS: |
+ self.assertAlmostEqual(a, b, 1) |
+ |
+ def assertReturnsIfImplemented(self, value, func, *args): |
+ try: |
+ res = func(*args) |
+ except NotImplementedError: |
+ pass |
+ else: |
+ return self.assertEqual(value, res) |
+ |
+# |
+# Return the value of a semaphore |
+# |
+ |
+def get_value(self): |
+ try: |
+ return self.get_value() |
+ except AttributeError: |
+ try: |
+ return self._Semaphore__value |
+ except AttributeError: |
+ try: |
+ return self._value |
+ except AttributeError: |
+ raise NotImplementedError |
+ |
+# |
+# Testcases |
+# |
+ |
+class _TestProcess(BaseTestCase): |
+ |
+ ALLOWED_TYPES = ('processes', 'threads') |
+ |
+ def test_current(self): |
+ if self.TYPE == 'threads': |
+ return |
+ |
+ current = self.current_process() |
+ authkey = current.get_authkey() |
+ |
+ self.assertTrue(current.is_alive()) |
+ self.assertTrue(not current.is_daemon()) |
+ self.assertTrue(isinstance(authkey, bytes)) |
+ self.assertTrue(len(authkey) > 0) |
+ self.assertEqual(current.get_ident(), os.getpid()) |
+ self.assertEqual(current.get_exitcode(), None) |
+ |
+ def _test(self, q, *args, **kwds): |
+ current = self.current_process() |
+ q.put(args) |
+ q.put(kwds) |
+ q.put(current.get_name()) |
+ if self.TYPE != 'threads': |
+ q.put(bytes(current.get_authkey())) |
+ q.put(current.pid) |
+ |
+ def test_process(self): |
+ q = self.Queue(1) |
+ e = self.Event() |
+ args = (q, 1, 2) |
+ kwargs = {'hello':23, 'bye':2.54} |
+ name = 'SomeProcess' |
+ p = self.Process( |
+ target=self._test, args=args, kwargs=kwargs, name=name |
+ ) |
+ p.set_daemon(True) |
+ current = self.current_process() |
+ |
+ if self.TYPE != 'threads': |
+ self.assertEquals(p.get_authkey(), current.get_authkey()) |
+ self.assertEquals(p.is_alive(), False) |
+ self.assertEquals(p.is_daemon(), True) |
+ self.assertTrue(p not in self.active_children()) |
+ self.assertTrue(type(self.active_children()) is list) |
+ self.assertEqual(p.get_exitcode(), None) |
+ |
+ p.start() |
+ |
+ self.assertEquals(p.get_exitcode(), None) |
+ self.assertEquals(p.is_alive(), True) |
+ self.assertTrue(p in self.active_children()) |
+ |
+ self.assertEquals(q.get(), args[1:]) |
+ self.assertEquals(q.get(), kwargs) |
+ self.assertEquals(q.get(), p.get_name()) |
+ if self.TYPE != 'threads': |
+ self.assertEquals(q.get(), current.get_authkey()) |
+ self.assertEquals(q.get(), p.pid) |
+ |
+ p.join() |
+ |
+ self.assertEquals(p.get_exitcode(), 0) |
+ self.assertEquals(p.is_alive(), False) |
+ self.assertTrue(p not in self.active_children()) |
+ |
+ def _test_terminate(self): |
+ time.sleep(1000) |
+ |
+ def test_terminate(self): |
+ if self.TYPE == 'threads': |
+ return |
+ |
+ p = self.Process(target=self._test_terminate) |
+ p.set_daemon(True) |
+ p.start() |
+ |
+ self.assertEqual(p.is_alive(), True) |
+ self.assertTrue(p in self.active_children()) |
+ self.assertEqual(p.get_exitcode(), None) |
+ |
+ p.terminate() |
+ |
+ join = TimingWrapper(p.join) |
+ self.assertEqual(join(), None) |
+ self.assertTimingAlmostEqual(join.elapsed, 0.0) |
+ |
+ self.assertEqual(p.is_alive(), False) |
+ self.assertTrue(p not in self.active_children()) |
+ |
+ p.join() |
+ |
+ # XXX sometimes get p.get_exitcode() == 0 on Windows ... |
+ #self.assertEqual(p.get_exitcode(), -signal.SIGTERM) |
+ |
+ def test_cpu_count(self): |
+ try: |
+ cpus = multiprocessing.cpu_count() |
+ except NotImplementedError: |
+ cpus = 1 |
+ self.assertTrue(type(cpus) is int) |
+ self.assertTrue(cpus >= 1) |
+ |
+ def test_active_children(self): |
+ self.assertEqual(type(self.active_children()), list) |
+ |
+ p = self.Process(target=time.sleep, args=(DELTA,)) |
+ self.assertTrue(p not in self.active_children()) |
+ |
+ p.start() |
+ self.assertTrue(p in self.active_children()) |
+ |
+ p.join() |
+ self.assertTrue(p not in self.active_children()) |
+ |
+ def _test_recursion(self, wconn, id): |
+ from multiprocessing import forking |
+ wconn.send(id) |
+ if len(id) < 2: |
+ for i in range(2): |
+ p = self.Process( |
+ target=self._test_recursion, args=(wconn, id+[i]) |
+ ) |
+ p.start() |
+ p.join() |
+ |
+ def test_recursion(self): |
+ rconn, wconn = self.Pipe(duplex=False) |
+ self._test_recursion(wconn, []) |
+ |
+ time.sleep(DELTA) |
+ result = [] |
+ while rconn.poll(): |
+ result.append(rconn.recv()) |
+ |
+ expected = [ |
+ [], |
+ [0], |
+ [0, 0], |
+ [0, 1], |
+ [1], |
+ [1, 0], |
+ [1, 1] |
+ ] |
+ self.assertEqual(result, expected) |
+ |
+# |
+# |
+# |
+ |
+class _UpperCaser(multiprocessing.Process): |
+ |
+ def __init__(self): |
+ multiprocessing.Process.__init__(self) |
+ self.child_conn, self.parent_conn = multiprocessing.Pipe() |
+ |
+ def run(self): |
+ self.parent_conn.close() |
+ for s in iter(self.child_conn.recv, None): |
+ self.child_conn.send(s.upper()) |
+ self.child_conn.close() |
+ |
+ def submit(self, s): |
+ assert type(s) is str |
+ self.parent_conn.send(s) |
+ return self.parent_conn.recv() |
+ |
+ def stop(self): |
+ self.parent_conn.send(None) |
+ self.parent_conn.close() |
+ self.child_conn.close() |
+ |
+class _TestSubclassingProcess(BaseTestCase): |
+ |
+ ALLOWED_TYPES = ('processes',) |
+ |
+ def test_subclassing(self): |
+ uppercaser = _UpperCaser() |
+ uppercaser.start() |
+ self.assertEqual(uppercaser.submit('hello'), 'HELLO') |
+ self.assertEqual(uppercaser.submit('world'), 'WORLD') |
+ uppercaser.stop() |
+ uppercaser.join() |
+ |
+# |
+# |
+# |
+ |
+def queue_empty(q): |
+ if hasattr(q, 'empty'): |
+ return q.empty() |
+ else: |
+ return q.qsize() == 0 |
+ |
+def queue_full(q, maxsize): |
+ if hasattr(q, 'full'): |
+ return q.full() |
+ else: |
+ return q.qsize() == maxsize |
+ |
+ |
+class _TestQueue(BaseTestCase): |
+ |
+ |
+ def _test_put(self, queue, child_can_start, parent_can_continue): |
+ child_can_start.wait() |
+ for i in range(6): |
+ queue.get() |
+ parent_can_continue.set() |
+ |
+ def test_put(self): |
+ MAXSIZE = 6 |
+ queue = self.Queue(maxsize=MAXSIZE) |
+ child_can_start = self.Event() |
+ parent_can_continue = self.Event() |
+ |
+ proc = self.Process( |
+ target=self._test_put, |
+ args=(queue, child_can_start, parent_can_continue) |
+ ) |
+ proc.set_daemon(True) |
+ proc.start() |
+ |
+ self.assertEqual(queue_empty(queue), True) |
+ self.assertEqual(queue_full(queue, MAXSIZE), False) |
+ |
+ queue.put(1) |
+ queue.put(2, True) |
+ queue.put(3, True, None) |
+ queue.put(4, False) |
+ queue.put(5, False, None) |
+ queue.put_nowait(6) |
+ |
+ # the values may be in buffer but not yet in pipe so sleep a bit |
+ time.sleep(DELTA) |
+ |
+ self.assertEqual(queue_empty(queue), False) |
+ self.assertEqual(queue_full(queue, MAXSIZE), True) |
+ |
+ put = TimingWrapper(queue.put) |
+ put_nowait = TimingWrapper(queue.put_nowait) |
+ |
+ self.assertRaises(Queue.Full, put, 7, False) |
+ self.assertTimingAlmostEqual(put.elapsed, 0) |
+ |
+ self.assertRaises(Queue.Full, put, 7, False, None) |
+ self.assertTimingAlmostEqual(put.elapsed, 0) |
+ |
+ self.assertRaises(Queue.Full, put_nowait, 7) |
+ self.assertTimingAlmostEqual(put_nowait.elapsed, 0) |
+ |
+ self.assertRaises(Queue.Full, put, 7, True, TIMEOUT1) |
+ self.assertTimingAlmostEqual(put.elapsed, TIMEOUT1) |
+ |
+ self.assertRaises(Queue.Full, put, 7, False, TIMEOUT2) |
+ self.assertTimingAlmostEqual(put.elapsed, 0) |
+ |
+ self.assertRaises(Queue.Full, put, 7, True, timeout=TIMEOUT3) |
+ self.assertTimingAlmostEqual(put.elapsed, TIMEOUT3) |
+ |
+ child_can_start.set() |
+ parent_can_continue.wait() |
+ |
+ self.assertEqual(queue_empty(queue), True) |
+ self.assertEqual(queue_full(queue, MAXSIZE), False) |
+ |
+ proc.join() |
+ |
+ def _test_get(self, queue, child_can_start, parent_can_continue): |
+ child_can_start.wait() |
+ queue.put(1) |
+ queue.put(2) |
+ queue.put(3) |
+ queue.put(4) |
+ queue.put(5) |
+ parent_can_continue.set() |
+ |
+ def test_get(self): |
+ queue = self.Queue() |
+ child_can_start = self.Event() |
+ parent_can_continue = self.Event() |
+ |
+ proc = self.Process( |
+ target=self._test_get, |
+ args=(queue, child_can_start, parent_can_continue) |
+ ) |
+ proc.set_daemon(True) |
+ proc.start() |
+ |
+ self.assertEqual(queue_empty(queue), True) |
+ |
+ child_can_start.set() |
+ parent_can_continue.wait() |
+ |
+ time.sleep(DELTA) |
+ self.assertEqual(queue_empty(queue), False) |
+ |
+ self.assertEqual(queue.get(), 1) |
+ self.assertEqual(queue.get(True, None), 2) |
+ self.assertEqual(queue.get(True), 3) |
+ self.assertEqual(queue.get(timeout=1), 4) |
+ self.assertEqual(queue.get_nowait(), 5) |
+ |
+ self.assertEqual(queue_empty(queue), True) |
+ |
+ get = TimingWrapper(queue.get) |
+ get_nowait = TimingWrapper(queue.get_nowait) |
+ |
+ self.assertRaises(Queue.Empty, get, False) |
+ self.assertTimingAlmostEqual(get.elapsed, 0) |
+ |
+ self.assertRaises(Queue.Empty, get, False, None) |
+ self.assertTimingAlmostEqual(get.elapsed, 0) |
+ |
+ self.assertRaises(Queue.Empty, get_nowait) |
+ self.assertTimingAlmostEqual(get_nowait.elapsed, 0) |
+ |
+ self.assertRaises(Queue.Empty, get, True, TIMEOUT1) |
+ self.assertTimingAlmostEqual(get.elapsed, TIMEOUT1) |
+ |
+ self.assertRaises(Queue.Empty, get, False, TIMEOUT2) |
+ self.assertTimingAlmostEqual(get.elapsed, 0) |
+ |
+ self.assertRaises(Queue.Empty, get, timeout=TIMEOUT3) |
+ self.assertTimingAlmostEqual(get.elapsed, TIMEOUT3) |
+ |
+ proc.join() |
+ |
+ def _test_fork(self, queue): |
+ for i in range(10, 20): |
+ queue.put(i) |
+ # note that at this point the items may only be buffered, so the |
+ # process cannot shutdown until the feeder thread has finished |
+ # pushing items onto the pipe. |
+ |
+ def test_fork(self): |
+ # Old versions of Queue would fail to create a new feeder |
+ # thread for a forked process if the original process had its |
+ # own feeder thread. This test checks that this no longer |
+ # happens. |
+ |
+ queue = self.Queue() |
+ |
+ # put items on queue so that main process starts a feeder thread |
+ for i in range(10): |
+ queue.put(i) |
+ |
+ # wait to make sure thread starts before we fork a new process |
+ time.sleep(DELTA) |
+ |
+ # fork process |
+ p = self.Process(target=self._test_fork, args=(queue,)) |
+ p.start() |
+ |
+ # check that all expected items are in the queue |
+ for i in range(20): |
+ self.assertEqual(queue.get(), i) |
+ self.assertRaises(Queue.Empty, queue.get, False) |
+ |
+ p.join() |
+ |
+ def test_qsize(self): |
+ q = self.Queue() |
+ try: |
+ self.assertEqual(q.qsize(), 0) |
+ except NotImplementedError: |
+ return |
+ q.put(1) |
+ self.assertEqual(q.qsize(), 1) |
+ q.put(5) |
+ self.assertEqual(q.qsize(), 2) |
+ q.get() |
+ self.assertEqual(q.qsize(), 1) |
+ q.get() |
+ self.assertEqual(q.qsize(), 0) |
+ |
+ def _test_task_done(self, q): |
+ for obj in iter(q.get, None): |
+ time.sleep(DELTA) |
+ q.task_done() |
+ |
+ def test_task_done(self): |
+ queue = self.JoinableQueue() |
+ |
+ if sys.version_info < (2, 5) and not hasattr(queue, 'task_done'): |
+ return |
+ |
+ workers = [self.Process(target=self._test_task_done, args=(queue,)) |
+ for i in xrange(4)] |
+ |
+ for p in workers: |
+ p.start() |
+ |
+ for i in xrange(10): |
+ queue.put(i) |
+ |
+ queue.join() |
+ |
+ for p in workers: |
+ queue.put(None) |
+ |
+ for p in workers: |
+ p.join() |
+ |
+# |
+# |
+# |
+ |
+class _TestLock(BaseTestCase): |
+ |
+ def test_lock(self): |
+ lock = self.Lock() |
+ self.assertEqual(lock.acquire(), True) |
+ self.assertEqual(lock.acquire(False), False) |
+ self.assertEqual(lock.release(), None) |
+ self.assertRaises((ValueError, threading.ThreadError), lock.release) |
+ |
+ def test_rlock(self): |
+ lock = self.RLock() |
+ self.assertEqual(lock.acquire(), True) |
+ self.assertEqual(lock.acquire(), True) |
+ self.assertEqual(lock.acquire(), True) |
+ self.assertEqual(lock.release(), None) |
+ self.assertEqual(lock.release(), None) |
+ self.assertEqual(lock.release(), None) |
+ self.assertRaises((AssertionError, RuntimeError), lock.release) |
+ |
+ |
+class _TestSemaphore(BaseTestCase): |
+ |
+ def _test_semaphore(self, sem): |
+ self.assertReturnsIfImplemented(2, get_value, sem) |
+ self.assertEqual(sem.acquire(), True) |
+ self.assertReturnsIfImplemented(1, get_value, sem) |
+ self.assertEqual(sem.acquire(), True) |
+ self.assertReturnsIfImplemented(0, get_value, sem) |
+ self.assertEqual(sem.acquire(False), False) |
+ self.assertReturnsIfImplemented(0, get_value, sem) |
+ self.assertEqual(sem.release(), None) |
+ self.assertReturnsIfImplemented(1, get_value, sem) |
+ self.assertEqual(sem.release(), None) |
+ self.assertReturnsIfImplemented(2, get_value, sem) |
+ |
+ def test_semaphore(self): |
+ sem = self.Semaphore(2) |
+ self._test_semaphore(sem) |
+ self.assertEqual(sem.release(), None) |
+ self.assertReturnsIfImplemented(3, get_value, sem) |
+ self.assertEqual(sem.release(), None) |
+ self.assertReturnsIfImplemented(4, get_value, sem) |
+ |
+ def test_bounded_semaphore(self): |
+ sem = self.BoundedSemaphore(2) |
+ self._test_semaphore(sem) |
+ # Currently fails on OS/X |
+ #if HAVE_GETVALUE: |
+ # self.assertRaises(ValueError, sem.release) |
+ # self.assertReturnsIfImplemented(2, get_value, sem) |
+ |
+ def test_timeout(self): |
+ if self.TYPE != 'processes': |
+ return |
+ |
+ sem = self.Semaphore(0) |
+ acquire = TimingWrapper(sem.acquire) |
+ |
+ self.assertEqual(acquire(False), False) |
+ self.assertTimingAlmostEqual(acquire.elapsed, 0.0) |
+ |
+ self.assertEqual(acquire(False, None), False) |
+ self.assertTimingAlmostEqual(acquire.elapsed, 0.0) |
+ |
+ self.assertEqual(acquire(False, TIMEOUT1), False) |
+ self.assertTimingAlmostEqual(acquire.elapsed, 0) |
+ |
+ self.assertEqual(acquire(True, TIMEOUT2), False) |
+ self.assertTimingAlmostEqual(acquire.elapsed, TIMEOUT2) |
+ |
+ self.assertEqual(acquire(timeout=TIMEOUT3), False) |
+ self.assertTimingAlmostEqual(acquire.elapsed, TIMEOUT3) |
+ |
+ |
+class _TestCondition(BaseTestCase): |
+ |
+ def f(self, cond, sleeping, woken, timeout=None): |
+ cond.acquire() |
+ sleeping.release() |
+ cond.wait(timeout) |
+ woken.release() |
+ cond.release() |
+ |
+ def check_invariant(self, cond): |
+ # this is only supposed to succeed when there are no sleepers |
+ if self.TYPE == 'processes': |
+ try: |
+ sleepers = (cond._sleeping_count.get_value() - |
+ cond._woken_count.get_value()) |
+ self.assertEqual(sleepers, 0) |
+ self.assertEqual(cond._wait_semaphore.get_value(), 0) |
+ except NotImplementedError: |
+ pass |
+ |
+ def test_notify(self): |
+ cond = self.Condition() |
+ sleeping = self.Semaphore(0) |
+ woken = self.Semaphore(0) |
+ |
+ p = self.Process(target=self.f, args=(cond, sleeping, woken)) |
+ p.set_daemon(True) |
+ p.start() |
+ |
+ p = threading.Thread(target=self.f, args=(cond, sleeping, woken)) |
+ p.setDaemon(True) |
+ p.start() |
+ |
+ # wait for both children to start sleeping |
+ sleeping.acquire() |
+ sleeping.acquire() |
+ |
+ # check no process/thread has woken up |
+ time.sleep(DELTA) |
+ self.assertReturnsIfImplemented(0, get_value, woken) |
+ |
+ # wake up one process/thread |
+ cond.acquire() |
+ cond.notify() |
+ cond.release() |
+ |
+ # check one process/thread has woken up |
+ time.sleep(DELTA) |
+ self.assertReturnsIfImplemented(1, get_value, woken) |
+ |
+ # wake up another |
+ cond.acquire() |
+ cond.notify() |
+ cond.release() |
+ |
+ # check other has woken up |
+ time.sleep(DELTA) |
+ self.assertReturnsIfImplemented(2, get_value, woken) |
+ |
+ # check state is not mucked up |
+ self.check_invariant(cond) |
+ p.join() |
+ |
+ def test_notify_all(self): |
+ cond = self.Condition() |
+ sleeping = self.Semaphore(0) |
+ woken = self.Semaphore(0) |
+ |
+ # start some threads/processes which will timeout |
+ for i in range(3): |
+ p = self.Process(target=self.f, |
+ args=(cond, sleeping, woken, TIMEOUT1)) |
+ p.set_daemon(True) |
+ p.start() |
+ |
+ t = threading.Thread(target=self.f, |
+ args=(cond, sleeping, woken, TIMEOUT1)) |
+ t.setDaemon(True) |
+ t.start() |
+ |
+ # wait for them all to sleep |
+ for i in xrange(6): |
+ sleeping.acquire() |
+ |
+ # check they have all timed out |
+ for i in xrange(6): |
+ woken.acquire() |
+ self.assertReturnsIfImplemented(0, get_value, woken) |
+ |
+ # check state is not mucked up |
+ self.check_invariant(cond) |
+ |
+ # start some more threads/processes |
+ for i in range(3): |
+ p = self.Process(target=self.f, args=(cond, sleeping, woken)) |
+ p.set_daemon(True) |
+ p.start() |
+ |
+ t = threading.Thread(target=self.f, args=(cond, sleeping, woken)) |
+ t.setDaemon(True) |
+ t.start() |
+ |
+ # wait for them to all sleep |
+ for i in xrange(6): |
+ sleeping.acquire() |
+ |
+ # check no process/thread has woken up |
+ time.sleep(DELTA) |
+ self.assertReturnsIfImplemented(0, get_value, woken) |
+ |
+ # wake them all up |
+ cond.acquire() |
+ cond.notify_all() |
+ cond.release() |
+ |
+ # check they have all woken |
+ time.sleep(DELTA) |
+ self.assertReturnsIfImplemented(6, get_value, woken) |
+ |
+ # check state is not mucked up |
+ self.check_invariant(cond) |
+ |
+ def test_timeout(self): |
+ cond = self.Condition() |
+ wait = TimingWrapper(cond.wait) |
+ cond.acquire() |
+ res = wait(TIMEOUT1) |
+ cond.release() |
+ self.assertEqual(res, None) |
+ self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1) |
+ |
+ |
+class _TestEvent(BaseTestCase): |
+ |
+ def _test_event(self, event): |
+ time.sleep(TIMEOUT2) |
+ event.set() |
+ |
+ def test_event(self): |
+ event = self.Event() |
+ wait = TimingWrapper(event.wait) |
+ |
+ # Removed temporaily, due to API shear, this does not |
+ # work with threading._Event objects. is_set == isSet |
+ #self.assertEqual(event.is_set(), False) |
+ |
+ self.assertEqual(wait(0.0), None) |
+ self.assertTimingAlmostEqual(wait.elapsed, 0.0) |
+ self.assertEqual(wait(TIMEOUT1), None) |
+ self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1) |
+ |
+ event.set() |
+ |
+ # See note above on the API differences |
+ # self.assertEqual(event.is_set(), True) |
+ self.assertEqual(wait(), None) |
+ self.assertTimingAlmostEqual(wait.elapsed, 0.0) |
+ self.assertEqual(wait(TIMEOUT1), None) |
+ self.assertTimingAlmostEqual(wait.elapsed, 0.0) |
+ # self.assertEqual(event.is_set(), True) |
+ |
+ event.clear() |
+ |
+ #self.assertEqual(event.is_set(), False) |
+ |
+ self.Process(target=self._test_event, args=(event,)).start() |
+ self.assertEqual(wait(), None) |
+ |
+# |
+# |
+# |
+ |
+class _TestValue(BaseTestCase): |
+ |
+ codes_values = [ |
+ ('i', 4343, 24234), |
+ ('d', 3.625, -4.25), |
+ ('h', -232, 234), |
+ ('c', latin('x'), latin('y')) |
+ ] |
+ |
+ def _test(self, values): |
+ for sv, cv in zip(values, self.codes_values): |
+ sv.value = cv[2] |
+ |
+ |
+ def test_value(self, raw=False): |
+ if self.TYPE != 'processes': |
+ return |
+ |
+ if raw: |
+ values = [self.RawValue(code, value) |
+ for code, value, _ in self.codes_values] |
+ else: |
+ values = [self.Value(code, value) |
+ for code, value, _ in self.codes_values] |
+ |
+ for sv, cv in zip(values, self.codes_values): |
+ self.assertEqual(sv.value, cv[1]) |
+ |
+ proc = self.Process(target=self._test, args=(values,)) |
+ proc.start() |
+ proc.join() |
+ |
+ for sv, cv in zip(values, self.codes_values): |
+ self.assertEqual(sv.value, cv[2]) |
+ |
+ def test_rawvalue(self): |
+ self.test_value(raw=True) |
+ |
+ def test_getobj_getlock(self): |
+ if self.TYPE != 'processes': |
+ return |
+ |
+ val1 = self.Value('i', 5) |
+ lock1 = val1.get_lock() |
+ obj1 = val1.get_obj() |
+ |
+ val2 = self.Value('i', 5, lock=None) |
+ lock2 = val2.get_lock() |
+ obj2 = val2.get_obj() |
+ |
+ lock = self.Lock() |
+ val3 = self.Value('i', 5, lock=lock) |
+ lock3 = val3.get_lock() |
+ obj3 = val3.get_obj() |
+ self.assertEqual(lock, lock3) |
+ |
+ arr4 = self.RawValue('i', 5) |
+ self.assertFalse(hasattr(arr4, 'get_lock')) |
+ self.assertFalse(hasattr(arr4, 'get_obj')) |
+ |
+ |
+class _TestArray(BaseTestCase): |
+ |
+ def f(self, seq): |
+ for i in range(1, len(seq)): |
+ seq[i] += seq[i-1] |
+ |
+ def test_array(self, raw=False): |
+ if self.TYPE != 'processes': |
+ return |
+ |
+ seq = [680, 626, 934, 821, 150, 233, 548, 982, 714, 831] |
+ if raw: |
+ arr = self.RawArray('i', seq) |
+ else: |
+ arr = self.Array('i', seq) |
+ |
+ self.assertEqual(len(arr), len(seq)) |
+ self.assertEqual(arr[3], seq[3]) |
+ self.assertEqual(list(arr[2:7]), list(seq[2:7])) |
+ |
+ arr[4:8] = seq[4:8] = array.array('i', [1, 2, 3, 4]) |
+ |
+ self.assertEqual(list(arr[:]), seq) |
+ |
+ self.f(seq) |
+ |
+ p = self.Process(target=self.f, args=(arr,)) |
+ p.start() |
+ p.join() |
+ |
+ self.assertEqual(list(arr[:]), seq) |
+ |
+ def test_rawarray(self): |
+ self.test_array(raw=True) |
+ |
+ def test_getobj_getlock_obj(self): |
+ if self.TYPE != 'processes': |
+ return |
+ |
+ arr1 = self.Array('i', range(10)) |
+ lock1 = arr1.get_lock() |
+ obj1 = arr1.get_obj() |
+ |
+ arr2 = self.Array('i', range(10), lock=None) |
+ lock2 = arr2.get_lock() |
+ obj2 = arr2.get_obj() |
+ |
+ lock = self.Lock() |
+ arr3 = self.Array('i', range(10), lock=lock) |
+ lock3 = arr3.get_lock() |
+ obj3 = arr3.get_obj() |
+ self.assertEqual(lock, lock3) |
+ |
+ arr4 = self.RawArray('i', range(10)) |
+ self.assertFalse(hasattr(arr4, 'get_lock')) |
+ self.assertFalse(hasattr(arr4, 'get_obj')) |
+ |
+# |
+# |
+# |
+ |
+class _TestContainers(BaseTestCase): |
+ |
+ ALLOWED_TYPES = ('manager',) |
+ |
+ def test_list(self): |
+ a = self.list(range(10)) |
+ self.assertEqual(a[:], range(10)) |
+ |
+ b = self.list() |
+ self.assertEqual(b[:], []) |
+ |
+ b.extend(range(5)) |
+ self.assertEqual(b[:], range(5)) |
+ |
+ self.assertEqual(b[2], 2) |
+ self.assertEqual(b[2:10], [2,3,4]) |
+ |
+ b *= 2 |
+ self.assertEqual(b[:], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4]) |
+ |
+ self.assertEqual(b + [5, 6], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4, 5, 6]) |
+ |
+ self.assertEqual(a[:], range(10)) |
+ |
+ d = [a, b] |
+ e = self.list(d) |
+ self.assertEqual( |
+ e[:], |
+ [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4]] |
+ ) |
+ |
+ f = self.list([a]) |
+ a.append('hello') |
+ self.assertEqual(f[:], [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 'hello']]) |
+ |
+ def test_dict(self): |
+ d = self.dict() |
+ indices = range(65, 70) |
+ for i in indices: |
+ d[i] = chr(i) |
+ self.assertEqual(d.copy(), dict((i, chr(i)) for i in indices)) |
+ self.assertEqual(sorted(d.keys()), indices) |
+ self.assertEqual(sorted(d.values()), [chr(i) for i in indices]) |
+ self.assertEqual(sorted(d.items()), [(i, chr(i)) for i in indices]) |
+ |
+ def test_namespace(self): |
+ n = self.Namespace() |
+ n.name = 'Bob' |
+ n.job = 'Builder' |
+ n._hidden = 'hidden' |
+ self.assertEqual((n.name, n.job), ('Bob', 'Builder')) |
+ del n.job |
+ self.assertEqual(str(n), "Namespace(name='Bob')") |
+ self.assertTrue(hasattr(n, 'name')) |
+ self.assertTrue(not hasattr(n, 'job')) |
+ |
+# |
+# |
+# |
+ |
+def sqr(x, wait=0.0): |
+ time.sleep(wait) |
+ return x*x |
+ |
+class _TestPool(BaseTestCase): |
+ |
+ def test_apply(self): |
+ papply = self.pool.apply |
+ self.assertEqual(papply(sqr, (5,)), sqr(5)) |
+ self.assertEqual(papply(sqr, (), {'x':3}), sqr(x=3)) |
+ |
+ def test_map(self): |
+ pmap = self.pool.map |
+ self.assertEqual(pmap(sqr, range(10)), map(sqr, range(10))) |
+ self.assertEqual(pmap(sqr, range(100), chunksize=20), |
+ map(sqr, range(100))) |
+ |
+ def test_async(self): |
+ res = self.pool.apply_async(sqr, (7, TIMEOUT1,)) |
+ get = TimingWrapper(res.get) |
+ self.assertEqual(get(), 49) |
+ self.assertTimingAlmostEqual(get.elapsed, TIMEOUT1) |
+ |
+ def test_async_timeout(self): |
+ res = self.pool.apply_async(sqr, (6, TIMEOUT2 + 0.2)) |
+ get = TimingWrapper(res.get) |
+ self.assertRaises(multiprocessing.TimeoutError, get, timeout=TIMEOUT2) |
+ self.assertTimingAlmostEqual(get.elapsed, TIMEOUT2) |
+ |
+ def test_imap(self): |
+ it = self.pool.imap(sqr, range(10)) |
+ self.assertEqual(list(it), map(sqr, range(10))) |
+ |
+ it = self.pool.imap(sqr, range(10)) |
+ for i in range(10): |
+ self.assertEqual(it.next(), i*i) |
+ self.assertRaises(StopIteration, it.next) |
+ |
+ it = self.pool.imap(sqr, range(1000), chunksize=100) |
+ for i in range(1000): |
+ self.assertEqual(it.next(), i*i) |
+ self.assertRaises(StopIteration, it.next) |
+ |
+ def test_imap_unordered(self): |
+ it = self.pool.imap_unordered(sqr, range(1000)) |
+ self.assertEqual(sorted(it), map(sqr, range(1000))) |
+ |
+ it = self.pool.imap_unordered(sqr, range(1000), chunksize=53) |
+ self.assertEqual(sorted(it), map(sqr, range(1000))) |
+ |
+ def test_make_pool(self): |
+ p = multiprocessing.Pool(3) |
+ self.assertEqual(3, len(p._pool)) |
+ p.close() |
+ p.join() |
+ |
+ def test_terminate(self): |
+ if self.TYPE == 'manager': |
+ # On Unix a forked process increfs each shared object to |
+ # which its parent process held a reference. If the |
+ # forked process gets terminated then there is likely to |
+ # be a reference leak. So to prevent |
+ # _TestZZZNumberOfObjects from failing we skip this test |
+ # when using a manager. |
+ return |
+ |
+ result = self.pool.map_async( |
+ time.sleep, [0.1 for i in range(10000)], chunksize=1 |
+ ) |
+ self.pool.terminate() |
+ join = TimingWrapper(self.pool.join) |
+ join() |
+ self.assertTrue(join.elapsed < 0.2) |
+ |
+# |
+# Test that manager has expected number of shared objects left |
+# |
+ |
+class _TestZZZNumberOfObjects(BaseTestCase): |
+ # Because test cases are sorted alphabetically, this one will get |
+ # run after all the other tests for the manager. It tests that |
+ # there have been no "reference leaks" for the manager's shared |
+ # objects. Note the comment in _TestPool.test_terminate(). |
+ ALLOWED_TYPES = ('manager',) |
+ |
+ def test_number_of_objects(self): |
+ EXPECTED_NUMBER = 1 # the pool object is still alive |
+ multiprocessing.active_children() # discard dead process objs |
+ gc.collect() # do garbage collection |
+ refs = self.manager._number_of_objects() |
+ if refs != EXPECTED_NUMBER: |
+ print self.manager._debugInfo() |
+ |
+ self.assertEqual(refs, EXPECTED_NUMBER) |
+ |
+# |
+# Test of creating a customized manager class |
+# |
+ |
+from multiprocessing.managers import BaseManager, BaseProxy, RemoteError |
+ |
+class FooBar(object): |
+ def f(self): |
+ return 'f()' |
+ def g(self): |
+ raise ValueError |
+ def _h(self): |
+ return '_h()' |
+ |
+def baz(): |
+ for i in xrange(10): |
+ yield i*i |
+ |
+class IteratorProxy(BaseProxy): |
+ _exposed_ = ('next', '__next__') |
+ def __iter__(self): |
+ return self |
+ def next(self): |
+ return self._callmethod('next') |
+ def __next__(self): |
+ return self._callmethod('__next__') |
+ |
+class MyManager(BaseManager): |
+ pass |
+ |
+MyManager.register('Foo', callable=FooBar) |
+MyManager.register('Bar', callable=FooBar, exposed=('f', '_h')) |
+MyManager.register('baz', callable=baz, proxytype=IteratorProxy) |
+ |
+ |
+class _TestMyManager(BaseTestCase): |
+ |
+ ALLOWED_TYPES = ('manager',) |
+ |
+ def test_mymanager(self): |
+ manager = MyManager() |
+ manager.start() |
+ |
+ foo = manager.Foo() |
+ bar = manager.Bar() |
+ baz = manager.baz() |
+ |
+ foo_methods = [name for name in ('f', 'g', '_h') if hasattr(foo, name)] |
+ bar_methods = [name for name in ('f', 'g', '_h') if hasattr(bar, name)] |
+ |
+ self.assertEqual(foo_methods, ['f', 'g']) |
+ self.assertEqual(bar_methods, ['f', '_h']) |
+ |
+ self.assertEqual(foo.f(), 'f()') |
+ self.assertRaises(ValueError, foo.g) |
+ self.assertEqual(foo._callmethod('f'), 'f()') |
+ self.assertRaises(RemoteError, foo._callmethod, '_h') |
+ |
+ self.assertEqual(bar.f(), 'f()') |
+ self.assertEqual(bar._h(), '_h()') |
+ self.assertEqual(bar._callmethod('f'), 'f()') |
+ self.assertEqual(bar._callmethod('_h'), '_h()') |
+ |
+ self.assertEqual(list(baz), [i*i for i in range(10)]) |
+ |
+ manager.shutdown() |
+ |
+# |
+# Test of connecting to a remote server and using xmlrpclib for serialization |
+# |
+ |
+_queue = Queue.Queue() |
+def get_queue(): |
+ return _queue |
+ |
+class QueueManager(BaseManager): |
+ '''manager class used by server process''' |
+QueueManager.register('get_queue', callable=get_queue) |
+ |
+class QueueManager2(BaseManager): |
+ '''manager class which specifies the same interface as QueueManager''' |
+QueueManager2.register('get_queue') |
+ |
+ |
+SERIALIZER = 'xmlrpclib' |
+ |
+class _TestRemoteManager(BaseTestCase): |
+ |
+ ALLOWED_TYPES = ('manager',) |
+ |
+ def _putter(self, address, authkey): |
+ manager = QueueManager2( |
+ address=address, authkey=authkey, serializer=SERIALIZER |
+ ) |
+ manager.connect() |
+ queue = manager.get_queue() |
+ queue.put(('hello world', None, True, 2.25)) |
+ |
+ def test_remote(self): |
+ authkey = os.urandom(32) |
+ |
+ manager = QueueManager( |
+ address=('localhost', 0), authkey=authkey, serializer=SERIALIZER |
+ ) |
+ manager.start() |
+ |
+ p = self.Process(target=self._putter, args=(manager.address, authkey)) |
+ p.start() |
+ |
+ manager2 = QueueManager2( |
+ address=manager.address, authkey=authkey, serializer=SERIALIZER |
+ ) |
+ manager2.connect() |
+ queue = manager2.get_queue() |
+ |
+ # Note that xmlrpclib will deserialize object as a list not a tuple |
+ self.assertEqual(queue.get(), ['hello world', None, True, 2.25]) |
+ |
+ # Because we are using xmlrpclib for serialization instead of |
+ # pickle this will cause a serialization error. |
+ self.assertRaises(Exception, queue.put, time.sleep) |
+ |
+ # Make queue finalizer run before the server is stopped |
+ del queue |
+ manager.shutdown() |
+ |
+# |
+# |
+# |
+ |
+SENTINEL = latin('') |
+ |
+class _TestConnection(BaseTestCase): |
+ |
+ ALLOWED_TYPES = ('processes', 'threads') |
+ |
+ def _echo(self, conn): |
+ for msg in iter(conn.recv_bytes, SENTINEL): |
+ conn.send_bytes(msg) |
+ conn.close() |
+ |
+ def test_connection(self): |
+ conn, child_conn = self.Pipe() |
+ |
+ p = self.Process(target=self._echo, args=(child_conn,)) |
+ p.set_daemon(True) |
+ p.start() |
+ |
+ seq = [1, 2.25, None] |
+ msg = latin('hello world') |
+ longmsg = msg * 10 |
+ arr = array.array('i', range(4)) |
+ |
+ if self.TYPE == 'processes': |
+ self.assertEqual(type(conn.fileno()), int) |
+ |
+ self.assertEqual(conn.send(seq), None) |
+ self.assertEqual(conn.recv(), seq) |
+ |
+ self.assertEqual(conn.send_bytes(msg), None) |
+ self.assertEqual(conn.recv_bytes(), msg) |
+ |
+ if self.TYPE == 'processes': |
+ buffer = array.array('i', [0]*10) |
+ expected = list(arr) + [0] * (10 - len(arr)) |
+ self.assertEqual(conn.send_bytes(arr), None) |
+ self.assertEqual(conn.recv_bytes_into(buffer), |
+ len(arr) * buffer.itemsize) |
+ self.assertEqual(list(buffer), expected) |
+ |
+ buffer = array.array('i', [0]*10) |
+ expected = [0] * 3 + list(arr) + [0] * (10 - 3 - len(arr)) |
+ self.assertEqual(conn.send_bytes(arr), None) |
+ self.assertEqual(conn.recv_bytes_into(buffer, 3 * buffer.itemsize), |
+ len(arr) * buffer.itemsize) |
+ self.assertEqual(list(buffer), expected) |
+ |
+ buffer = bytearray(latin(' ' * 40)) |
+ self.assertEqual(conn.send_bytes(longmsg), None) |
+ try: |
+ res = conn.recv_bytes_into(buffer) |
+ except multiprocessing.BufferTooShort, e: |
+ self.assertEqual(e.args, (longmsg,)) |
+ else: |
+ self.fail('expected BufferTooShort, got %s' % res) |
+ |
+ poll = TimingWrapper(conn.poll) |
+ |
+ self.assertEqual(poll(), False) |
+ self.assertTimingAlmostEqual(poll.elapsed, 0) |
+ |
+ self.assertEqual(poll(TIMEOUT1), False) |
+ self.assertTimingAlmostEqual(poll.elapsed, TIMEOUT1) |
+ |
+ conn.send(None) |
+ |
+ self.assertEqual(poll(TIMEOUT1), True) |
+ self.assertTimingAlmostEqual(poll.elapsed, 0) |
+ |
+ self.assertEqual(conn.recv(), None) |
+ |
+ really_big_msg = latin('X') * (1024 * 1024 * 16) # 16Mb |
+ conn.send_bytes(really_big_msg) |
+ self.assertEqual(conn.recv_bytes(), really_big_msg) |
+ |
+ conn.send_bytes(SENTINEL) # tell child to quit |
+ child_conn.close() |
+ |
+ if self.TYPE == 'processes': |
+ self.assertEqual(conn.readable, True) |
+ self.assertEqual(conn.writable, True) |
+ self.assertRaises(EOFError, conn.recv) |
+ self.assertRaises(EOFError, conn.recv_bytes) |
+ |
+ p.join() |
+ |
+ def test_duplex_false(self): |
+ reader, writer = self.Pipe(duplex=False) |
+ self.assertEqual(writer.send(1), None) |
+ self.assertEqual(reader.recv(), 1) |
+ if self.TYPE == 'processes': |
+ self.assertEqual(reader.readable, True) |
+ self.assertEqual(reader.writable, False) |
+ self.assertEqual(writer.readable, False) |
+ self.assertEqual(writer.writable, True) |
+ self.assertRaises(IOError, reader.send, 2) |
+ self.assertRaises(IOError, writer.recv) |
+ self.assertRaises(IOError, writer.poll) |
+ |
+ def test_spawn_close(self): |
+ # We test that a pipe connection can be closed by parent |
+ # process immediately after child is spawned. On Windows this |
+ # would have sometimes failed on old versions because |
+ # child_conn would be closed before the child got a chance to |
+ # duplicate it. |
+ conn, child_conn = self.Pipe() |
+ |
+ p = self.Process(target=self._echo, args=(child_conn,)) |
+ p.start() |
+ child_conn.close() # this might complete before child initializes |
+ |
+ msg = latin('hello') |
+ conn.send_bytes(msg) |
+ self.assertEqual(conn.recv_bytes(), msg) |
+ |
+ conn.send_bytes(SENTINEL) |
+ conn.close() |
+ p.join() |
+ |
+ def test_sendbytes(self): |
+ if self.TYPE != 'processes': |
+ return |
+ |
+ msg = latin('abcdefghijklmnopqrstuvwxyz') |
+ a, b = self.Pipe() |
+ |
+ a.send_bytes(msg) |
+ self.assertEqual(b.recv_bytes(), msg) |
+ |
+ a.send_bytes(msg, 5) |
+ self.assertEqual(b.recv_bytes(), msg[5:]) |
+ |
+ a.send_bytes(msg, 7, 8) |
+ self.assertEqual(b.recv_bytes(), msg[7:7+8]) |
+ |
+ a.send_bytes(msg, 26) |
+ self.assertEqual(b.recv_bytes(), latin('')) |
+ |
+ a.send_bytes(msg, 26, 0) |
+ self.assertEqual(b.recv_bytes(), latin('')) |
+ |
+ self.assertRaises(ValueError, a.send_bytes, msg, 27) |
+ |
+ self.assertRaises(ValueError, a.send_bytes, msg, 22, 5) |
+ |
+ self.assertRaises(ValueError, a.send_bytes, msg, 26, 1) |
+ |
+ self.assertRaises(ValueError, a.send_bytes, msg, -1) |
+ |
+ self.assertRaises(ValueError, a.send_bytes, msg, 4, -1) |
+ |
+ |
+class _TestListenerClient(BaseTestCase): |
+ |
+ ALLOWED_TYPES = ('processes', 'threads') |
+ |
+ def _test(self, address): |
+ conn = self.connection.Client(address) |
+ conn.send('hello') |
+ conn.close() |
+ |
+ def test_listener_client(self): |
+ for family in self.connection.families: |
+ l = self.connection.Listener(family=family) |
+ p = self.Process(target=self._test, args=(l.address,)) |
+ p.set_daemon(True) |
+ p.start() |
+ conn = l.accept() |
+ self.assertEqual(conn.recv(), 'hello') |
+ p.join() |
+ l.close() |
+ |
+# |
+# Test of sending connection and socket objects between processes |
+# |
+ |
+class _TestPicklingConnections(BaseTestCase): |
+ |
+ ALLOWED_TYPES = ('processes',) |
+ |
+ def _listener(self, conn, families): |
+ for fam in families: |
+ l = self.connection.Listener(family=fam) |
+ conn.send(l.address) |
+ new_conn = l.accept() |
+ conn.send(new_conn) |
+ |
+ if self.TYPE == 'processes': |
+ l = socket.socket() |
+ l.bind(('localhost', 0)) |
+ conn.send(l.getsockname()) |
+ l.listen(1) |
+ new_conn, addr = l.accept() |
+ conn.send(new_conn) |
+ |
+ conn.recv() |
+ |
+ def _remote(self, conn): |
+ for (address, msg) in iter(conn.recv, None): |
+ client = self.connection.Client(address) |
+ client.send(msg.upper()) |
+ client.close() |
+ |
+ if self.TYPE == 'processes': |
+ address, msg = conn.recv() |
+ client = socket.socket() |
+ client.connect(address) |
+ client.sendall(msg.upper()) |
+ client.close() |
+ |
+ conn.close() |
+ |
+ def test_pickling(self): |
+ try: |
+ multiprocessing.allow_connection_pickling() |
+ except ImportError: |
+ return |
+ |
+ families = self.connection.families |
+ |
+ lconn, lconn0 = self.Pipe() |
+ lp = self.Process(target=self._listener, args=(lconn0, families)) |
+ lp.start() |
+ lconn0.close() |
+ |
+ rconn, rconn0 = self.Pipe() |
+ rp = self.Process(target=self._remote, args=(rconn0,)) |
+ rp.start() |
+ rconn0.close() |
+ |
+ for fam in families: |
+ msg = ('This connection uses family %s' % fam).encode('ascii') |
+ address = lconn.recv() |
+ rconn.send((address, msg)) |
+ new_conn = lconn.recv() |
+ self.assertEqual(new_conn.recv(), msg.upper()) |
+ |
+ rconn.send(None) |
+ |
+ if self.TYPE == 'processes': |
+ msg = latin('This connection uses a normal socket') |
+ address = lconn.recv() |
+ rconn.send((address, msg)) |
+ if hasattr(socket, 'fromfd'): |
+ new_conn = lconn.recv() |
+ self.assertEqual(new_conn.recv(100), msg.upper()) |
+ else: |
+ # XXX On Windows with Py2.6 need to backport fromfd() |
+ discard = lconn.recv_bytes() |
+ |
+ lconn.send(None) |
+ |
+ rconn.close() |
+ lconn.close() |
+ |
+ lp.join() |
+ rp.join() |
+ |
+# |
+# |
+# |
+ |
+class _TestHeap(BaseTestCase): |
+ |
+ ALLOWED_TYPES = ('processes',) |
+ |
+ def test_heap(self): |
+ iterations = 5000 |
+ maxblocks = 50 |
+ blocks = [] |
+ |
+ # create and destroy lots of blocks of different sizes |
+ for i in xrange(iterations): |
+ size = int(random.lognormvariate(0, 1) * 1000) |
+ b = multiprocessing.heap.BufferWrapper(size) |
+ blocks.append(b) |
+ if len(blocks) > maxblocks: |
+ i = random.randrange(maxblocks) |
+ del blocks[i] |
+ |
+ # get the heap object |
+ heap = multiprocessing.heap.BufferWrapper._heap |
+ |
+ # verify the state of the heap |
+ all = [] |
+ occupied = 0 |
+ for L in heap._len_to_seq.values(): |
+ for arena, start, stop in L: |
+ all.append((heap._arenas.index(arena), start, stop, |
+ stop-start, 'free')) |
+ for arena, start, stop in heap._allocated_blocks: |
+ all.append((heap._arenas.index(arena), start, stop, |
+ stop-start, 'occupied')) |
+ occupied += (stop-start) |
+ |
+ all.sort() |
+ |
+ for i in range(len(all)-1): |
+ (arena, start, stop) = all[i][:3] |
+ (narena, nstart, nstop) = all[i+1][:3] |
+ self.assertTrue((arena != narena and nstart == 0) or |
+ (stop == nstart)) |
+ |
+# |
+# |
+# |
+ |
+try: |
+ from ctypes import Structure, Value, copy, c_int, c_double |
+except ImportError: |
+ Structure = object |
+ c_int = c_double = None |
+ |
+class _Foo(Structure): |
+ _fields_ = [ |
+ ('x', c_int), |
+ ('y', c_double) |
+ ] |
+ |
+class _TestSharedCTypes(BaseTestCase): |
+ |
+ ALLOWED_TYPES = ('processes',) |
+ |
+ def _double(self, x, y, foo, arr, string): |
+ x.value *= 2 |
+ y.value *= 2 |
+ foo.x *= 2 |
+ foo.y *= 2 |
+ string.value *= 2 |
+ for i in range(len(arr)): |
+ arr[i] *= 2 |
+ |
+ def test_sharedctypes(self, lock=False): |
+ if c_int is None: |
+ return |
+ |
+ x = Value('i', 7, lock=lock) |
+ y = Value(ctypes.c_double, 1.0/3.0, lock=lock) |
+ foo = Value(_Foo, 3, 2, lock=lock) |
+ arr = Array('d', range(10), lock=lock) |
+ string = Array('c', 20, lock=lock) |
+ string.value = 'hello' |
+ |
+ p = self.Process(target=self._double, args=(x, y, foo, arr, string)) |
+ p.start() |
+ p.join() |
+ |
+ self.assertEqual(x.value, 14) |
+ self.assertAlmostEqual(y.value, 2.0/3.0) |
+ self.assertEqual(foo.x, 6) |
+ self.assertAlmostEqual(foo.y, 4.0) |
+ for i in range(10): |
+ self.assertAlmostEqual(arr[i], i*2) |
+ self.assertEqual(string.value, latin('hellohello')) |
+ |
+ def test_synchronize(self): |
+ self.test_sharedctypes(lock=True) |
+ |
+ def test_copy(self): |
+ if c_int is None: |
+ return |
+ |
+ foo = _Foo(2, 5.0) |
+ bar = copy(foo) |
+ foo.x = 0 |
+ foo.y = 0 |
+ self.assertEqual(bar.x, 2) |
+ self.assertAlmostEqual(bar.y, 5.0) |
+ |
+# |
+# |
+# |
+ |
+class _TestFinalize(BaseTestCase): |
+ |
+ ALLOWED_TYPES = ('processes',) |
+ |
+ def _test_finalize(self, conn): |
+ class Foo(object): |
+ pass |
+ |
+ a = Foo() |
+ util.Finalize(a, conn.send, args=('a',)) |
+ del a # triggers callback for a |
+ |
+ b = Foo() |
+ close_b = util.Finalize(b, conn.send, args=('b',)) |
+ close_b() # triggers callback for b |
+ close_b() # does nothing because callback has already been called |
+ del b # does nothing because callback has already been called |
+ |
+ c = Foo() |
+ util.Finalize(c, conn.send, args=('c',)) |
+ |
+ d10 = Foo() |
+ util.Finalize(d10, conn.send, args=('d10',), exitpriority=1) |
+ |
+ d01 = Foo() |
+ util.Finalize(d01, conn.send, args=('d01',), exitpriority=0) |
+ d02 = Foo() |
+ util.Finalize(d02, conn.send, args=('d02',), exitpriority=0) |
+ d03 = Foo() |
+ util.Finalize(d03, conn.send, args=('d03',), exitpriority=0) |
+ |
+ util.Finalize(None, conn.send, args=('e',), exitpriority=-10) |
+ |
+ util.Finalize(None, conn.send, args=('STOP',), exitpriority=-100) |
+ |
+ # call mutliprocessing's cleanup function then exit process without |
+ # garbage collecting locals |
+ util._exit_function() |
+ conn.close() |
+ os._exit(0) |
+ |
+ def test_finalize(self): |
+ conn, child_conn = self.Pipe() |
+ |
+ p = self.Process(target=self._test_finalize, args=(child_conn,)) |
+ p.start() |
+ p.join() |
+ |
+ result = [obj for obj in iter(conn.recv, 'STOP')] |
+ self.assertEqual(result, ['a', 'b', 'd10', 'd03', 'd02', 'd01', 'e']) |
+ |
+# |
+# Test that from ... import * works for each module |
+# |
+ |
+class _TestImportStar(BaseTestCase): |
+ |
+ ALLOWED_TYPES = ('processes',) |
+ |
+ def test_import(self): |
+ modules = ( |
+ 'multiprocessing', 'multiprocessing.connection', |
+ 'multiprocessing.heap', 'multiprocessing.managers', |
+ 'multiprocessing.pool', 'multiprocessing.process', |
+ 'multiprocessing.reduction', 'multiprocessing.sharedctypes', |
+ 'multiprocessing.synchronize', 'multiprocessing.util' |
+ ) |
+ |
+ for name in modules: |
+ __import__(name) |
+ mod = sys.modules[name] |
+ |
+ for attr in getattr(mod, '__all__', ()): |
+ self.assertTrue( |
+ hasattr(mod, attr), |
+ '%r does not have attribute %r' % (mod, attr) |
+ ) |
+ |
+# |
+# Quick test that logging works -- does not test logging output |
+# |
+ |
+class _TestLogging(BaseTestCase): |
+ |
+ ALLOWED_TYPES = ('processes',) |
+ |
+ def test_enable_logging(self): |
+ logger = multiprocessing.get_logger() |
+ logger.setLevel(util.SUBWARNING) |
+ self.assertTrue(logger is not None) |
+ logger.debug('this will not be printed') |
+ logger.info('nor will this') |
+ logger.setLevel(LOG_LEVEL) |
+ |
+ def _test_level(self, conn): |
+ logger = multiprocessing.get_logger() |
+ conn.send(logger.getEffectiveLevel()) |
+ |
+ def test_level(self): |
+ LEVEL1 = 32 |
+ LEVEL2 = 37 |
+ |
+ logger = multiprocessing.get_logger() |
+ root_logger = logging.getLogger() |
+ root_level = root_logger.level |
+ |
+ reader, writer = multiprocessing.Pipe(duplex=False) |
+ |
+ logger.setLevel(LEVEL1) |
+ self.Process(target=self._test_level, args=(writer,)).start() |
+ self.assertEqual(LEVEL1, reader.recv()) |
+ |
+ logger.setLevel(logging.NOTSET) |
+ root_logger.setLevel(LEVEL2) |
+ self.Process(target=self._test_level, args=(writer,)).start() |
+ self.assertEqual(LEVEL2, reader.recv()) |
+ |
+ root_logger.setLevel(root_level) |
+ logger.setLevel(level=LOG_LEVEL) |
+ |
+# |
+# Functions used to create test cases from the base ones in this module |
+# |
+ |
+def get_attributes(Source, names): |
+ d = {} |
+ for name in names: |
+ obj = getattr(Source, name) |
+ if type(obj) == type(get_attributes): |
+ obj = staticmethod(obj) |
+ d[name] = obj |
+ return d |
+ |
+def create_test_cases(Mixin, type): |
+ result = {} |
+ glob = globals() |
+ Type = type[0].upper() + type[1:] |
+ |
+ for name in glob.keys(): |
+ if name.startswith('_Test'): |
+ base = glob[name] |
+ if type in base.ALLOWED_TYPES: |
+ newname = 'With' + Type + name[1:] |
+ class Temp(base, unittest.TestCase, Mixin): |
+ pass |
+ result[newname] = Temp |
+ Temp.__name__ = newname |
+ Temp.__module__ = Mixin.__module__ |
+ return result |
+ |
+# |
+# Create test cases |
+# |
+ |
+class ProcessesMixin(object): |
+ TYPE = 'processes' |
+ Process = multiprocessing.Process |
+ locals().update(get_attributes(multiprocessing, ( |
+ 'Queue', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', |
+ 'Condition', 'Event', 'Value', 'Array', 'RawValue', |
+ 'RawArray', 'current_process', 'active_children', 'Pipe', |
+ 'connection', 'JoinableQueue' |
+ ))) |
+ |
+testcases_processes = create_test_cases(ProcessesMixin, type='processes') |
+globals().update(testcases_processes) |
+ |
+ |
+class ManagerMixin(object): |
+ TYPE = 'manager' |
+ Process = multiprocessing.Process |
+ manager = object.__new__(multiprocessing.managers.SyncManager) |
+ locals().update(get_attributes(manager, ( |
+ 'Queue', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', |
+ 'Condition', 'Event', 'Value', 'Array', 'list', 'dict', |
+ 'Namespace', 'JoinableQueue' |
+ ))) |
+ |
+testcases_manager = create_test_cases(ManagerMixin, type='manager') |
+globals().update(testcases_manager) |
+ |
+ |
+class ThreadsMixin(object): |
+ TYPE = 'threads' |
+ Process = multiprocessing.dummy.Process |
+ locals().update(get_attributes(multiprocessing.dummy, ( |
+ 'Queue', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', |
+ 'Condition', 'Event', 'Value', 'Array', 'current_process', |
+ 'active_children', 'Pipe', 'connection', 'dict', 'list', |
+ 'Namespace', 'JoinableQueue' |
+ ))) |
+ |
+testcases_threads = create_test_cases(ThreadsMixin, type='threads') |
+globals().update(testcases_threads) |
+ |
+# |
+# |
+# |
+ |
+def test_main(run=None): |
+ if run is None: |
+ from test.test_support import run_unittest as run |
+ |
+ util.get_temp_dir() # creates temp directory for use by all processes |
+ |
+ multiprocessing.get_logger().setLevel(LOG_LEVEL) |
+ |
+ ProcessesMixin.pool = multiprocessing.Pool(4) |
+ ThreadsMixin.pool = multiprocessing.dummy.Pool(4) |
+ ManagerMixin.manager.__init__() |
+ ManagerMixin.manager.start() |
+ ManagerMixin.pool = ManagerMixin.manager.Pool(4) |
+ |
+ testcases = ( |
+ sorted(testcases_processes.values(), key=lambda tc:tc.__name__) + |
+ sorted(testcases_threads.values(), key=lambda tc:tc.__name__) + |
+ sorted(testcases_manager.values(), key=lambda tc:tc.__name__) |
+ ) |
+ |
+ loadTestsFromTestCase = unittest.defaultTestLoader.loadTestsFromTestCase |
+ suite = unittest.TestSuite(loadTestsFromTestCase(tc) for tc in testcases) |
+ run(suite) |
+ |
+ ThreadsMixin.pool.terminate() |
+ ProcessesMixin.pool.terminate() |
+ ManagerMixin.pool.terminate() |
+ ManagerMixin.manager.shutdown() |
+ |
+ del ProcessesMixin.pool, ThreadsMixin.pool, ManagerMixin.pool |
+ |
+def main(): |
+ test_main(unittest.TextTestRunner(verbosity=2).run) |
+ |
+if __name__ == '__main__': |
+ main() |
Property changes on: Lib/test/test_multiprocessing.py |
___________________________________________________________________ |
Name: svn:executable |
+ * |