3724f1e5a4ee27a761c32065dd6e016353cea192
# Very rudimentary test of threading module
# Create a bunch of threads, let each do some work, wait until all are done
from test
.test_support
import verbose
import dummy_threading
as _threading
class TestThread(_threading
.Thread
):
# Uncomment if testing another module, such as the real 'threading'
#delay = random.random() * 2
print 'task', self
.getName(), 'will run for', delay
, 'sec'
print running
, 'tasks are running'
print 'task', self
.getName(), 'done'
print self
.getName(), 'is finished.', running
, 'tasks are running'
for i
in range(numtasks
):
t
= TestThread(name
="<thread %d>"%i)
# This takes about n/3 seconds to run (about n/3 clumps of tasks, times
# about 1 second per clump).
# no more than 3 of the 10 can run at once
sema
= _threading
.BoundedSemaphore(value
=3)
mutex
= _threading
.RLock()
print 'waiting for all tasks to complete'
if __name__
== '__main__':