Tastatur-Interrupts mit python-multiprocessing-Pool

Dies ist ein Python bug. Beim warten auf einen Zustand, in dem threading.Zustand.wait(), KeyboardInterrupt nie gesendet. Repro:

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

Die KeyboardInterrupt-Ausnahme, werden nicht übermittelt, bis wait() kehrt zurück, und es kehrt nie zurück, so wird der interrupt nie passiert. KeyboardInterrupt sollte fast sicher unterbrechen, eine Bedingung warten.

Beachten Sie, dass dies nicht der Fall, wenn ein timeout angegeben ist; cond.warten(1), erhalten den interrupt sofort aus. Also, ein workaround ist die Angabe eines timeout. Um das zu tun, ersetzen

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

mit

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

oder ähnliches.

InformationsquelleAutor Glenn Maynard

Einige Gründe, nur Ausnahmen geerbt von der Basis Exception Klasse werden normal behandelt. Als workaround, können Sie re-raise your KeyboardInterrupt als Exception Beispiel:

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()

Normalerweise würden Sie die folgende Ausgabe erhalten:

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

Also, wenn Sie Treffer ^C erhalten Sie:

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

InformationsquelleAutor Andrey Vlasovskikh

Es scheint, es gibt zwei Probleme, Ausnahmen zu machen, während multiprocessing ärgerlich. Die erste (bemerkte Glenn) ist, dass Sie brauchen, um zu verwenden map_async mit einem timeout statt map um eine sofortige Antwort erhalten (D. H., nicht die Verarbeitung der gesamten Liste). Die zweite (bereits von Andrey) ist das multiprocessing nicht fangen von Ausnahmen, die nicht Erben aus Exception (z.B. SystemExit). Also hier ist meine Lösung, die sich mit beiden:

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

InformationsquelleAutor Paul Price

Fand ich, für die Zeit, die beste Lösung ist nicht zu verwenden, die multiprocessing.pool Funktion, sondern Rollen Sie Ihre eigenen pool-Funktionalität. Ich ein Beispiel, das die Fehler mit apply_async sowie ein Beispiel zeigt, wie Sie zu vermeiden, mit der pool-Funktionalität insgesamt.

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

InformationsquelleAutor bboe

Können Sie versuchen, mit der apply_async Methode, eine Pool-Objekt, wie folgt:

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)

Ausgabe:

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

Vorteil dieser Methode ist, dass die Ergebnisse verarbeitet, bevor die Unterbrechungen in den Ergebnissen zurückgegeben Wörterbuch:

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

InformationsquelleAutor bparker856