Bem-vindos à aula sobre concurrent.futures! Neste módulo, vamos explorar uma das ferramentas mais poderosas e fáceis de usar do Python para lidar com concorrência e paralelismo. Se você já tentou gerenciar threads ou processos manualmente, sabe que isso pode ser complicado e propenso a erros. O módulo concurrent.futures abstrai esses detalhes, fornecendo uma interface limpa e consistente para executar tarefas em paralelo, seja com threads (para operações de I/O) ou com processos (para tarefas que exigem CPU).

Nesta aula, vamos mergulhar fundo nos dois principais executores: ThreadPoolExecutor e ProcessPoolExecutor. Também vamos entender o que são Futures e como usá-los para obter resultados de tarefas assíncronas. Por fim, veremos padrões úteis, como o uso de as_completed e map, que tornam o código mais eficiente e elegante. Ao final, você estará apto a aplicar essas técnicas em seus próprios projetos para melhorar o desempenho e a responsividade.

ThreadPoolExecutor

O ThreadPoolExecutor é um executor que utiliza um pool de threads para executar chamadas de forma assíncrona. Ele é ideal para tarefas que passam a maior parte do tempo esperando por I/O, como leitura/escrita de arquivos, requisições HTTP, consultas a bancos de dados ou interação com APIs externas. Isso porque, enquanto uma thread está bloqueada esperando uma resposta, outras threads podem continuar executando, aproveitando melhor o tempo de CPU.

Para usar o ThreadPoolExecutor, você cria uma instância, especificando o número máximo de threads (max_workers). Em seguida, você submete tarefas usando o método submit, que retorna um objeto Future. Você também pode usar o método map para aplicar uma função a uma sequência de argumentos de forma concorrente. O contexto with garante que o pool seja encerrado corretamente após o uso.

Vamos ver um exemplo clássico: baixar várias páginas da web. Suponha que temos uma lista de URLs e queremos buscar o conteúdo de cada uma. Sem usar concorrência, faríamos uma requisição de cada vez, o que é lento. Com threads, podemos disparar várias requisições simultaneamente.

import concurrent.futures
import requests

urls = [
    'https://httpbin.org/delay/1',
    'https://httpbin.org/delay/2',
    'https://httpbin.org/delay/3'
]

def fetch_url(url):
    response = requests.get(url)
    return response.status_code

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    # Submetendo tarefas individualmente
    future_to_url = {executor.submit(fetch_url, url): url for url in urls}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            status = future.result()
            print(f'{url}: {status}')
        except Exception as e:
            print(f'{url} gerou uma exceção: {e}')

Neste código, usamos submit para cada URL e depois as_completed para iterar sobre os futures na ordem em que são concluídos. Note que o tempo total será aproximadamente o maior atraso (3 segundos), em vez da soma (6 segundos). Isso demonstra o ganho de desempenho.

Uma alternativa mais concisa é usar executor.map, que retorna um iterador dos resultados na ordem dos argumentos:

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    statuses = executor.map(fetch_url, urls)
    for url, status in zip(urls, statuses):
        print(f'{url}: {status}')

No entanto, map não permite capturar exceções de forma individual; se uma tarefa falhar, o iterador levanta a exceção no momento da iteração. Para maior controle, prefira submit + as_completed.

ProcessPoolExecutor

O ProcessPoolExecutor funciona de forma semelhante, mas usa processos em vez de threads. Isso é fundamental para tarefas que exigem muita CPU, como cálculos numéricos, processamento de imagens ou análise de dados. Por que não usar threads? Porque o Python possui o GIL (Global Interpreter Lock), que impede que múltiplas threads executem código Python nativo simultaneamente. Em tarefas CPU-bound, as threads acabam disputando o GIL e não há ganho real de paralelismo. Já os processos são independentes, cada um com seu próprio interpretador e GIL, permitindo verdadeira execução paralela em múltiplos núcleos.

Para usar ProcessPoolExecutor, a sintaxe é idêntica à do ThreadPoolExecutor. A diferença crucial é que, ao usar processos, a função que você passa deve ser serializável (pickleable). Além disso, o código deve estar protegido pelo bloco if __name__ == '__main__': em sistemas operacionais que usam o método de spawn (como Windows) para evitar a criação recursiva de processos.

Vamos ilustrar com um exemplo que calcula o quadrado de números grandes:

import concurrent.futures
import math

def is_prime(n):
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True

def check_prime(n):
    return n, is_prime(n)

if __name__ == '__main__':
    numbers = [15485863, 15485867, 15485869, 32452843, 32452867]
    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
        results = executor.map(check_prime, numbers)
        for number, prime in zip(numbers, results):
            print(f'{number} é primo? {prime}')

Neste exemplo, a função is_prime é CPU-bound. Usando processos, cada número é testado em paralelo, aproveitando todos os núcleos da máquina. Se usássemos threads, o GIL limitaria o ganho.

É importante notar que a criação de processos tem um custo maior do que threads, tanto de memória quanto de tempo de inicialização. Portanto, use ProcessPoolExecutor apenas para tarefas que realmente se beneficiarão de paralelismo de CPU. Para tarefas I/O-bound, threads são mais leves e suficientes.

