How to stop multiprocessing.Pool with Ctrl+c? (Python 3.10) [Solved]

I've spent several hours on research and no solution seems to work anymore. My function will take an estimated 30-50 minutes per process. If I spot flaws during the first few minutes, I don't want to wait for the processes to finish completely or have to close the terminal without killing the processes cleanly.

For simplicity, each process is assigned a function with an infinite loop. With Ctrl+c the complete main program should now be interrupted without getting stuck. The reason for the infinite loop is that every previous solution has worked with lists that are actually looped through. The keyboard interrupt is recognized, but only afterwards, when all processes have finished. I don't know if this has anything to do with the Python version, but that's not the requirement I would like.

I would like a simple and clean solution to this problem. That would probably help many others as well.
Thank you for your help.

Example:

from multiprocessing import Pool

def f(x):
    while True:
        pass

if __name__ == '__main__':
    with Pool(2) as p:
        p.map(f, [1, 2])

Problem Solved

In another forum this problem was discussed and solved in a simple and clean way. Here is the link for everyone else:

https://www.reddit.com/r/learnpython/comments/152sfp8/how_to_stop_multiprocessingpool_with_ctrlc_python/

Solution 1

In multiprocessing module there is an AsyncResult object. This includes various functions such as get(), wait() and ready(). While get() and wait() block, you can use ready() to wait for the result and still catch KeyboardInterrupts.

Unfortunately, only map_async() and apply_async() return an AsyncResult object.

from multiprocessing import Pool

def f(x):
    while True:
        pass

if __name__ == "__main__":
    with Pool(2) as p:
        result = p.map_async(f, [1, 2])
        while not result.ready():
            time.sleep(1)

Solution 2

This way you can also use functions that block:
> Playing around with signals it seems that from inside child processes,
> exceptions other than KeyboardInterrupt or SystemExit propagate to the
> main process. When ctrl+c is pressed, all processes (children and main
> process) run their SIGINT handler which by default just raises
> KeyboardInterrupt. To get control out of the children, the child
> interrupt handler must raise some other exception. The exception is
> not seen by the way because the main process has its hands full with
> its own KeyboardInterrupt.

Basically, the SIGINT handler can also return None.

from multiprocessing import Pool
from signal import signal, SIGINT

def initializer():
    signal(SIGINT, lambda: None)

def f(x):
    while True:
        pass

if __name__ == '__main__':
    with Pool(n, initializer) as p:
        p.map(f, [1, 2])

For your scenario using apply_async is better than map so you can implement it like this:

from multiprocessing import Pool
import signal

def f(x):
    while True:
        pass

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

if __name__ == '__main__':
    with Pool(2, init_worker) as p:
        processes = 
]
        try:
            # Check for keyboard interrupts while the processes are running
            for process in processes:
                process.get(timeout=5)  # Adjust the timeout as needed
        except KeyboardInterrupt:
            print("Keyboard interrupt received. Stopping processes...")
            p.terminate()
            p.join()

You need the "driver" and sub-processing code to somehow collaborate. An Event can be useful for this.

Let's say that your sub-process is running in some kind of loop, then you just need that process to check an Event from time to time. If the event is set then it can terminate gracefully.

Something like this:

from multiprocessing import Pool, Manager
from signal import signal, SIGINT, SIG_IGN
from time import sleep

N = 10

def do_work(t):
    e, x = t
    while True: # infinite loop
        sleep(1) # do some work
        if e.is_set(): # check the Event
            break # process terminates
    return x

def ignore():
    signal(SIGINT, SIG_IGN)

def main():
    with Manager() as manager:
        e = manager.Event() # event is not set initially
        with Pool(N, ignore) as pool:
            results = pool.map_async(do_work, [(e, i) for i in range(N)])
            print('Blocking... Interrupt (Ctrl-C or Ctrl-Break) to continue')
            while True:
                try:
                    for r in results.get(): # block
                        print(r)
                    break
                except KeyboardInterrupt:
                    e.set()

if __name__ == '__main__':
    main()

Output:

Blocking... Interrupt (Ctrl-C or Ctrl-Break) to continue
^C0
1
2
3
4
5
6
7
8
9