¿Cómo matas a los futuros una vez que han comenzado?

Solución:

Es un poco doloroso. Esencialmente, sus subprocesos de trabajo deben estar terminados antes de que su subproceso principal pueda salir. No puede salir a menos que ellos lo hagan. La solución típica es tener algún estado global, que cada hilo puede verificar para determinar si deben hacer más trabajo o no.

Aquí está la cita que explica por qué. En esencia, si los hilos salieron cuando lo hace el intérprete, podrían suceder cosas malas.

Aquí tienes un ejemplo práctico. Tenga en cuenta que Cc tarda como máximo 1 segundo en propagarse debido a la duración de suspensión del subproceso secundario.

#!/usr/bin/env python
from __future__ import print_function

import concurrent.futures
import time
import sys

quit = False
def wait_a_bit(name):
    while not quit:
        print("{n} is doing work...".format(n=name))
        time.sleep(1)

def setup():
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
    future1 = executor.submit(wait_a_bit, "Jack")
    future2 = executor.submit(wait_a_bit, "Jill")

    # main thread must be doing "work" to be able to catch a Ctrl+C 
    # http://www.luke.maurits.id.au/blog/post/threads-and-signals-in-python.html
    while (not (future1.done() and future2.done())):
        time.sleep(1)

if __name__ == "__main__":
    try:
        setup()
    except KeyboardInterrupt:
        quit = True

Encontré esto, pero el problema que tuve fue que muchos futuros (decenas de miles) estarían esperando para ejecutarse y simplemente presionando Ctrl-C los dejaba esperando, no saliendo realmente. Estaba usando concurrent.futures.wait para ejecutar un ciclo de progreso y necesitaba agregar un try ... except KeyboardInterrupt para manejar la cancelación de Futuros inconclusos.

POLL_INTERVAL = 5
with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
    futures = [pool.submit(do_work, arg) for arg in large_set_to_do_work_over]
    # next line returns instantly
    done, not_done = concurrent.futures.wait(futures, timeout=0)
    try:
        while not_done:
            # next line 'sleeps' this main thread, letting the thread pool run
            freshly_done, not_done = concurrent.futures.wait(not_done, timeout=POLL_INTERVAL)
            done |= freshly_done
            # more polling stats calculated here and printed every POLL_INTERVAL seconds...
    except KeyboardInterrupt:
        # only futures that are not done will prevent exiting
        for future in not_done:
            # cancel() returns False if it's already done or currently running,
            # and True if was able to cancel it; we don't need that return value
            _ = future.cancel()
         # wait for running futures that the above for loop couldn't cancel (note timeout)
         _ = concurrent.futures.wait(not_done, timeout=None)

Si eres no interesado en realizar un seguimiento exacto de lo que se hizo y lo que no (es decir, no quiere un ciclo de progreso), puede reemplazar la primera llamada de espera (la que tiene timeout=0) con not_done = futures y todavía dejar el while not_done: lógica.

los for future in not_done: El ciclo de cancelación probablemente pueda comportarse de manera diferente en función de ese valor de retorno (o escribirse como una comprensión), pero esperar a que los futuros se hagan o se cancelen no es realmente una espera: regresa instantáneamente. El último wait con timeout=None asegura que los trabajos en ejecución de la piscina realmente terminen.

Nuevamente, esto solo funciona correctamente si el do_work que se está llamando en realidad, eventualmente regresa dentro de un período de tiempo razonable. Eso estuvo bien para mí; de hecho, quiero estar seguro de que si do_work comienza, se ejecuta hasta su finalización. Si do_work es ‘interminable’, entonces necesitará algo como la respuesta de cdosborn que usa una variable visible para todos los hilos, indicándoles que se detengan.

---

---

Utiliza Nuestro Buscador