Interruptions du clavier avec le pool de multitraitement de python

Question

Comment puis-je gérer les événements KeyboardInterrupt avec les Pools multiprocesseurs de python ? Voici un exemple simple :

from multiprocessing import Pool
from time import sleep
from sys import exit

def slowly_square(i):
    sleep(1)
    return i*i

def go():
    pool = Pool(8)
    try:
        results = pool.map(slowly_square, range(40))
    except KeyboardInterrupt:
        # **** THIS PART NEVER EXECUTES. ****
        pool.terminate()
        print "You cancelled the program!"
        sys.exit(1)
    print "\nFinally, here are the results: ", results

if __name__ == "__main__":
    go()

Lors de l'exécution du code ci-dessus, le KeyboardInterrupt se lève lorsque j'appuie sur ^C mais le processus se bloque tout simplement à ce moment-là et je dois le tuer en externe.

Je veux pouvoir appuyer sur ^C à tout moment et faire en sorte que tous les processus se terminent gracieusement.

Answer 1

C'est un bug de Python. Lors de l'attente d'une condition dans threading.Condition.wait(), KeyboardInterrupt n'est jamais envoyé. Repro :

import threading
cond = threading.Condition(threading.Lock())
cond.acquire()
cond.wait(None)
print "done"

L'exception KeyboardInterrupt ne sera pas délivrée avant le retour de wait(), qui ne revient jamais, de sorte que l'interruption ne se produit jamais. KeyboardInterrupt devrait presque certainement interrompre une condition d'attente.

Notez que cela ne se produit pas si un timeout est spécifié ; cond.wait(1) recevra l'interruption immédiatement. Une solution de contournement consiste donc à spécifier un délai d'attente. Pour ce faire, remplacez

    results = pool.map(slowly_square, range(40))

avec

    results = pool.map_async(slowly_square, range(40)).get(9999999)

ou similaire.

Answer 2

D'après ce que j'ai découvert récemment, la meilleure solution consiste à configurer les processus de travail pour qu'ils ignorent complètement SIGINT, et à confier tout le code de nettoyage au processus parent. Cela résout le problème pour les processus travailleurs inactifs et occupés, et ne nécessite aucun code de gestion des erreurs dans vos processus enfants.

import signal

...

def init_worker():
    signal.signal(signal.SIGINT, signal.SIG_IGN)

...

def main()
    pool = multiprocessing.Pool(size, init_worker)

    ...

    except KeyboardInterrupt:
        pool.terminate()
        pool.join()

L'explication et le code d'exemple complet se trouvent à l'adresse suivante http://noswap.com/blog/python-multiprocessing-keyboardinterrupt/ y http://github.com/jreese/multiprocessing-keyboardinterrupt respectivement.

Answer 3

Pour certaines raisons, seules les exceptions héritées de la base Exception sont traitées normalement. En guise de solution de contournement, vous pouvez relancer la classe KeyboardInterrupt comme un Exception instance :

from multiprocessing import Pool
import time

class KeyboardInterruptError(Exception): pass

def f(x):
    try:
        time.sleep(x)
        return x
    except KeyboardInterrupt:
        raise KeyboardInterruptError()

def main():
    p = Pool(processes=4)
    try:
        print 'starting the pool map'
        print p.map(f, range(10))
        p.close()
        print 'pool map complete'
    except KeyboardInterrupt:
        print 'got ^C while pool mapping, terminating the pool'
        p.terminate()
        print 'pool is terminated'
    except Exception, e:
        print 'got exception: %r, terminating the pool' % (e,)
        p.terminate()
        print 'pool is terminated'
    finally:
        print 'joining pool processes'
        p.join()
        print 'join complete'
    print 'the end'

if __name__ == '__main__':
    main()

Normalement, vous devriez obtenir le résultat suivant :

staring the pool map
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
pool map complete
joining pool processes
join complete
the end

Donc si vous tapez ^C vous obtiendrez :

staring the pool map
got ^C while pool mapping, terminating the pool
pool is terminated
joining pool processes
join complete
the end

Answer 4

Cette structure simple fonctionne généralement pour Ctrl - C sur la piscine :

def signal_handle(_signal, frame):
    print "Stopping the Jobs."

signal.signal(signal.SIGINT, signal_handle)

Comme cela a été dit dans quelques posts similaires :

Capturer l'interruption du clavier en Python sans try-except

Answer 5

J'ai trouvé, pour l'instant, que la meilleure solution est de ne pas utiliser la fonctionnalité multiprocessing.pool mais plutôt de développer votre propre fonctionnalité de pool. J'ai fourni un exemple démontrant l'erreur avec apply_async ainsi qu'un exemple montrant comment éviter complètement l'utilisation de la fonctionnalité de pool.

http://www.bryceboe.com/2010/08/26/python-multiprocessing-and-keyboardinterrupt/