Mirror
/
cpython
mirror of https://github.com/python/cpython
4
1
Fork 0
Code Issues Releases Wiki Activity

Speed test_threading up from 14s to .5s, and avoid a deadlock on certain 

failures. The test for enumerate-after-join is now a little less rigorous, but
the bug it references says the error happened in the first couple iterations,
so 100 iterations should still be enough.

cProfile was useful for identifying the slow tests here.
This commit is contained in:
Jeffrey Yasskin 2008-03-21 18:48:04 +00:00
parent 413f588a36
commit 510eab5965
1 changed files with 28 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
Lib/test/test_threading.py
View File

@ -30,32 +30,28 @@ class TestThread(threading.Thread):
self.nrunning = nrunning self.nrunning = nrunning
def run(self): def run(self):
delay = random.random() * 2 delay = random.random() / 10000.0
if verbose: if verbose:
print 'task', self.getName(), 'will run for', delay, 'sec' print 'task %s will run for %.1f usec' % (
self.getName(), delay * 1e6)
self.sema.acquire() with self.sema:
with self.mutex:
self.mutex.acquire()
self.nrunning.inc() self.nrunning.inc()
if verbose: if verbose:
print self.nrunning.get(), 'tasks are running' print self.nrunning.get(), 'tasks are running'
self.testcase.assert_(self.nrunning.get() <= 3) self.testcase.assert_(self.nrunning.get() <= 3)
self.mutex.release()
time.sleep(delay) time.sleep(delay)
if verbose: if verbose:
print 'task', self.getName(), 'done' print 'task', self.getName(), 'done'
self.mutex.acquire() with self.mutex:
self.nrunning.dec() self.nrunning.dec()
self.testcase.assert_(self.nrunning.get() >= 0) self.testcase.assert_(self.nrunning.get() >= 0)
if verbose: if verbose:
print self.getName(), 'is finished.', self.nrunning.get(), \ print '%s is finished. %d tasks are running' % (
'tasks are running' self.getName(), self.nrunning.get())
self.mutex.release()
self.sema.release()
class ThreadTests(unittest.TestCase): class ThreadTests(unittest.TestCase):
@ -218,6 +214,10 @@ class ThreadTests(unittest.TestCase):
rc = subprocess.call([sys.executable, "-c", """if 1: rc = subprocess.call([sys.executable, "-c", """if 1:
import ctypes, sys, time, thread import ctypes, sys, time, thread
# This lock is used as a simple event variable.
ready = thread.allocate_lock()
ready.acquire()
# Module globals are cleared before __del__ is run # Module globals are cleared before __del__ is run
# So we save the functions in class dict # So we save the functions in class dict
class C: class C:
@ -229,10 +229,11 @@ class ThreadTests(unittest.TestCase):
def waitingThread(): def waitingThread():
x = C() x = C()
ready.release()
time.sleep(100) time.sleep(100)
thread.start_new_thread(waitingThread, ()) thread.start_new_thread(waitingThread, ())
time.sleep(1) # be sure the other thread is waiting ready.acquire() # Be sure the other thread is waiting.
sys.exit(42) sys.exit(42)
"""]) """])
self.assertEqual(rc, 42) self.assertEqual(rc, 42)
@ -242,9 +243,11 @@ class ThreadTests(unittest.TestCase):
# threading.enumerate() after it has been join()ed. # threading.enumerate() after it has been join()ed.
enum = threading.enumerate enum = threading.enumerate
old_interval = sys.getcheckinterval() old_interval = sys.getcheckinterval()
sys.setcheckinterval(1)
try: try:
for i in xrange(1, 1000): for i in xrange(1, 100):
# Try a couple times at each thread-switching interval
# to get more interleavings.
sys.setcheckinterval(i // 5)
t = threading.Thread(target=lambda: None) t = threading.Thread(target=lambda: None)
t.start() t.start()
t.join() t.join()