Notebook cluster manager now uses proper launchers.

IPython/config/loader.py

@@ -676,3 +676,27 b' class KVArgParseConfigLoader(ArgParseConfigLoader):'
             sub_parser.load_config(self.extra_args)
             self.config._merge(sub_parser.config)
             self.extra_args = sub_parser.extra_args
+
+
+def load_pyconfig_files(config_files, path):
+    """Load multiple Python config files, merging each of them in turn.
+
+    Parameters
+    ==========
+    config_files : list of str
+        List of config files names to load and merge into the config.
+    path : unicode
+        The full path to the location of the config files.
+    """
+    config = Config()
+    for cf in config_files:
+        loader = PyFileConfigLoader(cf, path=path)
+        try:
+            next_config = loader.load_config()
+        except ConfigFileNotFound:
+            pass
+        except:
+            raise
+        else:
+            config._merge(next_config)
+    return config

IPython/frontend/html/notebook/clustermanager.py

@@ -16,19 +16,20 b' Authors:'
 # Imports
 #-----------------------------------------------------------------------------
 
-import datetime
 import os
-import uuid
-import glob
 
 from tornado import web
 from zmq.eventloop import ioloop
 
 from IPython.config.configurable import LoggingConfigurable
-from IPython.utils.traitlets import Unicode, List, Dict, Bool
-from IPython.parallel.apps.launcher import IPClusterLauncher
-from IPython.core.profileapp import list_profiles_in, list_bundled_profiles
-from IPython.utils.path import get_ipython_dir, get_ipython_package_dir
+from IPython.config.loader import load_pyconfig_files
+from IPython.utils.traitlets import Dict, Instance, CFloat
+from IPython.parallel.apps.ipclusterapp import find_launcher_class
+from IPython.core.profileapp import list_profiles_in
+from IPython.core.profiledir import ProfileDir
+from IPython.utils.path import get_ipython_dir
+from IPython.utils.sysinfo import num_cpus
+
 
 #-----------------------------------------------------------------------------
 # Classes
@@ -38,55 +39,130 b' class ClusterManager(LoggingConfigurable):'
 
     profiles = Dict()
 
-
-    def list_profile_names(self):
+    delay = CFloat(1., config=True,
+        help="delay (in s) between starting the controller and the engines")
+
+    loop = Instance('zmq.eventloop.ioloop.IOLoop')
+    def _loop_default(self):
+        from zmq.eventloop.ioloop import IOLoop
+        return IOLoop.instance()
+
+    def load_cluster_config(self, profile_dir):
+        config_files = ['ipcontroller_config.py', 'ipengine_config.py', 'ipcluster_config.py']
+        config = load_pyconfig_files(config_files, profile_dir)
+        return config
+
+    def build_launchers(self, profile_dir):
+        config = self.load_cluster_config(profile_dir)
+        cont_clsname = config.IPClusterStart.get('controller_launcher_class','Local')
+        cont_class = find_launcher_class(cont_clsname,'Controller')
+        cl = cont_class(work_dir=u'.',config=config, profile_dir=profile_dir)
+
+        engine_clsname = config.IPClusterEngines.get('engine_launcher_class','Local')
+        engine_class = find_launcher_class(engine_clsname,'EngineSet')
+        esl = engine_class(work_dir=u'.',config=config, profile_dir=profile_dir)
+        n = config.IPClusterEngines.get('n', num_cpus())
+        n = getattr(esl, 'engine_count', n)
+        return cl, esl, n
+
+    def get_profile_dir(self, name, path):
+        p = ProfileDir.find_profile_dir_by_name(path,name=name)
+        return p.location
+
+    def update_profiles(self):
         """List all profiles in the ipython_dir and cwd.
         """
-        profiles = list_profiles_in(get_ipython_dir())
-        profiles += list_profiles_in(os.getcwdu())
-        return profiles
-
+        for path in [get_ipython_dir(), os.getcwdu()]:
+            for profile in list_profiles_in(path):
+                pd = self.get_profile_dir(profile, path)
+                if profile not in self.profiles:
+                    self.log.debug("Overwriting profile %s" % profile)
+                    self.profiles[profile] = {
+                        'profile': profile,
+                        'profile_dir': pd,
+                        'status': 'stopped'
+                    }
 
     def list_profiles(self):
