Add sometimes-failing test about a race condition in SemLock.release()

1 files changed, 26 insertions, 0 deletions

diff --git a/extra_tests/test_semlock.py b/extra_tests/test_semlock.py
new file mode 100644
index 0000000000..712d550028
--- /dev/null
+++ b/extra_tests/test_semlock.py
@@ -0,0 +1,26 @@
+from _multiprocessing import SemLock
+from threading import Thread
+import time
+
+
+def test_notify_all():
+    """A low-level variation on test_notify_all() in lib-python's
+    test_multiprocessing.py
+    """
+    N_THREADS = 1000
+    lock = SemLock(0, 1, 1)
+    results = []
+
+    def f(n):
+        if lock.acquire(timeout=5.):
+            results.append(n)
+            lock.release()
+
+    threads = [Thread(target=f, args=(i,)) for i in range(N_THREADS)]
+    with lock:
+        for t in threads:
+            t.start()
+        time.sleep(0.1)
+    for t in threads:
+        t.join()
+    assert len(results) == N_THREADS