Semi-working refactored ipcluster.

This version uses the NotificationCenter, which I am not sure I like.
But, it is important to see how all of this works.

IPython/config/loader.py

@@ -1,329 +1,330 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """A simple configuration system.
 
 Authors:
 
 * Brian Granger
 """
 
 #-----------------------------------------------------------------------------
 #  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 __builtin__
 import os
 import sys
 
 from IPython.external import argparse
 from IPython.utils.genutils import filefind
 
 #-----------------------------------------------------------------------------
 # Exceptions
 #-----------------------------------------------------------------------------
 
 
 class ConfigError(Exception):
     pass
 
 
 class ConfigLoaderError(ConfigError):
     pass
 
 
 #-----------------------------------------------------------------------------
 # Config class for holding config information
 #-----------------------------------------------------------------------------
 
 
 class Config(dict):
     """An attribute based dict that can do smart merges."""
 
     def __init__(self, *args, **kwds):
         dict.__init__(self, *args, **kwds)
         # This sets self.__dict__ = self, but it has to be done this way
         # because we are also overriding __setattr__.
         dict.__setattr__(self, '__dict__', self)
 
     def _merge(self, other):
         to_update = {}
         for k, v in other.items():
             if not self.has_key(k):
                 to_update[k] = v
             else: # I have this key
                 if isinstance(v, Config):
                     # Recursively merge common sub Configs
                     self[k]._merge(v)
                 else:
                     # Plain updates for non-Configs
                     to_update[k] = v
 
         self.update(to_update)
 
     def _is_section_key(self, key):
         if key[0].upper()==key[0] and not key.startswith('_'):
             return True
         else:
             return False
 
     def has_key(self, key):
         if self._is_section_key(key):
             return True
         else:
             return dict.has_key(self, key)
 
     def _has_section(self, key):
         if self._is_section_key(key):
             if dict.has_key(self, key):
                 return True
         return False
 
     def copy(self):
         return type(self)(dict.copy(self))
 
     def __copy__(self):
         return self.copy()
 
     def __deepcopy__(self, memo):
         import copy
         return type(self)(copy.deepcopy(self.items()))
 
     def __getitem__(self, key):
         # Because we use this for an exec namespace, we need to delegate
         # the lookup of names in __builtin__ to itself.  This means
         # that you can't have section or attribute names that are 
         # builtins.
         try:
             return getattr(__builtin__, key)
         except AttributeError:
             pass
         if self._is_section_key(key):
             try:
                 return dict.__getitem__(self, key)
             except KeyError:
                 c = Config()
                 dict.__setitem__(self, key, c)
                 return c
         else:
             return dict.__getitem__(self, key)
 
     def __setitem__(self, key, value):
         # Don't allow names in __builtin__ to be modified.
         if hasattr(__builtin__, key):
             raise ConfigError('Config variable names cannot have the same name '
                               'as a Python builtin: %s' % key)
         if self._is_section_key(key):
             if not isinstance(value, Config):
                 raise ValueError('values whose keys begin with an uppercase '
                                  'char must be Config instances: %r, %r' % (key, value))
         else:
             dict.__setitem__(self, key, value)
 
     def __getattr__(self, key):
         try:
             return self.__getitem__(key)
         except KeyError, e:
             raise AttributeError(e)
 
     def __setattr__(self, key, value):
         try:
             self.__setitem__(key, value)
         except KeyError, e:
             raise AttributeError(e)
 
     def __delattr__(self, key):
         try:
             dict.__delitem__(self, key)
         except KeyError, e:
             raise AttributeError(e)
 
 
 #-----------------------------------------------------------------------------
 # Config loading classes
 #-----------------------------------------------------------------------------
 
 
 class ConfigLoader(object):
     """A object for loading configurations from just about anywhere.
     
     The resulting configuration is packaged as a :class:`Struct`.
     
     Notes
     -----
     A :class:`ConfigLoader` does one thing: load a config from a source 
     (file, command line arguments) and returns the data as a :class:`Struct`.
     There are lots of things that :class:`ConfigLoader` does not do.  It does
     not implement complex logic for finding config files.  It does not handle
     default values or merge multiple configs.  These things need to be 
     handled elsewhere.
     """
 
     def __init__(self):
         """A base class for config loaders.
         
         Examples
         --------
         
         >>> cl = ConfigLoader()
         >>> config = cl.load_config()
         >>> config
         {}
         """
         self.clear()
 
     def clear(self):
         self.config = Config()
 
     def load_config(self):
         """Load a config from somewhere, return a Struct.
         
         Usually, this will cause self.config to be set and then returned.
         """
         return self.config
 
 
 class FileConfigLoader(ConfigLoader):
     """A base class for file based configurations.
 
     As we add more file based config loaders, the common logic should go
     here.
     """
     pass
 
 
 class PyFileConfigLoader(FileConfigLoader):
     """A config loader for pure python files.
     
     This calls execfile on a plain python file and looks for attributes
     that are all caps.  These attribute are added to the config Struct.
     """
 
     def __init__(self, filename, path=None):
         """Build a config loader for a filename and path.
 
         Parameters
         ----------
         filename : str
             The file name of the config file.
         path : str, list, tuple
             The path to search for the config file on, or a sequence of
             paths to try in order.
         """
         super(PyFileConfigLoader, self).__init__()
         self.filename = filename
         self.path = path
         self.full_filename = ''
         self.data = None
 
     def load_config(self):
         """Load the config from a file and return it as a Struct."""
         self._find_file()
         self._read_file_as_dict()
         self._convert_to_config()
         return self.config
 
     def _find_file(self):
         """Try to find the file by searching the paths."""
         self.full_filename = filefind(self.filename, self.path)
 
     def _read_file_as_dict(self):
         """Load the config file into self.config, with recursive loading."""
         # This closure is made available in the namespace that is used
         # to exec the config file.  This allows users to call
         # load_subconfig('myconfig.py') to load config files recursively.
         # It needs to be a closure because it has references to self.path
         # and self.config.  The sub-config is loaded with the same path
         # as the parent, but it uses an empty config which is then merged
         # with the parents.
         def load_subconfig(fname):
             loader = PyFileConfigLoader(fname, self.path)
             sub_config = loader.load_config()
             self.config._merge(sub_config)
 
         # Again, this needs to be a closure and should be used in config
         # files to get the config being loaded.
         def get_config():
             return self.config
 
         namespace = dict(load_subconfig=load_subconfig, get_config=get_config)
         execfile(self.full_filename, namespace)
 
     def _convert_to_config(self):
         if self.data is None:
             ConfigLoaderError('self.data does not exist')
 
 
 class CommandLineConfigLoader(ConfigLoader):
     """A config loader for command line arguments.
 
     As we add more command line based loaders, the common logic should go
     here.
     """
 
 
 class NoConfigDefault(object): pass
 NoConfigDefault = NoConfigDefault()
 
+
 class ArgParseConfigLoader(CommandLineConfigLoader):
     
     # arguments = [(('-f','--file'),dict(type=str,dest='file'))]
     arguments = ()
 
     def __init__(self, *args, **kw):
         """Create a config loader for use with argparse.
 
         The args and kwargs arguments here are passed onto the constructor
         of :class:`argparse.ArgumentParser`.
         """
         super(CommandLineConfigLoader, self).__init__()
         self.args = args
         self.kw = kw
 
     def load_config(self, args=None):
         """Parse command line arguments and return as a Struct."""
         self._create_parser()
         self._parse_args(args)
         self._convert_to_config()
         return self.config
 
     def get_extra_args(self):
         if hasattr(self, 'extra_args'):
             return self.extra_args
         else:
             return []
 
     def _create_parser(self):
         self.parser = argparse.ArgumentParser(*self.args, **self.kw)
         self._add_arguments()
         self._add_other_arguments()
 
     def _add_other_arguments(self):
         pass
 
     def _add_arguments(self):
         for argument in self.arguments:
             if not argument[1].has_key('default'):
                 argument[1]['default'] = NoConfigDefault
             self.parser.add_argument(*argument[0],**argument[1])
 
     def _parse_args(self, args=None):
         """self.parser->self.parsed_data"""
         if args is None:
             self.parsed_data, self.extra_args = self.parser.parse_known_args()
         else:
             self.parsed_data, self.extra_args = self.parser.parse_known_args(args)
 
     def _convert_to_config(self):
         """self.parsed_data->self.config"""
         for k, v in vars(self.parsed_data).items():
             if v is not NoConfigDefault:
                 exec_str = 'self.config.' + k + '= v'
                 exec exec_str in locals(), globals()
 

IPython/kernel/clusterdir.py

