bpo-10978: Semaphores can release multiple threads at a time (GH-15588)
diff --git a/Lib/test/lock_tests.py b/Lib/test/lock_tests.py
index 23a02e0..8885868 100644
--- a/Lib/test/lock_tests.py
+++ b/Lib/test/lock_tests.py
@@ -663,6 +663,38 @@
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))