update parallel apps to use ProfileDir

IPython/parallel/apps/clusterdir.py

@@ -1,540 +1,257 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 The IPython cluster directory
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  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 shutil
 import sys
 
 from subprocess import Popen, PIPE
 
-from IPython.config.loader import PyFileConfigLoader, Config
-from IPython.config.configurable import Configurable
-from IPython.config.application import Application
-from IPython.core.crashhandler import CrashHandler
-from IPython.core.newapplication import BaseIPythonApplication
+from IPython.config.loader import Config
 from IPython.core import release
-from IPython.utils.path import (
-    get_ipython_package_dir,
-    get_ipython_dir,
-    expand_path
+from IPython.core.crashhandler import CrashHandler
+from IPython.core.newapplication 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 ClusterDirError(Exception):
-    pass
-
-
 class PIDFileError(Exception):
     pass
 
 
 #-----------------------------------------------------------------------------
-# Class for managing cluster directories
-#-----------------------------------------------------------------------------
-
-class ClusterDir(Configurable):
-    """An object to manage the cluster directory and its resources.
-
-    The cluster directory is used by :command:`ipengine`, 
-    :command:`ipcontroller` and :command:`ipclsuter` to manage the
-    configuration, logging and security of these applications.
-
-    This object knows how to find, create and manage these directories. This
-    should be used by any code that want's to handle cluster directories.
-    """
-
-    security_dir_name = Unicode('security')
-    log_dir_name = Unicode('log')
-    pid_dir_name = Unicode('pid')
-    security_dir = Unicode(u'')
-    log_dir = Unicode(u'')
-    pid_dir = Unicode(u'')
-
-    auto_create = Bool(False,
-        help="""Whether to automatically create the ClusterDirectory if it does
-        not exist""")
-    overwrite = Bool(False,
-        help="""Whether to overwrite existing config files""")
-    location = Unicode(u'', config=True,
-        help="""Set the cluster dir. This overrides the logic used by the
-        `profile` option.""",
-        )
-    profile = Unicode(u'default', config=True,
-        help="""The string name of the profile to be used. This determines the name
-        of the cluster dir as: cluster_<profile>. The default profile is named
-        'default'.  The cluster directory is resolve this way if the
-        `cluster_dir` option is not used."""
-        )
-
-    _location_isset = Bool(False) # flag for detecting multiply set location
-    _new_dir = Bool(False) # flag for whether a new dir was created
-
-    def __init__(self, **kwargs):
-        # make sure auto_create,overwrite are set *before* location
-        for name in ('auto_create', 'overwrite'):
-            v = kwargs.pop(name, None)
-            if v is not None:
-                setattr(self, name, v)
-        super(ClusterDir, self).__init__(**kwargs)
-        if not self.location:
-            self._profile_changed('profile', 'default', self.profile)
-
-    def _location_changed(self, name, old, new):
-        if self._location_isset:
-            raise RuntimeError("Cannot set ClusterDir more than once.")
-        self._location_isset = True
-        if not os.path.isdir(new):
-            if self.auto_create:# or self.config.ClusterDir.auto_create:
-                os.makedirs(new)
-                self._new_dir = True
-            else:
-                raise ClusterDirError('Directory not found: %s' % new)
-        
-        # ensure config files exist:
-        self.copy_all_config_files(overwrite=self.overwrite)
-        self.security_dir = os.path.join(new, self.security_dir_name)
-        self.log_dir = os.path.join(new, self.log_dir_name)
-        self.pid_dir = os.path.join(new, self.pid_dir_name)
-        self.check_dirs()
-
-    def _profile_changed(self, name, old, new):
-        if self._location_isset:
-            raise RuntimeError("ClusterDir already set.  Cannot set by profile.")
-        self.location = os.path.join(get_ipython_dir(), 'cluster_'+new)
-
-    def _log_dir_changed(self, name, old, new):
-        self.check_log_dir()
-
-    def check_log_dir(self):
-        if not os.path.isdir(self.log_dir):
-            os.mkdir(self.log_dir)
-
-    def _security_dir_changed(self, name, old, new):
-        self.check_security_dir()
-
-    def check_security_dir(self):
-        if not os.path.isdir(self.security_dir):
-            os.mkdir(self.security_dir, 0700)
-        os.chmod(self.security_dir, 0700)
-
-    def _pid_dir_changed(self, name, old, new):
-        self.check_pid_dir()
-
-    def check_pid_dir(self):
-        if not os.path.isdir(self.pid_dir):
-            os.mkdir(self.pid_dir, 0700)
-        os.chmod(self.pid_dir, 0700)
-
-    def check_dirs(self):
-        self.check_security_dir()
-        self.check_log_dir()
-        self.check_pid_dir()
-
-    def copy_config_file(self, config_file, path=None, overwrite=False):
-        """Copy a default config file into the active cluster directory.
-
-        Default configuration files are kept in :mod:`IPython.config.default`.
-        This function moves these from that location to the working cluster
-        directory.
-        """
-        if path is None:
-            import IPython.config.default
-            path = IPython.config.default.__file__.split(os.path.sep)[:-1]
-            path = os.path.sep.join(path)
-        src = os.path.join(path, config_file)
-        dst = os.path.join(self.location, config_file)
-        if not os.path.isfile(dst) or overwrite:
-            shutil.copy(src, dst)
-
-    def copy_all_config_files(self, path=None, overwrite=False):
-        """Copy all config files into the active cluster directory."""
-        for f in [u'ipcontroller_config.py', u'ipengine_config.py',
-                  u'ipcluster_config.py']:
-            self.copy_config_file(f, path=path, overwrite=overwrite)
-
-    @classmethod
-    def create_cluster_dir(csl, cluster_dir):
-        """Create a new cluster directory given a full path.
-
-        Parameters
-        ----------
-        cluster_dir : str
-            The full path to the cluster directory.  If it does exist, it will
-            be used.  If not, it will be created.
-        """
-        return ClusterDir(location=cluster_dir)
-
-    @classmethod
-    def create_cluster_dir_by_profile(cls, path, profile=u'default'):
-        """Create a cluster dir by profile name and path.
-
-        Parameters
-        ----------
-        path : str
-            The path (directory) to put the cluster directory in.
-        profile : str
-            The name of the profile.  The name of the cluster directory will
-            be "cluster_<profile>".
-        """
-        if not os.path.isdir(path):
-            raise ClusterDirError('Directory not found: %s' % path)
-        cluster_dir = os.path.join(path, u'cluster_' + profile)
-        return ClusterDir(location=cluster_dir)
-
-    @classmethod
-    def find_cluster_dir_by_profile(cls, ipython_dir, profile=u'default'):
-        """Find an existing cluster dir by profile name, return its ClusterDir.
-
-        This searches through a sequence of paths for a cluster dir.  If it
-        is not found, a :class:`ClusterDirError` exception will be raised.
-
-        The search path algorithm is:
-        1. ``os.getcwd()``
-        2. ``ipython_dir``
-        3. The directories found in the ":" separated 
-           :env:`IPCLUSTER_DIR_PATH` environment variable.
-
-        Parameters
-        ----------
-        ipython_dir : unicode or str
-            The IPython directory to use.
-        profile : unicode or str
-            The name of the profile.  The name of the cluster directory
-            will be "cluster_<profile>".
-        """
-        dirname = u'cluster_' + profile
-        cluster_dir_paths = os.environ.get('IPCLUSTER_DIR_PATH','')
-        if cluster_dir_paths:
-            cluster_dir_paths = cluster_dir_paths.split(':')
-        else:
-            cluster_dir_paths = []
-        paths = [os.getcwd(), ipython_dir] + cluster_dir_paths
-        for p in paths:
-            cluster_dir = os.path.join(p, dirname)
-            if os.path.isdir(cluster_dir):
-                return ClusterDir(location=cluster_dir)
-        else:
-            raise ClusterDirError('Cluster directory not found in paths: %s' % dirname)
-
-    @classmethod
-    def find_cluster_dir(cls, cluster_dir):
-        """Find/create a cluster dir and return its ClusterDir.
-
-        This will create the cluster directory if it doesn't exist.
-
-        Parameters
-        ----------
-        cluster_dir : unicode or str
-            The path of the cluster directory.  This is expanded using
-            :func:`IPython.utils.genutils.expand_path`.
-        """
-        cluster_dir = expand_path(cluster_dir)
-        if not os.path.isdir(cluster_dir):
-            raise ClusterDirError('Cluster directory not found: %s' % cluster_dir)
-        return ClusterDir(location=cluster_dir)
-
-
-#-----------------------------------------------------------------------------
 # Crash handler for this application
 #-----------------------------------------------------------------------------
 
 
 _message_template = """\
 Oops, $self.app_name crashed. We do our best to make it stable, but...
 
 A crash report was automatically generated with the following information:
   - A verbatim copy of the crash traceback.
   - Data on your current $self.app_name configuration.
 
 It was left in the file named:
 \t'$self.crash_report_fname'
 If you can email this file to the developers, the information in it will help
 them in understanding and correcting the problem.
 
 You can mail it to: $self.contact_name at $self.contact_email
 with the subject '$self.app_name Crash Report'.
 
 If you want to do it now, the following command will work (under Unix):
 mail -s '$self.app_name Crash Report' $self.contact_email < $self.crash_report_fname
 
 To ensure accurate tracking of this issue, please file a report about it at:
 $self.bug_tracker
 """
 
-class ClusterDirCrashHandler(CrashHandler):
+class ParallelCrashHandler(CrashHandler):
     """sys.excepthook for IPython itself, leaves a detailed report on disk."""
 
     message_template = _message_template
 
     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(ClusterDirCrashHandler,self).__init__(
+        super(ParallelCrashHandler,self).__init__(
             app, contact_name, contact_email, bug_tracker
         )
 
 
 #-----------------------------------------------------------------------------
 # Main application
 #-----------------------------------------------------------------------------
-base_aliases = {
-    'profile' : "ClusterDir.profile",
-    'cluster_dir' : 'ClusterDir.location',
-    'log_level' : 'ClusterApplication.log_level',
-    'work_dir' : 'ClusterApplication.work_dir',
-    'log_to_file' : 'ClusterApplication.log_to_file',
-    'clean_logs' : 'ClusterApplication.clean_logs',
-    'log_url' : 'ClusterApplication.log_url',
-    'config' : 'ClusterApplication.config_file',
-}
+base_aliases = {}
+base_aliases.update(base_ip_aliases)
+base_aliases.update({
+    'profile_dir' : 'ProfileDir.location',
+    'log_level' : 'BaseParallelApplication.log_level',
+    'work_dir' : 'BaseParallelApplication.work_dir',
+    'log_to_file' : 'BaseParallelApplication.log_to_file',
+    'clean_logs' : 'BaseParallelApplication.clean_logs',
+    'log_url' : 'BaseParallelApplication.log_url',
+})
 
 base_flags = {
-    'debug' : ( {"ClusterApplication" : {"log_level" : logging.DEBUG}}, "set loglevel to DEBUG"),
-    'quiet' : ( {"ClusterApplication" : {"log_level" : logging.CRITICAL}}, "set loglevel to CRITICAL (minimal output)"),
-    'log-to-file' : ( {"ClusterApplication" : {"log_to_file" : True}}, "redirect log output to a file"),
+    'log-to-file' : ({'BaseParallelApplication' : Config({
+                    'log_to_file' : True}),
+            }, "send log output to a file")
 }
-for k,v in base_flags.iteritems():
-    base_flags[k] = (Config(v[0]),v[1])
-
-class ClusterApplication(BaseIPythonApplication):
-    """An application that puts everything into a cluster directory.
-
-    Instead of looking for things in the ipython_dir, this type of application
-    will use its own private directory called the "cluster directory"
-    for things like config files, log files, etc.
+base_flags.update(base_ip_flags)
 
-    The cluster directory is resolved as follows:
+class BaseParallelApplication(BaseIPythonApplication):
+    """The base Application for IPython.parallel apps
 
-    * If the ``cluster_dir`` option is given, it is used.
-    * If ``cluster_dir`` is not given, the application directory is 
-      resolve using the profile name as ``cluster_<profile>``. The search 
-      path for this directory is then i) cwd if it is found there
-      and ii) in ipython_dir otherwise.
+    Principle extensions to BaseIPyythonApplication:
 
