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