cleanup parallel traits

* eliminate use of CUnicode
* eliminate use of Str
* all remaining CStr will become CBytes

IPython/parallel/apps/clusterdir.py

@@ -1,539 +1,539 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.core import release
 from IPython.utils.path import (
     get_ipython_package_dir,
     get_ipython_dir,
     expand_path
 )
-from IPython.utils.traitlets import Unicode, Bool, CStr, Instance, Dict
+from IPython.utils.traitlets import Unicode, Bool, Instance, Dict
 
 #-----------------------------------------------------------------------------
 # 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):
     """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__(
             app, contact_name, contact_email, bug_tracker
         )
 
 
 #-----------------------------------------------------------------------------
 # Main application
 #-----------------------------------------------------------------------------
 base_aliases = {
     'profile' : "ClusterDir.profile",
     'cluster_dir' : 'ClusterDir.location',
     'auto_create' : 'ClusterDirApplication.auto_create',
     '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',
 }
 
 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"),
 }
 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.
 
     The cluster directory is resolved as follows:
 
     * 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.
 
     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.
     """
 
     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]
     
     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, shortname='--clean-logs', config=True,
         help="whether to cleanup old logfiles before starting")
 
-    log_url = CStr('', shortname='--log-url', config=True,
-        help="The ZMQ URL of the iplooger to aggregate logging.")
+    log_url = Unicode('', shortname='--log-url', config=True,
+        help="The ZMQ URL of the iplogger to aggregate logging.")
 
     config_file = Unicode(u'', config=True,
         help="""Path to ipcontroller configuration file.  The default is to use
          <appname>_config.py, as found by cluster-dir."""
     )
     
     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)
     
     def initialize(self, argv=None):
         """initialize the app"""
         self.init_crash_handler()
         self.parse_command_line(argv)
         cl_config = self.config
         self.init_clusterdir()
         if self.config_file:
             self.load_config_file(self.config_file)
         else:
             self.load_config_file(self.default_config_file_name, path=self.cluster_dir.location)
         # command-line should *override* config file, but command-line is necessary
         # to determine clusterdir, etc.
         self.update_config(cl_config)
         self.reinit_logging()
 
         self.to_work_dir()
         
     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)
 
     def load_config_file(self, filename, path=None):
         """Load a .py based config file by filename and path."""
         # use config.application.Application.load_config
         # instead of inflexible core.newapplication.BaseIPythonApplication.load_config
         return Application.load_config_file(self, filename, path=path)
     #
     # def load_default_config_file(self):
     #     """Load a .py based config file by filename and path."""
     #     return BaseIPythonApplication.load_config_file(self)
 
     # disable URL-logging
     def reinit_logging(self):
         # Remove old log files
         log_dir = self.cluster_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')
         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')
         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')
         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,537 +1,537 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.utils.importstring import import_item
-from IPython.utils.traitlets import Int, CStr, CUnicode, Str, Bool, CFloat, Dict, List
+from IPython.utils.traitlets import Int, Unicode, Bool, CFloat, Dict, List
 
 from IPython.parallel.apps.clusterdir import (
     ClusterApplication, ClusterDirError, ClusterDir,
     PIDFileError,
     base_flags, base_aliases
 )
 
 
 #-----------------------------------------------------------------------------
 # Module level variables
 #-----------------------------------------------------------------------------
 
 
 default_config_file_name = u'ipcluster_config.py'
 
 
 _description = """\
 Start an IPython cluster for parallel computing.\n\n
 
 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 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.
 """
 stop_help = """
 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.
 """
 engines_help = """
 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.
 """
 create_help = """
 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 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):
         # Find the search paths
         cluster_dir_paths = os.environ.get('IPCLUSTER_DIR_PATH','')
         if cluster_dir_paths:
             cluster_dir_paths = cluster_dir_paths.split(':')
         else:
             cluster_dir_paths = []
         
         ipython_dir = self.ipython_dir
         
         paths = [os.getcwd(), ipython_dir] + cluster_dir_paths
         paths = list(set(paths))
 
         self.log.info('Searching for cluster dirs 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]
                     start_cmd = 'ipcluster start profile=%s n=4' % profile
                     print start_cmd + " ==> " + full_path
     
     def start(self):
         self.list_cluster_dirs()
 
 create_flags = {}
 create_flags.update(base_flags)
 create_flags.update(boolean_flag('reset', 'IPClusterCreate.reset', 
                 "reset config files to defaults", "leave existing config files"))
 
 class IPClusterCreate(ClusterApplication):
     name = u'ipcluster'
     description = create_help
     auto_create_cluster_dir = Bool(True,
         help="whether to create the cluster_dir if it doesn't exist")
     default_config_file_name = 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'))
     
     classes = [ClusterDir]
     
     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',
 )
 
 class IPClusterStop(ClusterApplication):
     name = u'ipcluster'
     description = stop_help
     auto_create_cluster_dir = Bool(False)
     default_config_file_name = 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):
 
     name = u'ipcluster'
     description = engines_help
     usage = None
     default_config_file_name = 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
     
     n = Int(2, config=True,
         help="The number of engines to start.")
 
-    engine_launcher_class = Str('LocalEngineSetLauncher',
+    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
         )
         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
         )
 
     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
             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()
 
     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
     usage = None
     default_config_file_name = default_config_file_name
     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")
     classes = List()
     def _classes_default(self,):
         from IPython.parallel.apps import launcher
         return [ClusterDir]+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 = Str('LocalControllerLauncher',
+    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
         )
         
     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):
     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.initialize()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()
 

IPython/parallel/apps/ipcontrollerapp.py

@@ -1,401 +1,401 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.parallel.apps.clusterdir import (
     ClusterDir,
     ClusterApplication,
     base_flags
     # ClusterDirConfigLoader
 )
 from IPython.utils.importstring import import_item
-from IPython.utils.traitlets import Instance, Unicode, Bool, List, CStr, Dict
+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.scheduler import TaskScheduler,launch_scheduler
 from IPython.parallel.controller.sqlitedb import SQLiteDB
 
 from IPython.parallel.util import signal_children,disambiguate_ip_address, 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.
 """
 
 
 
 
 #-----------------------------------------------------------------------------
 # The main application
 #-----------------------------------------------------------------------------
 flags = {}
 flags.update(base_flags)
 flags.update({
     'usethreads' : ( {'IPControllerApp' : {'usethreads' : True}},
                     'Use threads instead of processes for the schedulers'),
     'sqlitedb' : ({'HubFactory' : {'db_class' : 'IPython.parallel.controller.sqlitedb.SQLiteDB'}},
                     'use the SQLiteDB backend'),
     'mongodb' : ({'HubFactory' : {'db_class' : 'IPython.parallel.controller.mongodb.MongoDB'}},
                     'use the MongoDB backend'),
     'dictdb' : ({'HubFactory' : {'db_class' : 'IPython.parallel.controller.dictdb.DictDB'}},
                     'use the in-memory DictDB backend'),
 })
 
 flags.update()
 
 class IPControllerApp(ClusterApplication):
 
     name = u'ipcontroller'
     description = _description
     # command_line_loader = IPControllerAppConfigLoader
     default_config_file_name = default_config_file_name
     classes = [ClusterDir, StreamSession, HubFactory, TaskScheduler, HeartMonitor, SQLiteDB] + maybe_mongo
     
     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"
     )
 
     usethreads = Bool(False, config=True,
         help='Use threads instead of processes for the schedulers',
         )
 
     # internal
     children = List()
-    mq_class = CStr('zmq.devices.ProcessMonitoredQueue')
+    mq_class = Unicode('zmq.devices.ProcessMonitoredQueue')
 
     def _usethreads_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',
         reuse_files = 'IPControllerApp.reuse_files',
         secure = 'IPControllerApp.secure',
         ssh = 'IPControllerApp.ssh_server',
         usethreads = 'IPControllerApp.usethreads',
         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',
         
     ))
     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)
         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:
             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:
             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):
         # This is the working dir by now.
         sys.path.insert(0, '')
         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)
             # 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(self.mq_class)
         
         hub = self.factory
         # maybe_inproc = 'inproc://monitor' if self.usethreads 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=self.log.name, loglevel=self.log_level,
                                                             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
         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 start_logging(self):
     #     super(IPControllerApp, self).start_logging()
     #     if self.config.Global.log_url:
     #         context = self.factory.context
     #         lsock = context.socket(zmq.PUB)
     #         lsock.connect(self.config.Global.log_url)
     #         handler = PUBHandler(lsock)
     #         handler.root_topic = 'controller'
     #         handler.setLevel(self.log_level)
     #         self.log.addHandler(handler)
     # #
     
     def initialize(self, argv=None):
         super(IPControllerApp, self).initialize(argv)
         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,289 +1,289 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.parallel.apps.clusterdir import (
     ClusterApplication,
     ClusterDir,
     base_aliases,
     # ClusterDirConfigLoader
 )
 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 Str, Bool, Unicode, Dict, List, CStr
+from IPython.utils.traitlets import Bool, Unicode, Dict, List, CStr
 
 
 #-----------------------------------------------------------------------------
 # 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.\n\n
 
 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.
 """
 
 
 #-----------------------------------------------------------------------------
 # 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 = Str('', config=True,
+    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 = Str('', config=True,
+    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):
 
     app_name = Unicode(u'ipengine')
     description = Unicode(_description)
     default_config_file_name = 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")
     
     startup_script = Unicode(u'', config=True,
         help='specify a script to be run at startup')
-    startup_command = Str('', config=True,
+    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.""",
         )
 
     url_file_name = Unicode(u'ipcontroller-engine.json')
 
     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',
 
         mpi = 'MPI.use',
 
         log_level = 'IPEngineApp.log_level',
     ))
 
     # 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.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.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)
         
         # self.start_logging()
 
         # Create the service hierarchy
         # self.main_service = service.MultiService()
         # self.engine_service.setServiceParent(self.main_service)
         # self.tub_service = Tub()
         # self.tub_service.setServiceParent(self.main_service)
         # # This needs to be called before the connection is initiated
         # self.main_service.startService()
 
         # This initiates the connection to the controller and calls
         # register_engine to tell the controller we are ready to do work
         # self.engine_connector = EngineConnector(self.tub_service)
 
         # self.log.info("Using furl file: %s" % self.master_config.Global.furl_file)
 
         # reactor.callWhenRunning(self.call_connect)
 
     # def start_logging(self):
     #     super(IPEngineApp, self).start_logging()
     #     if self.master_config.Global.log_url:
     #         context = self.engine.context
     #         lsock = context.socket(zmq.PUB)
     #         lsock.connect(self.master_config.Global.log_url)
     #         handler = EnginePUBHandler(self.engine, lsock)
     #         handler.setLevel(self.log_level)
     #         self.log.addHandler(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()
     
     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/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, Str, Int, List, Unicode, Dict, Instance, CUnicode
