Doc/includes/mp_benchmarks.py

   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()