Fire and forget processes

circus/commands/__init__.py

@@ -16,6 +16,7 @@
     reload,
     restart,
     rmwatcher,
+    run,
     sendsignal,
     set,
     start,

circus/commands/run.py

@@ -0,0 +1,53 @@
+from circus.commands.base import Command
+from circus.exc import ArgumentError
+
+
+class Run(Command):
+    """\
+        Run a watcher
+        ==============================
+
+        This command runs the process in a watcher
+
+
+        ZMQ Message
+        -----------
+
+        ::
+
+            {
+                "command": "run",
+                "properties": {
+                    "name": '<name>",
+                }
+            }
+
+        The response return the status "ok".
+
+        If the property name is present, the watcher will be run.
+
+        Command line
+        ------------
+
+        ::
+
+            $ circusctl run <name>
+
+        Options
+        +++++++
+
+        - <name>: name of the watcher
+
+    """
+    name = "run"
+
+    def message(self, *args, **opts):
+        if len(args) != 1:
+            raise ArgumentError("invalid number of arguments")
+
+        return self.make_message(name=args[0])
+
+    def execute(self, arbiter, props):
+        if 'name' in props:
+            watcher = self._get_watcher(arbiter, props['name'])
+            watcher.run()

circus/watcher.py

@@ -121,6 +121,7 @@ def __init__(self, name, cmd, args=None, numprocesses=1, warmup_delay=0.,
         self.args = args
         self._process_counter = 0
         self.stopped = stopped
+        self.running = False
         self.graceful_timeout = graceful_timeout
         self.prereload_fn = prereload_fn
         self.executable = None
@@ -267,35 +268,45 @@ def manage_processes(self):
         if self.stopped:
             return
 
-        if len(self.processes) < self.numprocesses:
-            self.spawn_processes()
-
-        processes = self.processes.values()
-        processes.sort()
-        while len(processes) > self.numprocesses:
-            process = processes.pop(0)
-            if process.status == STATUS_DEAD:
-                self.processes.pop(process.pid)
-            else:
-                self.processes.pop(process.pid)
-                self.kill_process(process)
-
+        self.shutdown_excess_processes()
+        self.spawn_needed_processes()
+
+    @util.debuglog
+    def shutdown_excess_processes(self):
+      """ If there are more running processes than numprocesses, kill the excess
+      """
+      processes = self.processes.values()
+      processes.sort()
+      while len(processes) > self.numprocesses:
+          process = processes.pop(0)
+          if process.status == STATUS_DEAD:
+              self.processes.pop(process.pid)
+          else:
+              self.processes.pop(process.pid)
+              self.kill_process(process)
+
     @util.debuglog
     def reap_and_manage_processes(self):
         """Reap & manage processes.
         """
         if self.stopped:
             return
+
         self.reap_processes()
         self.manage_processes()
 
     @util.debuglog
-    def spawn_processes(self):
+    def spawn_needed_processes(self):
         """Spawn processes.
         """
-        for i in range(self.numprocesses - len(self.processes)):
-            self.spawn_process()
-            time.sleep(self.warmup_delay)
+
+        if self.running:
+          return
+
+        if len(self.processes) < self.numprocesses:
+          for i in range(self.numprocesses - len(self.processes)):
+              self.spawn_process()
+              time.sleep(self.warmup_delay)
 
     def spawn_process(self):
         """Spawn process.
@@ -419,7 +430,10 @@ def send_signal_children(self, pid, signum):
     def status(self):
         if self.stopped:
             return "stopped"
-        return "active"
+        elif self.running:
+            return "running"
+        else:
+            return "active"
 
     @util.debuglog
     def process_info(self, pid):
@@ -472,10 +486,11 @@ def get_active_pids(self):
     def start(self):
         """Start.
         """
-        if not self.stopped:
+        if not self.stopped and not self.running:
             return
 
         self.stopped = False
+        self.running = False
         self._create_redirectors()
         self.reap_processes()
         self.manage_processes()
@@ -489,6 +504,29 @@ def start(self):
         logger.info('%s started' % self.name)
         self.notify_event("start", {"time": time.time()})
 
+    @util.debuglog
+    def run(self):
+        """Run.
+        """
+        if not self.stopped and not self.running:
+            return
+
+        self.stopped = False
+        self.running = False
+        self._create_redirectors()
+        self.reap_processes()
+        self.manage_processes()
+        self.running = True
+
+        if self.stdout_redirector is not None:
+            self.stdout_redirector.start()
+
+        if self.stderr_redirector is not None:
+            self.stderr_redirector.start()
+
+        logger.info('%s run' % self.name)
+        self.notify_event("run", {"time": time.time()})
+
     @util.debuglog
     def restart(self):
         """Restart.