import multiprocessing import time class Base(multiprocessing.Process): def __init__(self): super().__init__() self.stay_alive = multiprocessing.Event() def shutdown(self): self.stay_alive.clear() self.join() @property def state(self): if self.is_alive(): return f"{type(self).__name__} - run() running" elif not self.is_alive() and self.exitcode == 0: return f"{type(self).__name__} - run() completed successfully" elif not self.is_alive() and self.exitcode == 1: return f"{type(self).__name__} - run() completed unsuccessfully" elif not self.is_alive() and self.exitcode is None: return f"{type(self).__name__} - run() has not started" else: return f"{type(self).__name__} - idk" # def is_alive(self) -> bool: # # self._check_closed() # if self is multiprocessing.process._current_process: # return True # if self._popen is None: # return False # # returncode = self._popen.poll() # if returncode is None: # return True # else: # return False class A(Base): def run(self): self.stay_alive.set() while self.stay_alive.is_set(): time.sleep(0.1) print('A.run() closed') class B(A): def run(self): self.stay_alive.set() raise Exception class C(Base): def __init__(self, brother): super().__init__() self.brother = brother def run(self): print(f"brother - {self.brother.state}") # Show the states of a successful process a = A() print(a.state) a.start() print(a.state) time.sleep(1) c = C(a) # View state of A from sibling process c.start() time.sleep(2) a.shutdown() print(a.state) # b = B() c = C(b) # View state of A from sibling process c.start() print(b.state) c = C(b) b.start() # which will error time.sleep(1) c.start() # View state of A from sibling process print(b.state) print(1)