¿Cómo combinar Pool.map con Array (memoria compartida) en el multiproceso de Python?

Question

Tengo una matriz de datos muy grande (solo lectura) que quiero que varios procesos procesen en paralelo.

Me gusta la función Pool.map y me gustaría usarla para calcular funciones en esos datos en paralelo.

Vi que se puede usar la clase Value o Array para usar datos de memoria compartida entre procesos. Pero cuando trato de utilizar este recibo una RuntimeError: 'objetos SynchronizedString sólo deben ser compartidos entre los procesos a través de la herencia cuando se utiliza la función Pool.map:

Aquí es un ejemplo simplificado de lo que yo estoy tratando de hacer:

from sys import stdin 
from multiprocessing import Pool, Array 

def count_it(arr, key): 
    count = 0 
    for c in arr: 
    if c == key: 
     count += 1 
    return count 

if __name__ == '__main__': 
    testData = "abcabcs bsdfsdf gdfg dffdgdfg sdfsdfsd sdfdsfsdf" 
    # want to share it using shared memory 
    toShare = Array('c', testData) 

    # this works 
    print count_it(toShare, "a") 

    pool = Pool() 

    # RuntimeError here 
    print pool.map(count_it, [(toShare,key) for key in ["a", "b", "s", "d"]]) 

¿Alguien puede decirme qué estoy haciendo mal aquí?

Lo que me gustaría hacer es pasar información sobre una matriz asignada de memoria compartida recientemente creada a los procesos después de que se hayan creado en el grupo de procesos.

Answer 1

Tratando de nuevo como acabo de ver la generosidad;)

Básicamente creo que el mensaje de error significa lo que dijo - multiprocesamiento compartida matrices de memoria no puede pasado como argumentos (por decapado). No tiene sentido serializar los datos; el punto es que los datos son memoria compartida. Entonces tienes que hacer que la matriz compartida sea global. Creo que es mejor ponerlo como el atributo de un módulo, como en mi primera respuesta, pero simplemente dejarlo como una variable global en tu ejemplo también funciona bien. Tomando en cuenta su punto de no querer establecer los datos antes del tenedor, aquí hay un ejemplo modificado. Si quisiera tener más de una posible matriz compartida (y es por eso que quería pasar aShare como argumento) podría hacer una lista global de matrices compartidas, y simplemente pasar el índice a count_it (que se convertiría en for c in toShare[i]:).

from sys import stdin 
from multiprocessing import Pool, Array, Process 

def count_it(key): 
    count = 0 
    for c in toShare: 
    if c == key: 
     count += 1 
    return count 

if __name__ == '__main__': 
    # allocate shared array - want lock=False in this case since we 
    # aren't writing to it and want to allow multiple processes to access 
    # at the same time - I think with lock=True there would be little or 
    # no speedup 
    maxLength = 50 
    toShare = Array('c', maxLength, lock=False) 

    # fork 
    pool = Pool() 

    # can set data after fork 
    testData = "abcabcs bsdfsdf gdfg dffdgdfg sdfsdfsd sdfdsfsdf" 
    if len(testData) > maxLength: 
     raise ValueError, "Shared array too small to hold data" 
    toShare[:len(testData)] = testData 

    print pool.map(count_it, ["a", "b", "s", "d"]) 