-        profiles = self.list_profile_names()
-        result = [self.profile_info(p) for p in profiles]
+        self.update_profiles()
+        result = [self.profile_info(p) for p in self.profiles.keys()]
         return result
 
+    def check_profile(self, profile):
+        if profile not in self.profiles:
+            raise web.HTTPError(404, u'profile not found')
 
     def profile_info(self, profile):
-        if profile not in self.list_profile_names():
-            raise web.HTTPError(404, u'profile not found')
-        result = dict(profile=profile)
+        self.check_profile(profile)
+        result = {}
         data = self.profiles.get(profile)
-        if data is None:
-            result['status'] = 'stopped'
-        else:
-            result['status'] = 'running'
+        result['profile'] = profile
+        result['profile_dir'] = data['profile_dir']
+        result['status'] = data['status']
+        if 'n' in data:
             result['n'] = data['n']
         return result
 
-    def start_cluster(self, profile, n=4):
+    def start_cluster(self, profile, n=None):
         """Start a cluster for a given profile."""
-        if profile not in self.list_profile_names():
-            raise web.HTTPError(404, u'profile not found')
-        if profile in self.profiles:
+        self.check_profile(profile)
+        data = self.profiles[profile]
+        if data['status'] == 'running':
             raise web.HTTPError(409, u'cluster already running')
-        launcher = IPClusterLauncher(ipcluster_profile=profile, ipcluster_n=n)
-        launcher.start()
-        self.profiles[profile] = {
-            'launcher': launcher,
-            'n': n
-        }
+        cl, esl, default_n = self.build_launchers(data['profile_dir'])
+        n = n if n is not None else default_n
+        def clean_data():
+            data.pop('controller_launcher',None)
+            data.pop('engine_set_launcher',None)
+            data.pop('n',None)
+            data['status'] = 'stopped'
+        def engines_stopped(r):
+            self.log.debug('Engines stopped')
+            if cl.running:
+                cl.stop()
+            clean_data()
+        esl.on_stop(engines_stopped)
+        def controller_stopped(r):
+            self.log.debug('Controller stopped')
+            if esl.running:
+                esl.stop()
+            clean_data()
+        cl.on_stop(controller_stopped)
+
+        dc = ioloop.DelayedCallback(lambda: cl.start(), 0, self.loop)
+        dc.start()
+        dc = ioloop.DelayedCallback(lambda: esl.start(n), 1000*self.delay, self.loop)
+        dc.start()
+
+        self.log.debug('Cluster started')
+        data['controller_launcher'] = cl
+        data['engine_set_launcher'] = esl
+        data['n'] = n
+        data['status'] = 'running'
         return self.profile_info(profile)
 
     def stop_cluster(self, profile):
         """Stop a cluster for a given profile."""
-        if profile not in self.profiles:
+        self.check_profile(profile)
+        data = self.profiles[profile]
+        if data['status'] == 'stopped':
             raise web.HTTPError(409, u'cluster not running')
-        launcher = self.profiles.pop(profile)['launcher']
-        launcher.stop()
-        return self.profile_info(profile)
+        data = self.profiles[profile]
+        cl = data['controller_launcher']
+        esl = data['engine_set_launcher']
+        if cl.running:
+            cl.stop()
+        if esl.running:
+            esl.stop()
+        # Return a temp info dict, the real one is updated in the on_stop
+        # logic above.
+        result = {
+            'profile': data['profile'],
+            'profile_dir': data['profile_dir'],
+            'status': 'stopped'
+        }
+        return result
 
     def stop_all_clusters(self):
-        for p in self.profiles.values():
-            p['launcher'].stop()
+        for p in self.profiles.keys():
+            self.stop_cluster(profile)

IPython/frontend/html/notebook/handlers.py

@@ -690,8 +690,11 b' class ClusterActionHandler(AuthenticatedHandler):'
     def post(self, profile, action):
         cm = self.application.cluster_manager
         if action == 'start':
