asyncio.eager_task_factory permite que corrotinas comecem a executar imediatamente no momento em que uma tarefa é criada, antes de serem agendadas no loop de eventos. Essa mudança pode reduzir overhead em operações muito curtas, mas também altera a ordem de execução esperada e exige cuidado com efeitos colaterais, exceções e testes.
O que é execução eager
No comportamento tradicional do asyncio, criar uma tarefa com asyncio.create_task() agenda a corrotina para executar em uma próxima oportunidade do event loop. Com uma task factory eager, o início acontece de forma síncrona durante a criação da tarefa. Se a corrotina terminar sem precisar suspender, ela pode concluir sem sequer entrar na fila normal do loop.
Esse comportamento é útil para corrotinas que frequentemente retornam de cache, validam dados em memória ou concluem rapidamente. Em sistemas com milhares de chamadas pequenas, evitar o custo de agendamento pode representar uma melhoria mensurável. Porém, o ganho não deve ser presumido: ele precisa ser confirmado com benchmarks próximos da carga real.
Como ativar a task factory
import asyncio
async def buscar_cache(chave, cache):
if chave in cache:
return cache[chave]
await asyncio.sleep(0.01)
return None
async def main():
loop = asyncio.get_running_loop()
loop.set_task_factory(asyncio.eager_task_factory)
cache = {"curso": "Python"}
tarefa = asyncio.create_task(buscar_cache("curso", cache))
resultado = await tarefa
print(resultado)
asyncio.run(main())A configuração é aplicada ao loop atual. A partir daí, tarefas criadas nesse loop usam a factory configurada. Em aplicações maiores, é recomendável centralizar essa decisão na inicialização para evitar ambientes inconsistentes.
Quando a corrotina entra no event loop
Se a corrotina executada de forma eager encontra um await que realmente precisa aguardar, a tarefa passa a ser agendada normalmente. Portanto, o comportamento eager acelera principalmente o trecho inicial e as corrotinas que conseguem terminar sem bloqueio assíncrono.
Esse detalhe é importante em funções que consultam cache primeiro e só depois fazem I/O. Uma resposta em cache pode terminar imediatamente, enquanto uma ausência de cache segue pelo fluxo normal do loop.
Diferenças na ordem de execução
A principal mudança semântica está na ordem. Código que antes criava uma tarefa e executava outra instrução antes de a tarefa começar pode deixar de se comportar assim.
import asyncio
async def trabalho():
print("corrotina iniciou")
return 42
async def main():
loop = asyncio.get_running_loop()
loop.set_task_factory(asyncio.eager_task_factory)
print("antes")
tarefa = asyncio.create_task(trabalho())
print("depois")
print(await tarefa)
asyncio.run(main())Com execução eager, a mensagem da corrotina pode aparecer antes de depois. Isso pode expor dependências ocultas de ordem, especialmente em código com logs, métricas, mutação de estado compartilhado ou callbacks.
Tratamento de exceções
Exceções também podem surgir mais cedo. Uma corrotina que falha antes do primeiro bloqueio pode produzir o erro durante a criação e execução inicial da tarefa. Por isso, testes devem cobrir tanto caminhos que concluem imediatamente quanto caminhos que suspendem.
async def validar(valor):
if valor < 0:
raise ValueError("valor inválido")
return valorEm aplicações com grupos de tarefas, avalie como o comportamento interage com asyncio.Runner, cancelamento e encerramento limpo. Também vale revisar padrões de sincronização descritos em asyncio.Barrier.
Uso com caches
O caso clássico é uma função assíncrona que consulta um cache local antes de acessar rede ou banco de dados. Quando há acerto, a corrotina termina de forma síncrona; quando não há, ela entra no fluxo assíncrono normal.
async def obter_usuario(user_id, cache, cliente):
if user_id in cache:
return cache[user_id]
resposta = await cliente.get(f"/users/{user_id}")
cache[user_id] = resposta
return respostaEsse padrão pode reduzir overhead, mas o benefício depende da taxa de acerto e do volume. Faça medições antes e depois, usando dados representativos.
Riscos de estado compartilhado
Como a tarefa pode executar imediatamente, qualquer alteração em listas, dicionários, contadores ou objetos globais pode ocorrer antes do esperado. Prefira funções com efeitos colaterais explícitos, estados imutáveis quando possível e mecanismos de sincronização claros.
Para filas entre threads, veja queue.SimpleQueue. Para processamento paralelo de CPU, consulte InterpreterPoolExecutor.
Benchmark correto
Um benchmark útil deve separar três cenários: conclusão imediata, suspensão curta e I/O dominante. Em tarefas dominadas por rede, a diferença pode ser irrelevante. Em milhões de chamadas de cache, pode ser significativa.
import asyncio
import time
async def imediato():
return 1
async def medir(n):
inicio = time.perf_counter()
tarefas = [asyncio.create_task(imediato()) for _ in range(n)]
await asyncio.gather(*tarefas)
return time.perf_counter() - inicioRepita o teste, descarte aquecimento, compare versões do Python e observe uso de CPU e memória.
Testes recomendados
Crie testes que verifiquem ordem de logs, exceções antes do primeiro await, cancelamento, callbacks de conclusão e resultados de cache. Não teste apenas o valor final: teste também a sequência de eventos quando ela fizer parte do contrato.
Boas práticas
Ative a factory em um ponto central, documente a mudança, não dependa de ordem acidental, mantenha benchmarks e use observabilidade. O recurso deve ser tratado como otimização semântica, não apenas como uma opção de desempenho.
A documentação oficial de tarefas asyncio descreve a API e seus detalhes. Para compreender o modelo geral de concorrência, consulte também a documentação do asyncio.
Conclusão
asyncio.eager_task_factory pode reduzir o custo de tarefas curtas e caminhos de cache, mas muda quando a corrotina começa. Use o recurso quando houver ganho comprovado, testes sólidos e código preparado para uma ordem de execução mais imediata.
