Python Multiprozessing Race-Bedingung

Question

Ich entdeckte einen seltsamen Fehler bei der Verwendung concurrent.futures aus mehreren Textdateien lesen.

Hier ist ein kleines reproduzierbares Beispiel:

import os 
import concurrent.futures 

def read_file(file): 
    with open(os.path.join(data_dir, file),buffering=1000) as f: 
     for row in f: 
      try: 
       print(row) 
      except Exception as e: 
       print(str(e)) 

if __name__ == '__main__': 
    data_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), 'data')) 
    files = ['file1', 'file2'] 
    with concurrent.futures.ProcessPoolExecutor() as executor: 
     for file,_ in zip(files,executor.map(read_file,files)): 
      pass  

file1 und file2 beliebige Textdateien im Verzeichnis data.

mir die folgende Fehlermeldung erhalten (im Grunde ein Prozess versucht, data_dir Variable zu lesen, bevor sie zugeordnet ist):

concurrent.futures.process._RemoteTraceback: 
""" 
Traceback (most recent call last): 
    File "C:\Users\my_username\AppData\Local\Continuum\Anaconda3\lib\concurrent\futures\process.py", line 175, in _process_worker 
    r = call_item.fn(*call_item.args, **call_item.kwargs) 
    File "C:\Users\my_username\AppData\Local\Continuum\Anaconda3\lib\concurrent\futures\process.py", line 153, in _process_chunk 
    return [fn(*args) for args in chunk] 
    File "C:\Users\my_username\AppData\Local\Continuum\Anaconda3\lib\concurrent\futures\process.py", line 153, in <listcomp> 
    return [fn(*args) for args in chunk] 
    File "C:\Users\my_username\Downloads\example.py", line 5, in read_file 
    with open(os.path.join(data_dir, file),buffering=1000) as f: 
NameError: name 'data_dir' is not defined 
""" 

The above exception was the direct cause of the following exception: 

Traceback (most recent call last): 
    File "example.py", line 16, in <module> 
    for file,_ in zip(files,executor.map(read_file,files)): 
    File "C:\Users\my_username\AppData\Local\Continuum\Anaconda3\lib\concurrent\futures\_base.py", line 556, in result_iterator 
    yield future.result() 
    File "C:\Users\my_username\AppData\Local\Continuum\Anaconda3\lib\concurrent\futures\_base.py", line 405, in result 
    return self.__get_result() 
    File "C:\Users\my_username\AppData\Local\Continuum\Anaconda3\lib\concurrent\futures\_base.py", line 357, in __get_result 
    raise self._exception 
NameError: name 'data_dir' is not defined 

Wenn ich data_dir Zuordnung vor if __name__ == '__main__': Block platzieren, ich habe nicht diesen Fehler und die Code wird wie erwartet ausgeführt.

Was verursacht diesen Fehler? Offensichtlich wird data_dir zugewiesen, bevor irgendwelche asynchronen Aufrufe in beiden Fällen vorgenommen werden sollten.

Answer 1

fork() nicht auf Fenster, so Python Verwendung spawn neuen Prozess zu starten, die einen neuen Python-Interpreter Prozess startet, wird kein Speicher freigegeben werden, aber Python wird try to recreate Arbeiter Funktion Umwelt in dem neuen Prozess, deshalb Modulebene Variable funktioniert. Siehe doc for more detail.

Answer 2

ProcessPoolExecutor spaws ein neuer Python Prozess, das richtige Modul importiert und ruft die Funktion, die Sie zur Verfügung stellen. Als data_dir wird nur definiert, wenn Sie das Modul ausführen, nicht wenn Sie importieren es, der Fehler ist zu erwarten.

Bereitstellung der data_dir Dateideskriptor als Argument an read_fileMacht Arbeit, wie ich glaube, dass die Prozesse, die Dateideskriptoren von ihren Eltern erben. Das müsstest du allerdings überprüfen.

Wenn Sie jedoch eine ThreadPoolExecutor verwenden, sollte Ihr Beispiel funktionieren, da die erstellten Threads Speicher teilen.