multiprocessing em Python para tarefas CPU-intensivas

1. Visão Geral

Quando seu código Python precisa fazer muitos cálculos, por exemplo processar imagens, calcular checksums ou fazer simulações numéricas, você verá que usar thread não o torna praticamente mais rápido. A razão é o GIL (Global Interpreter Lock). Este artigo explica esse problema de forma concisa. Depois, o artigo mostra como usar o módulo multiprocessing e ProcessPoolExecutor para dividir o trabalho entre múltiplos processos. Depois de ler, você saberá como paralelizar uma tarefa CPU-bound, escrever testes para ela e evitar erros comuns.

Todos os exemplos usam Python 3.12+ e apenas a biblioteca padrão, não precisa instalar nada a mais.

2. Por que threading não ajuda

2.1. Uma tarefa CPU-intensiva

Vamos usar uma função que conta números primos escrita em Python puro. Esta é uma tarefa típica que consome CPU:

# cpu_task.py
import math

def count_primes(start: int, end: int) -> int:
    count = 0
    for n in range(max(start, 2), end):
        if all(n % d for d in range(2, math.isqrt(n) + 1)):
            count += 1
    return count

A função percorre cada número em [start, end) e tenta dividi-lo por números de 2 até sua raiz quadrada. A função não lê arquivos, não faz chamadas de rede. Ela apenas realiza cálculos.

2.2. Medindo tempo: sequencial e thread

# bench_threads.py
import time
from concurrent.futures import ThreadPoolExecutor
from cpu_task import count_primes

RANGES = [(i, i + 125_000) for i in range(0, 1_000_000, 125_000)]

def run_sequential() -> int:
    return sum(count_primes(a, b) for a, b in RANGES)

def run_threads() -> int:
    starts, ends = zip(*RANGES)
    with ThreadPoolExecutor(max_workers=8) as executor:
        return sum(executor.map(count_primes, starts, ends))

if __name__ == "__main__":
    for fn in (run_sequential, run_threads):
        t0 = time.perf_counter()
        result = fn()
        print(f"{fn.__name__}: {result} primes, {time.perf_counter() - t0:.2f}s")

RANGES divide o intervalo [0, 1_000_000) em 8 partes. run_threads atribui cada parte a um thread. Resultados em uma máquina com 8 núcleos (seus resultados podem ser diferentes):

run_sequential: 78498 primes, 6.41s
run_threads: 78498 primes, 6.58s

A versão com thread é ainda um pouco mais lenta. No CPython padrão, apenas um thread por vez pode executar bytecode Python porque ele deve manter o GIL. Oito threads apenas se alternam usando um único núcleo e ainda incorrem em overhead de alternância entre threads.

Nota: A partir do Python 3.13, existe uma compilação free-threaded (python3.13t) que remove o GIL. Esta versão ainda está em fase experimental e muitas extensões C ainda não a suportam. Para a maioria dos projetos atuais, usar processos ainda é a forma mais confiável.

3. multiprocessing básico

Cada processo tem seu próprio interpretador e seu próprio GIL. Portanto, N processos podem realmente ser executados em paralelo em N núcleos.

3.1. Criando Process manualmente

# basic_process.py
from multiprocessing import Process, Queue
from cpu_task import count_primes

def worker(start: int, end: int, results: Queue) -> None:
    results.put(count_primes(start, end))

if __name__ == "__main__":
    results = Queue()
    ranges = [(0, 500_000), (500_000, 1_000_000)]
    processes = [Process(target=worker, args=(a, b, results)) for a, b in ranges]

    for p in processes:
        p.start()
    total = sum(results.get() for _ in processes)
    for p in processes:
        p.join()

    print("Total primes:", total)

Explicação de cada parte:

• Process(target=..., args=...) descreve um processo filho. Ele não é executado até que você chame start().
• O processo filho não compartilha memória com o processo pai. Para enviar resultados de volta, você deve usar um canal de comunicação. Aqui é Queue.
• O programa chama results.get() antes de join(). Se você chamar join() primeiro enquanto a fila ainda tem dados grandes não lidos, o programa pode ficar em deadlock.

Total primes: 78498

Essa abordagem é flexível, mas verbosa. Na prática, você raramente gerencia Process manualmente.

3.2. Usando Pool

Pool cria um grupo de workers pronto e distribui o trabalho entre eles automaticamente:

# pool_example.py
from multiprocessing import Pool
from cpu_task import count_primes

RANGES = [(i, i + 125_000) for i in range(0, 1_000_000, 125_000)]

