add ipcluster engines; fix ssh process shutdown

IPython/zmq/parallel/ipclusterapp.py

@@ -63,6 +63,10 b' ALREADY_STARTED = 10'
 # file to be found.
 ALREADY_STOPPED = 11
 
+# This will be the exit code if ipcluster engines is run, but there is not .pid
+# file to be found.
+NO_CLUSTER = 12
+
 
 #-----------------------------------------------------------------------------
 # Command line options
@@ -164,6 +168,7 b' class IPClusterAppConfigLoader(ClusterDirConfigLoader):'
             otherwise use the '--cluster-dir' option.
             """
         )
+        
         paa = parser_start.add_argument
         paa('-n', '--number',
             type=int, dest='Global.n',
@@ -206,6 +211,35 b' class IPClusterAppConfigLoader(ClusterDirConfigLoader):'
             help="The signal number to use in stopping the cluster (default=2).",
             metavar="Global.signal")
 
+        # the "engines" subcommand parser
+        parser_engines = subparsers.add_parser(
+            'engines',
+            parents=[parent_parser1, parent_parser2],
+            argument_default=SUPPRESS,
+            help="Attach some engines to an existing controller or cluster.",
+            description=
+            """Start one or more engines to connect to an existing Cluster
+            by profile name or cluster directory.
+            Cluster directories contain configuration, log and
+            security related files and are named using the convention
+            'cluster_<profile>' and should be creating using the 'start'
+            subcommand of 'ipcluster'. If your cluster directory is in 
+            the cwd or the ipython directory, you can simply refer to it
+            using its profile name, 'ipclusterz engines -n 4 -p <profile>`,
+            otherwise use the '--cluster-dir' option.
+            """
+        )
+        paa = parser_engines.add_argument
+        paa('-n', '--number',
+            type=int, dest='Global.n',
+            help='The number of engines to start.',
+            metavar='Global.n')
+        paa('--daemon',
+            dest='Global.daemonize', action='store_true',
+            help='Daemonize the ipcluster program. This implies --log-to-file')
+        paa('--no-daemon',
+            dest='Global.daemonize', action='store_false',
+            help="Dont't daemonize the ipcluster program.")
 
 #-----------------------------------------------------------------------------
 # Main application
@@ -232,7 +266,7 b' class IPClusterApp(ApplicationWithClusterDir):'
         self.default_config.Global.delay = 1
         self.default_config.Global.reset_config = False
         self.default_config.Global.clean_logs = True
-        self.default_config.Global.signal = 2
+        self.default_config.Global.signal = signal.SIGINT
         self.default_config.Global.daemonize = False
 
     def find_resources(self):
@@ -255,6 +289,17 b' class IPClusterApp(ApplicationWithClusterDir):'
                     "'ipclusterz create -h' or 'ipclusterz list -h' for more "
                     "information about creating and listing cluster dirs."
                 )
+        elif subcommand=='engines':
+            self.auto_create_cluster_dir = False
+            try:
+                super(IPClusterApp, self).find_resources()
+            except ClusterDirError:
+                raise ClusterDirError(
+                    "Could not find a cluster directory. A cluster dir must "
+                    "be created before running 'ipclusterz start'.  Do "
+                    "'ipclusterz create -h' or 'ipclusterz list -h' for more "
+                    "information about creating and listing cluster dirs."
+                )
 
     def list_cluster_dirs(self):
         # Find the search paths
@@ -309,31 +354,30 b' class IPClusterApp(ApplicationWithClusterDir):'
             # reactor.callWhenRunning(self.start_launchers)
             dc = ioloop.DelayedCallback(self.start_launchers, 0, self.loop)
             dc.start()
+        if subcmd == 'engines':
+            self.start_logging()
+            self.loop = ioloop.IOLoop.instance()
+            # reactor.callWhenRunning(self.start_launchers)
+            engine_only = lambda : self.start_launchers(controller=False)
+            dc = ioloop.DelayedCallback(engine_only, 0, self.loop)
+            dc.start()
 
-    def start_launchers(self):
+    def start_launchers(self, controller=True):
         config = self.master_config
 
         # Create the launchers. In both bases, we set the work_dir of
         # the launcher to the cluster_dir. This is where the launcher's
         # subprocesses will be launched. It is not where the controller
         # and engine will be launched.
-        el_class = import_item(config.Global.engine_launcher)
-        self.engine_launcher = el_class(
-            work_dir=self.cluster_dir, config=config, logname=self.log.name
-        )
+        if controller:
         cl_class = import_item(config.Global.controller_launcher)
         self.controller_launcher = cl_class(
             work_dir=self.cluster_dir, config=config,
             logname=self.log.name
         )
-
-        # Setup signals
-        signal.signal(signal.SIGINT, self.sigint_handler)
-
         # Setup the observing of stopping. If the controller dies, shut
         # everything down as that will be completely fatal for the engines.
         self.controller_launcher.on_stop(self.stop_launchers)
-        # d1.addCallback(self.stop_launchers)
         # But, we don't monitor the stopping of engines. An engine dying
         # is just fine and in principle a user could start a new engine.
         # Also, if we did monitor engine stopping, it is difficult to
@@ -342,17 +386,22 b' class IPClusterApp(ApplicationWithClusterDir):'
         # might trigger on a single engine stopping, other wait until
         # all stop.  TODO: think more about how to handle this.
         
+        
+        el_class = import_item(config.Global.engine_launcher)
+        self.engine_launcher = el_class(
+            work_dir=self.cluster_dir, config=config, logname=self.log.name
+        )
+
+        # Setup signals
+        signal.signal(signal.SIGINT, self.sigint_handler)
+
         # Start the controller and engines
         self._stopping = False  # Make sure stop_launchers is not called 2x.
-        d = self.start_controller()
-        dc = ioloop.DelayedCallback(self.start_engines, 1000*config.Global.delay, self.loop)
+        if controller:
+            self.start_controller()
+        dc = ioloop.DelayedCallback(self.start_engines, 1000*config.Global.delay*controller, self.loop)
         dc.start()
         self.startup_message()
-        # d.addCallback(self.start_engines)
-        # d.addCallback(self.startup_message)
-        # If the controller or engines fail to start, stop everything
-        # d.addErrback(self.stop_launchers)
-        return d
 
     def startup_message(self, r=None):
         self.log.info("IPython cluster: started")
@@ -369,6 +418,7 b' class IPClusterApp(ApplicationWithClusterDir):'
     def start_engines(self, r=None):
         # self.log.info("In start_engines")
         config = self.master_config
+        
         d = self.engine_launcher.start(
             config.Global.n,
             cluster_dir=config.Global.cluster_dir
@@ -432,6 +482,8 b' class IPClusterApp(ApplicationWithClusterDir):'
             self.start_app_start()
         elif subcmd=='stop':
             self.start_app_stop()
+        elif subcmd=='engines':
+            self.start_app_engines()
 
     def start_app_start(self):
         """Start the app for the start subcommand."""
@@ -468,6 +520,29 b' class IPClusterApp(ApplicationWithClusterDir):'
             self.log.info("stopping...")
         self.remove_pid_file()
 
+    def start_app_engines(self):
+        """Start the app for the start subcommand."""
+        config = self.master_config
+        # First see if the cluster is already running
+        
+        # Now log and daemonize
+        self.log.info(
+            'Starting engines with [daemon=%r]' % config.Global.daemonize
+        )
+        # TODO: Get daemonize working on Windows or as a Windows Server.
+        if config.Global.daemonize:
+            if os.name=='posix':
+                from twisted.scripts._twistd_unix import daemonize
+                daemonize()
+
+        # Now write the new pid file AFTER our new forked pid is active.
+        # self.write_pid_file()
+        try:
+            self.loop.start()
+        except:
+            self.log.fatal("stopping...")
+        # self.remove_pid_file()
+    
     def start_app_stop(self):
         """Start the app for the stop subcommand."""
         config = self.master_config

IPython/zmq/parallel/launcher.py

@@ -28,12 +28,12 b' except ImportError:'
 
 from subprocess import Popen, PIPE, STDOUT
 try:
-    from subprocess import check_open
+    from subprocess import check_output
 except ImportError:
     # pre-2.7:
     from StringIO import StringIO
     
-    def check_open(*args, **kwargs):
+    def check_output(*args, **kwargs):
         sio = StringIO()
         kwargs.update(dict(stdout=PIPE, stderr=STDOUT))
         p = Popen(*args, **kwargs)
@@ -495,7 +495,7 b' class SSHLauncher(LocalProcessLauncher):'
     """
 
     ssh_cmd = List(['ssh'], config=True)