Futures

Um Future é um objeto que representa o resultado de uma tarefa que pode ainda não ter sido concluída. Ele encapsula a chamada assíncrona e fornece métodos para verificar se a tarefa terminou, aguardar o resultado, ou cancelar a tarefa. No módulo concurrent.futures, os objetos Future são retornados pelo método submit de um executor.

Os principais métodos de um Future são:

  • result(timeout=None): retorna o resultado da chamada. Se a chamada ainda não terminou, bloqueia até que termine ou até que o tempo limite expire. Se a chamada levantou uma exceção, ela é relançada aqui.
  • done(): retorna True se a chamada foi concluída (com sucesso ou com exceção) ou cancelada.
  • cancel(): tenta cancelar a chamada. Retorna True se o cancelamento foi bem-sucedido.
  • add_done_callback(fn): adiciona uma função de callback que será chamada quando o future estiver concluído. A função recebe o future como argumento.

Vejamos um exemplo que usa add_done_callback para tratar o resultado de forma assíncrona, sem bloquear o fluxo principal:

import concurrent.futures
import time

def tarefa_demorada(nome, segundos):
    time.sleep(segundos)
    return f"{nome} terminou após {segundos} segundos"

def callback(future):
    print(f"Callback executado: {future.result()}")

with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
    fut1 = executor.submit(tarefa_demorada, "Tarefa 1", 2)
    fut2 = executor.submit(tarefa_demorada, "Tarefa 2", 1)
    fut1.add_done_callback(callback)
    fut2.add_done_callback(callback)

print("Tarefas submetidas, aguardando callbacks...")

Ao executar, você verá que o programa principal não bloqueia, e os callbacks são chamados assim que cada tarefa termina. Isso é útil para programação reativa ou para atualizar interfaces de usuário.

Outro ponto importante: quando você usa executor.map, os resultados são retornados como um iterador, mas internamente também são futures. O método map apenas agrupa a lógica de submissão e coleta de resultados. Se você precisar de mais controle, como capturar exceções individuais ou processar resultados na ordem de conclusão, prefira submit + as_completed.

Padrões

Existem alguns padrões comuns ao usar concurrent.futures que podem tornar seu código mais eficiente e expressivo. Vamos explorar os principais.

Padrão 1: Mapa paralelo com executor.map

Quando você tem uma função e uma lista de argumentos, e o resultado de cada chamada é independente, executor.map é a escolha mais simples. Ele retorna um iterador que produz os resultados na mesma ordem dos argumentos.

import concurrent.futures

def dobro(x):
    return x * 2

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    resultados = list(executor.map(dobro, range(10)))
print(resultados)

Esse padrão é equivalente a um map nativo, mas com execução paralela.

Padrão 2: Submissão dinâmica com as_completed

Em muitos cenários, você não sabe de antemão quantas tarefas serão necessárias, ou quer processar os resultados assim que cada um estiver pronto, sem esperar pelas mais lentas. O padrão submit + as_completed é perfeito para isso.

import concurrent.futures
import random
import time

def tarefa(n):
    tempo = random.uniform(0.5, 2.0)
    time.sleep(tempo)
    return n, tempo

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(tarefa, i): i for i in range(10)}
    for future in concurrent.futures.as_completed(futures):
        n, tempo = future.result()
        print(f"Tarefa {n} terminou em {tempo:.2f}s")

Aqui, as tarefas são submetidas todas de uma vez, e as_completed nos dá os futures na ordem em que são concluídos. Isso é muito útil para agregação de resultados parciais.

Padrão 3: Tratamento de exceções

Quando uma tarefa levanta uma exceção, o Future a captura e a relança quando você chama result(). É importante tratar essas exceções para evitar que o programa quebre. Você pode usar try/except em torno de future.result().

def funcao_com_erro(x):
    if x == 3:
        raise ValueError("Número 3 não é permitido")
    return x * 2

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(funcao_com_erro, i) for i in range(5)]
    for future in concurrent.futures.as_completed(futures):
        try:
            resultado = future.result()
            print(f"Resultado: {resultado}")
        except Exception as e:
            print(f"Erro capturado: {e}")

Dessa forma, uma exceção em uma tarefa não impede o processamento das demais.

Padrão 4: Cancelamento e timeout

Às vezes você precisa cancelar tarefas ou impor um tempo máximo de espera. O Future.cancel() só funciona se a tarefa ainda não começou a executar. Para esperar com timeout, use future.result(timeout=...).

import concurrent.futures
import time

def tarefa_longa():
    time.sleep(10)
    return "Concluída"

with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
    future = executor.submit(tarefa_longa)
    try:
        resultado = future.result(timeout=2)
        print(resultado)
    except concurrent.futures.TimeoutError:
        print("Tarefa não terminou a tempo")
        future.cancel()

Isso é útil para evitar que o programa fique preso indefinidamente.

Padrão 5: Pipeline com múltiplos estágios

Você pode combinar múltiplos executores para criar um pipeline de processamento. Por exemplo, primeiro buscar dados da web (I/O-bound) e depois processá-los (CPU-bound).