if __name__ == "__main__":
    with Pool(processes=8) as pool:
        counts = pool.starmap(count_primes, RANGES)
    print(counts)
    print("Total:", sum(counts))

starmap descompõe cada tupla em RANGES nos parâmetros start, end. Os resultados são retornados na mesma ordem da entrada:

[10746, 9895, 9622, 9480, 9637, 9554, 9454, 9610]
Total: 78498

Se você não passar processes, Pool usa os.cpu_count() workers. O bloco with limpará os workers quando terminar.

4. ProcessPoolExecutor: uma API mais moderna

concurrent.futures.ProcessPoolExecutor tem a mesma interface que ThreadPoolExecutor. Por isso você pode mudar de thread para processo simplesmente alterando o nome da classe:

# parallel.py
import os
from concurrent.futures import ProcessPoolExecutor
from cpu_task import count_primes

def split_range(start: int, end: int, parts: int) -> list[tuple[int, int]]:
    step = (end - start + parts - 1) // parts
    return [(s, min(s + step, end)) for s in range(start, end, step)]

def parallel_count_primes(start: int, end: int, workers: int | None = None) -> int:
    workers = workers or os.cpu_count() or 1
    chunks = split_range(start, end, workers)
    starts, ends = zip(*chunks)
    with ProcessPoolExecutor(max_workers=workers) as executor:
        return sum(executor.map(count_primes, starts, ends))

if __name__ == "__main__":
    import time
    t0 = time.perf_counter()
    print(parallel_count_primes(0, 1_000_000, workers=8))
    print(f"{time.perf_counter() - t0:.2f}s")

78498
1.27s

O tempo diminui de cerca de 6,4 segundos para 1,3 segundos. A velocidade não aumenta exatamente 8 vezes por duas razões. Primeiro, os segmentos têm volumes de trabalho desiguais: segmentos com números maiores levam mais tempo para verificar. Segundo, iniciar processos e transmitir dados também consome tempo.

Se você quer processar cada resultado assim que fica pronto, use submit combinado com as_completed:

from concurrent.futures import ProcessPoolExecutor, as_completed
from cpu_task import count_primes

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        futures = {executor.submit(count_primes, a, a + 250_000): a
                   for a in range(0, 1_000_000, 250_000)}
        for future in as_completed(futures):
            print(f"Intervalo começando em {futures[future]}: {future.result()} primes")

Se o worker lança uma exceção, future.result() lançará essa mesma exceção no processo pai. Portanto, você pode capturar erros com try/except como de costume.

5. Muitas tarefas pequenas: chunksize e imap_unordered

Cada vez que você envia uma tarefa para um worker, Python precisa fazer pickle do parâmetro, enviar por pipe e fazer unpickle do resultado. Se você tem 100.000 tarefas muito pequenas, esse custo pode ser maior que o próprio cálculo. Você lida com isso agrupando tarefas em lotes com chunksize:

# chunks.py
from multiprocessing import Pool

def is_prime(n: int) -> bool:
    if n < 2:
        return False
    return all(n % d for d in range(2, int(n ** 0.5) + 1))

if __name__ == "__main__":
    numbers = range(1_000_000)
    with Pool() as pool:
        total = sum(pool.imap_unordered(is_prime, numbers, chunksize=10_000))
    print(total)

• chunksize=10_000 faz com que cada envio contenha 10.000 números em vez de 1. O número de comunicações entre processos reduz significativamente.
• imap_unordered retorna resultados de forma lazy e não mantém a ordem. Isso é útil quando você apenas precisa agregar resultados, como acumular uma soma, e quer economizar memória.
• ProcessPoolExecutor.map também aceita o parâmetro chunksize. O padrão é 1, então você deve ajustá-lo ao lidar com muitas tarefas pequenas.

6. Inicializando recursos uma vez por worker

Suponha que cada tarefa precise de uma tabela de busca cara para criar. Se você recria a tabela a cada chamada de função, é muito desperdiçador. O parâmetro initializer permite executar uma função exatamente uma vez quando o worker inicia:

# initializer_example.py
from concurrent.futures import ProcessPoolExecutor

_price_table: dict[str, float] = {}

def load_prices(currency: str) -> None:
    global _price_table
    rate = 25_000 if currency == "VND" else 1
    _price_table = {f"SKU-{i}": i * 1.5 * rate for i in range(100_000)}

def order_total(skus: list[str]) -> float:
    return sum(_price_table[s] for s in skus)

