mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-03 16:39:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Also rewrote the guts of asyncio.Semaphore (patch by manipopopo).

Browse source
This commit is contained in:
Guido van Rossum 2015-09-29 11:54:45 -07:00
parent d94c1b92ed
commit d455a50773
3 changed files with 66 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

52
Lib/test/test_asyncio/test_locks.py
View file

@ -7,7 +7,6 @@ import re
import asyncio
from asyncio import test_utils
STR_RGX_REPR = (
r'^<(?P<class>.*?) object at (?P<address>.*?)'
r'\[(?P<extras>'
@ -783,22 +782,20 @@ class SemaphoreTests(test_utils.TestCase):
test_utils.run_briefly(self.loop)
self.assertEqual(0, sem._value)
self.assertEqual([1, 2, 3], result)
self.assertEqual(3, len(result))
self.assertTrue(sem.locked())
self.assertEqual(1, len(sem._waiters))
self.assertEqual(0, sem._value)
self.assertTrue(t1.done())
self.assertTrue(t1.result())
self.assertTrue(t2.done())
self.assertTrue(t2.result())
self.assertTrue(t3.done())
self.assertTrue(t3.result())
self.assertFalse(t4.done())
race_tasks = [t2, t3, t4]
done_tasks = [t for t in race_tasks if t.done() and t.result()]
self.assertTrue(2, len(done_tasks))
# cleanup locked semaphore
sem.release()
self.loop.run_until_complete(t4)
self.loop.run_until_complete(asyncio.gather(*race_tasks))
def test_acquire_cancel(self):
sem = asyncio.Semaphore(loop=self.loop)
@ -809,7 +806,44 @@ class SemaphoreTests(test_utils.TestCase):
self.assertRaises(
asyncio.CancelledError,
self.loop.run_until_complete, acquire)
self.assertFalse(sem._waiters)
self.assertTrue((not sem._waiters) or
all(waiter.done() for waiter in sem._waiters))
def test_acquire_cancel_before_awoken(self):
sem = asyncio.Semaphore(value=0, loop=self.loop)
t1 = asyncio.Task(sem.acquire(), loop=self.loop)
t2 = asyncio.Task(sem.acquire(), loop=self.loop)
t3 = asyncio.Task(sem.acquire(), loop=self.loop)
t4 = asyncio.Task(sem.acquire(), loop=self.loop)
test_utils.run_briefly(self.loop)
sem.release()
t1.cancel()
t2.cancel()
test_utils.run_briefly(self.loop)
num_done = sum(t.done() for t in [t3, t4])
self.assertEqual(num_done, 1)
t3.cancel()
t4.cancel()
test_utils.run_briefly(self.loop)
def test_acquire_hang(self):
sem = asyncio.Semaphore(value=0, loop=self.loop)
t1 = asyncio.Task(sem.acquire(), loop=self.loop)
t2 = asyncio.Task(sem.acquire(), loop=self.loop)
test_utils.run_briefly(self.loop)
sem.release()
t1.cancel()
test_utils.run_briefly(self.loop)
self.assertTrue(sem.locked())
def test_release_not_acquired(self):
sem = asyncio.BoundedSemaphore(loop=self.loop)