1 #!/usr/bin/env python

     2 

     3 #

     4 # Unit tests for the multiprocessing package

     5 #

     6 

     7 import unittest

     8 import threading

     9 import Queue

    10 import time

    11 import sys

    12 import os

    13 import gc

    14 import signal

    15 import array

    16 import copy

    17 import socket

    18 import random

    19 import logging

    20 

    21 

    22 # Work around broken sem_open implementations

    23 try:

    24     import multiprocessing.synchronize

    25 except ImportError, e:

    26     from test.test_support import TestSkipped

    27     raise TestSkipped(e)

    28 

    29 import multiprocessing.dummy

    30 import multiprocessing.connection

    31 import multiprocessing.managers

    32 import multiprocessing.heap

    33 import multiprocessing.pool

    34 import _multiprocessing

    35 

    36 from multiprocessing import util

    37 

    38 #

    39 #

    40 #

    41 

    42 latin = str

    43 

    44 #

    45 # Constants

    46 #

    47 

    48 LOG_LEVEL = util.SUBWARNING

    49 #LOG_LEVEL = logging.WARNING

    50 

    51 DELTA = 0.1

    52 CHECK_TIMINGS = False     # making true makes tests take a lot longer

    53                           # and can sometimes cause some non-serious

    54                           # failures because some calls block a bit

    55                           # longer than expected

    56 if CHECK_TIMINGS:

    57     TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.82, 0.35, 1.4

    58 else:

    59     TIMEOUT1, TIMEOUT2, TIMEOUT3 = 0.1, 0.1, 0.1

    60 

    61 HAVE_GETVALUE = not getattr(_multiprocessing,

    62                             'HAVE_BROKEN_SEM_GETVALUE', False)

    63 

    64 #

    65 # Creates a wrapper for a function which records the time it takes to finish

    66 #

    67 

    68 class TimingWrapper(object):

    69 

    70     def __init__(self, func):

    71         self.func = func

    72         self.elapsed = None

    73 

    74     def __call__(self, *args, **kwds):

    75         t = time.time()

    76         try:

    77             return self.func(*args, **kwds)

    78         finally:

    79             self.elapsed = time.time() - t

    80 

    81 #

    82 # Base class for test cases

    83 #

    84 

    85 class BaseTestCase(object):

    86 

    87     ALLOWED_TYPES = ('processes', 'manager', 'threads')

    88 

    89     def assertTimingAlmostEqual(self, a, b):

    90         if CHECK_TIMINGS:

    91             self.assertAlmostEqual(a, b, 1)

    92 

    93     def assertReturnsIfImplemented(self, value, func, *args):

    94         try:

    95             res = func(*args)

    96         except NotImplementedError:

    97             pass

    98         else:

    99             return self.assertEqual(value, res)

   100 

   101 #

   102 # Return the value of a semaphore

   103 #

   104 

   105 def get_value(self):

   106     try:

   107         return self.get_value()

   108     except AttributeError:

   109         try:

   110             return self._Semaphore__value

   111         except AttributeError:

   112             try:

   113                 return self._value

   114             except AttributeError:

   115                 raise NotImplementedError

   116 

   117 #

   118 # Testcases

   119 #

   120 

   121 class _TestProcess(BaseTestCase):

   122 

   123     ALLOWED_TYPES = ('processes', 'threads')

   124 

   125     def test_current(self):

   126         if self.TYPE == 'threads':

   127             return

   128 

   129         current = self.current_process()

   130         authkey = current.authkey

   131 

   132         self.assertTrue(current.is_alive())

   133         self.assertTrue(not current.daemon)

   134         self.assertTrue(isinstance(authkey, bytes))

   135         self.assertTrue(len(authkey) > 0)

   136         self.assertEqual(current.ident, os.getpid())

   137         self.assertEqual(current.exitcode, None)

   138 

   139     def _test(self, q, *args, **kwds):

   140         current = self.current_process()

   141         q.put(args)

   142         q.put(kwds)

   143         q.put(current.name)

   144         if self.TYPE != 'threads':

   145             q.put(bytes(current.authkey))

   146             q.put(current.pid)

   147 

   148     def test_process(self):

   149         q = self.Queue(1)

   150         e = self.Event()

   151         args = (q, 1, 2)

   152         kwargs = {'hello':23, 'bye':2.54}

   153         name = 'SomeProcess'

   154         p = self.Process(

   155             target=self._test, args=args, kwargs=kwargs, name=name

   156             )

   157         p.daemon = True

   158         current = self.current_process()

   159 

   160         if self.TYPE != 'threads':

   161             self.assertEquals(p.authkey, current.authkey)

   162         self.assertEquals(p.is_alive(), False)

   163         self.assertEquals(p.daemon, True)

   164         self.assertTrue(p not in self.active_children())

   165         self.assertTrue(type(self.active_children()) is list)

   166         self.assertEqual(p.exitcode, None)

   167 

   168         p.start()

   169 

   170         self.assertEquals(p.exitcode, None)

   171         self.assertEquals(p.is_alive(), True)

   172         self.assertTrue(p in self.active_children())

   173 

   174         self.assertEquals(q.get(), args[1:])

   175         self.assertEquals(q.get(), kwargs)

   176         self.assertEquals(q.get(), p.name)

   177         if self.TYPE != 'threads':

   178             self.assertEquals(q.get(), current.authkey)

   179             self.assertEquals(q.get(), p.pid)

   180 

   181         p.join()

   182 

   183         self.assertEquals(p.exitcode, 0)

   184         self.assertEquals(p.is_alive(), False)

   185         self.assertTrue(p not in self.active_children())

   186 

   187     def _test_terminate(self):

   188         time.sleep(1000)

   189 

   190     def test_terminate(self):

   191         if self.TYPE == 'threads':

   192             return

   193 

   194         p = self.Process(target=self._test_terminate)

   195         p.daemon = True

   196         p.start()

   197 

   198         self.assertEqual(p.is_alive(), True)

   199         self.assertTrue(p in self.active_children())

   200         self.assertEqual(p.exitcode, None)

   201 

   202         p.terminate()

   203 

   204         join = TimingWrapper(p.join)

   205         self.assertEqual(join(), None)

   206         self.assertTimingAlmostEqual(join.elapsed, 0.0)

   207 

   208         self.assertEqual(p.is_alive(), False)

   209         self.assertTrue(p not in self.active_children())

   210 

   211         p.join()

   212 

   213         # XXX sometimes get p.exitcode == 0 on Windows ...

   214         #self.assertEqual(p.exitcode, -signal.SIGTERM)

   215 

   216     def test_cpu_count(self):

   217         try:

   218             cpus = multiprocessing.cpu_count()

   219         except NotImplementedError:

   220             cpus = 1

   221         self.assertTrue(type(cpus) is int)

   222         self.assertTrue(cpus >= 1)

   223 

   224     def test_active_children(self):

   225         self.assertEqual(type(self.active_children()), list)

   226 

   227         p = self.Process(target=time.sleep, args=(DELTA,))

   228         self.assertTrue(p not in self.active_children())

   229 

   230         p.start()

   231         self.assertTrue(p in self.active_children())

   232 

   233         p.join()

   234         self.assertTrue(p not in self.active_children())

   235 

   236     def _test_recursion(self, wconn, id):

   237         from multiprocessing import forking

   238         wconn.send(id)

   239         if len(id) < 2:

   240             for i in range(2):

   241                 p = self.Process(

   242                     target=self._test_recursion, args=(wconn, id+[i])

   243                     )

   244                 p.start()

   245                 p.join()

   246 

   247     def test_recursion(self):

   248         rconn, wconn = self.Pipe(duplex=False)

   249         self._test_recursion(wconn, [])

   250 

   251         time.sleep(DELTA)

   252         result = []

   253         while rconn.poll():

   254             result.append(rconn.recv())

   255 

   256         expected = [

   257             [],

   258               [0],

   259                 [0, 0],

   260                 [0, 1],

   261               [1],

   262                 [1, 0],

   263                 [1, 1]

   264             ]

   265         self.assertEqual(result, expected)

   266 

   267 #

   268 #

   269 #

   270 

   271 class _UpperCaser(multiprocessing.Process):

   272 

   273     def __init__(self):

   274         multiprocessing.Process.__init__(self)

   275         self.child_conn, self.parent_conn = multiprocessing.Pipe()

   276 

   277     def run(self):

   278         self.parent_conn.close()

   279         for s in iter(self.child_conn.recv, None):

   280             self.child_conn.send(s.upper())

   281         self.child_conn.close()

   282 

   283     def submit(self, s):

   284         assert type(s) is str

   285         self.parent_conn.send(s)

   286         return self.parent_conn.recv()

   287 

   288     def stop(self):

   289         self.parent_conn.send(None)

   290         self.parent_conn.close()

   291         self.child_conn.close()

   292 

   293 class _TestSubclassingProcess(BaseTestCase):

   294 

   295     ALLOWED_TYPES = ('processes',)

   296 

   297     def test_subclassing(self):

   298         uppercaser = _UpperCaser()

   299         uppercaser.start()

   300         self.assertEqual(uppercaser.submit('hello'), 'HELLO')

   301         self.assertEqual(uppercaser.submit('world'), 'WORLD')

   302         uppercaser.stop()

   303         uppercaser.join()

   304 

   305 #

   306 #

   307 #

   308 

   309 def queue_empty(q):

   310     if hasattr(q, 'empty'):

   311         return q.empty()

   312     else:

   313         return q.qsize() == 0

   314 

   315 def queue_full(q, maxsize):

   316     if hasattr(q, 'full'):

   317         return q.full()

   318     else:

   319         return q.qsize() == maxsize

   320 

   321 

   322 class _TestQueue(BaseTestCase):

   323 

   324 

   325     def _test_put(self, queue, child_can_start, parent_can_continue):

   326         child_can_start.wait()

   327         for i in range(6):

   328             queue.get()

   329         parent_can_continue.set()

   330 

   331     def test_put(self):

   332         MAXSIZE = 6

   333         queue = self.Queue(maxsize=MAXSIZE)

   334         child_can_start = self.Event()

   335         parent_can_continue = self.Event()

   336 

   337         proc = self.Process(

   338             target=self._test_put,

   339             args=(queue, child_can_start, parent_can_continue)

   340             )

   341         proc.daemon = True

   342         proc.start()

   343 

   344         self.assertEqual(queue_empty(queue), True)

   345         self.assertEqual(queue_full(queue, MAXSIZE), False)

   346 

   347         queue.put(1)

   348         queue.put(2, True)

   349         queue.put(3, True, None)

   350         queue.put(4, False)

   351         queue.put(5, False, None)

   352         queue.put_nowait(6)

   353 

   354         # the values may be in buffer but not yet in pipe so sleep a bit

   355         time.sleep(DELTA)

   356 

   357         self.assertEqual(queue_empty(queue), False)

   358         self.assertEqual(queue_full(queue, MAXSIZE), True)

   359 

   360         put = TimingWrapper(queue.put)

   361         put_nowait = TimingWrapper(queue.put_nowait)

   362 

   363         self.assertRaises(Queue.Full, put, 7, False)

   364         self.assertTimingAlmostEqual(put.elapsed, 0)

   365 

   366         self.assertRaises(Queue.Full, put, 7, False, None)

   367         self.assertTimingAlmostEqual(put.elapsed, 0)

   368 

   369         self.assertRaises(Queue.Full, put_nowait, 7)

   370         self.assertTimingAlmostEqual(put_nowait.elapsed, 0)

   371 

   372         self.assertRaises(Queue.Full, put, 7, True, TIMEOUT1)

   373         self.assertTimingAlmostEqual(put.elapsed, TIMEOUT1)

   374 

   375         self.assertRaises(Queue.Full, put, 7, False, TIMEOUT2)

   376         self.assertTimingAlmostEqual(put.elapsed, 0)

   377 

   378         self.assertRaises(Queue.Full, put, 7, True, timeout=TIMEOUT3)

   379         self.assertTimingAlmostEqual(put.elapsed, TIMEOUT3)

   380 

   381         child_can_start.set()

   382         parent_can_continue.wait()

   383 

   384         self.assertEqual(queue_empty(queue), True)

   385         self.assertEqual(queue_full(queue, MAXSIZE), False)

   386 

   387         proc.join()

   388 

   389     def _test_get(self, queue, child_can_start, parent_can_continue):

   390         child_can_start.wait()

   391         #queue.put(1)

   392         queue.put(2)

   393         queue.put(3)

   394         queue.put(4)

   395         queue.put(5)

   396         parent_can_continue.set()

   397 

   398     def test_get(self):

   399         queue = self.Queue()

   400         child_can_start = self.Event()

   401         parent_can_continue = self.Event()

   402 

   403         proc = self.Process(

   404             target=self._test_get,

   405             args=(queue, child_can_start, parent_can_continue)

   406             )

   407         proc.daemon = True

   408         proc.start()

   409 

   410         self.assertEqual(queue_empty(queue), True)

   411 

   412         child_can_start.set()

   413         parent_can_continue.wait()

   414 

   415         time.sleep(DELTA)

   416         self.assertEqual(queue_empty(queue), False)

   417 

   418         # Hangs unexpectedly, remove for now

   419         #self.assertEqual(queue.get(), 1)

   420         self.assertEqual(queue.get(True, None), 2)

   421         self.assertEqual(queue.get(True), 3)

   422         self.assertEqual(queue.get(timeout=1), 4)

   423         self.assertEqual(queue.get_nowait(), 5)

   424 

   425         self.assertEqual(queue_empty(queue), True)

   426 

   427         get = TimingWrapper(queue.get)

   428         get_nowait = TimingWrapper(queue.get_nowait)

   429 

   430         self.assertRaises(Queue.Empty, get, False)

   431         self.assertTimingAlmostEqual(get.elapsed, 0)

   432 

   433         self.assertRaises(Queue.Empty, get, False, None)

   434         self.assertTimingAlmostEqual(get.elapsed, 0)

   435 

   436         self.assertRaises(Queue.Empty, get_nowait)

   437         self.assertTimingAlmostEqual(get_nowait.elapsed, 0)

   438 

   439         self.assertRaises(Queue.Empty, get, True, TIMEOUT1)

   440         self.assertTimingAlmostEqual(get.elapsed, TIMEOUT1)

   441 

   442         self.assertRaises(Queue.Empty, get, False, TIMEOUT2)

   443         self.assertTimingAlmostEqual(get.elapsed, 0)

   444 

   445         self.assertRaises(Queue.Empty, get, timeout=TIMEOUT3)

   446         self.assertTimingAlmostEqual(get.elapsed, TIMEOUT3)

   447 

   448         proc.join()

   449 

   450     def _test_fork(self, queue):

   451         for i in range(10, 20):

   452             queue.put(i)

   453         # note that at this point the items may only be buffered, so the

   454         # process cannot shutdown until the feeder thread has finished

   455         # pushing items onto the pipe.

   456 

   457     def test_fork(self):

   458         # Old versions of Queue would fail to create a new feeder

   459         # thread for a forked process if the original process had its

   460         # own feeder thread.  This test checks that this no longer

   461         # happens.

   462 

   463         queue = self.Queue()

   464 

   465         # put items on queue so that main process starts a feeder thread

   466         for i in range(10):

   467             queue.put(i)

   468 

   469         # wait to make sure thread starts before we fork a new process

   470         time.sleep(DELTA)

   471 

   472         # fork process

   473         p = self.Process(target=self._test_fork, args=(queue,))

   474         p.start()

   475 

   476         # check that all expected items are in the queue

   477         for i in range(20):

   478             self.assertEqual(queue.get(), i)

   479         self.assertRaises(Queue.Empty, queue.get, False)

   480 

   481         p.join()

   482 

   483     def test_qsize(self):

   484         q = self.Queue()

   485         try:

   486             self.assertEqual(q.qsize(), 0)

   487         except NotImplementedError:

   488             return

   489         q.put(1)

   490         self.assertEqual(q.qsize(), 1)

   491         q.put(5)

   492         self.assertEqual(q.qsize(), 2)

   493         q.get()

   494         self.assertEqual(q.qsize(), 1)

   495         q.get()

   496         self.assertEqual(q.qsize(), 0)

   497 

   498     def _test_task_done(self, q):

   499         for obj in iter(q.get, None):

   500             time.sleep(DELTA)

   501             q.task_done()

   502 

   503     def test_task_done(self):

   504         queue = self.JoinableQueue()

   505 

   506         if sys.version_info < (2, 5) and not hasattr(queue, 'task_done'):

   507             return

   508 

   509         workers = [self.Process(target=self._test_task_done, args=(queue,))

   510                    for i in xrange(4)]

   511 

   512         for p in workers:

   513             p.start()

   514 

   515         for i in xrange(10):

   516             queue.put(i)

   517 

   518         queue.join()

   519 

   520         for p in workers:

   521             queue.put(None)

   522 

   523         for p in workers:

   524             p.join()

   525 

   526 #

   527 #

   528 #

   529 

   530 class _TestLock(BaseTestCase):

   531 

   532     def test_lock(self):

   533         lock = self.Lock()

   534         self.assertEqual(lock.acquire(), True)

   535         self.assertEqual(lock.acquire(False), False)

   536         self.assertEqual(lock.release(), None)

   537         self.assertRaises((ValueError, threading.ThreadError), lock.release)

   538 

   539     def test_rlock(self):

   540         lock = self.RLock()

   541         self.assertEqual(lock.acquire(), True)

   542         self.assertEqual(lock.acquire(), True)

   543         self.assertEqual(lock.acquire(), True)

   544         self.assertEqual(lock.release(), None)

   545         self.assertEqual(lock.release(), None)

   546         self.assertEqual(lock.release(), None)

   547         self.assertRaises((AssertionError, RuntimeError), lock.release)

   548 

   549 

   550 class _TestSemaphore(BaseTestCase):

   551 

   552     def _test_semaphore(self, sem):

   553         self.assertReturnsIfImplemented(2, get_value, sem)

   554         self.assertEqual(sem.acquire(), True)

   555         self.assertReturnsIfImplemented(1, get_value, sem)

   556         self.assertEqual(sem.acquire(), True)

   557         self.assertReturnsIfImplemented(0, get_value, sem)

   558         self.assertEqual(sem.acquire(False), False)

   559         self.assertReturnsIfImplemented(0, get_value, sem)

   560         self.assertEqual(sem.release(), None)

   561         self.assertReturnsIfImplemented(1, get_value, sem)

   562         self.assertEqual(sem.release(), None)

   563         self.assertReturnsIfImplemented(2, get_value, sem)

   564 

   565     def test_semaphore(self):

   566         sem = self.Semaphore(2)

   567         self._test_semaphore(sem)

   568         self.assertEqual(sem.release(), None)

   569         self.assertReturnsIfImplemented(3, get_value, sem)

   570         self.assertEqual(sem.release(), None)

   571         self.assertReturnsIfImplemented(4, get_value, sem)

   572 

   573     def test_bounded_semaphore(self):

   574         sem = self.BoundedSemaphore(2)

   575         self._test_semaphore(sem)

   576         # Currently fails on OS/X

   577         #if HAVE_GETVALUE:

   578         #    self.assertRaises(ValueError, sem.release)

   579         #    self.assertReturnsIfImplemented(2, get_value, sem)

   580 

   581     def test_timeout(self):

   582         if self.TYPE != 'processes':

   583             return

   584 

   585         sem = self.Semaphore(0)

   586         acquire = TimingWrapper(sem.acquire)

   587 

   588         self.assertEqual(acquire(False), False)

   589         self.assertTimingAlmostEqual(acquire.elapsed, 0.0)

   590 

   591         self.assertEqual(acquire(False, None), False)

   592         self.assertTimingAlmostEqual(acquire.elapsed, 0.0)

   593 

   594         self.assertEqual(acquire(False, TIMEOUT1), False)

   595         self.assertTimingAlmostEqual(acquire.elapsed, 0)

   596 

   597         self.assertEqual(acquire(True, TIMEOUT2), False)

   598         self.assertTimingAlmostEqual(acquire.elapsed, TIMEOUT2)

   599 

   600         self.assertEqual(acquire(timeout=TIMEOUT3), False)

   601         self.assertTimingAlmostEqual(acquire.elapsed, TIMEOUT3)

   602 

   603 

   604 class _TestCondition(BaseTestCase):

   605 

   606     def f(self, cond, sleeping, woken, timeout=None):

   607         cond.acquire()

   608         sleeping.release()

   609         cond.wait(timeout)

   610         woken.release()

   611         cond.release()

   612 

   613     def check_invariant(self, cond):

   614         # this is only supposed to succeed when there are no sleepers

   615         if self.TYPE == 'processes':

   616             try:

   617                 sleepers = (cond._sleeping_count.get_value() -

   618                             cond._woken_count.get_value())

   619                 self.assertEqual(sleepers, 0)

   620                 self.assertEqual(cond._wait_semaphore.get_value(), 0)

   621             except NotImplementedError:

   622                 pass

   623 

   624     def test_notify(self):

   625         cond = self.Condition()

   626         sleeping = self.Semaphore(0)

   627         woken = self.Semaphore(0)

   628 

   629         p = self.Process(target=self.f, args=(cond, sleeping, woken))

   630         p.daemon = True

   631         p.start()

   632 

   633         p = threading.Thread(target=self.f, args=(cond, sleeping, woken))

   634         p.daemon = True

   635         p.start()

   636 

   637         # wait for both children to start sleeping

   638         sleeping.acquire()

   639         sleeping.acquire()

   640 

   641         # check no process/thread has woken up

   642         time.sleep(DELTA)

   643         self.assertReturnsIfImplemented(0, get_value, woken)

   644 

   645         # wake up one process/thread

   646         cond.acquire()

   647         cond.notify()

   648         cond.release()

   649 

   650         # check one process/thread has woken up

   651         time.sleep(DELTA)

   652         self.assertReturnsIfImplemented(1, get_value, woken)

   653 

   654         # wake up another

   655         cond.acquire()

   656         cond.notify()

   657         cond.release()

   658 

   659         # check other has woken up

   660         time.sleep(DELTA)

   661         self.assertReturnsIfImplemented(2, get_value, woken)

   662 

   663         # check state is not mucked up

   664         self.check_invariant(cond)

   665         p.join()

   666 

   667     def test_notify_all(self):

   668         cond = self.Condition()

   669         sleeping = self.Semaphore(0)

   670         woken = self.Semaphore(0)

   671 

   672         # start some threads/processes which will timeout

   673         for i in range(3):

   674             p = self.Process(target=self.f,

   675                              args=(cond, sleeping, woken, TIMEOUT1))

   676             p.daemon = True

   677             p.start()

   678 

   679             t = threading.Thread(target=self.f,

   680                                  args=(cond, sleeping, woken, TIMEOUT1))

   681             t.daemon = True

   682             t.start()

   683 

   684         # wait for them all to sleep

   685         for i in xrange(6):

   686             sleeping.acquire()

   687 

   688         # check they have all timed out

   689         for i in xrange(6):

   690             woken.acquire()

   691         self.assertReturnsIfImplemented(0, get_value, woken)

   692 

   693         # check state is not mucked up

   694         self.check_invariant(cond)

   695 

   696         # start some more threads/processes

   697         for i in range(3):

   698             p = self.Process(target=self.f, args=(cond, sleeping, woken))

   699             p.daemon = True

   700             p.start()

   701 

   702             t = threading.Thread(target=self.f, args=(cond, sleeping, woken))

   703             t.daemon = True

   704             t.start()

   705 

   706         # wait for them to all sleep

   707         for i in xrange(6):

   708             sleeping.acquire()

   709 

   710         # check no process/thread has woken up

   711         time.sleep(DELTA)

   712         self.assertReturnsIfImplemented(0, get_value, woken)

   713 

   714         # wake them all up

   715         cond.acquire()

   716         cond.notify_all()

   717         cond.release()

   718 

   719         # check they have all woken

   720         time.sleep(DELTA)

   721         self.assertReturnsIfImplemented(6, get_value, woken)

   722 

   723         # check state is not mucked up

   724         self.check_invariant(cond)

   725 

   726     def test_timeout(self):

   727         cond = self.Condition()

   728         wait = TimingWrapper(cond.wait)

   729         cond.acquire()

   730         res = wait(TIMEOUT1)

   731         cond.release()

   732         self.assertEqual(res, None)

   733         self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1)

   734 

   735 

   736 class _TestEvent(BaseTestCase):

   737 

   738     def _test_event(self, event):

   739         time.sleep(TIMEOUT2)

   740         event.set()

   741 

   742     def test_event(self):

   743         event = self.Event()

   744         wait = TimingWrapper(event.wait)

   745 

   746         # Removed temporaily, due to API shear, this does not

   747         # work with threading._Event objects. is_set == isSet

   748         #self.assertEqual(event.is_set(), False)

   749 

   750         self.assertEqual(wait(0.0), None)

   751         self.assertTimingAlmostEqual(wait.elapsed, 0.0)

   752         self.assertEqual(wait(TIMEOUT1), None)

   753         self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1)

   754 

   755         event.set()

   756 

   757         # See note above on the API differences

   758         # self.assertEqual(event.is_set(), True)

   759         self.assertEqual(wait(), None)

   760         self.assertTimingAlmostEqual(wait.elapsed, 0.0)

   761         self.assertEqual(wait(TIMEOUT1), None)

   762         self.assertTimingAlmostEqual(wait.elapsed, 0.0)

   763         # self.assertEqual(event.is_set(), True)

   764 

   765         event.clear()

   766 

   767         #self.assertEqual(event.is_set(), False)

   768 

   769         self.Process(target=self._test_event, args=(event,)).start()

   770         self.assertEqual(wait(), None)

   771 

   772 #

   773 #

   774 #

   775 

   776 class _TestValue(BaseTestCase):

   777 

   778     codes_values = [

   779         ('i', 4343, 24234),

   780         ('d', 3.625, -4.25),

   781         ('h', -232, 234),

   782         ('c', latin('x'), latin('y'))

   783         ]

   784 

   785     def _test(self, values):

   786         for sv, cv in zip(values, self.codes_values):

   787             sv.value = cv[2]

   788 

   789 

   790     def test_value(self, raw=False):

   791         if self.TYPE != 'processes':

   792             return

   793 

   794         if raw:

   795             values = [self.RawValue(code, value)

   796                       for code, value, _ in self.codes_values]

   797         else:

   798             values = [self.Value(code, value)

   799                       for code, value, _ in self.codes_values]

   800 

   801         for sv, cv in zip(values, self.codes_values):

   802             self.assertEqual(sv.value, cv[1])

   803 

   804         proc = self.Process(target=self._test, args=(values,))

   805         proc.start()

   806         proc.join()

   807 

   808         for sv, cv in zip(values, self.codes_values):

   809             self.assertEqual(sv.value, cv[2])

   810 

   811     def test_rawvalue(self):

   812         self.test_value(raw=True)

   813 

   814     def test_getobj_getlock(self):

   815         if self.TYPE != 'processes':

   816             return

   817 

   818         val1 = self.Value('i', 5)

   819         lock1 = val1.get_lock()

   820         obj1 = val1.get_obj()

   821 

   822         val2 = self.Value('i', 5, lock=None)

   823         lock2 = val2.get_lock()

   824         obj2 = val2.get_obj()

   825 

   826         lock = self.Lock()

   827         val3 = self.Value('i', 5, lock=lock)

   828         lock3 = val3.get_lock()

   829         obj3 = val3.get_obj()

   830         self.assertEqual(lock, lock3)

   831 

   832         arr4 = self.RawValue('i', 5)

   833         self.assertFalse(hasattr(arr4, 'get_lock'))

   834         self.assertFalse(hasattr(arr4, 'get_obj'))

   835 

   836 

   837 class _TestArray(BaseTestCase):

   838 

   839     def f(self, seq):

   840         for i in range(1, len(seq)):

   841             seq[i] += seq[i-1]

   842 

   843     def test_array(self, raw=False):

   844         if self.TYPE != 'processes':

   845             return

   846 

   847         seq = [680, 626, 934, 821, 150, 233, 548, 982, 714, 831]

   848         if raw:

   849             arr = self.RawArray('i', seq)

   850         else:

   851             arr = self.Array('i', seq)

   852 

   853         self.assertEqual(len(arr), len(seq))

   854         self.assertEqual(arr[3], seq[3])

   855         self.assertEqual(list(arr[2:7]), list(seq[2:7]))

   856 

   857         arr[4:8] = seq[4:8] = array.array('i', [1, 2, 3, 4])

   858 

   859         self.assertEqual(list(arr[:]), seq)

   860 

   861         self.f(seq)

   862 

   863         p = self.Process(target=self.f, args=(arr,))

   864         p.start()

   865         p.join()

   866 

   867         self.assertEqual(list(arr[:]), seq)

   868 

   869     def test_rawarray(self):

   870         self.test_array(raw=True)

   871 

   872     def test_getobj_getlock_obj(self):

   873         if self.TYPE != 'processes':

   874             return

   875 

   876         arr1 = self.Array('i', range(10))

   877         lock1 = arr1.get_lock()

   878         obj1 = arr1.get_obj()

   879 

   880         arr2 = self.Array('i', range(10), lock=None)

   881         lock2 = arr2.get_lock()

   882         obj2 = arr2.get_obj()

   883 

   884         lock = self.Lock()

   885         arr3 = self.Array('i', range(10), lock=lock)

   886         lock3 = arr3.get_lock()

   887         obj3 = arr3.get_obj()

   888         self.assertEqual(lock, lock3)

   889 

   890         arr4 = self.RawArray('i', range(10))

   891         self.assertFalse(hasattr(arr4, 'get_lock'))

   892         self.assertFalse(hasattr(arr4, 'get_obj'))

   893 

   894 #

   895 #

   896 #

   897 

   898 class _TestContainers(BaseTestCase):

   899 

   900     ALLOWED_TYPES = ('manager',)

   901 

   902     def test_list(self):

   903         a = self.list(range(10))

   904         self.assertEqual(a[:], range(10))

   905 

   906         b = self.list()

   907         self.assertEqual(b[:], [])

   908 

   909         b.extend(range(5))

   910         self.assertEqual(b[:], range(5))

   911 

   912         self.assertEqual(b[2], 2)

   913         self.assertEqual(b[2:10], [2,3,4])

   914 

   915         b *= 2

   916         self.assertEqual(b[:], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4])

   917 

   918         self.assertEqual(b + [5, 6], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4, 5, 6])

   919 

   920         self.assertEqual(a[:], range(10))

   921 

   922         d = [a, b]

   923         e = self.list(d)

   924         self.assertEqual(

   925             e[:],

   926             [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4]]

   927             )

   928 

   929         f = self.list([a])

   930         a.append('hello')

   931         self.assertEqual(f[:], [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 'hello']])

   932 

   933     def test_dict(self):

   934         d = self.dict()

   935         indices = range(65, 70)

   936         for i in indices:

   937             d[i] = chr(i)

   938         self.assertEqual(d.copy(), dict((i, chr(i)) for i in indices))

   939         self.assertEqual(sorted(d.keys()), indices)

   940         self.assertEqual(sorted(d.values()), [chr(i) for i in indices])

   941         self.assertEqual(sorted(d.items()), [(i, chr(i)) for i in indices])

   942 

   943     def test_namespace(self):

   944         n = self.Namespace()

   945         n.name = 'Bob'

   946         n.job = 'Builder'

   947         n._hidden = 'hidden'

   948         self.assertEqual((n.name, n.job), ('Bob', 'Builder'))

   949         del n.job

   950         self.assertEqual(str(n), "Namespace(name='Bob')")

   951         self.assertTrue(hasattr(n, 'name'))

   952         self.assertTrue(not hasattr(n, 'job'))

   953 

   954 #

   955 #

   956 #

   957 

   958 def sqr(x, wait=0.0):

   959     time.sleep(wait)

   960     return x*x

   961 class _TestPool(BaseTestCase):

   962 

   963     def test_apply(self):

   964         papply = self.pool.apply

   965         self.assertEqual(papply(sqr, (5,)), sqr(5))

   966         self.assertEqual(papply(sqr, (), {'x':3}), sqr(x=3))

   967 

   968     def test_map(self):

   969         pmap = self.pool.map

   970         self.assertEqual(pmap(sqr, range(10)), map(sqr, range(10)))

   971         self.assertEqual(pmap(sqr, range(100), chunksize=20),

   972                          map(sqr, range(100)))

   973 

   974     def test_async(self):

   975         res = self.pool.apply_async(sqr, (7, TIMEOUT1,))

   976         get = TimingWrapper(res.get)

   977         self.assertEqual(get(), 49)

   978         self.assertTimingAlmostEqual(get.elapsed, TIMEOUT1)

   979 

   980     def test_async_timeout(self):

   981         res = self.pool.apply_async(sqr, (6, TIMEOUT2 + 0.2))

   982         get = TimingWrapper(res.get)

   983         self.assertRaises(multiprocessing.TimeoutError, get, timeout=TIMEOUT2)

   984         self.assertTimingAlmostEqual(get.elapsed, TIMEOUT2)

   985 

   986     def test_imap(self):

   987         it = self.pool.imap(sqr, range(10))

   988         self.assertEqual(list(it), map(sqr, range(10)))

   989 

   990         it = self.pool.imap(sqr, range(10))

   991         for i in range(10):

   992             self.assertEqual(it.next(), i*i)

   993         self.assertRaises(StopIteration, it.next)

   994 

   995         it = self.pool.imap(sqr, range(1000), chunksize=100)

   996         for i in range(1000):

   997             self.assertEqual(it.next(), i*i)

   998         self.assertRaises(StopIteration, it.next)

   999 

  1000     def test_imap_unordered(self):

  1001         it = self.pool.imap_unordered(sqr, range(1000))

  1002         self.assertEqual(sorted(it), map(sqr, range(1000)))

  1003 

  1004         it = self.pool.imap_unordered(sqr, range(1000), chunksize=53)

  1005         self.assertEqual(sorted(it), map(sqr, range(1000)))

  1006 

  1007     def test_make_pool(self):

  1008         p = multiprocessing.Pool(3)

  1009         self.assertEqual(3, len(p._pool))

  1010         p.close()

  1011         p.join()

  1012 

  1013     def test_terminate(self):

  1014         if self.TYPE == 'manager':

  1015             # On Unix a forked process increfs each shared object to

  1016             # which its parent process held a reference.  If the

  1017             # forked process gets terminated then there is likely to

  1018             # be a reference leak.  So to prevent

  1019             # _TestZZZNumberOfObjects from failing we skip this test

  1020             # when using a manager.

  1021             return

  1022 

  1023         result = self.pool.map_async(

  1024             time.sleep, [0.1 for i in range(10000)], chunksize=1

  1025             )

  1026         self.pool.terminate()

  1027         join = TimingWrapper(self.pool.join)

  1028         join()

  1029         self.assertTrue(join.elapsed < 0.2)

  1030 #

  1031 # Test that manager has expected number of shared objects left

  1032 #

  1033 

  1034 class _TestZZZNumberOfObjects(BaseTestCase):

  1035     # Because test cases are sorted alphabetically, this one will get

  1036     # run after all the other tests for the manager.  It tests that

  1037     # there have been no "reference leaks" for the manager's shared

  1038     # objects.  Note the comment in _TestPool.test_terminate().

  1039     ALLOWED_TYPES = ('manager',)

  1040 

  1041     def test_number_of_objects(self):

  1042         EXPECTED_NUMBER = 1                # the pool object is still alive

  1043         multiprocessing.active_children()  # discard dead process objs

  1044         gc.collect()                       # do garbage collection

  1045         refs = self.manager._number_of_objects()

  1046         if refs != EXPECTED_NUMBER:

  1047             print self.manager._debug_info()

  1048 

  1049         self.assertEqual(refs, EXPECTED_NUMBER)

  1050 

  1051 #

  1052 # Test of creating a customized manager class

  1053 #

  1054 

  1055 from multiprocessing.managers import BaseManager, BaseProxy, RemoteError

  1056 

  1057 class FooBar(object):

  1058     def f(self):

  1059         return 'f()'

  1060     def g(self):

  1061         raise ValueError

  1062     def _h(self):

  1063         return '_h()'

  1064 

  1065 def baz():

  1066     for i in xrange(10):

  1067         yield i*i

  1068 

  1069 class IteratorProxy(BaseProxy):

  1070     _exposed_ = ('next', '__next__')

  1071     def __iter__(self):

  1072         return self

  1073     def next(self):

  1074         return self._callmethod('next')

  1075     def __next__(self):

  1076         return self._callmethod('__next__')

  1077 

  1078 class MyManager(BaseManager):

  1079     pass

  1080 

  1081 MyManager.register('Foo', callable=FooBar)

  1082 MyManager.register('Bar', callable=FooBar, exposed=('f', '_h'))

  1083 MyManager.register('baz', callable=baz, proxytype=IteratorProxy)

  1084 

  1085 

  1086 class _TestMyManager(BaseTestCase):

  1087 

  1088     ALLOWED_TYPES = ('manager',)

  1089 

  1090     def test_mymanager(self):

  1091         manager = MyManager()

  1092         manager.start()

  1093 

  1094         foo = manager.Foo()

  1095         bar = manager.Bar()

  1096         baz = manager.baz()

  1097 

  1098         foo_methods = [name for name in ('f', 'g', '_h') if hasattr(foo, name)]

  1099         bar_methods = [name for name in ('f', 'g', '_h') if hasattr(bar, name)]

  1100 

  1101         self.assertEqual(foo_methods, ['f', 'g'])

  1102         self.assertEqual(bar_methods, ['f', '_h'])

  1103 

  1104         self.assertEqual(foo.f(), 'f()')

  1105         self.assertRaises(ValueError, foo.g)

  1106         self.assertEqual(foo._callmethod('f'), 'f()')

  1107         self.assertRaises(RemoteError, foo._callmethod, '_h')

  1108 

  1109         self.assertEqual(bar.f(), 'f()')

  1110         self.assertEqual(bar._h(), '_h()')

  1111         self.assertEqual(bar._callmethod('f'), 'f()')

  1112         self.assertEqual(bar._callmethod('_h'), '_h()')

  1113 

  1114         self.assertEqual(list(baz), [i*i for i in range(10)])

  1115 

  1116         manager.shutdown()

  1117 

  1118 #

  1119 # Test of connecting to a remote server and using xmlrpclib for serialization

  1120 #

  1121 

  1122 _queue = Queue.Queue()

  1123 def get_queue():

  1124     return _queue

  1125 

  1126 class QueueManager(BaseManager):

  1127     '''manager class used by server process'''

  1128 QueueManager.register('get_queue', callable=get_queue)

  1129 

  1130 class QueueManager2(BaseManager):

  1131     '''manager class which specifies the same interface as QueueManager'''

  1132 QueueManager2.register('get_queue')

  1133 

  1134 

  1135 SERIALIZER = 'xmlrpclib'