fix: added startDependeeProcesses function which spawns dependee processes before. This currently blocks the supervisor runforever loop but does the job

Author: Michael Hammann

supervisor/rpcinterface.py

@@ -25,6 +25,8 @@
 from supervisor.events import notify
 from supervisor.events import RemoteCommunicationEvent
 
+from supervisor.medusa import asyncore_25 as asyncore
+
 from supervisor.http import NOT_DONE_YET
 from supervisor.xmlrpc import (
     capped_int,
@@ -283,28 +285,9 @@ 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:
-        #    # keep track of RUNNING childs
-        #    running_childs = set()
-        #    # wait/loop until all childs are running
-        #    while set(process.config.depends_on.keys()) != running_childs:
-        #        for child in process.config.depends_on.values():
-        #            if child.get_state() != ProcessStates.RUNNING:
-        #                # potentially remove child, if it is in running list
-        #                if child.config.name in running_childs:
-        #                    running_childs.remove(child.config.name)
-        #                # check if it needs to be started
-        #                if child.state is not (ProcessStates.STARTING or ProcessStates.RUNNING):
-        #                    self.startProcess(child.group.config.name + ':' + child.config.name)
-        #                else:
-        #                    child.transition()
-        #                    msg = ("waiting on dependee process {} to reach running state - currently in {}"
-        #                    .format(child.config.name, getProcessStateDescription(child.state)))
-        #                    self.supervisord.options.logger.warn(msg)
-        #            else:
-        #                # child is running - add to set
-        #                running_childs.add(child.config.name)
-        #        time.sleep(0.5)
+        if process.config.depends_on is not None:
+            self.startDependeeProcesses(process)
+
 
         self._update('startProcess')
         group, process = self._getGroupAndProcess(name)
@@ -330,8 +313,6 @@ def startProcess(self, name, wait=True):
 
         process.spawn(self.supervisord)
 
-        #TODO make sure this process is properly spawned in the end and correctly handled now.
-
         # We call reap() in order to more quickly obtain the side effects of
         # process.finish(), which reap() eventually ends up calling.  This
         # might be the case if the spawn() was successful but then the process
@@ -378,6 +359,84 @@ def onwait():
 
         return True
 
+    def startDependeeProcesses(self, process, name):
+        """ Start a Process but make sure that dependee processes are started
+        before and in RUNNING state. This is a blocking call and a lot of code
+        is copied from supervisor runforever loop. This needs to be improved 
+        but is currently required to use depends_on in combination with
+        runningregex and rpcinterfaces such as supervisorctl.
+        """
+        # keep track of RUNNING childs
+        running_childs = set()
+        # wait/loop until all childs are running
+        while set(process.config.depends_on.keys()) != running_childs:
+            for dependee in process.config.depends_on.values():
+                if dependee.get_state() != ProcessStates.RUNNING:
+                    # potentially remove child, if it is in running list
+                    if dependee.config.name in running_childs:
+                        running_childs.remove(dependee.config.name)
+                    # check if it needs to be started
+                    if dependee.state is not (ProcessStates.STARTING or ProcessStates.RUNNING):
+                        self.startProcess(dependee.group.config.name + ':' + dependee.config.name)
+                    else:
+                        timeout = 1
+                        socket_map = self.supervisord.options.get_socket_map()
+                        combined_map = {}
+                        combined_map.update(socket_map)
+                        combined_map.update(self.supervisord.get_process_map())
+
+                        for fd, dispatcher in combined_map.items():
+                            if dispatcher.readable():
+                                self.supervisord.options.poller.register_readable(fd)
+                            if dispatcher.writable():
+                                self.supervisord.options.poller.register_writable(fd)
+
+                        r, w = self.supervisord.options.poller.poll(timeout)
+
+                        for fd in r:
+                            if fd in combined_map:
+                                try:
+                                    dispatcher = combined_map[fd]
+                                    self.supervisord.options.logger.blather(
+                                        'read event caused by %(dispatcher)r',
+                                        dispatcher=dispatcher)
+                                    dispatcher.handle_read_event()
+                                    if not dispatcher.readable():
+                                        self.supervisord.options.poller.unregister_readable(fd)
+                                except asyncore.ExitNow:
+                                    raise
+                                except:
+                                    combined_map[fd].handle_error()
+
+                        for fd in w:
+                            if fd in combined_map:
+                                try:
+                                    dispatcher = combined_map[fd]
+                                    self.supervisord.options.logger.blather(
+                                        'write event caused by %(dispatcher)r',
+                                        dispatcher=dispatcher)
+                                    dispatcher.handle_write_event()
+                                    if not dispatcher.writable():
+                                        self.supervisord.options.poller.unregister_writable(fd)
+                                except asyncore.ExitNow:
+                                    raise
+                                except:
+                                    combined_map[fd].handle_error()
+
+                        self.supervisord.reap()
+                        if dependee.spawnerr:
+                            dependee.state = ProcessStates.FATAL
+                            raise RPCError(Faults.SPAWN_ERROR, name)
+                        dependee.transition()
+
+                        msg = ("waiting on dependee process {} to reach running state - currently in {}"
+                        .format(dependee.config.name, getProcessStateDescription(dependee.state)))
+                        self.supervisord.options.logger.info(msg)
+                else:
+                    # child is running - add to set
+                    running_childs.add(dependee.config.name)
+            time.sleep(0.5)
+
     def startProcessGroup(self, name, wait=True):
         """ Start all processes in the group named 'name'