-    The config file for the application is to be put in the cluster
-    dir and named the value of the ``config_file_name`` class attribute.
+    * work_dir
+    * remote logging via pyzmq
+    * IOLoop instance
     """
 
-    crash_handler_class = ClusterDirCrashHandler
-    auto_create_cluster_dir = Bool(True, config=True,
-        help="whether to create the cluster_dir if it doesn't exist")
-    cluster_dir = Instance(ClusterDir)
-    classes = [ClusterDir]
+    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.")
 
-    config_file = Unicode(u'', config=True,
-        help="""Path to ip<appname> configuration file.  The default is to use
-         <appname>_config.py, as found by cluster-dir."""
-    )
-    def _config_file_paths_default(self):
-        # don't include profile dir
-        return [ os.getcwdu(), self.ipython_dir ]
-    
-    def _config_file_changed(self, name, old, new):
-        if os.pathsep in new:
-            path, new = new.rsplit(os.pathsep)
-            self.config_file_paths.insert(0, path)
-        self.config_file_name = new
-    
-    config_file_name = Unicode('')
+    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 init_clusterdir(self):
-        """This resolves the cluster directory.
-
-        This tries to find the cluster directory and if successful, it will
-        have done:
-        * Sets ``self.cluster_dir_obj`` to the :class:`ClusterDir` object for 
-          the application.
-        * Sets ``self.cluster_dir`` attribute of the application and config
-          objects.
-
-        The algorithm used for this is as follows:
-        1. Try ``Global.cluster_dir``.
-        2. Try using ``Global.profile``.
-        3. If both of these fail and ``self.auto_create_cluster_dir`` is
-           ``True``, then create the new cluster dir in the IPython directory.
-        4. If all fails, then raise :class:`ClusterDirError`.
-        """
-        try:
-            self.cluster_dir = ClusterDir(auto_create=self.auto_create_cluster_dir, config=self.config)
-        except ClusterDirError as e:
-            self.log.fatal("Error initializing cluster dir: %s"%e)
-            self.log.fatal("A cluster dir must be created before running this command.")
-            self.log.fatal("Do 'ipcluster create -h' or 'ipcluster list -h' for more "
-            "information about creating and listing cluster dirs."
-            )
-            self.exit(1)
-            
-        if self.cluster_dir._new_dir:
-            self.log.info('Creating new cluster dir: %s' % \
-                            self.cluster_dir.location)
-        else:
-            self.log.info('Using existing cluster dir: %s' % \
-                            self.cluster_dir.location)
-        
-        # insert after cwd:
-        self.config_file_paths.insert(1, self.cluster_dir.location)
-    
     def initialize(self, argv=None):
         """initialize the app"""
-        self.init_crash_handler()
-        self.parse_command_line(argv)
-        cl_config = self.config
-        self.init_clusterdir()
-        self.load_config_file()
-        # command-line should *override* config file, but command-line is necessary
-        # to determine clusterdir, etc.
-        self.update_config(cl_config)
+        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.cluster_dir.log_dir
+        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)
 
     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.cluster_dir.pid_dir, self.name + u'.pid')
+        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.cluster_dir.pid_dir, self.name + u'.pid')
+        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.cluster_dir.pid_dir, self.name + u'.pid')
+        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:
                 pid = int(f.read().strip())
                 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/ipclusterapp.py

@@ -1,540 +1,521 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 The ipcluster application.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  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 errno
 import logging
 import os
 import re
 import signal
 
 from subprocess import check_call, CalledProcessError, PIPE
 import zmq
 from zmq.eventloop import ioloop
 
 from IPython.config.application import Application, boolean_flag
 from IPython.config.loader import Config
-from IPython.core.newapplication import BaseIPythonApplication
+from IPython.core.newapplication import BaseIPythonApplication, ProfileDir
 from IPython.utils.importstring import import_item
 from IPython.utils.traitlets import Int, Unicode, Bool, CFloat, Dict, List
 
 from IPython.parallel.apps.clusterdir import (
-    ClusterApplication, ClusterDirError, ClusterDir,
+    BaseParallelApplication,
     PIDFileError,
     base_flags, base_aliases
 )
 
 
 #-----------------------------------------------------------------------------
 # Module level variables
 #-----------------------------------------------------------------------------
 
 
 default_config_file_name = u'ipcluster_config.py'
 
 
 _description = """Start an IPython cluster for parallel computing.
 
 An IPython cluster consists of 1 controller and 1 or more engines.
 This command automates the startup of these processes using a wide
 range of startup methods (SSH, local processes, PBS, mpiexec,
 Windows HPC Server 2008). To start a cluster with 4 engines on your
 local host simply do 'ipcluster start n=4'. For more complex usage
 you will typically do 'ipcluster create profile=mycluster', then edit
 configuration files, followed by 'ipcluster start profile=mycluster n=4'.
 """
 
 
 # Exit codes for ipcluster
 
 # This will be the exit code if the ipcluster appears to be running because
 # a .pid file exists
 ALREADY_STARTED = 10
 
 
 # This will be the exit code if ipcluster stop is run, but there is not .pid
 # file to be found.
 ALREADY_STOPPED = 11
 
 # This will be the exit code if ipcluster engines is run, but there is not .pid
 # file to be found.
 NO_CLUSTER = 12
 
 
 #-----------------------------------------------------------------------------
 # Main application
 #-----------------------------------------------------------------------------
 start_help = """Start an IPython cluster for parallel computing
 
 Start an ipython cluster by its profile name or cluster
 directory. Cluster directories contain configuration, log and
 security related files and are named using the convention
 'cluster_<profile>' and should be creating using the 'start'
 subcommand of 'ipcluster'. If your cluster directory is in 
 the cwd or the ipython directory, you can simply refer to it
 using its profile name, 'ipcluster start n=4 profile=<profile>`,
-otherwise use the 'cluster_dir' option.
+otherwise use the 'profile_dir' option.
 """
 stop_help = """Stop a running IPython cluster
 
 Stop a running ipython cluster by its profile name or cluster
 directory. Cluster directories are named using the convention
 'cluster_<profile>'. If your cluster directory is in 
 the cwd or the ipython directory, you can simply refer to it
 using its profile name, 'ipcluster stop profile=<profile>`, otherwise
-use the 'cluster_dir' option.
+use the 'profile_dir' option.
 """
 engines_help = """Start engines connected to an existing IPython cluster
 
 Start one or more engines to connect to an existing Cluster
 by profile name or cluster directory.
 Cluster directories contain configuration, log and
 security related files and are named using the convention
 'cluster_<profile>' and should be creating using the 'start'
 subcommand of 'ipcluster'. If your cluster directory is in 
 the cwd or the ipython directory, you can simply refer to it
 using its profile name, 'ipcluster engines n=4 profile=<profile>`,
-otherwise use the 'cluster_dir' option.
+otherwise use the 'profile_dir' option.
 """
 create_help = """Create an ipcluster profile by name
 
 Create an ipython cluster directory by its profile name or
 cluster directory path. Cluster directories contain
 configuration, log and security related files and are named
 using the convention 'cluster_<profile>'. By default they are
 located in your ipython directory. Once created, you will
 probably need to edit the configuration files in the cluster
 directory to configure your cluster. Most users will create a
 cluster directory by profile name,
 `ipcluster create profile=mycluster`, which will put the directory
 in `<ipython_dir>/cluster_mycluster`.
 """
 list_help = """List available cluster profiles
 
 List all available clusters, by cluster directory, that can
 be found in the current working directly or in the ipython
 directory. Cluster directories are named using the convention
 'cluster_<profile>'.
 """
 
 
 class IPClusterList(BaseIPythonApplication):
     name = u'ipcluster-list'
     description = list_help
     
     # empty aliases
     aliases=Dict()
     flags = Dict(base_flags)
     
     def _log_level_default(self):
         return 20
     
-    def list_cluster_dirs(self):
+    def list_profile_dirs(self):
         # Find the search paths
-        cluster_dir_paths = os.environ.get('IPCLUSTER_DIR_PATH','')
-        if cluster_dir_paths:
-            cluster_dir_paths = cluster_dir_paths.split(':')
+        profile_dir_paths = os.environ.get('IPYTHON_PROFILE_PATH','')
+        if profile_dir_paths:
+            profile_dir_paths = profile_dir_paths.split(':')
         else:
-            cluster_dir_paths = []
+            profile_dir_paths = []
         
         ipython_dir = self.ipython_dir
         
-        paths = [os.getcwd(), ipython_dir] + cluster_dir_paths
+        paths = [os.getcwd(), ipython_dir] + profile_dir_paths
         paths = list(set(paths))
 
-        self.log.info('Searching for cluster dirs in paths: %r' % paths)
+        self.log.info('Searching for cluster profiles in paths: %r' % paths)
         for path in paths:
             files = os.listdir(path)
             for f in files:
                 full_path = os.path.join(path, f)
-                if os.path.isdir(full_path) and f.startswith('cluster_'):
-                    profile = full_path.split('_')[-1]
+                if os.path.isdir(full_path) and f.startswith('profile_') and \
+                    os.path.isfile(os.path.join(full_path, 'ipcontroller_config.py')):
+                    profile = f.split('_')[-1]
                     start_cmd = 'ipcluster start profile=%s n=4' % profile
                     print start_cmd + " ==> " + full_path
     
     def start(self):
-        self.list_cluster_dirs()
+        self.list_profile_dirs()
+
+
+# `ipcluster create` will be deprecated when `ipython profile create` or equivalent exists
 
 create_flags = {}
 create_flags.update(base_flags)
-create_flags.update(boolean_flag('reset', 'IPClusterCreate.reset', 
+create_flags.update(boolean_flag('reset', 'IPClusterCreate.overwrite', 
                 "reset config files to defaults", "leave existing config files"))
 
-class IPClusterCreate(ClusterApplication):
-    name = u'ipcluster'
+class IPClusterCreate(BaseParallelApplication):
+    name = u'ipcluster-create'
     description = create_help
-    auto_create_cluster_dir = Bool(True,
-        help="whether to create the cluster_dir if it doesn't exist")
+    auto_create = Bool(True)
     config_file_name = Unicode(default_config_file_name)
     
-    reset = Bool(False, config=True,
-        help="Whether to reset config files as part of 'create'."
-        )
-    
     flags = Dict(create_flags)
     
-    aliases = Dict(dict(profile='ClusterDir.profile'))
+    aliases = Dict(dict(profile='BaseIPythonApplication.profile'))
     
-    classes = [ClusterDir]
+    classes = [ProfileDir]
     
-    def init_clusterdir(self):
-        super(IPClusterCreate, self).init_clusterdir()
-        self.log.info('Copying default config files to cluster directory '
-        '[overwrite=%r]' % (self.reset,))
-        self.cluster_dir.copy_all_config_files(overwrite=self.reset)
-    
-    def initialize(self, argv=None):
-        self.parse_command_line(argv)
-        self.init_clusterdir()
     
 stop_aliases = dict(
     signal='IPClusterStop.signal',
-    profile='ClusterDir.profile',
-    cluster_dir='ClusterDir.location',
+    profile='BaseIPythonApplication.profile',
+    profile_dir='ProfileDir.location',
 )
 
-class IPClusterStop(ClusterApplication):
+class IPClusterStop(BaseParallelApplication):
     name = u'ipcluster'
     description = stop_help
-    auto_create_cluster_dir = Bool(False)
     config_file_name = Unicode(default_config_file_name)
     
     signal = Int(signal.SIGINT, config=True,
         help="signal to use for stopping processes.")
         
     aliases = Dict(stop_aliases)
     
-    def init_clusterdir(self):
-        try:
-            super(IPClusterStop, self).init_clusterdir()
-        except ClusterDirError as e:
-            self.log.fatal("Failed ClusterDir init: %s"%e)
-            self.exit(1)
-    
     def start(self):
         """Start the app for the stop subcommand."""
         try:
             pid = self.get_pid_from_file()
         except PIDFileError:
             self.log.critical(
                 'Could not read pid file, cluster is probably not running.'
             )
             # Here I exit with a unusual exit status that other processes
             # can watch for to learn how I existed.
             self.remove_pid_file()
             self.exit(ALREADY_STOPPED)
         
         if not self.check_pid(pid):
             self.log.critical(
                 'Cluster [pid=%r] is not running.' % pid
             )
             self.remove_pid_file()
             # Here I exit with a unusual exit status that other processes
             # can watch for to learn how I existed.
             self.exit(ALREADY_STOPPED)
             
         elif os.name=='posix':
             sig = self.signal
             self.log.info(
                 "Stopping cluster [pid=%r] with [signal=%r]" % (pid, sig)
             )
             try:
                 os.kill(pid, sig)
             except OSError:
                 self.log.error("Stopping cluster failed, assuming already dead.",
                     exc_info=True)
                 self.remove_pid_file()
         elif os.name=='nt':
             try:
                 # kill the whole tree
                 p = check_call(['taskkill', '-pid', str(pid), '-t', '-f'], stdout=PIPE,stderr=PIPE)
             except (CalledProcessError, OSError):
                 self.log.error("Stopping cluster failed, assuming already dead.",
                     exc_info=True)
             self.remove_pid_file()
     
 engine_aliases = {}
 engine_aliases.update(base_aliases)
 engine_aliases.update(dict(
     n='IPClusterEngines.n',
     elauncher = 'IPClusterEngines.engine_launcher_class',
 ))
-class IPClusterEngines(ClusterApplication):
+class IPClusterEngines(BaseParallelApplication):
 
     name = u'ipcluster'
     description = engines_help
     usage = None
     config_file_name = Unicode(default_config_file_name)
     default_log_level = logging.INFO
-    auto_create_cluster_dir = Bool(False)
     classes = List()
     def _classes_default(self):
         from IPython.parallel.apps import launcher
         launchers = launcher.all_launchers
         eslaunchers = [ l for l in launchers if 'EngineSet' in l.__name__]
-        return [ClusterDir]+eslaunchers
+        return [ProfileDir]+eslaunchers
     
     n = Int(2, config=True,
         help="The number of engines to start.")
 
     engine_launcher_class = Unicode('LocalEngineSetLauncher',
         config=True,
         help="The class for launching a set of Engines."
         )
     daemonize = Bool(False, config=True,
         help='Daemonize the ipcluster program. This implies --log-to-file')
 
     def _daemonize_changed(self, name, old, new):
         if new:
             self.log_to_file = True
 
     aliases = Dict(engine_aliases)
     # flags = Dict(flags)
     _stopping = False
 
     def initialize(self, argv=None):
         super(IPClusterEngines, self).initialize(argv)
         self.init_signal()
         self.init_launchers()
     
     def init_launchers(self):
         self.engine_launcher = self.build_launcher(self.engine_launcher_class)
         self.engine_launcher.on_stop(lambda r: self.loop.stop())
     
     def init_signal(self):
         # Setup signals
         signal.signal(signal.SIGINT, self.sigint_handler)
     
     def build_launcher(self, clsname):
         """import and instantiate a Launcher based on importstring"""
         if '.' not in clsname:
             # not a module, presume it's the raw name in apps.launcher
             clsname = 'IPython.parallel.apps.launcher.'+clsname
         # print repr(clsname)
         klass = import_item(clsname)
 
         launcher = klass(
-            work_dir=self.cluster_dir.location, config=self.config, logname=self.log.name
+            work_dir=self.profile_dir.location, config=self.config, logname=self.log.name
         )
         return launcher
     
     def start_engines(self):
         self.log.info("Starting %i engines"%self.n)
         self.engine_launcher.start(
             self.n,
-            cluster_dir=self.cluster_dir.location
+            profile_dir=self.profile_dir.location
         )
 
     def stop_engines(self):
         self.log.info("Stopping Engines...")
         if self.engine_launcher.running:
             d = self.engine_launcher.stop()
             return d
         else:
             return None
 
     def stop_launchers(self, r=None):
         if not self._stopping:
             self._stopping = True
             self.log.error("IPython cluster: stopping")
             self.stop_engines()
             # Wait a few seconds to let things shut down.
             dc = ioloop.DelayedCallback(self.loop.stop, 4000, self.loop)
             dc.start()
 
     def sigint_handler(self, signum, frame):
         self.log.debug("SIGINT received, stopping launchers...")
         self.stop_launchers()
         
     def start_logging(self):
         # Remove old log files of the controller and engine
         if self.clean_logs:
-            log_dir = self.cluster_dir.log_dir
+            log_dir = self.profile_dir.log_dir
             for f in os.listdir(log_dir):
                 if re.match(r'ip(engine|controller)z-\d+\.(log|err|out)',f):
                     os.remove(os.path.join(log_dir, f))
         # This will remove old log files for ipcluster itself
-        # super(IPClusterApp, self).start_logging()
+        # super(IPBaseParallelApplication, self).start_logging()
 
     def start(self):
         """Start the app for the engines subcommand."""
         self.log.info("IPython cluster: started")
         # First see if the cluster is already running
         
         # Now log and daemonize
         self.log.info(
             'Starting engines with [daemon=%r]' % self.daemonize
         )
         # TODO: Get daemonize working on Windows or as a Windows Server.
         if self.daemonize:
             if os.name=='posix':
                 from twisted.scripts._twistd_unix import daemonize
                 daemonize()
 
         dc = ioloop.DelayedCallback(self.start_engines, 0, self.loop)
         dc.start()
         # Now write the new pid file AFTER our new forked pid is active.
         # self.write_pid_file()
         try:
             self.loop.start()
         except KeyboardInterrupt:
             pass
         except zmq.ZMQError as e:
             if e.errno == errno.EINTR:
                 pass
             else:
                 raise
 
 start_aliases = {}
 start_aliases.update(engine_aliases)
 start_aliases.update(dict(
     delay='IPClusterStart.delay',
     clean_logs='IPClusterStart.clean_logs',
 ))
 
 class IPClusterStart(IPClusterEngines):
 
     name = u'ipcluster'
     description = start_help
     default_log_level = logging.INFO
-    auto_create_cluster_dir = Bool(True, config=True,
-        help="whether to create the cluster_dir if it doesn't exist")
+    auto_create = Bool(True, config=True,
+        help="whether to create the profile_dir if it doesn't exist")
     classes = List()
     def _classes_default(self,):
         from IPython.parallel.apps import launcher
-        return [ClusterDir]+launcher.all_launchers
+        return [ProfileDir]+launcher.all_launchers
     
     clean_logs = Bool(True, config=True, 
         help="whether to cleanup old logs before starting")
 
     delay = CFloat(1., config=True,
         help="delay (in s) between starting the controller and the engines")
 
     controller_launcher_class = Unicode('LocalControllerLauncher',
         config=True,
         help="The class for launching a Controller."
         )
     reset = Bool(False, config=True,
         help="Whether to reset config files as part of '--create'."
         )
     
     # flags = Dict(flags)
     aliases = Dict(start_aliases)
 
     def init_launchers(self):
         self.controller_launcher = self.build_launcher(self.controller_launcher_class)
         self.engine_launcher = self.build_launcher(self.engine_launcher_class)
         self.controller_launcher.on_stop(self.stop_launchers)
     
     def start_controller(self):
         self.controller_launcher.start(
-            cluster_dir=self.cluster_dir.location
+            profile_dir=self.profile_dir.location
         )
         
     def stop_controller(self):
         # self.log.info("In stop_controller")
         if self.controller_launcher and self.controller_launcher.running:
             return self.controller_launcher.stop()
 
     def stop_launchers(self, r=None):
         if not self._stopping:
             self.stop_controller()
             super(IPClusterStart, self).stop_launchers()
 
     def start(self):
         """Start the app for the start subcommand."""
         # First see if the cluster is already running
         try:
             pid = self.get_pid_from_file()
         except PIDFileError:
             pass
         else:
             if self.check_pid(pid):
                 self.log.critical(
                     'Cluster is already running with [pid=%s]. '
                     'use "ipcluster stop" to stop the cluster.' % pid
                 )
                 # Here I exit with a unusual exit status that other processes
                 # can watch for to learn how I existed.
                 self.exit(ALREADY_STARTED)
             else:
                 self.remove_pid_file()
                 
 
         # Now log and daemonize
         self.log.info(
             'Starting ipcluster with [daemon=%r]' % self.daemonize
         )
         # TODO: Get daemonize working on Windows or as a Windows Server.
         if self.daemonize:
             if os.name=='posix':
                 from twisted.scripts._twistd_unix import daemonize
                 daemonize()
 
         dc = ioloop.DelayedCallback(self.start_controller, 0, self.loop)
         dc.start()
         dc = ioloop.DelayedCallback(self.start_engines, 1000*self.delay, self.loop)
         dc.start()
         # Now write the new pid file AFTER our new forked pid is active.
         self.write_pid_file()
         try:
             self.loop.start()
         except KeyboardInterrupt:
             pass
         except zmq.ZMQError as e:
             if e.errno == errno.EINTR:
                 pass
             else:
                 raise
         finally:
             self.remove_pid_file()
 
 base='IPython.parallel.apps.ipclusterapp.IPCluster'
 
-class IPClusterApp(Application):
+class IPBaseParallelApplication(Application):
     name = u'ipcluster'
     description = _description
     
     subcommands = {'create' : (base+'Create', create_help),
                 'list' : (base+'List', list_help),
                 'start' : (base+'Start', start_help),
                 'stop' : (base+'Stop', stop_help),
                 'engines' : (base+'Engines', engines_help),
     }
     
     # no aliases or flags for parent App
     aliases = Dict()
     flags = Dict()
     
     def start(self):
         if self.subapp is None:
             print "No subcommand specified! Must specify one of: %s"%(self.subcommands.keys())
             print
             self.print_subcommands()
             self.exit(1)
         else:
             return self.subapp.start()
 
 def launch_new_instance():
     """Create and run the IPython cluster."""
-    app = IPClusterApp()
+    app = IPBaseParallelApplication()
     app.initialize()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()
 

IPython/parallel/apps/ipcontrollerapp.py

@@ -1,404 +1,399 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 The IPython controller application.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  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 copy
 import os
-import logging
 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.loader import Config
-
-from IPython.parallel import factory
+from IPython.core.newapplication import ProfileDir
 
 from IPython.parallel.apps.clusterdir import (
-    ClusterDir,
-    ClusterApplication,
+    BaseParallelApplication,
     base_flags
-    # ClusterDirConfigLoader
 )
 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.parallel.streamsession import StreamSession
 from IPython.parallel.controller.heartmonitor import HeartMonitor
-from IPython.parallel.controller.hub import Hub, HubFactory
+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,disambiguate_ip_address, split_url
+from IPython.parallel.util import signal_children, split_url
 
 # 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 "cluster_<profile>". See the `profile`
-and `cluster_dir` options for details.
+and `profile_dir` options for details.
 """
 
 
 
 
 #-----------------------------------------------------------------------------
 # 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' : Config({'db_class' : 'IPython.parallel.controller.sqlitedb.SQLiteDB'})},
                     'use the SQLiteDB backend'),
     'mongodb' : ({'HubFactory' : Config({'db_class' : 'IPython.parallel.controller.mongodb.MongoDB'})},
                     'use the MongoDB backend'),
     'dictdb' : ({'HubFactory' : Config({'db_class' : 'IPython.parallel.controller.dictdb.DictDB'})},
                     'use the in-memory DictDB backend'),
     'reuse' : ({'IPControllerApp' : Config({'reuse_files' : True})},
                     'reuse existing json connection files')
 })
 
 flags.update()
 