@@ -1,251 +1,263 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
 #-----------------------------------------------------------------------------
 
 import os
 import shutil
 
 from IPython.core import release
 from IPython.config.loader import PyFileConfigLoader
 from IPython.core.application import Application
 from IPython.core.component import Component
 from IPython.config.loader import ArgParseConfigLoader, NoConfigDefault
 from IPython.utils.traitlets import Unicode
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 
 class ClusterDir(Component):
     """An object to manage the cluster directory and its resources.
 
     The cluster directory is used by :command:`ipcontroller`, 
     :command:`ipcontroller` and :command:`ipcontroller` 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')
     security_dir = Unicode()
     log_dir = Unicode('')
     location = Unicode('')
 
     def __init__(self, location):
         super(ClusterDir, self).__init__(None)
         self.location = location
 
     def _location_changed(self, name, old, new):
         if not os.path.isdir(new):
             os.makedirs(new, mode=0777)
         else:
             os.chmod(new, 0777)
         self.security_dir = os.path.join(new, self.security_dir_name)
         self.log_dir = os.path.join(new, self.log_dir_name)
+        self.check_dirs()
 
     def _log_dir_changed(self, name, old, new):
-        if not os.path.isdir(new):
-            os.mkdir(new, 0777)
+        self.check_log_dir()
+
+    def check_log_dir(self):
+        if not os.path.isdir(self.log_dir):
+            os.mkdir(self.log_dir, 0777)
         else:
-            os.chmod(new, 0777)
+            os.chmod(self.log_dir, 0777)
 
     def _security_dir_changed(self, name, old, new):
-        if not os.path.isdir(new):
-            os.mkdir(new, 0700)
+        self.check_security_dir()
+
+    def check_security_dir(self):
+        if not os.path.isdir(self.security_dir):
+            os.mkdir(self.security_dir, 0700)
         else:
-            os.chmod(new, 0700)
+            os.chmod(self.security_dir, 0700)
+
+    def check_dirs(self):
+        self.check_security_dir()
+        self.check_log_dir()
 
     def load_config_file(self, filename):
         """Load a config file from the top level of the cluster dir.
 
         Parameters
         ----------
         filename : unicode or str
             The filename only of the config file that must be located in
             the top-level of the cluster directory.
         """
         loader = PyFileConfigLoader(filename, self.location)
         return loader.load_config()
 
-    def copy_config_file(self, config_file, path=None):
+    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)
-        shutil.copy(src, dst)
+        if not os.path.isfile(dst) or overwrite:
+            shutil.copy(src, dst)
 
-    def copy_all_config_files(self):
+    def copy_all_config_files(self, path=None, overwrite=False):
         """Copy all config files into the active cluster directory."""
-        for f in ['ipcontroller_config.py', 'ipengine_config.py']:
-            self.copy_config_file(f)
+        for f in ['ipcontroller_config.py', 'ipengine_config.py',
+                  'ipcluster_config.py']:
+            self.copy_config_file(f, path=path, overwrite=overwrite)
 
     @classmethod
     def find_cluster_dir_by_profile(cls, path, profile='default'):
         """Find/create a cluster dir by profile name and return its ClusterDir.
 
         This will create the cluster directory if it doesn't exist.
 
         Parameters
         ----------
         path : unicode or str
             The directory path to look for the cluster directory in.
         profile : unicode or str
             The name of the profile.  The name of the cluster directory
             will be "cluster_<profile>".
         """
         dirname = 'cluster_' + profile
         cluster_dir = os.path.join(os.getcwd(), dirname)
         if os.path.isdir(cluster_dir):
             return ClusterDir(cluster_dir)
         else:
             if not os.path.isdir(path):
                 raise IOError("Directory doesn't exist: %s" % path)
             cluster_dir = os.path.join(path, dirname)
             return ClusterDir(cluster_dir)
 
     @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:`os.path.expandvars` and :func:`os.path.expanduser`.
         """
         cluster_dir = os.path.expandvars(os.path.expanduser(cluster_dir))
         return ClusterDir(cluster_dir)
 
 
 class AppWithClusterDirArgParseConfigLoader(ArgParseConfigLoader):
     """Default command line options for IPython cluster applications."""
 
     def _add_other_arguments(self):
         self.parser.add_argument('-ipythondir', '--ipython-dir', 
             dest='Global.ipythondir',type=str,
             help='Set to override default location of Global.ipythondir.',
             default=NoConfigDefault,
             metavar='Global.ipythondir')
         self.parser.add_argument('-p','-profile', '--profile',
             dest='Global.profile',type=str,
             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.',
             default=NoConfigDefault,
             metavar='Global.profile')
         self.parser.add_argument('-log_level', '--log-level',
             dest="Global.log_level",type=int,
             help='Set the log level (0,10,20,30,40,50).  Default is 30.',
             default=NoConfigDefault)
         self.parser.add_argument('-cluster_dir', '--cluster-dir',
             dest='Global.cluster_dir',type=str,
             help='Set the cluster dir. This overrides the logic used by the '
             '--profile option.',
             default=NoConfigDefault,
             metavar='Global.cluster_dir')
 
 
 class ApplicationWithClusterDir(Application):
     """An application that puts everything into a cluster directory.
 
     Instead of looking for things in the ipythondir, 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 ipythondir 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.
     """
 
     def create_default_config(self):
         super(ApplicationWithClusterDir, self).create_default_config()
         self.default_config.Global.profile = 'default'
         self.default_config.Global.cluster_dir = ''
 
     def create_command_line_config(self):
         """Create and return a command line config loader."""
         return AppWithClusterDirArgParseConfigLoader(
             description=self.description, 
             version=release.version
         )
 
     def find_config_file_name(self):
         """Find the config file name for this application."""
         # For this type of Application it should be set as a class attribute.
         if not hasattr(self, 'config_file_name'):
             self.log.critical("No config filename found")
 
     def find_config_file_paths(self):
         """This resolves the cluster directory and sets ``config_file_paths``.
 
         This does the following:
         * Create the :class:`ClusterDir` object for the application.
         * Set the ``cluster_dir`` attribute of the application and config
           objects.
         * Set ``config_file_paths`` to point to the cluster directory.
         """
 
         # Create the ClusterDir object for managing everything
         try:
             cluster_dir = self.command_line_config.Global.cluster_dir
         except AttributeError:
             cluster_dir = self.default_config.Global.cluster_dir
         cluster_dir = os.path.expandvars(os.path.expanduser(cluster_dir))
         if cluster_dir:
             # Just use cluster_dir
             self.cluster_dir_obj = ClusterDir.find_cluster_dir(cluster_dir)
         else:
             # Then look for a profile
             try:
                 self.profile = self.command_line_config.Global.profile
             except AttributeError:
                 self.profile = self.default_config.Global.profile
             self.cluster_dir_obj = ClusterDir.find_cluster_dir_by_profile(
                 self.ipythondir, self.profile)
 
         # Set the cluster directory
         self.cluster_dir = self.cluster_dir_obj.location
         
         # These have to be set because they could be different from the one
         # that we just computed.  Because command line has the highest
         # priority, this will always end up in the master_config.
         self.default_config.Global.cluster_dir = self.cluster_dir
         self.command_line_config.Global.cluster_dir = self.cluster_dir
-        self.log.info("Cluster directory set to: %s" % self.cluster_dir)
 
         # Set the search path to the cluster directory
         self.config_file_paths = (self.cluster_dir,)

IPython/kernel/engineconnector.py

@@ -1,92 +1,131 b''
 # encoding: utf-8
 
 """A class that manages the engines connection to the controller."""
 
 __docformat__ = "restructuredtext en"
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008  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 cPickle as pickle
 
 from twisted.python import log, failure
 from twisted.internet import defer
+from twisted.internet.defer import inlineCallbacks, returnValue
 
 from IPython.kernel.fcutil import find_furl
 from IPython.kernel.enginefc import IFCEngine
+from IPython.kernel.twistedutil import sleep_deferred
 
 #-------------------------------------------------------------------------------
 # The ClientConnector class
 #-------------------------------------------------------------------------------
 
+
+class EngineConnectorError(Exception):
+    pass
+
+
 class EngineConnector(object):
     """Manage an engines connection to a controller.
     
     This class takes a foolscap `Tub` and provides a `connect_to_controller` 
     method that will use the `Tub` to connect to a controller and register
     the engine with the controller.
     """
     
     def __init__(self, tub):
         self.tub = tub
-        
-    def connect_to_controller(self, engine_service, furl_or_file):
+
+    def connect_to_controller(self, engine_service, furl_or_file,
+                              delay=0.1, max_tries=10):
         """
         Make a connection to a controller specified by a furl.
         
         This method takes an `IEngineBase` instance and a foolcap URL and uses
         the `tub` attribute to make a connection to the controller.  The 
         foolscap URL contains all the information needed to connect to the 
         controller, including the ip and port as well as any encryption and 
         authentication information needed for the connection.
-        
+
         After getting a reference to the controller, this method calls the 
         `register_engine` method of the controller to actually register the 
         engine.
-        
-        :Parameters:
-            engine_service : IEngineBase
-                An instance of an `IEngineBase` implementer
-            furl_or_file : str
-                A furl or a filename containing a furl
+
+        This method will try to connect to the controller multiple times with
+        a delay in between.  Each time the FURL file is read anew.
+
+        Parameters
+        __________
+        engine_service : IEngineBase
+            An instance of an `IEngineBase` implementer
+        furl_or_file : str
+            A furl or a filename containing a furl
+        delay : float
+            The intial time to wait between connection attempts.  Subsequent
+            attempts have increasing delays.
+        max_tries : int
+            The maximum number of connection attempts.
         """
         if not self.tub.running:
             self.tub.startService()
         self.engine_service = engine_service
         self.engine_reference = IFCEngine(self.engine_service)
-        try:
-            self.furl = find_furl(furl_or_file)
-        except ValueError:
-            return defer.fail(failure.Failure())
+
+        d = self._try_to_connect(furl_or_file, delay, max_tries, attempt=0)
+        return d
+
+    @inlineCallbacks
+    def _try_to_connect(self, furl_or_file, delay, max_tries, attempt):
+        """Try to connect to the controller with retry logic."""
+        if attempt < max_tries:
+            log.msg("Attempting to connect to controller [%r]: %s" % \
+                (attempt, furl_or_file))
+            try:
+                self.furl = find_furl(furl_or_file)
+                # Uncomment this to see the FURL being tried.
+                # log.msg("FURL: %s" % self.furl)
+                rr = yield self.tub.getReference(self.furl)
+            except:
+                if attempt==max_tries-1:
+                    # This will propagate the exception all the way to the top
+                    # where it can be handled.
+                    raise
+                else:
+                    yield sleep_deferred(delay)
+                    yield self._try_to_connect(
+                        furl_or_file, 1.5*delay, max_tries, attempt+1
+                    )
+            else:
+                result = yield self._register(rr)
+                returnValue(result)
         else:
-            d = self.tub.getReference(self.furl)
-            d.addCallbacks(self._register, self._log_failure)
-            return d
-    
-    def _log_failure(self, reason):
-        log.err('EngineConnector: engine registration failed:')
-        log.err(reason)
-        return reason
+            raise EngineConnectorError(
+                'Could not connect to controller, max_tries (%r) exceeded. '
+                'This usually means that i) the controller was not started, '
+                'or ii) a firewall was blocking the engine from connecting '
+                'to the controller.' % max_tries
+            )
 
     def _register(self, rr):
         self.remote_ref = rr
         # Now register myself with the controller
         desired_id = self.engine_service.id
         d = self.remote_ref.callRemote('register_engine', self.engine_reference, 
             desired_id, os.getpid(), pickle.dumps(self.engine_service.properties,2))
-        return d.addCallbacks(self._reference_sent, self._log_failure)
+        return d.addCallback(self._reference_sent)
 
     def _reference_sent(self, registration_dict):
         self.engine_service.id = registration_dict['id']
         log.msg("engine registration succeeded, got id: %r" % self.engine_service.id)
         return self.engine_service.id
 

IPython/kernel/fcutil.py

