1 #

     2 # Simple benchmarks for the multiprocessing package

     3 #

     4 

     5 import time, sys, multiprocessing, threading, Queue, gc

     6 

     7 if sys.platform == 'win32':

     8     _timer = time.clock

     9 else:

    10     _timer = time.time

    11 

    12 delta = 1

    13 

    14 

    15 #### TEST_QUEUESPEED

    16 

    17 def queuespeed_func(q, c, iterations):

    18     a = '0' * 256

    19     c.acquire()

    20     c.notify()

    21     c.release()

    22 

    23     for i in xrange(iterations):

    24         q.put(a)

    25 

    26     q.put('STOP')

    27 

    28 def test_queuespeed(Process, q, c):

    29     elapsed = 0

    30     iterations = 1

    31 

    32     while elapsed < delta:

    33         iterations *= 2

    34 

    35         p = Process(target=queuespeed_func, args=(q, c, iterations))

    36         c.acquire()

    37         p.start()

    38         c.wait()

    39         c.release()

    40 

    41         result = None

    42         t = _timer()

    43 

    44         while result != 'STOP':

    45             result = q.get()

    46 

    47         elapsed = _timer() - t

    48 

    49         p.join()

    50 

    51     print iterations, 'objects passed through the queue in', elapsed, 'seconds'

    52     print 'average number/sec:', iterations/elapsed

    53 

    54 

    55 #### TEST_PIPESPEED

    56 

    57 def pipe_func(c, cond, iterations):

    58     a = '0' * 256

    59     cond.acquire()

    60     cond.notify()

    61     cond.release()

    62 

    63     for i in xrange(iterations):

    64         c.send(a)

    65 

    66     c.send('STOP')

    67 

    68 def test_pipespeed():

    69     c, d = multiprocessing.Pipe()

    70     cond = multiprocessing.Condition()

    71     elapsed = 0

    72     iterations = 1

    73 

    74     while elapsed < delta:

    75         iterations *= 2

    76 

    77         p = multiprocessing.Process(target=pipe_func,

    78                                     args=(d, cond, iterations))

    79         cond.acquire()

    80         p.start()

    81         cond.wait()

    82         cond.release()

    83 

    84         result = None

    85         t = _timer()

    86 

    87         while result != 'STOP':

    88             result = c.recv()

    89 

    90         elapsed = _timer() - t

    91         p.join()

    92 

    93     print iterations, 'objects passed through connection in',elapsed,'seconds'

    94     print 'average number/sec:', iterations/elapsed

    95 

    96 

    97 #### TEST_SEQSPEED

    98 

    99 def test_seqspeed(seq):

   100     elapsed = 0

   101     iterations = 1

   102 

   103     while elapsed < delta:

   104         iterations *= 2

   105 

   106         t = _timer()

   107 

   108         for i in xrange(iterations):

   109             a = seq[5]

   110 

   111         elapsed = _timer()-t

   112 

   113     print iterations, 'iterations in', elapsed, 'seconds'

   114     print 'average number/sec:', iterations/elapsed

   115 

   116 

   117 #### TEST_LOCK

   118 

   119 def test_lockspeed(l):

   120     elapsed = 0

   121     iterations = 1

   122 

   123     while elapsed < delta:

   124         iterations *= 2

   125 

   126         t = _timer()

   127 

   128         for i in xrange(iterations):

   129             l.acquire()

   130             l.release()

   131 

   132         elapsed = _timer()-t

   133 

   134     print iterations, 'iterations in', elapsed, 'seconds'

   135     print 'average number/sec:', iterations/elapsed

   136 

   137 

   138 #### TEST_CONDITION

   139 

   140 def conditionspeed_func(c, N):

   141     c.acquire()

   142     c.notify()

   143 

   144     for i in xrange(N):

   145         c.wait()

   146         c.notify()

   147 

   148     c.release()

   149 

   150 def test_conditionspeed(Process, c):

   151     elapsed = 0

   152     iterations = 1

   153 

   154     while elapsed < delta:

   155         iterations *= 2

   156 

   157         c.acquire()

   158         p = Process(target=conditionspeed_func, args=(c, iterations))

   159         p.start()

   160 

   161         c.wait()

   162 

   163         t = _timer()

   164 

   165         for i in xrange(iterations):

   166             c.notify()

   167             c.wait()

   168 

   169         elapsed = _timer()-t

   170 

   171         c.release()

   172         p.join()

   173 

   174     print iterations * 2, 'waits in', elapsed, 'seconds'

   175     print 'average number/sec:', iterations * 2 / elapsed

   176 

   177 ####

   178 

   179 def test():

   180     manager = multiprocessing.Manager()

   181 

   182     gc.disable()

   183 

   184     print '\n\t######## testing Queue.Queue\n'

   185     test_queuespeed(threading.Thread, Queue.Queue(),

   186                     threading.Condition())

   187     print '\n\t######## testing multiprocessing.Queue\n'

   188     test_queuespeed(multiprocessing.Process, multiprocessing.Queue(),

   189                     multiprocessing.Condition())

   190     print '\n\t######## testing Queue managed by server process\n'

   191     test_queuespeed(multiprocessing.Process, manager.Queue(),

   192                     manager.Condition())

   193     print '\n\t######## testing multiprocessing.Pipe\n'

   194     test_pipespeed()

   195 

   196     print

   197 

   198     print '\n\t######## testing list\n'

   199     test_seqspeed(range(10))

   200     print '\n\t######## testing list managed by server process\n'

   201     test_seqspeed(manager.list(range(10)))

   202     print '\n\t######## testing Array("i", ..., lock=False)\n'

   203     test_seqspeed(multiprocessing.Array('i', range(10), lock=False))

   204     print '\n\t######## testing Array("i", ..., lock=True)\n'

   205     test_seqspeed(multiprocessing.Array('i', range(10), lock=True))

   206 

   207     print

   208 

   209     print '\n\t######## testing threading.Lock\n'

   210     test_lockspeed(threading.Lock())

   211     print '\n\t######## testing threading.RLock\n'

   212     test_lockspeed(threading.RLock())

   213     print '\n\t######## testing multiprocessing.Lock\n'

   214     test_lockspeed(multiprocessing.Lock())

   215     print '\n\t######## testing multiprocessing.RLock\n'

   216     test_lockspeed(multiprocessing.RLock())

   217     print '\n\t######## testing lock managed by server process\n'

   218     test_lockspeed(manager.Lock())

   219     print '\n\t######## testing rlock managed by server process\n'

   220     test_lockspeed(manager.RLock())

   221 

   222     print

   223 

   224     print '\n\t######## testing threading.Condition\n'

   225     test_conditionspeed(threading.Thread, threading.Condition())

   226     print '\n\t######## testing multiprocessing.Condition\n'

   227     test_conditionspeed(multiprocessing.Process, multiprocessing.Condition())

   228     print '\n\t######## testing condition managed by a server process\n'

   229     test_conditionspeed(multiprocessing.Process, manager.Condition())

   230 

   231     gc.enable()

   232 

   233 if __name__ == '__main__':

   234     multiprocessing.freeze_support()

   235     test()