O teclado interrompe o pool de multiprocessamento do python

Como posso lidar com eventos KeyboardInterrupt com os pools de multiprocessamento do python? Aqui está um exemplo simples:

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

Ao executar o código acima, ele KeyboardInterrupté gerado quando pressiono ^C, mas o processo simplesmente trava nesse momento e eu tenho que eliminá-lo externamente.

Quero poder pressionar ^Ca qualquer momento e fazer com que todos os processos sejam encerrados normalmente.

Este é um bug do Python. Ao aguardar uma condição em threading.Condition.wait (), KeyboardInterrupt nunca é enviado. Repro:

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

A exceção KeyboardInterrupt não será entregue até wait () retornar e nunca retornar, portanto a interrupção nunca acontece. KeyboardInterrupt quase certamente deve interromper uma espera de condição.

Observe que isso não acontece se um tempo limite for especificado; cond.wait (1) receberá a interrupção imediatamente. Portanto, uma solução alternativa é especificar um tempo limite. Para fazer isso, substitua

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

com

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

ou similar.

Pelo que descobri recentemente, a melhor solução é configurar os processos de trabalho para ignorar completamente o SIGINT e restringir todo o código de limpeza ao processo pai. Isso corrige o problema para os processos de trabalho ocioso e ocupado e não requer nenhum código de manipulação de erros nos processos filhos.

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

A explicação e o código de exemplo completo podem ser encontrados em http://noswap.com/blog/python-multiprocessing-keyboardinterrupt/ e http://github.com/jreese/multiprocessing-keyboardinterrupt, respectivamente.

Por alguns motivos, apenas exceções herdadas da Exceptionclasse base são tratadas normalmente. Como solução alternativa, você pode aumentar novamente o seu KeyboardInterruptcomo uma Exceptioninstância:

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

Normalmente você obteria a seguinte saída:

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

Então, se você acertar ^C, receberá:

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

Geralmente essa estrutura simples funciona para Ctrl- Cno Pool:

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

signal.signal(signal.SIGINT, signal_handle)

Como foi afirmado em alguns posts semelhantes:

Capture a interrupção do teclado no Python sem tentar, exceto

A resposta votada não aborda a questão central, mas um efeito colateral semelhante.

Jesse Noller, o autor da biblioteca de multiprocessamento, explica como lidar corretamente com CTRL + C ao usar multiprocessing.Poolem uma postagem de blog antiga .

import signal
from multiprocessing import Pool


def initializer():
    """Ignore CTRL+C in the worker process."""
    signal.signal(signal.SIGINT, signal.SIG_IGN)


pool = Pool(initializer=initializer)

try:
    pool.map(perform_download, dowloads)
except KeyboardInterrupt:
    pool.terminate()
    pool.join()

Parece que há dois problemas que fazem exceções, enquanto o multiprocessamento é irritante. A primeira (observada por Glenn) é que você precisa usar map_asynccom um tempo limite em vez de mapobter uma resposta imediata (ou seja, não termine de processar a lista inteira). O segundo (observado por Andrey) é que o multiprocessamento não captura exceções que não são herdadas de Exception(por exemplo, SystemExit). Então, aqui está minha solução que lida com os dois:

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

Descobri, por enquanto, a melhor solução é não usar o recurso multiprocessing.pool, mas sim rolar sua própria funcionalidade de pool. Forneci um exemplo demonstrando o erro com apply_async, bem como um exemplo mostrando como evitar o uso completo da funcionalidade do pool.

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

Eu sou um novato em Python. Eu estava procurando em todos os lugares por respostas e encontrei este e alguns outros blogs e vídeos do youtube. Eu tentei copiar, colar o código do autor acima e reproduzi-lo no meu python 2.7.13 no windows 7 de 64 bits. É perto do que eu quero alcançar.

Fiz meus processos filhos ignorar o ControlC e fazer com que o processo pai fosse encerrado. Parece que ignorar o processo filho evita esse problema para mim.

#!/usr/bin/python

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


def slowly_square(i):
    try:
        print "<slowly_square> Sleeping and later running a square calculation..."
        sleep(1)
        return i * i
    except KeyboardInterrupt:
        print "<child processor> Don't care if you say CtrlC"
        pass


def go():
    pool = Pool(8)

    try:
        results = pool.map(slowly_square, range(40))
    except KeyboardInterrupt:
        pool.terminate()
        pool.close()
        print "You cancelled the program!"
        exit(1)
    print "Finally, here are the results", results


if __name__ == '__main__':
    go()

A parte iniciada em pool.terminate()nunca parece ser executada.

Você pode tentar usar o método apply_async de um objeto Pool, assim:

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)

Resultado:

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

Uma vantagem desse método é que os resultados processados ​​antes da interrupção serão retornados no dicionário de resultados:

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

Estranhamente, parece que você precisa lidar KeyboardInterruptcom as crianças também. Eu esperava que isso funcionasse como está escrito ... tente mudar slowly_squarepara:

def slowly_square(i):
    try:
        sleep(1)
        return i * i
    except KeyboardInterrupt:
        print 'You EVIL bastard!'
        return 0

Isso deve funcionar como você esperava.