@@ -1,236 +1,239 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 Foolscap related utilities.
 """
 
 #-----------------------------------------------------------------------------
 #  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 tempfile
 
 from twisted.internet import reactor, defer
 from twisted.python import log
 
 from foolscap import Tub, UnauthenticatedTub
 
 from IPython.config.loader import Config
 
 from IPython.kernel.configobjfactory import AdaptedConfiguredObjectFactory
 
 from IPython.kernel.error import SecurityError
 
 from IPython.utils.traitlets import Int, Str, Bool, Instance
 from IPython.utils.importstring import import_item
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 
 # We do this so if a user doesn't have OpenSSL installed, it will try to use
 # an UnauthenticatedTub.  But, they will still run into problems if they
 # try to use encrypted furls.
 try:
     import OpenSSL
 except:
     Tub = UnauthenticatedTub
     have_crypto = False
 else:
     have_crypto = True
 
 
 def check_furl_file_security(furl_file, secure):
     """Remove the old furl_file if changing security modes."""
     if os.path.isfile(furl_file):
         f = open(furl_file, 'r')
         oldfurl = f.read().strip()
         f.close()
         if (oldfurl.startswith('pb://') and not secure) or (oldfurl.startswith('pbu://') and secure):
             os.remove(furl_file)
 
 
 def is_secure(furl):
     """Is the given FURL secure or not."""
     if is_valid(furl):
         if furl.startswith("pb://"):
             return True
         elif furl.startswith("pbu://"):
             return False
     else:
         raise ValueError("invalid FURL: %s" % furl)
 
 
 def is_valid(furl):
     """Is the str a valid FURL or not."""
     if isinstance(furl, str):
         if furl.startswith("pb://") or furl.startswith("pbu://"):
             return True
     else:
         return False
 
 
 def find_furl(furl_or_file):
     """Find, validate and return a FURL in a string or file."""
     if isinstance(furl_or_file, str):
         if is_valid(furl_or_file):
             return furl_or_file
     if os.path.isfile(furl_or_file):
-        furl = open(furl_or_file, 'r').read().strip()
+        with open(furl_or_file, 'r') as f:
+            furl = f.read().strip()
         if is_valid(furl):
             return furl
-    raise ValueError("not a FURL or a file containing a FURL: %s" % furl_or_file)
+    raise ValueError("Not a FURL or a file containing a FURL: %s" % furl_or_file)
 
 
 def get_temp_furlfile(filename):
     """Return a temporary FURL file."""
     return tempfile.mktemp(dir=os.path.dirname(filename),
                            prefix=os.path.basename(filename))
 
 
 def make_tub(ip, port, secure, cert_file):
     """Create a listening tub given an ip, port, and cert_file location.
     
     Parameters
     ----------
     ip : str
         The ip address or hostname that the tub should listen on.  
         Empty means all interfaces.
     port : int
         The port that the tub should listen on.  A value of 0 means
         pick a random port
     secure: bool
         Will the connection be secure (in the Foolscap sense).
     cert_file: str
         A filename of a file to be used for theSSL certificate.
 
     Returns
     -------
     A tub, listener tuple.
     """
     if secure:
         if have_crypto:
             tub = Tub(certFile=cert_file)
         else:
             raise SecurityError("OpenSSL/pyOpenSSL is not available, so we "
                                 "can't run in secure mode.  Try running without "
                                 "security using 'ipcontroller -xy'.")
     else:
         tub = UnauthenticatedTub()
     
     # Set the strport based on the ip and port and start listening
     if ip == '':
         strport = "tcp:%i" % port
     else:
         strport = "tcp:%i:interface=%s" % (port, ip)
     log.msg("Starting listener with [secure=%r] on: %s" % (secure, strport))
     listener = tub.listenOn(strport)
     
     return tub, listener
 
 
 class FCServiceFactory(AdaptedConfiguredObjectFactory):
     """This class creates a tub with various services running in it.
 
     The basic idea is that :meth:`create` returns a running :class:`Tub`
     instance that has a number of Foolscap references registered in it.
     This class is a subclass of :class:`IPython.core.component.Component`
     so the IPython configuration and component system are used.
 
     Attributes
     ----------
     interfaces : Config
         A Config instance whose values are sub-Config objects having two
         keys: furl_file and interface_chain.
 
     The other attributes are the standard ones for Foolscap.
     """
 
     ip = Str('', config=True)
     port = Int(0, config=True)
     secure = Bool(True, config=True)
     cert_file = Str('', config=True)
     location = Str('', config=True)
     reuse_furls = Bool(False, config=True)
     interfaces = Instance(klass=Config, kw={}, allow_none=False, config=True)
 
     def __init__(self, config, adaptee):
         super(FCServiceFactory, self).__init__(config, adaptee)
         self._check_reuse_furls()
 
     def _ip_changed(self, name, old, new):
         if new == 'localhost' or new == '127.0.0.1':
             self.location = '127.0.0.1'
 
     def _check_reuse_furls(self):
         furl_files = [i.furl_file for i in self.interfaces.values()]
         for ff in furl_files:
             fullfile = self._get_security_file(ff)
             if self.reuse_furls:
                 log.msg("Reusing FURL file: %s" % fullfile)
             else:
                 if os.path.isfile(fullfile):
                     log.msg("Removing old FURL file: %s" % fullfile)
                     os.remove(fullfile)
 
     def _get_security_file(self, filename):
         return os.path.join(self.config.Global.security_dir, filename)
 
     def create(self):
         """Create and return the Foolscap tub with everything running."""
 
         self.tub, self.listener = make_tub(
             self.ip, self.port, self.secure, 
             self._get_security_file(self.cert_file)
         )
         # log.msg("Interfaces to register [%r]: %r" % \
         #     (self.__class__, self.interfaces))
         if not self.secure:
             log.msg("WARNING: running with no security: %s" % \
                 self.__class__.__name__)
         reactor.callWhenRunning(self.set_location_and_register)
         return self.tub
 
     def set_location_and_register(self):
         """Set the location for the tub and return a deferred."""
 
         if self.location == '':
             d = self.tub.setLocationAutomatically()
         else:
             d = defer.maybeDeferred(self.tub.setLocation,
                 "%s:%i" % (self.location, self.listener.getPortnum()))
         self.adapt_to_interfaces(d)
 
     def adapt_to_interfaces(self, d):
         """Run through the interfaces, adapt and register."""
 
         for ifname, ifconfig in self.interfaces.iteritems():
             ff = self._get_security_file(ifconfig.furl_file)
             log.msg("Adapting [%s] to interface: %s" % \
                 (self.adaptee.__class__.__name__, ifname))
             log.msg("Saving FURL for interface [%s] to file: %s" % (ifname, ff))
             check_furl_file_security(ff, self.secure)
             adaptee = self.adaptee
             for i in ifconfig.interface_chain:
                 adaptee = import_item(i)(adaptee)
             d.addCallback(self.register, adaptee, furl_file=ff)
 
     def register(self, empty, ref, furl_file):
         """Register the reference with the FURL file.
 
         The FURL file is created and then moved to make sure that when the
         file appears, the buffer has been flushed and the file closed.
         """
         temp_furl_file = get_temp_furlfile(furl_file)
         self.tub.registerReference(ref, furlFile=temp_furl_file)
         os.rename(temp_furl_file, furl_file)
 

IPython/kernel/ipcontrollerapp.py

@@ -1,268 +1,274 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
 #-----------------------------------------------------------------------------
 
 import copy
 import os
 import sys
 
 from twisted.application import service
 from twisted.internet import reactor
 from twisted.python import log
 
 from IPython.config.loader import Config, NoConfigDefault
 
 from IPython.kernel.clusterdir import (
     ApplicationWithClusterDir, 
     AppWithClusterDirArgParseConfigLoader
 )
 
 from IPython.core import release
 
 from IPython.utils.traitlets import Str, Instance
 
 from IPython.kernel import controllerservice
 
 from IPython.kernel.fcutil import FCServiceFactory
 
 #-----------------------------------------------------------------------------
 # Default interfaces
 #-----------------------------------------------------------------------------
 
 
 # The default client interfaces for FCClientServiceFactory.interfaces
 default_client_interfaces = Config()
 default_client_interfaces.Task.interface_chain = [
     'IPython.kernel.task.ITaskController',
     'IPython.kernel.taskfc.IFCTaskController'
 ]
 
 default_client_interfaces.Task.furl_file = 'ipcontroller-tc.furl'
 
 default_client_interfaces.MultiEngine.interface_chain = [
     'IPython.kernel.multiengine.IMultiEngine',
     'IPython.kernel.multienginefc.IFCSynchronousMultiEngine'
 ]
 
 default_client_interfaces.MultiEngine.furl_file = 'ipcontroller-mec.furl'
 
 # Make this a dict we can pass to Config.__init__ for the default
 default_client_interfaces = dict(copy.deepcopy(default_client_interfaces.items()))
 
 
 
 # The default engine interfaces for FCEngineServiceFactory.interfaces
 default_engine_interfaces = Config()
 default_engine_interfaces.Default.interface_chain = [
     'IPython.kernel.enginefc.IFCControllerBase'
 ]
 
 default_engine_interfaces.Default.furl_file = 'ipcontroller-engine.furl'
 
 # Make this a dict we can pass to Config.__init__ for the default
 default_engine_interfaces = dict(copy.deepcopy(default_engine_interfaces.items()))
 
 
 #-----------------------------------------------------------------------------
 # Service factories
 #-----------------------------------------------------------------------------
 
 
 class FCClientServiceFactory(FCServiceFactory):
     """A Foolscap implementation of the client services."""
 
     cert_file = Str('ipcontroller-client.pem', config=True)
     interfaces = Instance(klass=Config, kw=default_client_interfaces,
                           allow_none=False, config=True)
 
 
 class FCEngineServiceFactory(FCServiceFactory):
     """A Foolscap implementation of the engine services."""
 
     cert_file = Str('ipcontroller-engine.pem', config=True)
     interfaces = Instance(klass=dict, kw=default_engine_interfaces,
                           allow_none=False, config=True)
 
 
 #-----------------------------------------------------------------------------
 # The main application
 #-----------------------------------------------------------------------------
 
 
 cl_args = (
     # Client config
     (('--client-ip',), dict(
         type=str, dest='FCClientServiceFactory.ip', default=NoConfigDefault,
         help='The IP address or hostname the controller will listen on for '
         'client connections.',
         metavar='FCClientServiceFactory.ip')
     ),
     (('--client-port',), dict(
         type=int, dest='FCClientServiceFactory.port', default=NoConfigDefault,
         help='The port the controller will listen on for client connections. '
         'The default is to use 0, which will autoselect an open port.',
         metavar='FCClientServiceFactory.port')
     ),
     (('--client-location',), dict(
         type=str, dest='FCClientServiceFactory.location', default=NoConfigDefault,
         help='The hostname or IP that clients should connect to. This does '
         'not control which interface the controller listens on. Instead, this '
         'determines the hostname/IP that is listed in the FURL, which is how '
         'clients know where to connect. Useful if the controller is listening '
         'on multiple interfaces.',
         metavar='FCClientServiceFactory.location')
     ),
     # Engine config
     (('--engine-ip',), dict(
         type=str, dest='FCEngineServiceFactory.ip', default=NoConfigDefault,
         help='The IP address or hostname the controller will listen on for '
         'engine connections.',
         metavar='FCEngineServiceFactory.ip')
     ),
     (('--engine-port',), dict(
         type=int, dest='FCEngineServiceFactory.port', default=NoConfigDefault,
         help='The port the controller will listen on for engine connections. '
         'The default is to use 0, which will autoselect an open port.',
         metavar='FCEngineServiceFactory.port')
     ),
     (('--engine-location',), dict(
         type=str, dest='FCEngineServiceFactory.location', default=NoConfigDefault,
         help='The hostname or IP that engines should connect to. This does '
         'not control which interface the controller listens on. Instead, this '
         'determines the hostname/IP that is listed in the FURL, which is how '
         'engines know where to connect. Useful if the controller is listening '
         'on multiple interfaces.',
         metavar='FCEngineServiceFactory.location')
     ),
     # Global config
     (('--log-to-file',), dict(
         action='store_true', dest='Global.log_to_file', default=NoConfigDefault,
         help='Log to a file in the log directory (default is stdout)')
     ),
     (('-r','--reuse-furls'), dict(
         action='store_true', dest='Global.reuse_furls', default=NoConfigDefault,
         help='Try to reuse all FURL files. If this is not set all FURL files '
         'are deleted before the controller starts. This must be set if '
         'specific ports are specified by --engine-port or --client-port.')
     ),
     (('-ns','--no-security'), dict(
         action='store_false', dest='Global.secure', default=NoConfigDefault,
         help='Turn off SSL encryption for all connections.')
     )
 )
 
 
 class IPControllerAppCLConfigLoader(AppWithClusterDirArgParseConfigLoader):
 
     arguments = cl_args
 
 
 default_config_file_name = 'ipcontroller_config.py'
 
 
 class IPControllerApp(ApplicationWithClusterDir):
 
     name = 'ipcontroller'
     description = 'Start the IPython controller for parallel computing.'
     config_file_name = default_config_file_name
 
     def create_default_config(self):
         super(IPControllerApp, self).create_default_config()
         self.default_config.Global.reuse_furls = False
         self.default_config.Global.secure = True
         self.default_config.Global.import_statements = []
         self.default_config.Global.log_to_file = False
 
     def create_command_line_config(self):
         """Create and return a command line config loader."""
         return IPControllerAppCLConfigLoader(
             description=self.description, 
             version=release.version
         )
 
     def post_load_command_line_config(self):
         # Now setup reuse_furls
         c = self.command_line_config
         if hasattr(c.Global, 'reuse_furls'):
             c.FCClientServiceFactory.reuse_furls = c.Global.reuse_furls
             c.FCEngineServiceFactory.reuse_furls = c.Global.reuse_furls
             del c.Global.reuse_furls
         if hasattr(c.Global, 'secure'):
             c.FCClientServiceFactory.secure = c.Global.secure
             c.FCEngineServiceFactory.secure = c.Global.secure
             del c.Global.secure
 
     def pre_construct(self):
         # The log and security dirs were set earlier, but here we put them
         # into the config and log them.
         config = self.master_config
         sdir = self.cluster_dir_obj.security_dir
         self.security_dir = config.Global.security_dir = sdir
         ldir = self.cluster_dir_obj.log_dir
         self.log_dir = config.Global.log_dir = ldir
+        self.log.info("Cluster directory set to: %s" % self.cluster_dir)
         self.log.info("Log directory set to: %s" % self.log_dir)
         self.log.info("Security directory set to: %s" % self.security_dir)
 
     def construct(self):
         # I am a little hesitant to put these into InteractiveShell itself.
         # But that might be the place for them
         sys.path.insert(0, '')
 
         self.start_logging()
         self.import_statements()
         
         # Create the service hierarchy
         self.main_service = service.MultiService()
         # The controller service
         controller_service = controllerservice.ControllerService()
         controller_service.setServiceParent(self.main_service)
         # The client tub and all its refereceables
         csfactory = FCClientServiceFactory(self.master_config, controller_service)
         client_service = csfactory.create()
         client_service.setServiceParent(self.main_service)
         # The engine tub
         esfactory = FCEngineServiceFactory(self.master_config, controller_service)
         engine_service = esfactory.create()
         engine_service.setServiceParent(self.main_service)
 
     def start_logging(self):
         if self.master_config.Global.log_to_file:
             log_filename = self.name + '-' + str(os.getpid()) + '.log'
             logfile = os.path.join(self.log_dir, log_filename)
             open_log_file = open(logfile, 'w')
         else:
             open_log_file = sys.stdout
         log.startLogging(open_log_file)
 
     def import_statements(self):
         statements = self.master_config.Global.import_statements
         for s in statements:
             try:
                 log.msg("Executing statement: '%s'" % s)
                 exec s in globals(), locals()
             except:
                 log.msg("Error running statement: %s" % s)
 
     def start_app(self):
         # Start the controller service and set things running
         self.main_service.startService()
         reactor.run()
 
 
 def launch_new_instance():
     """Create and run the IPython controller"""
     app = IPControllerApp()
     app.start()
+
+
+if __name__ == '__main__':
+    launch_new_instance()
+

IPython/kernel/ipengineapp.py

@@ -1,245 +1,248 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
 #-----------------------------------------------------------------------------
 
 import os
 import sys
 
 from twisted.application import service
 from twisted.internet import reactor
 from twisted.python import log
 
 from IPython.config.loader import NoConfigDefault
 
 from IPython.kernel.clusterdir import (
     ApplicationWithClusterDir, 
     AppWithClusterDirArgParseConfigLoader
 )
 from IPython.core import release
 
 from IPython.utils.importstring import import_item
 
 from IPython.kernel.engineservice import EngineService
 from IPython.kernel.fcutil import Tub
 from IPython.kernel.engineconnector import EngineConnector
 
 #-----------------------------------------------------------------------------
 # The main application
 #-----------------------------------------------------------------------------
 
 
 cl_args = (
     # Controller config
     (('--furl-file',), dict(
         type=str, dest='Global.furl_file', default=NoConfigDefault,
         help='The full location of the file containing the FURL of the '
         'controller. If this is not given, the FURL file must be in the '
         'security directory of the cluster directory.  This location is '
         'resolved using the --profile and --app-dir options.',
         metavar='Global.furl_file')
     ),
     # MPI
     (('--mpi',), dict(
         type=str, dest='MPI.use', default=NoConfigDefault,
         help='How to enable MPI (mpi4py, pytrilinos, or empty string to disable).',
         metavar='MPI.use')
     ),
     # Global config
     (('--log-to-file',), dict(
         action='store_true', dest='Global.log_to_file', default=NoConfigDefault,
         help='Log to a file in the log directory (default is stdout)')
     )
 )
 
 
 class IPEngineAppCLConfigLoader(AppWithClusterDirArgParseConfigLoader):
 
     arguments = cl_args
 
 
 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
 """
 
 
 default_config_file_name = 'ipengine_config.py'
 
 
 class IPEngineApp(ApplicationWithClusterDir):
 
     name = 'ipengine'
     description = 'Start the IPython engine for parallel computing.'
     config_file_name = default_config_file_name
 
     def create_default_config(self):
         super(IPEngineApp, self).create_default_config()
 
         # Global config attributes
         self.default_config.Global.log_to_file = False
         self.default_config.Global.exec_lines = []
         # The log and security dir names must match that of the controller
         self.default_config.Global.log_dir_name = 'log'
         self.default_config.Global.security_dir_name = 'security'
         self.default_config.Global.shell_class = 'IPython.kernel.core.interpreter.Interpreter'
 
         # Configuration related to the controller
         # This must match the filename (path not included) that the controller
         # used for the FURL file.
         self.default_config.Global.furl_file_name = 'ipcontroller-engine.furl'
         # If given, this is the actual location of the controller's FURL file.
         # If not, this is computed using the profile, app_dir and furl_file_name
         self.default_config.Global.furl_file = ''
 
         # MPI related config attributes
         self.default_config.MPI.use = ''
         self.default_config.MPI.mpi4py = mpi4py_init
         self.default_config.MPI.pytrilinos = pytrilinos_init
 
     def create_command_line_config(self):
         """Create and return a command line config loader."""
         return IPEngineAppCLConfigLoader(
             description=self.description, 
             version=release.version
         )
 
     def post_load_command_line_config(self):
         pass
 
     def pre_construct(self):
         config = self.master_config
         sdir = self.cluster_dir_obj.security_dir
         self.security_dir = config.Global.security_dir = sdir
         ldir = self.cluster_dir_obj.log_dir
         self.log_dir = config.Global.log_dir = ldir
