Compartilhe grande matriz Numpy somente leitura entre processos de multiprocessamento

Eu tenho um 60GB SciPy Array (Matrix) que devo compartilhar entre 5+ multiprocessing Processobjetos. Eu vi numpy-sharedmem e li essa discussão na lista SciPy. Parece haver duas abordagens - numpy-sharedmeme usando multiprocessing.RawArray()ae mapeando NumPy dtypes para ctypes. Agora, numpy-sharedmemparece ser o caminho a percorrer, mas ainda não vi um bom exemplo de referência. Não preciso de nenhum tipo de bloqueio, já que a matriz (na verdade, uma matriz) será somente leitura. Agora, devido ao seu tamanho, gostaria de evitar uma cópia. Ele soa como o método correto é criar a única cópia da matriz como uma sharedmemmatriz e, em seguida, passá-lo para os Processobjetos? Algumas perguntas específicas:

1. Qual é a melhor maneira de realmente passar os identificadores de memória compartilhada para subes Process()? Eu preciso de uma fila apenas para passar um array? Um cachimbo seria melhor? Posso apenas passá-lo como um argumento para o Process()init da subclasse (onde estou assumindo que está em conserva)?

2. Na discussão que vinculei acima, há menção de numpy-sharedmemnão ser seguro para 64 bits. Definitivamente, estou usando algumas estruturas que não são endereçáveis ​​de 32 bits.

3. Existem vantagens e desvantagens na RawArray()abordagem? Mais lento, mais problemático?

4. Eu preciso de algum mapeamento ctype-para-dtype para o método numpy-sharedmem?

5. Alguém tem um exemplo de código OpenSource fazendo isso? Sou um aprendiz muito prático e é difícil fazer isso funcionar sem nenhum tipo de bom exemplo para olhar.

Se houver alguma informação adicional que eu possa fornecer para ajudar a esclarecer isso para outras pessoas, por favor, comente e eu adicionarei. Obrigado!

Isso precisa ser executado no Ubuntu Linux e Maybe Mac OS, mas a portabilidade não é uma grande preocupação.

@Velimir Mlaker deu uma ótima resposta. Achei que poderia adicionar alguns comentários e um pequeno exemplo.

(Não consegui encontrar muita documentação sobre sharedmem - estes são os resultados de minhas próprias experiências.)

1. Você precisa passar as alças quando o subprocesso está iniciando ou depois de ter iniciado? Se for apenas o primeiro, você pode usar os argumentos targete argspara Process. Isso é potencialmente melhor do que usar uma variável global.
2. Na página de discussão que você vinculou, parece que o suporte para Linux de 64 bits foi adicionado ao sharedmem há algum tempo, então pode não ser um problema.
3. Eu não sei sobre este.
4. Não. Consulte o exemplo abaixo.

Exemplo

#!/usr/bin/env python
from multiprocessing import Process
import sharedmem
import numpy

def do_work(data, start):
    data[start] = 0;

