mirror of
https://github.com/RustPython/RustPython.git
synced 2026-10-29 03:04:38 +08:00
104 lines
2.8 KiB
Python
104 lines
2.8 KiB
Python
"""Thread+process Condition notify must wake every waiter."""
|
|
|
|
import multiprocessing
|
|
import threading
|
|
from _multiprocessing import SemLock
|
|
|
|
|
|
def wait_on_condition(cond, started, woken):
|
|
cond.acquire()
|
|
started.release()
|
|
cond.wait()
|
|
woken.release()
|
|
cond.release()
|
|
|
|
|
|
def join_workers(workers):
|
|
for worker in workers:
|
|
worker.join(timeout=5)
|
|
assert not worker.is_alive()
|
|
|
|
|
|
def wait_sem(sem, msg):
|
|
assert sem.acquire(timeout=5), msg
|
|
|
|
|
|
def notify_all_six(ctx):
|
|
cond = ctx.Condition()
|
|
started = ctx.Semaphore(0)
|
|
woken = ctx.Semaphore(0)
|
|
|
|
workers = []
|
|
for _ in range(3):
|
|
proc = ctx.Process(target=wait_on_condition, args=(cond, started, woken))
|
|
proc.daemon = True
|
|
proc.start()
|
|
workers.append(proc)
|
|
|
|
thread = threading.Thread(target=wait_on_condition, args=(cond, started, woken))
|
|
thread.daemon = True
|
|
thread.start()
|
|
workers.append(thread)
|
|
|
|
for i in range(6):
|
|
wait_sem(started, f"waiter {i} did not enter Condition.wait()")
|
|
|
|
cond.acquire()
|
|
cond.notify_all()
|
|
cond.release()
|
|
|
|
for i in range(6):
|
|
wait_sem(woken, f"waiter {i} did not wake")
|
|
join_workers(workers)
|
|
|
|
|
|
def notify_one_then_the_other(ctx):
|
|
cond = ctx.Condition()
|
|
started = ctx.Semaphore(0)
|
|
woken = ctx.Semaphore(0)
|
|
|
|
proc = ctx.Process(target=wait_on_condition, args=(cond, started, woken))
|
|
proc.daemon = True
|
|
proc.start()
|
|
thread = threading.Thread(target=wait_on_condition, args=(cond, started, woken))
|
|
thread.daemon = True
|
|
thread.start()
|
|
|
|
wait_sem(started, "process did not enter Condition.wait()")
|
|
wait_sem(started, "thread did not enter Condition.wait()")
|
|
|
|
cond.acquire()
|
|
cond.notify()
|
|
cond.release()
|
|
wait_sem(woken, "first waiter did not wake")
|
|
|
|
cond.acquire()
|
|
cond.notify()
|
|
cond.release()
|
|
wait_sem(woken, "second waiter did not wake")
|
|
join_workers([proc, thread])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
multiprocessing.freeze_support()
|
|
# Low-level callers can supply every constructor parameter by name.
|
|
name = f"/rustpython_kw_{multiprocessing.current_process().pid}"
|
|
for args, kwargs in (
|
|
((1, 1, 1), {"name": name, "unlink": True}),
|
|
((), {"kind": 1, "value": 1, "maxvalue": 1, "name": name, "unlink": True}),
|
|
):
|
|
sem = SemLock(*args, **kwargs)
|
|
assert sem.acquire(False)
|
|
assert not sem.acquire(False)
|
|
sem.release()
|
|
del sem
|
|
methods = ["spawn"]
|
|
if "forkserver" in multiprocessing.get_all_start_methods():
|
|
methods.append("forkserver")
|
|
# Enough repeats to surface a lost-wakeup race in CI snippets.
|
|
for method in methods:
|
|
ctx = multiprocessing.get_context(method)
|
|
for _ in range(20):
|
|
notify_all_six(ctx)
|
|
notify_one_then_the_other(ctx)
|