+        self.log.info("Cluster directory set to: %s" % self.cluster_dir)
         self.log.info("Log directory set to: %s" % self.log_dir)
         self.log.info("Security directory set to: %s" % self.security_dir)
 
         self.find_cont_furl_file()
 
     def find_cont_furl_file(self):
+        """Set the furl 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 FURL file
-        if os.path.isfile(config.Global.furl_file):
-            return
-        else:
-            # We should know what the app dir is
+        if not config.Global.furl_file:
             try_this = os.path.join(
                 config.Global.cluster_dir, 
                 config.Global.security_dir,
                 config.Global.furl_file_name
             )
-            if os.path.isfile(try_this):
-                config.Global.furl_file = try_this
-                return
-            else:
-                self.log.critical("Could not find a valid controller FURL file.")
-                self.abort()
+            config.Global.furl_file = try_this
 
     def construct(self):
         # I am a little hesitant to put these into InteractiveShell itself.
         # But that might be the place for them
         sys.path.insert(0, '')
 
         self.start_mpi()
         self.start_logging()
 
         # Create the underlying shell class and EngineService
         shell_class = import_item(self.master_config.Global.shell_class)
         self.engine_service = EngineService(shell_class, mpi=mpi)
 
         self.exec_lines()
 
         # 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)
 
         log.msg("Using furl file: %s" % self.master_config.Global.furl_file)
 
         reactor.callWhenRunning(self.call_connect)
 
     def call_connect(self):
         d = self.engine_connector.connect_to_controller(
             self.engine_service, 
             self.master_config.Global.furl_file
         )
 
         def handle_error(f):
-            # If this print statement is replaced by a log.err(f) I get
-            # an unhandled error, which makes no sense.  I shouldn't have
-            # to use a print statement here.  My only thought is that
-            # at the beginning of the process the logging is still starting up
-            print "Error connecting to controller:", f.getErrorMessage()
+            log.msg('Error connecting to controller. This usually means that '
+            'i) the controller was not started, ii) a firewall was blocking '
+            'the engine from connecting to the controller or iii) the engine '
+            ' was not pointed at the right FURL file:')
+            log.msg(f.getErrorMessage())
             reactor.callLater(0.1, reactor.stop)
 
         d.addErrback(handle_error)
 
     def start_mpi(self):
         global mpi
         mpikey = self.master_config.MPI.use
         mpi_import_statement = self.master_config.MPI.get(mpikey, None)
         if mpi_import_statement is not None:
             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 start_logging(self):
         if self.master_config.Global.log_to_file:
             log_filename = self.name + '-' + str(os.getpid()) + '.log'
             logfile = os.path.join(self.log_dir, log_filename)
             open_log_file = open(logfile, 'w')
         else:
             open_log_file = sys.stdout
         log.startLogging(open_log_file)
 
     def exec_lines(self):
         for line in self.master_config.Global.exec_lines:
             try:
                 log.msg("Executing statement: '%s'" % line)
                 self.engine_service.execute(line)
             except:
                 log.msg("Error executing statement: %s" % line)
 
     def start_app(self):
         # Start the controller service and set things running
         reactor.run()
 
 
 def launch_new_instance():
     """Create and run the IPython controller"""
     app = IPEngineApp()
     app.start()
+
+
+if __name__ == '__main__':
+    launch_new_instance()
+

IPython/kernel/scripts/ipcluster

@@ -1,22 +1,18 b''
 #!/usr/bin/env python
 # encoding: utf-8
 
-"""ipcluster script"""
-
-__docformat__ = "restructuredtext en"
-
-#-------------------------------------------------------------------------------
-#  Copyright (C) 2008  The IPython Development Team
+#-----------------------------------------------------------------------------
+#  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
-#-------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------
+
 
-if __name__ == '__main__':
-	from IPython.kernel.scripts import ipcluster
-	ipcluster.main()
+from IPython.kernel.ipclusterapp import launch_new_instance
 
+launch_new_instance()

IPython/kernel/scripts/ipcluster.py

@@ -1,813 +1,811 b''
  #!/usr/bin/env python
 # encoding: utf-8
 
 """Start an IPython cluster = (controller + engines)."""
 
 #-----------------------------------------------------------------------------
-#  Copyright (C) 2008  The IPython Development Team
+#  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 sys
 import signal
 import tempfile
 pjoin = os.path.join
 
 from twisted.internet import reactor, defer
 from twisted.internet.protocol import ProcessProtocol
 from twisted.internet.error import ProcessDone, ProcessTerminated
 from twisted.internet.utils import getProcessOutput
-from twisted.python import failure, log
+from twisted.python import log
 
 from IPython.external import argparse
 from IPython.external import Itpl
 from IPython.utils.genutils import (
     get_ipython_dir, 
     get_log_dir, 
     get_security_dir, 
     num_cpus
 )
 from IPython.kernel.fcutil import have_crypto
 
 # Create various ipython directories if they don't exist.
 # This must be done before IPython.kernel.config is imported.
 from IPython.core.oldusersetup import user_setup
 if os.name == 'posix':
     rc_suffix = ''
 else:
     rc_suffix = '.ini'
 user_setup(get_ipython_dir(), rc_suffix, mode='install', interactive=False)
 get_log_dir()
 get_security_dir()
 
 from IPython.kernel.config import config_manager as kernel_config_manager
-from IPython.kernel.error import SecurityError, FileTimeoutError
-from IPython.kernel.fcutil import have_crypto
 from IPython.kernel.twistedutil import gatherBoth, wait_for_file
-from IPython.kernel.util import printer
+
 
 #-----------------------------------------------------------------------------
 # General process handling code
 #-----------------------------------------------------------------------------
 
 
 class ProcessStateError(Exception):
     pass
 
 class UnknownStatus(Exception):
     pass
 
 class LauncherProcessProtocol(ProcessProtocol):
     """
     A ProcessProtocol to go with the ProcessLauncher.
     """
     def __init__(self, process_launcher):
         self.process_launcher = process_launcher
     
     def connectionMade(self):
         self.process_launcher.fire_start_deferred(self.transport.pid)
     
     def processEnded(self, status):
         value = status.value
         if isinstance(value, ProcessDone):
             self.process_launcher.fire_stop_deferred(0)
         elif isinstance(value, ProcessTerminated):
             self.process_launcher.fire_stop_deferred(
                 {'exit_code':value.exitCode,
                  'signal':value.signal,
                  'status':value.status
                 }
             )
         else:
             raise UnknownStatus("unknown exit status, this is probably a bug in Twisted")
     
     def outReceived(self, data):
         log.msg(data)
     
     def errReceived(self, data):
         log.err(data)
 
 class ProcessLauncher(object):
     """
     Start and stop an external process in an asynchronous manner.
     
     Currently this uses deferreds to notify other parties of process state
     changes.  This is an awkward design and should be moved to using
     a formal NotificationCenter.
     """
     def __init__(self, cmd_and_args):
         self.cmd = cmd_and_args[0]
         self.args = cmd_and_args
         self._reset()
     
     def _reset(self):
         self.process_protocol = None
         self.pid = None
         self.start_deferred = None
         self.stop_deferreds = []
         self.state = 'before' # before, running, or after
     
     @property
     def running(self):
         if self.state == 'running':
             return True
         else:
             return False
     
     def fire_start_deferred(self, pid):
         self.pid = pid
         self.state = 'running'
         log.msg('Process %r has started with pid=%i' % (self.args, pid))
         self.start_deferred.callback(pid)
     
     def start(self):
         if self.state == 'before':
             self.process_protocol = LauncherProcessProtocol(self)
             self.start_deferred = defer.Deferred()
             self.process_transport = reactor.spawnProcess(
                 self.process_protocol,
                 self.cmd,
                 self.args,
                 env=os.environ
             )
             return self.start_deferred
         else:
-            s = 'the process has already been started and has state: %r' % \
+            s = 'The process has already been started and has state: %r' % \
                 self.state
             return defer.fail(ProcessStateError(s))
     
     def get_stop_deferred(self):
         if self.state == 'running' or self.state == 'before':
             d = defer.Deferred()
             self.stop_deferreds.append(d)
             return d
         else:
             s = 'this process is already complete'
             return defer.fail(ProcessStateError(s))
     
     def fire_stop_deferred(self, exit_code):
         log.msg('Process %r has stopped with %r' % (self.args, exit_code))
         self.state = 'after'
         for d in self.stop_deferreds:
             d.callback(exit_code)
     
     def signal(self, sig):
         """
         Send a signal to the process.
         
         The argument sig can be ('KILL','INT', etc.) or any signal number.
         """
         if self.state == 'running':
             self.process_transport.signalProcess(sig)
     
     # def __del__(self):
     #     self.signal('KILL')
     
     def interrupt_then_kill(self, delay=1.0):
         self.signal('INT')
         reactor.callLater(delay, self.signal, 'KILL')
 
 
 #-----------------------------------------------------------------------------
 # Code for launching controller and engines
 #-----------------------------------------------------------------------------
 
 
 class ControllerLauncher(ProcessLauncher):
     
     def __init__(self, extra_args=None):
         if sys.platform == 'win32':
             # This logic is needed because the ipcontroller script doesn't
             # always get installed in the same way or in the same location.
             from IPython.kernel.scripts import ipcontroller
             script_location = ipcontroller.__file__.replace('.pyc', '.py')
             # The -u option here turns on unbuffered output, which is required
             # on Win32 to prevent wierd conflict and problems with Twisted.
             # Also, use sys.executable to make sure we are picking up the 
             # right python exe.
             args = [sys.executable, '-u', script_location]
         else:
             args = ['ipcontroller']
         self.extra_args = extra_args
         if extra_args is not None:
             args.extend(extra_args)
         
         ProcessLauncher.__init__(self, args)
 
 
 class EngineLauncher(ProcessLauncher):
     
     def __init__(self, extra_args=None):
         if sys.platform == 'win32':
             # This logic is needed because the ipcontroller script doesn't
             # always get installed in the same way or in the same location.
             from IPython.kernel.scripts import ipengine
             script_location = ipengine.__file__.replace('.pyc', '.py')
             # The -u option here turns on unbuffered output, which is required
             # on Win32 to prevent wierd conflict and problems with Twisted.
             # Also, use sys.executable to make sure we are picking up the 
             # right python exe.
             args = [sys.executable, '-u', script_location]
         else:
             args = ['ipengine']
         self.extra_args = extra_args
         if extra_args is not None:
             args.extend(extra_args)
         
         ProcessLauncher.__init__(self, args)
 
 
 class LocalEngineSet(object):
     
     def __init__(self, extra_args=None):
         self.extra_args = extra_args
         self.launchers = []
     
     def start(self, n):
         dlist = []
         for i in range(n):
             el = EngineLauncher(extra_args=self.extra_args)
             d = el.start()
             self.launchers.append(el)
             dlist.append(d)
         dfinal = gatherBoth(dlist, consumeErrors=True)
         dfinal.addCallback(self._handle_start)
         return dfinal
     
     def _handle_start(self, r):
         log.msg('Engines started with pids: %r' % r)
         return r
     
     def _handle_stop(self, r):
         log.msg('Engines received signal: %r' % r)
         return r
     
     def signal(self, sig):
         dlist = []
         for el in self.launchers:
             d = el.get_stop_deferred()
             dlist.append(d)
             el.signal(sig)
         dfinal = gatherBoth(dlist, consumeErrors=True)
         dfinal.addCallback(self._handle_stop)
         return dfinal
     
     def interrupt_then_kill(self, delay=1.0):
         dlist = []
         for el in self.launchers:
             d = el.get_stop_deferred()
             dlist.append(d)
             el.interrupt_then_kill(delay)
         dfinal = gatherBoth(dlist, consumeErrors=True)
         dfinal.addCallback(self._handle_stop)
         return dfinal
 
 
 class BatchEngineSet(object):
     
     # Subclasses must fill these in.  See PBSEngineSet
     submit_command = ''
     delete_command = ''
     job_id_regexp = ''
     
     def __init__(self, template_file, **kwargs):
         self.template_file = template_file
         self.context = {}
         self.context.update(kwargs)
         self.batch_file = self.template_file+'-run'
     
     def parse_job_id(self, output):
         m = re.match(self.job_id_regexp, output)
         if m is not None:
             job_id = m.group()
         else:
             raise Exception("job id couldn't be determined: %s" % output)
         self.job_id = job_id
         log.msg('Job started with job id: %r' % job_id)
         return job_id
     
     def write_batch_script(self, n):
         self.context['n'] = n
         template = open(self.template_file, 'r').read()
         log.msg('Using template for batch script: %s' % self.template_file)
         script_as_string = Itpl.itplns(template, self.context)
         log.msg('Writing instantiated batch script: %s' % self.batch_file)
         f = open(self.batch_file,'w')
         f.write(script_as_string)
         f.close()
     
     def handle_error(self, f):
         f.printTraceback()
         f.raiseException()
     
     def start(self, n):
         self.write_batch_script(n)
         d = getProcessOutput(self.submit_command,
             [self.batch_file],env=os.environ)
         d.addCallback(self.parse_job_id)
         d.addErrback(self.handle_error)
         return d
     
     def kill(self):
         d = getProcessOutput(self.delete_command,
             [self.job_id],env=os.environ)
         return d
 
 class PBSEngineSet(BatchEngineSet):
     
     submit_command = 'qsub'
     delete_command = 'qdel'
     job_id_regexp = '\d+'
     
     def __init__(self, template_file, **kwargs):
         BatchEngineSet.__init__(self, template_file, **kwargs)
 
 
 sshx_template="""#!/bin/sh
 "$@" &> /dev/null &
 echo $!
 """
 
 engine_killer_template="""#!/bin/sh
 ps -fu `whoami` | grep '[i]pengine' | awk '{print $2}' | xargs kill -TERM
 """
 
 class SSHEngineSet(object):
     sshx_template=sshx_template
     engine_killer_template=engine_killer_template
     
     def __init__(self, engine_hosts, sshx=None, ipengine="ipengine"):
         """Start a controller on localhost and engines using ssh.
         
         The engine_hosts argument is a dict with hostnames as keys and
         the number of engine (int) as values.  sshx is the name of a local
         file that will be used to run remote commands.  This file is used
         to setup the environment properly.
         """
         
         self.temp_dir = tempfile.gettempdir()
         if sshx is not None:
             self.sshx = sshx
         else:
             # Write the sshx.sh file locally from our template.
             self.sshx = os.path.join(
                 self.temp_dir,
                 '%s-main-sshx.sh' % os.environ['USER']
             )
             f = open(self.sshx, 'w')
             f.writelines(self.sshx_template)
             f.close()
         self.engine_command = ipengine
         self.engine_hosts = engine_hosts
         # Write the engine killer script file locally from our template.
         self.engine_killer = os.path.join(
             self.temp_dir,  
             '%s-local-engine_killer.sh' % os.environ['USER']
         )
         f = open(self.engine_killer, 'w')
         f.writelines(self.engine_killer_template)
         f.close()
     
     def start(self, send_furl=False):
         dlist = []
         for host in self.engine_hosts.keys():
             count = self.engine_hosts[host]
             d = self._start(host, count, send_furl)
             dlist.append(d)
         return gatherBoth(dlist, consumeErrors=True)
     
     def _start(self, hostname, count=1, send_furl=False):
         if send_furl:
             d = self._scp_furl(hostname)
         else:
             d = defer.succeed(None)
         d.addCallback(lambda r: self._scp_sshx(hostname))
         d.addCallback(lambda r: self._ssh_engine(hostname, count))
         return d
         
     def _scp_furl(self, hostname):
         scp_cmd = "scp ~/.ipython/security/ipcontroller-engine.furl %s:.ipython/security/" % (hostname)
         cmd_list = scp_cmd.split()
         cmd_list[1] = os.path.expanduser(cmd_list[1])
         log.msg('Copying furl file: %s' % scp_cmd)
         d = getProcessOutput(cmd_list[0], cmd_list[1:], env=os.environ) 
         return d
     
     def _scp_sshx(self, hostname):
         scp_cmd = "scp %s %s:%s/%s-sshx.sh" % (
             self.sshx, hostname, 
             self.temp_dir, os.environ['USER']
         )
         print
         log.msg("Copying sshx: %s" % scp_cmd)
         sshx_scp = scp_cmd.split()
         d = getProcessOutput(sshx_scp[0], sshx_scp[1:], env=os.environ)
         return d
     
     def _ssh_engine(self, hostname, count):
         exec_engine = "ssh %s sh %s/%s-sshx.sh %s" % (
             hostname, self.temp_dir, 
             os.environ['USER'], self.engine_command
         )
         cmds = exec_engine.split()
         dlist = []
         log.msg("about to start engines...")
         for i in range(count):
             log.msg('Starting engines: %s' % exec_engine)
             d = getProcessOutput(cmds[0], cmds[1:], env=os.environ)
             dlist.append(d)
         return gatherBoth(dlist, consumeErrors=True)
     
     def kill(self):
         dlist = []
         for host in self.engine_hosts.keys():
             d = self._killall(host)
             dlist.append(d)
         return gatherBoth(dlist, consumeErrors=True)
     
     def _killall(self, hostname):
         d = self._scp_engine_killer(hostname)
         d.addCallback(lambda r: self._ssh_kill(hostname))
         # d.addErrback(self._exec_err)
         return d
 
     def _scp_engine_killer(self, hostname):
         scp_cmd = "scp %s %s:%s/%s-engine_killer.sh" % (
             self.engine_killer, 
             hostname, 
             self.temp_dir, 
             os.environ['USER']
         )
         cmds = scp_cmd.split()
         log.msg('Copying engine_killer: %s' % scp_cmd)
         d = getProcessOutput(cmds[0], cmds[1:], env=os.environ)
         return d
     
     def _ssh_kill(self, hostname):
         kill_cmd = "ssh %s sh %s/%s-engine_killer.sh" % (
             hostname,
             self.temp_dir, 
             os.environ['USER']
         )
         log.msg('Killing engine: %s' % kill_cmd)
         kill_cmd = kill_cmd.split()
         d = getProcessOutput(kill_cmd[0], kill_cmd[1:], env=os.environ)
         return d
 
     def _exec_err(self, r):
         log.msg(r)
 
 #-----------------------------------------------------------------------------
 # Main functions for the different types of clusters
 #-----------------------------------------------------------------------------
 
 # TODO:
 # The logic in these codes should be moved into classes like LocalCluster
 # MpirunCluster, PBSCluster, etc.  This would remove alot of the duplications.
 # The main functions should then just parse the command line arguments, create
 # the appropriate class and call a 'start' method.
 
 
 def check_security(args, cont_args):
     """Check to see if we should run with SSL support."""
     if (not args.x or not args.y) and not have_crypto:
         log.err("""
 OpenSSL/pyOpenSSL is not available, so we can't run in secure mode.
 Try running ipcluster with the -xy flags:  ipcluster local -xy -n 4""")
         reactor.stop()
         return False
     if args.x:
         cont_args.append('-x')
     if args.y:
         cont_args.append('-y')
     return True
 
 
 def check_reuse(args, cont_args):
     """Check to see if we should try to resuse FURL files."""
     if args.r:
         cont_args.append('-r')
         if args.client_port == 0 or args.engine_port == 0:
             log.err("""
 To reuse FURL files, you must also set the client and engine ports using
 the --client-port and --engine-port options.""")
             reactor.stop()
             return False
         cont_args.append('--client-port=%i' % args.client_port)
         cont_args.append('--engine-port=%i' % args.engine_port)
     return True
 
 
 def _err_and_stop(f):
     """Errback to log a failure and halt the reactor on a fatal error."""
     log.err(f)
     reactor.stop()
 
 
 def _delay_start(cont_pid, start_engines, furl_file, reuse):
     """Wait for controller to create FURL files and the start the engines."""
     if not reuse:
         if os.path.isfile(furl_file):
             os.unlink(furl_file)
     log.msg('Waiting for controller to finish starting...')
     d = wait_for_file(furl_file, delay=0.2, max_tries=50)
     d.addCallback(lambda _: log.msg('Controller started'))
     d.addCallback(lambda _: start_engines(cont_pid))
     return d
 
 
 def main_local(args):
     cont_args = []
     cont_args.append('--logfile=%s' % pjoin(args.logdir,'ipcontroller'))
     
     # Check security settings before proceeding
     if not check_security(args, cont_args):
         return
     
     # See if we are reusing FURL files
     if not check_reuse(args, cont_args):
         return
     
     cl = ControllerLauncher(extra_args=cont_args)
     dstart = cl.start()
     def start_engines(cont_pid):
         engine_args = []
         engine_args.append('--logfile=%s' % \
             pjoin(args.logdir,'ipengine%s-' % cont_pid))
         eset = LocalEngineSet(extra_args=engine_args)
         def shutdown(signum, frame):
             log.msg('Stopping local cluster')
             # We are still playing with the times here, but these seem
             # to be reliable in allowing everything to exit cleanly.
             eset.interrupt_then_kill(0.5)
             cl.interrupt_then_kill(0.5)
             reactor.callLater(1.0, reactor.stop)
         signal.signal(signal.SIGINT,shutdown)
         d = eset.start(args.n)
         return d
     config = kernel_config_manager.get_config_obj()
     furl_file = config['controller']['engine_furl_file']
     dstart.addCallback(_delay_start, start_engines, furl_file, args.r)
     dstart.addErrback(_err_and_stop)
 
 
 def main_mpi(args):
     cont_args = []
     cont_args.append('--logfile=%s' % pjoin(args.logdir,'ipcontroller'))
     
     # Check security settings before proceeding
     if not check_security(args, cont_args):
         return
     
     # See if we are reusing FURL files
     if not check_reuse(args, cont_args):
         return
     
     cl = ControllerLauncher(extra_args=cont_args)
     dstart = cl.start()
     def start_engines(cont_pid):
         raw_args = [args.cmd]
         raw_args.extend(['-n',str(args.n)])
         raw_args.append('ipengine')
         raw_args.append('-l')
         raw_args.append(pjoin(args.logdir,'ipengine%s-' % cont_pid))
         if args.mpi:
             raw_args.append('--mpi=%s' % args.mpi)
         eset = ProcessLauncher(raw_args)
         def shutdown(signum, frame):
             log.msg('Stopping local cluster')
             # We are still playing with the times here, but these seem
             # to be reliable in allowing everything to exit cleanly.
             eset.interrupt_then_kill(1.0)
             cl.interrupt_then_kill(1.0)
             reactor.callLater(2.0, reactor.stop)
         signal.signal(signal.SIGINT,shutdown)
         d = eset.start()
         return d
     config = kernel_config_manager.get_config_obj()
     furl_file = config['controller']['engine_furl_file']
     dstart.addCallback(_delay_start, start_engines, furl_file, args.r)
     dstart.addErrback(_err_and_stop)
 
 
 def main_pbs(args):
     cont_args = []
     cont_args.append('--logfile=%s' % pjoin(args.logdir,'ipcontroller'))
     
     # Check security settings before proceeding
     if not check_security(args, cont_args):
         return
     
     # See if we are reusing FURL files
     if not check_reuse(args, cont_args):
         return
     
     cl = ControllerLauncher(extra_args=cont_args)
     dstart = cl.start()
     def start_engines(r):
         pbs_set =  PBSEngineSet(args.pbsscript)
         def shutdown(signum, frame):
             log.msg('Stopping pbs cluster')
             d = pbs_set.kill()
             d.addBoth(lambda _: cl.interrupt_then_kill(1.0))
             d.addBoth(lambda _: reactor.callLater(2.0, reactor.stop))
         signal.signal(signal.SIGINT,shutdown)
         d = pbs_set.start(args.n)
         return d
     config = kernel_config_manager.get_config_obj()
     furl_file = config['controller']['engine_furl_file']
     dstart.addCallback(_delay_start, start_engines, furl_file, args.r)
     dstart.addErrback(_err_and_stop)
 
 
 def main_ssh(args):
     """Start a controller on localhost and engines using ssh.
     
     Your clusterfile should look like::
     
         send_furl = False # True, if you want 
         engines = {
             'engine_host1' : engine_count, 
             'engine_host2' : engine_count2
         } 
     """
     clusterfile = {}
     execfile(args.clusterfile, clusterfile)
     if not clusterfile.has_key('send_furl'):
         clusterfile['send_furl'] = False
         
     cont_args = []
     cont_args.append('--logfile=%s' % pjoin(args.logdir,'ipcontroller'))
     
     # Check security settings before proceeding
     if not check_security(args, cont_args):
         return
     
     # See if we are reusing FURL files
     if not check_reuse(args, cont_args):
         return
     
     cl = ControllerLauncher(extra_args=cont_args)
     dstart = cl.start()
     def start_engines(cont_pid):
         ssh_set = SSHEngineSet(clusterfile['engines'], sshx=args.sshx)
         def shutdown(signum, frame):
             d = ssh_set.kill()
             cl.interrupt_then_kill(1.0)
             reactor.callLater(2.0, reactor.stop)
         signal.signal(signal.SIGINT,shutdown)
         d = ssh_set.start(clusterfile['send_furl'])
         return d
     config = kernel_config_manager.get_config_obj()
     furl_file = config['controller']['engine_furl_file']
     dstart.addCallback(_delay_start, start_engines, furl_file, args.r)
     dstart.addErrback(_err_and_stop)
 
 
 def get_args():
     base_parser = argparse.ArgumentParser(add_help=False)
     base_parser.add_argument(
         '-r',
         action='store_true',
         dest='r',
         help='try to reuse FURL files.  Use with --client-port and --engine-port'
     )
     base_parser.add_argument(
         '--client-port',
         type=int,
         dest='client_port',
         help='the port the controller will listen on for client connections',
         default=0
     )
     base_parser.add_argument(
         '--engine-port',
         type=int,
         dest='engine_port',
         help='the port the controller will listen on for engine connections',
         default=0
     )
     base_parser.add_argument(
         '-x',
         action='store_true',
         dest='x',
         help='turn off client security'
     )
     base_parser.add_argument(
         '-y',
         action='store_true',
         dest='y',
         help='turn off engine security'
     )
     base_parser.add_argument(
         "--logdir", 
         type=str,
         dest="logdir",
         help="directory to put log files (default=$IPYTHONDIR/log)",
         default=pjoin(get_ipython_dir(),'log')
     )
     base_parser.add_argument(
         "-n",
         "--num", 
         type=int,
         dest="n",
         default=2,
         help="the number of engines to start"
     )
     
     parser = argparse.ArgumentParser(
         description='IPython cluster startup.  This starts a controller and\
         engines using various approaches.  Use the IPYTHONDIR environment\
         variable to change your IPython directory from the default of\
         .ipython or _ipython.  The log and security subdirectories of your\
         IPython directory will be used by this script for log files and\
         security files.'
     )
     subparsers = parser.add_subparsers(
         help='available cluster types.  For help, do "ipcluster TYPE --help"')
     
     parser_local = subparsers.add_parser(
         'local',
         help='run a local cluster',
         parents=[base_parser]
     )
     parser_local.set_defaults(func=main_local)
     
     parser_mpirun = subparsers.add_parser(
         'mpirun',
         help='run a cluster using mpirun (mpiexec also works)',
         parents=[base_parser]
     )
     parser_mpirun.add_argument(
         "--mpi",
         type=str,
         dest="mpi", # Don't put a default here to allow no MPI support
         help="how to call MPI_Init (default=mpi4py)"
     )
     parser_mpirun.set_defaults(func=main_mpi, cmd='mpirun')
 
     parser_mpiexec = subparsers.add_parser(
         'mpiexec',
         help='run a cluster using mpiexec (mpirun also works)',
         parents=[base_parser]
     )
     parser_mpiexec.add_argument(
         "--mpi",
         type=str,
         dest="mpi", # Don't put a default here to allow no MPI support
         help="how to call MPI_Init (default=mpi4py)"
     )
     parser_mpiexec.set_defaults(func=main_mpi, cmd='mpiexec')
     
     parser_pbs = subparsers.add_parser(
         'pbs', 
         help='run a pbs cluster',
         parents=[base_parser]
     )
     parser_pbs.add_argument(
         '--pbs-script',
         type=str, 
         dest='pbsscript',
         help='PBS script template',
         default='pbs.template'
     )
     parser_pbs.set_defaults(func=main_pbs)
     
     parser_ssh = subparsers.add_parser(
         'ssh',
         help='run a cluster using ssh, should have ssh-keys setup',
         parents=[base_parser]
     )
     parser_ssh.add_argument(
         '--clusterfile', 
         type=str,
         dest='clusterfile',
         help='python file describing the cluster',
         default='clusterfile.py',
     )
     parser_ssh.add_argument(
         '--sshx', 
         type=str,
         dest='sshx',
         help='sshx launcher helper'
     )
     parser_ssh.set_defaults(func=main_ssh)
     
     args = parser.parse_args()
     return args
 
 def main():
     args = get_args()
     reactor.callWhenRunning(args.func, args)
     log.startLogging(sys.stdout)
     reactor.run()
 
 if __name__ == '__main__':
     main()

IPython/kernel/twistedutil.py

@@ -1,249 +1,256 b''
 #!/usr/bin/env python
 # encoding: utf-8
 
 """Things directly related to all of twisted."""
 
 __docformat__ = "restructuredtext en"
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008  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, sys
 import threading, Queue, atexit
 
 import twisted
 from twisted.internet import defer, reactor
 from twisted.python import log, failure
 
 from IPython.kernel.error import FileTimeoutError
 
 #-------------------------------------------------------------------------------
 # Classes related to twisted and threads
 #-------------------------------------------------------------------------------
 
 
 class ReactorInThread(threading.Thread):
     """Run the twisted reactor in a different thread.
     
     For the process to be able to exit cleanly, do the following:
     
     rit = ReactorInThread()
     rit.setDaemon(True)
     rit.start()
     
     """
     
     def run(self):
         reactor.run(installSignalHandlers=0)
         # self.join()
         
     def stop(self):
         # I don't think this does anything useful.
         blockingCallFromThread(reactor.stop)
         self.join()
 
 if(twisted.version.major >= 8):
     import twisted.internet.threads
     def blockingCallFromThread(f, *a, **kw):
         """
         Run a function in the reactor from a thread, and wait for the result
         synchronously, i.e. until the callback chain returned by the function get a
         result.
         
         Delegates to twisted.internet.threads.blockingCallFromThread(reactor, f, *a, **kw),
         passing twisted.internet.reactor for the first argument.
         
         @param f: the callable to run in the reactor thread
         @type f: any callable.
         @param a: the arguments to pass to C{f}.
         @param kw: the keyword arguments to pass to C{f}.
 
         @return: the result of the callback chain.
         @raise: any error raised during the callback chain.
         """
         return twisted.internet.threads.blockingCallFromThread(reactor, f, *a, **kw)
     
 else:   
     def blockingCallFromThread(f, *a, **kw):
         """
         Run a function in the reactor from a thread, and wait for the result
         synchronously, i.e. until the callback chain returned by the function get a
         result.
 
         @param f: the callable to run in the reactor thread
         @type f: any callable.
         @param a: the arguments to pass to C{f}.
         @param kw: the keyword arguments to pass to C{f}.
 
         @return: the result of the callback chain.
         @raise: any error raised during the callback chain.
         """
         from twisted.internet import reactor
         queue = Queue.Queue()
         def _callFromThread():
             result = defer.maybeDeferred(f, *a, **kw)
             result.addBoth(queue.put)
         
         reactor.callFromThread(_callFromThread)
         result = queue.get()
         if isinstance(result, failure.Failure):
             # This makes it easier for the debugger to get access to the instance
             try:
                 result.raiseException()
             except Exception, e:
                 raise e
         return result
     
 
 
 #-------------------------------------------------------------------------------
 # Things for managing deferreds
 #-------------------------------------------------------------------------------
 
 
 def parseResults(results):
     """Pull out results/Failures from a DeferredList."""
     return [x[1] for x in results]
 
 def gatherBoth(dlist, fireOnOneCallback=0, 
                       fireOnOneErrback=0,
                       consumeErrors=0,
                       logErrors=0):
     """This is like gatherBoth, but sets consumeErrors=1."""
     d = DeferredList(dlist, fireOnOneCallback, fireOnOneErrback,
                      consumeErrors, logErrors)
     if not fireOnOneCallback:
         d.addCallback(parseResults)
     return d
 
 SUCCESS = True
 FAILURE = False
 
 class DeferredList(defer.Deferred):
     """I combine a group of deferreds into one callback.
 
     I track a list of L{Deferred}s for their callbacks, and make a single
     callback when they have all completed, a list of (success, result)
     tuples, 'success' being a boolean.
 
     Note that you can still use a L{Deferred} after putting it in a
     DeferredList.  For example, you can suppress 'Unhandled error in Deferred'
     messages by adding errbacks to the Deferreds *after* putting them in the
     DeferredList, as a DeferredList won't swallow the errors.  (Although a more
     convenient way to do this is simply to set the consumeErrors flag)
     
     Note: This is a modified version of the twisted.internet.defer.DeferredList
     """
 
     fireOnOneCallback = 0
     fireOnOneErrback = 0
 
     def __init__(self, deferredList, fireOnOneCallback=0, fireOnOneErrback=0,
                  consumeErrors=0, logErrors=0):
         """Initialize a DeferredList.
 
         @type deferredList:  C{list} of L{Deferred}s
         @param deferredList: The list of deferreds to track.
         @param fireOnOneCallback: (keyword param) a flag indicating that
                              only one callback needs to be fired for me to call
                              my callback
         @param fireOnOneErrback: (keyword param) a flag indicating that
                             only one errback needs to be fired for me to call
                             my errback
         @param consumeErrors: (keyword param) a flag indicating that any errors
                             raised in the original deferreds should be
                             consumed by this DeferredList.  This is useful to
                             prevent spurious warnings being logged.
         """
         self.resultList = [None] * len(deferredList)
         defer.Deferred.__init__(self)
         if len(deferredList) == 0 and not fireOnOneCallback:
             self.callback(self.resultList)
 
         # These flags need to be set *before* attaching callbacks to the
         # deferreds, because the callbacks use these flags, and will run
         # synchronously if any of the deferreds are already fired.
         self.fireOnOneCallback = fireOnOneCallback
         self.fireOnOneErrback = fireOnOneErrback
         self.consumeErrors = consumeErrors
         self.logErrors = logErrors
         self.finishedCount = 0
 
         index = 0
         for deferred in deferredList:
             deferred.addCallbacks(self._cbDeferred, self._cbDeferred,
                                   callbackArgs=(index,SUCCESS),
                                   errbackArgs=(index,FAILURE))
             index = index + 1
 
     def _cbDeferred(self, result, index, succeeded):
         """(internal) Callback for when one of my deferreds fires.
         """
         self.resultList[index] = (succeeded, result)
 
         self.finishedCount += 1
         if not self.called:
             if succeeded == SUCCESS and self.fireOnOneCallback:
                 self.callback((result, index))
             elif succeeded == FAILURE and self.fireOnOneErrback:
                 # We have modified this to fire the errback chain with the actual
                 # Failure instance the originally occured rather than twisted's
                 # FirstError which wraps the failure
                 self.errback(result)
             elif self.finishedCount == len(self.resultList):
                 self.callback(self.resultList)
 
         if succeeded == FAILURE and self.logErrors:
             log.err(result)
         if succeeded == FAILURE and self.consumeErrors:
             result = None
 
         return result
 
 
 def wait_for_file(filename, delay=0.1, max_tries=10):
     """Wait (poll) for a file to be created.
     
     This method returns a Deferred that will fire when a file exists. It
     works by polling os.path.isfile in time intervals specified by the
     delay argument.  If `max_tries` is reached, it will errback with a 
     `FileTimeoutError`.
     
     Parameters
     ----------
     filename : str
         The name of the file to wait for.
     delay : float
         The time to wait between polls.
     max_tries : int
         The max number of attempts before raising `FileTimeoutError`
     
     Returns
     -------
     d : Deferred
         A Deferred instance that will fire when the file exists.
     """
     
     d = defer.Deferred()
     
     def _test_for_file(filename, attempt=0):
         if attempt >= max_tries:
             d.errback(FileTimeoutError(
                 'timeout waiting for file to be created: %s' % filename
             ))
         else:
             if os.path.isfile(filename):
                 d.callback(True)
             else:
                 reactor.callLater(delay, _test_for_file, filename, attempt+1)
     
     _test_for_file(filename)
     return d
+
+
+def sleep_deferred(seconds):
+    """Sleep without blocking the event loop."""
+    d = defer.Deferred()
+    reactor.callLater(seconds, d.callback, seconds)
+    return d

IPython/utils/notification.py

@@ -1,125 +1,143 b''
+#!/usr/bin/env python
 # encoding: utf-8
-
-"""The IPython Core Notification Center.
+"""
+The IPython Core Notification Center.
 
 See docs/source/development/notification_blueprint.txt for an overview of the
 notification module.
+
+Authors:
+
+* Barry Wark
+* Brian Granger
 """
 