def split_work(num):
    n = 20
    width  = n/num
    shared = sharedmem.empty(n)
    shared[:] = numpy.random.rand(1, n)[0]
    print "values are %s" % shared

    processes = [Process(target=do_work, args=(shared, i*width)) for i in xrange(num)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

    print "values are %s" % shared
    print "type is %s" % type(shared[0])

if __name__ == '__main__':
    split_work(4)

Resultado

values are [ 0.81397784  0.59667692  0.10761908  0.6736734   0.46349645  0.98340718
  0.44056863  0.10701816  0.67167752  0.29158274  0.22242552  0.14273156
  0.34912309  0.43812636  0.58484507  0.81697513  0.57758441  0.4284959
  0.7292129   0.06063283]
values are [ 0.          0.59667692  0.10761908  0.6736734   0.46349645  0.
  0.44056863  0.10701816  0.67167752  0.29158274  0.          0.14273156
  0.34912309  0.43812636  0.58484507  0.          0.57758441  0.4284959
  0.7292129   0.06063283]
type is <type 'numpy.float64'>

Esta questão relacionada pode ser útil.

Se você estiver no Linux (ou qualquer sistema compatível com POSIX), você pode definir este array como uma variável global. multiprocessingestá usando fork()no Linux quando inicia um novo processo filho. Um processo filho recém-gerado compartilha automaticamente a memória com seu pai, desde que ele não a altere ( mecanismo de cópia na gravação ).

Já que você está dizendo "Eu não preciso de nenhum tipo de bloqueio, uma vez que o array (na verdade, uma matriz) será somente leitura", tirar vantagem desse comportamento seria uma abordagem muito simples, mas extremamente eficiente: todos os processos filhos acessarão os mesmos dados na memória física ao ler este grande array numpy.

Não entregar sua matriz para o Process()construtor, este irá instruir multiprocessingpara pickleos dados para a criança, o que seria extremamente ineficiente ou impossível no seu caso. No Linux, logo após fork()o filho é uma cópia exata do pai usando a mesma memória física, então tudo que você precisa fazer é certificar-se de que a variável Python 'contendo' a matriz está acessível a partir da targetfunção que você entrega Process(). Isso você normalmente pode conseguir com uma variável 'global'.

Código de exemplo:

from multiprocessing import Process
from numpy import random


global_array = random.random(10**4)


def child():
    print sum(global_array)


def main():
    processes = [Process(target=child) for _ in xrange(10)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()


if __name__ == "__main__":
    main()

No Windows - que não oferece suporte fork()- multiprocessingestá usando a chamada de API win32 CreateProcess. Ele cria um processo inteiramente novo a partir de qualquer executável. É por isso que no Windows é necessário separar os dados para o filho se precisar de dados que foram criados durante o tempo de execução do pai.

Você pode estar interessado em um pequeno pedaço de código que escrevi: github.com/vmlaker/benchmark-sharedmem

O único arquivo de interesse é main.py. É um benchmark de numpy-sharedmem - o código simplesmente passa arrays (ou numpyou sharedmem) para processos gerados, via Pipe. Os trabalhadores apenas solicitam sum()os dados. Eu estava interessado apenas em comparar os tempos de comunicação de dados entre as duas implementações.

Também escrevi outro código mais complexo: github.com/vmlaker/sherlock .

Aqui eu uso o módulo numpy-sharedmem para processamento de imagem em tempo real com OpenCV - as imagens são matrizes NumPy, de acordo com a cv2API mais recente do OpenCV . As imagens, na verdade suas referências, são compartilhadas entre os processos por meio do objeto de dicionário criado a partir multiprocessing.Manager(em oposição ao uso de Queue ou Pipe). Estou obtendo grandes melhorias de desempenho quando comparado ao uso de matrizes NumPy simples.

Tubulação vs. fila :

Na minha experiência, o IPC com Pipe é mais rápido do que o Queue. E isso faz sentido, já que o Queue adiciona um bloqueio para torná-lo seguro para vários produtores / consumidores. Pipe não. Mas se você tiver apenas dois processos conversando, é seguro usar o Pipe ou, como diz a documentação:

... não há risco de corrupção de processos que usam extremidades diferentes do tubo ao mesmo tempo.

sharedmemsegurança :

O principal problema com o sharedmemmódulo é a possibilidade de vazamento de memória ao sair do programa. Isso é descrito em uma longa discussão aqui . Embora em 10 de abril de 2011 Sturla mencione uma correção de vazamento de memória, ainda experimentei vazamentos desde então, usando os dois repositórios, o próprio Sturla Molden no GitHub ( github.com/sturlamolden/sharedmem-numpy ) e Chris Lee-Messer no Bitbucket ( bitbucket.org/cleemesser/numpy-sharedmem ).

Se sua matriz for tão grande, você pode usar numpy.memmap. Por exemplo, se você tem um array armazenado em disco, digamos 'test.array', você pode usar processos simultâneos para acessar os dados nele mesmo no modo de "gravação", mas seu caso é mais simples, pois você só precisa do modo de "leitura".

Criando a matriz:

a = np.memmap('test.array', dtype='float32', mode='w+', shape=(100000,1000))

Você pode então preencher essa matriz da mesma forma que faria com uma matriz comum. Por exemplo:

a[:10,:100]=1.
a[10:,100:]=2.

Os dados são armazenados no disco quando você exclui a variável a.

Mais tarde, você pode usar vários processos que irão acessar os dados em test.array:

# read-only mode
b = np.memmap('test.array', dtype='float32', mode='r', shape=(100000,1000))

# read and writing mode
c = np.memmap('test.array', dtype='float32', mode='r+', shape=(100000,1000))

Respostas relacionadas:

• Trabalhando com big data em python e numpy, não há memória RAM suficiente, como salvar resultados parciais em disco?

• É possível mapear dados descontínuos no disco para uma matriz com python?

Você também pode achar útil dar uma olhada na documentação do pyro como se você pudesse particionar sua tarefa apropriadamente, você poderia usá-lo para executar seções diferentes em máquinas diferentes, bem como em núcleos diferentes na mesma máquina.

Por que não usar multithreading? Os recursos do processo principal podem ser compartilhados por seus threads nativamente, portanto, multithreading é obviamente a melhor maneira de compartilhar objetos pertencentes ao processo principal.

Se você se preocupa com o mecanismo GIL do python, talvez possa recorrer ao nogilde numba.