from abc import ABC, abstractmethod from enum import Enum, Flag, auto def _sigchld_handler(signum, process): if not signum == signal.SIGCHLD: return Process.propagate_signal(signum) class Process(ABC): # export __processes = [] class State(Enum): Running = auto() Shutdown = auto() Done = auto() class Flags(Flag): FailOnExitWithoutShutdown = auto() def __init__(self): self.__state = Running self.__flags = Flags.FailOnExitWithoutShutdown if len(self.__processes) == 0: self._signals().add_handler(signals.SIGCHLD, _sigchld_handler) self.__processes.add(self) @classmethod def propagate_signal(cls, signum): for p in cls.__processes: p.__signal(signum) def signal(self, signum): if signum == signals.SIGCHLD: self.exited() @abstractmethod def _pid(self): pass @classmethod @abstractmethod def signals(cls): pass # to be reimplemented def _request_shutdown(self): pass # to be reimplemented def name(self): return str(self._pid()) def request_shutdown(self): if not self.__state == Shutdown: self.__state = Shutdown self._request_shutdown() def exited(self): if self.__state == Process.State.Running: slog(ERR, 'process "{}" exited unexpectedly'.format(process.name())) if __flags & Process.Flags.FailOnExitWithoutShutdown: slog(ERR, 'exiting') exit(1) self.__state = Process.State.Done self.__processes.erase(self) if len(self.__processes) == 0: self._signals().remove_handler(signals.SIGCHLD) # FIXME: broken logic