-class IPControllerApp(ClusterApplication):
+class IPControllerApp(BaseParallelApplication):
 
     name = u'ipcontroller'
     description = _description
     config_file_name = Unicode(default_config_file_name)
-    classes = [ClusterDir, StreamSession, HubFactory, TaskScheduler, HeartMonitor, SQLiteDB] + maybe_mongo
+    classes = [ProfileDir, StreamSession, 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""")
     
-    auto_create_cluster_dir = Bool(True, config=True,
-        help="Whether to create cluster_dir if it exists.")
     reuse_files = Bool(False, config=True,
         help='Whether to reuse existing json connection files [default: False]'
     )
     secure = Bool(True, config=True,
         help='Whether to use exec_keys for extra authentication [default: True]'
     )
     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""",
     )
     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',
         )
 
     # 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(dict(
-        config = 'IPControllerApp.config_file',
-        # file = 'IPControllerApp.url_file',
         log_level = 'IPControllerApp.log_level',
         log_url = 'IPControllerApp.log_url',
         reuse_files = 'IPControllerApp.reuse_files',
         secure = 'IPControllerApp.secure',
         ssh = 'IPControllerApp.ssh_server',
         use_threads = 'IPControllerApp.use_threads',
         import_statements = 'IPControllerApp.import_statements',
         location = 'IPControllerApp.location',
 
         ident = 'StreamSession.session',
         user = 'StreamSession.username',
         exec_key = 'StreamSession.keyfile',
 
         url = 'HubFactory.url',
         ip = 'HubFactory.ip',
         transport = 'HubFactory.transport',
         port = 'HubFactory.regport',
 
         ping = 'HeartMonitor.period',
 
         scheme = 'TaskScheduler.scheme_name',
         hwm = 'TaskScheduler.hwm',
 
 
-        profile = "ClusterDir.profile",
-        cluster_dir = 'ClusterDir.location',
+        profile = "BaseIPythonApplication.profile",
+        profile_dir = 'ProfileDir.location',
         
     ))
     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:
                 location = socket.gethostbyname_ex(socket.gethostname())[2][-1]
             cdict['location'] = location
-        fname = os.path.join(self.cluster_dir.security_dir, fname)
+        fname = os.path.join(self.profile_dir.security_dir, fname)
         with open(fname, 'w') 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.cluster_dir.security_dir, 'ipcontroller-engine.json')) as f:
+        with open(os.path.join(self.profile_dir.security_dir, 'ipcontroller-engine.json')) as f:
             cfg = json.loads(f.read())
         key = c.StreamSession.key = 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']
         
         # load client config
-        with open(os.path.join(self.cluster_dir.security_dir, 'ipcontroller-client.json')) as f:
+        with open(os.path.join(self.profile_dir.security_dir, 'ipcontroller-client.json')) 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
         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.cluster_dir.security_dir, self.exec_key)
