Doc/includes/mp_synchronize.py

   1 #
   2 # A test file for the `multiprocessing` package
   3 #
   4 # Copyright (c) 2006-2008, R Oudkerk
   5 # All rights reserved.
   6 #
   7
   8 import time, sys, random
   9 from Queue import Empty
  10
  11 import multiprocessing               # may get overwritten
  12
  13
  14 #### TEST_VALUE
  15
  16 def value_func(running, mutex):
  17     random.seed()
  18     time.sleep(random.random()*4)
  19
  20     mutex.acquire()
  21     print '\n\t\t\t' + str(multiprocessing.current_process()) + ' has finished'
  22     running.value -= 1
  23     mutex.release()
  24
  25 def test_value():
  26     TASKS = 10
  27     running = multiprocessing.Value('i', TASKS)
  28     mutex = multiprocessing.Lock()
  29
  30     for i in range(TASKS):
  31         p = multiprocessing.Process(target=value_func, args=(running, mutex))
  32         p.start()
  33
  34     while running.value > 0:
  35         time.sleep(0.08)
  36         mutex.acquire()
  37         print running.value,
  38         sys.stdout.flush()
  39         mutex.release()
  40
  41     print
  42     print 'No more running processes'
  43
  44
  45 #### TEST_QUEUE
  46
  47 def queue_func(queue):
  48     for i in range(30):
  49         time.sleep(0.5 * random.random())
  50         queue.put(i*i)
  51     queue.put('STOP')
  52
  53 def test_queue():
  54     q = multiprocessing.Queue()
  55
  56     p = multiprocessing.Process(target=queue_func, args=(q,))
  57     p.start()
  58
  59     o = None
  60     while o != 'STOP':
  61         try:
  62             o = q.get(timeout=0.3)
  63             print o,
  64             sys.stdout.flush()
  65         except Empty:
  66             print 'TIMEOUT'
  67
  68     print
  69
  70
  71 #### TEST_CONDITION
  72
  73 def condition_func(cond):
  74     cond.acquire()
  75     print '\t' + str(cond)
  76     time.sleep(2)
  77     print '\tchild is notifying'
  78     print '\t' + str(cond)
  79     cond.notify()
  80     cond.release()
  81
  82 def test_condition():
  83     cond = multiprocessing.Condition()
  84
  85     p = multiprocessing.Process(target=condition_func, args=(cond,))
  86     print cond
  87
  88     cond.acquire()
  89     print cond
  90     cond.acquire()
  91     print cond
  92
  93     p.start()
  94
  95     print 'main is waiting'
  96     cond.wait()
  97     print 'main has woken up'
  98
  99     print cond
 100     cond.release()
 101     print cond
 102     cond.release()
 103
 104     p.join()
 105     print cond
 106
 107
 108 #### TEST_SEMAPHORE
 109
 110 def semaphore_func(sema, mutex, running):
 111     sema.acquire()
 112
 113     mutex.acquire()
 114     running.value += 1
 115     print running.value, 'tasks are running'
 116     mutex.release()
 117
 118     random.seed()
 119     time.sleep(random.random()*2)
 120
 121     mutex.acquire()
 122     running.value -= 1
 123     print '%s has finished' % multiprocessing.current_process()
 124     mutex.release()
 125
 126     sema.release()
 127
 128 def test_semaphore():
 129     sema = multiprocessing.Semaphore(3)
 130     mutex = multiprocessing.RLock()
 131     running = multiprocessing.Value('i', 0)
 132
 133     processes = [
 134         multiprocessing.Process(target=semaphore_func,
 135                                 args=(sema, mutex, running))
 136         for i in range(10)
 137         ]
 138
 139     for p in processes:
 140         p.start()
 141
 142     for p in processes:
 143         p.join()
 144
 145
 146 #### TEST_JOIN_TIMEOUT
 147
 148 def join_timeout_func():
 149     print '\tchild sleeping'
 150     time.sleep(5.5)
 151     print '\n\tchild terminating'
 152
 153 def test_join_timeout():
 154     p = multiprocessing.Process(target=join_timeout_func)
 155     p.start()
 156
 157     print 'waiting for process to finish'
 158
 159     while 1:
 160         p.join(timeout=1)
 161         if not p.is_alive():
 162             break
 163         print '.',
 164         sys.stdout.flush()
 165
 166
 167 #### TEST_EVENT
 168
 169 def event_func(event):
 170     print '\t%r is waiting' % multiprocessing.current_process()
 171     event.wait()
 172     print '\t%r has woken up' % multiprocessing.current_process()
 173
 174 def test_event():
 175     event = multiprocessing.Event()
 176
 177     processes = [multiprocessing.Process(target=event_func, args=(event,))
 178                  for i in range(5)]
 179
 180     for p in processes:
 181         p.start()
 182
 183     print 'main is sleeping'
 184     time.sleep(2)
 185
 186     print 'main is setting event'
 187     event.set()
 188
 189     for p in processes:
 190         p.join()
 191
 192
 193 #### TEST_SHAREDVALUES
 194
 195 def sharedvalues_func(values, arrays, shared_values, shared_arrays):
 196     for i in range(len(values)):
 197         v = values[i][1]
 198         sv = shared_values[i].value
 199         assert v == sv
 200
 201     for i in range(len(values)):
 202         a = arrays[i][1]
 203         sa = list(shared_arrays[i][:])
 204         assert a == sa
 205
 206     print 'Tests passed'
 207
 208 def test_sharedvalues():
 209     values = [
 210         ('i', 10),
 211         ('h', -2),
 212         ('d', 1.25)
 213         ]
 214     arrays = [
 215         ('i', range(100)),
 216         ('d', [0.25 * i for i in range(100)]),
 217         ('H', range(1000))
 218         ]
 219
 220     shared_values = [multiprocessing.Value(id, v) for id, v in values]
 221     shared_arrays = [multiprocessing.Array(id, a) for id, a in arrays]
 222
 223     p = multiprocessing.Process(
 224         target=sharedvalues_func,
 225         args=(values, arrays, shared_values, shared_arrays)
 226         )
 227     p.start()
 228     p.join()
 229
 230     assert p.exitcode == 0
 231
 232
 233 ####
 234
 235 def test(namespace=multiprocessing):
 236     global multiprocessing
 237
 238     multiprocessing = namespace
 239
 240     for func in [ test_value, test_queue, test_condition,
 241                   test_semaphore, test_join_timeout, test_event,
 242                   test_sharedvalues ]:
 243
 244         print '\n\t######## %s\n' % func.__name__
 245         func()
 246
 247     ignore = multiprocessing.active_children()      # cleanup any old processes
 248     if hasattr(multiprocessing, '_debug_info'):
 249         info = multiprocessing._debug_info()
 250         if info:
 251             print info
 252             raise ValueError('there should be no positive refcounts left')
 253
 254
 255 if __name__ == '__main__':
 256     multiprocessing.freeze_support()
 257
 258     assert len(sys.argv) in (1, 2)
 259
 260     if len(sys.argv) == 1 or sys.argv[1] == 'processes':
 261         print ' Using processes '.center(79, '-')
 262         namespace = multiprocessing
 263     elif sys.argv[1] == 'manager':
 264         print ' Using processes and a manager '.center(79, '-')
 265         namespace = multiprocessing.Manager()
 266         namespace.Process = multiprocessing.Process
 267         namespace.current_process = multiprocessing.current_process
 268         namespace.active_children = multiprocessing.active_children
 269     elif sys.argv[1] == 'threads':
 270         print ' Using threads '.center(79, '-')
 271         import multiprocessing.dummy as namespace
 272     else:
 273         print 'Usage:\n\t%s [processes | manager | threads]' % sys.argv[0]
 274         raise SystemExit(2)
 275
 276     test(namespace)