Python: multiprocessing.map: Wenn ein Prozess eine Ausnahme auslöst, warum werden dann die finally-Blöcke anderer Prozesse nicht aufgerufen?

Question

Mein Verständnis ist, dass finally-Klauseln müssen *immer* ausgeführt werden, wenn der try-Block betreten wurde.

import random

from multiprocessing import Pool
from time import sleep

def Process(x):
  try:
    print x
    sleep(random.random())
    raise Exception('Exception: ' + x)
  finally:
    print 'Finally: ' + x

Pool(3).map(Process, ['1','2','3'])

Die erwartete Ausgabe ist, dass für jedes x, das auf Zeile 8 einzeln ausgegeben wird, es ein Vorkommen von 'Finally x' geben muss.

Beispiel-Ausgabe:

$ python bug.py 
1
2
3
Finally: 2
Traceback (most recent call last):
  File "bug.py", line 14, in 
    Pool(3).map(Process, ['1','2','3'])
  File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/pool.py", line 225, in map
    return self.map_async(func, iterable, chunksize).get()
  File "/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/pool.py", line 522, in get
    raise self._value
Exception: Exception: 2

Es scheint, dass eine Ausnahme, die einen Prozess beendet, auch den Eltern- und Geschwisterprozess beendet, obwohl in anderen Prozessen noch weitere Arbeit erforderlich ist.

Warum liege ich falsch? Warum ist das korrekt? Wenn dies korrekt ist, wie sollte man Ressourcen sicher in multiprozessiertem Python aufräumen?

Answer 1

Kurze Antwort: SIGTERM hat Vorrang vor finally.

Lange Antwort: Aktivieren Sie das Logging mit mp.log_to_stderr():

import random
import multiprocessing as mp
import time
import logging

logger=mp.log_to_stderr(logging.DEBUG)

def Process(x):
    try:
        logger.info(x)
        time.sleep(random.random())
        raise Exception('Exception: ' + x)
    finally:
        logger.info('Finally: ' + x)

result=mp.Pool(3).map(Process, ['1','2','3'])

Die Protokollierung enthält:

[DEBUG/MainProcess] terminating workers

Was diesem Code in multiprocessing.pool._terminate_pool entspricht:

    if pool and hasattr(pool[0], 'terminate'):
        debug('terminating workers')
        for p in pool:
            p.terminate()

Jedes p in pool ist ein multiprocessing.Process und das Aufrufen von terminate (zumindest auf Nicht-Windows-Maschinen) ruft SIGTERM auf:

aus multiprocessing/forking.py:

class Popen(object)
    def terminate(self):
        ...
            try:
                os.kill(self.pid, signal.SIGTERM)
            except OSError, e:
                if self.wait(timeout=0.1) is None:
                    raise

Es kommt also darauf an, was passiert, wenn einem Python-Prozess in einer try-Suite ein SIGTERM gesendet wird.

Betrachten Sie das folgende Beispiel (test.py):

import time    
def worker():
    try:
        time.sleep(100)        
    finally:
        print('enter finally')
        time.sleep(2) 
        print('exit finally')    
worker()

Wenn Sie es ausführen und dann ein SIGTERM senden, endet der Prozess sofort, ohne die finally-Suite zu betreten, wie durch keine Ausgabe und keine Verzögerung belegt.

In einem Terminal:

% test.py

Im zweiten Terminal:

% pkill -TERM -f "test.py"

Ergebnis im ersten Terminal:

Terminated

Vergleichen Sie das mit dem, was passiert, wenn der Prozess ein SIGINT erhalten hat (C-c):

Im zweiten Terminal:

% pkill -INT -f "test.py"

Ergebnis im ersten Terminal:

enter finally
exit finally
Traceback (most recent call last):
  File "/home/unutbu/pybin/test.py", line 14, in 
    worker()
  File "/home/unutbu/pybin/test.py", line 8, in worker
    time.sleep(100)        
KeyboardInterrupt

Fazit: SIGTERM hat Vorrang vor finally.

Answer 2

Die Antwort von unutbu erklärt eindeutig warum das beobachtete Verhalten auftritt. Es sollte jedoch betont werden, dass SIGTERM nur gesendet wird, weil die Funktion multiprocessing.pool._terminate_pool implementiert ist. Wenn Sie vermeiden können, Pool zu verwenden, dann können Sie das gewünschte Verhalten erhalten. Hier ist ein beispielhaftes Beispiel:

from multiprocessing import Process
from time import sleep
import random

def f(x):
    try:
        sleep(random.random()*10)
        raise Exception
    except:
        print "Ausnahme im Prozess aufgefangen:", x
        # Sorgen Sie dafür, dass dies länger als die Ausnahme-Klausel in der Hauptfunktion dauert.
        sleep(3)
    finally:
        print "Prozessbereinigung:", x

if __name__ == '__main__':
    processes = []
    for i in range(4):
        p = Process(target=f, args=(i,))
        p.start()
        processes.append(p)
    try:
        for process in processes:
            process.join()
    except:
        print "Ausnahme in der Hauptfunktion aufgefangen."
    finally:
        print "Hauptbereinigung."

Nachdem ein SIGINT gesendet wurde, lautet der Beispielausgang:

Ausnahme im Prozess aufgefangen: 0
^C
Prozessbereinigung: 0
Ausnahme in der Hauptfunktion aufgefangen.
Hauptbereinigung.
Ausnahme im Prozess aufgefangen: 1
Ausnahme im Prozess aufgefangen: 2
Ausnahme im Prozess aufgefangen: 3
Prozessbereinigung: 1
Prozessbereinigung: 2
Prozessbereinigung: 3

Beachten Sie, dass die finally-Klausel für alle Prozesse ausgeführt wird. Wenn Sie gemeinsam genutzten Speicher benötigen, sollten Sie erwägen, Queue, Pipe, Manager oder einen externen Speicher wie redis oder sqlite3 zu verwenden.

Answer 3

finally gibt die ursprüngliche Ausnahme wieder , es sei denn, Sie return davon. Die Ausnahme wird dann von Pool.map erhoben und beendet Ihre gesamte Anwendung. Die Unterprozesse werden beendet und es werden keine anderen Ausnahmen angezeigt.

Sie können ein return hinzufügen, um die Ausnahme zu unterdrücken:

def Process(x):
  try:
    print x
    sleep(random.random())
    raise Exception('Ausnahme: ' + x)
  finally:
    print 'Schließlich: ' + x
    return

Dann sollte None in Ihrem map-Ergebnis stehen, wenn eine Ausnahme aufgetreten ist.