if __name__ == "__main__":
    orders = [["SKU-1", "SKU-2"], ["SKU-10", "SKU-20", "SKU-30"]]
    with ProcessPoolExecutor(max_workers=2, initializer=load_prices,
                             initargs=("VND",)) as executor:
        print(list(executor.map(order_total, orders)))

[112500.0, 2250000.0]

A variável _price_table é uma variável global em cada worker. O processo pai não vê essas mudanças. Esta é uma forma segura de cada worker ter seus próprios recursos sem precisar enviar a tabela junto com cada tarefa.

7. Escrevendo testes para código paralelo

Você pode testar código que usa processos com pytest como faria com código normal. A condição é que a função worker deve ser definida no nível do módulo:

# test_parallel.py
from cpu_task import count_primes
from parallel import parallel_count_primes, split_range

def test_count_primes_known_value():
    assert count_primes(0, 100) == 25

def test_split_range_covers_whole_interval():
    chunks = split_range(0, 10, 3)
    assert chunks == [(0, 4), (4, 8), (8, 10)]

def test_parallel_matches_sequential():
    expected = count_primes(0, 50_000)
    assert parallel_count_primes(0, 50_000, workers=4) == expected

pip install pytest
pytest -q test_parallel.py

...                                                    [100%]
3 passed in 0.62s

O teste mais importante é test_parallel_matches_sequential. Ele garante que dividir o intervalo não perde ou duplica nenhum elemento. Este é um erro muito comum ao escrever split_range por conta própria.

8. Erros comuns e quando não usar

Esquecer if __name__ == "__main__":.

Com o método de início spawn (padrão no Windows e macOS) ou forkserver (padrão no Linux a partir do Python 3.14), o processo filho importará o módulo principal novamente. Se o código que cria o pool está no nível do módulo, cada processo filho criará um novo pool. Então você terá RuntimeError ou processos sendo criados continuamente.

Passando lambda ou funções aninhadas.

Tudo que é enviado para um worker deve ser pickleável:

from concurrent.futures import ProcessPoolExecutor

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        print(list(executor.map(lambda x: x * x, range(5))))

_pickle.PicklingError: Can't pickle <function <lambda> at 0x...>: attribute lookup <lambda> on __main__ failed

Você deve definir a função com def no nível do módulo. Se você precisar fixar alguns parâmetros, use functools.partial.

Passando dados muito grandes.

Uma lista com milhões de elementos passada para cada tarefa será pickleada muitas vezes. O tempo de cópia pode dominar a parte do cálculo. Existem algumas formas de contornar isso: enviar apenas índices ou caminhos de arquivo, usar initializer, ou usar multiprocessing.shared_memory com dados em forma de array.

Dependendo do comportamento de fork.

Código que funciona no Linux antigo graças ao fork (o processo filho herda variáveis globais) pode falhar no macOS, Windows ou Python 3.14. Para ter um comportamento consistente, escolha explicitamente o método de início:

import multiprocessing as mp

if __name__ == "__main__":
    mp.set_start_method("spawn")

Nota: Você não deve usar multiprocessing para tarefas I/O-bound como fazer chamadas HTTP ou consultar um banco de dados. Enquanto aguarda I/O, a thread ainda libera o GIL, portanto ThreadPoolExecutor ou asyncio são muito mais leves. Também não use multiprocessing quando cada tarefa leva apenas alguns microsegundos sem agrupar, ou quando o cálculo já está em uma biblioteca C que libera o GIL (NumPy, funções de hash em dados grandes…). Nesses casos, thread ou vetorização geralmente é suficiente.

Criando muitos workers.

Definir o número de workers maior que o número de núcleos da CPU não torna o código CPU-bound mais rápido. Apenas desperdiça RAM, porque cada processo é um interpretador completo. Começar com os.cpu_count() é razoável.

9. Conclusão

O GIL faz com que threads não acelerem código Python puro CPU-intensivo. Cada processo tem seu próprio GIL, portanto dividir o trabalho entre vários processos é uma forma simples de usar todos os núcleos da CPU. Na maioria dos casos, ProcessPoolExecutor é a escolha mais concisa. Pool é útil quando você precisa de imap_unordered ou starmap. Process manual é apenas para situações especiais.

Ao usar multiprocessing, você precisa lembrar dos seguintes pontos:

• Sempre tenha if __name__ == "__main__":.
• Apenas passe funções e dados que podem ser pickleados.
• Aumente chunksize quando tiver muitas tarefas pequenas.
• Use initializer para recursos custosos.
• Escreva um teste comparando o resultado paralelo com o sequencial para garantir que dividir o intervalo está correto.