threading.RLock.locked() lets you check whether a reentrant lock is currently acquired. The method is small, but it is valuable for diagnostics, tests, metrics, and state inspection in concurrent applications. This guide explains what a reentrant lock is, how locked() behaves, where it helps, and why it should not be used as a substitute for atomic synchronization.
What is an RLock?
An RLock is a reentrant synchronization primitive from Python’s threading module. Unlike a basic Lock, the same thread may acquire it multiple times without immediately deadlocking itself. Internally, the lock tracks the owning thread and a recursion count. Every successful acquire() by the owner increases that count, and every release() decreases it. The lock becomes available to other threads only when the count reaches zero.
This behavior is useful when synchronized methods call other synchronized methods that protect the same state. For broader context, see Threading in Python and the Python GIL.
How locked() works
The locked() method returns True when the reentrant lock is acquired and False when it is free. It does not tell you which thread owns the lock, how many times it has been acquired, or how long it has remained held. Think of the result as a momentary snapshot.
import threading
lock = threading.RLock()
print(lock.locked()) # False
lock.acquire()
print(lock.locked()) # True
lock.release()
print(lock.locked()) # False
In production code, prefer a context manager:
with lock:
update_cache()
The with statement releases the lock even if the protected block raises an exception. This is the same resource-management principle discussed in using with in Python.
Useful use cases
locked() is useful in unit tests, internal health endpoints, debug logs, and observability dashboards. A service with a cache protected by an RLock may expose whether an update is currently running. A test can verify that a callback releases the lock after an exception. A diagnostic command can include lock state alongside queue size and worker count.
def service_state():
return {
"cache_update_running": cache_lock.locked(),
"queue_size": len(queue),
}
The value must still be treated as approximate. Another thread can acquire or release the lock immediately after the check.
Do not check before acting
A common mistake is to use locked() as permission to enter a critical section:
if not lock.locked():
lock.acquire()
try:
process()
finally:
lock.release()
This is a check-then-act race. The lock may be free during the check and acquired by another thread before the following operation. If the goal is to acquire without waiting, use the atomic nonblocking form:
if lock.acquire(blocking=False):
try:
process()
finally:
lock.release()
else:
record_busy_state()
Here, testing and acquisition happen in one synchronization operation.
Understanding the recursion count
Because an RLock is reentrant, locked() stays true until every acquisition has a matching release.
lock.acquire()
lock.acquire()
print(lock.locked()) # True
lock.release()
print(lock.locked()) # still True
lock.release()
print(lock.locked()) # False
This detail matters when debugging apparent lock leaks. A code path may call release() once and still leave the lock acquired because it was entered more than once. Tools such as faulthandler and pdb can help investigate stuck threads.
A thread-safe class example
import threading
class SafeCounter:
def __init__(self):
self._value = 0
self._lock = threading.RLock()
def increment(self):
with self._lock:
self._value += 1
self._log_value()
def _log_value(self):
with self._lock:
print(f"value={self._value}")
def busy(self):
return self._lock.locked()
increment() acquires the lock and then calls another method that acquires the same lock. This is exactly the kind of nested ownership that reentrancy supports. A basic lock could deadlock the thread against itself.
Testing exception safety
def fail_inside(lock):
with lock:
raise RuntimeError("simulated failure")
lock = threading.RLock()
try:
fail_inside(lock)
except RuntimeError:
pass
assert lock.locked() is False
This test confirms that the context manager releases the resource. For a broader testing workflow, see unit testing in Python.
Observability without unsafe decisions
Lock state is useful as a metric, but it should not be interpreted as a durable guarantee. In a busy service, the result can change before it is logged or transmitted. A boolean “busy” metric can reveal frequent contention, but it does not show how long threads wait or how long the critical section lasts.
For stronger observability, record timestamps before and after acquire(), measure wait time, and measure hold duration. Track percentiles rather than only averages. Long hold times often indicate excessive work inside the critical section, blocking I/O, or nested calls that should be moved outside the protected block.
Lock versus RLock
Use Lock when reentrancy is unnecessary. It is simpler and can expose design problems earlier. Use RLock when the same thread must safely cross multiple synchronized layers that share one protected resource. Both primitives can report whether they are locked, but an RLock requires you to remember its ownership count.
Important limitations
Do not use locked() to decide whether shared data can be accessed safely. Do not inspect private CPython attributes to discover the owner or recursion level. Do not scatter manual release() calls across many branches. Prefer narrow with blocks and document which fields each lock protects. Keep slow network calls, disk I/O, and callbacks outside the critical section whenever possible.
Compatibility and official references
Libraries supporting multiple Python versions should verify availability in their test matrix before relying on this method. The official threading documentation describes synchronization primitives, ownership rules, and context-manager behavior. The Python 3.14 release notes provide version-specific changes.
Practical checklist
Use locked() for diagnostics and tests. Use acquire() or with for synchronization. Use blocking=False for an atomic try-lock operation. Match every manual acquisition with a release. Keep critical sections small. Add timing metrics when contention matters. Test exception paths explicitly. Avoid depending on implementation details.
Conclusion
threading.RLock.locked() provides a clear, lightweight view of whether a reentrant lock is currently held. Its best uses are diagnostics, observability, and assertions. It is not a transaction boundary and should never replace proper acquisition. By separating state inspection from synchronization, using context managers, and respecting the recursion count, you can build concurrent Python code that is easier to test, debug, and maintain.
