Lib/test/test_capi.py

   1 # Run the _testcapi module tests (tests for the Python/C API):  by defn,
   2 # these are all functions _testcapi exports whose name begins with 'test_'.
   3
   4 from __future__ import with_statement
   5 import sys
   6 import time
   7 import random
   8 import unittest
   9 from test import test_support
  10 try:
  11     import threading
  12 except ImportError:
  13     threading = None
  14 import _testcapi
  15
  16 @unittest.skipUnless(threading, 'Threading required for this test.')
  17 class TestPendingCalls(unittest.TestCase):
  18
  19     def pendingcalls_submit(self, l, n):
  20         def callback():
  21             #this function can be interrupted by thread switching so let's
  22             #use an atomic operation
  23             l.append(None)
  24
  25         for i in range(n):
  26             time.sleep(random.random()*0.02) #0.01 secs on average
  27             #try submitting callback until successful.
  28             #rely on regular interrupt to flush queue if we are
  29             #unsuccessful.
  30             while True:
  31                 if _testcapi._pending_threadfunc(callback):
  32                     break;
  33
  34     def pendingcalls_wait(self, l, n, context = None):
  35         #now, stick around until l[0] has grown to 10
  36         count = 0;
  37         while len(l) != n:
  38             #this busy loop is where we expect to be interrupted to
  39             #run our callbacks.  Note that callbacks are only run on the
  40             #main thread
  41             if False and test_support.verbose:
  42                 print "(%i)"%(len(l),),
  43             for i in xrange(1000):
  44                 a = i*i
  45             if context and not context.event.is_set():
  46                 continue
  47             count += 1
  48             self.assertTrue(count < 10000,
  49                 "timeout waiting for %i callbacks, got %i"%(n, len(l)))
  50         if False and test_support.verbose:
  51             print "(%i)"%(len(l),)
  52
  53     def test_pendingcalls_threaded(self):
  54         #do every callback on a separate thread
  55         n = 32 #total callbacks
  56         threads = []
  57         class foo(object):pass
  58         context = foo()
  59         context.l = []
  60         context.n = 2 #submits per thread
  61         context.nThreads = n // context.n
  62         context.nFinished = 0
  63         context.lock = threading.Lock()
  64         context.event = threading.Event()
  65
  66         for i in range(context.nThreads):
  67             t = threading.Thread(target=self.pendingcalls_thread, args = (context,))
  68             t.start()
  69             threads.append(t)
  70
  71         self.pendingcalls_wait(context.l, n, context)
  72
  73         for t in threads:
  74             t.join()
  75
  76     def pendingcalls_thread(self, context):
  77         try:
  78             self.pendingcalls_submit(context.l, context.n)
  79         finally:
  80             with context.lock:
  81                 context.nFinished += 1
  82                 nFinished = context.nFinished
  83                 if False and test_support.verbose:
  84                     print "finished threads: ", nFinished
  85             if nFinished == context.nThreads:
  86                 context.event.set()
  87
  88     def test_pendingcalls_non_threaded(self):
  89         #again, just using the main thread, likely they will all be dispathced at
  90         #once.  It is ok to ask for too many, because we loop until we find a slot.
  91         #the loop can be interrupted to dispatch.
  92         #there are only 32 dispatch slots, so we go for twice that!
  93         l = []
  94         n = 64
  95         self.pendingcalls_submit(l, n)
  96         self.pendingcalls_wait(l, n)
  97
  98
  99 def test_main():
 100
 101     for name in dir(_testcapi):
 102         if name.startswith('test_'):
 103             test = getattr(_testcapi, name)
 104             if test_support.verbose:
 105                 print "internal", name
 106             try:
 107                 test()
 108             except _testcapi.error:
 109                 raise test_support.TestFailed, sys.exc_info()[1]
 110
 111     # some extra thread-state tests driven via _testcapi
 112     def TestThreadState():
 113         if test_support.verbose:
 114             print "auto-thread-state"
 115
 116         idents = []
 117
 118         def callback():
 119             idents.append(thread.get_ident())
 120
 121         _testcapi._test_thread_state(callback)
 122         a = b = callback
 123         time.sleep(1)
 124         # Check our main thread is in the list exactly 3 times.
 125         if idents.count(thread.get_ident()) != 3:
 126             raise test_support.TestFailed, \
 127                   "Couldn't find main thread correctly in the list"
 128
 129     if threading:
 130         import thread
 131         import time
 132         TestThreadState()
 133         t=threading.Thread(target=TestThreadState)
 134         t.start()
 135         t.join()
 136
 137     test_support.run_unittest(TestPendingCalls)
 138
 139 if __name__ == "__main__":
 140     test_main()