parallel.apps cleanup per review

STY: added spaces in string formatting calls
STY: expanded helpstring for cluster-id

ENH: allow BatchClusterAppMixin to be the second inherited class

IPython/parallel/apps/baseapp.py

@@ -1,257 +1,261 b''
 # encoding: utf-8
 """
 The Base Application class for IPython.parallel apps
 
 Authors:
 
 * Brian Granger
 * Min RK
 
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 from __future__ import with_statement
 
 import os
 import logging
 import re
 import sys
 
 from subprocess import Popen, PIPE
 
 from IPython.core import release
 from IPython.core.crashhandler import CrashHandler
 from IPython.core.application import (
     BaseIPythonApplication,
     base_aliases as base_ip_aliases,
     base_flags as base_ip_flags
 )
 from IPython.utils.path import expand_path
 
 from IPython.utils.traitlets import Unicode, Bool, Instance, Dict, List
 
 #-----------------------------------------------------------------------------
 # Module errors
 #-----------------------------------------------------------------------------
 
 class PIDFileError(Exception):
     pass
 
 
 #-----------------------------------------------------------------------------
 # Crash handler for this application
 #-----------------------------------------------------------------------------
 
 class ParallelCrashHandler(CrashHandler):
     """sys.excepthook for IPython itself, leaves a detailed report on disk."""
 
     def __init__(self, app):
         contact_name = release.authors['Min'][0]
         contact_email = release.authors['Min'][1]
         bug_tracker = 'http://github.com/ipython/ipython/issues'
         super(ParallelCrashHandler,self).__init__(
             app, contact_name, contact_email, bug_tracker
         )
 
 
 #-----------------------------------------------------------------------------
 # Main application
 #-----------------------------------------------------------------------------
 base_aliases = {}
 base_aliases.update(base_ip_aliases)
 base_aliases.update({
     'profile-dir' : 'ProfileDir.location',
     'work-dir' : 'BaseParallelApplication.work_dir',
     'log-to-file' : 'BaseParallelApplication.log_to_file',
     'clean-logs' : 'BaseParallelApplication.clean_logs',
     'log-url' : 'BaseParallelApplication.log_url',
     'cluster-id' : 'BaseParallelApplication.cluster_id',
 })
 
 base_flags = {
     'log-to-file' : (
         {'BaseParallelApplication' : {'log_to_file' : True}},
         "send log output to a file"
     )
 }
 base_flags.update(base_ip_flags)
 
 class BaseParallelApplication(BaseIPythonApplication):
     """The base Application for IPython.parallel apps
     
     Principle extensions to BaseIPyythonApplication:
     
     * work_dir
     * remote logging via pyzmq
     * IOLoop instance
     """
 
     crash_handler_class = ParallelCrashHandler
     
     def _log_level_default(self):
         # temporarily override default_log_level to INFO
         return logging.INFO
 
     work_dir = Unicode(os.getcwdu(), config=True,
         help='Set the working dir for the process.'
     )
     def _work_dir_changed(self, name, old, new):
         self.work_dir = unicode(expand_path(new))
 
     log_to_file = Bool(config=True,
         help="whether to log to a file")
 
     clean_logs = Bool(False, config=True,
         help="whether to cleanup old logfiles before starting")
 
     log_url = Unicode('', config=True,
         help="The ZMQ URL of the iplogger to aggregate logging.")
 
     cluster_id = Unicode('', config=True,
         help="""String id to add to runtime files, to prevent name collisions when
-        using multiple clusters with a single profile.
+        using multiple clusters with a single profile simultaneously.
         
         When set, files will be named like: 'ipcontroller-<cluster_id>-engine.json'
+        
+        Since this is text inserted into filenames, typical recommendations apply:
+        Simple character strings are ideal, and spaces are not recommended (but should
+        generally work).
         """
     )
     def _cluster_id_changed(self, name, old, new):
         self.name = self.__class__.name
         if new:
             self.name += '-%s'%new
     
     def _config_files_default(self):
         return ['ipcontroller_config.py', 'ipengine_config.py', 'ipcluster_config.py']
     
     loop = Instance('zmq.eventloop.ioloop.IOLoop')
     def _loop_default(self):
         from zmq.eventloop.ioloop import IOLoop
         return IOLoop.instance()
 
     aliases = Dict(base_aliases)
     flags = Dict(base_flags)
     
     def initialize(self, argv=None):
         """initialize the app"""
         super(BaseParallelApplication, self).initialize(argv)
         self.to_work_dir()
         self.reinit_logging()
         
     def to_work_dir(self):
         wd = self.work_dir
         if unicode(wd) != os.getcwdu():
             os.chdir(wd)
             self.log.info("Changing to working dir: %s" % wd)
         # This is the working dir by now.
         sys.path.insert(0, '')
 
     def reinit_logging(self):
         # Remove old log files
         log_dir = self.profile_dir.log_dir
         if self.clean_logs:
             for f in os.listdir(log_dir):
                 if re.match(r'%s-\d+\.(log|err|out)'%self.name,f):
                     os.remove(os.path.join(log_dir, f))
         if self.log_to_file:
             # Start logging to the new log file
             log_filename = self.name + u'-' + str(os.getpid()) + u'.log'
             logfile = os.path.join(log_dir, log_filename)
             open_log_file = open(logfile, 'w')
         else:
             open_log_file = None
         if open_log_file is not None:
             self.log.removeHandler(self._log_handler)
             self._log_handler = logging.StreamHandler(open_log_file)
             self._log_formatter = logging.Formatter("[%(name)s] %(message)s")
             self._log_handler.setFormatter(self._log_formatter)
             self.log.addHandler(self._log_handler)
         # do not propagate log messages to root logger
         # ipcluster app will sometimes print duplicate messages during shutdown
         # if this is 1 (default):
         self.log.propagate = False
 
     def write_pid_file(self, overwrite=False):
         """Create a .pid file in the pid_dir with my pid.
 
         This must be called after pre_construct, which sets `self.pid_dir`.
         This raises :exc:`PIDFileError` if the pid file exists already.
         """
         pid_file = os.path.join(self.profile_dir.pid_dir, self.name + u'.pid')
         if os.path.isfile(pid_file):
             pid = self.get_pid_from_file()
             if not overwrite:
                 raise PIDFileError(
                     'The pid file [%s] already exists. \nThis could mean that this '
                     'server is already running with [pid=%s].' % (pid_file, pid)
                 )
         with open(pid_file, 'w') as f:
             self.log.info("Creating pid file: %s" % pid_file)
             f.write(repr(os.getpid())+'\n')
 
     def remove_pid_file(self):
         """Remove the pid file.
 
         This should be called at shutdown by registering a callback with
         :func:`reactor.addSystemEventTrigger`. This needs to return
         ``None``.
         """
         pid_file = os.path.join(self.profile_dir.pid_dir, self.name + u'.pid')
         if os.path.isfile(pid_file):
             try:
                 self.log.info("Removing pid file: %s" % pid_file)
                 os.remove(pid_file)
             except:
                 self.log.warn("Error removing the pid file: %s" % pid_file)
 
     def get_pid_from_file(self):
         """Get the pid from the pid file.
 
         If the  pid file doesn't exist a :exc:`PIDFileError` is raised.
         """
         pid_file = os.path.join(self.profile_dir.pid_dir, self.name + u'.pid')
         if os.path.isfile(pid_file):
             with open(pid_file, 'r') as f:
                 s = f.read().strip()
                 try:
                     pid = int(s)
                 except:
                     raise PIDFileError("invalid pid file: %s (contents: %r)"%(pid_file, s))
                 return pid
         else:
             raise PIDFileError('pid file not found: %s' % pid_file)
     
     def check_pid(self, pid):
         if os.name == 'nt':
             try:
                 import ctypes
                 # returns 0 if no such process (of ours) exists
                 # positive int otherwise
                 p = ctypes.windll.kernel32.OpenProcess(1,0,pid)
             except Exception:
                 self.log.warn(
                     "Could not determine whether pid %i is running via `OpenProcess`. "
                     " Making the likely assumption that it is."%pid
                 )
                 return True
             return bool(p)
         else:
             try:
                 p = Popen(['ps','x'], stdout=PIPE, stderr=PIPE)
                 output,_ = p.communicate()
             except OSError:
                 self.log.warn(
                     "Could not determine whether pid %i is running via `ps x`. "
                     " Making the likely assumption that it is."%pid
                 )
                 return True
             pids = map(int, re.findall(r'^\W*\d+', output, re.MULTILINE))
             return pid in pids

IPython/parallel/apps/ipcontrollerapp.py

@@ -1,452 +1,452 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 The IPython controller application.
 
 Authors:
 
 * Brian Granger
 * MinRK
 
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 from __future__ import with_statement
 
 import os
 import socket
 import stat
 import sys
 import uuid
 
 from multiprocessing import Process
 
 import zmq
 from zmq.devices import ProcessMonitoredQueue
 from zmq.log.handlers import PUBHandler
 from zmq.utils import jsonapi as json
 
 from IPython.config.application import boolean_flag
 from IPython.core.profiledir import ProfileDir
 
 from IPython.parallel.apps.baseapp import (
     BaseParallelApplication,
     base_aliases,
     base_flags,
 )
 from IPython.utils.importstring import import_item
 from IPython.utils.traitlets import Instance, Unicode, Bool, List, Dict
 
 # from IPython.parallel.controller.controller import ControllerFactory
 from IPython.zmq.session import Session
 from IPython.parallel.controller.heartmonitor import HeartMonitor
 from IPython.parallel.controller.hub import HubFactory
 from IPython.parallel.controller.scheduler import TaskScheduler,launch_scheduler
 from IPython.parallel.controller.sqlitedb import SQLiteDB
 
 from IPython.parallel.util import signal_children, split_url, asbytes
 
 # conditional import of MongoDB backend class
 
 try:
     from IPython.parallel.controller.mongodb import MongoDB
 except ImportError:
     maybe_mongo = []
 else:
     maybe_mongo = [MongoDB]
 
 
 #-----------------------------------------------------------------------------
 # Module level variables
 #-----------------------------------------------------------------------------
 
 
 #: The default config file name for this application
 default_config_file_name = u'ipcontroller_config.py'
 
 
 _description = """Start the IPython controller for parallel computing.
 
 The IPython controller provides a gateway between the IPython engines and
 clients. The controller needs to be started before the engines and can be
 configured using command line options or using a cluster directory. Cluster
 directories contain config, log and security files and are usually located in
 your ipython directory and named as "profile_name". See the `profile`
 and `profile-dir` options for details.
 """
 
 _examples = """
 ipcontroller --ip=192.168.0.1 --port=1000  # listen on ip, port for engines
 ipcontroller --scheme=pure  # use the pure zeromq scheduler
 """
 
 
 #-----------------------------------------------------------------------------
 # The main application
 #-----------------------------------------------------------------------------
 flags = {}
 flags.update(base_flags)
 flags.update({
     'usethreads' : ( {'IPControllerApp' : {'use_threads' : True}},
                     'Use threads instead of processes for the schedulers'),
     'sqlitedb' : ({'HubFactory' : {'db_class' : 'IPython.parallel.controller.sqlitedb.SQLiteDB'}},
                     'use the SQLiteDB backend'),
     'mongodb' : ({'HubFactory' : {'db_class' : 'IPython.parallel.controller.mongodb.MongoDB'}},
                     'use the MongoDB backend'),
     'dictdb' : ({'HubFactory' : {'db_class' : 'IPython.parallel.controller.dictdb.DictDB'}},
                     'use the in-memory DictDB backend'),
     'reuse' : ({'IPControllerApp' : {'reuse_files' : True}},
                     'reuse existing json connection files')
 })
 
 flags.update(boolean_flag('secure', 'IPControllerApp.secure',
     "Use HMAC digests for authentication of messages.",
     "Don't authenticate messages."
 ))
 aliases = dict(
     secure = 'IPControllerApp.secure',
     ssh = 'IPControllerApp.ssh_server',
     enginessh = 'IPControllerApp.engine_ssh_server',
     location = 'IPControllerApp.location',
 
     ident = 'Session.session',
     user = 'Session.username',
     keyfile = 'Session.keyfile',
 
     url = 'HubFactory.url',
     ip = 'HubFactory.ip',
     transport = 'HubFactory.transport',
     port = 'HubFactory.regport',
 
     ping = 'HeartMonitor.period',
 
     scheme = 'TaskScheduler.scheme_name',
     hwm = 'TaskScheduler.hwm',
 )
 aliases.update(base_aliases)
 
 
 class IPControllerApp(BaseParallelApplication):
 
     name = u'ipcontroller'
     description = _description
     examples = _examples
     config_file_name = Unicode(default_config_file_name)
     classes = [ProfileDir, Session, HubFactory, TaskScheduler, HeartMonitor, SQLiteDB] + maybe_mongo
     
     # change default to True
     auto_create = Bool(True, config=True,
         help="""Whether to create profile dir if it doesn't exist.""")
     
     reuse_files = Bool(False, config=True,
         help='Whether to reuse existing json connection files.'
     )
     secure = Bool(True, config=True,
         help='Whether to use HMAC digests for extra message authentication.'
     )
     ssh_server = Unicode(u'', config=True,
         help="""ssh url for clients to use when connecting to the Controller
         processes. It should be of the form: [user@]server[:port]. The
         Controller's listening addresses must be accessible from the ssh server""",
     )
     engine_ssh_server = Unicode(u'', config=True,
         help="""ssh url for engines to use when connecting to the Controller
         processes. It should be of the form: [user@]server[:port]. The
         Controller's listening addresses must be accessible from the ssh server""",
     )
     location = Unicode(u'', config=True,
         help="""The external IP or domain name of the Controller, used for disambiguating
         engine and client connections.""",
     )
     import_statements = List([], config=True,
         help="import statements to be run at startup.  Necessary in some environments"
     )
 
     use_threads = Bool(False, config=True,
         help='Use threads instead of processes for the schedulers',
     )
 
     engine_json_file = Unicode('ipcontroller-engine.json', config=True,
         help="JSON filename where engine connection info will be stored.")
     client_json_file = Unicode('ipcontroller-client.json', config=True,
         help="JSON filename where client connection info will be stored.")
     
     def _cluster_id_changed(self, name, old, new):
         super(IPControllerApp, self)._cluster_id_changed(name, old, new)