-__docformat__ = "restructuredtext en"
-                                                                              
 #-----------------------------------------------------------------------------
-#  Copyright (C) 2008  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.                    
+#  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.
+#-----------------------------------------------------------------------------
+
+#-----------------------------------------------------------------------------
+# Code
 #-----------------------------------------------------------------------------
 
-# Tell nose to skip the testing of this module
-__test__ = {}
+
+class NotificationError(Exception):
+    pass
+
 
 class NotificationCenter(object):
-    """Synchronous notification center
+    """Synchronous notification center.
     
     Examples
     --------
-    >>> import IPython.kernel.core.notification as notification
-    >>> def callback(theType, theSender, args={}):
-    ...     print theType,theSender,args
-    ...     
-    >>> notification.sharedCenter.add_observer(callback, 'NOTIFICATION_TYPE', None)
-    >>> notification.sharedCenter.post_notification('NOTIFICATION_TYPE', object()) # doctest:+ELLIPSIS
-    NOTIFICATION_TYPE ...
-        
+    Here is a simple example of how to use this::
+
+        import IPython.kernel.core.notification as notification
+        def callback(ntype, theSender, args={}):
+            print ntype,theSender,args
+
+        notification.sharedCenter.add_observer(callback, 'NOTIFICATION_TYPE', None)
+        notification.sharedCenter.post_notification('NOTIFICATION_TYPE', object()) # doctest:+ELLIPSIS
+        NOTIFICATION_TYPE ...
     """
     def __init__(self):
         super(NotificationCenter, self).__init__()
         self._init_observers()
-    
-    
+
     def _init_observers(self):
         """Initialize observer storage"""
-        
+
         self.registered_types = set() #set of types that are observed
         self.registered_senders = set() #set of senders that are observed
         self.observers = {} #map (type,sender) => callback (callable)
-    
-    
-    def post_notification(self, theType, sender, **kwargs):
-        """Post notification (type,sender,**kwargs) to all registered
-        observers.
 