+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
     # 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)
         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)
         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)
         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)
             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)
         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)
         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 = CUnicode('', config=True,
+    hostname = Unicode('', config=True,
         help="hostname on which to launch the program")
-    user = CUnicode('', config=True,
+    user = Unicode('', config=True,
         help="username for ssh")
-    location = CUnicode('', config=True,
+    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)
         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.
         `n` is ignored, and the `engines` config property is used instead.
         """
         
         self.cluster_dir = unicode(cluster_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)
                 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 = Str(r'\d+', config=True,
+    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 = CUnicode(u'ipython_job.xml', config=True,
+    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 = CUnicode(u'')
-    scheduler = CUnicode('', config=True,
+    job_file = Unicode(u'')
+    scheduler = Unicode('', config=True,
         help="The hostname of the scheduler to submit the job to.")
-    job_cmd = CUnicode(find_job_cmd(), config=True,
+    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 = CUnicode(u'ipcontroller_job.xml', config=True,
+    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.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)
 
     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)
         return super(WindowsHPCControllerLauncher, self).start(1)
 
 
 class WindowsHPCEngineSetLauncher(WindowsHPCLauncher):
 
-    job_file_name = CUnicode(u'ipengineset_job.xml', config=True,
+    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.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)
 
     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)
         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 = CUnicode('', config=True,
+    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 = CUnicode('', config=True,
+    batch_template = Unicode('', config=True,
         help="The string that is the batch script template itself.")
-    batch_template_file = CUnicode(u'', config=True,
+    batch_template_file = Unicode(u'', config=True,
         help="The file that contains the batch template.")
-    batch_file_name = CUnicode(u'batch_script', config=True,
+    batch_file_name = Unicode(u'batch_script', config=True,
         help="The filename of the instantiated batch script.")
-    queue = CUnicode(u'', config=True,
+    queue = Unicode(u'', config=True,
         help="The PBS Queue.")
     
     # not configurable, override in subclasses
     # PBS Job Array regex
-    job_array_regexp = CUnicode('')
-    job_array_template = CUnicode('')
+    job_array_regexp = Unicode('')
+    job_array_template = Unicode('')
     # PBS Queue regex
-    queue_regexp = CUnicode('')
-    queue_template = CUnicode('')
+    queue_regexp = Unicode('')
+    queue_template = Unicode('')
     # The default batch template, override in subclasses
-    default_template = CUnicode('')
+    default_template = Unicode('')
     # The full path to the instantiated batch script.
-    batch_file = CUnicode(u'')
+    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):
         """Start n copies of the process using a batch system."""
         # Here we save profile and cluster_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)
         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 = CUnicode(r'\d+', config=True,
+    job_id_regexp = Unicode(r'\d+', config=True,
         help="Regular expresion for identifying the job ID [r'\d+']")
     
-    batch_file = CUnicode(u'')
-    job_array_regexp = CUnicode('#PBS\W+-t\W+[\w\d\-\$]+')
-    job_array_template = CUnicode('#PBS -t 1-$n')
-    queue_regexp = CUnicode('#PBS\W+-q\W+\$?\w+')
-    queue_template = CUnicode('#PBS -q $queue')
+    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 = CUnicode(u'pbs_controller', config=True,
+    batch_file_name = Unicode(u'pbs_controller', config=True,
         help="batch file name for the controller job.")
-    default_template= CUnicode("""#!/bin/sh
+    default_template= Unicode("""#!/bin/sh
 #PBS -V
 #PBS -N ipcontroller
 %s --log-to-file cluster_dir $cluster_dir
 """%(' '.join(ipcontroller_cmd_argv)))
 
     def start(self, cluster_dir):
         """Start the controller by profile or cluster_dir."""
         self.log.info("Starting PBSControllerLauncher: %r" % self.args)
         return super(PBSControllerLauncher, self).start(1, cluster_dir)
 
 
 class PBSEngineSetLauncher(PBSLauncher):
     """Launch Engines using PBS"""
-    batch_file_name = CUnicode(u'pbs_engines', config=True,
+    batch_file_name = Unicode(u'pbs_engines', config=True,
         help="batch file name for the engine(s) job.")
-    default_template= CUnicode(u"""#!/bin/sh
+    default_template= Unicode(u"""#!/bin/sh
 #PBS -V
 #PBS -N ipengine
 %s cluster_dir $cluster_dir
 """%(' '.join(ipengine_cmd_argv)))
 
     def start(self, n, cluster_dir):
         """Start n engines by profile or cluster_dir."""
         self.log.info('Starting %i engines with PBSEngineSetLauncher: %r' % (n, self.args))
         return super(PBSEngineSetLauncher, self).start(n, cluster_dir)
 
 #SGE is very similar to PBS
 
 class SGELauncher(PBSLauncher):
     """Sun GridEngine is a PBS clone with slightly different syntax"""
-    job_array_regexp = CUnicode('#$$\W+-t\W+[\w\d\-\$]+')
-    job_array_template = CUnicode('#$$ -t 1-$n')
-    queue_regexp = CUnicode('#$$\W+-q\W+\$?\w+')
-    queue_template = CUnicode('#$$ -q $queue')
+    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 = CUnicode(u'sge_controller', config=True,
+    batch_file_name = Unicode(u'sge_controller', config=True,
         help="batch file name for the ipontroller job.")
-    default_template= CUnicode(u"""#$$ -V
+    default_template= Unicode(u"""#$$ -V
 #$$ -S /bin/sh
 #$$ -N ipcontroller
 %s --log-to-file cluster_dir=$cluster_dir
 """%(' '.join(ipcontroller_cmd_argv)))
 
     def start(self, cluster_dir):
         """Start the controller by profile or cluster_dir."""
         self.log.info("Starting PBSControllerLauncher: %r" % self.args)
         return super(PBSControllerLauncher, self).start(1, cluster_dir)
 
 class SGEEngineSetLauncher(SGELauncher):
     """Launch Engines with SGE"""
-    batch_file_name = CUnicode(u'sge_engines', config=True,
+    batch_file_name = Unicode(u'sge_engines', config=True,
         help="batch file name for the engine(s) job.")
-    default_template = CUnicode("""#$$ -V
+    default_template = Unicode("""#$$ -V
 #$$ -S /bin/sh
 #$$ -N ipengine
 %s cluster_dir=$cluster_dir
 """%(' '.join(ipengine_cmd_argv)))
 
     def start(self, n, cluster_dir):
         """Start n engines by profile or cluster_dir."""
         self.log.info('Starting %i engines with SGEEngineSetLauncher: %r' % (n, self.args))
         return super(SGEEngineSetLauncher, self).start(n, cluster_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 = Str('start')
+    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/apps/logwatcher.py

@@ -1,98 +1,98 b''
 #!/usr/bin/env python
 """A simple logger object that consolidates messages incoming from ipcluster processes."""
 
 #-----------------------------------------------------------------------------
 #  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 logging
 import sys
 
 import zmq
 from zmq.eventloop import ioloop, zmqstream
 
-from IPython.utils.traitlets import Int, Str, Instance, List
+from IPython.utils.traitlets import Int, Unicode, Instance, List
 
 from IPython.parallel.factory import LoggingFactory
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 
 class LogWatcher(LoggingFactory):
     """A simple class that receives messages on a SUB socket, as published
     by subclasses of `zmq.log.handlers.PUBHandler`, and logs them itself.
     
     This can subscribe to multiple topics, but defaults to all topics.
     """
     # configurables
     topics = List([''], config=True)
-    url = Str('tcp://127.0.0.1:20202', config=True)
+    url = Unicode('tcp://127.0.0.1:20202', config=True)
     
     # internals
     context = Instance(zmq.Context, (), {})
     stream = Instance('zmq.eventloop.zmqstream.ZMQStream')
     loop = Instance('zmq.eventloop.ioloop.IOLoop')
     def _loop_default(self):
         return ioloop.IOLoop.instance()
     
     def __init__(self, **kwargs):
         super(LogWatcher, self).__init__(**kwargs)
         s = self.context.socket(zmq.SUB)
         s.bind(self.url)
         self.stream = zmqstream.ZMQStream(s, self.loop)
         self.subscribe()
         self.on_trait_change(self.subscribe, 'topics')
     
     def start(self):
         self.stream.on_recv(self.log_message)
     
     def stop(self):
         self.stream.stop_on_recv()
     
     def subscribe(self):
         """Update our SUB socket's subscriptions."""
         self.stream.setsockopt(zmq.UNSUBSCRIBE, '')
         for topic in self.topics:
             self.log.debug("Subscribing to: %r"%topic)
             self.stream.setsockopt(zmq.SUBSCRIBE, topic)
     
     def _extract_level(self, topic_str):
         """Turn 'engine.0.INFO.extra' into (logging.INFO, 'engine.0.extra')"""
         topics = topic_str.split('.')
         for idx,t in enumerate(topics):
             level = getattr(logging, t, None)
             if level is not None:
                 break
         
         if level is None:
             level = logging.INFO
         else:
             topics.pop(idx)
         
         return level, '.'.join(topics)
             
             
     def log_message(self, raw):
         """receive and parse a message, then log it."""
         if len(raw) != 2 or '.' not in raw[0]:
             self.log.error("Invalid log message: %s"%raw)
             return
         else:
             topic, msg = raw
             # don't newline, since log messages always newline:
             topic,level_name = topic.rsplit('.',1)
             level,topic = self._extract_level(topic)
             if msg[-1] == '\n':
                 msg = msg[:-1]
             logging.log(level, "[%s] %s" % (topic, msg))
 

IPython/parallel/apps/winhpcjob.py

@@ -1,314 +1,314 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 Job and task components for writing .xml files that the Windows HPC Server 
 2008 can use to start jobs.
 """
 
 #-----------------------------------------------------------------------------
 #  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 os
 import re
 import uuid
 
 from xml.etree import ElementTree as ET
 
 from IPython.config.configurable import Configurable
 from IPython.utils.traitlets import (
-    Str, Int, List, Instance,
-    Enum, Bool, CStr
+    Unicode, Int, List, Instance,
+    Enum, Bool
 )
 
 #-----------------------------------------------------------------------------
 # Job and Task classes
 #-----------------------------------------------------------------------------
 
 
 def as_str(value):
     if isinstance(value, str):
         return value
     elif isinstance(value, bool):
         if value:
             return 'true'
         else:
             return 'false'
     elif isinstance(value, (int, float)):
         return repr(value)
     else:
         return value
 
 
 def indent(elem, level=0):
     i = "\n" + level*"  "
     if len(elem):
         if not elem.text or not elem.text.strip():
             elem.text = i + "  "
         if not elem.tail or not elem.tail.strip():
             elem.tail = i
         for elem in elem:
             indent(elem, level+1)
         if not elem.tail or not elem.tail.strip():
             elem.tail = i
     else:
         if level and (not elem.tail or not elem.tail.strip()):
             elem.tail = i
 
 
 def find_username():
     domain = os.environ.get('USERDOMAIN')
     username = os.environ.get('USERNAME','')
     if domain is None:
         return username
     else:
         return '%s\\%s' % (domain, username)
 
 
 class WinHPCJob(Configurable):
 
-    job_id = Str('')
-    job_name = Str('MyJob', config=True)
+    job_id = Unicode('')
+    job_name = Unicode('MyJob', config=True)
     min_cores = Int(1, config=True)
     max_cores = Int(1, config=True)
     min_sockets = Int(1, config=True)
     max_sockets = Int(1, config=True)
     min_nodes = Int(1, config=True)
     max_nodes = Int(1, config=True)
-    unit_type = Str("Core", config=True)
+    unit_type = Unicode("Core", config=True)
     auto_calculate_min = Bool(True, config=True)
     auto_calculate_max = Bool(True, config=True)
     run_until_canceled = Bool(False, config=True)
     is_exclusive = Bool(False, config=True)
-    username = Str(find_username(), config=True)
-    job_type = Str('Batch', config=True)
+    username = Unicode(find_username(), config=True)
+    job_type = Unicode('Batch', config=True)
     priority = Enum(('Lowest','BelowNormal','Normal','AboveNormal','Highest'),
         default_value='Highest', config=True)
-    requested_nodes = Str('', config=True)
-    project = Str('IPython', config=True)
-    xmlns = Str('http://schemas.microsoft.com/HPCS2008/scheduler/')
-    version = Str("2.000")
+    requested_nodes = Unicode('', config=True)
+    project = Unicode('IPython', config=True)
+    xmlns = Unicode('http://schemas.microsoft.com/HPCS2008/scheduler/')
+    version = Unicode("2.000")
     tasks = List([])
 
     @property
     def owner(self):
         return self.username
 
     def _write_attr(self, root, attr, key):
         s = as_str(getattr(self, attr, ''))
         if s:
             root.set(key, s)
 
     def as_element(self):
         # We have to add _A_ type things to get the right order than
         # the MSFT XML parser expects.
         root = ET.Element('Job')
         self._write_attr(root, 'version', '_A_Version')
         self._write_attr(root, 'job_name', '_B_Name')
         self._write_attr(root, 'unit_type', '_C_UnitType')
         self._write_attr(root, 'min_cores', '_D_MinCores')
         self._write_attr(root, 'max_cores', '_E_MaxCores')
         self._write_attr(root, 'min_sockets', '_F_MinSockets')
         self._write_attr(root, 'max_sockets', '_G_MaxSockets')
         self._write_attr(root, 'min_nodes', '_H_MinNodes')
         self._write_attr(root, 'max_nodes', '_I_MaxNodes')
         self._write_attr(root, 'run_until_canceled', '_J_RunUntilCanceled')
         self._write_attr(root, 'is_exclusive', '_K_IsExclusive')
         self._write_attr(root, 'username', '_L_UserName')
         self._write_attr(root, 'job_type', '_M_JobType')
         self._write_attr(root, 'priority', '_N_Priority')
         self._write_attr(root, 'requested_nodes', '_O_RequestedNodes')
         self._write_attr(root, 'auto_calculate_max', '_P_AutoCalculateMax')
         self._write_attr(root, 'auto_calculate_min', '_Q_AutoCalculateMin')
         self._write_attr(root, 'project', '_R_Project')
         self._write_attr(root, 'owner', '_S_Owner')
         self._write_attr(root, 'xmlns', '_T_xmlns')
         dependencies = ET.SubElement(root, "Dependencies")
         etasks = ET.SubElement(root, "Tasks")
         for t in self.tasks:
             etasks.append(t.as_element())
         return root
 
     def tostring(self):
         """Return the string representation of the job description XML."""
         root = self.as_element()
         indent(root)
         txt = ET.tostring(root, encoding="utf-8")
         # Now remove the tokens used to order the attributes.
         txt = re.sub(r'_[A-Z]_','',txt)
         txt = '<?xml version="1.0" encoding="utf-8"?>\n' + txt
         return txt
 
     def write(self, filename):
         """Write the XML job description to a file."""
         txt = self.tostring()
         with open(filename, 'w') as f:
             f.write(txt)
 
     def add_task(self, task):
         """Add a task to the job.
 
         Parameters
         ----------
         task : :class:`WinHPCTask`
             The task object to add.
         """
         self.tasks.append(task)
 
 
 class WinHPCTask(Configurable):
 
-    task_id = Str('')
-    task_name = Str('')
-    version = Str("2.000")
+    task_id = Unicode('')
+    task_name = Unicode('')
+    version = Unicode("2.000")
     min_cores = Int(1, config=True)
     max_cores = Int(1, config=True)
     min_sockets = Int(1, config=True)
     max_sockets = Int(1, config=True)
     min_nodes = Int(1, config=True)
     max_nodes = Int(1, config=True)
-    unit_type = Str("Core", config=True)
-    command_line = CStr('', config=True)
-    work_directory = CStr('', config=True)
+    unit_type = Unicode("Core", config=True)
+    command_line = Unicode('', config=True)
+    work_directory = Unicode('', config=True)
     is_rerunnaable = Bool(True, config=True)
-    std_out_file_path = CStr('', config=True)
-    std_err_file_path = CStr('', config=True)
+    std_out_file_path = Unicode('', config=True)
+    std_err_file_path = Unicode('', config=True)
     is_parametric = Bool(False, config=True)
     environment_variables = Instance(dict, args=(), config=True)
 
     def _write_attr(self, root, attr, key):
         s = as_str(getattr(self, attr, ''))
         if s:
             root.set(key, s)
 
     def as_element(self):
         root = ET.Element('Task')
         self._write_attr(root, 'version', '_A_Version')
         self._write_attr(root, 'task_name', '_B_Name')
         self._write_attr(root, 'min_cores', '_C_MinCores')
         self._write_attr(root, 'max_cores', '_D_MaxCores')
         self._write_attr(root, 'min_sockets', '_E_MinSockets')
         self._write_attr(root, 'max_sockets', '_F_MaxSockets')
         self._write_attr(root, 'min_nodes', '_G_MinNodes')
         self._write_attr(root, 'max_nodes', '_H_MaxNodes')
         self._write_attr(root, 'command_line', '_I_CommandLine')
         self._write_attr(root, 'work_directory', '_J_WorkDirectory')
         self._write_attr(root, 'is_rerunnaable', '_K_IsRerunnable')
         self._write_attr(root, 'std_out_file_path', '_L_StdOutFilePath')
         self._write_attr(root, 'std_err_file_path', '_M_StdErrFilePath')
         self._write_attr(root, 'is_parametric', '_N_IsParametric')
         self._write_attr(root, 'unit_type', '_O_UnitType')
         root.append(self.get_env_vars())
         return root
 
     def get_env_vars(self):
         env_vars = ET.Element('EnvironmentVariables')
         for k, v in self.environment_variables.iteritems():
             variable = ET.SubElement(env_vars, "Variable")
             name = ET.SubElement(variable, "Name")
             name.text = k
             value = ET.SubElement(variable, "Value")
             value.text = v
         return env_vars
 
 
 
 # By declaring these, we can configure the controller and engine separately!
 
 class IPControllerJob(WinHPCJob):
-    job_name = Str('IPController', config=False)
+    job_name = Unicode('IPController', config=False)
     is_exclusive = Bool(False, config=True)
-    username = Str(find_username(), config=True)
+    username = Unicode(find_username(), config=True)
     priority = Enum(('Lowest','BelowNormal','Normal','AboveNormal','Highest'),
         default_value='Highest', config=True)
-    requested_nodes = Str('', config=True)
-    project = Str('IPython', config=True)
+    requested_nodes = Unicode('', config=True)
+    project = Unicode('IPython', config=True)
 
 
 class IPEngineSetJob(WinHPCJob):
-    job_name = Str('IPEngineSet', config=False)
+    job_name = Unicode('IPEngineSet', config=False)
     is_exclusive = Bool(False, config=True)
-    username = Str(find_username(), config=True)
+    username = Unicode(find_username(), config=True)
     priority = Enum(('Lowest','BelowNormal','Normal','AboveNormal','Highest'),
         default_value='Highest', config=True)
-    requested_nodes = Str('', config=True)
-    project = Str('IPython', config=True)
+    requested_nodes = Unicode('', config=True)
+    project = Unicode('IPython', config=True)
 
 
 class IPControllerTask(WinHPCTask):
 
-    task_name = Str('IPController', config=True)
+    task_name = Unicode('IPController', config=True)
     controller_cmd = List(['ipcontroller.exe'], config=True)
     controller_args = List(['--log-to-file', '--log-level', '40'], config=True)
     # I don't want these to be configurable
-    std_out_file_path = CStr('', config=False)
-    std_err_file_path = CStr('', config=False)
+    std_out_file_path = Unicode('', config=False)
+    std_err_file_path = Unicode('', config=False)
     min_cores = Int(1, config=False)
     max_cores = Int(1, config=False)
     min_sockets = Int(1, config=False)
     max_sockets = Int(1, config=False)
     min_nodes = Int(1, config=False)
     max_nodes = Int(1, config=False)
-    unit_type = Str("Core", config=False)
-    work_directory = CStr('', config=False)
+    unit_type = Unicode("Core", config=False)
+    work_directory = Unicode('', config=False)
 
     def __init__(self, config=None):
         super(IPControllerTask, self).__init__(config=config)
         the_uuid = uuid.uuid1()
         self.std_out_file_path = os.path.join('log','ipcontroller-%s.out' % the_uuid)
         self.std_err_file_path = os.path.join('log','ipcontroller-%s.err' % the_uuid)
 
     @property
     def command_line(self):
         return ' '.join(self.controller_cmd + self.controller_args)
 
 
 class IPEngineTask(WinHPCTask):
 
-    task_name = Str('IPEngine', config=True)
+    task_name = Unicode('IPEngine', config=True)
     engine_cmd = List(['ipengine.exe'], config=True)
     engine_args = List(['--log-to-file', '--log-level', '40'], config=True)
     # I don't want these to be configurable
-    std_out_file_path = CStr('', config=False)
-    std_err_file_path = CStr('', config=False)
+    std_out_file_path = Unicode('', config=False)
+    std_err_file_path = Unicode('', config=False)
     min_cores = Int(1, config=False)
     max_cores = Int(1, config=False)
     min_sockets = Int(1, config=False)
     max_sockets = Int(1, config=False)
     min_nodes = Int(1, config=False)
     max_nodes = Int(1, config=False)
-    unit_type = Str("Core", config=False)
-    work_directory = CStr('', config=False)
+    unit_type = Unicode("Core", config=False)
+    work_directory = Unicode('', config=False)
 
     def __init__(self, config=None):
         super(IPEngineTask,self).__init__(config=config)
         the_uuid = uuid.uuid1()
         self.std_out_file_path = os.path.join('log','ipengine-%s.out' % the_uuid)
         self.std_err_file_path = os.path.join('log','ipengine-%s.err' % the_uuid)
     
     @property
     def command_line(self):
         return ' '.join(self.engine_cmd + self.engine_args)
 
 
 # j = WinHPCJob(None)
 # j.job_name = 'IPCluster'
 # j.username = 'GNET\\bgranger'
 # j.requested_nodes = 'GREEN'
 # 
 # t = WinHPCTask(None)
 # t.task_name = 'Controller'
 # t.command_line = r"\\blue\domainusers$\bgranger\Python\Python25\Scripts\ipcontroller.exe --log-to-file -p default --log-level 10"
 # t.work_directory = r"\\blue\domainusers$\bgranger\.ipython\cluster_default"
 # t.std_out_file_path = 'controller-out.txt'
 # t.std_err_file_path = 'controller-err.txt'
 # t.environment_variables['PYTHONPATH'] = r"\\blue\domainusers$\bgranger\Python\Python25\Lib\site-packages"
 # j.add_task(t)
 

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, CUnicode, 
-                                    Dict, List, Bool, Str, Set)
+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 .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=CUnicode('default')
+    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=Str()
+    _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,
             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)
         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):
         if ipython_dir is None:
             ipython_dir = get_ipython_dir()
         if cluster_dir is not None:
             try:
                 self._cd = ClusterDir.find_cluster_dir(cluster_dir)
                 return
             except ClusterDirError:
                 pass
         elif profile is not None:
             try:
                 self._cd = ClusterDir.find_cluster_dir_by_profile(
                     ipython_dir, profile)
                 return
             except ClusterDirError:
                 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/dictdb.py

@@ -1,180 +1,180 b''
 """A Task logger that presents our DB interface, 
 but exists entirely in memory and implemented with dicts.
 
 TaskRecords are dicts of the form:
 {
     'msg_id' : str(uuid),
     'client_uuid' : str(uuid),
     'engine_uuid' : str(uuid) or None,
     'header' : dict(header),
     'content': dict(content),
     'buffers': list(buffers),
     'submitted': datetime,
     'started': datetime or None,
     'completed': datetime or None,
     'resubmitted': datetime or None,
     'result_header' : dict(header) or None,
     'result_content' : dict(content) or None,
     'result_buffers' : list(buffers) or None,
 }
 With this info, many of the special categories of tasks can be defined by query:
 
 pending:  completed is None
 client's outstanding: client_uuid = uuid && completed is None
 MIA: arrived is None (and completed is None)
 etc.
 
 EngineRecords are dicts of the form:
 {
     'eid' : int(id),
     'uuid': str(uuid)
 }
 This may be extended, but is currently.
 
 We support a subset of mongodb operators:
     $lt,$gt,$lte,$gte,$ne,$in,$nin,$all,$mod,$exists
 """
 #-----------------------------------------------------------------------------
 #  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.
 #-----------------------------------------------------------------------------
 
 
 from datetime import datetime
 
 from IPython.config.configurable import Configurable
 
-from IPython.utils.traitlets import Dict, CUnicode
+from IPython.utils.traitlets import Dict, Unicode
 
 filters = {
  '$lt' : lambda a,b: a < b,
  '$gt' : lambda a,b: b > a,
  '$eq' : lambda a,b: a == b,
  '$ne' : lambda a,b: a != b,
  '$lte': lambda a,b: a <= b,
  '$gte': lambda a,b: a >= b,
  '$in' : lambda a,b: a in b,
  '$nin': lambda a,b: a not in b,
  '$all': lambda a,b: all([ a in bb for bb in b ]),
  '$mod': lambda a,b: a%b[0] == b[1],
  '$exists' : lambda a,b: (b and a is not None) or (a is None and not b)
 }
 
 
 class CompositeFilter(object):
     """Composite filter for matching multiple properties."""
     
     def __init__(self, dikt):
         self.tests = []
         self.values = []
         for key, value in dikt.iteritems():
             self.tests.append(filters[key])
             self.values.append(value)
 
     def __call__(self, value):
         for test,check in zip(self.tests, self.values):
             if not test(value, check):
                 return False
         return True
 
 class BaseDB(Configurable):
     """Empty Parent class so traitlets work on DB."""
     # base configurable traits:
-    session = CUnicode("")
+    session = Unicode("")
 
 class DictDB(BaseDB):
     """Basic in-memory dict-based object for saving Task Records.
     
     This is the first object to present the DB interface
     for logging tasks out of memory.
     
     The interface is based on MongoDB, so adding a MongoDB
     backend should be straightforward.
     """
     
     _records = Dict()
     
     def _match_one(self, rec, tests):
         """Check if a specific record matches tests."""
         for key,test in tests.iteritems():
             if not test(rec.get(key, None)):
                 return False
         return True
         
     def _match(self, check):
         """Find all the matches for a check dict."""
         matches = []
         tests = {}
         for k,v in check.iteritems():
             if isinstance(v, dict):
                 tests[k] = CompositeFilter(v)
             else:
                 tests[k] = lambda o: o==v
         
         for rec in self._records.itervalues():
             if self._match_one(rec, tests):
                 matches.append(rec)
         return matches
     
     def _extract_subdict(self, rec, keys):
         """extract subdict of keys"""
         d = {}
         d['msg_id'] = rec['msg_id']
         for key in keys:
             d[key] = rec[key]
         return d
     
     def add_record(self, msg_id, rec):
         """Add a new Task Record, by msg_id."""
         if self._records.has_key(msg_id):
             raise KeyError("Already have msg_id %r"%(msg_id))
         self._records[msg_id] = rec
     
     def get_record(self, msg_id):
         """Get a specific Task Record, by msg_id."""
         if not self._records.has_key(msg_id):
             raise KeyError("No such msg_id %r"%(msg_id))
         return self._records[msg_id]
     
     def update_record(self, msg_id, rec):
         """Update the data in an existing record."""
         self._records[msg_id].update(rec)
     
     def drop_matching_records(self, check):
         """Remove a record from the DB."""
         matches = self._match(check)
         for m in matches:
             del self._records[m['msg_id']]
         
     def drop_record(self, msg_id):
         """Remove a record from the DB."""
         del self._records[msg_id]
         
     
     def find_records(self, check, keys=None):
         """Find records matching a query dict, optionally extracting subset of keys.
         
         Returns dict keyed by msg_id 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.
         """
         matches = self._match(check)
         if keys:
             return [ self._extract_subdict(rec, keys) for rec in matches ]
         else:
             return matches
         
     
     def get_history(self):
         """get all msg_ids, ordered by time submitted."""
         msg_ids = self._records.keys()
         return sorted(msg_ids, key=lambda m: self._records[m]['submitted'])

IPython/parallel/controller/heartmonitor.py

@@ -1,165 +1,165 b''
 #!/usr/bin/env python
 """
 A multi-heart Heartbeat system using PUB and XREP sockets. pings are sent out on the PUB,
 and hearts are tracked based on their XREQ identities.
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-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.
 #-----------------------------------------------------------------------------
 
 from __future__ import print_function
 import time
 import uuid
 
 import zmq
-from zmq.devices import ProcessDevice, ThreadDevice
+from zmq.devices import ThreadDevice
 from zmq.eventloop import ioloop, zmqstream
 
-from IPython.utils.traitlets import Set, Instance, CFloat, Bool, CStr
+from IPython.utils.traitlets import Set, Instance, CFloat
 from IPython.parallel.factory import LoggingFactory
 
 class Heart(object):
     """A basic heart object for responding to a HeartMonitor.
     This is a simple wrapper with defaults for the most common
     Device model for responding to heartbeats.
     
     It simply builds a threadsafe zmq.FORWARDER Device, defaulting to using 
     SUB/XREQ for in/out.
     
     You can specify the XREQ's IDENTITY via the optional heart_id argument."""
     device=None
     id=None
     def __init__(self, in_addr, out_addr, in_type=zmq.SUB, out_type=zmq.XREQ, heart_id=None):
         self.device = ThreadDevice(zmq.FORWARDER, in_type, out_type)
         self.device.daemon=True
         self.device.connect_in(in_addr)
         self.device.connect_out(out_addr)
         if in_type == zmq.SUB:
             self.device.setsockopt_in(zmq.SUBSCRIBE, "")
         if heart_id is None:
             heart_id = str(uuid.uuid4())
         self.device.setsockopt_out(zmq.IDENTITY, heart_id)
         self.id = heart_id
     
     def start(self):
         return self.device.start()
         
 class HeartMonitor(LoggingFactory):
     """A basic HeartMonitor class
     pingstream: a PUB stream
     pongstream: an XREP stream
     period: the period of the heartbeat in milliseconds"""
     
     period=CFloat(1000, config=True,
         help='The frequency at which the Hub pings the engines for heartbeats '
         ' (in ms) [default: 100]',
     )
     
     pingstream=Instance('zmq.eventloop.zmqstream.ZMQStream')
     pongstream=Instance('zmq.eventloop.zmqstream.ZMQStream')
     loop = Instance('zmq.eventloop.ioloop.IOLoop')
     def _loop_default(self):
         return ioloop.IOLoop.instance()
     
     # not settable:
     hearts=Set()
     responses=Set()
     on_probation=Set()
     last_ping=CFloat(0)
     _new_handlers = Set()
     _failure_handlers = Set()
     lifetime = CFloat(0)
     tic = CFloat(0)
     
     def __init__(self, **kwargs):
         super(HeartMonitor, self).__init__(**kwargs)
         
         self.pongstream.on_recv(self.handle_pong)
     
     def start(self):
         self.caller = ioloop.PeriodicCallback(self.beat, self.period, self.loop)
         self.caller.start()
     
     def add_new_heart_handler(self, handler):
         """add a new handler for new hearts"""
         self.log.debug("heartbeat::new_heart_handler: %s"%handler)
         self._new_handlers.add(handler)
         
     def add_heart_failure_handler(self, handler):
         """add a new handler for heart failure"""
         self.log.debug("heartbeat::new heart failure handler: %s"%handler)
         self._failure_handlers.add(handler)
             
     def beat(self):
         self.pongstream.flush() 
         self.last_ping = self.lifetime
         
         toc = time.time()
         self.lifetime += toc-self.tic
         self.tic = toc
         # self.log.debug("heartbeat::%s"%self.lifetime)
         goodhearts = self.hearts.intersection(self.responses)
         missed_beats = self.hearts.difference(goodhearts)
         heartfailures = self.on_probation.intersection(missed_beats)
         newhearts = self.responses.difference(goodhearts)
         map(self.handle_new_heart, newhearts)
         map(self.handle_heart_failure, heartfailures)
         self.on_probation = missed_beats.intersection(self.hearts)
         self.responses = set()
         # print self.on_probation, self.hearts
         # self.log.debug("heartbeat::beat %.3f, %i beating hearts"%(self.lifetime, len(self.hearts)))
         self.pingstream.send(str(self.lifetime))
     
     def handle_new_heart(self, heart):
         if self._new_handlers:
             for handler in self._new_handlers:
                 handler(heart)
         else:
             self.log.info("heartbeat::yay, got new heart %s!"%heart)
         self.hearts.add(heart)
     
     def handle_heart_failure(self, heart):
         if self._failure_handlers:
             for handler in self._failure_handlers:
                 try:
                     handler(heart)
                 except Exception as e:
                     self.log.error("heartbeat::Bad Handler! %s"%handler, exc_info=True)
                     pass
         else:
             self.log.info("heartbeat::Heart %s failed :("%heart)
         self.hearts.remove(heart)
         
     
     def handle_pong(self, msg):
         "a heart just beat"
         if msg[1] == str(self.lifetime):
             delta = time.time()-self.tic
             # self.log.debug("heartbeat::heart %r took %.2f ms to respond"%(msg[0], 1000*delta))
             self.responses.add(msg[0])
         elif msg[1] == str(self.last_ping):
             delta = time.time()-self.tic + (self.lifetime-self.last_ping)
             self.log.warn("heartbeat::heart %r missed a beat, and took %.2f ms to respond"%(msg[0], 1000*delta))
             self.responses.add(msg[0])
         else:
             self.log.warn("heartbeat::got bad heartbeat (possibly old?): %s (current=%.3f)"%
             (msg[1],self.lifetime))
 
 
 if __name__ == '__main__':
     loop = ioloop.IOLoop.instance()
     context = zmq.Context()
     pub = context.socket(zmq.PUB)
     pub.bind('tcp://127.0.0.1:5555')
     xrep = context.socket(zmq.XREP)
     xrep.bind('tcp://127.0.0.1:5556')
     
     outstream = zmqstream.ZMQStream(pub, loop)
     instream = zmqstream.ZMQStream(xrep, loop)
     
     hb = HeartMonitor(loop, outstream, instream)
     
     loop.start()

IPython/parallel/controller/hub.py

@@ -1,1293 +1,1293 b''
 #!/usr/bin/env python
 """The IPython Controller Hub with 0MQ
 This is the master object that handles connections from engines and clients,
 and monitors traffic through the various queues.
 """
 #-----------------------------------------------------------------------------
 #  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
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 import sys
 import time
 from datetime import datetime
 
 import zmq
 from zmq.eventloop import ioloop
 from zmq.eventloop.zmqstream import ZMQStream
 
 # internal:
 from IPython.utils.importstring import import_item
 from IPython.utils.traitlets import (
-        HasTraits, Instance, Int, CStr, Str, Dict, Set, List, Bool, Tuple
+        HasTraits, Instance, Int, Unicode, Dict, Set, Tuple, CStr
         )
 
 from IPython.parallel import error, util
 from IPython.parallel.factory import RegistrationFactory, LoggingFactory
 
 from .heartmonitor import HeartMonitor
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 def _passer(*args, **kwargs):
     return
 
 def _printer(*args, **kwargs):
     print (args)
     print (kwargs)
 
 def empty_record():
     """Return an empty dict with all record keys."""
     return {
         'msg_id' : None,
         'header' : None,
         'content': None,
         'buffers': None,
         'submitted': None,
         'client_uuid' : None,
         'engine_uuid' : None,
         'started': None,
         'completed': None,
         'resubmitted': None,
         'result_header' : None,
         'result_content' : None,
         'result_buffers' : None,
         'queue' : None,
         'pyin' : None,
         'pyout': None,
         'pyerr': None,
         'stdout': '',
         'stderr': '',
     }
     
 def init_record(msg):
     """Initialize a TaskRecord based on a request."""
     header = msg['header']
     return {
         'msg_id' : header['msg_id'],
         'header' : header,
         'content': msg['content'],
         'buffers': msg['buffers'],
         'submitted': datetime.strptime(header['date'], util.ISO8601),
         'client_uuid' : None,
         'engine_uuid' : None,
         'started': None,
         'completed': None,
         'resubmitted': None,
         'result_header' : None,
         'result_content' : None,
         'result_buffers' : None,
         'queue' : None,
         'pyin' : None,
         'pyout': None,
         'pyerr': None,
         'stdout': '',
         'stderr': '',
     }
 
 
 class EngineConnector(HasTraits):
     """A simple object for accessing the various zmq connections of an object.
     Attributes are:
     id (int): engine ID
     uuid (str): uuid (unused?)
     queue (str): identity of queue's XREQ socket
     registration (str): identity of registration XREQ socket
     heartbeat (str): identity of heartbeat XREQ socket
     """
     id=Int(0)
-    queue=Str()
-    control=Str()
-    registration=Str()
-    heartbeat=Str()
+    queue=CStr()
+    control=CStr()
+    registration=CStr()
+    heartbeat=CStr()
     pending=Set()
 
 class HubFactory(RegistrationFactory):
     """The Configurable for setting up a Hub."""
     
     # port-pairs for monitoredqueues:
     hb = Tuple(Int,Int,config=True,
         help="""XREQ/SUB Port pair for Engine heartbeats""")
     def _hb_default(self):
         return tuple(util.select_random_ports(2))
 
     mux = Tuple(Int,Int,config=True,
         help="""Engine/Client Port pair for MUX queue""")
     
     def _mux_default(self):
         return tuple(util.select_random_ports(2))
     
     task = Tuple(Int,Int,config=True,
         help="""Engine/Client Port pair for Task queue""")
     def _task_default(self):
         return tuple(util.select_random_ports(2))
 
     control = Tuple(Int,Int,config=True,
         help="""Engine/Client Port pair for Control queue""")
     
     def _control_default(self):
         return tuple(util.select_random_ports(2))
 
     iopub = Tuple(Int,Int,config=True,
         help="""Engine/Client Port pair for IOPub relay""")
     
     def _iopub_default(self):
         return tuple(util.select_random_ports(2))
     
     # single ports:
     mon_port = Int(config=True,
         help="""Monitor (SUB) port for queue traffic""")
 
     def _mon_port_default(self):
         return util.select_random_ports(1)[0]
     
     notifier_port = Int(config=True,
         help="""PUB port for sending engine status notifications""")
 
     def _notifier_port_default(self):
         return util.select_random_ports(1)[0]
     
-    engine_ip = CStr('127.0.0.1', config=True,
+    engine_ip = Unicode('127.0.0.1', config=True,
         help="IP on which to listen for engine connections. [default: loopback]")
-    engine_transport = CStr('tcp', config=True,
+    engine_transport = Unicode('tcp', config=True,
         help="0MQ transport for engine connections. [default: tcp]")
     
-    client_ip = CStr('127.0.0.1', config=True,
+    client_ip = Unicode('127.0.0.1', config=True,
         help="IP on which to listen for client connections. [default: loopback]")
-    client_transport = CStr('tcp', config=True,
+    client_transport = Unicode('tcp', config=True,
         help="0MQ transport for client connections. [default : tcp]")
     
-    monitor_ip = CStr('127.0.0.1', config=True,
+    monitor_ip = Unicode('127.0.0.1', config=True,
         help="IP on which to listen for monitor messages. [default: loopback]")
-    monitor_transport = CStr('tcp', config=True,
+    monitor_transport = Unicode('tcp', config=True,
         help="0MQ transport for monitor messages. [default : tcp]")
     
-    monitor_url = CStr('')
+    monitor_url = Unicode('')
     
-    db_class = CStr('IPython.parallel.controller.dictdb.DictDB', config=True,
+    db_class = Unicode('IPython.parallel.controller.dictdb.DictDB', config=True,
         help="""The class to use for the DB backend""")
     
     # not configurable
     db = Instance('IPython.parallel.controller.dictdb.BaseDB')
     heartmonitor = Instance('IPython.parallel.controller.heartmonitor.HeartMonitor')
     
     def _ip_changed(self, name, old, new):
         self.engine_ip = new
         self.client_ip = new
         self.monitor_ip = new
         self._update_monitor_url()
     
     def _update_monitor_url(self):
         self.monitor_url = "%s://%s:%i"%(self.monitor_transport, self.monitor_ip, self.mon_port)
     
     def _transport_changed(self, name, old, new):
         self.engine_transport = new
         self.client_transport = new
         self.monitor_transport = new
         self._update_monitor_url()
         
     def __init__(self, **kwargs):
         super(HubFactory, self).__init__(**kwargs)
         self._update_monitor_url()
         # self.on_trait_change(self._sync_ips, 'ip')
         # self.on_trait_change(self._sync_transports, 'transport')
         # self.subconstructors.append(self.construct_hub)
     
     
     def construct(self):
         self.init_hub()
     
     def start(self):
         self.heartmonitor.start()
         self.log.info("Heartmonitor started")
     
     def init_hub(self):
         """construct"""
         client_iface = "%s://%s:"%(self.client_transport, self.client_ip) + "%i"
         engine_iface = "%s://%s:"%(self.engine_transport, self.engine_ip) + "%i"
         
         ctx = self.context
         loop = self.loop
         
         # Registrar socket
         q = ZMQStream(ctx.socket(zmq.XREP), loop)
         q.bind(client_iface % self.regport)
         self.log.info("Hub listening on %s for registration."%(client_iface%self.regport))
         if self.client_ip != self.engine_ip:
             q.bind(engine_iface % self.regport)
             self.log.info("Hub listening on %s for registration."%(engine_iface%self.regport))
         
         ### Engine connections ###
 
         # heartbeat
         hpub = ctx.socket(zmq.PUB)
         hpub.bind(engine_iface % self.hb[0])
         hrep = ctx.socket(zmq.XREP)
         hrep.bind(engine_iface % self.hb[1])
         self.heartmonitor = HeartMonitor(loop=loop, pingstream=ZMQStream(hpub,loop), pongstream=ZMQStream(hrep,loop), 
                                 config=self.config)
 
         ### Client connections ###
         # Notifier socket
         n = ZMQStream(ctx.socket(zmq.PUB), loop)
         n.bind(client_iface%self.notifier_port)
 
         ### build and launch the queues ###
 
         # monitor socket
         sub = ctx.socket(zmq.SUB)
         sub.setsockopt(zmq.SUBSCRIBE, "")
         sub.bind(self.monitor_url)
         sub.bind('inproc://monitor')
         sub = ZMQStream(sub, loop)
         
         # connect the db
         self.log.info('Hub using DB backend: %r'%(self.db_class.split()[-1]))
         # cdir = self.config.Global.cluster_dir
-        self.db = import_item(self.db_class)(session=self.session.session, config=self.config)
+        self.db = import_item(str(self.db_class))(session=self.session.session, config=self.config)
         time.sleep(.25)
         try:
             scheme = self.config.TaskScheduler.scheme_name
         except AttributeError:
             from .scheduler import TaskScheduler
             scheme = TaskScheduler.scheme_name.get_default_value()
         # build connection dicts
         self.engine_info = {
             'control' : engine_iface%self.control[1],
             'mux': engine_iface%self.mux[1],
             'heartbeat': (engine_iface%self.hb[0], engine_iface%self.hb[1]),
             'task' : engine_iface%self.task[1],
             'iopub' : engine_iface%self.iopub[1],
             # 'monitor' : engine_iface%self.mon_port,
             }
 
         self.client_info = {
             'control' : client_iface%self.control[0],
             'mux': client_iface%self.mux[0],
             'task' : (scheme, client_iface%self.task[0]),
             'iopub' : client_iface%self.iopub[0],
             'notification': client_iface%self.notifier_port
             }
         self.log.debug("Hub engine addrs: %s"%self.engine_info)
         self.log.debug("Hub client addrs: %s"%self.client_info)
 
         # resubmit stream
         r = ZMQStream(ctx.socket(zmq.XREQ), loop)
         url = util.disambiguate_url(self.client_info['task'][-1])
         r.setsockopt(zmq.IDENTITY, self.session.session)
         r.connect(url)
 
         self.hub = Hub(loop=loop, session=self.session, monitor=sub, heartmonitor=self.heartmonitor,
                 query=q, notifier=n, resubmit=r, db=self.db,
                 engine_info=self.engine_info, client_info=self.client_info,
                 logname=self.log.name)
     
 
 class Hub(LoggingFactory):
     """The IPython Controller Hub with 0MQ connections
     
     Parameters
     ==========
     loop: zmq IOLoop instance
     session: StreamSession object
     <removed> context: zmq context for creating new connections (?)
     queue: ZMQStream for monitoring the command queue (SUB)
     query: ZMQStream for engine registration and client queries requests (XREP)
     heartbeat: HeartMonitor object checking the pulse of the engines
     notifier: ZMQStream for broadcasting engine registration changes (PUB)
     db: connection to db for out of memory logging of commands
                 NotImplemented
     engine_info: dict of zmq connection information for engines to connect
                 to the queues.
     client_info: dict of zmq connection information for engines to connect
                 to the queues.
     """
     # internal data structures:
     ids=Set() # engine IDs
     keytable=Dict()
     by_ident=Dict()
     engines=Dict()
     clients=Dict()
     hearts=Dict()
     pending=Set()
     queues=Dict()  # pending msg_ids keyed by engine_id
     tasks=Dict() # pending msg_ids submitted as tasks, keyed by client_id
     completed=Dict() # completed msg_ids keyed by engine_id
     all_completed=Set() # completed msg_ids keyed by engine_id
     dead_engines=Set() # completed msg_ids keyed by engine_id
     unassigned=Set() # set of task msg_ds not yet assigned a destination
     incoming_registrations=Dict()
     registration_timeout=Int()
     _idcounter=Int(0)
     
     # objects from constructor:
     loop=Instance(ioloop.IOLoop)
     query=Instance(ZMQStream)
     monitor=Instance(ZMQStream)
     notifier=Instance(ZMQStream)
     resubmit=Instance(ZMQStream)
     heartmonitor=Instance(HeartMonitor)
     db=Instance(object)
     client_info=Dict()
     engine_info=Dict()
     
     
     def __init__(self, **kwargs):
         """
         # universal:
         loop: IOLoop for creating future connections
         session: streamsession for sending serialized data
         # engine:
         queue: ZMQStream for monitoring queue messages
         query: ZMQStream for engine+client registration and client requests
         heartbeat: HeartMonitor object for tracking engines
         # extra:
         db: ZMQStream for db connection (NotImplemented)
         engine_info: zmq address/protocol dict for engine connections
         client_info: zmq address/protocol dict for client connections
         """
         
         super(Hub, self).__init__(**kwargs)
         self.registration_timeout = max(5000, 2*self.heartmonitor.period)
         
         # validate connection dicts:
         for k,v in self.client_info.iteritems():
             if k == 'task':
                 util.validate_url_container(v[1])
             else:
                 util.validate_url_container(v)
         # util.validate_url_container(self.client_info)
         util.validate_url_container(self.engine_info)
         
         # register our callbacks
         self.query.on_recv(self.dispatch_query)
         self.monitor.on_recv(self.dispatch_monitor_traffic)
         
         self.heartmonitor.add_heart_failure_handler(self.handle_heart_failure)
         self.heartmonitor.add_new_heart_handler(self.handle_new_heart)
         
         self.monitor_handlers = { 'in' : self.save_queue_request,
                                 'out': self.save_queue_result,
                                 'intask': self.save_task_request,
                                 'outtask': self.save_task_result,
                                 'tracktask': self.save_task_destination,
                                 'incontrol': _passer,
                                 'outcontrol': _passer,
                                 'iopub': self.save_iopub_message,
         }
         
         self.query_handlers = {'queue_request': self.queue_status,
                                 'result_request': self.get_results,
                                 'history_request': self.get_history,
                                 'db_request': self.db_query,
                                 'purge_request': self.purge_results,
                                 'load_request': self.check_load,
                                 'resubmit_request': self.resubmit_task,
                                 'shutdown_request': self.shutdown_request,
                                 'registration_request' : self.register_engine,
                                 'unregistration_request' : self.unregister_engine,
                                 'connection_request': self.connection_request,
         }
         
         # ignore resubmit replies
         self.resubmit.on_recv(lambda msg: None, copy=False)
 
         self.log.info("hub::created hub")
     
     @property
     def _next_id(self):
         """gemerate a new ID.
         
         No longer reuse old ids, just count from 0."""
         newid = self._idcounter
         self._idcounter += 1
         return newid
         # newid = 0
         # incoming = [id[0] for id in self.incoming_registrations.itervalues()]
         # # print newid, self.ids, self.incoming_registrations
         # while newid in self.ids or newid in incoming:
         #     newid += 1
         # return newid
     
     #-----------------------------------------------------------------------------
     # message validation
     #-----------------------------------------------------------------------------
     
     def _validate_targets(self, targets):
         """turn any valid targets argument into a list of integer ids"""
         if targets is None:
             # default to all
             targets = self.ids
             
         if isinstance(targets, (int,str,unicode)):
             # only one target specified
             targets = [targets]
         _targets = []
         for t in targets:
             # map raw identities to ids
             if isinstance(t, (str,unicode)):
                 t = self.by_ident.get(t, t)
             _targets.append(t)
         targets = _targets
         bad_targets = [ t for t in targets if t not in self.ids ]
         if bad_targets:
             raise IndexError("No Such Engine: %r"%bad_targets)
         if not targets:
             raise IndexError("No Engines Registered")
         return targets
     
     #-----------------------------------------------------------------------------
     # dispatch methods (1 per stream)
     #-----------------------------------------------------------------------------
     
     # def dispatch_registration_request(self, msg):
     #     """"""
     #     self.log.debug("registration::dispatch_register_request(%s)"%msg)
     #     idents,msg = self.session.feed_identities(msg)
     #     if not idents:
     #         self.log.error("Bad Query Message: %s"%msg, exc_info=True)
     #         return
     #     try:
     #         msg = self.session.unpack_message(msg,content=True)
     #     except:
     #         self.log.error("registration::got bad registration message: %s"%msg, exc_info=True)
     #         return
     #     
     #     msg_type = msg['msg_type']
     #     content = msg['content']
     #     
     #     handler = self.query_handlers.get(msg_type, None)
     #     if handler is None:
     #         self.log.error("registration::got bad registration message: %s"%msg)
     #     else:
     #         handler(idents, msg)
     
     def dispatch_monitor_traffic(self, msg):
         """all ME and Task queue messages come through here, as well as
         IOPub traffic."""
         self.log.debug("monitor traffic: %r"%msg[:2])
         switch = msg[0]
         idents, msg = self.session.feed_identities(msg[1:])
         if not idents:
             self.log.error("Bad Monitor Message: %r"%msg)
             return
         handler = self.monitor_handlers.get(switch, None)
         if handler is not None:
             handler(idents, msg)
         else:
             self.log.error("Invalid monitor topic: %r"%switch)
         
     
     def dispatch_query(self, msg):
         """Route registration requests and queries from clients."""
         idents, msg = self.session.feed_identities(msg)
         if not idents:
             self.log.error("Bad Query Message: %r"%msg)
             return
         client_id = idents[0]
         try:
             msg = self.session.unpack_message(msg, content=True)
         except:
             content = error.wrap_exception()
             self.log.error("Bad Query Message: %r"%msg, exc_info=True)
             self.session.send(self.query, "hub_error", ident=client_id, 
                     content=content)
             return
         
         # print client_id, header, parent, content
         #switch on message type:
         msg_type = msg['msg_type']
         self.log.info("client::client %r requested %r"%(client_id, msg_type))
         handler = self.query_handlers.get(msg_type, None)
         try:
             assert handler is not None, "Bad Message Type: %r"%msg_type
         except:
             content = error.wrap_exception()
             self.log.error("Bad Message Type: %r"%msg_type, exc_info=True)
             self.session.send(self.query, "hub_error", ident=client_id, 
                     content=content)
             return
 
         else:
             handler(idents, msg)
             
     def dispatch_db(self, msg):
         """"""
         raise NotImplementedError
     
     #---------------------------------------------------------------------------
     # handler methods (1 per event)
     #---------------------------------------------------------------------------
     
     #----------------------- Heartbeat --------------------------------------
     
     def handle_new_heart(self, heart):
         """handler to attach to heartbeater.
         Called when a new heart starts to beat.
         Triggers completion of registration."""
         self.log.debug("heartbeat::handle_new_heart(%r)"%heart)
         if heart not in self.incoming_registrations:
             self.log.info("heartbeat::ignoring new heart: %r"%heart)
         else:
             self.finish_registration(heart)
         
     
     def handle_heart_failure(self, heart):
         """handler to attach to heartbeater.
         called when a previously registered heart fails to respond to beat request.
         triggers unregistration"""
         self.log.debug("heartbeat::handle_heart_failure(%r)"%heart)
         eid = self.hearts.get(heart, None)
         queue = self.engines[eid].queue
         if eid is None:
             self.log.info("heartbeat::ignoring heart failure %r"%heart)
         else:
             self.unregister_engine(heart, dict(content=dict(id=eid, queue=queue)))
     
     #----------------------- MUX Queue Traffic ------------------------------
     
     def save_queue_request(self, idents, msg):
         if len(idents) < 2:
             self.log.error("invalid identity prefix: %s"%idents)
             return
         queue_id, client_id = idents[:2]
         try:
             msg = self.session.unpack_message(msg, content=False)
         except:
             self.log.error("queue::client %r sent invalid message to %r: %s"%(client_id, queue_id, msg), exc_info=True)
             return
         
         eid = self.by_ident.get(queue_id, None)
         if eid is None:
             self.log.error("queue::target %r not registered"%queue_id)
             self.log.debug("queue::    valid are: %s"%(self.by_ident.keys()))
             return
             
         header = msg['header']
         msg_id = header['msg_id']
         record = init_record(msg)
         record['engine_uuid'] = queue_id
         record['client_uuid'] = client_id
         record['queue'] = 'mux'
 
         try:
             # it's posible iopub arrived first:
             existing = self.db.get_record(msg_id)
             for key,evalue in existing.iteritems():
                 rvalue = record.get(key, None)
                 if evalue and rvalue and evalue != rvalue:
                     self.log.warn("conflicting initial state for record: %r:%r <%r> %r"%(msg_id, rvalue, key, evalue))
                 elif evalue and not rvalue:
                     record[key] = evalue
             self.db.update_record(msg_id, record)
         except KeyError:
             self.db.add_record(msg_id, record)
         
         self.pending.add(msg_id)
         self.queues[eid].append(msg_id)
     
     def save_queue_result(self, idents, msg):
         if len(idents) < 2:
             self.log.error("invalid identity prefix: %s"%idents)
             return
             
         client_id, queue_id = idents[:2]
         try:
             msg = self.session.unpack_message(msg, content=False)
         except:
             self.log.error("queue::engine %r sent invalid message to %r: %s"%(
                     queue_id,client_id, msg), exc_info=True)
             return
         
         eid = self.by_ident.get(queue_id, None)
         if eid is None:
             self.log.error("queue::unknown engine %r is sending a reply: "%queue_id)
             # self.log.debug("queue::       %s"%msg[2:])
             return
         
         parent = msg['parent_header']
         if not parent:
             return
         msg_id = parent['msg_id']
         if msg_id in self.pending:
             self.pending.remove(msg_id)
             self.all_completed.add(msg_id)
             self.queues[eid].remove(msg_id)
             self.completed[eid].append(msg_id)
         elif msg_id not in self.all_completed:
             # it could be a result from a dead engine that died before delivering the
             # result
             self.log.warn("queue:: unknown msg finished %s"%msg_id)
             return
         # update record anyway, because the unregistration could have been premature
         rheader = msg['header']
         completed = datetime.strptime(rheader['date'], util.ISO8601)
         started = rheader.get('started', None)
         if started is not None:
             started = datetime.strptime(started, util.ISO8601)
         result = {
             'result_header' : rheader,
             'result_content': msg['content'],
             'started' : started,
             'completed' : completed
         }
 
         result['result_buffers'] = msg['buffers']
         try:
             self.db.update_record(msg_id, result)
         except Exception:
             self.log.error("DB Error updating record %r"%msg_id, exc_info=True)
         
             
     #--------------------- Task Queue Traffic ------------------------------
     
     def save_task_request(self, idents, msg):
         """Save the submission of a task."""
         client_id = idents[0]
         
         try:
             msg = self.session.unpack_message(msg, content=False)
         except:
             self.log.error("task::client %r sent invalid task message: %s"%(
                     client_id, msg), exc_info=True)
             return
         record = init_record(msg)
 
         record['client_uuid'] = client_id
         record['queue'] = 'task'
         header = msg['header']
         msg_id = header['msg_id']
         self.pending.add(msg_id)
         self.unassigned.add(msg_id)
         try:
             # it's posible iopub arrived first:
             existing = self.db.get_record(msg_id)
             if existing['resubmitted']:
                 for key in ('submitted', 'client_uuid', 'buffers'):
                     # don't clobber these keys on resubmit
                     # submitted and client_uuid should be different
                     # and buffers might be big, and shouldn't have changed
                     record.pop(key)
                     # still check content,header which should not change
                     # but are not expensive to compare as buffers
 
             for key,evalue in existing.iteritems():
                 if key.endswith('buffers'):
                     # don't compare buffers
                     continue
                 rvalue = record.get(key, None)
                 if evalue and rvalue and evalue != rvalue:
                     self.log.warn("conflicting initial state for record: %r:%r <%r> %r"%(msg_id, rvalue, key, evalue))
                 elif evalue and not rvalue:
                     record[key] = evalue
             self.db.update_record(msg_id, record)
         except KeyError:
             self.db.add_record(msg_id, record)
         except Exception:
             self.log.error("DB Error saving task request %r"%msg_id, exc_info=True)
     
     def save_task_result(self, idents, msg):
         """save the result of a completed task."""
         client_id = idents[0]
         try:
             msg = self.session.unpack_message(msg, content=False)
         except:
             self.log.error("task::invalid task result message send to %r: %s"%(
                     client_id, msg), exc_info=True)
             raise
             return
         
         parent = msg['parent_header']
         if not parent:
             # print msg
             self.log.warn("Task %r had no parent!"%msg)
             return
         msg_id = parent['msg_id']
         if msg_id in self.unassigned:
             self.unassigned.remove(msg_id)
         
         header = msg['header']
         engine_uuid = header.get('engine', None)
         eid = self.by_ident.get(engine_uuid, None)
         
         if msg_id in self.pending:
             self.pending.remove(msg_id)
             self.all_completed.add(msg_id)
             if eid is not None:
                 self.completed[eid].append(msg_id)
                 if msg_id in self.tasks[eid]:
                     self.tasks[eid].remove(msg_id)
             completed = datetime.strptime(header['date'], util.ISO8601)
             started = header.get('started', None)
             if started is not None:
                 started = datetime.strptime(started, util.ISO8601)
             result = {
                 'result_header' : header,
                 'result_content': msg['content'],
                 'started' : started,
                 'completed' : completed,
                 'engine_uuid': engine_uuid
             }
 
             result['result_buffers'] = msg['buffers']
             try:
                 self.db.update_record(msg_id, result)
             except Exception:
                 self.log.error("DB Error saving task request %r"%msg_id, exc_info=True)
             
         else:
             self.log.debug("task::unknown task %s finished"%msg_id)
     
     def save_task_destination(self, idents, msg):
         try:
             msg = self.session.unpack_message(msg, content=True)
         except:
             self.log.error("task::invalid task tracking message", exc_info=True)
             return
         content = msg['content']
         # print (content)
         msg_id = content['msg_id']
         engine_uuid = content['engine_id']
         eid = self.by_ident[engine_uuid]
         
         self.log.info("task::task %s arrived on %s"%(msg_id, eid))
         if msg_id in self.unassigned:
             self.unassigned.remove(msg_id)
         # else:
         #     self.log.debug("task::task %s not listed as MIA?!"%(msg_id))
         
         self.tasks[eid].append(msg_id)
         # self.pending[msg_id][1].update(received=datetime.now(),engine=(eid,engine_uuid))
         try:
             self.db.update_record(msg_id, dict(engine_uuid=engine_uuid))
         except Exception:
             self.log.error("DB Error saving task destination %r"%msg_id, exc_info=True)
             
     
     def mia_task_request(self, idents, msg):
         raise NotImplementedError
         client_id = idents[0]
         # content = dict(mia=self.mia,status='ok')
         # self.session.send('mia_reply', content=content, idents=client_id)
         
         
     #--------------------- IOPub Traffic ------------------------------
     
     def save_iopub_message(self, topics, msg):
         """save an iopub message into the db"""
         # print (topics)
         try:
             msg = self.session.unpack_message(msg, content=True)
         except:
             self.log.error("iopub::invalid IOPub message", exc_info=True)
             return
         
         parent = msg['parent_header']
         if not parent:
             self.log.error("iopub::invalid IOPub message: %s"%msg)
             return
         msg_id = parent['msg_id']
         msg_type = msg['msg_type']
         content = msg['content']
         
         # ensure msg_id is in db
         try:
             rec = self.db.get_record(msg_id)
         except KeyError:
             rec = empty_record()
             rec['msg_id'] = msg_id
             self.db.add_record(msg_id, rec)
         # stream
         d = {}
         if msg_type == 'stream':
             name = content['name']
             s = rec[name] or ''
             d[name] = s + content['data']
             
         elif msg_type == 'pyerr':
             d['pyerr'] = content
         elif msg_type == 'pyin':
             d['pyin'] = content['code']
         else:
             d[msg_type] = content.get('data', '')
         
         try:
             self.db.update_record(msg_id, d)
         except Exception:
             self.log.error("DB Error saving iopub message %r"%msg_id, exc_info=True)
         
     
             
     #-------------------------------------------------------------------------
     # Registration requests
     #-------------------------------------------------------------------------
         
     def connection_request(self, client_id, msg):
         """Reply with connection addresses for clients."""
         self.log.info("client::client %s connected"%client_id)
         content = dict(status='ok')
         content.update(self.client_info)
         jsonable = {}
         for k,v in self.keytable.iteritems():
             if v not in self.dead_engines:
                 jsonable[str(k)] = v
         content['engines'] = jsonable
         self.session.send(self.query, 'connection_reply', content, parent=msg, ident=client_id)
     
     def register_engine(self, reg, msg):
         """Register a new engine."""
         content = msg['content']
         try:
             queue = content['queue']
         except KeyError:
             self.log.error("registration::queue not specified", exc_info=True)
             return
         heart = content.get('heartbeat', None)
         """register a new engine, and create the socket(s) necessary"""
         eid = self._next_id
         # print (eid, queue, reg, heart)
         
         self.log.debug("registration::register_engine(%i, %r, %r, %r)"%(eid, queue, reg, heart))
         
         content = dict(id=eid,status='ok')
         content.update(self.engine_info)
         # check if requesting available IDs:
         if queue in self.by_ident:
             try:
                 raise KeyError("queue_id %r in use"%queue)
             except:
                 content = error.wrap_exception()
                 self.log.error("queue_id %r in use"%queue, exc_info=True)
         elif heart in self.hearts: # need to check unique hearts?
             try:
                 raise KeyError("heart_id %r in use"%heart)
             except:
                 self.log.error("heart_id %r in use"%heart, exc_info=True)
                 content = error.wrap_exception()
         else:
             for h, pack in self.incoming_registrations.iteritems():
                 if heart == h:
                     try:
                         raise KeyError("heart_id %r in use"%heart)
                     except:
                         self.log.error("heart_id %r in use"%heart, exc_info=True)
                         content = error.wrap_exception()
                     break
                 elif queue == pack[1]:
                     try:
                         raise KeyError("queue_id %r in use"%queue)
                     except:
                         self.log.error("queue_id %r in use"%queue, exc_info=True)
                         content = error.wrap_exception()
                     break
         
         msg = self.session.send(self.query, "registration_reply", 
                 content=content, 
                 ident=reg)
         
         if content['status'] == 'ok':
             if heart in self.heartmonitor.hearts:
                 # already beating
                 self.incoming_registrations[heart] = (eid,queue,reg[0],None)
                 self.finish_registration(heart)
             else:
                 purge = lambda : self._purge_stalled_registration(heart)
                 dc = ioloop.DelayedCallback(purge, self.registration_timeout, self.loop)
                 dc.start()
                 self.incoming_registrations[heart] = (eid,queue,reg[0],dc)
         else:
             self.log.error("registration::registration %i failed: %s"%(eid, content['evalue']))
         return eid
     
     def unregister_engine(self, ident, msg):
         """Unregister an engine that explicitly requested to leave."""
         try:
             eid = msg['content']['id']
         except:
             self.log.error("registration::bad engine id for unregistration: %s"%ident, exc_info=True)
             return
         self.log.info("registration::unregister_engine(%s)"%eid)
         # print (eid)
         uuid = self.keytable[eid]
         content=dict(id=eid, queue=uuid)
         self.dead_engines.add(uuid)
         # self.ids.remove(eid)
         # uuid = self.keytable.pop(eid)
         # 
         # ec = self.engines.pop(eid)
         # self.hearts.pop(ec.heartbeat)
         # self.by_ident.pop(ec.queue)
         # self.completed.pop(eid)
         handleit = lambda : self._handle_stranded_msgs(eid, uuid)
         dc = ioloop.DelayedCallback(handleit, self.registration_timeout, self.loop)
         dc.start()
         ############## TODO: HANDLE IT ################
         
         if self.notifier:
             self.session.send(self.notifier, "unregistration_notification", content=content)
     
     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
         that the result failed and later receive the actual result.
         """
         
         outstanding = self.queues[eid]
         
         for msg_id in outstanding:
             self.pending.remove(msg_id)
             self.all_completed.add(msg_id)
             try:
                 raise error.EngineError("Engine %r died while running task %r"%(eid, msg_id))
             except:
                 content = error.wrap_exception()
             # build a fake header:
             header = {}
             header['engine'] = uuid
             header['date'] = datetime.now()
             rec = dict(result_content=content, result_header=header, result_buffers=[])
             rec['completed'] = header['date']
             rec['engine_uuid'] = uuid
             try:
                 self.db.update_record(msg_id, rec)
             except Exception:
                 self.log.error("DB Error handling stranded msg %r"%msg_id, exc_info=True)
                 
     
     def finish_registration(self, heart):
         """Second half of engine registration, called after our HeartMonitor
         has received a beat from the Engine's Heart."""
         try: 
             (eid,queue,reg,purge) = self.incoming_registrations.pop(heart)
         except KeyError:
             self.log.error("registration::tried to finish nonexistant registration", exc_info=True)
             return
         self.log.info("registration::finished registering engine %i:%r"%(eid,queue))
         if purge is not None:
             purge.stop()
         control = queue
         self.ids.add(eid)
         self.keytable[eid] = queue
         self.engines[eid] = EngineConnector(id=eid, queue=queue, registration=reg, 
                                     control=control, heartbeat=heart)
         self.by_ident[queue] = eid
         self.queues[eid] = list()
         self.tasks[eid] = list()
         self.completed[eid] = list()
         self.hearts[heart] = eid
         content = dict(id=eid, queue=self.engines[eid].queue)
         if self.notifier:
             self.session.send(self.notifier, "registration_notification", content=content)
         self.log.info("engine::Engine Connected: %i"%eid)
     
     def _purge_stalled_registration(self, heart):
         if heart in self.incoming_registrations:
             eid = self.incoming_registrations.pop(heart)[0]
             self.log.info("registration::purging stalled registration: %i"%eid)
         else:
             pass
             
     #-------------------------------------------------------------------------
     # Client Requests
     #-------------------------------------------------------------------------
     
     def shutdown_request(self, client_id, msg):
         """handle shutdown request."""
         self.session.send(self.query, 'shutdown_reply', content={'status': 'ok'}, ident=client_id)
         # also notify other clients of shutdown
         self.session.send(self.notifier, 'shutdown_notice', content={'status': 'ok'})
         dc = ioloop.DelayedCallback(lambda : self._shutdown(), 1000, self.loop)
         dc.start()
     
     def _shutdown(self):
         self.log.info("hub::hub shutting down.")
         time.sleep(0.1)
         sys.exit(0)
         
     
     def check_load(self, client_id, msg):
         content = msg['content']
         try:
             targets = content['targets']
             targets = self._validate_targets(targets)
         except:
             content = error.wrap_exception()
             self.session.send(self.query, "hub_error", 
                     content=content, ident=client_id)
             return
         
         content = dict(status='ok')
         # loads = {}
         for t in targets:
             content[bytes(t)] = len(self.queues[t])+len(self.tasks[t])
         self.session.send(self.query, "load_reply", content=content, ident=client_id)
             
     
     def queue_status(self, client_id, msg):
         """Return the Queue status of one or more targets.
         if verbose: return the msg_ids
         else: return len of each type.
         keys: queue (pending MUX jobs)
             tasks (pending Task jobs)
             completed (finished jobs from both queues)"""
         content = msg['content']
         targets = content['targets']
         try:
             targets = self._validate_targets(targets)
         except:
             content = error.wrap_exception()
             self.session.send(self.query, "hub_error", 
                     content=content, ident=client_id)
             return
         verbose = content.get('verbose', False)
         content = dict(status='ok')
         for t in targets:
             queue = self.queues[t]
             completed = self.completed[t]
             tasks = self.tasks[t]
             if not verbose:
                 queue = len(queue)
                 completed = len(completed)
                 tasks = len(tasks)
             content[bytes(t)] = {'queue': queue, 'completed': completed , 'tasks': tasks}
         content['unassigned'] = list(self.unassigned) if verbose else len(self.unassigned)
         
         self.session.send(self.query, "queue_reply", content=content, ident=client_id)
     
     def purge_results(self, client_id, msg):
         """Purge results from memory. This method is more valuable before we move
         to a DB based message storage mechanism."""
         content = msg['content']
         msg_ids = content.get('msg_ids', [])
         reply = dict(status='ok')
         if msg_ids == 'all':
             try:
                 self.db.drop_matching_records(dict(completed={'$ne':None}))
             except Exception:
                 reply = error.wrap_exception()
         else:
             pending = filter(lambda m: m in self.pending, msg_ids)
             if pending:
                 try:
                     raise IndexError("msg pending: %r"%pending[0])
                 except:
                     reply = error.wrap_exception()
             else:
                 try:
                     self.db.drop_matching_records(dict(msg_id={'$in':msg_ids}))
                 except Exception:
                     reply = error.wrap_exception()
 
             if reply['status'] == 'ok':
                 eids = content.get('engine_ids', [])
                 for eid in eids:
                     if eid not in self.engines:
                         try:
                             raise IndexError("No such engine: %i"%eid)
                         except:
                             reply = error.wrap_exception()
                         break
                     msg_ids = self.completed.pop(eid)
                     uid = self.engines[eid].queue
                     try:
                         self.db.drop_matching_records(dict(engine_uuid=uid, completed={'$ne':None}))
                     except Exception:
                         reply = error.wrap_exception()
                         break
         
         self.session.send(self.query, 'purge_reply', content=reply, ident=client_id)
     
     def resubmit_task(self, client_id, msg):
         """Resubmit one or more tasks."""
         def finish(reply):
             self.session.send(self.query, 'resubmit_reply', content=reply, ident=client_id)
 
         content = msg['content']
         msg_ids = content['msg_ids']
         reply = dict(status='ok')
         try:
             records = self.db.find_records({'msg_id' : {'$in' : msg_ids}}, keys=[
                 'header', 'content', 'buffers'])
         except Exception:
             self.log.error('db::db error finding tasks to resubmit', exc_info=True)
             return finish(error.wrap_exception())
 
         # validate msg_ids
         found_ids = [ rec['msg_id'] for rec in records ]
         invalid_ids = filter(lambda m: m in self.pending, found_ids)
         if len(records) > len(msg_ids):
             try:
                 raise RuntimeError("DB appears to be in an inconsistent state."
                     "More matching records were found than should exist")
             except Exception:
                 return finish(error.wrap_exception())
         elif len(records) < len(msg_ids):
             missing = [ m for m in msg_ids if m not in found_ids ]
             try:
                 raise KeyError("No such msg(s): %s"%missing)
             except KeyError:
                 return finish(error.wrap_exception())
         elif invalid_ids:
             msg_id = invalid_ids[0]
             try:
                 raise ValueError("Task %r appears to be inflight"%(msg_id))
             except Exception:
                 return finish(error.wrap_exception())
 
         # clear the existing records
         rec = empty_record()
         map(rec.pop, ['msg_id', 'header', 'content', 'buffers', 'submitted'])
         rec['resubmitted'] = datetime.now()
         rec['queue'] = 'task'
         rec['client_uuid'] = client_id[0]
         try:
             for msg_id in msg_ids:
                 self.all_completed.discard(msg_id)
                 self.db.update_record(msg_id, rec)
         except Exception:
             self.log.error('db::db error upating record', exc_info=True)
             reply = error.wrap_exception()
         else:
             # send the messages
             for rec in records:
                 header = rec['header']
                 msg = self.session.msg(header['msg_type'])
                 msg['content'] = rec['content']
                 msg['header'] = header
                 msg['msg_id'] = rec['msg_id']
                 self.session.send(self.resubmit, msg, buffers=rec['buffers'])
 
         finish(dict(status='ok'))
 
     
     def _extract_record(self, rec):
         """decompose a TaskRecord dict into subsection of reply for get_result"""
         io_dict = {}
         for key in 'pyin pyout pyerr stdout stderr'.split():
                 io_dict[key] = rec[key]
         content = { 'result_content': rec['result_content'],
                             'header': rec['header'],
                             'result_header' : rec['result_header'],
                             'io' : io_dict,
                           }
         if rec['result_buffers']:
             buffers = map(str, rec['result_buffers'])
         else:
             buffers = []
         
         return content, buffers
     
     def get_results(self, client_id, msg):
         """Get the result of 1 or more messages."""
         content = msg['content']
         msg_ids = sorted(set(content['msg_ids']))
         statusonly = content.get('status_only', False)
         pending = []
         completed = []
         content = dict(status='ok')
         content['pending'] = pending
         content['completed'] = completed
         buffers = []
         if not statusonly:
             try:
                 matches = self.db.find_records(dict(msg_id={'$in':msg_ids}))
                 # turn match list into dict, for faster lookup
                 records = {}
                 for rec in matches:
                     records[rec['msg_id']] = rec
             except Exception:
                 content = error.wrap_exception()
                 self.session.send(self.query, "result_reply", content=content, 
                                                     parent=msg, ident=client_id)
                 return
         else:
             records = {}
         for msg_id in msg_ids:
             if msg_id in self.pending:
                 pending.append(msg_id)
             elif msg_id in self.all_completed:
                 completed.append(msg_id)
                 if not statusonly:
                     c,bufs = self._extract_record(records[msg_id])
                     content[msg_id] = c
                     buffers.extend(bufs)
             elif msg_id in records:
                 if rec['completed']:
                     completed.append(msg_id)
                     c,bufs = self._extract_record(records[msg_id])
                     content[msg_id] = c
                     buffers.extend(bufs)
                 else:
                     pending.append(msg_id)
             else:
                 try:
                     raise KeyError('No such message: '+msg_id)
                 except:
                     content = error.wrap_exception()
                 break
         self.session.send(self.query, "result_reply", content=content, 
                                             parent=msg, ident=client_id,
                                             buffers=buffers)
 
     def get_history(self, client_id, msg):
         """Get a list of all msg_ids in our DB records"""
         try:
             msg_ids = self.db.get_history()
         except Exception as e:
             content = error.wrap_exception()
         else:
             content = dict(status='ok', history=msg_ids)
         
         self.session.send(self.query, "history_reply", content=content, 
                                             parent=msg, ident=client_id)
 
     def db_query(self, client_id, msg):
         """Perform a raw query on the task record database."""
         content = msg['content']
         query = content.get('query', {})
         keys = content.get('keys', None)
         query = util.extract_dates(query)
         buffers = []
         empty = list()
         
         try:
             records = self.db.find_records(query, keys)
         except Exception as e:
             content = error.wrap_exception()
         else:
             # extract buffers from reply content:
             if keys is not None:
                 buffer_lens = [] if 'buffers' in keys else None
                 result_buffer_lens = [] if 'result_buffers' in keys else None
             else:
                 buffer_lens = []
                 result_buffer_lens = []
             
             for rec in records:
                 # buffers may be None, so double check
                 if buffer_lens is not None:
                     b = rec.pop('buffers', empty) or empty
                     buffer_lens.append(len(b))
                     buffers.extend(b)
                 if result_buffer_lens is not None:
                     rb = rec.pop('result_buffers', empty) or empty
                     result_buffer_lens.append(len(rb))
                     buffers.extend(rb)
             content = dict(status='ok', records=records, buffer_lens=buffer_lens,
                                     result_buffer_lens=result_buffer_lens)
         
         self.session.send(self.query, "db_reply", content=content, 
                                             parent=msg, ident=client_id,
                                             buffers=buffers)
 

IPython/parallel/controller/mongodb.py

@@ -1,112 +1,112 b''
 """A TaskRecord backend using mongodb"""
 #-----------------------------------------------------------------------------
 #  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.
 #-----------------------------------------------------------------------------
 
 from pymongo import Connection
 from pymongo.binary import Binary
 
-from IPython.utils.traitlets import Dict, List, CUnicode, CStr, Instance
+from IPython.utils.traitlets import Dict, List, Unicode, Instance
 
 from .dictdb import BaseDB
 
 #-----------------------------------------------------------------------------
 # MongoDB class
 #-----------------------------------------------------------------------------
 
 class MongoDB(BaseDB):
     """MongoDB TaskRecord backend."""
     
     connection_args = List(config=True,
         help="""Positional arguments to be passed to pymongo.Connection.  Only
         necessary if the default mongodb configuration does not point to your
         mongod instance.""")
     connection_kwargs = Dict(config=True,
         help="""Keyword arguments to be passed to pymongo.Connection.  Only
         necessary if the default mongodb configuration does not point to your
         mongod instance."""
     )
-    database = CUnicode(config=True,
+    database = Unicode(config=True,
         help="""The MongoDB database name to use for storing tasks for this session. If unspecified,
         a new database will be created with the Hub's IDENT.  Specifying the database will result
         in tasks from previous sessions being available via Clients' db_query and
         get_result methods.""")
 
     _connection = Instance(Connection) # pymongo connection
     
     def __init__(self, **kwargs):
         super(MongoDB, self).__init__(**kwargs)
         if self._connection is None:
             self._connection = Connection(*self.connection_args, **self.connection_kwargs)
         if not self.database:
             self.database = self.session
         self._db = self._connection[self.database]
         self._records = self._db['task_records']
         self._records.ensure_index('msg_id', unique=True)
         self._records.ensure_index('submitted') # for sorting history
         # for rec in self._records.find
     
     def _binary_buffers(self, rec):
         for key in ('buffers', 'result_buffers'):
             if rec.get(key, None):
                 rec[key] = map(Binary, rec[key])
         return rec
     
     def add_record(self, msg_id, rec):
         """Add a new Task Record, by msg_id."""
         # print rec
         rec = self._binary_buffers(rec)
         self._records.insert(rec)
     
     def get_record(self, msg_id):
         """Get a specific Task Record, by msg_id."""
         r = self._records.find_one({'msg_id': msg_id})
         if not r:
             # r will be '' if nothing is found
             raise KeyError(msg_id)
         return r
     
     def update_record(self, msg_id, rec):
         """Update the data in an existing record."""
         rec = self._binary_buffers(rec)
 
         self._records.update({'msg_id':msg_id}, {'$set': rec})
     
     def drop_matching_records(self, check):
         """Remove a record from the DB."""
         self._records.remove(check)
         
     def drop_record(self, msg_id):
         """Remove a record from the DB."""
         self._records.remove({'msg_id':msg_id})
     
     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 and 'msg_id' not in keys:
             keys.append('msg_id')
         matches = list(self._records.find(check,keys))
         for rec in matches:
             rec.pop('_id')
         return matches
 
     def get_history(self):
         """get all msg_ids, ordered by time submitted."""
         cursor = self._records.find({},{'msg_id':1}).sort('submitted')
         return [ rec['msg_id'] for rec in cursor ]
 
 

IPython/parallel/controller/sqlitedb.py

@@ -1,333 +1,333 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 CUnicode, CStr, Instance, List
+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 = CUnicode('tasks.db', config=True,
+    filename = Unicode('tasks.db', config=True,
         help="""The filename of the sqlite task database. [default: 'tasks.db']""")
-    location = CUnicode('', config=True,
+    location = Unicode('', config=True,
         help="""The directory containing the sqlite task database.  The default
         is to use the cluster_dir location.""")
-    table = CUnicode("", config=True,
+    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
             else:
                 self.location = '.'
         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

IPython/parallel/engine/engine.py

@@ -1,166 +1,166 b''
 #!/usr/bin/env python
 """A simple engine that talks to a controller over 0MQ.
 it handles registration, etc. and launches a kernel
 connected to the Controller's Schedulers.
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-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.
 #-----------------------------------------------------------------------------
 
 from __future__ import print_function
 
 import sys
 import time
 
 import zmq
 from zmq.eventloop import ioloop, zmqstream
 
 # internal
-from IPython.utils.traitlets import Instance, Str, Dict, Int, Type, CFloat
+from IPython.utils.traitlets import Instance, Dict, Int, Type, CFloat, Unicode
 # from IPython.utils.localinterfaces import LOCALHOST 
 
 from IPython.parallel.controller.heartmonitor import Heart
 from IPython.parallel.factory import RegistrationFactory
 from IPython.parallel.streamsession import Message
 from IPython.parallel.util import disambiguate_url
 
 from .streamkernel import Kernel
 
 class EngineFactory(RegistrationFactory):
     """IPython engine"""
     
     # configurables:
     out_stream_factory=Type('IPython.zmq.iostream.OutStream', config=True,
         help="""The OutStream for handling stdout/err.
         Typically 'IPython.zmq.iostream.OutStream'""")
     display_hook_factory=Type('IPython.zmq.displayhook.DisplayHook', config=True,
         help="""The class for handling displayhook.
         Typically 'IPython.zmq.displayhook.DisplayHook'""")
-    location=Str(config=True,
+    location=Unicode(config=True,
         help="""The location (an IP address) of the controller.  This is
         used for disambiguating URLs, to determine whether
         loopback should be used to connect or the public address.""")
     timeout=CFloat(2,config=True,
         help="""The time (in seconds) to wait for the Controller to respond
         to registration requests before giving up.""")
     
     # not configurable:
     user_ns=Dict()
     id=Int(allow_none=True)
     registrar=Instance('zmq.eventloop.zmqstream.ZMQStream')
     kernel=Instance(Kernel)
     
     
     def __init__(self, **kwargs):
         super(EngineFactory, self).__init__(**kwargs)
         self.ident = self.session.session
         ctx = self.context
         
         reg = ctx.socket(zmq.XREQ)
         reg.setsockopt(zmq.IDENTITY, self.ident)
         reg.connect(self.url)
         self.registrar = zmqstream.ZMQStream(reg, self.loop)
         
     def register(self):
         """send the registration_request"""
         
         self.log.info("registering")
         content = dict(queue=self.ident, heartbeat=self.ident, control=self.ident)
         self.registrar.on_recv(self.complete_registration)
         # print (self.session.key)
         self.session.send(self.registrar, "registration_request",content=content)
     
     def complete_registration(self, msg):
         # print msg
         self._abort_dc.stop()
         ctx = self.context
         loop = self.loop
         identity = self.ident
         
         idents,msg = self.session.feed_identities(msg)
         msg = Message(self.session.unpack_message(msg))
         
         if msg.content.status == 'ok':
             self.id = int(msg.content.id)
             
             # create Shell Streams (MUX, Task, etc.):
             queue_addr = msg.content.mux
             shell_addrs = [ str(queue_addr) ]
             task_addr = msg.content.task
             if task_addr:
                 shell_addrs.append(str(task_addr))
             
             # Uncomment this to go back to two-socket model
             # shell_streams = []
             # for addr in shell_addrs:
             #     stream = zmqstream.ZMQStream(ctx.socket(zmq.XREP), loop)
             #     stream.setsockopt(zmq.IDENTITY, identity)
             #     stream.connect(disambiguate_url(addr, self.location))
             #     shell_streams.append(stream)
             
             # Now use only one shell stream for mux and tasks
             stream = zmqstream.ZMQStream(ctx.socket(zmq.XREP), loop)
             stream.setsockopt(zmq.IDENTITY, identity)
             shell_streams = [stream]
             for addr in shell_addrs:
                 stream.connect(disambiguate_url(addr, self.location))
             # end single stream-socket
             
             # control stream:
             control_addr = str(msg.content.control)
             control_stream = zmqstream.ZMQStream(ctx.socket(zmq.XREP), loop)
             control_stream.setsockopt(zmq.IDENTITY, identity)
             control_stream.connect(disambiguate_url(control_addr, self.location))
             
             # create iopub stream:
             iopub_addr = msg.content.iopub
             iopub_stream = zmqstream.ZMQStream(ctx.socket(zmq.PUB), loop)
             iopub_stream.setsockopt(zmq.IDENTITY, identity)
             iopub_stream.connect(disambiguate_url(iopub_addr, self.location))
             
             # launch heartbeat
             hb_addrs = msg.content.heartbeat
             # print (hb_addrs)
             
             # # Redirect input streams and set a display hook.
             if self.out_stream_factory:
                 sys.stdout = self.out_stream_factory(self.session, iopub_stream, u'stdout')
                 sys.stdout.topic = 'engine.%i.stdout'%self.id
                 sys.stderr = self.out_stream_factory(self.session, iopub_stream, u'stderr')
                 sys.stderr.topic = 'engine.%i.stderr'%self.id
             if self.display_hook_factory:
                 sys.displayhook = self.display_hook_factory(self.session, iopub_stream)
                 sys.displayhook.topic = 'engine.%i.pyout'%self.id
             
             self.kernel = Kernel(config=self.config, int_id=self.id, ident=self.ident, session=self.session, 
                     control_stream=control_stream, shell_streams=shell_streams, iopub_stream=iopub_stream, 
                     loop=loop, user_ns = self.user_ns, log=self.log)
             self.kernel.start()
             hb_addrs = [ disambiguate_url(addr, self.location) for addr in hb_addrs ]
             heart = Heart(*map(str, hb_addrs), heart_id=identity)
             # ioloop.DelayedCallback(heart.start, 1000, self.loop).start()
             heart.start()
             
             
         else:
             self.log.fatal("Registration Failed: %s"%msg)
             raise Exception("Registration Failed: %s"%msg)
         
         self.log.info("Completed registration with id %i"%self.id)
     
     
     def abort(self):
         self.log.fatal("Registration timed out after %.1f seconds"%self.timeout)
         self.session.send(self.registrar, "unregistration_request", content=dict(id=self.id))
         time.sleep(1)
         sys.exit(255)
     
     def start(self):
         dc = ioloop.DelayedCallback(self.register, 0, self.loop)
         dc.start()
         self._abort_dc = ioloop.DelayedCallback(self.abort, self.timeout*1000, self.loop)
         self._abort_dc.start()
 

IPython/parallel/engine/streamkernel.py

@@ -1,433 +1,433 b''
 #!/usr/bin/env python
 """
 Kernel adapted from kernel.py to use ZMQ Streams
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-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
 #-----------------------------------------------------------------------------
 
 # Standard library imports.
 from __future__ import print_function
 
 import sys
 import time
 
 from code import CommandCompiler
 from datetime import datetime
 from pprint import pprint
 
 # System library imports.
 import zmq
 from zmq.eventloop import ioloop, zmqstream
 
 # Local imports.
-from IPython.utils.traitlets import Instance, List, Int, Dict, Set, Str, CStr
+from IPython.utils.traitlets import Instance, List, Int, Dict, Set, Unicode
 from IPython.zmq.completer import KernelCompleter
 
 from IPython.parallel.error import wrap_exception
 from IPython.parallel.factory import SessionFactory
 from IPython.parallel.util import serialize_object, unpack_apply_message, ISO8601
 
 def printer(*args):
     pprint(args, stream=sys.__stdout__)
 
 
 class _Passer(zmqstream.ZMQStream):
     """Empty class that implements `send()` that does nothing.
     
     Subclass ZMQStream for StreamSession typechecking
     
     """
     def __init__(self, *args, **kwargs):
         pass
     
     def send(self, *args, **kwargs):
         pass
     send_multipart = send
     
 
 #-----------------------------------------------------------------------------
 # Main kernel class
 #-----------------------------------------------------------------------------
 
 class Kernel(SessionFactory):
 
     #---------------------------------------------------------------------------
     # Kernel interface
     #---------------------------------------------------------------------------
     
     # kwargs:
-    exec_lines = List(CStr, config=True,
+    exec_lines = List(Unicode, config=True,
         help="List of lines to execute")
 
     int_id = Int(-1)
     user_ns = Dict(config=True,  help="""Set the user's namespace of the Kernel""")
     
     control_stream = Instance(zmqstream.ZMQStream)
     task_stream = Instance(zmqstream.ZMQStream)
     iopub_stream = Instance(zmqstream.ZMQStream)
     client = Instance('IPython.parallel.Client')
     
     # internals
     shell_streams = List()
     compiler = Instance(CommandCompiler, (), {})
     completer = Instance(KernelCompleter)
     
     aborted = Set()
     shell_handlers = Dict()
     control_handlers = Dict()
     
     def _set_prefix(self):
         self.prefix = "engine.%s"%self.int_id
     
     def _connect_completer(self):
         self.completer = KernelCompleter(self.user_ns)
     
     def __init__(self, **kwargs):
         super(Kernel, self).__init__(**kwargs)
         self._set_prefix()
         self._connect_completer()
         
         self.on_trait_change(self._set_prefix, 'id')
         self.on_trait_change(self._connect_completer, 'user_ns')
         
         # Build dict of handlers for message types
         for msg_type in ['execute_request', 'complete_request', 'apply_request', 
                 'clear_request']:
             self.shell_handlers[msg_type] = getattr(self, msg_type)
         
         for msg_type in ['shutdown_request', 'abort_request']+self.shell_handlers.keys():
             self.control_handlers[msg_type] = getattr(self, msg_type)
         
         self._initial_exec_lines()
     
     def _wrap_exception(self, method=None):
         e_info = dict(engine_uuid=self.ident, engine_id=self.int_id, method=method)
         content=wrap_exception(e_info)
         return content
     
     def _initial_exec_lines(self):
         s = _Passer()
         content = dict(silent=True, user_variable=[],user_expressions=[])
         for line in self.exec_lines:
             self.log.debug("executing initialization: %s"%line)
             content.update({'code':line})
             msg = self.session.msg('execute_request', content)
             self.execute_request(s, [], msg)
         
         
     #-------------------- control handlers -----------------------------
     def abort_queues(self):
         for stream in self.shell_streams:
             if stream:
                 self.abort_queue(stream)
     
     def abort_queue(self, stream):
         while True:
             try:
                 msg = self.session.recv(stream, zmq.NOBLOCK,content=True)
             except zmq.ZMQError as e:
                 if e.errno == zmq.EAGAIN:
                     break
                 else:
                     return
             else:
                 if msg is None:
                     return
                 else:
                     idents,msg = msg
                 
                 # assert self.reply_socketly_socket.rcvmore(), "Unexpected missing message part."
                 # msg = self.reply_socket.recv_json()
             self.log.info("Aborting:")
             self.log.info(str(msg))
             msg_type = msg['msg_type']
             reply_type = msg_type.split('_')[0] + '_reply'
             # reply_msg = self.session.msg(reply_type, {'status' : 'aborted'}, msg)
             # self.reply_socket.send(ident,zmq.SNDMORE)
             # self.reply_socket.send_json(reply_msg)
             reply_msg = self.session.send(stream, reply_type, 
                         content={'status' : 'aborted'}, parent=msg, ident=idents)[0]
             self.log.debug(str(reply_msg))
             # We need to wait a bit for requests to come in. This can probably
             # be set shorter for true asynchronous clients.
             time.sleep(0.05)
     
     def abort_request(self, stream, ident, parent):
         """abort a specifig msg by id"""
         msg_ids = parent['content'].get('msg_ids', None)
         if isinstance(msg_ids, basestring):
             msg_ids = [msg_ids]
         if not msg_ids:
             self.abort_queues()
         for mid in msg_ids:
             self.aborted.add(str(mid))
         
         content = dict(status='ok')
         reply_msg = self.session.send(stream, 'abort_reply', content=content, 
                 parent=parent, ident=ident)
         self.log.debug(str(reply_msg))
     
     def shutdown_request(self, stream, ident, parent):
         """kill ourself.  This should really be handled in an external process"""
         try:
             self.abort_queues()
         except:
             content = self._wrap_exception('shutdown')
         else:
             content = dict(parent['content'])
             content['status'] = 'ok'
         msg = self.session.send(stream, 'shutdown_reply',
                                 content=content, parent=parent, ident=ident)
         self.log.debug(str(msg))
         dc = ioloop.DelayedCallback(lambda : sys.exit(0), 1000, self.loop)
         dc.start()
     
     def dispatch_control(self, msg):
         idents,msg = self.session.feed_identities(msg, copy=False)
         try:
             msg = self.session.unpack_message(msg, content=True, copy=False)
         except:
             self.log.error("Invalid Message", exc_info=True)
             return
         
         header = msg['header']
         msg_id = header['msg_id']
         
         handler = self.control_handlers.get(msg['msg_type'], None)
         if handler is None:
             self.log.error("UNKNOWN CONTROL MESSAGE TYPE: %r"%msg['msg_type'])
         else:
             handler(self.control_stream, idents, msg)
     
 
     #-------------------- queue helpers ------------------------------
     
     def check_dependencies(self, dependencies):
         if not dependencies:
             return True
         if len(dependencies) == 2 and dependencies[0] in 'any all'.split():
             anyorall = dependencies[0]
             dependencies = dependencies[1]
         else:
             anyorall = 'all'
         results = self.client.get_results(dependencies,status_only=True)
         if results['status'] != 'ok':
             return False
         
         if anyorall == 'any':
             if not results['completed']:
                 return False
         else:
             if results['pending']:
                 return False
         
         return True
     
     def check_aborted(self, msg_id):
         return msg_id in self.aborted
     
     #-------------------- queue handlers -----------------------------
     
     def clear_request(self, stream, idents, parent):
         """Clear our namespace."""
         self.user_ns = {}
         msg = self.session.send(stream, 'clear_reply', ident=idents, parent=parent, 
                 content = dict(status='ok'))
         self._initial_exec_lines()
     
     def execute_request(self, stream, ident, parent):
         self.log.debug('execute request %s'%parent)
         try:
             code = parent[u'content'][u'code']
         except:
             self.log.error("Got bad msg: %s"%parent, exc_info=True)
             return
         self.session.send(self.iopub_stream, u'pyin', {u'code':code},parent=parent,
                             ident='%s.pyin'%self.prefix)
         started = datetime.now().strftime(ISO8601)
         try:
             comp_code = self.compiler(code, '<zmq-kernel>')
             # allow for not overriding displayhook
             if hasattr(sys.displayhook, 'set_parent'):
                 sys.displayhook.set_parent(parent)
                 sys.stdout.set_parent(parent)
                 sys.stderr.set_parent(parent)
             exec comp_code in self.user_ns, self.user_ns
         except:
             exc_content = self._wrap_exception('execute')
             # exc_msg = self.session.msg(u'pyerr', exc_content, parent)
             self.session.send(self.iopub_stream, u'pyerr', exc_content, parent=parent,
                             ident='%s.pyerr'%self.prefix)
             reply_content = exc_content
         else:
             reply_content = {'status' : 'ok'}
         
         reply_msg = self.session.send(stream, u'execute_reply', reply_content, parent=parent, 
                     ident=ident, subheader = dict(started=started))
         self.log.debug(str(reply_msg))
         if reply_msg['content']['status'] == u'error':
             self.abort_queues()
 
     def complete_request(self, stream, ident, parent):
         matches = {'matches' : self.complete(parent),
                    'status' : 'ok'}
         completion_msg = self.session.send(stream, 'complete_reply',
                                            matches, parent, ident)
         # print >> sys.__stdout__, completion_msg
 
     def complete(self, msg):
         return self.completer.complete(msg.content.line, msg.content.text)
     
     def apply_request(self, stream, ident, parent):
         # flush previous reply, so this request won't block it
         stream.flush(zmq.POLLOUT)
         
         try:
             content = parent[u'content']
             bufs = parent[u'buffers']
             msg_id = parent['header']['msg_id']
             # bound = parent['header'].get('bound', False)
         except:
             self.log.error("Got bad msg: %s"%parent, exc_info=True)
             return
         # pyin_msg = self.session.msg(u'pyin',{u'code':code}, parent=parent)
         # self.iopub_stream.send(pyin_msg)
         # self.session.send(self.iopub_stream, u'pyin', {u'code':code},parent=parent)
         sub = {'dependencies_met' : True, 'engine' : self.ident,
                 'started': datetime.now().strftime(ISO8601)}
         try:
             # allow for not overriding displayhook
             if hasattr(sys.displayhook, 'set_parent'):
                 sys.displayhook.set_parent(parent)
                 sys.stdout.set_parent(parent)
                 sys.stderr.set_parent(parent)
             # exec "f(*args,**kwargs)" in self.user_ns, self.user_ns
             working = self.user_ns
             # suffix = 
             prefix = "_"+str(msg_id).replace("-","")+"_"
             
             f,args,kwargs = unpack_apply_message(bufs, working, copy=False)
             # if bound:
             #     bound_ns = Namespace(working)
             #     args = [bound_ns]+list(args)
 
             fname = getattr(f, '__name__', 'f')
             
             fname = prefix+"f"
             argname = prefix+"args"
             kwargname = prefix+"kwargs"
             resultname = prefix+"result"
             
             ns = { fname : f, argname : args, kwargname : kwargs , resultname : None }
             # print ns
             working.update(ns)
             code = "%s=%s(*%s,**%s)"%(resultname, fname, argname, kwargname)
             try:
                 exec code in working,working
                 result = working.get(resultname)
             finally:
                 for key in ns.iterkeys():
                     working.pop(key)
             # if bound:
             #     working.update(bound_ns)
             
             packed_result,buf = serialize_object(result)
             result_buf = [packed_result]+buf
         except:
             exc_content = self._wrap_exception('apply')
             # exc_msg = self.session.msg(u'pyerr', exc_content, parent)
             self.session.send(self.iopub_stream, u'pyerr', exc_content, parent=parent,
                                 ident='%s.pyerr'%self.prefix)
             reply_content = exc_content
             result_buf = []
             
             if exc_content['ename'] == 'UnmetDependency':
                 sub['dependencies_met'] = False
         else:
             reply_content = {'status' : 'ok'}
         
         # put 'ok'/'error' status in header, for scheduler introspection:
         sub['status'] = reply_content['status']
         
         reply_msg = self.session.send(stream, u'apply_reply', reply_content, 
                     parent=parent, ident=ident,buffers=result_buf, subheader=sub)
         
         # flush i/o
         # should this be before reply_msg is sent, like in the single-kernel code, 
         # or should nothing get in the way of real results?
         sys.stdout.flush()
         sys.stderr.flush()
     
     def dispatch_queue(self, stream, msg):
         self.control_stream.flush()
         idents,msg = self.session.feed_identities(msg, copy=False)
         try:
             msg = self.session.unpack_message(msg, content=True, copy=False)
         except:
             self.log.error("Invalid Message", exc_info=True)
             return
             
         
         header = msg['header']
         msg_id = header['msg_id']
         if self.check_aborted(msg_id):
             self.aborted.remove(msg_id)
             # is it safe to assume a msg_id will not be resubmitted?
             reply_type = msg['msg_type'].split('_')[0] + '_reply'
             status = {'status' : 'aborted'}
             reply_msg = self.session.send(stream, reply_type, subheader=status,
                         content=status, parent=msg, ident=idents)
             return
         handler = self.shell_handlers.get(msg['msg_type'], None)
         if handler is None:
             self.log.error("UNKNOWN MESSAGE TYPE: %r"%msg['msg_type'])
         else:
             handler(stream, idents, msg)
     
     def start(self):
         #### stream mode:
         if self.control_stream:
             self.control_stream.on_recv(self.dispatch_control, copy=False)
             self.control_stream.on_err(printer)
         
         def make_dispatcher(stream):
             def dispatcher(msg):
                 return self.dispatch_queue(stream, msg)
             return dispatcher
         
         for s in self.shell_streams:
             s.on_recv(make_dispatcher(s), copy=False)
             s.on_err(printer)
         
         if self.iopub_stream:
             self.iopub_stream.on_err(printer)
         
         #### while True mode:
         # while True:
         #     idle = True
         #     try:
         #         msg = self.shell_stream.socket.recv_multipart(
         #                     zmq.NOBLOCK, copy=False)
         #     except zmq.ZMQError, e:
         #         if e.errno != zmq.EAGAIN:
         #             raise e
         #     else:
         #         idle=False
         #         self.dispatch_queue(self.shell_stream, msg)
         #             
         #     if not self.task_stream.empty():
         #         idle=False
         #         msg = self.task_stream.recv_multipart()
         #         self.dispatch_queue(self.task_stream, msg)
         #     if idle:
         #         # don't busywait
         #         time.sleep(1e-3)
 

IPython/parallel/factory.py

@@ -1,95 +1,95 b''
 """Base config factories."""
 
 #-----------------------------------------------------------------------------
 #  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 logging
 import os
 
 from zmq.eventloop.ioloop import IOLoop
 
 from IPython.config.configurable import Configurable
-from IPython.utils.traitlets import Str,Int,Instance, CUnicode, CStr
+from IPython.utils.traitlets import Int, Instance, Unicode
 
 import IPython.parallel.streamsession as ss
 from IPython.parallel.util import select_random_ports
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 class LoggingFactory(Configurable):
     """A most basic class, that has a `log` (type:`Logger`) attribute, set via a `logname` Trait."""
     log = Instance('logging.Logger', ('ZMQ', logging.WARN))
-    logname = CUnicode('ZMQ')
+    logname = Unicode('ZMQ')
     def _logname_changed(self, name, old, new):
         self.log = logging.getLogger(new)
     
 
 class SessionFactory(LoggingFactory):
     """The Base factory from which every factory in IPython.parallel inherits"""
     
     # not configurable:
     context = Instance('zmq.Context', (), {})
     session = Instance('IPython.parallel.streamsession.StreamSession')
     loop = Instance('zmq.eventloop.ioloop.IOLoop', allow_none=False)
     def _loop_default(self):
         return IOLoop.instance()
     
     
     def __init__(self, **kwargs):
         super(SessionFactory, self).__init__(**kwargs)
         
         # construct the session
         self.session = ss.StreamSession(**kwargs)
     
 
 class RegistrationFactory(SessionFactory):
     """The Base Configurable for objects that involve registration."""
     
-    url = Str('', config=True,
+    url = Unicode('', config=True,
         help="""The 0MQ url used for registration. This sets transport, ip, and port
         in one variable. For example: url='tcp://127.0.0.1:12345' or
         url='epgm://*:90210'""") # url takes precedence over ip,regport,transport
-    transport = Str('tcp', config=True,
+    transport = Unicode('tcp', config=True,
         help="""The 0MQ transport for communications.  This will likely be
         the default of 'tcp', but other values include 'ipc', 'epgm', 'inproc'.""")
-    ip = Str('127.0.0.1', config=True,
+    ip = Unicode('127.0.0.1', config=True,
         help="""The IP address for registration.  This is generally either
         '127.0.0.1' for loopback only or '*' for all interfaces.
         [default: '127.0.0.1']""")
     regport = Int(config=True,
         help="""The port on which the Hub listens for registration.""")
     def _regport_default(self):
         return select_random_ports(1)[0]
     
     def __init__(self, **kwargs):
         super(RegistrationFactory, self).__init__(**kwargs)
         self._propagate_url()
         self._rebuild_url()
         self.on_trait_change(self._propagate_url, 'url')
         self.on_trait_change(self._rebuild_url, 'ip')
         self.on_trait_change(self._rebuild_url, 'transport')
         self.on_trait_change(self._rebuild_url, 'regport')
     
     def _rebuild_url(self):
         self.url = "%s://%s:%i"%(self.transport, self.ip, self.regport)
         
     def _propagate_url(self):
         """Ensure self.url contains full transport://interface:port"""
         if self.url:
             iface = self.url.split('://',1)
             if len(iface) == 2:
                 self.transport,iface = iface
             iface = iface.split(':')
             self.ip = iface[0]
             if iface[1]:
                 self.regport = int(iface[1])

IPython/parallel/streamsession.py

@@ -1,446 +1,446 b''
 #!/usr/bin/env python
 """edited session.py to work with streams, and move msg_type to the header
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-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 os
 import pprint
 import uuid
 from datetime import datetime
 
 try:
     import cPickle
     pickle = cPickle
 except:
     cPickle = None
     import pickle
 
 import zmq
 from zmq.utils import jsonapi
 from zmq.eventloop.zmqstream import ZMQStream
 
 from IPython.config.configurable import Configurable
 from IPython.utils.importstring import import_item
-from IPython.utils.traitlets import Str, CStr, CUnicode, Bool, Any
+from IPython.utils.traitlets import CStr, Unicode, Bool, Any
 
 from .util import ISO8601
 
 
 def squash_unicode(obj):
     """coerce unicode back to bytestrings."""
     if isinstance(obj,dict):
         for key in obj.keys():
             obj[key] = squash_unicode(obj[key])
             if isinstance(key, unicode):
                 obj[squash_unicode(key)] = obj.pop(key)
     elif isinstance(obj, list):
         for i,v in enumerate(obj):
             obj[i] = squash_unicode(v)
     elif isinstance(obj, unicode):
         obj = obj.encode('utf8')
     return obj
 
 def _date_default(obj):
     if isinstance(obj, datetime):
         return obj.strftime(ISO8601)
     else:
         raise TypeError("%r is not JSON serializable"%obj)
 
 _default_key = 'on_unknown' if jsonapi.jsonmod.__name__ == 'jsonlib' else 'default'
 json_packer = lambda obj: jsonapi.dumps(obj, **{_default_key:_date_default})
 json_unpacker = lambda s: squash_unicode(jsonapi.loads(s))
 
 pickle_packer = lambda o: pickle.dumps(o,-1)
 pickle_unpacker = pickle.loads
 
 default_packer = json_packer
 default_unpacker = json_unpacker
 
 
 DELIM="<IDS|MSG>"
 
 class Message(object):
     """A simple message object that maps dict keys to attributes.
 
     A Message can be created from a dict and a dict from a Message instance
     simply by calling dict(msg_obj)."""
     
     def __init__(self, msg_dict):
         dct = self.__dict__
         for k, v in dict(msg_dict).iteritems():
             if isinstance(v, dict):
                 v = Message(v)
             dct[k] = v
 
     # Having this iterator lets dict(msg_obj) work out of the box.
     def __iter__(self):
         return iter(self.__dict__.iteritems())
     
     def __repr__(self):
         return repr(self.__dict__)
 
     def __str__(self):
         return pprint.pformat(self.__dict__)
 
     def __contains__(self, k):
         return k in self.__dict__
 
     def __getitem__(self, k):
         return self.__dict__[k]
 
 
 def msg_header(msg_id, msg_type, username, session):
     date=datetime.now().strftime(ISO8601)
     return locals()
 
 def extract_header(msg_or_header):
     """Given a message or header, return the header."""
     if not msg_or_header:
         return {}
     try:
         # See if msg_or_header is the entire message.
         h = msg_or_header['header']
     except KeyError:
         try:
             # See if msg_or_header is just the header
             h = msg_or_header['msg_id']
         except KeyError:
             raise
         else:
             h = msg_or_header
     if not isinstance(h, dict):
         h = dict(h)
     return h
 
 class StreamSession(Configurable):
     """tweaked version of IPython.zmq.session.Session, for development in Parallel"""
     debug=Bool(False, config=True, help="""Debug output in the StreamSession""")
-    packer = Str('json',config=True,
+    packer = Unicode('json',config=True,
             help="""The name of the packer for serializing messages.
             Should be one of 'json', 'pickle', or an import name
             for a custom serializer.""")
     def _packer_changed(self, name, old, new):
         if new.lower() == 'json':
             self.pack = json_packer
             self.unpack = json_unpacker
         elif new.lower() == 'pickle':
             self.pack = pickle_packer
             self.unpack = pickle_unpacker
         else:
             self.pack = import_item(new)
 
-    unpacker = Str('json',config=True,
+    unpacker = Unicode('json',config=True,
         help="""The name of the unpacker for unserializing messages.
         Only used with custom functions for `packer`.""")
     def _unpacker_changed(self, name, old, new):
         if new.lower() == 'json':
             self.pack = json_packer
             self.unpack = json_unpacker
         elif new.lower() == 'pickle':
             self.pack = pickle_packer
             self.unpack = pickle_unpacker
         else:
             self.unpack = import_item(new)
         
     session = CStr('',config=True,
         help="""The UUID identifying this session.""")
     def _session_default(self):
-        return str(uuid.uuid4())
-    username = CUnicode(os.environ.get('USER','username'),config=True,
+        return bytes(uuid.uuid4())
+    username = Unicode(os.environ.get('USER','username'),config=True,
         help="""Username for the Session. Default is your system username.""")
     key = CStr('', config=True,
         help="""execution key, for extra authentication.""")
 
-    keyfile = CUnicode('', config=True,
+    keyfile = Unicode('', config=True,
         help="""path to file containing execution key.""")
     def _keyfile_changed(self, name, old, new):
         with open(new, 'rb') as f:
             self.key = f.read().strip()
 
     pack = Any(default_packer) # the actual packer function
     def _pack_changed(self, name, old, new):
         if not callable(new):
             raise TypeError("packer must be callable, not %s"%type(new))
         
     unpack = Any(default_unpacker) # the actual packer function
     def _unpack_changed(self, name, old, new):
         if not callable(new):
             raise TypeError("packer must be callable, not %s"%type(new))
 
     def __init__(self, **kwargs):
         super(StreamSession, self).__init__(**kwargs)
         self.none = self.pack({})
 
     @property
     def msg_id(self):
         """always return new uuid"""
         return str(uuid.uuid4())
 
     def msg_header(self, msg_type):
         return msg_header(self.msg_id, msg_type, self.username, self.session)
 
     def msg(self, msg_type, content=None, parent=None, subheader=None):
         msg = {}
         msg['header'] = self.msg_header(msg_type)
         msg['msg_id'] = msg['header']['msg_id']
         msg['parent_header'] = {} if parent is None else extract_header(parent)
         msg['msg_type'] = msg_type
         msg['content'] = {} if content is None else content
         sub = {} if subheader is None else subheader
         msg['header'].update(sub)
         return msg
 
     def check_key(self, msg_or_header):
         """Check that a message's header has the right key"""
         if not self.key:
             return True
         header = extract_header(msg_or_header)
         return header.get('key', '') == self.key
             
 
     def serialize(self, msg, ident=None):
         content = msg.get('content', {})
         if content is None:
             content = self.none
         elif isinstance(content, dict):
             content = self.pack(content)
         elif isinstance(content, bytes):
             # content is already packed, as in a relayed message
             pass
         elif isinstance(content, unicode):
             # should be bytes, but JSON often spits out unicode
             content = content.encode('utf8')
         else:
             raise TypeError("Content incorrect type: %s"%type(content))
 
         to_send = []
 
         if isinstance(ident, list):
             # accept list of idents
             to_send.extend(ident)
         elif ident is not None:
             to_send.append(ident)
         to_send.append(DELIM)
         if self.key:
             to_send.append(self.key)
         to_send.append(self.pack(msg['header']))
         to_send.append(self.pack(msg['parent_header']))
         to_send.append(content)
 
         return to_send
         
     def send(self, stream, msg_or_type, content=None, buffers=None, parent=None, subheader=None, ident=None, track=False):
         """Build and send a message via stream or socket.
         
         Parameters
         ----------
         
         stream : zmq.Socket or ZMQStream
             the socket-like object used to send the data
         msg_or_type : str or Message/dict
             Normally, msg_or_type will be a msg_type unless a message is being sent more
             than once.
         
         content : dict or None
             the content of the message (ignored if msg_or_type is a message)
         buffers : list or None
             the already-serialized buffers to be appended to the message
         parent : Message or dict or None
             the parent or parent header describing the parent of this message
         subheader : dict or None
             extra header keys for this message's header
         ident : bytes or list of bytes
             the zmq.IDENTITY routing path
         track : bool
             whether to track.  Only for use with Sockets, because ZMQStream objects cannot track messages.
         
         Returns
         -------
         msg : message dict
             the constructed message
         (msg,tracker) : (message dict, MessageTracker)
             if track=True, then a 2-tuple will be returned, the first element being the constructed
             message, and the second being the MessageTracker
             
         """
 
         if not isinstance(stream, (zmq.Socket, ZMQStream)):
             raise TypeError("stream must be Socket or ZMQStream, not %r"%type(stream))
         elif track and isinstance(stream, ZMQStream):
             raise TypeError("ZMQStream cannot track messages")
         
         if isinstance(msg_or_type, (Message, dict)):
             # we got a Message, not a msg_type
             # don't build a new Message
             msg = msg_or_type
         else:
             msg = self.msg(msg_or_type, content, parent, subheader)
         
         buffers = [] if buffers is None else buffers
         to_send = self.serialize(msg, ident)
         flag = 0
         if buffers:
             flag = zmq.SNDMORE
             _track = False
         else:
             _track=track
         if track:
             tracker = stream.send_multipart(to_send, flag, copy=False, track=_track)
         else:
             tracker = stream.send_multipart(to_send, flag, copy=False)
         for b in buffers[:-1]:
             stream.send(b, flag, copy=False)
         if buffers:
             if track:
                 tracker = stream.send(buffers[-1], copy=False, track=track)
             else:
                 tracker = stream.send(buffers[-1], copy=False)
                 
         # omsg = Message(msg)
         if self.debug:
             pprint.pprint(msg)
             pprint.pprint(to_send)
             pprint.pprint(buffers)
         
         msg['tracker'] = tracker
         
         return msg
     
     def send_raw(self, stream, msg, flags=0, copy=True, ident=None):
         """Send a raw message via ident path.
         
         Parameters
         ----------
         msg : list of sendable buffers"""
         to_send = []
         if isinstance(ident, bytes):
             ident = [ident]
         if ident is not None:
             to_send.extend(ident)
         to_send.append(DELIM)
         if self.key:
             to_send.append(self.key)
         to_send.extend(msg)
         stream.send_multipart(msg, flags, copy=copy)
     
     def recv(self, socket, mode=zmq.NOBLOCK, content=True, copy=True):
         """receives and unpacks a message
         returns [idents], msg"""
         if isinstance(socket, ZMQStream):
             socket = socket.socket
         try:
             msg = socket.recv_multipart(mode, copy=copy)
         except zmq.ZMQError as e:
             if e.errno == zmq.EAGAIN:
                 # We can convert EAGAIN to None as we know in this case
                 # recv_multipart won't return None.
                 return None
             else:
                 raise
         # return an actual Message object
         # determine the number of idents by trying to unpack them.
         # this is terrible:
         idents, msg = self.feed_identities(msg, copy)
         try:
             return idents, self.unpack_message(msg, content=content, copy=copy)
         except Exception as e:
             print (idents, msg)
             # TODO: handle it
             raise e
     
     def feed_identities(self, msg, copy=True):
         """feed until DELIM is reached, then return the prefix as idents and remainder as
         msg. This is easily broken by setting an IDENT to DELIM, but that would be silly.
         
         Parameters
         ----------
         msg : a list of Message or bytes objects
             the message to be split
         copy : bool
             flag determining whether the arguments are bytes or Messages
         
         Returns
         -------
         (idents,msg) : two lists
             idents will always be a list of bytes - the indentity prefix
             msg will be a list of bytes or Messages, unchanged from input
             msg should be unpackable via self.unpack_message at this point.
         """
         ikey = int(self.key != '')
         minlen = 3 + ikey
         msg = list(msg)
         idents = []
         while len(msg) > minlen:
             if copy:
                 s = msg[0]
             else:
                 s = msg[0].bytes
             if s == DELIM:
                 msg.pop(0)
                 break
             else:
                 idents.append(s)
                 msg.pop(0)
                 
         return idents, msg
     
     def unpack_message(self, msg, content=True, copy=True):
         """Return a message object from the format
         sent by self.send.
         
         Parameters:
         -----------
         
         content : bool (True)
             whether to unpack the content dict (True), 
             or leave it serialized (False)
         
         copy : bool (True)
             whether to return the bytes (True), 
             or the non-copying Message object in each place (False)
         
         """
         ikey = int(self.key != '')
         minlen = 3 + ikey
         message = {}
         if not copy:
             for i in range(minlen):
                 msg[i] = msg[i].bytes
         if ikey:
             if not self.key == msg[0]:
                 raise KeyError("Invalid Session Key: %s"%msg[0])
         if not len(msg) >= minlen:
             raise TypeError("malformed message, must have at least %i elements"%minlen)
         message['header'] = self.unpack(msg[ikey+0])
         message['msg_type'] = message['header']['msg_type']
         message['parent_header'] = self.unpack(msg[ikey+1])
         if content:
             message['content'] = self.unpack(msg[ikey+2])
         else:
             message['content'] = msg[ikey+2]
         
         message['buffers'] = msg[ikey+3:]# [ m.buffer for m in msg[3:] ]
         return message
         
 
 def test_msg2obj():
     am = dict(x=1)
     ao = Message(am)
     assert ao.x == am['x']
 
     am['y'] = dict(z=1)
     ao = Message(am)
     assert ao.y.z == am['y']['z']
     
     k1, k2 = 'y', 'z'
     assert ao[k1][k2] == am[k1][k2]
     
     am2 = dict(ao)
     assert am['x'] == am2['x']
     assert am['y']['z'] == am2['y']['z']