-        self.engine_json_file = "%s-engine.json"%self.name
-        self.client_json_file = "%s-client.json"%self.name
+        self.engine_json_file = "%s-engine.json" % self.name
+        self.client_json_file = "%s-client.json" % self.name
 
 
     # internal
     children = List()
     mq_class = Unicode('zmq.devices.ProcessMonitoredQueue')
 
     def _use_threads_changed(self, name, old, new):
         self.mq_class = 'zmq.devices.%sMonitoredQueue'%('Thread' if new else 'Process')
 
     aliases = Dict(aliases)
     flags = Dict(flags)
     
 
     def save_connection_dict(self, fname, cdict):
         """save a connection dict to json file."""
         c = self.config
         url = cdict['url']
         location = cdict['location']
         if not location:
             try:
                 proto,ip,port = split_url(url)
             except AssertionError:
                 pass
             else:
                 try:
                     location = socket.gethostbyname_ex(socket.gethostname())[2][-1]
                 except (socket.gaierror, IndexError):
                     self.log.warn("Could not identify this machine's IP, assuming 127.0.0.1."
                     " You may need to specify '--location=<external_ip_address>' to help"
                     " IPython decide when to connect via loopback.")
                     location = '127.0.0.1'
             cdict['location'] = location
         fname = os.path.join(self.profile_dir.security_dir, fname)
         with open(fname, 'wb') as f:
             f.write(json.dumps(cdict, indent=2))
         os.chmod(fname, stat.S_IRUSR|stat.S_IWUSR)
     
     def load_config_from_json(self):
         """load config from existing json connector files."""
         c = self.config
         # load from engine config
         with open(os.path.join(self.profile_dir.security_dir, self.engine_json_file)) as f:
             cfg = json.loads(f.read())
         key = c.Session.key = asbytes(cfg['exec_key'])
         xport,addr = cfg['url'].split('://')
         c.HubFactory.engine_transport = xport
         ip,ports = addr.split(':')
         c.HubFactory.engine_ip = ip
         c.HubFactory.regport = int(ports)
         self.location = cfg['location']
         if not self.engine_ssh_server:
             self.engine_ssh_server = cfg['ssh']
         # load client config
         with open(os.path.join(self.profile_dir.security_dir, self.client_json_file)) as f:
             cfg = json.loads(f.read())
         assert key == cfg['exec_key'], "exec_key mismatch between engine and client keys"
         xport,addr = cfg['url'].split('://')
         c.HubFactory.client_transport = xport
         ip,ports = addr.split(':')
         c.HubFactory.client_ip = ip
         if not self.ssh_server:
             self.ssh_server = cfg['ssh']
         assert int(ports) == c.HubFactory.regport, "regport mismatch"
     
     def init_hub(self):
         c = self.config
         
         self.do_import_statements()
         reusing = self.reuse_files
         if reusing:
             try:
                 self.load_config_from_json()
             except (AssertionError,IOError):
                 reusing=False
         # check again, because reusing may have failed:
         if reusing:
             pass
         elif self.secure:
             key = str(uuid.uuid4())
             # keyfile = os.path.join(self.profile_dir.security_dir, self.exec_key)
             # with open(keyfile, 'w') as f:
             #     f.write(key)
             # os.chmod(keyfile, stat.S_IRUSR|stat.S_IWUSR)
             c.Session.key = asbytes(key)
         else:
             key = c.Session.key = b''
         
         try:
             self.factory = HubFactory(config=c, log=self.log)
             # self.start_logging()
             self.factory.init_hub()
         except:
             self.log.error("Couldn't construct the Controller", exc_info=True)
             self.exit(1)
         
         if not reusing:
             # save to new json config files
             f = self.factory
             cdict = {'exec_key' : key,
                     'ssh' : self.ssh_server,
                     'url' : "%s://%s:%s"%(f.client_transport, f.client_ip, f.regport),
                     'location' : self.location
                     }
             self.save_connection_dict(self.client_json_file, cdict)
             edict = cdict
             edict['url']="%s://%s:%s"%((f.client_transport, f.client_ip, f.regport))
             edict['ssh'] = self.engine_ssh_server
             self.save_connection_dict(self.engine_json_file, edict)
 
     #
     def init_schedulers(self):
         children = self.children
         mq = import_item(str(self.mq_class))
         
         hub = self.factory
         # maybe_inproc = 'inproc://monitor' if self.use_threads else self.monitor_url
         # IOPub relay (in a Process)
         q = mq(zmq.PUB, zmq.SUB, zmq.PUB, b'N/A',b'iopub')
         q.bind_in(hub.client_info['iopub'])
         q.bind_out(hub.engine_info['iopub'])
         q.setsockopt_out(zmq.SUBSCRIBE, b'')
         q.connect_mon(hub.monitor_url)
         q.daemon=True
         children.append(q)
 
         # Multiplexer Queue (in a Process)
         q = mq(zmq.ROUTER, zmq.ROUTER, zmq.PUB, b'in', b'out')
         q.bind_in(hub.client_info['mux'])
         q.setsockopt_in(zmq.IDENTITY, b'mux')
         q.bind_out(hub.engine_info['mux'])
         q.connect_mon(hub.monitor_url)
         q.daemon=True
         children.append(q)
 
         # Control Queue (in a Process)
         q = mq(zmq.ROUTER, zmq.ROUTER, zmq.PUB, b'incontrol', b'outcontrol')
         q.bind_in(hub.client_info['control'])
         q.setsockopt_in(zmq.IDENTITY, b'control')
         q.bind_out(hub.engine_info['control'])
         q.connect_mon(hub.monitor_url)
         q.daemon=True
         children.append(q)
         try:
             scheme = self.config.TaskScheduler.scheme_name
         except AttributeError:
             scheme = TaskScheduler.scheme_name.get_default_value()
         # Task Queue (in a Process)
         if scheme == 'pure':
             self.log.warn("task::using pure XREQ Task scheduler")
             q = mq(zmq.ROUTER, zmq.DEALER, zmq.PUB, b'intask', b'outtask')
             # q.setsockopt_out(zmq.HWM, hub.hwm)
             q.bind_in(hub.client_info['task'][1])
             q.setsockopt_in(zmq.IDENTITY, b'task')
             q.bind_out(hub.engine_info['task'])
             q.connect_mon(hub.monitor_url)
             q.daemon=True
             children.append(q)
         elif scheme == 'none':
             self.log.warn("task::using no Task scheduler")
 
         else:
             self.log.info("task::using Python %s Task scheduler"%scheme)
             sargs = (hub.client_info['task'][1], hub.engine_info['task'],
                                 hub.monitor_url, hub.client_info['notification'])
             kwargs = dict(logname='scheduler', loglevel=self.log_level,
                             log_url = self.log_url, config=dict(self.config))
             if 'Process' in self.mq_class:
                 # run the Python scheduler in a Process
                 q = Process(target=launch_scheduler, args=sargs, kwargs=kwargs)
                 q.daemon=True
                 children.append(q)
             else:
                 # single-threaded Controller
                 kwargs['in_thread'] = True
                 launch_scheduler(*sargs, **kwargs)
 
     
     def save_urls(self):
         """save the registration urls to files."""
         c = self.config
         
         sec_dir = self.profile_dir.security_dir
         cf = self.factory
         
         with open(os.path.join(sec_dir, 'ipcontroller-engine.url'), 'w') as f:
             f.write("%s://%s:%s"%(cf.engine_transport, cf.engine_ip, cf.regport))
         
         with open(os.path.join(sec_dir, 'ipcontroller-client.url'), 'w') as f:
             f.write("%s://%s:%s"%(cf.client_transport, cf.client_ip, cf.regport))
         
     
     def do_import_statements(self):
         statements = self.import_statements
         for s in statements:
             try:
                 self.log.msg("Executing statement: '%s'" % s)
                 exec s in globals(), locals()
             except:
                 self.log.msg("Error running statement: %s" % s)
 
     def forward_logging(self):
         if self.log_url:
             self.log.info("Forwarding logging to %s"%self.log_url)
             context = zmq.Context.instance()
             lsock = context.socket(zmq.PUB)
             lsock.connect(self.log_url)
             handler = PUBHandler(lsock)
             self.log.removeHandler(self._log_handler)
             handler.root_topic = 'controller'
             handler.setLevel(self.log_level)
             self.log.addHandler(handler)
             self._log_handler = handler
     # #
     
     def initialize(self, argv=None):
         super(IPControllerApp, self).initialize(argv)
         self.forward_logging()
         self.init_hub()
         self.init_schedulers()
     
     def start(self):
         # Start the subprocesses:
         self.factory.start()
         child_procs = []
         for child in self.children:
             child.start()
             if isinstance(child, ProcessMonitoredQueue):
                 child_procs.append(child.launcher)
             elif isinstance(child, Process):
                 child_procs.append(child)
         if child_procs:
             signal_children(child_procs)
 
         self.write_pid_file(overwrite=True)
 
         try:
             self.factory.loop.start()
         except KeyboardInterrupt:
             self.log.critical("Interrupted, Exiting...\n")
             
 
 
 def launch_new_instance():
     """Create and run the IPython controller"""
     if sys.platform == 'win32':
         # make sure we don't get called from a multiprocessing subprocess
         # this can result in infinite Controllers being started on Windows
         # which doesn't have a proper fork, so multiprocessing is wonky
         
         # this only comes up when IPython has been installed using vanilla
         # setuptools, and *not* distribute.
         import multiprocessing
         p = multiprocessing.current_process()
         # the main process has name 'MainProcess'
         # subprocesses will have names like 'Process-1'
         if p.name != 'MainProcess':
             # we are a subprocess, don't start another Controller!
             return
     app = IPControllerApp.instance()
     app.initialize()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()