-        Implementation notes:
+    def post_notification(self, ntype, sender, *args, **kwargs):
+        """Post notification to all registered observers.
+
+        The registered callback will be called as::
+
+            callback(ntype, sender, *args, **kwargs)
+
+        Parameters
+        ----------
+        ntype : hashable
+            The notification type.
+        sender : hashable
+            The object sending the notification.
+        *args : tuple
+            The positional arguments to be passed to the callback.
+        **kwargs : dict
+            The keyword argument to be passed to the callback.
 
+        Notes
+        -----
         * If no registered observers, performance is O(1).
         * Notificaiton order is undefined.
         * Notifications are posted synchronously.
         """
-        
-        if(theType==None or sender==None):
-            raise Exception("NotificationCenter.post_notification requires \
-                type and sender.")
-        
+
+        if(ntype==None or sender==None):
+            raise NotificationError(
+                "Notification type and sender are required.")
+
         # If there are no registered observers for the type/sender pair
-        if((theType not in self.registered_types and 
+        if((ntype not in self.registered_types and 
                 None not in self.registered_types) or
             (sender not in self.registered_senders and 
                 None not in self.registered_senders)):
             return
-        
-        for o in self._observers_for_notification(theType, sender):
-            o(theType, sender, args=kwargs)
-    
-        
-    def _observers_for_notification(self, theType, sender):
+
+        for o in self._observers_for_notification(ntype, sender):
+            o(ntype, sender, *args, **kwargs)
+
+    def _observers_for_notification(self, ntype, sender):
         """Find all registered observers that should recieve notification"""
-        
+
         keys = (
-                    (theType,sender),
-                    (theType, None),
-                    (None, sender),  
-                    (None,None)
-                )
-        
-        
+                   (ntype,sender),
+                   (ntype, None),
+                   (None, sender),
+                   (None,None)
+               )
+
         obs = set()
         for k in keys:
             obs.update(self.observers.get(k, set()))
-        
+
         return obs
-    
-    
-    def add_observer(self, callback, theType, sender):
+
+    def add_observer(self, callback, ntype, sender):
         """Add an observer callback to this notification center.
         
         The given callback will be called upon posting of notifications of
-        the given type/sender and will receive any additional kwargs passed
+        the given type/sender and will receive any additional arguments passed
         to post_notification.
         
         Parameters
         ----------
-        observerCallback : callable
-            Callable. Must take at least two arguments::
-                observerCallback(type, sender, args={})
-        
-        theType : hashable
+        callback : callable
+            The callable that will be called by :meth:`post_notification`
+            as ``callback(ntype, sender, *args, **kwargs)
+        ntype : hashable
             The notification type. If None, all notifications from sender
             will be posted.
-        
         sender : hashable
-            The notification sender. If None, all notifications of theType
+            The notification sender. If None, all notifications of ntype
             will be posted.
         """
         assert(callback != None)
-        self.registered_types.add(theType)
+        self.registered_types.add(ntype)
         self.registered_senders.add(sender)
-        self.observers.setdefault((theType,sender), set()).add(callback)
+        self.observers.setdefault((ntype,sender), set()).add(callback)
     
     def remove_all_observers(self):
         """Removes all observers from this notification center"""
         
         self._init_observers()
     
 
 
-sharedCenter = NotificationCenter()
+shared_center = NotificationCenter()

IPython/utils/tests/test_notification.py

@@ -1,161 +1,154 b''
 # encoding: utf-8
 
 """This file contains unittests for the notification.py module."""
 
 #-----------------------------------------------------------------------------
 #  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
 #-----------------------------------------------------------------------------
 
-# Tell nose to skip this module
-__test__ = {}
+import unittest
 
-from twisted.trial import unittest
-import IPython.kernel.core.notification as notification
+from IPython.utils.notification import (
+    NotificationCenter,
+    NotificationError,
+    shared_center
+)
 
 #-----------------------------------------------------------------------------
 # Support Classes
 #-----------------------------------------------------------------------------
 
+
 class Observer(object):
-    """docstring for Observer"""
-    def __init__(self, expectedType, expectedSender, 
-                    center=notification.sharedCenter, **kwargs):
+
+    def __init__(self, expected_ntype, expected_sender, 
+                    center=shared_center, *args, **kwargs):
         super(Observer, self).__init__()
-        self.expectedType = expectedType
-        self.expectedSender = expectedSender
-        self.expectedKwArgs = kwargs
+        self.expected_ntype = expected_ntype
+        self.expected_sender = expected_sender
+        self.expected_args = args
+        self.expected_kwargs = kwargs
         self.recieved = False
         center.add_observer(self.callback, 
-                            self.expectedType, 
-                            self.expectedSender)
+                            self.expected_ntype, 
+                            self.expected_sender)
     