-    ssh_args = List([], config=True)
+    ssh_args = List(['-tt'], config=True)
     program = List(['date'], config=True)
     program_args = List([], config=True)
     hostname = Str('', config=True)
@@ -513,12 +513,21 b' class SSHLauncher(LocalProcessLauncher):'
                self.program + self.program_args
 
     def start(self, cluster_dir, hostname=None, user=None):
+        print self.config
         if hostname is not None:
             self.hostname = hostname
         if user is not None:
             self.user = user
+        print (self.location, hostname, user)
         return super(SSHLauncher, self).start()
 
+    def signal(self, sig):
+        if self.state == 'running':
+            # send escaped ssh connection-closer
+            self.process.stdin.write('~.')
+            self.process.stdin.flush()
+        
+
 
 class SSHControllerLauncher(SSHLauncher):
 
@@ -568,9 +577,9 b' class WindowsHPCLauncher(BaseLauncher):'
     scheduler = Str('', config=True)
     job_cmd = Str(find_job_cmd(), config=True)
 
-    def __init__(self, work_dir=u'.', config=None):
+    def __init__(self, work_dir=u'.', config=None, **kwargs):
         super(WindowsHPCLauncher, self).__init__(
-            work_dir=work_dir, config=config
+            work_dir=work_dir, config=config, **kwargs
         )
 
     @property
@@ -730,9 +739,9 b' class BatchSystemLauncher(BaseLauncher):'
     # The full path to the instantiated batch script.
     batch_file = Unicode(u'')
 
-    def __init__(self, work_dir=u'.', config=None):
+    def __init__(self, work_dir=u'.', config=None, **kwargs):
         super(BatchSystemLauncher, self).__init__(
-            work_dir=work_dir, config=config
+            work_dir=work_dir, config=config, **kwargs
         )
         self.batch_file = os.path.join(self.work_dir, self.batch_file_name)
         self.context = {}
@@ -766,7 +775,7 b' class BatchSystemLauncher(BaseLauncher):'
         return job_id
 
     def stop(self):
-        output = Popen([self.delete_command, self.job_id], env=os.environ, stderr=STDOUT)
+        output = check_output([self.delete_command, self.job_id], env=os.environ, stderr=STDOUT)
         self.notify_stop(output)  # Pass the output of the kill cmd
         return output