paralelo - procesos concurrentes python

Multiprocesamiento de IOError: longitud de mensaje erróneo

Question

Obtengo un IOError: bad message length al pasar argumentos grandes a la función de map . ¿Cómo puedo evitar esto? El error se produce cuando configuro N=1500 o mayor.

El código es:

import numpy as np
import multiprocessing

def func(args):
    i=args[0]
    images=args[1]
    print i
    return 0

N=1500       #N=1000 works fine

images=[]
for i in np.arange(N):
    images.append(np.random.random_integers(1,100,size=(500,500)))

iter_args=[]
for i in range(0,1):
    iter_args.append([i,images])

pool=multiprocessing.Pool()
print pool
pool.map(func,iter_args)

En los documentos de multiprocessing está la función recv_bytes que genera un IOError. ¿Podría ser por esto? ( https://python.readthedocs.org/en/v2.7.2/library/multiprocessing.html )

EDITAR Si utilizo images como una matriz numpy en lugar de una lista, obtengo un error diferente: SystemError: NULL result without error in PyObject_Call . Un código un poco diferente:

import numpy as np
import multiprocessing

def func(args):
    i=args[0]
    images=args[1]
    print i
    return 0

N=1500       #N=1000 works fine

images=[]
for i in np.arange(N):
    images.append(np.random.random_integers(1,100,size=(500,500)))
images=np.array(images)                                            #new

iter_args=[]
for i in range(0,1):
    iter_args.append([i,images])

pool=multiprocessing.Pool()
print pool
pool.map(func,iter_args)

EDIT2 La función real que utilizo es:

def func(args):
    i=args[0]
    images=args[1]
    image=np.mean(images,axis=0)
    np.savetxt("image%d.txt"%(i),image)
    return 0

Además, los iter_args no contienen el mismo conjunto de imágenes:

iter_args=[]
for i in range(0,1):
    rand_ind=np.random.random_integers(0,N-1,N)
    iter_args.append([i,images[rand_ind]])

Answer 1

Cuando ejecuto tu programa, en realidad me da un error claro:

OSError: [Errno 12] Cannot allocate memory

Como mencionaron otros usuarios, la solución a su problema es simplemente agregar memoria (mucho) o cambiar la forma en que su programa está manejando las imágenes.

La razón por la que está utilizando tanta memoria es porque asigna su memoria para sus imágenes en un nivel de módulo. Por lo tanto, cuando el multiprocesamiento inicia el proceso, también se copian todas las imágenes (lo que no es libre de acuerdo con los objetos de memoria compartida en el multiprocesamiento de Python ), esto no es necesario porque también está dando las imágenes como un argumento a la función que el módulo multiproceso También copias que utilizan ipc y pickle, esto probablemente resultaría en una falta de memoria. Pruebe una de las soluciones propuestas dadas por los otros usuarios.

Answer 2

Es probable que Python cargue sus datos en su memoria RAM y necesita que esta memoria esté disponible. ¿Has comprobado el uso de memoria de tu computadora?

Además, como mencionó Patrick, estás cargando 3 GB de datos, asegúrate de usar la versión de 64 bits de Python cuando llegues al límite de memoria de 32 bits. Esto podría causar que su proceso se bloquee: Python de 32 contra 64 bits

Otra mejora sería usar Python 3.4 en lugar de 2.7. La implementación de Python 3 parece estar optimizada para rangos muy grandes, vea el rendimiento de la lista / generador Python3 vs Python2

Answer 3

Estás creando un grupo y enviando todas las imágenes a la vez a func (). Si puede trabajar con una sola imagen a la vez, intente algo como esto, que se ejecuta hasta completar con N = 10000 en 35s con Python 2.7.10 para mí:

import numpy as np
import multiprocessing

def func(args):
    i = args[0]
    img = args[1]
    print "{}: {} {}".format(i, img.shape, img.sum())
    return 0

N=10000

images = ((i, np.random.random_integers(1,100,size=(500,500))) for i in xrange(N))
pool=multiprocessing.Pool(4)
pool.imap(func, images)
pool.close()
pool.join()

La clave aquí es usar iteradores para que no tenga que guardar todos los datos en la memoria de una vez. Por ejemplo, convertí imágenes de una matriz que contenía todos los datos en una expresión generadora para crear la imagen solo cuando fuera necesario. Puede modificar esto para cargar sus imágenes desde el disco o lo que sea. También usé pool.imap en lugar de pool.map.

Si puede, intente cargar los datos de imagen en la función de trabajo. En este momento, debe serializar todos los datos y enviarlos a otro proceso. Si los datos de su imagen son más grandes, esto podría ser un cuello de botella.

[actualizar ahora que sabemos que func tiene que manejar todas las imágenes a la vez]

Podrías hacer una media iterativa en tus imágenes. Aquí hay una solución sin utilizar multiprocesamiento. Para usar el multiprocesamiento, puede dividir sus imágenes en partes y agruparlas en la piscina.

import numpy as np

N=10000
shape = (500,500)

def func(images):
    average = np.full(shape, 0)
    for i, img in images:
        average += img / N
    return average

images = ((i, np.full(shape,i)) for i in range(N))

print func(images)

Answer 4

Esto es lo que resolvió el problema: declarar las imágenes globales.

import numpy as np
import multiprocessing


N=1500       #N=1000 works fine

images=[]
for i in np.arange(N):
    images.append(np.random.random_integers(1,100,size=(500,500)))

def func(args):
    i=args[0]
    images=images
    print i
    return 0

iter_args=[]
for i in range(0,1):
    iter_args.append([i])

pool=multiprocessing.Pool()
print pool
pool.map(func,iter_args)