fix: restructured queue, added timeout parameter, depends_on works now with rpcinterfaces and autostart

Author: Michael Hammann

supervisor/options.py

@@ -940,6 +940,7 @@ def get(section, opt, *args, **kwargs):
                 raise ValueError(
                     f"program section {section} has invalid runningregex value. Error {e}")
         depends_on = get(section, 'depends_on', None)
+        spawn_timeout = int(get(section, 'spawn_timeout', 60))
 
         # find uid from "user" option
         user = get(section, 'user', None)
@@ -1068,6 +1069,7 @@ def get(section, opt, *args, **kwargs):
                 serverurl=serverurl,
                 runningregex=runningregex,
                 depends_on=depends_on,
+                spawn_timeout=spawn_timeout,
             )
 
             programs.append(pconfig)
@@ -1886,7 +1888,8 @@ class ProcessConfig(Config):
         'stderr_events_enabled', 'stderr_syslog',
         'stopsignal', 'stopwaitsecs', 'stopasgroup', 'killasgroup',
         'exitcodes', 'redirect_stderr' ]
-    optional_param_names = [ 'environment', 'serverurl', 'runningregex', 'depends_on' ]
+    optional_param_names = [ 'environment', 'serverurl', 
+        'runningregex', 'depends_on', 'spawn_timeout' ]
 
     def __init__(self, options, **params):
         self.options = options

supervisor/process.py

@@ -192,6 +192,22 @@ def record_spawnerr(self, msg):
         self.spawnerr = msg
         self.config.options.logger.info("spawnerr: %s" % msg)
 
+    def queue_all_dependee_processes(self, supervisor):
+        if self.config.name not in supervisor.process_spawn_set:
+            supervisor.process_spawn_queue.append(self)
+            supervisor.process_spawn_set.add(self.config.name)
+        # all dependees that are not in queue and not in STARTING need to be added to queue.
+        if self.config.depends_on is not None:
+            for dependee in self.config.depends_on.values():
+                if dependee.state is not ProcessStates.RUNNING:
+                    ready_to_be_spawned = False
+                if dependee.state is not ProcessStates.STARTING:
+                    if dependee.config.name not in supervisor.process_spawn_set:
+                        supervisor.process_spawn_queue.append(dependee)
+                        supervisor.process_spawn_set.add(dependee.config.name)
+                        dependee.queue_all_dependee_processes(supervisor)
+
+
     def spawn(self, supervisor=None):
         """Start the subprocess.  It must not be running already.
 
@@ -202,14 +218,15 @@ def spawn(self, supervisor=None):
         # check if the process is dependent upon any other process and if so,
         # make sure that one is in the RUNNING state
         # spawn if all dependees are running - else add to queue if not in queue already
-        if self.config.depends_on is not None and any([dependee.state is not ProcessStates.RUNNING for dependee in self.config.depends_on.values()]):
-            # all dependees that are not in queue and not in STARTING need to be added to queue.
-            for dependee in self.config.depends_on.values():
-                if dependee.state is not (ProcessStates.STARTING or ProcessStates.RUNNING):
-                    if dependee.config.name not in supervisor.waiting_to_be_spawned:
-                        supervisor.process_queue.append(dependee)
-                        supervisor.waiting_to_be_spawned.add(dependee.config.name)
-        else:
+        ready_to_be_spawned = True
+        if self.config.depends_on is not None:
+            # if all dependees are running - no problem process can be spawned
+            if any([dependee.state is not ProcessStates.RUNNING for dependee in 
+                    self.config.depends_on.values()]):
+                ready_to_be_spawned = False
+                # all dependees not in queue and not in STARTING need to be added to queue.
+                self.queue_all_dependee_processes(supervisor)
+        if ready_to_be_spawned:
             # if runningregex is used, set the log_offset correctly
             if self.config.runningregex is not None:
                 try:

supervisor/rpcinterface.py

@@ -285,9 +285,6 @@ def startProcess(self, name, wait=True):
         """
         ## check if the process is dependent upon any other process and if so make sure that one is in the RUNNING state
         group, process = self._getGroupAndProcess(name)
-        if process.config.depends_on is not None:
-            self.startDependeeProcesses(process)
-
 
         self._update('startProcess')
         group, process = self._getGroupAndProcess(name)
@@ -312,6 +309,10 @@ def startProcess(self, name, wait=True):
                            "%s is in an unknown process state" % name)
 
         process.spawn(self.supervisord)
+        # if process has dependees, return succesfull start - 
+        # errors will be handled in main loop and inside process.spawn()
+        if process.config.depends_on is not None:
+            return True
 
         # We call reap() in order to more quickly obtain the side effects of
         # process.finish(), which reap() eventually ends up calling.  This

supervisor/supervisord.py

@@ -59,8 +59,8 @@ def __init__(self, options):
         self.options = options
         self.process_groups = {}
         self.ticks = {}
-        self.process_queue = deque()
-        self.waiting_to_be_spawned = set()
+        self.process_spawn_queue = deque()
+        self.process_spawn_set = set()
 
     def main(self):
         if not self.options.first:
