Lib/test/test_thread.py

   1 # Very rudimentary test of thread module
   2
   3 # Create a bunch of threads, let each do some work, wait until all are done
   4
   5 from test_support import verbose
   6 import random
   7 import thread
   8 import time
   9
  10 mutex = thread.allocate_lock()
  11 rmutex = thread.allocate_lock() # for calls to random
  12 running = 0
  13 done = thread.allocate_lock()
  14 done.acquire()
  15
  16 numtasks = 10
  17
  18 def task(ident):
  19         global running
  20         rmutex.acquire()
  21         delay = random.random() * numtasks
  22         rmutex.release()
  23         if verbose:
  24             print 'task', ident, 'will run for', round(delay, 1), 'sec'
  25         time.sleep(delay)
  26         if verbose:
  27             print 'task', ident, 'done'
  28         mutex.acquire()
  29         running = running - 1
  30         if running == 0:
  31                 done.release()
  32         mutex.release()
  33
  34 next_ident = 0
  35 def newtask():
  36         global next_ident, running
  37         mutex.acquire()
  38         next_ident = next_ident + 1
  39         if verbose:
  40             print 'creating task', next_ident
  41         thread.start_new_thread(task, (next_ident,))
  42         running = running + 1
  43         mutex.release()
  44
  45 for i in range(numtasks):
  46         newtask()
  47
  48 print 'waiting for all tasks to complete'
  49 done.acquire()
  50 print 'all tasks done'
  51
  52 class barrier:
  53         def __init__(self, n):
  54                 self.n = n
  55                 self.waiting = 0
  56                 self.checkin  = thread.allocate_lock()
  57                 self.checkout = thread.allocate_lock()
  58                 self.checkout.acquire()
  59
  60         def enter(self):
  61                 checkin, checkout = self.checkin, self.checkout
  62
  63                 checkin.acquire()
  64                 self.waiting = self.waiting + 1
  65                 if self.waiting == self.n:
  66                         self.waiting = self.n - 1
  67                         checkout.release()
  68                         return
  69                 checkin.release()
  70
  71                 checkout.acquire()
  72                 self.waiting = self.waiting - 1
  73                 if self.waiting == 0:
  74                         checkin.release()
  75                         return
  76                 checkout.release()
  77
  78 numtrips = 3
  79 def task2(ident):
  80         global running
  81         for i in range(numtrips):
  82                 if ident == 0:
  83                         # give it a good chance to enter the next
  84                         # barrier before the others are all out
  85                         # of the current one
  86                         delay = 0.001
  87                 else:
  88                         rmutex.acquire()
  89                         delay = random.random() * numtasks
  90                         rmutex.release()
  91                 if verbose:
  92                     print 'task', ident, 'will run for', round(delay, 1), 'sec'
  93                 time.sleep(delay)
  94                 if verbose:
  95                     print 'task', ident, 'entering barrier', i
  96                 bar.enter()
  97                 if verbose:
  98                     print 'task', ident, 'leaving barrier', i
  99         mutex.acquire()
 100         running = running - 1
 101         if running == 0:
 102                 done.release()
 103         mutex.release()
 104
 105 print '\n*** Barrier Test ***'
 106 if done.acquire(0):
 107         raise ValueError, "'done' should have remained acquired"
 108 bar = barrier(numtasks)
 109 running = numtasks
 110 for i in range(numtasks):
 111         thread.start_new_thread(task2, (i,))
 112 done.acquire()
 113 print 'all tasks done'