-    def callback(self, theType, sender, args={}):
-        """callback"""
-        
-        assert(theType == self.expectedType or
-                self.expectedType == None)
-        assert(sender == self.expectedSender or
-                self.expectedSender == None)
-        assert(args == self.expectedKwArgs)
+    def callback(self, ntype, sender, *args, **kwargs):
+        assert(ntype == self.expected_ntype or
+                self.expected_ntype == None)
+        assert(sender == self.expected_sender or
+                self.expected_sender == None)
+        assert(args == self.expected_args)
+        assert(kwargs == self.expected_kwargs)
         self.recieved = True
     
     def verify(self):
-        """verify"""
-        
         assert(self.recieved)
     
     def reset(self):
-        """reset"""
-        
         self.recieved = False
 
 
 class Notifier(object):
-    """docstring for Notifier"""
-    def __init__(self, theType, **kwargs):
+
+    def __init__(self, ntype, **kwargs):
         super(Notifier, self).__init__()
-        self.theType = theType
+        self.ntype = ntype
         self.kwargs = kwargs
-    
-    def post(self, center=notification.sharedCenter):
-        """fire"""
-        
-        center.post_notification(self.theType, self,
+
+    def post(self, center=shared_center):
+
+        center.post_notification(self.ntype, self,
             **self.kwargs)
 
+
 #-----------------------------------------------------------------------------
 # Tests
 #-----------------------------------------------------------------------------
 
+
 class NotificationTests(unittest.TestCase):
-    """docstring for NotificationTests"""
-    
+
     def tearDown(self):
-        notification.sharedCenter.remove_all_observers()
+        shared_center.remove_all_observers()
     
     def test_notification_delivered(self):
         """Test that notifications are delivered"""
-        expectedType = 'EXPECTED_TYPE'
-        sender = Notifier(expectedType)
-        observer = Observer(expectedType, sender)
-        
+
+        expected_ntype = 'EXPECTED_TYPE'
+        sender = Notifier(expected_ntype)
+        observer = Observer(expected_ntype, sender)
+
         sender.post()
-        
         observer.verify()
-    
+
     def test_type_specificity(self):
         """Test that observers are registered by type"""
-        
-        expectedType = 1
-        unexpectedType = "UNEXPECTED_TYPE"
-        sender = Notifier(expectedType)
-        unexpectedSender = Notifier(unexpectedType)
-        observer = Observer(expectedType, sender)
-    
+
+        expected_ntype = 1
+        unexpected_ntype = "UNEXPECTED_TYPE"
+        sender = Notifier(expected_ntype)
+        unexpected_sender = Notifier(unexpected_ntype)
+        observer = Observer(expected_ntype, sender)
+
         sender.post()
-        unexpectedSender.post()
-    
+        unexpected_sender.post()
         observer.verify()
-    
+
     def test_sender_specificity(self):
         """Test that observers are registered by sender"""
         
-        expectedType = "EXPECTED_TYPE"
-        sender1 = Notifier(expectedType)
-        sender2 = Notifier(expectedType)
-        observer = Observer(expectedType, sender1)
-        
+        expected_ntype = "EXPECTED_TYPE"
+        sender1 = Notifier(expected_ntype)
+        sender2 = Notifier(expected_ntype)
+        observer = Observer(expected_ntype, sender1)
+
         sender1.post()
         sender2.post()
-        
+
         observer.verify()
-    
+
     def test_remove_all_observers(self):
         """White-box test for remove_all_observers"""
-        
+
         for i in xrange(10):
-            Observer('TYPE', None, center=notification.sharedCenter)
-        
-        self.assert_(len(notification.sharedCenter.observers[('TYPE',None)]) >= 10, 
+            Observer('TYPE', None, center=shared_center)
+
+        self.assert_(len(shared_center.observers[('TYPE',None)]) >= 10, 
             "observers registered")
-        
-        notification.sharedCenter.remove_all_observers()
-        
-        self.assert_(len(notification.sharedCenter.observers) == 0, "observers removed")
+
+        shared_center.remove_all_observers()
+        self.assert_(len(shared_center.observers) == 0, "observers removed")
 
     def test_any_sender(self):
-        """test_any_sender"""
-        
-        expectedType = "EXPECTED_TYPE"
-        sender1 = Notifier(expectedType)
-        sender2 = Notifier(expectedType)
-        observer = Observer(expectedType, None)
-        
-        
+        expected_ntype = "EXPECTED_TYPE"
+        sender1 = Notifier(expected_ntype)
+        sender2 = Notifier(expected_ntype)
+        observer = Observer(expected_ntype, None)
+
         sender1.post()
         observer.verify()
-        
+
         observer.reset()
         sender2.post()
         observer.verify()
 
     def test_post_performance(self):
         """Test that post_notification, even with many registered irrelevant
         observers is fast"""
         
         for i in xrange(10):
             Observer("UNRELATED_TYPE", None)
-        
+
         o = Observer('EXPECTED_TYPE', None)
-        
-        notification.sharedCenter.post_notification('EXPECTED_TYPE', self)
-        
+        shared_center.post_notification('EXPECTED_TYPE', self)
         o.verify()
 
+

setup.py

@@ -1,210 +1,210 b''
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 """Setup script for IPython.
 
 Under Posix environments it works like a typical setup.py script. 
 Under Windows, the command sdist is not supported, since IPython 
 requires utilities which are not available under Windows."""
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008  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
 #-------------------------------------------------------------------------------
 
 # Stdlib imports
 import os
 import sys
 
 from glob import glob
 
 # BEFORE importing distutils, remove MANIFEST. distutils doesn't properly
 # update it when the contents of directories change.
 if os.path.exists('MANIFEST'): os.remove('MANIFEST')
 
 from distutils.core import setup
 
 from IPython.utils.genutils import target_update
 
 from setupbase import (
     setup_args, 
     find_packages, 
     find_package_data, 
     find_scripts,
     find_data_files,
     check_for_dependencies
 )
 
 isfile = os.path.isfile
 pjoin = os.path.join
 
 #-------------------------------------------------------------------------------
 # Handle OS specific things
 #-------------------------------------------------------------------------------
 
 if os.name == 'posix':
     os_name = 'posix'
 elif os.name in ['nt','dos']:
     os_name = 'windows'
 else:
     print 'Unsupported operating system:',os.name
     sys.exit(1)
 
 # Under Windows, 'sdist' has not been supported.  Now that the docs build with
 # Sphinx it might work, but let's not turn it on until someone confirms that it
 # actually works.
 if os_name == 'windows' and 'sdist' in sys.argv:
     print 'The sdist command is not available under Windows.  Exiting.'
     sys.exit(1)
 
 #-------------------------------------------------------------------------------
 # Things related to the IPython documentation
 #-------------------------------------------------------------------------------
 
 # update the manuals when building a source dist
 if len(sys.argv) >= 2 and sys.argv[1] in ('sdist','bdist_rpm'):
     import textwrap
 
     # List of things to be updated. Each entry is a triplet of args for
     # target_update()
     to_update = [
                   # FIXME - Disabled for now: we need to redo an automatic way
                   # of generating the magic info inside the rst.                
                   #('docs/magic.tex',
                   #['IPython/Magic.py'],
                   #"cd doc && ./update_magic.sh" ),
 
                  ('docs/man/ipcluster.1.gz',
                   ['docs/man/ipcluster.1'],
                   'cd docs/man && gzip -9c ipcluster.1 > ipcluster.1.gz'),
 
                  ('docs/man/ipcontroller.1.gz',
                   ['docs/man/ipcontroller.1'],
                   'cd docs/man && gzip -9c ipcontroller.1 > ipcontroller.1.gz'),
 
                  ('docs/man/ipengine.1.gz',
                   ['docs/man/ipengine.1'],
                   'cd docs/man && gzip -9c ipengine.1 > ipengine.1.gz'),
 
                  ('docs/man/ipython.1.gz',
                   ['docs/man/ipython.1'],
                   'cd docs/man && gzip -9c ipython.1 > ipython.1.gz'),
 
                  ('docs/man/ipython-wx.1.gz',
                   ['docs/man/ipython-wx.1'],
                   'cd docs/man && gzip -9c ipython-wx.1 > ipython-wx.1.gz'),
 
                  ('docs/man/ipythonx.1.gz',
                   ['docs/man/ipythonx.1'],
                   'cd docs/man && gzip -9c ipythonx.1 > ipythonx.1.gz'),
 
                  ('docs/man/irunner.1.gz',
                   ['docs/man/irunner.1'],
                   'cd docs/man && gzip -9c irunner.1 > irunner.1.gz'),
 
                  ('docs/man/pycolor.1.gz',
                   ['docs/man/pycolor.1'],
                   'cd docs/man && gzip -9c pycolor.1 > pycolor.1.gz'),
                  ]
 
     # Only build the docs if sphinx is present
     try:
         import sphinx
     except ImportError:
         pass
     else:
         # The Makefile calls the do_sphinx scripts to build html and pdf, so
         # just one target is enough to cover all manual generation
 
         # First, compute all the dependencies that can force us to rebuild the
         # docs.  Start with the main release file that contains metadata
         docdeps = ['IPython/core/release.py']
         # Inculde all the reST sources
         pjoin = os.path.join
         for dirpath,dirnames,filenames in os.walk('docs/source'):
             if dirpath in ['_static','_templates']:
                 continue
             docdeps += [ pjoin(dirpath,f) for f in filenames
                          if f.endswith('.txt') ]
         # and the examples
         for dirpath,dirnames,filenames in os.walk('docs/example'):
             docdeps += [ pjoin(dirpath,f) for f in filenames
                          if not f.endswith('~') ]
         # then, make them all dependencies for the main PDF (the html will get
         # auto-generated as well).
         to_update.append(
             ('docs/dist/ipython.pdf',
              docdeps,
              "cd docs && make dist")
             )
         
     [ target_update(*t) for t in to_update ]
 
     
 #---------------------------------------------------------------------------
 # Find all the packages, package data, scripts and data_files
 #---------------------------------------------------------------------------
 
 packages = find_packages()
 package_data = find_package_data()
 scripts = find_scripts()
 data_files = find_data_files()
 
 #---------------------------------------------------------------------------
 # Handle dependencies and setuptools specific things
 #---------------------------------------------------------------------------
 
 # This dict is used for passing extra arguments that are setuptools 
 # specific to setup
 setuptools_extra_args = {}
 
 if 'setuptools' in sys.modules:
     setuptools_extra_args['zip_safe'] = False
     setuptools_extra_args['entry_points'] = {
         'console_scripts': [
             'ipython = IPython.core.ipapp:launch_new_instance',
             'pycolor = IPython.utils.PyColorize:main',
             'ipcontroller = IPython.kernel.ipcontrollerapp:launch_new_instance',
             'ipengine = IPython.kernel.ipengineapp:launch_new_instance',
-            'ipcluster = IPython.kernel.scripts.ipcluster:main',
+            'ipcluster = IPython.kernel.ipclusterapp:launch_new_instance',
             'ipythonx = IPython.frontend.wx.ipythonx:main',
             'iptest = IPython.testing.iptest:main',
             'irunner = IPython.lib.irunner:main'
         ]
     }
     setup_args['extras_require'] = dict(
         kernel = [
             'zope.interface>=3.4.1',
             'Twisted>=8.0.1',
             'foolscap>=0.2.6'
         ],
         doc='Sphinx>=0.3',
         test='nose>=0.10.1',
         security='pyOpenSSL>=0.6'
     )
     # Allow setuptools to handle the scripts
     scripts = []
 else:
     # If we are running without setuptools, call this function which will
     # check for dependencies an inform the user what is needed.  This is
     # just to make life easy for users.
     check_for_dependencies()
 
 
 #---------------------------------------------------------------------------
 # Do the actual setup now
 #---------------------------------------------------------------------------
 
 setup_args['packages'] = packages
 setup_args['package_data'] = package_data
 setup_args['scripts'] = scripts
 setup_args['data_files'] = data_files
 setup_args.update(setuptools_extra_args)
 
 if __name__ == '__main__':
     setup(**setup_args)