@@ -270,33 +270,7 @@ def runforever(self):
             for group in pgroups:
                 group.transition(self)
 
-            # check for processes that are sheduled but not started
-            if self.process_queue:
-                # spawn if all dependencies are running or process has no dependencies
-                if self.process_queue[-1].config.depends_on is not None:
-                    if all([dependee.get_state() == ProcessStates.RUNNING for dependee in self.process_queue[-1].config.depends_on.values()]):
-                        ready_process = self.process_queue.pop()
-                        ready_process.spawn(self)
-                        try:
-                            self.waiting_to_be_spawned.remove(ready_process.config.name)
-                        except KeyError:
-                            pass
-                    # add any dependee which is not RUNNING or STARTING to queue
-                    else:
-                        for dependee in self.process_queue[-1].config.depends_on.values():
-                            if dependee.state is not (ProcessStates.STARTING or ProcessStates.RUNNING):
-                                # Should we check if the process is already in the queue?
-                                if dependee.config.name not in self.waiting_to_be_spawned:
-                                    self.process_queue.append(dependee)
-                                    self.waiting_to_be_spawned.add(dependee.config.name)
-
-                else:
-                    ready_process = self.process_queue.pop()
-                    ready_process.spawn(self)
-                    try:
-                        self.waiting_to_be_spawned.remove(ready_process.config.name)
-                    except KeyError:
-                        pass
+            self.spawn_dependee_queue()
 
             self.reap()
             self.handle_signal()
@@ -373,6 +347,76 @@ def handle_signal(self):
     def get_state(self):
         return self.options.mood
 
+    def spawn_dependee_queue(self):
+        """
+        check for processes that are sheduled but not started
+        if the queue is well ordered just pop and spawn all that are ready, 
+        Currently ordering is not considered when adding to queue - 
+        therefore just iterate over queue and spawn all ready processes.
+        """
+        if self.process_spawn_queue:
+            for i in range(len(self.process_spawn_queue)-1, -1, -1):
+                if self.process_spawn_queue[i].config.depends_on is not None:
+                    if any([dependee.state is ProcessStates.FATAL for dependee in
+                            self.process_spawn_queue[i].config.depends_on.values()]):
+                        for process in self.process_spawn_queue:
+                            process.record_spawnerr(
+                                'Dependee process did not start - set FATAL state for {}'
+                                .format(process.config.name))
+                            process.change_state(ProcessStates.FATAL)
+                        self.process_spawn_set = set()
+                        self.process_spawn_queue.clear()
+                        break
+                    if all([dependee.state is ProcessStates.RUNNING for dependee in
+                            self.process_spawn_queue[i].config.depends_on.values()]):
+                        self.spawn_me = self.process_spawn_queue[i]
+                        del self.process_spawn_queue[i]
+                        self.spawn_me.spawn(self)
+                        self.last_spawned = time.time()
+                        self.last_print = time.time()
+                else:
+                    self.spawn_me = self.process_spawn_queue[i]
+                    del self.process_spawn_queue[i]
+                    self.spawn_me.spawn(self)
+                    self.last_spawned = time.time()
+                    self.last_print = time.time()
+
+            try:
+                if (time.time() - self.last_spawned) >= self.spawn_me.config.spawn_timeout:
+                    msg = (
+                        "timeout: dependee process {} in {} state for over {} seconds, {} are not spawned"
+                            .format(self.spawn_me.config.name,
+                                    getProcessStateDescription(self.spawn_me.state),
+                                    self.spawn_me.config.spawn_timeout,
+                                    [process.config.name for process in self.process_spawn_queue]))
+                    self.process_spawn_queue[i].config.options.logger.warn(msg)
+                    self.spawn_me.record_spawnerr(
+                        'Process {} stayed for over {} seconds in STARTING'.format(
+                        self.spawn_me.config.name, self.spawn_me.config.spawn_timeout))
+                    self.spawn_me.change_state(ProcessStates.FATAL)
+                    for process in self.process_spawn_queue:
+                        process.record_spawnerr(
+                            'Dependee process {} did not start - set FATAL state for {}'.format(
+                            self.spawn_me.config.name, process.config.name))
+                        process.change_state(ProcessStates.FATAL)
+                    self.process_spawn_set = set()
+                    self.process_spawn_queue.clear()
+                    self.last_spawned = time.time()
+                elif (time.time() - self.last_print) >= 5:
+                    msg = ("waiting for dependee process {} in {} state to be RUNNING"
+                            .format(self.spawn_me.config.name,
+                                    getProcessStateDescription(self.spawn_me.state)))
+                    self.process_spawn_queue[0].config.options.logger.info(msg)
+                    self.last_print = time.time()
+            except UnboundLocalError:
+                self.last_spawned = time.time()
+                self.last_print = time.time()
+                
+
+        # empty set - all processes from queue are running
+        else:
+                self.process_spawn_set = set()
+
 def timeslice(period, when):
     return int(when - (when % period))