Stowage/cpython
2
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-03 23:21:29 +00:00
Code Issues Projects Releases Packages Wiki Activity

Give test_multiprocessing better chance of avoiding timeout failures on Windows

Browse source
This commit is contained in:
Richard Oudkerk 2012-05-04 09:44:39 +01:00
parent 9bd9d74996
commit 009b15e2c3
1 changed files with 16 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

23
Lib/test/test_multiprocessing.py
View file

@ -2699,35 +2699,43 @@ def test_wait_socket_slow(self):
def test_wait_timeout(self):
from multiprocessing.connection import wait
expected = 1
expected = 5
a, b = multiprocessing.Pipe()
start = time.time()
res = wait([a, b], 1)
res = wait([a, b], expected)
delta = time.time() - start
self.assertEqual(res, [])
self.assertLess(delta, expected + 0.5)
self.assertGreater(delta, expected - 0.5)
self.assertLess(delta, expected + 1)
self.assertGreater(delta, expected - 1)
b.send(None)
start = time.time()
res = wait([a, b], 1)
res = wait([a, b], 20)
delta = time.time() - start
self.assertEqual(res, [a])
self.assertLess(delta, 0.4)
@classmethod
def signal_and_sleep(cls, sem, period):
sem.release()
time.sleep(period)
def test_wait_integer(self):
from multiprocessing.connection import wait
expected = 5
expected = 3
sem = multiprocessing.Semaphore(0)
a, b = multiprocessing.Pipe()
p = multiprocessing.Process(target=time.sleep, args=(expected,))
p = multiprocessing.Process(target=self.signal_and_sleep,
args=(sem, expected))
p.start()
self.assertIsInstance(p.sentinel, int)
self.assertTrue(sem.acquire(timeout=20))
start = time.time()
res = wait([a, p.sentinel, b], expected + 20)
@ -2755,6 +2763,7 @@ def test_wait_integer(self):
self.assertEqual(res, [a, p.sentinel, b])
self.assertLess(delta, 0.4)
p.terminate()
p.join()