add the multiprocessing package to fulfill PEP 371
diff --git a/Lib/test/test_multiprocessing.py b/Lib/test/test_multiprocessing.py
new file mode 100644
index 0000000..523809b
--- /dev/null
+++ b/Lib/test/test_multiprocessing.py
@@ -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()