O método threading.RLock.locked() permite verificar se um bloqueio reentrante está atualmente adquirido. Embora pareça uma operação simples, ele ajuda bastante em diagnósticos, testes, métricas e validações de estado em aplicações concorrentes. Neste guia, você vai entender como o método funciona, quando ele é útil, quais cuidados tomar e como combiná-lo com boas práticas de sincronização.
O que é um RLock?
Um RLock, ou bloqueio reentrante, é uma primitiva de sincronização do módulo threading. Diferentemente de um Lock comum, ele pode ser adquirido várias vezes pela mesma thread sem causar deadlock imediato. Internamente, o objeto mantém o identificador da thread proprietária e um contador de aquisições. Cada chamada a acquire() incrementa esse contador, e cada chamada a release() o reduz. O bloqueio só fica realmente livre quando o contador retorna a zero.
Esse comportamento é útil quando métodos sincronizados chamam outros métodos sincronizados da mesma classe. Sem reentrância, a thread poderia bloquear a si própria. Para uma introdução mais ampla a concorrência, veja também Threading no Python e o que é o GIL.
Como funciona RLock.locked()
O método locked() retorna True quando o objeto está adquirido e False quando está livre. Ele não informa qual thread possui o bloqueio, quantas vezes ele foi adquirido ou por quanto tempo está retido. Portanto, deve ser entendido como uma fotografia momentânea do estado global do objeto.
import threading
lock = threading.RLock()
print(lock.locked()) # False
lock.acquire()
print(lock.locked()) # True
lock.release()
print(lock.locked()) # False
Em código real, prefira usar o gerenciador de contexto:
with lock:
# seção crítica
atualizar_cache()
O with garante a liberação mesmo quando ocorre uma exceção. Esse padrão segue as boas práticas apresentadas em como usar with no Python, ainda que o recurso sincronizado aqui seja um lock e não um arquivo.
Quando usar o método
locked() é especialmente útil em testes automatizados, endpoints de diagnóstico, logs de observabilidade e verificações defensivas. Imagine um serviço que possui um cache protegido por RLock. Um painel interno pode exibir se uma atualização está em andamento. Da mesma forma, um teste pode confirmar que determinado callback libera o bloqueio mesmo quando lança uma exceção.
def estado_do_servico():
return {
"cache_em_atualizacao": cache_lock.locked(),
"fila": len(fila),
}
Apesar disso, o valor não deve ser usado como autorização para entrar em uma seção crítica. Entre a leitura de locked() e uma ação posterior, outra thread pode adquirir ou liberar o bloqueio. Esse é um exemplo clássico de condição de corrida do tipo “verificar e depois agir”.
O erro de verificar antes de adquirir
Evite este padrão:
if not lock.locked():
lock.acquire()
try:
processar()
finally:
lock.release()
Mesmo que locked() retorne False, outra thread pode adquirir o objeto antes da chamada a acquire(). O código ainda pode funcionar porque acquire() bloqueará, mas a verificação anterior não oferece garantia alguma. Se a intenção for não esperar, use a própria operação atômica:
if lock.acquire(blocking=False):
try:
processar()
finally:
lock.release()
else:
registrar_ocupado()
Essa abordagem combina teste e aquisição em uma única operação segura.
Reentrância e contador interno
Uma diferença importante é que locked() continua retornando True enquanto o contador interno for maior que zero. Se a mesma thread adquirir o RLock duas vezes, será necessário liberar duas vezes.
lock.acquire()
lock.acquire()
print(lock.locked()) # True
lock.release()
print(lock.locked()) # ainda True
lock.release()
print(lock.locked()) # False
Esse detalhe é essencial ao depurar travamentos. Um código pode aparentemente chamar release() e ainda manter o bloqueio, porque houve mais aquisições do que liberações. Para investigar esse tipo de problema, recursos como faulthandler e pdb também podem ajudar.
Exemplo em uma classe thread-safe
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()
O método incrementar() chama _registrar(), e ambos adquirem o mesmo bloqueio. Isso funciona porque o lock é reentrante. Com um Lock simples, a mesma thread poderia ficar parada aguardando um recurso que ela própria já possui.
Testando a liberação após exceções
def executar_com_falha(lock):
with lock:
raise RuntimeError("falha simulada")
lock = threading.RLock()
try:
executar_com_falha(lock)
except RuntimeError:
pass
assert lock.locked() is False
Esse teste verifica uma propriedade importante: o gerenciador de contexto libera corretamente o recurso. Em suítes maiores, combine essa abordagem com práticas descritas em testes unitários no Python.
Observabilidade sem decisões perigosas
O método funciona bem para métricas, mas os dados devem ser interpretados como aproximados. Em sistemas muito concorrentes, o estado pode mudar logo após a leitura. Uma métrica como “lock ocupado” é útil para detectar contenção frequente, porém não substitui medições de duração, tamanho de fila ou tempo médio de espera.
Para observabilidade mais robusta, registre o instante antes de chamar acquire(), calcule o tempo de espera depois da aquisição e meça quanto tempo a seção crítica permanece ativa. Esses valores ajudam a localizar gargalos reais.
Diferença entre Lock e RLock
Use Lock quando não houver necessidade de reentrância. Ele é conceitualmente mais simples e reduz o risco de esconder uma arquitetura confusa. Use RLock quando uma mesma thread precisar atravessar camadas sincronizadas que reutilizam o mesmo recurso. O método locked() é útil nos dois contextos, mas a interpretação do RLock exige lembrar do contador de aquisições.
Cuidados importantes
Não use locked() para decidir se é seguro acessar dados compartilhados. Não tente descobrir a thread proprietária por meios privados. Não dependa de atributos internos do CPython, pois eles podem mudar. Evite liberar o bloqueio manualmente em muitos pontos diferentes; prefira concentrar aquisição e liberação em blocos with. Por fim, documente claramente quais atributos são protegidos por cada lock.
Compatibilidade e documentação
Antes de usar o método em bibliotecas distribuídas para várias versões do Python, confira a versão mínima suportada e execute testes na sua matriz de ambientes. A referência oficial de threading descreve o comportamento das primitivas de sincronização. Também vale consultar o What’s New do Python 3.14 para mudanças recentes.
Conclusão
threading.RLock.locked() é uma ferramenta pequena, mas útil para diagnóstico, testes e observabilidade. Seu retorno representa apenas o estado naquele instante e não deve ser usado para implementar uma sequência “verificar e agir”. Para controlar concorrência com segurança, use acquire(), preferencialmente por meio de with, e trate locked() como informação auxiliar. Com essa distinção, você consegue depurar locks reentrantes, validar liberações e construir serviços concorrentes mais previsíveis.
