Como fazer programação paralela em Python?

141

Para C ++, podemos usar o OpenMP para fazer programação paralela; no entanto, o OpenMP não funcionará para Python. O que devo fazer se quiser fazer paralelo com algumas partes do meu programa python?

A estrutura do código pode ser considerada como:

solve1(A)
solve2(B)

Onde solve1e solve2são duas funções independentes. Como executar esse tipo de código em paralelo, em vez de em sequência, a fim de reduzir o tempo de execução? Espero que alguém possa me ajudar. Muito obrigado antecipadamente. O código é:

def solve(Q, G, n):
    i = 0
    tol = 10 ** -4

    while i < 1000:
        inneropt, partition, x = setinner(Q, G, n)
        outeropt = setouter(Q, G, n)

        if (outeropt - inneropt) / (1 + abs(outeropt) + abs(inneropt)) < tol:
            break

        node1 = partition[0]
        node2 = partition[1]

        G = updateGraph(G, node1, node2)

        if i == 999:
            print "Maximum iteration reaches"
    print inneropt

Onde setinner e setouter são duas funções independentes. É aí que eu quero fazer um paralelo ...

ilovecp3
fonte
31
Dê uma olhada no multiprocessamento . Nota: Os encadeamentos do Python não são adequados para tarefas ligadas à CPU, somente para E / S.
9000
4
@ 9000 +100 internets para mencionar as tarefas dependentes da CPU versus E / S.
precisa
@ 9000 Na verdade, os threads não são adequados para tarefas ligadas à CPU, tanto quanto eu sei! Processos é o caminho a percorrer ao executar tarefas reais vinculadas à CPU.
Omar Al-Ithawi
6
@OmarIthawi: por que, os threads funcionam bem se você tem muitos núcleos de CPU (como de costume agora). Em seguida, seu processo pode executar vários threads carregando todos esses núcleos em paralelo e compartilhando dados comuns entre eles implicitamente (ou seja, sem ter uma área de memória compartilhada explícita ou sistema de mensagens entre processos).
9000
1
@ user2134774: Bem, sim, meu segundo comentário faz pouco sentido. Provavelmente, as únicas extensões C que liberam o GIL podem se beneficiar disso; por exemplo, partes do NumPy e Pandas fazem isso. Em outros casos, está errado (mas não posso editá-lo agora).
9000

Respostas:

162

Você pode usar o módulo de multiprocessamento . Nesse caso, eu posso usar um pool de processamento:

from multiprocessing import Pool
pool = Pool()
result1 = pool.apply_async(solve1, [A])    # evaluate "solve1(A)" asynchronously
result2 = pool.apply_async(solve2, [B])    # evaluate "solve2(B)" asynchronously
answer1 = result1.get(timeout=10)
answer2 = result2.get(timeout=10)

Isso gerará processos que podem fazer um trabalho genérico para você. Como não passamos processes, ele gerará um processo para cada núcleo de CPU em sua máquina. Cada núcleo da CPU pode executar um processo simultaneamente.

Se você deseja mapear uma lista para uma única função, faça o seguinte:

args = [A, B]
results = pool.map(solve1, args)

Não use threads porque o GIL bloqueia qualquer operação em objetos python.