-            n = int(self.get_argument('n', default=4))
-            data = cm.start_cluster(profile, n)
+            n = self.get_argument('n',default=None)
+            if n is None:
+                data = cm.start_cluster(profile)
+            else:
+                data = cm.start_cluster(profile,int(n))
         if action == 'stop':
             data = cm.stop_cluster(profile)
         self.finish(jsonapi.dumps(data))

IPython/frontend/html/notebook/notebookapp.py

@@ -406,6 +406,7 b' class NotebookApp(BaseIPythonApplication):'
         self.notebook_manager = NotebookManager(config=self.config, log=self.log)
         self.notebook_manager.list_notebooks()
         self.cluster_manager = ClusterManager(config=self.config, log=self.log)
+        self.cluster_manager.update_profiles()
 
     def init_logging(self):
         super(NotebookApp, self).init_logging()

IPython/frontend/html/notebook/static/js/clusterlist.js

@@ -116,11 +116,9 b' var IPython = (function (IPython) {'
             append(status_col);
         start_button.click(function (e) {
             var n = that.element.find('.engine_num_input').val();
-            console.log(n);
-            if (!/^\d+$/.test(n)) {
+            if (!/^\d+$/.test(n) && n.length>0) {
                 status_col.html('invalid engine #');
             } else {
-                console.log('ajax...');
                 var settings = {
                     cache : false,
                     data : {n:n},

IPython/parallel/apps/ipclusterapp.py

@@ -106,8 +106,35 b' NO_CLUSTER = 12'
 
 
 #-----------------------------------------------------------------------------
+# Utilities
+#-----------------------------------------------------------------------------
+
+def find_launcher_class(clsname, kind):
+    """Return a launcher for a given clsname and kind.
+
+    Parameters
+    ==========
+    clsname : str
+        The full name of the launcher class, either with or without the
+        module path, or an abbreviation (MPI, SSH, SGE, PBS, LSF,
+        WindowsHPC).
+    kind : str
+        Either 'EngineSet' or 'Controller'.
+    """
+    if '.' not in clsname:
+        # not a module, presume it's the raw name in apps.launcher
+        if kind and kind not in clsname:
+            # doesn't match necessary full class name, assume it's
+            # just 'PBS' or 'MPI' prefix:
+            clsname = clsname + kind + 'Launcher'
+        clsname = 'IPython.parallel.apps.launcher.'+clsname
+    klass = import_item(clsname)
+    return klass
+
+#-----------------------------------------------------------------------------
 # Main application
 #-----------------------------------------------------------------------------
+
 start_help = """Start an IPython cluster for parallel computing
 
 Start an ipython cluster by its profile name or cluster
@@ -303,15 +330,8 b' class IPClusterEngines(BaseParallelApplication):'
 
     def build_launcher(self, clsname, kind=None):
         """import and instantiate a Launcher based on importstring"""
-        if '.' not in clsname:
-            # not a module, presume it's the raw name in apps.launcher
-            if kind and kind not in clsname:
-                # doesn't match necessary full class name, assume it's
-                # just 'PBS' or 'MPI' prefix:
-                clsname = clsname + kind + 'Launcher'
-            clsname = 'IPython.parallel.apps.launcher.'+clsname
         try:
-            klass = import_item(clsname)
+            klass = find_launcher_class(clsname, kind)
         except (ImportError, KeyError):
             self.log.fatal("Could not import launcher class: %r"%clsname)
             self.exit(1)

IPython/parallel/apps/launcher.py

@@ -1167,12 +1167,12 b' class IPClusterLauncher(LocalProcessLauncher):'
         ['--clean-logs=True', '--log-to-file', '--log-level=%i'%logging.INFO], config=True,
         help="Command line arguments to pass to ipcluster.")
     ipcluster_subcommand = Unicode('start')
-    ipcluster_profile = Unicode('default')
-    ipcluster_n = Integer(2)
+    profile = Unicode('default')
+    n = Integer(2)
 
     def find_args(self):
         return self.ipcluster_cmd + [self.ipcluster_subcommand] + \
-            ['--n=%i'%self.ipcluster_n, '--profile=%s'%self.ipcluster_profile] + \
+            ['--n=%i'%self.n, '--profile=%s'%self.profile] + \
             self.ipcluster_args
 
     def start(self):