Python Multiprocessing Pool Karte und Imap

Question

Ich habe ein multiprocessing Skript mit pool.map, das funktioniert. Das Problem ist, dass nicht alle Prozesse so lange dauern, bis einige Prozesse einschlafen, weil sie warten, bis alle Prozesse beendet sind (dasselbe Problem wie in this question). Einige Dateien sind in weniger als einer Sekunde fertig, andere benötigen Minuten (oder Stunden).

Wenn ich das Handbuch (and this post) richtig verstehe, wartet pool.imap nicht auf alle Prozesse zu beenden, wenn man fertig ist, stellt es eine neue Datei zur Verfügung. Wenn ich das versuche, beschleunigt das Skript die zu verarbeitenden Dateien, die kleinen werden wie erwartet verarbeitet, die großen Dateien (deren Verarbeitung länger dauert) werden erst am Ende beendet (werden ohne vorherige Ankündigung beendet). Ist das normales Verhalten für pool.imap, oder muss ich weitere Befehle/Parameter hinzufügen? Wenn ich den time.sleep(100) Teil in den else Teil als Test hinzufüge, verarbeitet es größere Dateien, aber die anderen Prozesse schlafen ein. Irgendwelche Vorschläge ? Danke

def process_file(infile): 
    #read infile 
    #compare things in infile 
    #acquire Lock, save things in outfile, release Lock 
    #delete infile 

def main(): 
    #nprocesses = 8 
    global filename 
    pathlist = ['tmp0', 'tmp1', 'tmp2', 'tmp3', 'tmp4', 'tmp5', 'tmp6', 'tmp7', 'tmp8', 'tmp9'] 
    for d in pathlist: 
     os.chdir(d)  
     todolist = [] 
     for infile in os.listdir(): 
      todolist.append(infile) 
     try: 
      p = Pool(processes=nprocesses) 
      p.imap(process_file, todolist) 
     except KeyboardInterrupt:     
      print("Shutting processes down") 
      # Optionally try to gracefully shut down the worker processes here.  
      p.close() 
      p.terminate() 
      p.join() 
     except StopIteration: 
      continue  
     else: 
      time.sleep(100) 
      os.chdir('..') 
     p.close() 
     p.join() 

if __name__ == '__main__': 
    main()  

Answer 1

Da Sie bereits alle Ihre Dateien in eine Liste eingetragen haben, könnten Sie sie direkt in eine Warteschlange stellen. Die Warteschlange wird dann mit Ihren Unterprozessen geteilt, die die Dateinamen aus der Warteschlange übernehmen und ihre Aufgaben erledigen. Keine Notwendigkeit, es zweimal zu tun (zuerst in Liste, dann Pickle-Liste von Pool.imap). Pool.imap tut genau dasselbe, aber ohne dass du es weißt.

def process_file(inqueue): 
    for infile in iter(inqueue.get, "STOP"): 
     #do stuff until inqueue.get returns "STOP" 
    #read infile 
    #compare things in infile 
    #acquire Lock, save things in outfile, release Lock 
    #delete infile 

def main(): 
    nprocesses = 8 
    global filename 
    pathlist = ['tmp0', 'tmp1', 'tmp2', 'tmp3', 'tmp4', 'tmp5', 'tmp6', 'tmp7', 'tmp8', 'tmp9'] 
    for d in pathlist: 
     os.chdir(d)  
     todolist = Queue() 
     for infile in os.listdir(): 
      todolist.put(infile) 
     process = [Process(target=process_file, 
         args=(todolist) for x in range(nprocesses)] 
     for p in process: 
      #task the processes to stop when all files are handled 
      #"STOP" is at the very end of queue 
      todolist.put("STOP") 
     for p in process: 
      p.start() 
     for p in process: 
      p.join()  
if __name__ == '__main__': 
    main()