esgmapfile clean shutdown on SIGTERM

[alaniwi]

Is it possible to make esgmapfile kill the processes from the pool if the main process receives a SIGTERM? Currently only the main process dies, leaving the workers running. They will die with "broken pipe" when they try to send some output, but if they are checksumming large files then it may be some time until they do this.

[glevava]

It seems that multiprocessing.pool.terminate() does the job : https://docs.python.org/3/library/multiprocessing.html#multiprocessing.pool.Pool.terminate
Will see how to implement it.

[alaniwi]

Okay, then I think you will need to install a signal handler for SIGTERM, in which you can execute that call. I will see if I can produce a short example. I would need to experiment a little because the one complication I can anticipate is that if you install a handler in the master process before forking subprocesses, then maybe the child processes inherit the signal handler, which we would not want.

[alaniwi]

Okay, I have an example. It is a little bit longer than I hoped.

The other problem seems to be that if you install a handler, then python will not execute the handler until the python statement currently running has completed, so if you are executing a pool.join() statement which is blocking until the workers finish, then it will wait until the end. So instead my example code has a loop, so that while it is waiting it is repeatedly doing the time.sleep(1), and if a signal arrives then it will complete the current call and then run the handler. By the time that pool.join() is called, it will not block because the workers have already finished.

import time
import signal
import os
from multiprocessing import Pool


def double_it(x):
    time.sleep(5)
    return x * 2


class Runner:

    def __init__(self):
        self.pool = None

    def _handle_sigterm(self, signum, frame):
        if self.pool:
            self.pool.terminate()
        os._exit(1)
            
    def run(self):
        self.pool = Pool(4)
        result = self.pool.map_async(double_it, range(10))
        old_handler = signal.signal(signal.SIGTERM, self._handle_sigterm)
        while not result.ready():
            time.sleep(1)
        signal.signal(signal.SIGTERM, old_handler)
        self.pool.close()
        self.pool.join()
        return result.get()

print(os.getpid())
m = Runner()
print(m.run())