Você pode usar o asyncio
, em que corotinas/tarefas (Task
) são semelhantes a threads mas não são realmente threads como definidas pelo SO, e então não têm problemas de mudança de contexto e nem de estado externo ao seu próprio programa.
Suponhamos que você queira calcular o quadrado de uma lista dos números de 0 a 4, mas quer cancelar a execução do cálculo de 2^2 se o resultado de alguma outra função for 9, sem ter dentro da corotina de cálculo nenhuma checagem, como você pediu.
Se faria assim com asyncio
:
import asyncio
import random
# Função que simula algum trabalho de cada tarefa/corotina (semelhante a uma thread)
async def retornar_quadrado(n):
if n == 2:
await asyncio.sleep(10) # A função que será cancelada demora mais.
else:
await asyncio.sleep(random.uniform(0.5, 2)) # As outras demoram de 0.5 a 2 segundos.
return n ** 2
async def main():
# Criar uma lista de tarefas
tarefas = []
for i in range(5):
tarefa = asyncio.ensure_future(retornar_quadrado(i))
tarefas.append(tarefa)
for tarefa_completa in asyncio.as_completed(tarefas):
try:
n_quadrado = await tarefa_completa
print(n_quadrado)
except asyncio.CancelledError:
continue
# Se descobrimos que uma das respostas é 9, cancelamos tarefas[2]
if n_quadrado == 9 and not tarefas[2].cancelled():
print('Cancelando tarefas[2]...')
tarefas[2].cancel()
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
Um exemplo de resultado:
16
9
Cancelando tarefas[2]...
0
1
Como pode ver, assim que encontramos o 9, cancelamos o cálculo de 2^2 e ele nunca foi exibido. Se remover a linha tarefas[2].cancel()
, verá que o resultado 4
será exibido, e também que a função demorará os 10 segundos que esse caso especial requer.