Matt Williamson
fonte
1
faz pool.maptambém aceita dicionários como args? Ou apenas listas simples?
O BNDR
Apenas listas, eu acho. Mas você pode simplesmente passar em dict.items (), que será uma lista de tuplas de valores-chave
Matt Williamson
Infelizmente, isso termina em um erro `tipo unhashable: 'list' '
The Bndr
além do meu último comentário: `dict.items ()` work. O erro aumenta, porque eu tive que mudar o manuseio da variável insight o processo-funktion. Infelizmente, a mensagem de erro não foi muito útil ... Então: obrigado por sua dica. :-)
The Bndr
2
O que é o tempo limite aqui?
gamma
26

Isso pode ser feito de maneira muito elegante com Ray .

Para paralelizar seu exemplo, você precisa definir suas funções com o @ray.remotedecorador e depois invocá-las .remote.

import ray

ray.init()

# Define the functions.

@ray.remote
def solve1(a):
    return 1

@ray.remote
def solve2(b):
    return 2

# Start two tasks in the background.
x_id = solve1.remote(0)
y_id = solve2.remote(1)

# Block until the tasks are done and get the results.
x, y = ray.get([x_id, y_id])

Há várias vantagens disso sobre o módulo de multiprocessamento .

  1. O mesmo código será executado em uma máquina multicore e em um cluster de máquinas.
  2. Os processos compartilham dados de forma eficiente por meio de memória compartilhada e serialização de cópia zero .
  3. As mensagens de erro são propagadas de maneira adequada.
  4. Essas chamadas de função podem ser compostas juntas, por exemplo,

    @ray.remote
    def f(x):
        return x + 1
    
    x_id = f.remote(1)
    y_id = f.remote(x_id)
    z_id = f.remote(y_id)
    ray.get(z_id)  # returns 4
  5. Além de chamar funções remotamente, as classes podem ser instanciadas remotamente como atores .

Observe que Ray é uma estrutura que eu tenho ajudado a desenvolver.

Robert Nishihara
fonte
eu continuo recebendo um erro que diz "Não foi possível encontrar uma versão que satisfaz a exigência de raios (de versões:) Nenhuma distribuição de correspondência encontrada para ray" ao tentar instalar o pacote em Python
alwaysaskingquestions
2
Normalmente, esse tipo de erro significa que você precisa atualizar pip. Eu sugiro tentar pip install --upgrade pip. Se você precisar usá sudo-lo, é possível que a versão pipque você está usando para instalar raynão seja a mesma que está sendo atualizada. Você pode verificar com pip --version. Além disso, o Windows atualmente não é suportado, portanto, se você estiver no Windows, esse provavelmente é o problema.
Robert Nishihara
1
Apenas uma observação: isso se destina principalmente à distribuição de trabalhos simultâneos em várias máquinas.
21818 Matt Williamson
2
Na verdade, ele é otimizado para o gabinete de máquina única e a configuração de cluster. Muitas decisões de projeto (por exemplo, memória compartilhada, serialização de cópia zero) são direcionadas para oferecer suporte a máquinas únicas.
Robert Nishihara
2
Seria ótimo se os documentos apontassem isso mais. Ao ler os documentos, tive a impressão de que ele não era realmente destinado ao gabinete de uma única máquina.
Sledge
4

A solução, como outros já disseram, é usar vários processos. Qual estrutura é mais apropriada, no entanto, depende de muitos fatores. Além dos já mencionados, há o charm4py e o mpi4py (eu sou o desenvolvedor do charm4py).

Existe uma maneira mais eficiente de implementar o exemplo acima do que usar a abstração do conjunto de trabalhadores. O loop principal envia os mesmos parâmetros (incluindo o gráfico completo G) repetidamente para os trabalhadores em cada uma das 1000 iterações. Como pelo menos um trabalhador residirá em um processo diferente, isso envolve copiar e enviar os argumentos para os outros processos. Isso pode ser muito caro, dependendo do tamanho dos objetos. Em vez disso, faz sentido que os funcionários armazenem o estado e simplesmente enviem as informações atualizadas.

Por exemplo, no charm4py, isso pode ser feito assim:

class Worker(Chare):

    def __init__(self, Q, G, n):
        self.G = G
        ...

    def setinner(self, node1, node2):
        self.updateGraph(node1, node2)
        ...


def solve(Q, G, n):
    # create 2 workers, each on a different process, passing the initial state
    worker_a = Chare(Worker, onPE=0, args=[Q, G, n])
    worker_b = Chare(Worker, onPE=1, args=[Q, G, n])
    while i < 1000:
        result_a = worker_a.setinner(node1, node2, ret=True)  # execute setinner on worker A
        result_b = worker_b.setouter(node1, node2, ret=True)  # execute setouter on worker B

        inneropt, partition, x = result_a.get()  # wait for result from worker A
        outeropt = result_b.get()  # wait for result from worker B
        ...

Observe que, para este exemplo, realmente precisamos apenas de um trabalhador. O loop principal pode executar uma das funções e fazer com que o trabalhador execute a outra. Mas meu código ajuda a ilustrar algumas coisas:

  1. O trabalhador A é executado no processo 0 (igual ao loop principal). Enquanto result_a.get()está bloqueado aguardando o resultado, o trabalhador A faz o cálculo no mesmo processo.
  2. Os argumentos são passados ​​automaticamente por referência ao trabalhador A, pois está no mesmo processo (não há cópia envolvida).
Juan Galvez
fonte
2

Em alguns casos, é possível paralelizar loops automaticamente usando o Numba , embora ele funcione apenas com um pequeno subconjunto do Python:

from numba import njit, prange

@njit(parallel=True)
def prange_test(A):
    s = 0
    # Without "parallel=True" in the jit-decorator
    # the prange statement is equivalent to range
    for i in prange(A.shape[0]):
        s += A[i]
    return s

Infelizmente, parece que o Numba funciona apenas com matrizes Numpy, mas não com outros objetos Python. Em teoria, também pode ser possível compilar Python para C ++ e paralelizar automaticamente usando o compilador Intel C ++ , embora eu ainda não tenha tentado isso.

Anderson Green
fonte
2

Você pode usar a joblibbiblioteca para fazer computação e multiprocessamento paralelos.

from joblib import Parallel, delayed

Você pode simplesmente criar uma função fooque deseja executar paralelamente e com base no seguinte trecho de código implementar o processamento paralelo:

output = Parallel(n_jobs=num_cores)(delayed(foo)(i) for i in input)

Onde num_corespode ser obtido da multiprocessingbiblioteca, como segue:

import multiprocessing

num_cores = multiprocessing.cpu_count()

Se você possui uma função com mais de um argumento de entrada e deseja apenas repetir um dos argumentos por uma lista, pode usar o comando partial função da functoolsbiblioteca da seguinte maneira:

from joblib import Parallel, delayed
import multiprocessing
from functools import partial
def foo(arg1, arg2, arg3, arg4):
    '''
    body of the function
    '''
    return output
input = [11,32,44,55,23,0,100,...] # arbitrary list
num_cores = multiprocessing.cpu_count()
foo_ = partial(foo, arg2=arg2, arg3=arg3, arg4=arg4)
# arg1 is being fetched from input list
output = Parallel(n_jobs=num_cores)(delayed(foo_)(i) for i in input)

Você pode encontrar uma explicação completa do multiprocessamento python e R com alguns exemplos aqui .

vahab najari
fonte