Les Interruptions de clavier avec python multitraitement Piscine

C'est un Python bug. Lors de l'attente d'une condition dans le filetage.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 livré qu'en attendre() retourne, et il ne revient jamais, de sorte que l'interruption n'arrive jamais. KeyboardInterrupt devraient presque certainement interrompre un état d'attente.

Noter que cela n'arrive pas si un délai est spécifié; cond.attendre(1) recevront l'interrompre immédiatement. Donc, une solution de contournement est de spécifier un délai. Pour ce faire, remplacer

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

avec

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

ou similaire.

OriginalL'auteur Glenn Maynard

Pour certaines raisons, les seules exceptions hérité de la base Exception classe sont traitées normalement. Comme solution de contournement, vous pouvez ré-augmenter votre KeyboardInterrupt comme un Exception exemple:

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 frappez ^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

OriginalL'auteur Andrey Vlasovskikh

Il semble qu'il y a deux questions qui font exceptions lors de multitraitement ennuyeux. Le premier (noté par Glenn), c'est que vous devez utiliser map_async avec un délai d'attente au lieu de map afin d'obtenir une réponse immédiate (c'est à dire, ne pas terminer le traitement de l'ensemble de la liste). La deuxième (noté par Andrey), c'est que le multitraitement ne pas intercepter des exceptions qui n'héritent pas de Exception (par exemple, SystemExit). Voici donc ma solution qui traite à la fois de ces:

import sys
import functools
import traceback
import multiprocessing

def _poolFunctionWrapper(function, arg):
    """Run function under the pool

    Wrapper around function to catch exceptions that don't inherit from
    Exception (which aren't caught by multiprocessing, so that you end
    up hitting the timeout).
    """
    try:
        return function(arg)
    except:
        cls, exc, tb = sys.exc_info()
        if issubclass(cls, Exception):
            raise # No worries
        # Need to wrap the exception with something multiprocessing will recognise
        import traceback
        print "Unhandled exception %s (%s):\n%s" % (cls.__name__, exc, traceback.format_exc())
        raise Exception("Unhandled exception: %s (%s)" % (cls.__name__, exc))

def _runPool(pool, timeout, function, iterable):
    """Run the pool

    Wrapper around pool.map_async, to handle timeout.  This is required so as to
    trigger an immediate interrupt on the KeyboardInterrupt (Ctrl-C); see
    http://stackoverflow.com/questions/1408356/keyboard-interrupts-with-pythons-multiprocessing-pool

    Further wraps the function in _poolFunctionWrapper to catch exceptions
    that don't inherit from Exception.
    """
    return pool.map_async(functools.partial(_poolFunctionWrapper, function), iterable).get(timeout)

def myMap(function, iterable, numProcesses=1, timeout=9999):
    """Run the function on the iterable, optionally with multiprocessing"""
    if numProcesses > 1:
        pool = multiprocessing.Pool(processes=numProcesses, maxtasksperchild=1)
        mapFunc = functools.partial(_runPool, pool, timeout)
    else:
        pool = None
        mapFunc = map
    results = mapFunc(function, iterable)
    if pool is not None:
        pool.close()
        pool.join()
    return results

OriginalL'auteur Paul Price

J'ai trouvé, pour le moment, la meilleure solution est de ne pas utiliser le multitraitement.piscine fonctionnalité, mais plutôt rouler votre propre piscine fonctionnalité. J'ai donné un exemple montrant que l'erreur avec apply_async, ainsi qu'un exemple montrant comment éviter l'utilisation de la piscine de la fonctionnalité tout à fait.

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

OriginalL'auteur bboe

Vous pouvez essayer d'utiliser le apply_async méthode d'un objet Pool, comme ceci:

import multiprocessing
import time
from datetime import datetime


def test_func(x):
    time.sleep(2)
    return x**2


def apply_multiprocessing(input_list, input_function):
    pool_size = 5
    pool = multiprocessing.Pool(processes=pool_size, maxtasksperchild=10)

    try:
        jobs = {}
        for value in input_list:
            jobs[value] = pool.apply_async(input_function, [value])

        results = {}
        for value, result in jobs.items():
            try:
                results[value] = result.get()
            except KeyboardInterrupt:
                print "Interrupted by user"
                pool.terminate()
                break
            except Exception as e:
                results[value] = e
        return results
    except Exception:
        raise
    finally:
        pool.close()
        pool.join()


if __name__ == "__main__":
    iterations = range(100)
    t0 = datetime.now()
    results1 = apply_multiprocessing(iterations, test_func)
    t1 = datetime.now()
    print results1
    print "Multi: {}".format(t1 - t0)

    t2 = datetime.now()
    results2 = {i: test_func(i) for i in iterations}
    t3 = datetime.now()
    print results2
    print "Non-multi: {}".format(t3 - t2)

De sortie:

100
Multiprocessing run time: 0:00:41.131000
100
Non-multiprocessing run time: 0:03:20.688000

Un avantage de cette méthode est que les résultats traitées avant l'interruption sera renvoyé dans les résultats de dictionnaire:

>>> apply_multiprocessing(range(100), test_func)
Interrupted by user
{0: 0, 1: 1, 2: 4, 3: 9, 4: 16, 5: 25}

OriginalL'auteur bparker856