+            # 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.StreamSession.key = key
         else:
             key = c.StreamSession.key = ''
         
         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('ipcontroller-client.json', cdict)
             edict = cdict
             edict['url']="%s://%s:%s"%((f.client_transport, f.client_ip, f.regport))
             self.save_connection_dict('ipcontroller-engine.json', 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, 'N/A','iopub')
         q.bind_in(hub.client_info['iopub'])
         q.bind_out(hub.engine_info['iopub'])
         q.setsockopt_out(zmq.SUBSCRIBE, '')
         q.connect_mon(hub.monitor_url)
         q.daemon=True
         children.append(q)
 
         # Multiplexer Queue (in a Process)
         q = mq(zmq.XREP, zmq.XREP, zmq.PUB, 'in', 'out')
         q.bind_in(hub.client_info['mux'])
         q.setsockopt_in(zmq.IDENTITY, '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.XREP, zmq.XREP, zmq.PUB, 'incontrol', 'outcontrol')
         q.bind_in(hub.client_info['control'])
         q.setsockopt_in(zmq.IDENTITY, '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.XREP, zmq.XREQ, zmq.PUB, 'intask', 'outtask')
             # q.setsockopt_out(zmq.HWM, hub.hwm)
             q.bind_in(hub.client_info['task'][1])
             q.setsockopt_in(zmq.IDENTITY, '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))
             q = Process(target=launch_scheduler, args=sargs, kwargs=kwargs)
             q.daemon=True
             children.append(q)
 
     
     def save_urls(self):
         """save the registration urls to files."""
         c = self.config
         
-        sec_dir = self.cluster_dir.security_dir
+        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"""
     app = IPControllerApp()
     app.initialize()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()

IPython/parallel/apps/ipengineapp.py

@@ -1,277 +1,272 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 The IPython engine application
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  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 zmq
 from zmq.eventloop import ioloop
 
+from IPython.core.newapplication import ProfileDir
 from IPython.parallel.apps.clusterdir import (
-    ClusterApplication,
-    ClusterDir,
-    # ClusterDirConfigLoader
+    BaseParallelApplication,
 )
 from IPython.zmq.log import EnginePUBHandler
 
 from IPython.config.configurable import Configurable
 from IPython.parallel.streamsession import StreamSession
 from IPython.parallel.engine.engine import EngineFactory
 from IPython.parallel.engine.streamkernel import Kernel
 from IPython.parallel.util import disambiguate_url
 
 from IPython.utils.importstring import import_item
 from IPython.utils.traitlets import Bool, Unicode, Dict, List
 
 
 #-----------------------------------------------------------------------------
 # 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 "cluster_<profile>".
-See the `profile` and `cluster_dir` options for details.
+See the `profile` and `profile_dir` options for details.
 """
 
 
 #-----------------------------------------------------------------------------
 # 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):
         # 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
 #-----------------------------------------------------------------------------
 
 
-class IPEngineApp(ClusterApplication):
+class IPEngineApp(BaseParallelApplication):
 
     app_name = Unicode(u'ipengine')
     description = Unicode(_description)
     config_file_name = Unicode(default_config_file_name)
-    classes = List([ClusterDir, StreamSession, EngineFactory, Kernel, MPI])
-
-    auto_create_cluster_dir = Bool(False,
-        help="whether to create the cluster_dir if it doesn't exist")
+    classes = List([ProfileDir, StreamSession, 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 `cluster_dir` options.""",
+        resolved using the `profile` or `profile_dir` options.""",
         )
 
     url_file_name = Unicode(u'ipcontroller-engine.json')
     log_url = Unicode('', config=True,
         help="""The URL for the iploggerapp instance, for forwarding
         logging to a central location.""")
 
     aliases = Dict(dict(
-        config = 'IPEngineApp.config_file',
         file = 'IPEngineApp.url_file',
         c = 'IPEngineApp.startup_command',
         s = 'IPEngineApp.startup_script',
 
         ident = 'StreamSession.session',
         user = 'StreamSession.username',
         exec_key = 'StreamSession.keyfile',
 
         url = 'EngineFactory.url',
         ip = 'EngineFactory.ip',
         transport = 'EngineFactory.transport',
         port = 'EngineFactory.regport',
         location = 'EngineFactory.location',
 
         timeout = 'EngineFactory.timeout',
 
-        profile = "ClusterDir.profile",
-        cluster_dir = 'ClusterDir.location',
+        profile = "IPEngineApp.profile",
+        profile_dir = 'ProfileDir.location',
 
         mpi = 'MPI.use',
 
         log_level = 'IPEngineApp.log_level',
         log_url = 'IPEngineApp.log_url'
     ))
 
     # 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.cluster_dir, 
+    #             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 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.config
         # Find the actual controller key file
         if not self.url_file:
             self.url_file = os.path.join(
-                self.cluster_dir.security_dir,
+                self.profile_dir.security_dir,
                 self.url_file_name
             )
     def init_engine(self):
         # This is the working dir by now.
         sys.path.insert(0, '')
         config = self.config
         # print config
         self.find_url_file()
 
         # if os.path.exists(config.Global.key_file) and config.Global.secure:
         #     config.SessionFactory.exec_key = config.Global.key_file
         if os.path.exists(self.url_file):
             with open(self.url_file) as f:
                 d = json.loads(f.read())
                 for k,v in d.iteritems():
                     if isinstance(v, unicode):
                         d[k] = v.encode()
             if d['exec_key']:
                 config.StreamSession.key = d['exec_key']
             d['url'] = disambiguate_url(d['url'], d['location'])
             config.EngineFactory.url = d['url']
             config.EngineFactory.location = d['location']
         
         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()
     app.initialize()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()
 

IPython/parallel/apps/iploggerapp.py

@@ -1,97 +1,96 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 A simple IPython logger application
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 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 os
 import sys
 
 import zmq
 
+from IPython.core.newapplication import ProfileDir
 from IPython.utils.traitlets import Bool, Dict, Unicode
 
 from IPython.parallel.apps.clusterdir import (
-    ClusterApplication,
-    ClusterDir,
+    BaseParallelApplication,
     base_aliases
 )
 from IPython.parallel.apps.logwatcher import LogWatcher
 
 #-----------------------------------------------------------------------------
 # Module level variables
 #-----------------------------------------------------------------------------
 
 #: The default config file name for this application
 default_config_file_name = u'iplogger_config.py'
 
 _description = """Start an IPython logger for parallel computing.
 
 IPython controllers and engines (and your own processes) can broadcast log messages
 by registering a `zmq.log.handlers.PUBHandler` with the `logging` module. The
 logger 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 "cluster_<profile>".
-See the `profile` and `cluster_dir` options for details.
+See the `profile` and `profile_dir` options for details.
 """
 
 
 #-----------------------------------------------------------------------------
 # Main application
 #-----------------------------------------------------------------------------
 aliases = {}
 aliases.update(base_aliases)
 aliases.update(dict(url='LogWatcher.url', topics='LogWatcher.topics'))
 
-class IPLoggerApp(ClusterApplication):
+class IPLoggerApp(BaseParallelApplication):
 
     name = u'iploggerz'
     description = _description
     config_file_name = Unicode(default_config_file_name)
-    auto_create_cluster_dir = Bool(False)
     
-    classes = [LogWatcher, ClusterDir]
+    classes = [LogWatcher, ProfileDir]
     aliases = Dict(aliases)
 
     def initialize(self, argv=None):
         super(IPLoggerApp, self).initialize(argv)
         self.init_watcher()
     
     def init_watcher(self):
         try:
             self.watcher = LogWatcher(config=self.config, logname=self.log.name)
         except:
             self.log.error("Couldn't start the LogWatcher", exc_info=True)
             self.exit(1)
         self.log.info("Listening for log messages on %r"%self.watcher.url)
         
 
     def start(self):
         self.watcher.start()
         try:
             self.watcher.loop.start()
         except KeyboardInterrupt:
             self.log.critical("Logging Interrupted, shutting down...\n")
 
 
 def launch_new_instance():
     """Create and run the IPython LogWatcher"""
     app = IPLoggerApp()
     app.initialize()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()
 

IPython/parallel/apps/launcher.py

@@ -1,1070 +1,1070 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 Facilities for launching IPython processes asynchronously.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  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
 
 # 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.external import Itpl
 # from IPython.config.configurable import Configurable
 from IPython.utils.traitlets import Any, Int, List, Unicode, Dict, Instance
 from IPython.utils.path import get_ipython_module_path
 from IPython.utils.process import find_cmd, pycmd2argv, FindCmdError
 
 from IPython.parallel.factory import LoggingFactory
 
 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(LoggingFactory):
     """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 cluster_dir, but may not be. any work_dir
+    # 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.
 
         This must return a deferred that fires with information about the
         process starting (like a pid, job id, etc.).
         """
         raise NotImplementedError('start must be implemented in a subclass')
 
     def stop(self):
         """Stop the process and notify observers of stopping.
 
         This must return a deferred that fires with information about the
         processing stopping, like errors that occur while the process is
         attempting to be shut down. This deferred won't fire when the process
         actually stops. To observe the actual process stopping, see
         :func:`observe_stop`.
         """
         raise NotImplementedError('stop must be implemented in a subclass')
 
     def on_stop(self, f):
         """Get a deferred that will fire when the process stops.
 
         The deferred will fire with data that contains information about
         the exit status of the process.
         """
         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 all the deferreds from :func:`observe_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.
 
         Return a semi-meaningless deferred after signaling the process.
 
         Parameters
         ----------
         sig : str or int
             'KILL', 'INT', etc., or any signal number
         """
         raise NotImplementedError('signal must be implemented in a subclass')
 
 
 #-----------------------------------------------------------------------------
 # 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.start_deferred = 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):
     """Launch a controller as a regular external process."""
 
     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,
         help="""command-line args to pass to ipcontroller""")
 
     def find_args(self):
         return self.controller_cmd + self.controller_args
 
-    def start(self, cluster_dir):
-        """Start the controller by cluster_dir."""
-        self.controller_args.extend(['cluster_dir=%s'%cluster_dir])
-        self.cluster_dir = unicode(cluster_dir)
+    def start(self, profile_dir):
+        """Start the controller by profile_dir."""
+        self.controller_args.extend(['profile_dir=%s'%profile_dir])
+        self.profile_dir = unicode(profile_dir)
         self.log.info("Starting LocalControllerLauncher: %r" % self.args)
         return super(LocalControllerLauncher, self).start()
 
 
 class LocalEngineLauncher(LocalProcessLauncher):
     """Launch a single engine as a regular externall process."""
 
     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,
         help="command-line arguments to pass to ipengine"
     )
 
     def find_args(self):
         return self.engine_cmd + self.engine_args
 
-    def start(self, cluster_dir):
-        """Start the engine by cluster_dir."""
-        self.engine_args.extend(['cluster_dir=%s'%cluster_dir])
-        self.cluster_dir = unicode(cluster_dir)
+    def start(self, profile_dir):
+        """Start the engine by profile_dir."""
+        self.engine_args.extend(['profile_dir=%s'%profile_dir])
+        self.profile_dir = unicode(profile_dir)
         return super(LocalEngineLauncher, self).start()
 
 
 class LocalEngineSetLauncher(BaseLauncher):
     """Launch a set of engines as regular external processes."""
 
     # Command line arguments for ipengine.
     engine_args = List(
         ['--log-to-file','log_level=%i'%logging.INFO], config=True,
         help="command-line arguments to pass to ipengine"
     )
     # 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, cluster_dir):
-        """Start n engines by profile or cluster_dir."""
-        self.cluster_dir = unicode(cluster_dir)
+    def start(self, n, profile_dir):
+        """Start n engines by profile or profile_dir."""
+        self.profile_dir = unicode(profile_dir)
         dlist = []
         for i in range(n):
             el = self.launcher_class(work_dir=self.work_dir, config=self.config, logname=self.log.name)
             # Copy the engine args over to each engine launcher.
             el.engine_args = copy.deepcopy(self.engine_args)
             el.on_stop(self._notice_engine_stopped)
-            d = el.start(cluster_dir)
+            d = el.start(profile_dir)
             if i==0:
                 self.log.info("Starting LocalEngineSetLauncher: %r" % el.args)
             self.launchers[i] = el
             dlist.append(d)
         self.notify_start(dlist)
         # The consumeErrors here could be dangerous
         # dfinal = gatherBoth(dlist, consumeErrors=True)
         # dfinal.addCallback(self.notify_start)
         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)
         # dfinal = gatherBoth(dlist, consumeErrors=True)
         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)
         # dfinal = gatherBoth(dlist, consumeErrors=True)
         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'], config=True,
         help="The program to start via mpiexec.")
     program_args = List([], config=True,
         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):
     """Launch a controller using mpiexec."""
 
     controller_cmd = List(ipcontroller_cmd_argv, config=True,
         help="Popen command to launch the Contropper"
     )
     controller_args = List(['--log-to-file','log_level=%i'%logging.INFO], config=True,
         help="Command line arguments to pass to ipcontroller."
     )
     n = Int(1)
 
-    def start(self, cluster_dir):
-        """Start the controller by cluster_dir."""
-        self.controller_args.extend(['cluster_dir=%s'%cluster_dir])
-        self.cluster_dir = unicode(cluster_dir)
+    def start(self, profile_dir):
+        """Start the controller by profile_dir."""
+        self.controller_args.extend(['profile_dir=%s'%profile_dir])
+        self.profile_dir = unicode(profile_dir)
         self.log.info("Starting MPIExecControllerLauncher: %r" % self.args)
         return super(MPIExecControllerLauncher, self).start(1)
 
     def find_args(self):
         return self.mpi_cmd + ['-n', self.n] + self.mpi_args + \
                self.controller_cmd + self.controller_args
 
 
 class MPIExecEngineSetLauncher(MPIExecLauncher):
 
     program = List(ipengine_cmd_argv, config=True,
         help="Popen command for ipengine"
     )
     program_args = List(
         ['--log-to-file','log_level=%i'%logging.INFO], config=True,
         help="Command line arguments for ipengine."
     )
     n = Int(1)
 
-    def start(self, n, cluster_dir):
-        """Start n engines by profile or cluster_dir."""
-        self.program_args.extend(['cluster_dir=%s'%cluster_dir])
-        self.cluster_dir = unicode(cluster_dir)
+    def start(self, n, profile_dir):
+        """Start n engines by profile or profile_dir."""
+        self.program_args.extend(['profile_dir=%s'%profile_dir])
+        self.profile_dir = unicode(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 working again.
 
 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'], config=True,
         help="Program to launch via ssh")
     program_args = List([], config=True,
         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, cluster_dir, hostname=None, user=None):
-        self.cluster_dir = unicode(cluster_dir)
+    def start(self, profile_dir, hostname=None, user=None):
+        self.profile_dir = unicode(profile_dir)
         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):
 
     program = List(ipcontroller_cmd_argv, config=True,
         help="remote ipcontroller command.")
     program_args = List(['--reuse-files', '--log-to-file','log_level=%i'%logging.INFO], config=True,
         help="Command line arguments to ipcontroller.")
 
 
 class SSHEngineLauncher(SSHLauncher):
     program = List(ipengine_cmd_argv, config=True,
         help="remote ipengine command.")
     # Command line arguments for ipengine.
     program_args = List(
         ['--log-to-file','log_level=%i'%logging.INFO], config=True,
         help="Command line arguments to ipengine."
     )
     
 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, cluster_dir):
-        """Start engines by profile or cluster_dir.
+    def start(self, n, profile_dir):
+        """Start engines by profile or profile_dir.
         `n` is ignored, and the `engines` config property is used instead.
         """
         
-        self.cluster_dir = unicode(cluster_dir)
+        self.profile_dir = unicode(profile_dir)
         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):
                 el = self.launcher_class(work_dir=self.work_dir, config=self.config, logname=self.log.name)
                 
                 # Copy the engine args over to each engine launcher.
                 i
                 el.program_args = args
                 el.on_stop(self._notice_engine_stopped)
-                d = el.start(cluster_dir, user=user, hostname=host)
+                d = el.start(profile_dir, 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),))
         # Twisted will raise DeprecationWarnings if we try to pass unicode to this
         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):
 
     job_file_name = Unicode(u'ipcontroller_job.xml', config=True,
         help="WinHPC xml job file.")
     extra_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.cluster_dir
-        # Add the cluster_dir and from self.start().
+        t.work_directory = self.profile_dir
+        # Add the profile_dir and from self.start().
         t.controller_args.extend(self.extra_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.cluster_dir, self.job_file_name)
+        return os.path.join(self.profile_dir, self.job_file_name)
 
-    def start(self, cluster_dir):
-        """Start the controller by cluster_dir."""
-        self.extra_args = ['cluster_dir=%s'%cluster_dir]
-        self.cluster_dir = unicode(cluster_dir)
+    def start(self, profile_dir):
+        """Start the controller by profile_dir."""
+        self.extra_args = ['profile_dir=%s'%profile_dir]
+        self.profile_dir = unicode(profile_dir)
         return super(WindowsHPCControllerLauncher, self).start(1)
 
 
 class WindowsHPCEngineSetLauncher(WindowsHPCLauncher):
 
     job_file_name = Unicode(u'ipengineset_job.xml', config=True,
         help="jobfile for ipengines job")
     extra_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.cluster_dir
-            # Add the cluster_dir and from self.start().
+            t.work_directory = self.profile_dir
+            # Add the profile_dir and from self.start().
             t.engine_args.extend(self.extra_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.cluster_dir, self.job_file_name)
+        return os.path.join(self.profile_dir, self.job_file_name)
 
-    def start(self, n, cluster_dir):
-        """Start the controller by cluster_dir."""
-        self.extra_args = ['cluster_dir=%s'%cluster_dir]
-        self.cluster_dir = unicode(cluster_dir)
+    def start(self, n, profile_dir):
+        """Start the controller by profile_dir."""
+        self.extra_args = ['profile_dir=%s'%profile_dir]
+        self.profile_dir = unicode(profile_dir)
         return super(WindowsHPCEngineSetLauncher, self).start(n)
 
 
 #-----------------------------------------------------------------------------
 # Batch (PBS) system launchers
 #-----------------------------------------------------------------------------
 
 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 Itpl. 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.")
     
     # 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()
 
     
     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.context['n'] = n
         self.context['queue'] = self.queue
         print self.context
         # 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
         
         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 = Itpl.itplns(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, cluster_dir):
+    def start(self, n, profile_dir):
         """Start n copies of the process using a batch system."""
-        # Here we save profile and cluster_dir in the context so they
+        # Here we save profile and profile_dir in the context so they
         # can be used in the batch script template as ${profile} and
-        # ${cluster_dir}
-        self.context['cluster_dir'] = cluster_dir
-        self.cluster_dir = unicode(cluster_dir)
+        # ${profile_dir}
+        self.context['profile_dir'] = profile_dir
+        self.profile_dir = unicode(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(PBSLauncher):
     """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 cluster_dir $cluster_dir
+%s --log-to-file profile_dir $profile_dir
 """%(' '.join(ipcontroller_cmd_argv)))
 
-    def start(self, cluster_dir):
-        """Start the controller by profile or cluster_dir."""
+    def start(self, profile_dir):
+        """Start the controller by profile or profile_dir."""
         self.log.info("Starting PBSControllerLauncher: %r" % self.args)
-        return super(PBSControllerLauncher, self).start(1, cluster_dir)
+        return super(PBSControllerLauncher, self).start(1, profile_dir)
 
 
 class PBSEngineSetLauncher(PBSLauncher):
     """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 cluster_dir $cluster_dir
+%s profile_dir $profile_dir
 """%(' '.join(ipengine_cmd_argv)))
 
-    def start(self, n, cluster_dir):
-        """Start n engines by profile or cluster_dir."""
+    def start(self, n, profile_dir):
+        """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, cluster_dir)
+        return super(PBSEngineSetLauncher, self).start(n, profile_dir)
 
 #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\W+[\w\d\-\$]+')
     job_array_template = Unicode('#$$ -t 1-$n')
     queue_regexp = Unicode('#$$\W+-q\W+\$?\w+')
     queue_template = Unicode('#$$ -q $queue')
 
 class SGEControllerLauncher(SGELauncher):
     """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 cluster_dir=$cluster_dir
+%s --log-to-file profile_dir=$profile_dir
 """%(' '.join(ipcontroller_cmd_argv)))
 
-    def start(self, cluster_dir):
-        """Start the controller by profile or cluster_dir."""
+    def start(self, profile_dir):
+        """Start the controller by profile or profile_dir."""
         self.log.info("Starting PBSControllerLauncher: %r" % self.args)
-        return super(PBSControllerLauncher, self).start(1, cluster_dir)
+        return super(PBSControllerLauncher, self).start(1, profile_dir)
 
 class SGEEngineSetLauncher(SGELauncher):
     """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 cluster_dir=$cluster_dir
+%s profile_dir=$profile_dir
 """%(' '.join(ipengine_cmd_argv)))
 
-    def start(self, n, cluster_dir):
-        """Start n engines by profile or cluster_dir."""
+    def start(self, n, profile_dir):
+        """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, cluster_dir)
+        return super(SGEEngineSetLauncher, self).start(n, profile_dir)
 
 
 #-----------------------------------------------------------------------------
 # 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,
 ]
 all_launchers = local_launchers + mpi_launchers + ssh_launchers + winhpc_launchers\
                 + pbs_launchers + sge_launchers
  No newline at end of file

IPython/parallel/client/client.py

@@ -1,1356 +1,1356 b''
 """A semi-synchronous Client for the ZMQ cluster"""
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010  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 os
 import json
 import time
 import warnings
 from datetime import datetime
 from getpass import getpass
 from pprint import pprint
 
 pjoin = os.path.join
 
 import zmq
 # from zmq.eventloop import ioloop, zmqstream
 
 from IPython.utils.path import get_ipython_dir
 from IPython.utils.traitlets import (HasTraits, Int, Instance, Unicode,
                                     Dict, List, Bool, Set)
 from IPython.external.decorator import decorator
 from IPython.external.ssh import tunnel
 
 from IPython.parallel import error
 from IPython.parallel import streamsession as ss
 from IPython.parallel import util
 
 from .asyncresult import AsyncResult, AsyncHubResult
-from IPython.parallel.apps.clusterdir import ClusterDir, ClusterDirError
+from IPython.core.newapplication import ProfileDir, ProfileDirError
 from .view import DirectView, LoadBalancedView
 
 #--------------------------------------------------------------------------
 # Decorators for Client methods
 #--------------------------------------------------------------------------
 
 @decorator
 def spin_first(f, self, *args, **kwargs):
     """Call spin() to sync state prior to calling the method."""
     self.spin()
     return f(self, *args, **kwargs)
 
 
 #--------------------------------------------------------------------------
 # Classes
 #--------------------------------------------------------------------------
 
 class Metadata(dict):
     """Subclass of dict for initializing metadata values.
     
     Attribute access works on keys.
     
     These objects have a strict set of keys - errors will raise if you try
     to add new keys.
     """
     def __init__(self, *args, **kwargs):
         dict.__init__(self)
         md = {'msg_id' : None,
               'submitted' : None,
               'started' : None,
               'completed' : None,
               'received' : None,
               'engine_uuid' : None,
               'engine_id' : None,
               'follow' : None,
               'after' : None,
               'status' : None,
 
               'pyin' : None,
               'pyout' : None,
               'pyerr' : None,
               'stdout' : '',
               'stderr' : '',
             }
         self.update(md)
         self.update(dict(*args, **kwargs))
     
     def __getattr__(self, key):
         """getattr aliased to getitem"""
         if key in self.iterkeys():
             return self[key]
         else:
             raise AttributeError(key)
 
     def __setattr__(self, key, value):
         """setattr aliased to setitem, with strict"""
         if key in self.iterkeys():
             self[key] = value
         else:
             raise AttributeError(key)
     
     def __setitem__(self, key, value):
         """strict static key enforcement"""
         if key in self.iterkeys():
             dict.__setitem__(self, key, value)
         else:
             raise KeyError(key)
         
 
 class Client(HasTraits):
     """A semi-synchronous client to the IPython ZMQ cluster
     
     Parameters
     ----------
     
     url_or_file : bytes; zmq url or path to ipcontroller-client.json
         Connection information for the Hub's registration.  If a json connector
         file is given, then likely no further configuration is necessary.
         [Default: use profile]
     profile : bytes
         The name of the Cluster profile to be used to find connector information.
         [Default: 'default']
     context : zmq.Context
         Pass an existing zmq.Context instance, otherwise the client will create its own.
     username : bytes
         set username to be passed to the Session object
     debug : bool
         flag for lots of message printing for debug purposes
  
     #-------------- ssh related args ----------------
     # These are args for configuring the ssh tunnel to be used
     # credentials are used to forward connections over ssh to the Controller
     # Note that the ip given in `addr` needs to be relative to sshserver
     # The most basic case is to leave addr as pointing to localhost (127.0.0.1),
     # and set sshserver as the same machine the Controller is on. However, 
     # the only requirement is that sshserver is able to see the Controller
     # (i.e. is within the same trusted network).
     
     sshserver : str
         A string of the form passed to ssh, i.e. 'server.tld' or '[e-mail]:port'
         If keyfile or password is specified, and this is not, it will default to
         the ip given in addr.
     sshkey : str; path to public ssh key file
         This specifies a key to be used in ssh login, default None.
         Regular default ssh keys will be used without specifying this argument.
     password : str 
         Your ssh password to sshserver. Note that if this is left None,
         you will be prompted for it if passwordless key based login is unavailable.
     paramiko : bool
         flag for whether to use paramiko instead of shell ssh for tunneling.
         [default: True on win32, False else]
     
     ------- exec authentication args -------
     If even localhost is untrusted, you can have some protection against
     unauthorized execution by using a key.  Messages are still sent
     as cleartext, so if someone can snoop your loopback traffic this will
     not help against malicious attacks.
     
     exec_key : str
         an authentication key or file containing a key
         default: None
     
     
     Attributes
     ----------
     
     ids : list of int engine IDs
         requesting the ids attribute always synchronizes
         the registration state. To request ids without synchronization,
         use semi-private _ids attributes.
     
     history : list of msg_ids
         a list of msg_ids, keeping track of all the execution
         messages you have submitted in order.
     
     outstanding : set of msg_ids
         a set of msg_ids that have been submitted, but whose
         results have not yet been received.
     
     results : dict
         a dict of all our results, keyed by msg_id
     
     block : bool
         determines default behavior when block not specified
         in execution methods
     
     Methods
     -------
     
     spin
         flushes incoming results and registration state changes
         control methods spin, and requesting `ids` also ensures up to date
     
     wait
         wait on one or more msg_ids
     
     execution methods
         apply
         legacy: execute, run
     
     data movement
         push, pull, scatter, gather
     
     query methods
         queue_status, get_result, purge, result_status
     
     control methods
         abort, shutdown
     
     """
     
     
     block = Bool(False)
     outstanding = Set()
     results = Instance('collections.defaultdict', (dict,))
     metadata = Instance('collections.defaultdict', (Metadata,))
     history = List()
     debug = Bool(False)
     profile=Unicode('default')
     
     _outstanding_dict = Instance('collections.defaultdict', (set,))
     _ids = List()
     _connected=Bool(False)
     _ssh=Bool(False)
     _context = Instance('zmq.Context')
     _config = Dict()
     _engines=Instance(util.ReverseDict, (), {})
     # _hub_socket=Instance('zmq.Socket')
     _query_socket=Instance('zmq.Socket')
     _control_socket=Instance('zmq.Socket')
     _iopub_socket=Instance('zmq.Socket')
     _notification_socket=Instance('zmq.Socket')
     _mux_socket=Instance('zmq.Socket')
     _task_socket=Instance('zmq.Socket')
     _task_scheme=Unicode()
     _closed = False
     _ignored_control_replies=Int(0)
     _ignored_hub_replies=Int(0)
     
-    def __init__(self, url_or_file=None, profile='default', cluster_dir=None, ipython_dir=None,
+    def __init__(self, url_or_file=None, profile='default', profile_dir=None, ipython_dir=None,
             context=None, username=None, debug=False, exec_key=None,
             sshserver=None, sshkey=None, password=None, paramiko=None,
             timeout=10
             ):
         super(Client, self).__init__(debug=debug, profile=profile)
         if context is None:
             context = zmq.Context.instance()
         self._context = context
             
         
-        self._setup_cluster_dir(profile, cluster_dir, ipython_dir)
+        self._setup_profile_dir(profile, profile_dir, ipython_dir)
         if self._cd is not None:
             if url_or_file is None:
                 url_or_file = pjoin(self._cd.security_dir, 'ipcontroller-client.json')
         assert url_or_file is not None, "I can't find enough information to connect to a hub!"\
             " Please specify at least one of url_or_file or profile."
         
         try:
             util.validate_url(url_or_file)
         except AssertionError:
             if not os.path.exists(url_or_file):
                 if self._cd:
                     url_or_file = os.path.join(self._cd.security_dir, url_or_file)
                 assert os.path.exists(url_or_file), "Not a valid connection file or url: %r"%url_or_file
             with open(url_or_file) as f:
                 cfg = json.loads(f.read())
         else:
             cfg = {'url':url_or_file}
         
         # sync defaults from args, json:
         if sshserver:
             cfg['ssh'] = sshserver
         if exec_key:
             cfg['exec_key'] = exec_key
         exec_key = cfg['exec_key']
         sshserver=cfg['ssh']
         url = cfg['url']
         location = cfg.setdefault('location', None)
         cfg['url'] = util.disambiguate_url(cfg['url'], location)
         url = cfg['url']
         
         self._config = cfg
         
         self._ssh = bool(sshserver or sshkey or password)
         if self._ssh and sshserver is None:
             # default to ssh via localhost
             sshserver = url.split('://')[1].split(':')[0]
         if self._ssh and password is None:
             if tunnel.try_passwordless_ssh(sshserver, sshkey, paramiko):
                 password=False
             else:
                 password = getpass("SSH Password for %s: "%sshserver)
         ssh_kwargs = dict(keyfile=sshkey, password=password, paramiko=paramiko)
         if exec_key is not None and os.path.isfile(exec_key):
             arg = 'keyfile'
         else:
             arg = 'key'
         key_arg = {arg:exec_key}
         if username is None:
             self.session = ss.StreamSession(**key_arg)
         else:
             self.session = ss.StreamSession(username=username, **key_arg)
         self._query_socket = self._context.socket(zmq.XREQ)
         self._query_socket.setsockopt(zmq.IDENTITY, self.session.session)
         if self._ssh:
             tunnel.tunnel_connection(self._query_socket, url, sshserver, **ssh_kwargs)
         else:
             self._query_socket.connect(url)
         
         self.session.debug = self.debug
         
         self._notification_handlers = {'registration_notification' : self._register_engine,
                                     'unregistration_notification' : self._unregister_engine,
                                     'shutdown_notification' : lambda msg: self.close(),
                                     }
         self._queue_handlers = {'execute_reply' : self._handle_execute_reply,
                                 'apply_reply' : self._handle_apply_reply}
         self._connect(sshserver, ssh_kwargs, timeout)
         
     def __del__(self):
         """cleanup sockets, but _not_ context."""
         self.close()
     
-    def _setup_cluster_dir(self, profile, cluster_dir, ipython_dir):
+    def _setup_profile_dir(self, profile, profile_dir, ipython_dir):
         if ipython_dir is None:
             ipython_dir = get_ipython_dir()
-        if cluster_dir is not None:
+        if profile_dir is not None:
             try:
-                self._cd = ClusterDir.find_cluster_dir(cluster_dir)
+                self._cd = ProfileDir.find_profile_dir(profile_dir)
                 return
-            except ClusterDirError:
+            except ProfileDirError:
                 pass
         elif profile is not None:
             try:
-                self._cd = ClusterDir.find_cluster_dir_by_profile(
+                self._cd = ProfileDir.find_profile_dir_by_name(
                     ipython_dir, profile)
                 return
-            except ClusterDirError:
+            except ProfileDirError:
                 pass
         self._cd = None
     
     def _update_engines(self, engines):
         """Update our engines dict and _ids from a dict of the form: {id:uuid}."""
         for k,v in engines.iteritems():
             eid = int(k)
             self._engines[eid] = bytes(v) # force not unicode
             self._ids.append(eid)
         self._ids = sorted(self._ids)
         if sorted(self._engines.keys()) != range(len(self._engines)) and \
                         self._task_scheme == 'pure' and self._task_socket:
             self._stop_scheduling_tasks()
     
     def _stop_scheduling_tasks(self):
         """Stop scheduling tasks because an engine has been unregistered
         from a pure ZMQ scheduler.
         """
         self._task_socket.close()
         self._task_socket = None
         msg = "An engine has been unregistered, and we are using pure " +\
               "ZMQ task scheduling.  Task farming will be disabled."
         if self.outstanding:
             msg += " If you were running tasks when this happened, " +\
                    "some `outstanding` msg_ids may never resolve."
         warnings.warn(msg, RuntimeWarning)
     
     def _build_targets(self, targets):
         """Turn valid target IDs or 'all' into two lists:
         (int_ids, uuids).
         """
         if not self._ids:
             # flush notification socket if no engines yet, just in case
             if not self.ids:
                 raise error.NoEnginesRegistered("Can't build targets without any engines")
         
         if targets is None:
             targets = self._ids
         elif isinstance(targets, str):
             if targets.lower() == 'all':
                 targets = self._ids
             else:
                 raise TypeError("%r not valid str target, must be 'all'"%(targets))
         elif isinstance(targets, int):
             if targets < 0:
                 targets = self.ids[targets]
             if targets not in self._ids:
                 raise IndexError("No such engine: %i"%targets)
             targets = [targets]
         
         if isinstance(targets, slice):
             indices = range(len(self._ids))[targets]
             ids = self.ids
             targets = [ ids[i] for i in indices ]
         
         if not isinstance(targets, (tuple, list, xrange)):
             raise TypeError("targets by int/slice/collection of ints only, not %s"%(type(targets)))
             
         return [self._engines[t] for t in targets], list(targets)
     
     def _connect(self, sshserver, ssh_kwargs, timeout):
         """setup all our socket connections to the cluster. This is called from
         __init__."""
         
         # Maybe allow reconnecting?
         if self._connected:
             return
         self._connected=True
 
         def connect_socket(s, url):
             url = util.disambiguate_url(url, self._config['location'])
             if self._ssh:
                 return tunnel.tunnel_connection(s, url, sshserver, **ssh_kwargs)
             else:
                 return s.connect(url)
             
         self.session.send(self._query_socket, 'connection_request')
         r,w,x = zmq.select([self._query_socket],[],[], timeout)
         if not r:
             raise error.TimeoutError("Hub connection request timed out")
         idents,msg = self.session.recv(self._query_socket,mode=0)
         if self.debug:
             pprint(msg)
         msg = ss.Message(msg)
         content = msg.content
         self._config['registration'] = dict(content)
         if content.status == 'ok':
             if content.mux:
                 self._mux_socket = self._context.socket(zmq.XREQ)
                 self._mux_socket.setsockopt(zmq.IDENTITY, self.session.session)
                 connect_socket(self._mux_socket, content.mux)
             if content.task:
                 self._task_scheme, task_addr = content.task
                 self._task_socket = self._context.socket(zmq.XREQ)
                 self._task_socket.setsockopt(zmq.IDENTITY, self.session.session)
                 connect_socket(self._task_socket, task_addr)
             if content.notification:
                 self._notification_socket = self._context.socket(zmq.SUB)
                 connect_socket(self._notification_socket, content.notification)
                 self._notification_socket.setsockopt(zmq.SUBSCRIBE, b'')
             # if content.query:
             #     self._query_socket = self._context.socket(zmq.XREQ)
             #     self._query_socket.setsockopt(zmq.IDENTITY, self.session.session)
             #     connect_socket(self._query_socket, content.query)
             if content.control:
                 self._control_socket = self._context.socket(zmq.XREQ)
                 self._control_socket.setsockopt(zmq.IDENTITY, self.session.session)
                 connect_socket(self._control_socket, content.control)
             if content.iopub:
                 self._iopub_socket = self._context.socket(zmq.SUB)
                 self._iopub_socket.setsockopt(zmq.SUBSCRIBE, b'')
                 self._iopub_socket.setsockopt(zmq.IDENTITY, self.session.session)
                 connect_socket(self._iopub_socket, content.iopub)
             self._update_engines(dict(content.engines))
         else:
             self._connected = False
             raise Exception("Failed to connect!")
     
     #--------------------------------------------------------------------------
     # handlers and callbacks for incoming messages
     #--------------------------------------------------------------------------
     
     def _unwrap_exception(self, content):
         """unwrap exception, and remap engine_id to int."""
         e = error.unwrap_exception(content)
         # print e.traceback
         if e.engine_info:
             e_uuid = e.engine_info['engine_uuid']
             eid = self._engines[e_uuid]
             e.engine_info['engine_id'] = eid
         return e
     
     def _extract_metadata(self, header, parent, content):
         md = {'msg_id' : parent['msg_id'],
               'received' : datetime.now(),
               'engine_uuid' : header.get('engine', None),
               'follow' : parent.get('follow', []),
               'after' : parent.get('after', []),
               'status' : content['status'],
             }
         
         if md['engine_uuid'] is not None:
             md['engine_id'] = self._engines.get(md['engine_uuid'], None)
         
         if 'date' in parent:
             md['submitted'] = datetime.strptime(parent['date'], util.ISO8601)
         if 'started' in header:
             md['started'] = datetime.strptime(header['started'], util.ISO8601)
         if 'date' in header:
             md['completed'] = datetime.strptime(header['date'], util.ISO8601)
         return md
     
     def _register_engine(self, msg):
         """Register a new engine, and update our connection info."""
         content = msg['content']
         eid = content['id']
         d = {eid : content['queue']}
         self._update_engines(d)
 
     def _unregister_engine(self, msg):
         """Unregister an engine that has died."""
         content = msg['content']
         eid = int(content['id'])
         if eid in self._ids:
             self._ids.remove(eid)
             uuid = self._engines.pop(eid)
             
             self._handle_stranded_msgs(eid, uuid)
                 
         if self._task_socket and self._task_scheme == 'pure':
             self._stop_scheduling_tasks()
     
     def _handle_stranded_msgs(self, eid, uuid):
         """Handle messages known to be on an engine when the engine unregisters.
         
         It is possible that this will fire prematurely - that is, an engine will
         go down after completing a result, and the client will be notified
         of the unregistration and later receive the successful result.
         """
         
         outstanding = self._outstanding_dict[uuid]
         
         for msg_id in list(outstanding):
             if msg_id in self.results:
                 # we already 
                 continue
             try:
                 raise error.EngineError("Engine %r died while running task %r"%(eid, msg_id))
             except:
                 content = error.wrap_exception()
             # build a fake message:
             parent = {}
             header = {}
             parent['msg_id'] = msg_id
             header['engine'] = uuid
             header['date'] = datetime.now().strftime(util.ISO8601)
             msg = dict(parent_header=parent, header=header, content=content)
             self._handle_apply_reply(msg)
     
     def _handle_execute_reply(self, msg):
         """Save the reply to an execute_request into our results.
         
         execute messages are never actually used. apply is used instead.
         """
         
         parent = msg['parent_header']
         msg_id = parent['msg_id']
         if msg_id not in self.outstanding:
             if msg_id in self.history:
                 print ("got stale result: %s"%msg_id)
             else:
                 print ("got unknown result: %s"%msg_id)
         else:
             self.outstanding.remove(msg_id)
         self.results[msg_id] = self._unwrap_exception(msg['content'])
     
     def _handle_apply_reply(self, msg):
         """Save the reply to an apply_request into our results."""
         parent = msg['parent_header']
         msg_id = parent['msg_id']
         if msg_id not in self.outstanding:
             if msg_id in self.history:
                 print ("got stale result: %s"%msg_id)
                 print self.results[msg_id]
                 print msg
             else:
                 print ("got unknown result: %s"%msg_id)
         else:
             self.outstanding.remove(msg_id)
         content = msg['content']
         header = msg['header']
         
         # construct metadata:
         md = self.metadata[msg_id]
         md.update(self._extract_metadata(header, parent, content))
         # is this redundant?
         self.metadata[msg_id] = md
         
         e_outstanding = self._outstanding_dict[md['engine_uuid']]
         if msg_id in e_outstanding:
             e_outstanding.remove(msg_id)
         
         # construct result:
         if content['status'] == 'ok':
             self.results[msg_id] = util.unserialize_object(msg['buffers'])[0]
         elif content['status'] == 'aborted':
             self.results[msg_id] = error.TaskAborted(msg_id)
         elif content['status'] == 'resubmitted':
             # TODO: handle resubmission
             pass
         else:
             self.results[msg_id] = self._unwrap_exception(content)
     
     def _flush_notifications(self):
         """Flush notifications of engine registrations waiting
         in ZMQ queue."""
         msg = self.session.recv(self._notification_socket, mode=zmq.NOBLOCK)
         while msg is not None:
             if self.debug:
                 pprint(msg)
             msg = msg[-1]
             msg_type = msg['msg_type']
             handler = self._notification_handlers.get(msg_type, None)
             if handler is None:
                 raise Exception("Unhandled message type: %s"%msg.msg_type)
             else:
                 handler(msg)
             msg = self.session.recv(self._notification_socket, mode=zmq.NOBLOCK)
     
     def _flush_results(self, sock):
         """Flush task or queue results waiting in ZMQ queue."""
         msg = self.session.recv(sock, mode=zmq.NOBLOCK)
         while msg is not None:
             if self.debug:
                 pprint(msg)
             msg = msg[-1]
             msg_type = msg['msg_type']
             handler = self._queue_handlers.get(msg_type, None)
             if handler is None:
                 raise Exception("Unhandled message type: %s"%msg.msg_type)
             else:
                 handler(msg)
             msg = self.session.recv(sock, mode=zmq.NOBLOCK)
     
     def _flush_control(self, sock):
         """Flush replies from the control channel waiting
         in the ZMQ queue.
         
         Currently: ignore them."""
         if self._ignored_control_replies <= 0:
             return
         msg = self.session.recv(sock, mode=zmq.NOBLOCK)
         while msg is not None:
             self._ignored_control_replies -= 1
             if self.debug:
                 pprint(msg)
             msg = self.session.recv(sock, mode=zmq.NOBLOCK)
     
     def _flush_ignored_control(self):
         """flush ignored control replies"""
         while self._ignored_control_replies > 0:
             self.session.recv(self._control_socket)
             self._ignored_control_replies -= 1
     
     def _flush_ignored_hub_replies(self):
         msg = self.session.recv(self._query_socket, mode=zmq.NOBLOCK)
         while msg is not None:
             msg = self.session.recv(self._query_socket, mode=zmq.NOBLOCK)
     
     def _flush_iopub(self, sock):
         """Flush replies from the iopub channel waiting
         in the ZMQ queue.
         """
         msg = self.session.recv(sock, mode=zmq.NOBLOCK)
         while msg is not None:
             if self.debug:
                 pprint(msg)
             msg = msg[-1]
             parent = msg['parent_header']
             msg_id = parent['msg_id']
             content = msg['content']
             header = msg['header']
             msg_type = msg['msg_type']
             
             # init metadata:
             md = self.metadata[msg_id]
             
             if msg_type == 'stream':
                 name = content['name']
                 s = md[name] or ''
                 md[name] = s + content['data']
             elif msg_type == 'pyerr':
                 md.update({'pyerr' : self._unwrap_exception(content)})
             elif msg_type == 'pyin':
                 md.update({'pyin' : content['code']})
             else:
                 md.update({msg_type : content.get('data', '')})
             
             # reduntant?
             self.metadata[msg_id] = md
             
             msg = self.session.recv(sock, mode=zmq.NOBLOCK)
     
     #--------------------------------------------------------------------------
     # len, getitem
     #--------------------------------------------------------------------------
     
     def __len__(self):
         """len(client) returns # of engines."""
         return len(self.ids)
     
     def __getitem__(self, key):
         """index access returns DirectView multiplexer objects
         
         Must be int, slice, or list/tuple/xrange of ints"""
         if not isinstance(key, (int, slice, tuple, list, xrange)):
             raise TypeError("key by int/slice/iterable of ints only, not %s"%(type(key)))
         else:
             return self.direct_view(key)
     
     #--------------------------------------------------------------------------
     # Begin public methods
     #--------------------------------------------------------------------------
     
     @property
     def ids(self):
         """Always up-to-date ids property."""
         self._flush_notifications()
         # always copy:
         return list(self._ids)
         
     def close(self):
         if self._closed:
             return
         snames = filter(lambda n: n.endswith('socket'), dir(self))
         for socket in map(lambda name: getattr(self, name), snames):
             if isinstance(socket, zmq.Socket) and not socket.closed:
                 socket.close()
         self._closed = True
     
     def spin(self):
         """Flush any registration notifications and execution results
         waiting in the ZMQ queue.
         """
         if self._notification_socket:
             self._flush_notifications()
         if self._mux_socket:
             self._flush_results(self._mux_socket)
         if self._task_socket:
             self._flush_results(self._task_socket)
         if self._control_socket:
             self._flush_control(self._control_socket)
         if self._iopub_socket:
             self._flush_iopub(self._iopub_socket)
         if self._query_socket:
             self._flush_ignored_hub_replies()
     
     def wait(self, jobs=None, timeout=-1):
         """waits on one or more `jobs`, for up to `timeout` seconds.
         
         Parameters
         ----------
         
         jobs : int, str, or list of ints and/or strs, or one or more AsyncResult objects
                 ints are indices to self.history
                 strs are msg_ids
                 default: wait on all outstanding messages
         timeout : float
                 a time in seconds, after which to give up.
                 default is -1, which means no timeout
         
         Returns
         -------
         
         True : when all msg_ids are done
         False : timeout reached, some msg_ids still outstanding
         """
         tic = time.time()
         if jobs is None:
             theids = self.outstanding
         else:
             if isinstance(jobs, (int, str, AsyncResult)):
                 jobs = [jobs]
             theids = set()
             for job in jobs:
                 if isinstance(job, int):
                     # index access
                     job = self.history[job]
                 elif isinstance(job, AsyncResult):
                     map(theids.add, job.msg_ids)
                     continue
                 theids.add(job)
         if not theids.intersection(self.outstanding):
             return True
         self.spin()
         while theids.intersection(self.outstanding):
             if timeout >= 0 and ( time.time()-tic ) > timeout:
                 break
             time.sleep(1e-3)
             self.spin()
         return len(theids.intersection(self.outstanding)) == 0
     
     #--------------------------------------------------------------------------
     # Control methods
     #--------------------------------------------------------------------------
     
     @spin_first
     def clear(self, targets=None, block=None):
         """Clear the namespace in target(s)."""
         block = self.block if block is None else block
         targets = self._build_targets(targets)[0]
         for t in targets:
             self.session.send(self._control_socket, 'clear_request', content={}, ident=t)
         error = False
         if block:
             self._flush_ignored_control()
             for i in range(len(targets)):
                 idents,msg = self.session.recv(self._control_socket,0)
                 if self.debug:
                     pprint(msg)
                 if msg['content']['status'] != 'ok':
                     error = self._unwrap_exception(msg['content'])
         else:
             self._ignored_control_replies += len(targets)
         if error:
             raise error
         
     
     @spin_first
     def abort(self, jobs=None, targets=None, block=None):
         """Abort specific jobs from the execution queues of target(s).
         
         This is a mechanism to prevent jobs that have already been submitted
         from executing.
         
         Parameters
         ----------
         
         jobs : msg_id, list of msg_ids, or AsyncResult
             The jobs to be aborted
         
         
         """
         block = self.block if block is None else block
         targets = self._build_targets(targets)[0]
         msg_ids = []
         if isinstance(jobs, (basestring,AsyncResult)):
             jobs = [jobs]
         bad_ids = filter(lambda obj: not isinstance(obj, (basestring, AsyncResult)), jobs)
         if bad_ids:
             raise TypeError("Invalid msg_id type %r, expected str or AsyncResult"%bad_ids[0])
         for j in jobs:
             if isinstance(j, AsyncResult):
                 msg_ids.extend(j.msg_ids)
             else:
                 msg_ids.append(j)
         content = dict(msg_ids=msg_ids)
         for t in targets:
             self.session.send(self._control_socket, 'abort_request', 
                     content=content, ident=t)
         error = False
         if block:
             self._flush_ignored_control()
             for i in range(len(targets)):
                 idents,msg = self.session.recv(self._control_socket,0)
                 if self.debug:
                     pprint(msg)
                 if msg['content']['status'] != 'ok':
                     error = self._unwrap_exception(msg['content'])
         else:
             self._ignored_control_replies += len(targets)
         if error:
             raise error
     
     @spin_first
     def shutdown(self, targets=None, restart=False, hub=False, block=None):
         """Terminates one or more engine processes, optionally including the hub."""
         block = self.block if block is None else block
         if hub:
             targets = 'all'
         targets = self._build_targets(targets)[0]
         for t in targets:
             self.session.send(self._control_socket, 'shutdown_request', 
                         content={'restart':restart},ident=t)
         error = False
         if block or hub:
             self._flush_ignored_control()
             for i in range(len(targets)):
                 idents,msg = self.session.recv(self._control_socket, 0)
                 if self.debug:
                     pprint(msg)
                 if msg['content']['status'] != 'ok':
                     error = self._unwrap_exception(msg['content'])
         else:
             self._ignored_control_replies += len(targets)
         
         if hub:
             time.sleep(0.25)
             self.session.send(self._query_socket, 'shutdown_request')
             idents,msg = self.session.recv(self._query_socket, 0)
             if self.debug:
                 pprint(msg)
             if msg['content']['status'] != 'ok':
                 error = self._unwrap_exception(msg['content'])
         
         if error:
             raise error
     
     #--------------------------------------------------------------------------
     # Execution related methods
     #--------------------------------------------------------------------------
     
     def _maybe_raise(self, result):
         """wrapper for maybe raising an exception if apply failed."""
         if isinstance(result, error.RemoteError):
             raise result
         
         return result
     
     def send_apply_message(self, socket, f, args=None, kwargs=None, subheader=None, track=False,
                             ident=None):
         """construct and send an apply message via a socket.
         
         This is the principal method with which all engine execution is performed by views.
         """
                             
         assert not self._closed, "cannot use me anymore, I'm closed!"
         # defaults:
         args = args if args is not None else []
         kwargs = kwargs if kwargs is not None else {}
         subheader = subheader if subheader is not None else {}
         
         # validate arguments
         if not callable(f):
             raise TypeError("f must be callable, not %s"%type(f))
         if not isinstance(args, (tuple, list)):
             raise TypeError("args must be tuple or list, not %s"%type(args))
         if not isinstance(kwargs, dict):
             raise TypeError("kwargs must be dict, not %s"%type(kwargs))
         if not isinstance(subheader, dict):
             raise TypeError("subheader must be dict, not %s"%type(subheader))
         
         bufs = util.pack_apply_message(f,args,kwargs)
         
         msg = self.session.send(socket, "apply_request", buffers=bufs, ident=ident,
                             subheader=subheader, track=track)
         
         msg_id = msg['msg_id']
         self.outstanding.add(msg_id)
         if ident:
             # possibly routed to a specific engine
             if isinstance(ident, list):
                 ident = ident[-1]
             if ident in self._engines.values():
                 # save for later, in case of engine death
                 self._outstanding_dict[ident].add(msg_id)
         self.history.append(msg_id)
         self.metadata[msg_id]['submitted'] = datetime.now()
         
         return msg
 
     #--------------------------------------------------------------------------
     # construct a View object
     #--------------------------------------------------------------------------
     
     def load_balanced_view(self, targets=None):
         """construct a DirectView object.
         
         If no arguments are specified, create a LoadBalancedView
         using all engines.
         
         Parameters
         ----------
         
         targets: list,slice,int,etc. [default: use all engines]
             The subset of engines across which to load-balance
         """
         if targets is not None:
             targets = self._build_targets(targets)[1]
         return LoadBalancedView(client=self, socket=self._task_socket, targets=targets)
     
     def direct_view(self, targets='all'):
         """construct a DirectView object.
         
         If no targets are specified, create a DirectView
         using all engines.
         
         Parameters
         ----------
         
         targets: list,slice,int,etc. [default: use all engines]
             The engines to use for the View
         """
         single = isinstance(targets, int)
         targets = self._build_targets(targets)[1]
         if single:
             targets = targets[0]
         return DirectView(client=self, socket=self._mux_socket, targets=targets)
     
     #--------------------------------------------------------------------------
     # Query methods
     #--------------------------------------------------------------------------
     
     @spin_first
     def get_result(self, indices_or_msg_ids=None, block=None):
         """Retrieve a result by msg_id or history index, wrapped in an AsyncResult object.
         
         If the client already has the results, no request to the Hub will be made.
         
         This is a convenient way to construct AsyncResult objects, which are wrappers
         that include metadata about execution, and allow for awaiting results that
         were not submitted by this Client.
         
         It can also be a convenient way to retrieve the metadata associated with
         blocking execution, since it always retrieves
         
         Examples
         --------
         ::
         
             In [10]: r = client.apply()
         
         Parameters
         ----------
         
         indices_or_msg_ids : integer history index, str msg_id, or list of either
             The indices or msg_ids of indices to be retrieved
         
         block : bool
             Whether to wait for the result to be done
         
         Returns
         -------
         
         AsyncResult
             A single AsyncResult object will always be returned.
         
         AsyncHubResult
             A subclass of AsyncResult that retrieves results from the Hub
         
         """
         block = self.block if block is None else block
         if indices_or_msg_ids is None:
             indices_or_msg_ids = -1
         
         if not isinstance(indices_or_msg_ids, (list,tuple)):
             indices_or_msg_ids = [indices_or_msg_ids]
         
         theids = []
         for id in indices_or_msg_ids:
             if isinstance(id, int):
                 id = self.history[id]
             if not isinstance(id, str):
                 raise TypeError("indices must be str or int, not %r"%id)
             theids.append(id)
         
         local_ids = filter(lambda msg_id: msg_id in self.history or msg_id in self.results, theids)
         remote_ids = filter(lambda msg_id: msg_id not in local_ids, theids)
         
         if remote_ids:
             ar = AsyncHubResult(self, msg_ids=theids)
         else:
             ar = AsyncResult(self, msg_ids=theids)
         
         if block:
             ar.wait()
         
         return ar
 
     @spin_first
     def resubmit(self, indices_or_msg_ids=None, subheader=None, block=None):
         """Resubmit one or more tasks.
 
         in-flight tasks may not be resubmitted.
 
         Parameters
         ----------
 
         indices_or_msg_ids : integer history index, str msg_id, or list of either
             The indices or msg_ids of indices to be retrieved
 
         block : bool
             Whether to wait for the result to be done
 
         Returns
         -------
 
         AsyncHubResult
             A subclass of AsyncResult that retrieves results from the Hub
 
         """
         block = self.block if block is None else block
         if indices_or_msg_ids is None:
             indices_or_msg_ids = -1
 
         if not isinstance(indices_or_msg_ids, (list,tuple)):
             indices_or_msg_ids = [indices_or_msg_ids]
 
         theids = []
         for id in indices_or_msg_ids:
             if isinstance(id, int):
                 id = self.history[id]
             if not isinstance(id, str):
                 raise TypeError("indices must be str or int, not %r"%id)
             theids.append(id)
 
         for msg_id in theids:
             self.outstanding.discard(msg_id)
             if msg_id in self.history:
                 self.history.remove(msg_id)
             self.results.pop(msg_id, None)
             self.metadata.pop(msg_id, None)
         content = dict(msg_ids = theids)
 
         self.session.send(self._query_socket, 'resubmit_request', content)
 
         zmq.select([self._query_socket], [], [])
         idents,msg = self.session.recv(self._query_socket, zmq.NOBLOCK)
         if self.debug:
             pprint(msg)
         content = msg['content']
         if content['status'] != 'ok':
             raise self._unwrap_exception(content)
 
         ar = AsyncHubResult(self, msg_ids=theids)
 
         if block:
             ar.wait()
 
         return ar
     
     @spin_first
     def result_status(self, msg_ids, status_only=True):
         """Check on the status of the result(s) of the apply request with `msg_ids`.
         
         If status_only is False, then the actual results will be retrieved, else
         only the status of the results will be checked.
         
         Parameters
         ----------
         
         msg_ids : list of msg_ids
             if int:
                 Passed as index to self.history for convenience.
         status_only : bool (default: True)
             if False:
                 Retrieve the actual results of completed tasks.
         
         Returns
         -------
         
         results : dict
             There will always be the keys 'pending' and 'completed', which will
             be lists of msg_ids that are incomplete or complete. If `status_only`
             is False, then completed results will be keyed by their `msg_id`.
         """
         if not isinstance(msg_ids, (list,tuple)):
             msg_ids = [msg_ids]
             
         theids = []
         for msg_id in msg_ids:
             if isinstance(msg_id, int):
                 msg_id = self.history[msg_id]
             if not isinstance(msg_id, basestring):
                 raise TypeError("msg_ids must be str, not %r"%msg_id)
             theids.append(msg_id)
         
         completed = []
         local_results = {}
         
         # comment this block out to temporarily disable local shortcut:
         for msg_id in theids:
             if msg_id in self.results:
                 completed.append(msg_id)
                 local_results[msg_id] = self.results[msg_id]
                 theids.remove(msg_id)
         
         if theids: # some not locally cached
             content = dict(msg_ids=theids, status_only=status_only)
             msg = self.session.send(self._query_socket, "result_request", content=content)
             zmq.select([self._query_socket], [], [])
             idents,msg = self.session.recv(self._query_socket, zmq.NOBLOCK)
             if self.debug:
                 pprint(msg)
             content = msg['content']
             if content['status'] != 'ok':
                 raise self._unwrap_exception(content)
             buffers = msg['buffers']
         else:
             content = dict(completed=[],pending=[])
         
         content['completed'].extend(completed)
         
         if status_only:
             return content
         
         failures = []
         # load cached results into result:
         content.update(local_results)
         # update cache with results:
         for msg_id in sorted(theids):
             if msg_id in content['completed']:
                 rec = content[msg_id]
                 parent = rec['header']
                 header = rec['result_header']
                 rcontent = rec['result_content']
                 iodict = rec['io']
                 if isinstance(rcontent, str):
                     rcontent = self.session.unpack(rcontent)
                 
                 md = self.metadata[msg_id]
                 md.update(self._extract_metadata(header, parent, rcontent))
                 md.update(iodict)
                 
                 if rcontent['status'] == 'ok':
                     res,buffers = util.unserialize_object(buffers)
                 else:
                     print rcontent
                     res = self._unwrap_exception(rcontent)
                     failures.append(res)
                 
                 self.results[msg_id] = res
                 content[msg_id] = res
         
         if len(theids) == 1 and failures:
                 raise failures[0]
         
         error.collect_exceptions(failures, "result_status")
         return content
 
     @spin_first
     def queue_status(self, targets='all', verbose=False):
         """Fetch the status of engine queues.
         
         Parameters
         ----------
         
         targets : int/str/list of ints/strs
                 the engines whose states are to be queried.
                 default : all
         verbose : bool
                 Whether to return lengths only, or lists of ids for each element
         """
         engine_ids = self._build_targets(targets)[1]
         content = dict(targets=engine_ids, verbose=verbose)
         self.session.send(self._query_socket, "queue_request", content=content)
         idents,msg = self.session.recv(self._query_socket, 0)
         if self.debug:
             pprint(msg)
         content = msg['content']
         status = content.pop('status')
         if status != 'ok':
             raise self._unwrap_exception(content)
         content = util.rekey(content)
         if isinstance(targets, int):
             return content[targets]
         else:
             return content
         
     @spin_first
     def purge_results(self, jobs=[], targets=[]):
         """Tell the Hub to forget results.
         
         Individual results can be purged by msg_id, or the entire
         history of specific targets can be purged.
         
         Parameters
         ----------
         
         jobs : str or list of str or AsyncResult objects
                 the msg_ids whose results should be forgotten.
         targets : int/str/list of ints/strs
                 The targets, by uuid or int_id, whose entire history is to be purged.
                 Use `targets='all'` to scrub everything from the Hub's memory.
                 
                 default : None
         """
         if not targets and not jobs:
             raise ValueError("Must specify at least one of `targets` and `jobs`")
         if targets:
             targets = self._build_targets(targets)[1]
         
         # construct msg_ids from jobs
         msg_ids = []
         if isinstance(jobs, (basestring,AsyncResult)):
             jobs = [jobs]
         bad_ids = filter(lambda obj: not isinstance(obj, (basestring, AsyncResult)), jobs)
         if bad_ids:
             raise TypeError("Invalid msg_id type %r, expected str or AsyncResult"%bad_ids[0])
         for j in jobs:
             if isinstance(j, AsyncResult):
                 msg_ids.extend(j.msg_ids)
             else:
                 msg_ids.append(j)
         
         content = dict(targets=targets, msg_ids=msg_ids)
         self.session.send(self._query_socket, "purge_request", content=content)
         idents, msg = self.session.recv(self._query_socket, 0)
         if self.debug:
             pprint(msg)
         content = msg['content']
         if content['status'] != 'ok':
             raise self._unwrap_exception(content)
 
     @spin_first
     def hub_history(self):
         """Get the Hub's history
         
         Just like the Client, the Hub has a history, which is a list of msg_ids.
         This will contain the history of all clients, and, depending on configuration,
         may contain history across multiple cluster sessions.
         
         Any msg_id returned here is a valid argument to `get_result`.
         
         Returns
         -------
         
         msg_ids : list of strs
                 list of all msg_ids, ordered by task submission time.
         """
         
         self.session.send(self._query_socket, "history_request", content={})
         idents, msg = self.session.recv(self._query_socket, 0)
         
         if self.debug:
             pprint(msg)
         content = msg['content']
         if content['status'] != 'ok':
             raise self._unwrap_exception(content)
         else:
             return content['history']
 
     @spin_first
     def db_query(self, query, keys=None):
         """Query the Hub's TaskRecord database
         
         This will return a list of task record dicts that match `query`
         
         Parameters
         ----------
         
         query : mongodb query dict
             The search dict. See mongodb query docs for details.
         keys : list of strs [optional]
             The subset of keys to be returned.  The default is to fetch everything but buffers.
             'msg_id' will *always* be included.
         """
         if isinstance(keys, basestring):
             keys = [keys]
         content = dict(query=query, keys=keys)
         self.session.send(self._query_socket, "db_request", content=content)
         idents, msg = self.session.recv(self._query_socket, 0)
         if self.debug:
             pprint(msg)
         content = msg['content']
         if content['status'] != 'ok':
             raise self._unwrap_exception(content)
         
         records = content['records']
         buffer_lens = content['buffer_lens']
         result_buffer_lens = content['result_buffer_lens']
         buffers = msg['buffers']
         has_bufs = buffer_lens is not None
         has_rbufs = result_buffer_lens is not None
         for i,rec in enumerate(records):
             # relink buffers
             if has_bufs:
                 blen = buffer_lens[i]
                 rec['buffers'], buffers = buffers[:blen],buffers[blen:]
             if has_rbufs:
                 blen = result_buffer_lens[i]
                 rec['result_buffers'], buffers = buffers[:blen],buffers[blen:]
             # turn timestamps back into times
             for key in 'submitted started completed resubmitted'.split():
                 maybedate = rec.get(key, None)
                 if maybedate and util.ISO8601_RE.match(maybedate):
                     rec[key] = datetime.strptime(maybedate, util.ISO8601)
             
         return records
 
 __all__ = [ 'Client' ]

IPython/parallel/controller/sqlitedb.py

@@ -1,333 +1,339 b''
 """A TaskRecord backend using sqlite3"""
 #-----------------------------------------------------------------------------
 #  Copyright (C) 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.
 #-----------------------------------------------------------------------------
 
 import json
 import os
 import cPickle as pickle
 from datetime import datetime
 
 import sqlite3
 
 from zmq.eventloop import ioloop
 
 from IPython.utils.traitlets import Unicode, Instance, List
 from .dictdb import BaseDB
 from IPython.parallel.util import ISO8601
 
 #-----------------------------------------------------------------------------
 # SQLite operators, adapters, and converters
 #-----------------------------------------------------------------------------
 
 operators = {
  '$lt' : "<",
  '$gt' : ">",
  # null is handled weird with ==,!=
  '$eq' : "=",
  '$ne' : "!=",
  '$lte': "<=",
  '$gte': ">=",
  '$in' : ('=', ' OR '),
  '$nin': ('!=', ' AND '),
  # '$all': None,
  # '$mod': None,
  # '$exists' : None
 }
 null_operators = {
 '=' : "IS NULL",
 '!=' : "IS NOT NULL",
 }
 
 def _adapt_datetime(dt):
     return dt.strftime(ISO8601)
 
 def _convert_datetime(ds):
     if ds is None:
         return ds
     else:
         return datetime.strptime(ds, ISO8601)
 
 def _adapt_dict(d):
     return json.dumps(d)
 
 def _convert_dict(ds):
     if ds is None:
         return ds
     else:
         return json.loads(ds)
 
 def _adapt_bufs(bufs):
     # this is *horrible*
     # copy buffers into single list and pickle it:
     if bufs and isinstance(bufs[0], (bytes, buffer)):
         return sqlite3.Binary(pickle.dumps(map(bytes, bufs),-1))
     elif bufs:
         return bufs
     else:
         return None
 
 def _convert_bufs(bs):
     if bs is None:
         return []
     else:
         return pickle.loads(bytes(bs))
 
 #-----------------------------------------------------------------------------
 # SQLiteDB class
 #-----------------------------------------------------------------------------
 
 class SQLiteDB(BaseDB):
     """SQLite3 TaskRecord backend."""
     
     filename = Unicode('tasks.db', config=True,
         help="""The filename of the sqlite task database. [default: 'tasks.db']""")
     location = Unicode('', config=True,
         help="""The directory containing the sqlite task database.  The default
         is to use the cluster_dir location.""")
     table = Unicode("", config=True,
         help="""The SQLite Table to use for storing tasks for this session. If unspecified,
         a new table will be created with the Hub's IDENT.  Specifying the table will result
         in tasks from previous sessions being available via Clients' db_query and
         get_result methods.""")
     
     _db = Instance('sqlite3.Connection')
     _keys = List(['msg_id' ,
             'header' ,
             'content',
             'buffers',
             'submitted',
             'client_uuid' ,
             'engine_uuid' ,
             'started',
             'completed',
             'resubmitted',
             'result_header' ,
             'result_content' ,
             'result_buffers' ,
             'queue' ,
             'pyin' ,
             'pyout',
             'pyerr',
             'stdout',
             'stderr',
         ])
     
     def __init__(self, **kwargs):
         super(SQLiteDB, self).__init__(**kwargs)
         if not self.table:
             # use session, and prefix _, since starting with # is illegal
             self.table = '_'+self.session.replace('-','_')
         if not self.location:
-            if hasattr(self.config.Global, 'cluster_dir'):
-                self.location = self.config.Global.cluster_dir
+            # get current profile
+            from IPython.core.newapplication import BaseIPythonApplication
+            if BaseIPythonApplication.initialized():
+                app = BaseIPythonApplication.instance()
+                if app.profile_dir is not None:
+                    self.location = app.profile_dir.location
             else:
-                self.location = '.'
+                    self.location = u'.'
+            else:
+                self.location = u'.'
         self._init_db()
         
         # register db commit as 2s periodic callback
         # to prevent clogging pipes
         # assumes we are being run in a zmq ioloop app
         loop = ioloop.IOLoop.instance()
         pc = ioloop.PeriodicCallback(self._db.commit, 2000, loop)
         pc.start()
     
     def _defaults(self, keys=None):
         """create an empty record"""
         d = {}
         keys = self._keys if keys is None else keys
         for key in keys:
             d[key] = None
         return d
     
     def _init_db(self):
         """Connect to the database and get new session number."""
         # register adapters
         sqlite3.register_adapter(datetime, _adapt_datetime)
         sqlite3.register_converter('datetime', _convert_datetime)
         sqlite3.register_adapter(dict, _adapt_dict)
         sqlite3.register_converter('dict', _convert_dict)
         sqlite3.register_adapter(list, _adapt_bufs)
         sqlite3.register_converter('bufs', _convert_bufs)
         # connect to the db
         dbfile = os.path.join(self.location, self.filename)
         self._db = sqlite3.connect(dbfile, detect_types=sqlite3.PARSE_DECLTYPES, 
             # isolation_level = None)#,
              cached_statements=64)
         # print dir(self._db)
         
         self._db.execute("""CREATE TABLE IF NOT EXISTS %s 
                 (msg_id text PRIMARY KEY,
                 header dict text,
                 content dict text,
                 buffers bufs blob,
                 submitted datetime text,
                 client_uuid text,
                 engine_uuid text,
                 started datetime text,
                 completed datetime text,
                 resubmitted datetime text,
                 result_header dict text,
                 result_content dict text,
                 result_buffers bufs blob,
                 queue text,
                 pyin text,
                 pyout text,
                 pyerr text,
                 stdout text,
                 stderr text)
                 """%self.table)
         self._db.commit()
     
     def _dict_to_list(self, d):
         """turn a mongodb-style record dict into a list."""
         
         return [ d[key] for key in self._keys ]
     
     def _list_to_dict(self, line, keys=None):
         """Inverse of dict_to_list"""
         keys = self._keys if keys is None else keys
         d = self._defaults(keys)
         for key,value in zip(keys, line):
             d[key] = value
         
         return d
     
     def _render_expression(self, check):
         """Turn a mongodb-style search dict into an SQL query."""
         expressions = []
         args = []
         
         skeys = set(check.keys())
         skeys.difference_update(set(self._keys))
         skeys.difference_update(set(['buffers', 'result_buffers']))
         if skeys:
             raise KeyError("Illegal testing key(s): %s"%skeys)
         
         for name,sub_check in check.iteritems():
             if isinstance(sub_check, dict):
                 for test,value in sub_check.iteritems():
                     try:
                         op = operators[test]
                     except KeyError:
                         raise KeyError("Unsupported operator: %r"%test)
                     if isinstance(op, tuple):
                         op, join = op
                     
                     if value is None and op in null_operators:
                             expr = "%s %s"%null_operators[op]
                     else:
                         expr = "%s %s ?"%(name, op)
                         if isinstance(value, (tuple,list)):
                             if op in null_operators and any([v is None for v in value]):
                                 # equality tests don't work with NULL
                                 raise ValueError("Cannot use %r test with NULL values on SQLite backend"%test)
                             expr = '( %s )'%( join.join([expr]*len(value)) )
                             args.extend(value)
                         else:
                             args.append(value)
                     expressions.append(expr)
             else:
                 # it's an equality check
                 if sub_check is None:
                     expressions.append("%s IS NULL")
                 else:
                     expressions.append("%s = ?"%name)
                     args.append(sub_check)
         
         expr = " AND ".join(expressions)
         return expr, args
     
     def add_record(self, msg_id, rec):
         """Add a new Task Record, by msg_id."""
         d = self._defaults()
         d.update(rec)
         d['msg_id'] = msg_id
         line = self._dict_to_list(d)
         tups = '(%s)'%(','.join(['?']*len(line)))
         self._db.execute("INSERT INTO %s VALUES %s"%(self.table, tups), line)
         # self._db.commit()
     
     def get_record(self, msg_id):
         """Get a specific Task Record, by msg_id."""
         cursor = self._db.execute("""SELECT * FROM %s WHERE msg_id==?"""%self.table, (msg_id,))
         line = cursor.fetchone()
         if line is None:
             raise KeyError("No such msg: %r"%msg_id)
         return self._list_to_dict(line)
     
     def update_record(self, msg_id, rec):
         """Update the data in an existing record."""
         query = "UPDATE %s SET "%self.table
         sets = []
         keys = sorted(rec.keys())
         values = []
         for key in keys:
             sets.append('%s = ?'%key)
             values.append(rec[key])
         query += ', '.join(sets)
         query += ' WHERE msg_id == ?'
         values.append(msg_id)
         self._db.execute(query, values)
         # self._db.commit()
     
     def drop_record(self, msg_id):
         """Remove a record from the DB."""
         self._db.execute("""DELETE FROM %s WHERE msg_id==?"""%self.table, (msg_id,))
         # self._db.commit()
     
     def drop_matching_records(self, check):
         """Remove a record from the DB."""
         expr,args = self._render_expression(check)
         query = "DELETE FROM %s WHERE %s"%(self.table, expr)
         self._db.execute(query,args)
         # self._db.commit()
         
     def find_records(self, check, keys=None):
         """Find records matching a query dict, optionally extracting subset of keys.
         
         Returns list of matching records.
         
         Parameters
         ----------
         
         check: dict
             mongodb-style query argument
         keys: list of strs [optional]
             if specified, the subset of keys to extract.  msg_id will *always* be
             included.
         """
         if keys:
             bad_keys = [ key for key in keys if key not in self._keys ]
             if bad_keys:
                 raise KeyError("Bad record key(s): %s"%bad_keys)
         
         if keys:
             # ensure msg_id is present and first:
             if 'msg_id' in keys:
                 keys.remove('msg_id')
             keys.insert(0, 'msg_id')
             req = ', '.join(keys)
         else:
             req = '*'
         expr,args = self._render_expression(check)
         query = """SELECT %s FROM %s WHERE %s"""%(req, self.table, expr)
         cursor = self._db.execute(query, args)
         matches = cursor.fetchall()
         records = []
         for line in matches:
             rec = self._list_to_dict(line, keys)
             records.append(rec)
         return records
     
     def get_history(self):
         """get all msg_ids, ordered by time submitted."""
         query = """SELECT msg_id FROM %s ORDER by submitted ASC"""%self.table
         cursor = self._db.execute(query)
         # will be a list of length 1 tuples
         return [ tup[0] for tup in cursor.fetchall()]
 
 __all__ = ['SQLiteDB']
  No newline at end of file