0001 # Very rudimentary test of threading module 0002 0003 # Create a bunch of threads, let each do some work, wait until all are done 0004 0005 from test.test_support import verbose 0006 import random 0007 import dummy_threading as _threading 0008 import time 0009 0010 0011 class TestThread(_threading.Thread): 0012 0013 def run(self): 0014 global running 0015 # Uncomment if testing another module, such as the real 'threading' 0016 # module. 0017 #delay = random.random() * 2 0018 delay = 0 0019 if verbose: 0020 print 'task', self.getName(), 'will run for', delay, 'sec' 0021 sema.acquire() 0022 mutex.acquire() 0023 running = running + 1 0024 if verbose: 0025 print running, 'tasks are running' 0026 mutex.release() 0027 time.sleep(delay) 0028 if verbose: 0029 print 'task', self.getName(), 'done' 0030 mutex.acquire() 0031 running = running - 1 0032 if verbose: 0033 print self.getName(), 'is finished.', running, 'tasks are running' 0034 mutex.release() 0035 sema.release() 0036 0037 def starttasks(): 0038 for i in range(numtasks): 0039 t = TestThread(name="<thread %d>"%i) 0040 threads.append(t) 0041 t.start() 0042 0043 0044 def test_main(): 0045 # This takes about n/3 seconds to run (about n/3 clumps of tasks, times 0046 # about 1 second per clump). 0047 global numtasks 0048 numtasks = 10 0049 0050 # no more than 3 of the 10 can run at once 0051 global sema 0052 sema = _threading.BoundedSemaphore(value=3) 0053 global mutex 0054 mutex = _threading.RLock() 0055 global running 0056 running = 0 0057 0058 global threads 0059 threads = [] 0060 0061 starttasks() 0062 0063 if verbose: 0064 print 'waiting for all tasks to complete' 0065 for t in threads: 0066 t.join() 0067 if verbose: 0068 print 'all tasks done' 0069 0070 0071 0072 if __name__ == '__main__': 0073 test_main() 0074
Generated by PyXR 0.9.4