IPython/parallel/apps/ipengineapp.py

@@ -1,344 +1,344 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 The IPython engine application
 
 Authors:
 
 * Brian Granger
 * MinRK
 
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import json
 import os
 import sys
 import time
 
 import zmq
 from zmq.eventloop import ioloop
 
 from IPython.core.profiledir import ProfileDir
 from IPython.parallel.apps.baseapp import (
     BaseParallelApplication,
     base_aliases,
     base_flags,
 )
 from IPython.zmq.log import EnginePUBHandler
 
 from IPython.config.configurable import Configurable
 from IPython.zmq.session import Session
 from IPython.parallel.engine.engine import EngineFactory
 from IPython.parallel.engine.streamkernel import Kernel
 from IPython.parallel.util import disambiguate_url, asbytes
 
 from IPython.utils.importstring import import_item
 from IPython.utils.traitlets import Bool, Unicode, Dict, List, Float
 
 
 #-----------------------------------------------------------------------------
 # Module level variables
 #-----------------------------------------------------------------------------
 
 #: The default config file name for this application
 default_config_file_name = u'ipengine_config.py'
 
 _description = """Start an IPython engine for parallel computing.
 
 IPython engines run in parallel and perform computations on behalf of a client
 and controller. A controller needs to be started before the engines. The
 engine can be configured using command line options or using a cluster
 directory. Cluster directories contain config, log and security files and are
 usually located in your ipython directory and named as "profile_name".
 See the `profile` and `profile-dir` options for details.
 """
 
 _examples = """
 ipengine --ip=192.168.0.1 --port=1000     # connect to hub at ip and port
 ipengine --log-to-file --log-level=DEBUG  # log to a file with DEBUG verbosity
 """
 
 #-----------------------------------------------------------------------------
 # MPI configuration
 #-----------------------------------------------------------------------------
 
 mpi4py_init = """from mpi4py import MPI as mpi
 mpi.size = mpi.COMM_WORLD.Get_size()
 mpi.rank = mpi.COMM_WORLD.Get_rank()
 """
 
 
 pytrilinos_init = """from PyTrilinos import Epetra
 class SimpleStruct:
 pass
 mpi = SimpleStruct()
 mpi.rank = 0
 mpi.size = 0
 """
 
 class MPI(Configurable):
     """Configurable for MPI initialization"""
     use = Unicode('', config=True,
         help='How to enable MPI (mpi4py, pytrilinos, or empty string to disable).'
         )
 
