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().

1 files changed, 27 insertions, 0 deletions

diff --git a/Lib/test/test_thread.py b/Lib/test/test_thread.py
index 66ad22f25a..4945047f6d 100644
--- a/Lib/test/test_thread.py
+++ b/Lib/test/test_thread.py
@@ -4,6 +4,7 @@ import random
 from test import test_support
 import thread
 import time
+import weakref
 
 
 NUMTASKS = 10
@@ -101,6 +102,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):