Skip to content

Commit

Permalink
test_thread uses support.sleeping_retry() (#93849)
Browse files Browse the repository at this point in the history
test_thread.test_count() now fails if it takes longer than
LONG_TIMEOUT seconds.
  • Loading branch information
vstinner authored Jun 15, 2022
1 parent d318346 commit bddbd80
Showing 1 changed file with 11 additions and 7 deletions.
18 changes: 11 additions & 7 deletions Lib/test/test_thread.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@

NUMTASKS = 10
NUMTRIPS = 3
POLL_SLEEP = 0.010 # seconds = 10 ms

_print_mutex = thread.allocate_lock()

Expand Down Expand Up @@ -121,19 +120,24 @@ def task():

with threading_helper.wait_threads_exit():
thread.start_new_thread(task, ())
while not started:
time.sleep(POLL_SLEEP)
for _ in support.sleeping_retry(support.LONG_TIMEOUT):
if started:
break
self.assertEqual(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.
# 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(POLL_SLEEP)

for _ in support.sleeping_retry(support.LONG_TIMEOUT):
if done:
break
support.gc_collect() # For PyPy or other GCs.
self.assertEqual(thread._count(), orig)

Expand Down

0 comments on commit bddbd80

Please sign in to comment.