import concurrent.futures

def buscar_dados(url):
    # Simula I/O
    return f"dados de {url}"

def processar_dados(dados):
    # Simula CPU-bound
    return dados.upper()

urls = ["site1", "site2", "site3"]
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as io_executor:
    dados = list(io_executor.map(buscar_dados, urls))

with concurrent.futures.ProcessPoolExecutor(max_workers=2) as cpu_executor:
    resultados = list(cpu_executor.map(processar_dados, dados))

print(resultados)

Esse padrão permite otimizar cada estágio com o executor adequado.

Referências

Exercícios

  1. Escreva um programa que use ThreadPoolExecutor para baixar o conteúdo de 5 URLs fictícias (simule com time.sleep) e imprima o tamanho (em caracteres) de cada conteúdo. Use executor.map.

    ✓ Resposta: 
    import concurrent.futures
import time

def baixar(url):
    time.sleep(1)
    return f"Conteúdo de {url}"

urls = ["https://exemplo.com/1", "https://exemplo.com/2", "https://exemplo.com/3", "https://exemplo.com/4", "https://exemplo.com/5"]

with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    tamanhos = [len(conteudo) for conteudo in executor.map(baixar, urls)]

for url, tamanho in zip(urls, tamanhos):
    print(f"{url}: {tamanho} caracteres")

  2. Crie uma função que calcule o fatorial de um número. Use ProcessPoolExecutor para calcular o fatorial de vários números em paralelo (ex.: 10, 20, 30, 40). Imprima os resultados.

    ✓ Resposta: 
    import concurrent.futures
import math

def fatorial(n):
    return math.factorial(n)

if __name__ == '__main__':
    numeros = [10, 20, 30, 40]
    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
        resultados = executor.map(fatorial, numeros)
        for n, resultado in zip(numeros, resultados):
            print(f"Fatorial de {n} = {resultado}")

  3. Explique a diferença entre executor.map e executor.submit + as_completed. Dê um exemplo em que a segunda abordagem é mais vantajosa.

    ✓ Resposta:

    executor.map retorna um iterador que produz os resultados na ordem dos argumentos, mas só começa a produzir quando o primeiro resultado está pronto? Na verdade, ele aguarda até que todos os resultados estejam prontos para manter a ordem? Não, ele aguarda cada resultado na ordem, mas as tarefas são executadas em paralelo. O problema é que se uma tarefa demora muito, o iterador fica bloqueado esperando por ela, mesmo que outras já tenham terminado. Já submit + as_completed permite processar os resultados na ordem em que são concluídos, o que é melhor quando as tarefas têm durações variáveis e você quer agir assim que cada uma termina.

    Exemplo: processar respostas de múltiplas requisições HTTP onde algumas são mais lentas. Com as_completed, você pode atualizar a interface do usuário conforme cada resposta chega, em vez de esperar todas.

  4. Implemente um programa que use ThreadPoolExecutor para verificar se uma lista de números é par ou ímpar. Use add_done_callback para imprimir o resultado de cada número assim que a verificação for concluída.

    ✓ Resposta: 
    import concurrent.futures

def par_ou_impar(n):
    return n, "par" if n % 2 == 0 else "ímpar"

def callback(future):
    n, resultado = future.result()
    print(f"{n} é {resultado}")

numeros = [1, 2, 3, 4, 5, 6]
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(par_ou_impar, n) for n in numeros]
    for future in futures:
        future.add_done_callback(callback)

  5. Escreva um código que use ProcessPoolExecutor para calcular a soma dos quadrados de uma lista de 1 a 100. Compare o tempo de execução com a versão sequencial (use time.time()).

    ✓ Resposta: 
    import concurrent.futures
import time

def soma_quadrados(intervalo):
    return sum(x*x for x in intervalo)

if __name__ == '__main__':
    numeros = list(range(1, 101))
    # divide em 4 partes
    partes = [numeros[i::4] for i in range(4)]

    inicio = time.time()
    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
        resultados = executor.map(soma_quadrados, partes)
        total = sum(resultados)
    fim = time.time()
    print(f"Total (paralelo): {total}, tempo: {fim - inicio:.4f}s")

    inicio = time.time()
    total_seq = soma_quadrados(numeros)
    fim = time.time()
    print(f"Total (sequencial): {total_seq}, tempo: {fim - inicio:.4f}s")

Boas Práticas e Observações Finais

Ao trabalhar com concurrent.futures, lembre-se de escolher o executor adequado ao tipo de tarefa: ThreadPoolExecutor para I/O-bound e ProcessPoolExecutor para CPU-bound. Sempre use o gerenciador de contexto with para garantir que os recursos sejam liberados. Para tarefas que dependem de estado compartilhado, prefira processos com objetos Queue ou Pipe se necessário, mas em muitos casos a abordagem de map/reduce com futures é suficiente. Além disso, monitore o número de workers: muitos podem sobrecarregar o sistema, poucos podem não aproveitar bem os recursos. Finalmente, teste seu código em diferentes cenários para garantir que não há condições de corrida ou deadlocks.