mp_synchronize.py

#
# A test file for the `multiprocessing` package
#
# Copyright (c) 2006-2008, R Oudkerk
# All rights reserved.
#

importtime, sys, random
fromQueueimportEmpty

importmultiprocessing# may get overwritten


#### TEST_VALUE

defvalue_func(running, mutex):
random.seed()
time.sleep(random.random()*4)

mutex.acquire()
print'\n\t\t\t'+str(multiprocessing.current_process()) +' has finished'
running.value-=1
mutex.release()

deftest_value():
TASKS=10
running=multiprocessing.Value('i', TASKS)
mutex=multiprocessing.Lock()

foriinrange(TASKS):
p=multiprocessing.Process(target=value_func, args=(running, mutex))
p.start()

whilerunning.value>0:
time.sleep(0.08)
mutex.acquire()
printrunning.value,
sys.stdout.flush()
mutex.release()

print
print'No more running processes'


#### TEST_QUEUE

defqueue_func(queue):
foriinrange(30):
time.sleep(0.5*random.random())
queue.put(i*i)
queue.put('STOP')

deftest_queue():
q=multiprocessing.Queue()

p=multiprocessing.Process(target=queue_func, args=(q,))
p.start()

o=None
whileo!='STOP':
try:
o=q.get(timeout=0.3)
printo,
sys.stdout.flush()
exceptEmpty:
print'TIMEOUT'

print


#### TEST_CONDITION

defcondition_func(cond):
cond.acquire()
print'\t'+str(cond)
time.sleep(2)
print'\tchild is notifying'
print'\t'+str(cond)
cond.notify()
cond.release()

deftest_condition():
cond=multiprocessing.Condition()

p=multiprocessing.Process(target=condition_func, args=(cond,))
printcond

cond.acquire()
printcond
cond.acquire()
printcond

p.start()

print'main is waiting'
cond.wait()
print'main has woken up'

printcond
cond.release()
printcond
cond.release()

p.join()
printcond


#### TEST_SEMAPHORE

defsemaphore_func(sema, mutex, running):
sema.acquire()

mutex.acquire()
running.value+=1
printrunning.value, 'tasks are running'
mutex.release()

random.seed()
time.sleep(random.random()*2)

mutex.acquire()
running.value-=1
print'%s has finished'%multiprocessing.current_process()
mutex.release()

sema.release()

deftest_semaphore():
sema=multiprocessing.Semaphore(3)
mutex=multiprocessing.RLock()
running=multiprocessing.Value('i', 0)

processes= [
multiprocessing.Process(target=semaphore_func,
args=(sema, mutex, running))
foriinrange(10)
 ]

forpinprocesses:
p.start()

forpinprocesses:
p.join()


#### TEST_JOIN_TIMEOUT

defjoin_timeout_func():
print'\tchild sleeping'
time.sleep(5.5)
print'\n\tchild terminating'

deftest_join_timeout():
p=multiprocessing.Process(target=join_timeout_func)
p.start()

print'waiting for process to finish'

while1:
p.join(timeout=1)
ifnotp.is_alive():
break
print'.',
sys.stdout.flush()


#### TEST_EVENT

defevent_func(event):
print'\t%r is waiting'%multiprocessing.current_process()
event.wait()
print'\t%r has woken up'%multiprocessing.current_process()

deftest_event():
event=multiprocessing.Event()

processes= [multiprocessing.Process(target=event_func, args=(event,))
foriinrange(5)]

forpinprocesses:
p.start()

print'main is sleeping'
time.sleep(2)

print'main is setting event'
event.set()

forpinprocesses:
p.join()


#### TEST_SHAREDVALUES

defsharedvalues_func(values, arrays, shared_values, shared_arrays):
foriinrange(len(values)):
v=values[i][1]
sv=shared_values[i].value
assertv==sv

foriinrange(len(values)):
a=arrays[i][1]
sa=list(shared_arrays[i][:])
asserta==sa

print'Tests passed'

deftest_sharedvalues():
values= [
 ('i', 10),
 ('h', -2),
 ('d', 1.25)
 ]
arrays= [
 ('i', range(100)),
 ('d', [0.25*iforiinrange(100)]),
 ('H', range(1000))
 ]

shared_values= [multiprocessing.Value(id, v) forid, vinvalues]
shared_arrays= [multiprocessing.Array(id, a) forid, ainarrays]

p=multiprocessing.Process(
target=sharedvalues_func,
args=(values, arrays, shared_values, shared_arrays)
 )
p.start()
p.join()

assertp.exitcode==0


####

deftest(namespace=multiprocessing):
globalmultiprocessing

multiprocessing=namespace

forfuncin [ test_value, test_queue, test_condition,
test_semaphore, test_join_timeout, test_event,
test_sharedvalues ]:

print'\n\t######## %s\n'%func.__name__
func()

ignore=multiprocessing.active_children() # cleanup any old processes
ifhasattr(multiprocessing, '_debug_info'):
info=multiprocessing._debug_info()
ifinfo:
printinfo
raiseValueError('there should be no positive refcounts left')


if__name__=='__main__':
multiprocessing.freeze_support()

assertlen(sys.argv) in (1, 2)

iflen(sys.argv) ==1orsys.argv[1] =='processes':
print' Using processes '.center(79, '-')
namespace=multiprocessing
elifsys.argv[1] =='manager':
print' Using processes and a manager '.center(79, '-')
namespace=multiprocessing.Manager()
namespace.Process=multiprocessing.Process
namespace.current_process=multiprocessing.current_process
namespace.active_children=multiprocessing.active_children
elifsys.argv[1] =='threads':
print' Using threads '.center(79, '-')
importmultiprocessing.dummyasnamespace
else:
print'Usage:\n\t%s [processes | manager | threads]'%sys.argv[0]
raiseSystemExit(2)

test(namespace)