-    def _on_use_changed(self, old, new):
+    def _use_changed(self, name, old, new):
         # load default init script if it's not set
         if not self.init_script:
             self.init_script = self.default_inits.get(new, '')
 
     init_script = Unicode('', config=True,
         help="Initialization code for MPI")
 
     default_inits = Dict({'mpi4py' : mpi4py_init, 'pytrilinos':pytrilinos_init},
         config=True)
 
 
 #-----------------------------------------------------------------------------
 # Main application
 #-----------------------------------------------------------------------------
 aliases = dict(
     file = 'IPEngineApp.url_file',
     c = 'IPEngineApp.startup_command',
     s = 'IPEngineApp.startup_script',
 
     ident = 'Session.session',
     user = 'Session.username',
     keyfile = 'Session.keyfile',
 
     url = 'EngineFactory.url',
     ssh = 'EngineFactory.sshserver',
     sshkey = 'EngineFactory.sshkey',
     ip = 'EngineFactory.ip',
     transport = 'EngineFactory.transport',
     port = 'EngineFactory.regport',
     location = 'EngineFactory.location',
 
     timeout = 'EngineFactory.timeout',
 
     mpi = 'MPI.use',
 
 )
 aliases.update(base_aliases)
 
 
 class IPEngineApp(BaseParallelApplication):
 
     name = 'ipengine'
     description = _description
     examples = _examples
     config_file_name = Unicode(default_config_file_name)
     classes = List([ProfileDir, Session, EngineFactory, Kernel, MPI])
 
     startup_script = Unicode(u'', config=True,
         help='specify a script to be run at startup')
     startup_command = Unicode('', config=True,
             help='specify a command to be run at startup')
 
     url_file = Unicode(u'', config=True,
         help="""The full location of the file containing the connection information for
         the controller. If this is not given, the file must be in the
         security directory of the cluster directory.  This location is
         resolved using the `profile` or `profile_dir` options.""",
         )
     wait_for_url_file = Float(5, config=True,
         help="""The maximum number of seconds to wait for url_file to exist.
         This is useful for batch-systems and shared-filesystems where the
         controller and engine are started at the same time and it
         may take a moment for the controller to write the connector files.""")
 
     url_file_name = Unicode(u'ipcontroller-engine.json', config=True)
 
     def _cluster_id_changed(self, name, old, new):
         if new:
-            base = 'ipcontroller-%s'%new
+            base = 'ipcontroller-%s' % new
         else:
             base = 'ipcontroller'
-        self.url_file_name = "%s-engine.json"%base
+        self.url_file_name = "%s-engine.json" % base
 
     log_url = Unicode('', config=True,
         help="""The URL for the iploggerapp instance, for forwarding
         logging to a central location.""")
 
     aliases = Dict(aliases)
 
     # def find_key_file(self):
     #     """Set the key file.
     # 
     #     Here we don't try to actually see if it exists for is valid as that
     #     is hadled by the connection logic.
     #     """
     #     config = self.master_config
     #     # Find the actual controller key file
     #     if not config.Global.key_file:
     #         try_this = os.path.join(
     #             config.Global.profile_dir, 
     #             config.Global.security_dir,
     #             config.Global.key_file_name
     #         )
     #         config.Global.key_file = try_this
         
     def find_url_file(self):
         """Set the url file.
 
         Here we don't try to actually see if it exists for is valid as that
         is hadled by the connection logic.
         """
         config = self.config
         # Find the actual controller key file
         if not self.url_file:
             self.url_file = os.path.join(
                 self.profile_dir.security_dir,
                 self.url_file_name
             )
     
     def load_connector_file(self):
         """load config from a JSON connector file,
         at a *lower* priority than command-line/config files.
         """
         
         self.log.info("Loading url_file %r"%self.url_file)
         config = self.config
         
         with open(self.url_file) as f:
             d = json.loads(f.read())
         
         try:
             config.Session.key
         except AttributeError:
             if d['exec_key']:
                 config.Session.key = asbytes(d['exec_key'])
                 
         try:
             config.EngineFactory.location
         except AttributeError:
             config.EngineFactory.location = d['location']
         
         d['url'] = disambiguate_url(d['url'], config.EngineFactory.location)
         try:
             config.EngineFactory.url
         except AttributeError:
             config.EngineFactory.url = d['url']
         
         try:
             config.EngineFactory.sshserver
         except AttributeError:
             config.EngineFactory.sshserver = d['ssh']
         
     def init_engine(self):
         # This is the working dir by now.
         sys.path.insert(0, '')
         config = self.config
         # print config
         self.find_url_file()
         
         # was the url manually specified?
         keys = set(self.config.EngineFactory.keys())
         keys = keys.union(set(self.config.RegistrationFactory.keys()))
         
         if keys.intersection(set(['ip', 'url', 'port'])):
             # Connection info was specified, don't wait for the file
             url_specified = True
             self.wait_for_url_file = 0
         else:
             url_specified = False
 
         if self.wait_for_url_file and not os.path.exists(self.url_file):
             self.log.warn("url_file %r not found"%self.url_file)
             self.log.warn("Waiting up to %.1f seconds for it to arrive."%self.wait_for_url_file)
             tic = time.time()
             while not os.path.exists(self.url_file) and (time.time()-tic < self.wait_for_url_file):
                 # wait for url_file to exist, for up to 10 seconds
                 time.sleep(0.1)
             
         if os.path.exists(self.url_file):
             self.load_connector_file()
         elif not url_specified:
             self.log.critical("Fatal: url file never arrived: %s"%self.url_file)
             self.exit(1)
         
         
         try:
             exec_lines = config.Kernel.exec_lines
         except AttributeError:
             config.Kernel.exec_lines = []
             exec_lines = config.Kernel.exec_lines
         
         if self.startup_script:
             enc = sys.getfilesystemencoding() or 'utf8'
             cmd="execfile(%r)"%self.startup_script.encode(enc)
             exec_lines.append(cmd)
         if self.startup_command:
             exec_lines.append(self.startup_command)
 
         # Create the underlying shell class and Engine
         # shell_class = import_item(self.master_config.Global.shell_class)
         # print self.config
         try:
             self.engine = EngineFactory(config=config, log=self.log)
         except:
             self.log.error("Couldn't start the Engine", exc_info=True)
             self.exit(1)
     
     def forward_logging(self):
         if self.log_url:
             self.log.info("Forwarding logging to %s"%self.log_url)
             context = self.engine.context
             lsock = context.socket(zmq.PUB)
             lsock.connect(self.log_url)
             self.log.removeHandler(self._log_handler)
             handler = EnginePUBHandler(self.engine, lsock)
             handler.setLevel(self.log_level)
             self.log.addHandler(handler)
             self._log_handler = handler
     
     def init_mpi(self):
         global mpi
         self.mpi = MPI(config=self.config)
 
         mpi_import_statement = self.mpi.init_script
         if mpi_import_statement:
             try:
                 self.log.info("Initializing MPI:")
                 self.log.info(mpi_import_statement)
                 exec mpi_import_statement in globals()
             except:
                 mpi = None
         else:
             mpi = None
 
     def initialize(self, argv=None):
         super(IPEngineApp, self).initialize(argv)
         self.init_mpi()
         self.init_engine()
         self.forward_logging()
     
     def start(self):
         self.engine.start()
         try:
             self.engine.loop.start()
         except KeyboardInterrupt:
             self.log.critical("Engine Interrupted, shutting down...\n")
 
 
 def launch_new_instance():
     """Create and run the IPython engine"""
     app = IPEngineApp.instance()
     app.initialize()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()
 

IPython/parallel/apps/launcher.py

