threading.RLock.locked() permite comprobar si un bloqueo reentrante está adquirido en ese momento. Aunque es un método pequeño, resulta útil para diagnósticos, pruebas, métricas y observabilidad en aplicaciones concurrentes. En esta guía aprenderás qué es un RLock, cómo funciona locked(), cuándo conviene usarlo y por qué no debe reemplazar una operación de sincronización atómica.
Qué es un RLock
Un RLock es un bloqueo reentrante del módulo threading. A diferencia de un Lock básico, la misma hebra puede adquirirlo varias veces sin bloquearse inmediatamente a sí misma. El objeto mantiene internamente la hebra propietaria y un contador de adquisiciones. Cada llamada correcta a acquire() incrementa el contador y cada release() lo reduce. El recurso solo queda disponible para otras hebras cuando el contador vuelve a cero.
Este comportamiento es útil cuando un método sincronizado llama a otro método de la misma clase que protege el mismo estado. Para ampliar el contexto, consulta Threading en Python y qué es el GIL.
Cómo funciona locked()
El método devuelve True cuando el bloqueo está adquirido y False cuando está libre. No indica qué hebra lo posee, cuántas veces fue adquirido ni cuánto tiempo lleva retenido. Por eso, el resultado debe interpretarse como una fotografía instantánea.
import threading
lock = threading.RLock()
print(lock.locked()) # False
lock.acquire()
print(lock.locked()) # True
lock.release()
print(lock.locked()) # False
En código real, es preferible usar un gestor de contexto:
with lock:
actualizar_cache()
La sentencia with garantiza la liberación incluso si ocurre una excepción. Es el mismo principio de gestión segura de recursos explicado en cómo usar with en Python.
Casos de uso útiles
locked() es útil en pruebas unitarias, endpoints internos de salud, registros de depuración y paneles de observabilidad. Un servicio con una caché protegida por un RLock puede mostrar si existe una actualización en curso. Una prueba puede confirmar que un callback libera el bloqueo después de una excepción. Un comando de diagnóstico puede incluir este estado junto con el tamaño de una cola y la cantidad de workers.
def estado_servicio():
return {
"cache_actualizandose": cache_lock.locked(),
"tamano_cola": len(cola),
}
Sin embargo, el valor es aproximado. Otra hebra puede adquirir o liberar el recurso inmediatamente después de la consulta.
No comprobar antes de actuar
Un error común consiste en usar locked() como permiso para entrar en una sección crítica:
if not lock.locked():
lock.acquire()
try:
procesar()
finally:
lock.release()
Este patrón sufre una condición de carrera de tipo “comprobar y luego actuar”. El bloqueo puede estar libre durante la consulta y ser adquirido por otra hebra antes de la siguiente instrucción. Si deseas intentar adquirirlo sin esperar, usa la operación atómica:
if lock.acquire(blocking=False):
try:
procesar()
finally:
lock.release()
else:
registrar_ocupado()
Así, la comprobación y la adquisición forman parte de una única operación de sincronización.
El contador de reentrancia
Como RLock es reentrante, locked() seguirá devolviendo True hasta que todas las adquisiciones tengan su liberación correspondiente.
lock.acquire()
lock.acquire()
print(lock.locked()) # True
lock.release()
print(lock.locked()) # todavía True
lock.release()
print(lock.locked()) # False
Este detalle es importante al investigar bloqueos aparentes. Un flujo puede llamar una vez a release() y mantener el recurso ocupado porque fue adquirido dos veces. Herramientas como faulthandler y pdb ayudan a diagnosticar hebras detenidas.
Ejemplo de clase segura para hebras
import threading
class ContadorSeguro:
def __init__(self):
self._valor = 0
self._lock = threading.RLock()
def incrementar(self):
with self._lock:
self._valor += 1
self._registrar()
def _registrar(self):
with self._lock:
print(f"valor={self._valor}")
def ocupado(self):
return self._lock.locked()
incrementar() adquiere el bloqueo y llama a otro método que vuelve a adquirirlo. La reentrancia permite esta composición. Con un Lock normal, la hebra podría quedar esperando un recurso que ella misma posee.
Probar la liberación tras excepciones
def fallar_dentro(lock):
with lock:
raise RuntimeError("fallo simulado")
lock = threading.RLock()
try:
fallar_dentro(lock)
except RuntimeError:
pass
assert lock.locked() is False
La prueba confirma que el gestor de contexto libera el recurso. Para diseñar una suite más completa, revisa pruebas unitarias en Python.
Observabilidad sin decisiones inseguras
El estado del bloqueo sirve como métrica, pero no representa una garantía duradera. En un servicio con mucha concurrencia puede cambiar antes de ser registrado o enviado. Una métrica booleana permite detectar contención frecuente, pero no muestra cuánto esperan las hebras ni cuánto dura la sección crítica.
Para una observabilidad mejor, registra el instante anterior y posterior a acquire(), calcula el tiempo de espera y mide cuánto permanece adquirido el recurso. Los percentiles suelen ser más informativos que un promedio. Tiempos largos pueden indicar trabajo excesivo dentro de la sección crítica, operaciones de red, acceso a disco o callbacks lentos.
Lock frente a RLock
Usa Lock cuando no necesites reentrancia. Es más simple y puede revelar problemas de diseño. Usa RLock cuando la misma hebra deba atravesar varias capas sincronizadas que comparten el mismo recurso. En ambos casos, locked() informa el estado actual, pero en un bloqueo reentrante debes recordar que existe un contador interno.
Limitaciones importantes
No uses locked() para decidir si puedes acceder a datos compartidos sin protección. No dependas de atributos privados de CPython para descubrir el propietario o el contador. Evita distribuir llamadas manuales a release() por muchas ramas. Prefiere bloques with pequeños, documenta qué datos protege cada lock y mueve operaciones lentas fuera de la sección crítica.
Compatibilidad y referencias oficiales
Las bibliotecas compatibles con varias versiones de Python deben verificar la disponibilidad del método en su matriz de pruebas. La documentación oficial de threading explica las primitivas de sincronización y sus reglas. Las notas de Python 3.14 describen cambios recientes.
Lista práctica
Usa locked() para diagnósticos y pruebas. Usa acquire() o with para sincronizar. Usa blocking=False para intentar adquirir de forma atómica. Empareja cada adquisición manual con una liberación. Mantén las secciones críticas pequeñas. Mide tiempos cuando exista contención. Prueba explícitamente las rutas con excepciones y evita detalles internos de implementación.
Conclusión
threading.RLock.locked() ofrece una forma clara de observar si un bloqueo reentrante está adquirido. Su mejor uso está en diagnósticos, métricas y aserciones. No sustituye una operación de adquisición ni convierte una secuencia de instrucciones en atómica. Al separar observación y sincronización, usar gestores de contexto y respetar el contador de reentrancia, puedes construir código concurrente más predecible, fácil de probar y sencillo de mantener.
