diff options
author | Antoine Pitrou <solipsis@pitrou.net> | 2009-10-30 17:25:12 +0000 |
---|---|---|
committer | Antoine Pitrou <solipsis@pitrou.net> | 2009-10-30 17:25:12 +0000 |
commit | 65c9c6426b719d885f12938a2c5ab938558406ab (patch) | |
tree | 4e4dbbf398d78a6bf2003d3a56635b9f0fa7582d /Lib/test/test_thread.py | |
parent | a2d1fe0b843b27130d0e3194ac2f517692e89c85 (diff) | |
download | cpython-git-65c9c6426b719d885f12938a2c5ab938558406ab.tar.gz |
Merged revisions 75958 via svnmerge from
svn+ssh://pythondev@svn.python.org/python/trunk
........
r75958 | antoine.pitrou | 2009-10-30 18:07:08 +0100 (ven., 30 oct. 2009) | 7 lines
Issue #7222: Make thread "reaping" more reliable so that reference
leak-chasing test runs give sensible results. The previous method of
reaping threads could return successfully while some Thread objects were
still referenced. This also introduces a new private function:
:func:hread._count().
........
Diffstat (limited to 'Lib/test/test_thread.py')
-rw-r--r-- | Lib/test/test_thread.py | 27 |
1 files changed, 27 insertions, 0 deletions
diff --git a/Lib/test/test_thread.py b/Lib/test/test_thread.py index 73d87b8cf3..c25668d781 100644 --- a/Lib/test/test_thread.py +++ b/Lib/test/test_thread.py @@ -4,6 +4,7 @@ import random from test import support import _thread as thread import time +import weakref NUMTASKS = 10 @@ -99,6 +100,32 @@ class ThreadRunningTests(BasicThreadTest): thread.stack_size(0) + def test__count(self): + # Test the _count() function. + orig = thread._count() + mut = thread.allocate_lock() + mut.acquire() + started = [] + def task(): + started.append(None) + mut.acquire() + mut.release() + thread.start_new_thread(task, ()) + while not started: + time.sleep(0.01) + self.assertEquals(thread._count(), orig + 1) + # Allow the task to finish. + mut.release() + # The only reliable way to be sure that the thread ended from the + # interpreter's point of view is to wait for the function object to be + # destroyed. + done = [] + wr = weakref.ref(task, lambda _: done.append(None)) + del task + while not done: + time.sleep(0.01) + self.assertEquals(thread._count(), orig) + class Barrier: def __init__(self, num_threads): |