bpo-10978: Semaphores can release multiple threads at a time (GH-15588)

This commit is contained in:
Raymond Hettinger 2019-08-29 01:45:19 -07:00 committed by GitHub
parent 0dac68f1e5
commit 35f6301d68
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 56 additions and 13 deletions

View File

@ -802,11 +802,14 @@ Semaphores also support the :ref:`context management protocol <with-locks>`.
.. versionchanged:: 3.2
The *timeout* parameter is new.
.. method:: release()
.. method:: release(n=1)
Release a semaphore, incrementing the internal counter by one. When it
was zero on entry and another thread is waiting for it to become larger
than zero again, wake up that thread.
Release a semaphore, incrementing the internal counter by *n*. When it
was zero on entry and other threads are waiting for it to become larger
than zero again, wake up *n* of those threads.
.. versionchanged:: 3.9
Added the *n* parameter to release multiple waiting threads at once.
.. class:: BoundedSemaphore(value=1)

View File

@ -663,6 +663,38 @@ class BaseSemaphoreTests(BaseTestCase):
b.wait_for_finished()
self.assertEqual(sem_results, [True] * (6 + 7 + 6 + 1))
def test_multirelease(self):
sem = self.semtype(7)
sem.acquire()
results1 = []
results2 = []
phase_num = 0
def f():
sem.acquire()
results1.append(phase_num)
sem.acquire()
results2.append(phase_num)
b = Bunch(f, 10)
b.wait_for_started()
while len(results1) + len(results2) < 6:
_wait()
self.assertEqual(results1 + results2, [0] * 6)
phase_num = 1
sem.release(7)
while len(results1) + len(results2) < 13:
_wait()
self.assertEqual(sorted(results1 + results2), [0] * 6 + [1] * 7)
phase_num = 2
sem.release(6)
while len(results1) + len(results2) < 19:
_wait()
self.assertEqual(sorted(results1 + results2), [0] * 6 + [1] * 7 + [2] * 6)
# The semaphore is still locked
self.assertFalse(sem.acquire(False))
# Final release, to let the last thread finish
sem.release()
b.wait_for_finished()
def test_try_acquire(self):
sem = self.semtype(2)
self.assertTrue(sem.acquire(False))

View File

@ -439,16 +439,19 @@ class Semaphore:
__enter__ = acquire
def release(self):
"""Release a semaphore, incrementing the internal counter by one.
def release(self, n=1):
"""Release a semaphore, incrementing the internal counter by one or more.
When the counter is zero on entry and another thread is waiting for it
to become larger than zero again, wake up that thread.
"""
if n < 1:
raise ValueError('n must be one or more')
with self._cond:
self._value += 1
self._cond.notify()
self._value += n
for i in range(n):
self._cond.notify()
def __exit__(self, t, v, tb):
self.release()
@ -475,8 +478,8 @@ class BoundedSemaphore(Semaphore):
Semaphore.__init__(self, value)
self._initial_value = value
def release(self):
"""Release a semaphore, incrementing the internal counter by one.
def release(self, n=1):
"""Release a semaphore, incrementing the internal counter by one or more.
When the counter is zero on entry and another thread is waiting for it
to become larger than zero again, wake up that thread.
@ -485,11 +488,14 @@ class BoundedSemaphore(Semaphore):
raise a ValueError.
"""
if n < 1:
raise ValueError('n must be one or more')
with self._cond:
if self._value >= self._initial_value:
if self._value + n > self._initial_value:
raise ValueError("Semaphore released too many times")
self._value += 1
self._cond.notify()
self._value += n
for i in range(n):
self._cond.notify()
class Event:

View File

@ -0,0 +1,2 @@
Semaphores and BoundedSemaphores can now release more than one waiting
thread at a time.