Merge pull request #3302 from radical-cybertools/fix/process_cleanup

ensure scheduler sub-process terminates

src/radical/pilot/agent/scheduler/base.py

@@ -2,6 +2,7 @@
 __copyright__ = 'Copyright 2013-2022, The RADICAL-Cybertools Team'
 __license__   = 'MIT'
 
+import os
 import copy
 import time
 import queue
@@ -272,6 +273,10 @@ def initialize(self):
         self._p.daemon = True
         self._p.start()
 
+        self._pwatcher = ru.PWatcher(uid='%s.pw' % self._uid, log=self._log)
+        self._pwatcher.watch(os.getpid())
+        self._pwatcher.watch(self._p.pid)
+
 
     # --------------------------------------------------------------------------
     #
@@ -619,6 +624,10 @@ def _schedule_tasks(self):
 
         self._scheduler_process = True
 
+        self._pwatcher = ru.PWatcher(uid='%s.pw' % self._uid, log=self._log)
+        self._pwatcher.watch(os.getpid())
+        self._pwatcher.watch(os.getppid())
+
         # ZMQ endpoints will not have survived the fork. Specifically the
         # registry client of the component base class will have to reconnect.
         # Note that `self._reg` of the base class is a *pointer* to the sesison

tests/unit_tests/test_scheduler/test_base.py

@@ -58,6 +58,7 @@ def _mock_schedule_tasks(*args, **kwargs):
         sched.nodes               = []
         sched._partitions         = {}
         sched._scheduler_process  = False
+        sched._uid                = 'uid'
 
         sched._session = mock.Mock()