[EDIT: Lo anterior no funciona en las ventanas, porque de no usar tenedor. Sin embargo, el siguiente no funciona en Windows, sigue utilizando la piscina, así que creo que esto es lo más cercano a lo que quiere:

from sys import stdin 
from multiprocessing import Pool, Array, Process 
import mymodule 

def count_it(key): 
    count = 0 
    for c in mymodule.toShare: 
    if c == key: 
     count += 1 
    return count 

def initProcess(share): 
    mymodule.toShare = share 

if __name__ == '__main__': 
    # allocate shared array - want lock=False in this case since we 
    # aren't writing to it and want to allow multiple processes to access 
    # at the same time - I think with lock=True there would be little or 
    # no speedup 
    maxLength = 50 
    toShare = Array('c', maxLength, lock=False) 

    # fork 
    pool = Pool(initializer=initProcess,initargs=(toShare,)) 

    # can set data after fork 
    testData = "abcabcs bsdfsdf gdfg dffdgdfg sdfsdfsd sdfdsfsdf" 
    if len(testData) > maxLength: 
     raise ValueError, "Shared array too small to hold data" 
    toShare[:len(testData)] = testData 

    print pool.map(count_it, ["a", "b", "s", "d"]) 

No sé por qué mapa no se Conserve en vinagre la matriz, pero Proceso y piscina habrá - Creo quizás se haya transferido en el punto de la inicialización del subproceso en Windows. Tenga en cuenta que los datos aún están configurados después de la horquilla.

Answer 2

Si los datos es de sólo lectura sólo lo hacen una variable en un módulo antes el tenedor de la piscina. Entonces, todos los procesos secundarios deberían poder acceder a él, y no se copiará siempre que no se escriba en él.

import myglobals # anything (empty .py file) 
myglobals.data = [] 

def count_it(key): 
    count = 0 
    for c in myglobals.data: 
     if c == key: 
      count += 1 
    return count 

if __name__ == '__main__': 
myglobals.data = "abcabcs bsdfsdf gdfg dffdgdfg sdfsdfsd sdfdsfsdf" 

pool = Pool() 
print pool.map(count_it, ["a", "b", "s", "d"]) 

Si quiere tratar de usar una matriz aunque se puede tratar con el argumento de palabra clave lock=False (bien es cierto por defecto).

Answer 3

El problema que veo es que Pool no admite el decapado de datos compartidos a través de su lista de argumentos. Eso es lo que significa el mensaje de error por "los objetos solo deben compartirse entre procesos a través de la herencia". Los datos compartidos deben heredarse, es decir, ser globales si desea compartirlos utilizando la clase Pool.

Si necesita pasarlos explícitamente, puede que tenga que usar multiprocesamiento.Proceso. Aquí está su ejemplo reelaborado:

from multiprocessing import Process, Array, Queue 

def count_it(q, arr, key): 
    count = 0 
    for c in arr: 
    if c == key: 
     count += 1 
    q.put((key, count)) 

if __name__ == '__main__': 
    testData = "abcabcs bsdfsdf gdfg dffdgdfg sdfsdfsd sdfdsfsdf" 
    # want to share it using shared memory 
    toShare = Array('c', testData) 

    q = Queue() 
    keys = ['a', 'b', 's', 'd'] 
    workers = [Process(target=count_it, args = (q, toShare, key)) 
    for key in keys] 

    for p in workers: 
    p.start() 
    for p in workers: 
    p.join() 
    while not q.empty(): 
    print q.get(), 

Output: ('s', 9) ('a', 2) ('b', 3) ('d', 12)

El orden de los elementos de la cola puede variar.

Para que esto sea más genérico y similar a la piscina, se podría crear un número fijo N de Procesos, dividir la lista de claves en N trozos, y luego usar una función de contenedor como el objetivo del proceso, la cual llamará count_it para cada clave en la lista que se pasa, como:

def wrapper(q, arr, keys): 
    for k in keys: 
    count_it(q, arr, k) 

Answer 4

The multiprocessing.sharedctypes module provides functions for allocating ctypes objects from shared memory which can be inherited by child processes.

Entonces su uso de sharedctypes es incorrecto. ¿Desea heredar esta matriz del proceso principal o prefiere pasarla explícitamente? En el primer caso, debe crear una variable global como sugieren otras respuestas. Pero no necesita usar sharedctypes para pasarlo explícitamente, solo pase el original testData.

Por cierto, su uso de Pool.map() es incorrecto. Tiene la misma interfaz que la función incorporada map() (¿la confundiste con starmap()?). A continuación está el ejemplo de trabajo con, pasando matriz explícitamente:

from multiprocessing import Pool 

def count_it((arr, key)): 
    count = 0 
    for c in arr: 
     if c == key: 
      count += 1 
    return count 

if __name__ == '__main__': 
    testData = "abcabcs bsdfsdf gdfg dffdgdfg sdfsdfsd sdfdsfsdf" 
    pool = Pool() 
    print pool.map(count_it, [(testData, key) for key in ["a", "b", "s", "d"]])