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.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 -= 1
 101     # Must release mutex before releasing done, else the main thread can
 102     # exit and set mutex to None as part of global teardown; then
 103     # mutex.release() raises AttributeError.
 104     finished = running == 0
 105     mutex.release()
 106     if finished:
 107         done.release()
 108
 109 print '\n*** Barrier Test ***'
 110 if done.acquire(0):
 111     raise ValueError, "'done' should have remained acquired"
 112 bar = barrier(numtasks)
 113 running = numtasks
 114 for i in range(numtasks):
 115     thread.start_new_thread(task2, (i,))
 116 done.acquire()
 117 print 'all tasks done'
 118
 119 # not all platforms support changing thread stack size
 120 print '\n*** Changing thread stack size ***'
 121 if thread.stack_size() != 0:
 122     raise ValueError, "initial stack_size not 0"
 123
 124 thread.stack_size(0)
 125 if thread.stack_size() != 0:
 126     raise ValueError, "stack_size not reset to default"
 127
 128 from os import name as os_name
 129 if os_name in ("nt", "os2", "posix"):
 130
 131     tss_supported = 1
 132     try:
 133         thread.stack_size(4096)
 134     except ValueError:
 135         print 'caught expected ValueError setting stack_size(4096)'
 136     except thread.error:
 137         tss_supported = 0
 138         print 'platform does not support changing thread stack size'
 139
 140     if tss_supported:
 141         failed = lambda s, e: s != e
 142         fail_msg = "stack_size(%d) failed - should succeed"
 143         for tss in (262144, 0x100000, 0):
 144             thread.stack_size(tss)
 145             if failed(thread.stack_size(), tss):
 146                 raise ValueError, fail_msg % tss
 147             print 'successfully set stack_size(%d)' % tss
 148
 149         for tss in (262144, 0x100000):
 150             print 'trying stack_size = %d' % tss
 151             next_ident = 0
 152             for i in range(numtasks):
 153                 newtask()
 154
 155             print 'waiting for all tasks to complete'
 156             done.acquire()
 157             print 'all tasks done'
 158
 159         # reset stack size to default
 160         thread.stack_size(0)