ada/cpython
0
0
Fork 0
mirror of https://github.com/AdaCore/cpython.git synced 2026-02-12 12:57:15 -08:00
Code Issues Packages Projects Releases Wiki Activity

Use test.support.start_threads() in threaded lru_cache tests.

Browse Source
This commit is contained in:
Serhiy Storchaka
2015-05-30 15:49:42 +03:00
parent 2806518e0c bf2b3b72d3
commit f49c42324f
1 changed files with 9 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
Lib/test/test_functools.py
View File

@@ -1120,14 +1120,10 @@ class TestLRU:
sys.setswitchinterval(1e-6)
try:
# create 5 threads in order to fill cache
threads = []
for k in range(5):
t = threading.Thread(target=full, args=[f, k, k])
t.start()
threads.append(t)
for t in threads:
t.join()
threads = [threading.Thread(target=full, args=[f, k, k])
for k in range(5)]
with support.start_threads(threads):
pass
hits, misses, maxsize, currsize = f.cache_info()
self.assertEqual(hits, 45)
@@ -1135,16 +1131,11 @@ class TestLRU:
self.assertEqual(currsize, 5)
# create 5 threads in order to fill cache and 1 to clear it
cleaner = threading.Thread(target=clear, args=[f])
cleaner.start()
threads = [cleaner]
for k in range(5):
t = threading.Thread(target=full, args=[f, k, k])
t.start()
threads.append(t)
for t in threads:
t.join()
threads = [threading.Thread(target=clear, args=[f])]
threads += [threading.Thread(target=full, args=[f, k, k])
for k in range(5)]
with support.start_threads(threads):
pass
finally:
sys.setswitchinterval(orig_si)