Issue #14428: Rewrite test_process_time_threads() test

time.process_time() is CPU time, not a number of seconds.
This commit is contained in:
Victor Stinner 2012-04-29 19:46:19 +02:00
parent b63fd2a408
commit 1fc3ec91cc
1 changed files with 41 additions and 8 deletions

View File

@ -392,20 +392,53 @@ class TimeTestCase(unittest.TestCase):
@unittest.skipUnless(threading,
'need threading')
def test_process_time_threads(self):
class BusyThread(threading.Thread):
def run(self):
while not self.stop:
pass
def factorial(n):
if n >= 2:
return n * factorial(n-1)
else:
return 1
thread = BusyThread()
thread.stop = False
def use_cpu(n, loops):
for loop in range(loops):
factorial(n)
class FactorialThread(threading.Thread):
def __init__(self, n, loops):
threading.Thread.__init__(self)
self.n = n
self.loops = loops
def run(self):
use_cpu(self.n, self.loops)
# Calibrate use_cpu() to use at least 1 ms of system time
n = 50
loops = 1
resolution = time.get_clock_info('process_time').resolution
min_rdt = max(resolution, 0.001)
while 1:
rt1 = time.time()
t1 = time.process_time()
use_cpu(n, loops)
t2 = time.process_time()
rt2 = time.time()
rdt = rt2 - rt1
if rdt >= min_rdt:
break
loops *= 2
busy = t2 - t1
# Ensure that time.process_time() includes the CPU time of all threads
thread = FactorialThread(n, loops)
t1 = time.process_time()
thread.start()
time.sleep(0.2)
# Use sleep() instead of thread.join() because thread.join() time may
# be included in time.process_time() depending on its implementation
time.sleep(rdt * 2)
t2 = time.process_time()
thread.stop = True
thread.join()
self.assertGreater(t2 - t1, 0.1)
self.assertGreaterEqual(t2 - t1, busy)
@unittest.skipUnless(hasattr(time, 'monotonic'),
'need time.monotonic')