@@ -1,1169 +1,1176 b''
 # encoding: utf-8
 """
 Facilities for launching IPython processes asynchronously.
 
 Authors:
 
 * Brian Granger
 * MinRK
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import copy
 import logging
 import os
 import re
 import stat
 import time
 
 # signal imports, handling various platforms, versions
 
 from signal import SIGINT, SIGTERM
 try:
     from signal import SIGKILL
 except ImportError:
     # Windows
     SIGKILL=SIGTERM
 
 try:
     # Windows >= 2.7, 3.2
     from signal import CTRL_C_EVENT as SIGINT
 except ImportError:
     pass
 
 from subprocess import Popen, PIPE, STDOUT
 try:
     from subprocess import check_output
 except ImportError:
     # pre-2.7, define check_output with Popen
     def check_output(*args, **kwargs):
         kwargs.update(dict(stdout=PIPE))
         p = Popen(*args, **kwargs)
         out,err = p.communicate()
         return out
 
 from zmq.eventloop import ioloop
 
 from IPython.config.application import Application
 from IPython.config.configurable import LoggingConfigurable
 from IPython.utils.text import EvalFormatter
 from IPython.utils.traitlets import (
     Any, Int, CFloat, List, Unicode, Dict, Instance, HasTraits,
 )
 from IPython.utils.path import get_ipython_module_path
 from IPython.utils.process import find_cmd, pycmd2argv, FindCmdError
 
 from .win32support import forward_read_events
 
 from .winhpcjob import IPControllerTask, IPEngineTask, IPControllerJob, IPEngineSetJob
 
 WINDOWS = os.name == 'nt'
 
 #-----------------------------------------------------------------------------
 # Paths to the kernel apps
 #-----------------------------------------------------------------------------
 
 
 ipcluster_cmd_argv = pycmd2argv(get_ipython_module_path(
     'IPython.parallel.apps.ipclusterapp'
 ))
 
 ipengine_cmd_argv = pycmd2argv(get_ipython_module_path(
     'IPython.parallel.apps.ipengineapp'
 ))
 
 ipcontroller_cmd_argv = pycmd2argv(get_ipython_module_path(
     'IPython.parallel.apps.ipcontrollerapp'
 ))
 
 #-----------------------------------------------------------------------------
 # Base launchers and errors
 #-----------------------------------------------------------------------------
 
 
 class LauncherError(Exception):
     pass
 
 
 class ProcessStateError(LauncherError):
     pass
 
 
 class UnknownStatus(LauncherError):
     pass
 
 
 class BaseLauncher(LoggingConfigurable):
     """An asbtraction for starting, stopping and signaling a process."""
 
     # In all of the launchers, the work_dir is where child processes will be
     # run. This will usually be the profile_dir, but may not be. any work_dir
     # passed into the __init__ method will override the config value.
     # This should not be used to set the work_dir for the actual engine
     # and controller. Instead, use their own config files or the
     # controller_args, engine_args attributes of the launchers to add
     # the work_dir option.
     work_dir = Unicode(u'.')
     loop = Instance('zmq.eventloop.ioloop.IOLoop')
     
     start_data = Any()
     stop_data = Any()
     
     def _loop_default(self):
         return ioloop.IOLoop.instance()
 
     def __init__(self, work_dir=u'.', config=None, **kwargs):
         super(BaseLauncher, self).__init__(work_dir=work_dir, config=config, **kwargs)
         self.state = 'before' # can be before, running, after
         self.stop_callbacks = []
         self.start_data = None
         self.stop_data = None
 
     @property
     def args(self):
         """A list of cmd and args that will be used to start the process.
 
         This is what is passed to :func:`spawnProcess` and the first element
         will be the process name.
         """
         return self.find_args()
 
     def find_args(self):
         """The ``.args`` property calls this to find the args list.
 
         Subcommand should implement this to construct the cmd and args.
         """
         raise NotImplementedError('find_args must be implemented in a subclass')
 
     @property
     def arg_str(self):
         """The string form of the program arguments."""
         return ' '.join(self.args)
 
     @property
     def running(self):
         """Am I running."""
         if self.state == 'running':
             return True
         else:
             return False
 
     def start(self):
         """Start the process."""
         raise NotImplementedError('start must be implemented in a subclass')
 
     def stop(self):
         """Stop the process and notify observers of stopping.
 
         This method will return None immediately.
         To observe the actual process stopping, see :meth:`on_stop`.
         """
         raise NotImplementedError('stop must be implemented in a subclass')
 
     def on_stop(self, f):
         """Register a callback to be called with this Launcher's stop_data
         when the process actually finishes.
         """
         if self.state=='after':
             return f(self.stop_data)
         else:
             self.stop_callbacks.append(f)
 
     def notify_start(self, data):
         """Call this to trigger startup actions.
 
         This logs the process startup and sets the state to 'running'.  It is
         a pass-through so it can be used as a callback.
         """
 
         self.log.info('Process %r started: %r' % (self.args[0], data))
         self.start_data = data
         self.state = 'running'
         return data
 
     def notify_stop(self, data):
         """Call this to trigger process stop actions.
 
         This logs the process stopping and sets the state to 'after'. Call
         this to trigger callbacks registered via :meth:`on_stop`."""
 
         self.log.info('Process %r stopped: %r' % (self.args[0], data))
         self.stop_data = data
         self.state = 'after'
         for i in range(len(self.stop_callbacks)):
             d = self.stop_callbacks.pop()
             d(data)
         return data
 
     def signal(self, sig):
         """Signal the process.
 
         Parameters
         ----------
         sig : str or int
             'KILL', 'INT', etc., or any signal number
         """
         raise NotImplementedError('signal must be implemented in a subclass')
 
 class ClusterAppMixin(HasTraits):
     """MixIn for cluster args as traits"""
     cluster_args = List([])
     profile_dir=Unicode('')
     cluster_id=Unicode('')
     def _profile_dir_changed(self, name, old, new):
         self.cluster_args = []
         if self.profile_dir:
             self.cluster_args.extend(['--profile-dir', self.profile_dir])
         if self.cluster_id:
             self.cluster_args.extend(['--cluster-id', self.cluster_id])
     _cluster_id_changed = _profile_dir_changed
 
 class ControllerMixin(ClusterAppMixin):
     controller_cmd = List(ipcontroller_cmd_argv, config=True,
         help="""Popen command to launch ipcontroller.""")
     # Command line arguments to ipcontroller.
-    controller_args = List(['--log-to-file','--log-level=%i'%logging.INFO], config=True,
+    controller_args = List(['--log-to-file','--log-level=%i' % logging.INFO], config=True,
         help="""command-line args to pass to ipcontroller""")
 
 class EngineMixin(ClusterAppMixin):
     engine_cmd = List(ipengine_cmd_argv, config=True,
         help="""command to launch the Engine.""")
     # Command line arguments for ipengine.
-    engine_args = List(['--log-to-file','--log-level=%i'%logging.INFO], config=True,
+    engine_args = List(['--log-to-file','--log-level=%i' % logging.INFO], config=True,
         help="command-line arguments to pass to ipengine"
     )
 
 #-----------------------------------------------------------------------------
 # Local process launchers
 #-----------------------------------------------------------------------------
 
 
 class LocalProcessLauncher(BaseLauncher):
     """Start and stop an external process in an asynchronous manner.
 
     This will launch the external process with a working directory of
     ``self.work_dir``.
     """
 
     # This is used to to construct self.args, which is passed to 
     # spawnProcess.
     cmd_and_args = List([])
     poll_frequency = Int(100) # in ms
 
     def __init__(self, work_dir=u'.', config=None, **kwargs):
         super(LocalProcessLauncher, self).__init__(
             work_dir=work_dir, config=config, **kwargs
         )
         self.process = None
         self.poller = None
 
     def find_args(self):
         return self.cmd_and_args
 
     def start(self):
         if self.state == 'before':
             self.process = Popen(self.args,
                 stdout=PIPE,stderr=PIPE,stdin=PIPE,
                 env=os.environ,
                 cwd=self.work_dir
             )
             if WINDOWS:
                 self.stdout = forward_read_events(self.process.stdout)
                 self.stderr = forward_read_events(self.process.stderr)
             else:
                 self.stdout = self.process.stdout.fileno()
                 self.stderr = self.process.stderr.fileno()
             self.loop.add_handler(self.stdout, self.handle_stdout, self.loop.READ)
             self.loop.add_handler(self.stderr, self.handle_stderr, self.loop.READ)
             self.poller = ioloop.PeriodicCallback(self.poll, self.poll_frequency, self.loop)
             self.poller.start()
             self.notify_start(self.process.pid)
         else:
             s = 'The process was already started and has state: %r' % self.state
             raise ProcessStateError(s)
 
     def stop(self):
         return self.interrupt_then_kill()
 
     def signal(self, sig):
         if self.state == 'running':
             if WINDOWS and sig != SIGINT:
                 # use Windows tree-kill for better child cleanup
                 check_output(['taskkill', '-pid', str(self.process.pid), '-t', '-f'])
             else:
                 self.process.send_signal(sig)
 
     def interrupt_then_kill(self, delay=2.0):
         """Send INT, wait a delay and then send KILL."""
         try:
             self.signal(SIGINT)
         except Exception:
             self.log.debug("interrupt failed")
             pass
         self.killer  = ioloop.DelayedCallback(lambda : self.signal(SIGKILL), delay*1000, self.loop)
         self.killer.start()
 
     # callbacks, etc:
     
     def handle_stdout(self, fd, events):
         if WINDOWS:
             line = self.stdout.recv()
         else:
             line = self.process.stdout.readline()
         # a stopped process will be readable but return empty strings
         if line:
             self.log.info(line[:-1])
         else:
             self.poll()
     
     def handle_stderr(self, fd, events):
         if WINDOWS:
             line = self.stderr.recv()
         else:
             line = self.process.stderr.readline()
         # a stopped process will be readable but return empty strings
         if line:
             self.log.error(line[:-1])
         else:
             self.poll()
     
     def poll(self):
         status = self.process.poll()
         if status is not None:
             self.poller.stop()
             self.loop.remove_handler(self.stdout)
             self.loop.remove_handler(self.stderr)
             self.notify_stop(dict(exit_code=status, pid=self.process.pid))
         return status
 
 class LocalControllerLauncher(LocalProcessLauncher, ControllerMixin):
     """Launch a controller as a regular external process."""
 
     def find_args(self):
         return self.controller_cmd + self.cluster_args + self.controller_args
 
     def start(self):
         """Start the controller by profile_dir."""
         self.log.info("Starting LocalControllerLauncher: %r" % self.args)
         return super(LocalControllerLauncher, self).start()
 
 
 class LocalEngineLauncher(LocalProcessLauncher, EngineMixin):
     """Launch a single engine as a regular externall process."""
 
     def find_args(self):
         return self.engine_cmd + self.cluster_args + self.engine_args
 
 
 class LocalEngineSetLauncher(LocalEngineLauncher):
     """Launch a set of engines as regular external processes."""
 
     delay = CFloat(0.1, config=True,
         help="""delay (in seconds) between starting each engine after the first.
         This can help force the engines to get their ids in order, or limit
         process flood when starting many engines."""
     )
     
     # launcher class
     launcher_class = LocalEngineLauncher
     
     launchers = Dict()
     stop_data = Dict()
     
     def __init__(self, work_dir=u'.', config=None, **kwargs):
         super(LocalEngineSetLauncher, self).__init__(
             work_dir=work_dir, config=config, **kwargs
         )
         self.stop_data = {}
 
     def start(self, n):
         """Start n engines by profile or profile_dir."""
         dlist = []
         for i in range(n):
             if i > 0:
                 time.sleep(self.delay)
             el = self.launcher_class(work_dir=self.work_dir, config=self.config, log=self.log,
                                     profile_dir=self.profile_dir, cluster_id=self.cluster_id,
             )
             
             # Copy the engine args over to each engine launcher.
             el.engine_cmd = copy.deepcopy(self.engine_cmd)
             el.engine_args = copy.deepcopy(self.engine_args)
             el.on_stop(self._notice_engine_stopped)
             d = el.start()
             if i==0:
                 self.log.info("Starting LocalEngineSetLauncher: %r" % el.args)
             self.launchers[i] = el
             dlist.append(d)
         self.notify_start(dlist)
         return dlist
 
     def find_args(self):
         return ['engine set']
 
     def signal(self, sig):
         dlist = []
         for el in self.launchers.itervalues():
             d = el.signal(sig)
             dlist.append(d)
         return dlist
 
     def interrupt_then_kill(self, delay=1.0):
         dlist = []
         for el in self.launchers.itervalues():
             d = el.interrupt_then_kill(delay)
             dlist.append(d)
         return dlist
 
     def stop(self):
         return self.interrupt_then_kill()
     
     def _notice_engine_stopped(self, data):
         pid = data['pid']
         for idx,el in self.launchers.iteritems():
             if el.process.pid == pid:
                 break
         self.launchers.pop(idx)
         self.stop_data[idx] = data
         if not self.launchers:
             self.notify_stop(self.stop_data)
 
 
 #-----------------------------------------------------------------------------
 # MPIExec launchers
 #-----------------------------------------------------------------------------
 
 
 class MPIExecLauncher(LocalProcessLauncher):
     """Launch an external process using mpiexec."""
 
     mpi_cmd = List(['mpiexec'], config=True,
         help="The mpiexec command to use in starting the process."
     )
     mpi_args = List([], config=True,
         help="The command line arguments to pass to mpiexec."
     )
     program = List(['date'],
         help="The program to start via mpiexec.")
     program_args = List([],
         help="The command line argument to the program."
     )
     n = Int(1)
 
     def find_args(self):
         """Build self.args using all the fields."""
         return self.mpi_cmd + ['-n', str(self.n)] + self.mpi_args + \
                self.program + self.program_args
 
     def start(self, n):
         """Start n instances of the program using mpiexec."""
         self.n = n
         return super(MPIExecLauncher, self).start()
 
 
 class MPIExecControllerLauncher(MPIExecLauncher, ControllerMixin):
     """Launch a controller using mpiexec."""
 
     # alias back to *non-configurable* program[_args] for use in find_args()
     # this way all Controller/EngineSetLaunchers have the same form, rather
     # than *some* having `program_args` and others `controller_args`
     @property
     def program(self):
         return self.controller_cmd
     
     @property
     def program_args(self):
         return self.cluster_args + self.controller_args
 
     def start(self):
         """Start the controller by profile_dir."""
         self.log.info("Starting MPIExecControllerLauncher: %r" % self.args)
         return super(MPIExecControllerLauncher, self).start(1)
 
 
 class MPIExecEngineSetLauncher(MPIExecLauncher, EngineMixin):
     """Launch engines using mpiexec"""
 
     # alias back to *non-configurable* program[_args] for use in find_args()
     # this way all Controller/EngineSetLaunchers have the same form, rather
     # than *some* having `program_args` and others `controller_args`
     @property
     def program(self):
         return self.engine_cmd
     
     @property
     def program_args(self):
         return self.cluster_args + self.engine_args
 
     def start(self, n):
         """Start n engines by profile or profile_dir."""
         self.n = n
         self.log.info('Starting MPIExecEngineSetLauncher: %r' % self.args)
         return super(MPIExecEngineSetLauncher, self).start(n)
 
 #-----------------------------------------------------------------------------
 # SSH launchers
 #-----------------------------------------------------------------------------
 
 # TODO: Get SSH Launcher back to level of sshx in 0.10.2
 
 class SSHLauncher(LocalProcessLauncher):
     """A minimal launcher for ssh.
 
     To be useful this will probably have to be extended to use the ``sshx``
     idea for environment variables.  There could be other things this needs
     as well.
     """
 
     ssh_cmd = List(['ssh'], config=True,
         help="command for starting ssh")
     ssh_args = List(['-tt'], config=True,
         help="args to pass to ssh")
     program = List(['date'],
         help="Program to launch via ssh")
     program_args = List([],
         help="args to pass to remote program")
     hostname = Unicode('', config=True,
         help="hostname on which to launch the program")
     user = Unicode('', config=True,
         help="username for ssh")
     location = Unicode('', config=True,
         help="user@hostname location for ssh in one setting")
 
     def _hostname_changed(self, name, old, new):
         if self.user:
             self.location = u'%s@%s' % (self.user, new)
         else:
             self.location = new
 
     def _user_changed(self, name, old, new):
         self.location = u'%s@%s' % (new, self.hostname)
 
     def find_args(self):
         return self.ssh_cmd + self.ssh_args + [self.location] + \
                self.program + self.program_args
 
     def start(self, hostname=None, user=None):
         if hostname is not None:
             self.hostname = hostname
         if user is not None:
             self.user = 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, ControllerMixin):
 
     # alias back to *non-configurable* program[_args] for use in find_args()
     # this way all Controller/EngineSetLaunchers have the same form, rather
     # than *some* having `program_args` and others `controller_args`
     @property
     def program(self):
         return self.controller_cmd
     
     @property
     def program_args(self):
         return self.cluster_args + self.controller_args
 
 
 class SSHEngineLauncher(SSHLauncher, EngineMixin):
 
     # alias back to *non-configurable* program[_args] for use in find_args()
     # this way all Controller/EngineSetLaunchers have the same form, rather
     # than *some* having `program_args` and others `controller_args`
     @property
     def program(self):
         return self.engine_cmd
     
     @property
     def program_args(self):
         return self.cluster_args + self.engine_args
     
     
 class SSHEngineSetLauncher(LocalEngineSetLauncher):
     launcher_class = SSHEngineLauncher
     engines = Dict(config=True,
         help="""dict of engines to launch.  This is a dict by hostname of ints,
         corresponding to the number of engines to start on that host.""")
     
     def start(self, n):
         """Start engines by profile or profile_dir.
         `n` is ignored, and the `engines` config property is used instead.
         """
         
         dlist = []
         for host, n in self.engines.iteritems():
             if isinstance(n, (tuple, list)):
                 n, args = n
             else:
                 args = copy.deepcopy(self.engine_args)
             
             if '@' in host:
                 user,host = host.split('@',1)
             else:
                 user=None
             for i in range(n):
                 if i > 0:
                     time.sleep(self.delay)
                 el = self.launcher_class(work_dir=self.work_dir, config=self.config, log=self.log,
                                         profile_dir=self.profile_dir, cluster_id=self.cluster_id,
                 )
                 
                 # Copy the engine args over to each engine launcher.
                 el.engine_cmd = self.engine_cmd
                 el.engine_args = args
                 el.on_stop(self._notice_engine_stopped)
                 d = el.start(user=user, hostname=host)
                 if i==0:
                     self.log.info("Starting SSHEngineSetLauncher: %r" % el.args)
                 self.launchers[host+str(i)] = el
                 dlist.append(d)
         self.notify_start(dlist)
         return dlist
     
 
 
 #-----------------------------------------------------------------------------
 # Windows HPC Server 2008 scheduler launchers
 #-----------------------------------------------------------------------------
 
 
 # This is only used on Windows.
 def find_job_cmd():
     if WINDOWS:
         try:
             return find_cmd('job')
         except (FindCmdError, ImportError):
             # ImportError will be raised if win32api is not installed
             return 'job'
     else:
         return 'job'
 
 
 class WindowsHPCLauncher(BaseLauncher):
 
     job_id_regexp = Unicode(r'\d+', config=True,
         help="""A regular expression used to get the job id from the output of the
         submit_command. """
         )
     job_file_name = Unicode(u'ipython_job.xml', config=True,
         help="The filename of the instantiated job script.")
     # The full path to the instantiated job script. This gets made dynamically
     # by combining the work_dir with the job_file_name.
     job_file = Unicode(u'')
     scheduler = Unicode('', config=True,
         help="The hostname of the scheduler to submit the job to.")
     job_cmd = Unicode(find_job_cmd(), config=True,
         help="The command for submitting jobs.")
 
     def __init__(self, work_dir=u'.', config=None, **kwargs):
         super(WindowsHPCLauncher, self).__init__(
             work_dir=work_dir, config=config, **kwargs
         )
 
     @property
     def job_file(self):
         return os.path.join(self.work_dir, self.job_file_name)
 
     def write_job_file(self, n):
         raise NotImplementedError("Implement write_job_file in a subclass.")
 
     def find_args(self):
         return [u'job.exe']
         
     def parse_job_id(self, output):
         """Take the output of the submit command and return the job id."""
         m = re.search(self.job_id_regexp, output)
         if m is not None:
             job_id = m.group()
         else:
             raise LauncherError("Job id couldn't be determined: %s" % output)
         self.job_id = job_id
         self.log.info('Job started with job id: %r' % job_id)
         return job_id
 
     def start(self, n):
         """Start n copies of the process using the Win HPC job scheduler."""
         self.write_job_file(n)
         args = [
             'submit',
             '/jobfile:%s' % self.job_file,
             '/scheduler:%s' % self.scheduler
         ]
         self.log.info("Starting Win HPC Job: %s" % (self.job_cmd + ' ' + ' '.join(args),))
 
         output = check_output([self.job_cmd]+args,
             env=os.environ,
             cwd=self.work_dir,
             stderr=STDOUT
         )
         job_id = self.parse_job_id(output)
         self.notify_start(job_id)
         return job_id
 
     def stop(self):
         args = [
             'cancel',
             self.job_id,
             '/scheduler:%s' % self.scheduler
         ]
         self.log.info("Stopping Win HPC Job: %s" % (self.job_cmd + ' ' + ' '.join(args),))
         try:
             output = check_output([self.job_cmd]+args,
                 env=os.environ,
                 cwd=self.work_dir,
                 stderr=STDOUT
             )
         except:
             output = 'The job already appears to be stoppped: %r' % self.job_id
         self.notify_stop(dict(job_id=self.job_id, output=output))  # Pass the output of the kill cmd
         return output
 
 
 class WindowsHPCControllerLauncher(WindowsHPCLauncher, ClusterAppMixin):
 
     job_file_name = Unicode(u'ipcontroller_job.xml', config=True,
         help="WinHPC xml job file.")
     controller_args = List([], config=False,
         help="extra args to pass to ipcontroller")
 
     def write_job_file(self, n):
         job = IPControllerJob(config=self.config)
 
         t = IPControllerTask(config=self.config)
         # The tasks work directory is *not* the actual work directory of 
         # the controller. It is used as the base path for the stdout/stderr
         # files that the scheduler redirects to.
         t.work_directory = self.profile_dir
         # Add the profile_dir and from self.start().
         t.controller_args.extend(self.cluster_args)
         t.controller_args.extend(self.controller_args)
         job.add_task(t)
 
         self.log.info("Writing job description file: %s" % self.job_file)
         job.write(self.job_file)
 
     @property
     def job_file(self):
         return os.path.join(self.profile_dir, self.job_file_name)
 
     def start(self):
         """Start the controller by profile_dir."""
         return super(WindowsHPCControllerLauncher, self).start(1)
 
 
 class WindowsHPCEngineSetLauncher(WindowsHPCLauncher, ClusterAppMixin):
 
     job_file_name = Unicode(u'ipengineset_job.xml', config=True,
         help="jobfile for ipengines job")
     engine_args = List([], config=False,
         help="extra args to pas to ipengine")
 
     def write_job_file(self, n):
         job = IPEngineSetJob(config=self.config)
 
         for i in range(n):
             t = IPEngineTask(config=self.config)
             # The tasks work directory is *not* the actual work directory of 
             # the engine. It is used as the base path for the stdout/stderr
             # files that the scheduler redirects to.
             t.work_directory = self.profile_dir
             # Add the profile_dir and from self.start().
             t.controller_args.extend(self.cluster_args)
             t.controller_args.extend(self.engine_args)
             job.add_task(t)
 
         self.log.info("Writing job description file: %s" % self.job_file)
         job.write(self.job_file)
 
     @property
     def job_file(self):
         return os.path.join(self.profile_dir, self.job_file_name)
 
     def start(self, n):
         """Start the controller by profile_dir."""
         return super(WindowsHPCEngineSetLauncher, self).start(n)
 
 
 #-----------------------------------------------------------------------------
 # Batch (PBS) system launchers
 #-----------------------------------------------------------------------------
 
 class BatchClusterAppMixin(ClusterAppMixin):
-    """ClusterApp mixin that updates context dict, rather than args"""
-    context = Dict({'profile_dir':'', 'cluster_id':''})
+    """ClusterApp mixin that updates the self.context dict, rather than cl-args."""
     def _profile_dir_changed(self, name, old, new):
         self.context[name] = new
     _cluster_id_changed = _profile_dir_changed
 
+    def _profile_dir_default(self):
+        self.context['profile_dir'] = ''
+        return ''
+    def _cluster_id_default(self):
+        self.context['cluster_id'] = ''
+        return ''
+
+
 class BatchSystemLauncher(BaseLauncher):
     """Launch an external process using a batch system.
 
     This class is designed to work with UNIX batch systems like PBS, LSF,
     GridEngine, etc.  The overall model is that there are different commands
     like qsub, qdel, etc. that handle the starting and stopping of the process.
 
     This class also has the notion of a batch script. The ``batch_template``
     attribute can be set to a string that is a template for the batch script.
     This template is instantiated using string formatting. Thus the template can
     use {n} fot the number of instances. Subclasses can add additional variables
     to the template dict.
     """
 
     # Subclasses must fill these in.  See PBSEngineSet
     submit_command = List([''], config=True,
         help="The name of the command line program used to submit jobs.")
     delete_command = List([''], config=True,
         help="The name of the command line program used to delete jobs.")
     job_id_regexp = Unicode('', config=True,
         help="""A regular expression used to get the job id from the output of the
         submit_command.""")
     batch_template = Unicode('', config=True,
         help="The string that is the batch script template itself.")
     batch_template_file = Unicode(u'', config=True,
         help="The file that contains the batch template.")
     batch_file_name = Unicode(u'batch_script', config=True,
         help="The filename of the instantiated batch script.")
     queue = Unicode(u'', config=True,
         help="The PBS Queue.")
     
     def _queue_changed(self, name, old, new):
         self.context[name] = new
     
     n = Int(1)
     _n_changed = _queue_changed
     
     # not configurable, override in subclasses
     # PBS Job Array regex
     job_array_regexp = Unicode('')
     job_array_template = Unicode('')
     # PBS Queue regex
     queue_regexp = Unicode('')
     queue_template = Unicode('')
     # The default batch template, override in subclasses
     default_template = Unicode('')
     # The full path to the instantiated batch script.
     batch_file = Unicode(u'')
     # the format dict used with batch_template:
     context = Dict()
     # the Formatter instance for rendering the templates:
     formatter = Instance(EvalFormatter, (), {})
 
     
     def find_args(self):
         return self.submit_command + [self.batch_file]
     
     def __init__(self, work_dir=u'.', config=None, **kwargs):
         super(BatchSystemLauncher, self).__init__(
             work_dir=work_dir, config=config, **kwargs
         )
         self.batch_file = os.path.join(self.work_dir, self.batch_file_name)
 
     def parse_job_id(self, output):
         """Take the output of the submit command and return the job id."""
         m = re.search(self.job_id_regexp, output)
         if m is not None:
             job_id = m.group()
         else:
             raise LauncherError("Job id couldn't be determined: %s" % output)
         self.job_id = job_id
         self.log.info('Job submitted with job id: %r' % job_id)
         return job_id
 
     def write_batch_script(self, n):
         """Instantiate and write the batch script to the work_dir."""
         self.n = n
         # first priority is batch_template if set
         if self.batch_template_file and not self.batch_template:
             # second priority is batch_template_file
             with open(self.batch_template_file) as f:
                 self.batch_template = f.read()
         if not self.batch_template:
             # third (last) priority is default_template
             self.batch_template = self.default_template
         
             # add jobarray or queue lines to user-specified template
             # note that this is *only* when user did not specify a template.
             regex = re.compile(self.job_array_regexp)
             # print regex.search(self.batch_template)
             if not regex.search(self.batch_template):
                 self.log.info("adding job array settings to batch script")
                 firstline, rest = self.batch_template.split('\n',1)
                 self.batch_template = u'\n'.join([firstline, self.job_array_template, rest])
         
             regex = re.compile(self.queue_regexp)
             # print regex.search(self.batch_template)
             if self.queue and not regex.search(self.batch_template):
                 self.log.info("adding PBS queue settings to batch script")
                 firstline, rest = self.batch_template.split('\n',1)
                 self.batch_template = u'\n'.join([firstline, self.queue_template, rest])
             
         script_as_string = self.formatter.format(self.batch_template, **self.context)
         self.log.info('Writing instantiated batch script: %s' % self.batch_file)
     
         with open(self.batch_file, 'w') as f:
             f.write(script_as_string)
         os.chmod(self.batch_file, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
 
     def start(self, n):
         """Start n copies of the process using a batch system."""
         # Here we save profile_dir in the context so they
         # can be used in the batch script template as {profile_dir}
         self.write_batch_script(n)
         output = check_output(self.args, env=os.environ)
         
         job_id = self.parse_job_id(output)
         self.notify_start(job_id)
         return job_id
 
     def stop(self):
         output = check_output(self.delete_command+[self.job_id], env=os.environ)
         self.notify_stop(dict(job_id=self.job_id, output=output)) # Pass the output of the kill cmd
         return output
 
 
 class PBSLauncher(BatchSystemLauncher):
     """A BatchSystemLauncher subclass for PBS."""
 
     submit_command = List(['qsub'], config=True,
         help="The PBS submit command ['qsub']")
     delete_command = List(['qdel'], config=True,
         help="The PBS delete command ['qsub']")
     job_id_regexp = Unicode(r'\d+', config=True,
         help="Regular expresion for identifying the job ID [r'\d+']")
     
     batch_file = Unicode(u'')
     job_array_regexp = Unicode('#PBS\W+-t\W+[\w\d\-\$]+')
     job_array_template = Unicode('#PBS -t 1-{n}')
     queue_regexp = Unicode('#PBS\W+-q\W+\$?\w+')
     queue_template = Unicode('#PBS -q {queue}')
 
 
-class PBSControllerLauncher(BatchClusterAppMixin, PBSLauncher):
+class PBSControllerLauncher(PBSLauncher, BatchClusterAppMixin):
     """Launch a controller using PBS."""
 
     batch_file_name = Unicode(u'pbs_controller', config=True,
         help="batch file name for the controller job.")
     default_template= Unicode("""#!/bin/sh
 #PBS -V
 #PBS -N ipcontroller
 %s --log-to-file --profile-dir="{profile_dir}" --cluster-id="{cluster_id}"
 """%(' '.join(ipcontroller_cmd_argv)))
 
     
     def start(self):
         """Start the controller by profile or profile_dir."""
         self.log.info("Starting PBSControllerLauncher: %r" % self.args)
         return super(PBSControllerLauncher, self).start(1)
 
 
-class PBSEngineSetLauncher(BatchClusterAppMixin, PBSLauncher):
+class PBSEngineSetLauncher(PBSLauncher, BatchClusterAppMixin):
     """Launch Engines using PBS"""
     batch_file_name = Unicode(u'pbs_engines', config=True,
         help="batch file name for the engine(s) job.")
     default_template= Unicode(u"""#!/bin/sh
 #PBS -V
 #PBS -N ipengine
 %s --profile-dir="{profile_dir}" --cluster-id="{cluster_id}"
 """%(' '.join(ipengine_cmd_argv)))
 
     def start(self, n):
         """Start n engines by profile or profile_dir."""
         self.log.info('Starting %i engines with PBSEngineSetLauncher: %r' % (n, self.args))
         return super(PBSEngineSetLauncher, self).start(n)
 
 #SGE is very similar to PBS
 
 class SGELauncher(PBSLauncher):
     """Sun GridEngine is a PBS clone with slightly different syntax"""
     job_array_regexp = Unicode('#\$\W+\-t')
     job_array_template = Unicode('#$ -t 1-{n}')
     queue_regexp = Unicode('#\$\W+-q\W+\$?\w+')
     queue_template = Unicode('#$ -q {queue}')
 
-class SGEControllerLauncher(BatchClusterAppMixin, SGELauncher):
+class SGEControllerLauncher(SGELauncher, BatchClusterAppMixin):
     """Launch a controller using SGE."""
 
     batch_file_name = Unicode(u'sge_controller', config=True,
         help="batch file name for the ipontroller job.")
     default_template= Unicode(u"""#$ -V
 #$ -S /bin/sh
 #$ -N ipcontroller
 %s --log-to-file --profile-dir="{profile_dir}" --cluster-id="{cluster_id}"
 """%(' '.join(ipcontroller_cmd_argv)))
 
     def start(self):
         """Start the controller by profile or profile_dir."""
         self.log.info("Starting PBSControllerLauncher: %r" % self.args)
         return super(SGEControllerLauncher, self).start(1)
 
-class SGEEngineSetLauncher(BatchClusterAppMixin, SGELauncher):
+class SGEEngineSetLauncher(SGELauncher, BatchClusterAppMixin):
     """Launch Engines with SGE"""
     batch_file_name = Unicode(u'sge_engines', config=True,
         help="batch file name for the engine(s) job.")
     default_template = Unicode("""#$ -V
 #$ -S /bin/sh
 #$ -N ipengine
 %s --profile-dir="{profile_dir}" --cluster-id="{cluster_id}"
 """%(' '.join(ipengine_cmd_argv)))
 
     def start(self, n):
         """Start n engines by profile or profile_dir."""
         self.log.info('Starting %i engines with SGEEngineSetLauncher: %r' % (n, self.args))
         return super(SGEEngineSetLauncher, self).start(n)
 
 
 # LSF launchers
 
 class LSFLauncher(BatchSystemLauncher):
     """A BatchSystemLauncher subclass for LSF."""
     
     submit_command = List(['bsub'], config=True,
                           help="The PBS submit command ['bsub']")
     delete_command = List(['bkill'], config=True,
                           help="The PBS delete command ['bkill']")
     job_id_regexp = Unicode(r'\d+', config=True,
                             help="Regular expresion for identifying the job ID [r'\d+']")
     
     batch_file = Unicode(u'')
     job_array_regexp = Unicode('#BSUB[ \t]-J+\w+\[\d+-\d+\]')
     job_array_template = Unicode('#BSUB -J ipengine[1-{n}]')
     queue_regexp = Unicode('#BSUB[ \t]+-q[ \t]+\w+')
     queue_template = Unicode('#BSUB -q {queue}')
     
     def start(self, n):
         """Start n copies of the process using LSF batch system.
         This cant inherit from the base class because bsub expects
         to be piped a shell script in order to honor the #BSUB directives :
         bsub < script
         """
         # Here we save profile_dir in the context so they
         # can be used in the batch script template as {profile_dir}
         self.write_batch_script(n)
         #output = check_output(self.args, env=os.environ)
         piped_cmd = self.args[0]+'<\"'+self.args[1]+'\"'
         p = Popen(piped_cmd, shell=True,env=os.environ,stdout=PIPE)
         output,err = p.communicate()
         job_id = self.parse_job_id(output)
         self.notify_start(job_id)
         return job_id
 
 
-class LSFControllerLauncher(BatchClusterAppMixin, LSFLauncher):
+class LSFControllerLauncher(LSFLauncher, BatchClusterAppMixin):
     """Launch a controller using LSF."""
     
     batch_file_name = Unicode(u'lsf_controller', config=True,
                               help="batch file name for the controller job.")
     default_template= Unicode("""#!/bin/sh
     #BSUB -J ipcontroller
     #BSUB -oo ipcontroller.o.%%J 
     #BSUB -eo ipcontroller.e.%%J 
     %s --log-to-file --profile-dir="{profile_dir}" --cluster-id="{cluster_id}"
     """%(' '.join(ipcontroller_cmd_argv)))
     
     def start(self):
         """Start the controller by profile or profile_dir."""
         self.log.info("Starting LSFControllerLauncher: %r" % self.args)
         return super(LSFControllerLauncher, self).start(1)
 
 
-class LSFEngineSetLauncher(BatchClusterAppMixin, LSFLauncher):
+class LSFEngineSetLauncher(LSFLauncher, BatchClusterAppMixin):
     """Launch Engines using LSF"""
     batch_file_name = Unicode(u'lsf_engines', config=True,
                               help="batch file name for the engine(s) job.")
     default_template= Unicode(u"""#!/bin/sh
     #BSUB -oo ipengine.o.%%J 
     #BSUB -eo ipengine.e.%%J 
     %s --profile-dir="{profile_dir}" --cluster-id="{cluster_id}"
     """%(' '.join(ipengine_cmd_argv)))
     
     def start(self, n):
         """Start n engines by profile or profile_dir."""
         self.log.info('Starting %i engines with LSFEngineSetLauncher: %r' % (n, self.args))
         return super(LSFEngineSetLauncher, self).start(n)
 
 
 #-----------------------------------------------------------------------------
 # A launcher for ipcluster itself!
 #-----------------------------------------------------------------------------
 
 
 class IPClusterLauncher(LocalProcessLauncher):
     """Launch the ipcluster program in an external process."""
 
     ipcluster_cmd = List(ipcluster_cmd_argv, config=True,
         help="Popen command for ipcluster")
     ipcluster_args = List(
         ['--clean-logs', '--log-to-file', '--log-level=%i'%logging.INFO], config=True,
         help="Command line arguments to pass to ipcluster.")
     ipcluster_subcommand = Unicode('start')
     ipcluster_n = Int(2)
 
     def find_args(self):
         return self.ipcluster_cmd + [self.ipcluster_subcommand] + \
             ['--n=%i'%self.ipcluster_n] + self.ipcluster_args
 
     def start(self):
         self.log.info("Starting ipcluster: %r" % self.args)
         return super(IPClusterLauncher, self).start()
 
 #-----------------------------------------------------------------------------
 # Collections of launchers
 #-----------------------------------------------------------------------------
 
 local_launchers = [
     LocalControllerLauncher,
     LocalEngineLauncher,
     LocalEngineSetLauncher,
 ]
 mpi_launchers = [
     MPIExecLauncher,
     MPIExecControllerLauncher,
     MPIExecEngineSetLauncher,
 ]
 ssh_launchers = [
     SSHLauncher,
     SSHControllerLauncher,
     SSHEngineLauncher,
     SSHEngineSetLauncher,
 ]
 winhpc_launchers = [
     WindowsHPCLauncher,
     WindowsHPCControllerLauncher,
     WindowsHPCEngineSetLauncher,
 ]
 pbs_launchers = [
     PBSLauncher,
     PBSControllerLauncher,
     PBSEngineSetLauncher,
 ]
 sge_launchers = [
     SGELauncher,
     SGEControllerLauncher,
     SGEEngineSetLauncher,
 ]
 lsf_launchers = [
     LSFLauncher,
     LSFControllerLauncher,
     LSFEngineSetLauncher,
 ]
 all_launchers = local_launchers + mpi_launchers + ssh_launchers + winhpc_launchers\
                 + pbs_launchers + sge_launchers + lsf_launchers