Verwendung eines Multiprozessor-Pools von Arbeitern

Question

Ich habe den folgenden Code geschrieben, um meinen faulen zweiten CPU-Kern zum Laufen zu bringen. Der Code sucht zunächst die gewünschten "sea"-Dateien in der Verzeichnishierarchie und führt dann eine Reihe von externen Skripten aus, um diese binären "sea"-Dateien zu verarbeiten und 50 bis 100 Text- und Binärdateien zu erzeugen. Wie der Titel der Frage nahelegt, geschieht dies parallel, um die Verarbeitungsgeschwindigkeit zu erhöhen.

Diese Frage stammt aus der langen Diskussion, die wir auf der IPython-Benutzerliste mit dem Titel "" geführt haben. ipcluster kann nicht gestartet werden ". Ich beginne mit meinen Experimenten zu den Parallelverarbeitungsfunktionalitäten von IPython.

Das Problem ist, dass ich diesen Code nicht richtig zum Laufen bringen kann. Wenn die Ordner, die "sea"-Dateien enthalten, nur "sea"-Dateien enthalten, beendet das Skript seine Ausführung, ohne externe Skriptläufe vollständig durchzuführen. (Angenommen, ich muss 30-50 externe Skripte ausführen, aber mein Skript mit aktiviertem Multiprocessing erschöpft sich erst nach der Ausführung des ersten Skripts in dieser externen Skriptkette). Interessanterweise läuft das Skript, wenn ich es in einem bereits verarbeiteten Ordner ausführe (d. h. in den zuvor verarbeiteten "See"-Dateien und den Ausgabedateien, die sich bereits in diesem Ordner befinden), aber dieses Mal erhalte ich Geschwindigkeitssteigerungen von etwa dem 2,4- bis 2,7-fachen der linearen Verarbeitungszeiten. Das ist nicht unbedingt zu erwarten, da ich nur eine Core 2 Duo 2.5 Ghz CPU in meinem Laptop habe. Obwohl ich einen CUDA-gesteuerten Grafikprozessor habe, hat das nichts mit meinem derzeitigen Kampf gegen parallele Berechnungen zu tun :)

Was könnte Ihrer Meinung nach die Ursache für dieses Problem sein?

Ich danke Ihnen für alle Kommentare und Vorschläge.

#!/usr/bin/env python

from multiprocessing import Pool
from subprocess import call
import os

def find_sea_files():

   file_list, path_list = [], []
   init = os.getcwd()

   for root, dirs, files in os.walk('.'):
      dirs.sort()
      for file in files:
          if file.endswith('.sea'):
              file_list.append(file)
              os.chdir(root)
              path_list.append(os.getcwd())
              os.chdir(init)

   return file_list, path_list

def process_all(pf):
   os.chdir(pf[0])
   call(['postprocessing_saudi', pf[1]])

if __name__ == '__main__':
   pool = Pool(processes=2)              # start 2 worker processes
   files, paths = find_sea_files()
   pathfile = [[paths[i],files[i]] for i in range(len(files))]
   pool.map(process_all, pathfile)

Answer 1

Ich würde damit beginnen, ein besseres Gefühl dafür zu bekommen, was im Arbeitsprozess vor sich geht. Das Multiprocessing-Modul verfügt über eine Protokollierung für seine Unterprozesse, falls Sie diese benötigen. Da Sie den Code vereinfacht haben, um das Problem einzugrenzen, würde ich einfach mit ein paar Druckanweisungen debuggen, etwa so (oder Sie können PrettyPrint die pf Array):

def process_all(pf):
   print "PID: ", os.getpid()
   print "Script Dir: ", pf[0]
   print "Script: ", pf[1]
   os.chdir(pf[0])
   call(['postprocessing_saudi', pf[1]])

if __name__ == '__main__':
   pool = Pool(processes=2)
   files, paths = find_sea_files()
   pathfile = [[paths[i],files[i]] for i in range(len(files))]
   pool.map(process_all, pathfile, 1) # Ensure the chunk size is 1
   pool.close()
   pool.join()

Die Version von Python, mit der ich dies erreicht habe, ist 2.6.4.

Answer 2

Mir fallen da mehrere Dinge ein:

1) Haben Sie die Pfaddateien ausgedruckt? Sind Sie sicher, dass sie alle richtig erstellt wurden?

a) Ich frage, weil Ihr os.walk ein wenig interessant ist; die dirs.sort() sollte in Ordnung sein, scheint aber ziemlich unnötig zu sein. os.chdir() sollte im Allgemeinen nicht verwendet werden; die Wiederherstellung debe in Ordnung sein, aber im Allgemeinen sollten Sie einfach Root an init anhängen.

2) Ich habe gesehen, dass multiprocessing auf python2.6 Probleme beim Erzeugen von Subprozessen aus Pools hat. (Ich hatte ein Skript, das Multiprocessing verwendet, um Subprozesse zu erzeugen. Diese Subprozesse konnten dann Multiprocessing nicht korrekt nutzen (der Pool blockierte)). Versuchen Sie Python 2.5 mit dem Mulitprocessing Backport.

3) Versuchen picloud Modul cloud.mp (das Multiprocessing verpackt, aber Pools etwas anders behandelt) und sehen Sie, ob das funktioniert.

Sie würden tun

cloud.mp.join(cloud.mp.map(process_all, pathfile))

(Haftungsausschluss: Ich bin einer der Entwickler von PiCloud)