NumPy vs. Multiprozessing und mmap

Question

Ich verwende Python multiprocessing-Modul, um große Anzahl Arrays parallel zu verarbeiten. Die Arrays sind im Master-Prozess unter Verwendung von numpy.load(mmap_mode='r') memory-mapped. Danach gibt multiprocessing.Pool() den Prozess ab (ich nehme an).

Alles scheint gut zu funktionieren, es sei denn ich wie immer Linien bin:

Attribute (" 'NoneType' Objekt hat kein Attribut 'sagen'",) in <bound method memmap.__del__ of memmap([ 0.57735026, 0.57735026, 0.57735026, 0. , 0. , 0. , 0. , 0. , 0. , 0. , 0. , 0. ], dtype=float32)> ignoriert

in der Unittest-Protokolle. Die Tests laufen trotzdem gut.

Irgendeine Idee, was dort vor sich geht?

Verwenden von Python 2.7.2, OS X, NumPy 1.6.1.

---

UPDATE:

Nach einigem Debugging, gejagt I, die Ursache zu einem Code-Weg, der eine (kleine Scheibe) diesen speicherabgebildete numpy Array als Eingabe für einen Anruf Pool.imap wurde mit.

Offenbar ist das "Problem" mit der Art und Weise übergibt seine Eingabe an die neuen Prozesse: Es verwendet Gurke. Dies funktioniert nicht mit mmap ed numpy Arrays, und etwas im Inneren bricht, was zu dem Fehler führt.

Ich fand this reply von Robert Kern, die das gleiche Problem zu behandeln scheint. Er schlägt vor, einen speziellen Codepfad zu erstellen, wenn der Eingang imap von einem Speicher-abgebildeten Array kommt: Speicher-Mapping des gleichen Arrays manuell im erzeugten Prozess.

Das wäre so kompliziert und hässlich, dass ich lieber mit dem Fehler und den zusätzlichen Speicherkopien leben würde. Gibt es eine andere Möglichkeit, die beim Ändern des vorhandenen Codes leichter wäre?

Answer 1

Mein üblicher Ansatz (wenn Sie mit zusätzlichen Speicherkopien leben können) besteht darin, alle E/A-Vorgänge in einem Prozess auszuführen und diese dann an einen Pool von Arbeitsthreads zu senden. Um ein Segment eines memmapped Array in den Speicher zu laden, tun Sie einfach x = np.array(data[yourslice]) (data[yourslice].copy() tut dies nicht wirklich, was zu einiger Verwirrung führen kann.).

Zuerst weg, lassen Sie uns einige Testdaten erzeugen:

import numpy as np 
np.random.random(10000).tofile('data.dat') 

Sie Ihre Fehler mit so etwas wie dieses reproduzieren kann:

import numpy as np 
import multiprocessing 

def main(): 
    data = np.memmap('data.dat', dtype=np.float, mode='r') 
    pool = multiprocessing.Pool() 
    results = pool.imap(calculation, chunks(data)) 
    results = np.fromiter(results, dtype=np.float) 

def chunks(data, chunksize=100): 
    """Overly-simple chunker...""" 
    intervals = range(0, data.size, chunksize) + [None] 
    for start, stop in zip(intervals[:-1], intervals[1:]): 
     yield data[start:stop] 

def calculation(chunk): 
    """Dummy calculation.""" 
    return chunk.mean() - chunk.std() 

if __name__ == '__main__': 
    main() 

Und wenn Sie gerade wechseln np.array(data[start:stop]) statt Nachgeben, werden Sie das Problem beheben:

import numpy as np 
import multiprocessing 

def main(): 
    data = np.memmap('data.dat', dtype=np.float, mode='r') 
    pool = multiprocessing.Pool() 
    results = pool.imap(calculation, chunks(data)) 
    results = np.fromiter(results, dtype=np.float) 

def chunks(data, chunksize=100): 
    """Overly-simple chunker...""" 
    intervals = range(0, data.size, chunksize) + [None] 
    for start, stop in zip(intervals[:-1], intervals[1:]): 
     yield np.array(data[start:stop]) 

def calculation(chunk): 
    """Dummy calculation.""" 
    return chunk.mean() - chunk.std() 

if __name__ == '__main__': 
    main() 

natürlich, das macht einen zusätzlichen In-Memory-Kopie jedes Chunks.

Auf lange Sicht werden Sie wahrscheinlich feststellen, dass es einfacher ist, von memmapped Dateien zu wechseln und zu etwas wie HDF zu wechseln. Dies gilt insbesondere, wenn Ihre Daten mehrdimensional sind. (Ich würde empfehlen h5py, aber pyTables ist nett, wenn Ihre Daten "Tabelle-wie" sind.)

Viel Glück, auf jeden Fall!