O parâmetro eager_start em asyncio.TaskGroup.create_task() oferece um controle mais explícito sobre o momento em que uma corrotina começa a executar. Em aplicações assíncronas, detalhes de agendamento podem alterar a ordem de efeitos colaterais, o momento em que erros aparecem e até o desempenho de operações muito curtas. Entender esse recurso ajuda a escrever concorrência estruturada com menos surpresas.
Neste guia, você verá o que significa iniciar uma tarefa de forma ansiosa, como isso se relaciona com TaskGroup, quando o recurso é útil, quais riscos devem ser considerados e como testar seu comportamento. O objetivo não é apenas mostrar a sintaxe, mas explicar como tomar decisões seguras em sistemas reais.
O papel do TaskGroup
asyncio.TaskGroup organiza várias tarefas como uma unidade. Ao entrar no bloco, você cria tarefas; ao sair, o grupo espera todas terminarem. Se uma delas falhar, as demais são canceladas de modo coordenado e os erros são reunidos. Esse modelo reduz tarefas órfãs e torna a vida útil da concorrência visível no código.
Antes de avançar, vale revisar asyncio.Barrier no Python, asyncio.eager_task_factory no Python, asyncio.Queue.shutdown no Python e sys.monitoring no Python. Esses conteúdos complementam os conceitos de sincronização, execução imediata, encerramento e observabilidade.
O que eager_start altera
Normalmente, criar uma tarefa agenda sua corrotina para execução pelo event loop. A corrotina começa quando o loop recebe a oportunidade de executá-la. Com início ansioso, a execução pode começar imediatamente durante a criação da tarefa, avançando até o primeiro ponto de suspensão. Se a corrotina terminar sem bloquear, o trabalho pode ser concluído sem uma rodada adicional do loop.
Isso pode reduzir overhead em corrotinas pequenas, como consultas a cache, validações em memória e adaptadores que frequentemente retornam antes de realizar I/O. Porém, também modifica a ordem observável do programa. Um trecho que antes criava várias tarefas e só depois via seus efeitos pode passar a executar parte de cada corrotina no próprio ponto de criação.
Exemplo básico
import asyncio
async def consultar_cache(chave):
print(f"início: {chave}")
if chave == "usuario:1":
return {"nome": "Ana"}
await asyncio.sleep(0.1)
return None
async def main():
async with asyncio.TaskGroup() as grupo:
tarefa = grupo.create_task(
consultar_cache("usuario:1"),
eager_start=True,
)
print(tarefa.result())
asyncio.run(main())A ideia é simples: quando a implementação e a versão do Python aceitam o parâmetro, a corrotina pode avançar imediatamente. Como o caminho de cache não contém um await, ele pode terminar rapidamente. Em um caminho com I/O, a execução avança até encontrar uma suspensão e depois continua pelo event loop.
Compatibilidade entre versões
Recursos recentes de asyncio exigem atenção à versão mínima suportada. Antes de usar eager_start, verifique a documentação da versão instalada e execute testes na mesma versão usada em produção. Uma biblioteca distribuída para vários ambientes pode precisar detectar a assinatura disponível ou manter um caminho compatível sem o argumento.
Evite capturar qualquer TypeError indiscriminadamente, pois ele também pode vir de dentro da corrotina ou de outro erro de programação. Prefira verificar a versão suportada, inspecionar a assinatura de forma controlada ou declarar claramente a versão mínima no projeto.
Quando o início ansioso ajuda
O melhor cenário costuma envolver corrotinas curtas que frequentemente terminam de forma síncrona. Consultas a cache local, leitura de estado já carregado, normalização simples e deduplicação em memória são exemplos. Nesses casos, evitar uma passagem extra pelo scheduler pode reduzir uma pequena parcela do custo.
O ganho deve ser medido. Use benchmarks representativos e compare latência, throughput e consumo de CPU. O artigo sobre perf_counter_ns no Python mostra como medir operações pequenas com repetições e cuidados estatísticos. Uma otimização de microssegundos pode não compensar maior complexidade sem dados reais.
Mudanças na ordem de execução
Considere um laço que cria tarefas e registra mensagens antes e depois de cada criação. Com agendamento convencional, é comum que todas as mensagens do criador apareçam antes das mensagens internas das corrotinas. Com início ansioso, parte da corrotina pode rodar entre essas mensagens.
async def trabalho(numero):
print("corrotina", numero)
await asyncio.sleep(0)
async def executar():
async with asyncio.TaskGroup() as grupo:
for numero in range(3):
print("antes", numero)
grupo.create_task(trabalho(numero), eager_start=True)
print("depois", numero)Não escreva lógica que dependa de uma ordem acidental de prints ou efeitos colaterais. Se a ordem é requisito, represente-a com sincronização explícita, filas, eventos ou dependências claras.
Exceções e cancelamento
Uma corrotina iniciada de forma ansiosa pode gerar uma exceção muito cedo. Dentro de TaskGroup, a concorrência estruturada continua sendo responsável por coordenar falhas e cancelamentos, mas o momento exato em que o erro se torna observável pode mudar. Teste caminhos de falha antes e depois do primeiro await.
Também preserve o comportamento de cancelamento. Não engula asyncio.CancelledError sem uma razão sólida. Blocos finally devem liberar recursos, e operações críticas de limpeza precisam ser curtas e previsíveis. O grupo depende da cooperação das tarefas para encerrar corretamente.
Não bloqueie o event loop
Início ansioso não torna código pesado mais apropriado para o event loop. Se o trecho executado antes do primeiro await realiza cálculo intenso, parsing enorme ou compressão demorada, ele bloqueia a thread do loop imediatamente. Isso pode aumentar a latência de todas as outras conexões.
Para trabalho de CPU, considere processos, interpretadores ou executores apropriados. Veja InterpreterPoolExecutor no Python e kill_workers no ProcessPoolExecutor para estratégias de paralelismo e encerramento de workers.
Estratégia de adoção
Comece em um ponto isolado, com métricas e testes. Documente por que o início ansioso foi escolhido. Compare a execução em cenários de cache hit, cache miss, falha e cancelamento. Observe a ordem de callbacks e logs. Depois, expanda apenas quando o ganho for consistente.
Uma boa função candidata deve ser pequena, não bloquear, ter poucos efeitos colaterais antes do primeiro await e ser fácil de testar. Corrotinas que dependem de uma sequência implícita, atualizam estado global ou executam trabalho pesado são candidatas ruins.
Testes recomendados
Crie testes que cubram conclusão síncrona, suspensão no primeiro I/O, exceção antes do primeiro await, exceção depois da suspensão e cancelamento do grupo. Não teste apenas a ordem textual de logs; teste invariantes do domínio, como número de resultados, recursos fechados e estado final consistente.
import asyncio
async def valor_imediato():
return 42
async def teste():
async with asyncio.TaskGroup() as grupo:
tarefa = grupo.create_task(valor_imediato(), eager_start=True)
assert tarefa.result() == 42Observabilidade
Como o momento de execução pode mudar, logs devem conter identificadores de tarefa, operação e requisição. Métricas de latência ajudam a distinguir ganho real de simples alteração na ordem dos eventos. Ferramentas de profiling e monitoramento devem ser testadas para assegurar que tarefas concluídas muito rapidamente continuam visíveis.
Boas práticas
Use eager_start de maneira explícita, não como padrão indiscriminado. Confirme a compatibilidade da versão, mantenha o trecho antes do primeiro await leve, preserve cancelamentos, evite depender de ordem implícita e meça resultados. Concorrência mais rápida só é útil quando continua compreensível e correta.
Conclusão
TaskGroup já oferece uma base sólida para concorrência estruturada. O controle de início ansioso acrescenta uma ferramenta de otimização e semântica, especialmente para corrotinas que frequentemente terminam sem I/O. Ao mesmo tempo, ele torna ainda mais importante compreender ordem de execução, falhas e bloqueios.
Consulte a documentação oficial de tarefas asyncio e a PEP 654 sobre grupos de exceções. Aplique o recurso somente após testar no ambiente alvo e confirmar que a alteração melhora o sistema sem sacrificar previsibilidade.
