Concurrency Coordination Reports
Lock Guard Coordination Report
Try a nonblocking lock several times and report whether the guard stayed open or blocked a later attempt.
Lock Guard Coordination Report
lock_guard_coordination_report.py
import threading
holds =
guard = threading.Lock()
acquired = 0
blocked = 0
for _ in range(holds):
if guard.acquire(blocking=False):
acquired += 1
else:
blocked += 1
status = "open" if acquired == 0 else "blocked" if blocked > 0 else "guarded"
if acquired > 0:
guard.release()
print("holds=" + str(holds) + " acquired=" + str(acquired) + " blocked=" + str(blocked) + " status=" + status)
import threading
holds =
guard = threading.Lock()
acquired = 0
blocked = 0
for _ in range(holds):
if guard.acquire(blocking=False):
acquired += 1
else:
blocked += 1
status = "open" if acquired == 0 else "blocked" if blocked > 0 else "guarded"
if acquired > 0:
guard.release()
print("holds=" + str(holds) + " acquired=" + str(acquired) + " blocked=" + str(blocked) + " status=" + status)
import threading
holds =
guard = threading.Lock()
acquired = 0
blocked = 0
for _ in range(holds):
if guard.acquire(blocking=False):
acquired += 1
else:
blocked += 1
status = "open" if acquired == 0 else "blocked" if blocked > 0 else "guarded"
if acquired > 0:
guard.release()
print("holds=" + str(holds) + " acquired=" + str(acquired) + " blocked=" + str(blocked) + " status=" + status)
nonblocking lock
A `threading.Lock` acquired without blocking reports whether the guard was free, so a second attempt that fails shows the lock is already held.